Commit 1906791e authored by Jakob Otto's avatar Jakob Otto

Merge branch 'master' into topic/transport-worker-dispatcher

parents 6ccd5d8c b759768a
...@@ -27,6 +27,8 @@ namespace net { ...@@ -27,6 +27,8 @@ namespace net {
class multiplexer; class multiplexer;
class socket_manager; class socket_manager;
template <class Application, class IdType = unit_t>
class transport_worker;
struct network_socket; struct network_socket;
struct pipe_socket; struct pipe_socket;
......
...@@ -65,7 +65,8 @@ public: ...@@ -65,7 +65,8 @@ public:
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
worker_.init(parent); if (auto err = worker_.init(parent))
return err;
parent.mask_add(operation::read); parent.mask_add(operation::read);
return none; return none;
} }
......
...@@ -29,8 +29,8 @@ ...@@ -29,8 +29,8 @@
namespace caf { namespace caf {
namespace net { namespace net {
/// implements a worker for the udp_transport policy /// Implements a worker for transport protocols.
template <class Application, class IdType = unit_t> template <class Application, class IdType>
class transport_worker { class transport_worker {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
} }
template <class Parent> template <class Parent>
void handle_data(Parent& parent, span<byte> data) { void handle_data(Parent& parent, span<const byte> data) {
application_.handle_data(parent, data); application_.handle_data(parent, data);
} }
......
...@@ -54,8 +54,8 @@ public: ...@@ -54,8 +54,8 @@ public:
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Header, class... Ts> template <class... Ts>
void write_packet(const Header& header, span<const byte> payload, void write_packet(span<const byte> header, span<const byte> payload,
Ts&&... xs) { Ts&&... xs) {
object_.write_packet(parent_, header, payload, std::forward<Ts>(xs)...); object_.write_packet(parent_, header, payload, std::forward<Ts>(xs)...);
} }
......
...@@ -37,6 +37,8 @@ struct dummy_socket { ...@@ -37,6 +37,8 @@ struct dummy_socket {
// nop // nop
} }
dummy_socket(const dummy_socket&) = default;
dummy_socket& operator=(const dummy_socket& other) { dummy_socket& operator=(const dummy_socket& other) {
id = other.id; id = other.id;
closed = other.closed; closed = other.closed;
......
...@@ -101,10 +101,8 @@ public: ...@@ -101,10 +101,8 @@ public:
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) { std::unique_ptr<net::endpoint_manager::message> msg) {
header_type header{static_cast<uint32_t>(msg->payload.size())}; header_type header{static_cast<uint32_t>(msg->payload.size())};
std::vector<byte> header_buf(sizeof(header_type));
std::vector<byte> payload(msg->payload.begin(), msg->payload.end()); std::vector<byte> payload(msg->payload.begin(), msg->payload.end());
memcpy(header_buf.data(), &header, header_buf.size()); parent.write_packet(as_bytes(make_span(&header, 1)), make_span(payload));
parent.write_packet(make_span(header_buf), make_span(payload));
} }
static expected<std::vector<byte>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
......
...@@ -122,8 +122,8 @@ public: ...@@ -122,8 +122,8 @@ public:
// nop // nop
} }
template <class Header> void write_packet(span<const byte> header, span<const byte> payload,
void write_packet(Header header, span<const byte> payload, ip_endpoint ep) { ip_endpoint ep) {
auto& buf = res_->packet_buffer; auto& buf = res_->packet_buffer;
buf.insert(buf.begin(), header.begin(), header.end()); buf.insert(buf.begin(), header.begin(), header.end());
buf.insert(buf.begin(), payload.begin(), payload.end()); buf.insert(buf.begin(), payload.begin(), payload.end());
...@@ -173,9 +173,7 @@ CAF_TEST(construction and initialization) { ...@@ -173,9 +173,7 @@ CAF_TEST(construction and initialization) {
} }
CAF_TEST(handle_data) { CAF_TEST(handle_data) {
auto test_span = make_span(reinterpret_cast<byte*>( auto test_span = as_bytes(make_span(hello_test));
const_cast<char*>(hello_test.data())),
hello_test.size());
worker.handle_data(transport, test_span); worker.handle_data(transport, test_span);
auto& buf = application_results->data_buffer; auto& buf = application_results->data_buffer;
string_view result{reinterpret_cast<char*>(buf.data()), buf.size()}; string_view result{reinterpret_cast<char*>(buf.data()), buf.size()};
...@@ -193,9 +191,8 @@ CAF_TEST(write_message) { ...@@ -193,9 +191,8 @@ CAF_TEST(write_message) {
const_cast<char*>(hello_test.data())), const_cast<char*>(hello_test.data())),
hello_test.size()); hello_test.size());
std::vector<byte> payload(test_span.begin(), test_span.end()); std::vector<byte> payload(test_span.begin(), test_span.end());
auto message = caf::detail::make_unique<endpoint_manager::message>(std::move( auto message = detail::make_unique<endpoint_manager::message>(std::move(elem),
elem), payload);
payload);
worker.write_message(transport, std::move(message)); worker.write_message(transport, std::move(message));
auto& buf = transport_results->packet_buffer; auto& buf = transport_results->packet_buffer;
string_view result{reinterpret_cast<char*>(buf.data()), buf.size()}; string_view result{reinterpret_cast<char*>(buf.data()), buf.size()};
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment