Commit 0bd912b4 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #67

Move serialization to multiplexer
parents 006f5ce1 061042f7
...@@ -37,7 +37,6 @@ public: ...@@ -37,7 +37,6 @@ public:
void kill_proxy(execution_unit* ctx, error rsn) override; void kill_proxy(execution_unit* ctx, error rsn) override;
private: private:
endpoint_manager::serialize_fun_type sf_;
endpoint_manager_ptr dst_; endpoint_manager_ptr dst_;
}; };
......
...@@ -136,8 +136,6 @@ public: ...@@ -136,8 +136,6 @@ public:
// nop // nop
} }
static expected<buffer_type> serialize(actor_system& sys, const message& x);
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path); strong_actor_ptr resolve_local_path(string_view path);
......
...@@ -44,12 +44,6 @@ public: ...@@ -44,12 +44,6 @@ public:
using super = socket_manager; using super = socket_manager;
/// Represents either an error or a serialized payload.
using maybe_buffer = expected<std::vector<byte>>;
/// A function type for serializing message payloads.
using serialize_fun_type = maybe_buffer (*)(actor_system&, const message&);
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
endpoint_manager(socket handle, const multiplexer_ptr& parent, endpoint_manager(socket handle, const multiplexer_ptr& parent,
...@@ -71,8 +65,7 @@ public: ...@@ -71,8 +65,7 @@ public:
void resolve(uri locator, actor listener); void resolve(uri locator, actor listener);
/// Enqueues a message to the endpoint. /// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver, void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver);
std::vector<byte> payload);
/// Enqueues an event to the endpoint. /// Enqueues an event to the endpoint.
template <class... Ts> template <class... Ts>
...@@ -85,9 +78,6 @@ public: ...@@ -85,9 +78,6 @@ public:
/// Initializes the manager before adding it to the multiplexer's event loop. /// Initializes the manager before adding it to the multiplexer's event loop.
virtual error init() = 0; virtual error init() = 0;
/// @returns the protocol-specific function for serializing payloads.
virtual serialize_fun_type serialize_fun() const noexcept = 0;
protected: protected:
bool enqueue(endpoint_manager_queue::element* ptr); bool enqueue(endpoint_manager_queue::element* ptr);
......
...@@ -62,8 +62,8 @@ public: ...@@ -62,8 +62,8 @@ public:
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
template <class... Ts> template <class... Ts>
uint64_t set_timeout(actor_clock::time_point tp, std::string type, uint64_t
Ts&&... xs) { set_timeout(actor_clock::time_point tp, std::string type, Ts&&... xs) {
auto act = actor_cast<abstract_actor*>(timeout_proxy_); auto act = actor_cast<abstract_actor*>(timeout_proxy_);
CAF_ASSERT(act != nullptr); CAF_ASSERT(act != nullptr);
sys_.clock().set_multi_timeout(tp, act, std::move(type), next_timeout_id_); sys_.clock().set_multi_timeout(tp, act, std::move(type), next_timeout_id_);
...@@ -119,10 +119,6 @@ public: ...@@ -119,10 +119,6 @@ public:
transport_.handle_error(code); transport_.handle_error(code);
} }
serialize_fun_type serialize_fun() const noexcept override {
return application_type::serialize;
}
private: private:
transport_type transport_; transport_type transport_;
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include <string> #include <string>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/detail/serialized_size.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp" #include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp" #include "caf/intrusive/fifo_inbox.hpp"
...@@ -112,7 +113,7 @@ public: ...@@ -112,7 +113,7 @@ public:
// nop // nop
} }
task_size_type task_size(const event&) const noexcept { static constexpr task_size_type task_size(const event&) noexcept {
return 1; return 1;
} }
}; };
...@@ -125,11 +126,7 @@ public: ...@@ -125,11 +126,7 @@ public:
/// ID of the receiving actor. /// ID of the receiving actor.
strong_actor_ptr receiver; strong_actor_ptr receiver;
/// Serialized representation of of `msg->content()`. message(mailbox_element_ptr msg, strong_actor_ptr receiver);
std::vector<byte> payload;
message(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
~message() override; ~message() override;
...@@ -153,9 +150,8 @@ public: ...@@ -153,9 +150,8 @@ public:
// nop // nop
} }
static task_size_type task_size(const message& x) noexcept { static task_size_type task_size(const message& msg) noexcept {
// Return at least 1 if the payload is empty. return detail::serialized_size(msg.msg->content());
return x.payload.size() + static_cast<task_size_type>(x.payload.empty());
} }
}; };
......
...@@ -25,7 +25,7 @@ ...@@ -25,7 +25,7 @@
namespace caf::net { namespace caf::net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst) actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) { : super(cfg), dst_(std::move(dst)) {
CAF_ASSERT(dst_ != nullptr); CAF_ASSERT(dst_ != nullptr);
dst_->enqueue_event(node(), id()); dst_->enqueue_event(node(), id());
} }
...@@ -34,14 +34,11 @@ actor_proxy_impl::~actor_proxy_impl() { ...@@ -34,14 +34,11 @@ actor_proxy_impl::~actor_proxy_impl() {
// nop // nop
} }
void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) { void actor_proxy_impl::enqueue(mailbox_element_ptr msg, execution_unit*) {
CAF_PUSH_AID(0); CAF_PUSH_AID(0);
CAF_ASSERT(what != nullptr); CAF_ASSERT(msg != nullptr);
CAF_LOG_SEND_EVENT(what); CAF_LOG_SEND_EVENT(msg);
if (auto payload = sf_(home_system(), what->content())) dst_->enqueue(std::move(msg), ctrl());
dst_->enqueue(std::move(what), ctrl(), std::move(*payload));
else
CAF_LOG_ERROR("unable to serialize payload: " << payload.error());
} }
void actor_proxy_impl::kill_proxy(execution_unit* ctx, error rsn) { void actor_proxy_impl::kill_proxy(execution_unit* ctx, error rsn) {
......
...@@ -52,14 +52,14 @@ error application::write_message( ...@@ -52,14 +52,14 @@ error application::write_message(
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr); CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content())); CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
auto payload_prefix = writer.next_payload_buffer();
binary_serializer sink{system(), payload_prefix};
const auto& src = ptr->msg->sender; const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver; const auto& dst = ptr->receiver;
if (dst == nullptr) { if (dst == nullptr) {
// TODO: valid? // TODO: valid?
return none; return none;
} }
auto payload_buf = writer.next_payload_buffer();
binary_serializer sink{system(), payload_buf};
if (src != nullptr) { if (src != nullptr) {
auto src_id = src->id(); auto src_id = src->id();
system().registry().put(src_id, src); system().registry().put(src_id, src);
...@@ -69,13 +69,14 @@ error application::write_message( ...@@ -69,13 +69,14 @@ error application::write_message(
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages)) if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err; return err;
} }
if (auto err = sink(ptr->msg->content()))
return err;
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::actor_message, to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_prefix.size() static_cast<uint32_t>(payload_buf.size()),
+ ptr->payload.size()),
ptr->msg->mid.integer_value()}, ptr->msg->mid.integer_value()},
hdr); hdr);
writer.write_packet(hdr, payload_prefix, ptr->payload); writer.write_packet(hdr, payload_buf);
return none; return none;
} }
...@@ -118,15 +119,6 @@ void application::local_actor_down(packet_writer& writer, actor_id id, ...@@ -118,15 +119,6 @@ void application::local_actor_down(packet_writer& writer, actor_id id,
writer.write_packet(hdr, payload); writer.write_packet(hdr, payload);
} }
expected<std::vector<byte>> application::serialize(actor_system& sys,
const message& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
}
strong_actor_ptr application::resolve_local_path(string_view path) { strong_actor_ptr application::resolve_local_path(string_view path) {
CAF_LOG_TRACE(CAF_ARG(path)); CAF_LOG_TRACE(CAF_ARG(path));
// We currently support two path formats: `id/<actor_id>` and `name/<atom>`. // We currently support two path formats: `id/<actor_id>` and `name/<atom>`.
......
...@@ -60,11 +60,9 @@ void endpoint_manager::resolve(uri locator, actor listener) { ...@@ -60,11 +60,9 @@ void endpoint_manager::resolve(uri locator, actor listener) {
} }
void endpoint_manager::enqueue(mailbox_element_ptr msg, void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver, strong_actor_ptr receiver) {
std::vector<byte> payload) {
using message_type = endpoint_manager_queue::message; using message_type = endpoint_manager_queue::message;
auto ptr = new message_type(std::move(msg), std::move(receiver), auto ptr = new message_type(std::move(msg), std::move(receiver));
std::move(payload));
enqueue(ptr); enqueue(ptr);
} }
......
...@@ -56,12 +56,10 @@ size_t endpoint_manager_queue::event::task_size() const noexcept { ...@@ -56,12 +56,10 @@ size_t endpoint_manager_queue::event::task_size() const noexcept {
} }
endpoint_manager_queue::message::message(mailbox_element_ptr msg, endpoint_manager_queue::message::message(mailbox_element_ptr msg,
strong_actor_ptr receiver, strong_actor_ptr receiver)
std::vector<byte> payload)
: element(element_type::message), : element(element_type::message),
msg(std::move(msg)), msg(std::move(msg)),
receiver(std::move(receiver)), receiver(std::move(receiver)) {
payload(std::move(payload)) {
// nop // nop
} }
......
...@@ -119,10 +119,14 @@ public: ...@@ -119,10 +119,14 @@ public:
return none; return none;
} }
template <class Transport> template <class Parent>
void write_message(Transport& transport, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
transport.write_packet(msg->payload); auto payload_buf = parent.next_payload_buffer();
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
} }
template <class Parent> template <class Parent>
...@@ -139,10 +143,8 @@ public: ...@@ -139,10 +143,8 @@ public:
auto nid = make_node_id(uri); auto nid = make_node_id(uri);
actor_config cfg; actor_config cfg;
endpoint_manager_ptr ptr{&parent.manager()}; endpoint_manager_ptr ptr{&parent.manager()};
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid, auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(
&parent.system(), aid, nid, &parent.system(), cfg, std::move(ptr));
cfg,
std::move(ptr));
anon_send(listener, resolve_atom_v, std::string{path.begin(), path.end()}, anon_send(listener, resolve_atom_v, std::string{path.begin(), path.end()},
p); p);
} }
...@@ -166,14 +168,6 @@ public: ...@@ -166,14 +168,6 @@ public:
CAF_FAIL("handle_error called: " << to_string(sec)); CAF_FAIL("handle_error called: " << to_string(sec));
} }
static expected<buffer_type> serialize(actor_system& sys, const message& x) {
buffer_type result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
}
private: private:
buffer_ptr rec_buf_; buffer_ptr rec_buf_;
}; };
...@@ -204,10 +198,9 @@ CAF_TEST(receive) { ...@@ -204,10 +198,9 @@ CAF_TEST(receive) {
using transport_type = datagram_transport<dummy_application_factory>; using transport_type = datagram_transport<dummy_application_factory>;
if (auto err = nonblocking(recv_socket, true)) if (auto err = nonblocking(recv_socket, true))
CAF_FAIL("nonblocking() returned an error: " << err); CAF_FAIL("nonblocking() returned an error: " << err);
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(
transport_type{recv_socket, mpx, sys,
dummy_application_factory{ transport_type{recv_socket, dummy_application_factory{shared_buf}});
shared_buf}});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>(); auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>();
CAF_CHECK(mgr_impl != nullptr); CAF_CHECK(mgr_impl != nullptr);
...@@ -227,10 +220,9 @@ CAF_TEST(resolve and proxy communication) { ...@@ -227,10 +220,9 @@ CAF_TEST(resolve and proxy communication) {
using transport_type = datagram_transport<dummy_application_factory>; using transport_type = datagram_transport<dummy_application_factory>;
buffer_type recv_buf(1024); buffer_type recv_buf(1024);
auto uri = unbox(make_uri("test:/id/42")); auto uri = unbox(make_uri("test:/id/42"));
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(
transport_type{send_socket, mpx, sys,
dummy_application_factory{ transport_type{send_socket, dummy_application_factory{shared_buf}});
shared_buf}});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>(); auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>();
CAF_CHECK(mgr_impl != nullptr); CAF_CHECK(mgr_impl != nullptr);
......
...@@ -60,15 +60,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -60,15 +60,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application { class dummy_application {
public: public:
static expected<std::vector<byte>> serialize(actor_system& sys,
const message& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
}
template <class Parent> template <class Parent>
error init(Parent&) { error init(Parent&) {
return none; return none;
...@@ -77,7 +68,11 @@ public: ...@@ -77,7 +68,11 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
parent.write_packet(msg->payload); auto payload_buf = parent.next_payload_buffer();
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
} }
template <class Parent> template <class Parent>
...@@ -116,11 +111,6 @@ class dummy_application_factory { ...@@ -116,11 +111,6 @@ class dummy_application_factory {
public: public:
using application_type = dummy_application; using application_type = dummy_application;
static expected<std::vector<byte>> serialize(actor_system& sys,
const message& x) {
return dummy_application::serialize(sys, x);
}
template <class Parent> template <class Parent>
error init(Parent&) { error init(Parent&) {
return none; return none;
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp" #include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
...@@ -61,15 +62,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -61,15 +62,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}; };
class dummy_application { class dummy_application {
public: // nop
static expected<std::vector<byte>> serialize(actor_system& sys,
const message& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
}
}; };
class dummy_transport { class dummy_transport {
...@@ -107,8 +100,9 @@ public: ...@@ -107,8 +100,9 @@ public:
template <class Manager> template <class Manager>
bool handle_write_event(Manager& mgr) { bool handle_write_event(Manager& mgr) {
for (auto x = mgr.next_message(); x != nullptr; x = mgr.next_message()) { for (auto x = mgr.next_message(); x != nullptr; x = mgr.next_message()) {
auto& payload = x->payload; binary_serializer sink{mgr.system(), buf_};
buf_.insert(buf_.end(), payload.begin(), payload.end()); if (auto err = sink(x->msg->payload))
CAF_FAIL("serializing failed: " << err);
} }
auto res = write(handle_, buf_); auto res = write(handle_, buf_);
if (auto num_bytes = get_if<size_t>(&res)) { if (auto num_bytes = get_if<size_t>(&res)) {
...@@ -128,9 +122,8 @@ public: ...@@ -128,9 +122,8 @@ public:
auto hid = string_view("0011223344556677889900112233445566778899"); auto hid = string_view("0011223344556677889900112233445566778899");
auto nid = unbox(make_node_id(42, hid)); auto nid = unbox(make_node_id(42, hid));
actor_config cfg; actor_config cfg;
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid, auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(
&mgr.system(), cfg, aid, nid, &mgr.system(), cfg, &mgr);
&mgr);
std::string path{locator.path().begin(), locator.path().end()}; std::string path{locator.path().begin(), locator.path().end()};
anon_send(listener, resolve_atom_v, std::move(path), p); anon_send(listener, resolve_atom_v, std::move(path), p);
} }
......
...@@ -92,8 +92,12 @@ public: ...@@ -92,8 +92,12 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) { std::unique_ptr<endpoint_manager_queue::message> msg) {
parent.write_packet(ptr->payload); auto payload_buf = parent.next_payload_buffer();
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
} }
template <class Parent> template <class Parent>
...@@ -110,10 +114,8 @@ public: ...@@ -110,10 +114,8 @@ public:
auto nid = unbox(make_node_id(42, hid)); auto nid = unbox(make_node_id(42, hid));
actor_config cfg; actor_config cfg;
endpoint_manager_ptr ptr{&parent.manager()}; endpoint_manager_ptr ptr{&parent.manager()};
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid, auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(
&parent.system(), aid, nid, &parent.system(), cfg, std::move(ptr));
cfg,
std::move(ptr));
anon_send(listener, resolve_atom_v, std::string{path.begin(), path.end()}, anon_send(listener, resolve_atom_v, std::string{path.begin(), path.end()},
p); p);
} }
...@@ -137,14 +139,6 @@ public: ...@@ -137,14 +139,6 @@ public:
// nop // nop
} }
static expected<buffer_type> serialize(actor_system& sys, const message& x) {
buffer_type result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
}
private: private:
buffer_ptr rec_buf_; buffer_ptr rec_buf_;
}; };
...@@ -155,10 +149,9 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) ...@@ -155,10 +149,9 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
using transport_type = stream_transport<dummy_application>; using transport_type = stream_transport<dummy_application>;
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(
transport_type{recv_socket_guard.release(), mpx, sys,
dummy_application{ transport_type{recv_socket_guard.release(), dummy_application{shared_buf}});
shared_buf}});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>(); auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>();
CAF_CHECK(mgr_impl != nullptr); CAF_CHECK(mgr_impl != nullptr);
...@@ -177,10 +170,9 @@ CAF_TEST(receive) { ...@@ -177,10 +170,9 @@ CAF_TEST(receive) {
CAF_TEST(resolve and proxy communication) { CAF_TEST(resolve and proxy communication) {
using transport_type = stream_transport<dummy_application>; using transport_type = stream_transport<dummy_application>;
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(
transport_type{send_socket_guard.release(), mpx, sys,
dummy_application{ transport_type{send_socket_guard.release(), dummy_application{shared_buf}});
shared_buf}});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
run(); run();
mgr->resolve(unbox(make_uri("test:/id/42")), self); mgr->resolve(unbox(make_uri("test:/id/42")), self);
......
...@@ -66,8 +66,8 @@ struct string_application_header { ...@@ -66,8 +66,8 @@ struct string_application_header {
/// @relates header /// @relates header
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, typename Inspector::result_type
string_application_header& hdr) { inspect(Inspector& f, string_application_header& hdr) {
return f(meta::type_name("sa_header"), hdr.payload); return f(meta::type_name("sa_header"), hdr.payload);
} }
...@@ -75,8 +75,8 @@ class string_application { ...@@ -75,8 +75,8 @@ class string_application {
public: public:
using header_type = string_application_header; using header_type = string_application_header;
string_application(actor_system& sys, std::shared_ptr<std::vector<byte>> buf) string_application(std::shared_ptr<std::vector<byte>> buf)
: sys_(sys), buf_(std::move(buf)) { : buf_(std::move(buf)) {
// nop // nop
} }
...@@ -86,8 +86,8 @@ public: ...@@ -86,8 +86,8 @@ public:
} }
template <class Parent> template <class Parent>
void handle_packet(Parent&, header_type&, span<const byte> payload) { void handle_packet(Parent& parent, header_type&, span<const byte> payload) {
binary_deserializer source{sys_, payload}; binary_deserializer source{parent.system(), payload};
message msg; message msg;
if (auto err = msg.load(source)) if (auto err = msg.load(source))
CAF_FAIL("unable to deserialize message: " << err); CAF_FAIL("unable to deserialize message: " << err);
...@@ -105,24 +105,18 @@ public: ...@@ -105,24 +105,18 @@ public:
if (ptr->msg == nullptr) if (ptr->msg == nullptr)
return; return;
auto header_buf = parent.next_header_buffer(); auto header_buf = parent.next_header_buffer();
binary_serializer sink{sys_, header_buf}; auto payload_buf = parent.next_payload_buffer();
header_type header{static_cast<uint32_t>(ptr->payload.size())}; binary_serializer payload_sink{parent.system(), payload_buf};
if (auto err = sink(header)) if (auto err = payload_sink(ptr->msg->payload))
CAF_FAIL("serializing failed: " << err); CAF_FAIL("serializing failed: " << err);
parent.write_packet(header_buf, ptr->payload); binary_serializer header_sink{parent.system(), header_buf};
} if (auto err
= header_sink(header_type{static_cast<uint32_t>(payload_buf.size())}))
static expected<std::vector<byte>> serialize(actor_system& sys, CAF_FAIL("serializing failed: " << err);
const message& x) { parent.write_packet(header_buf, payload_buf);
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
} }
private: private:
actor_system& sys_;
std::shared_ptr<std::vector<byte>> buf_; std::shared_ptr<std::vector<byte>> buf_;
}; };
...@@ -131,9 +125,8 @@ class stream_string_application : public Base { ...@@ -131,9 +125,8 @@ class stream_string_application : public Base {
public: public:
using header_type = typename Base::header_type; using header_type = typename Base::header_type;
stream_string_application(actor_system& sys, stream_string_application(std::shared_ptr<std::vector<byte>> buf)
std::shared_ptr<std::vector<byte>> buf) : Base(std::move(buf)), await_payload_(false) {
: Base(sys, std::move(buf)), await_payload_(false) {
// nop // nop
} }
...@@ -208,8 +201,8 @@ private: ...@@ -208,8 +201,8 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
using application_type = extend<string_application>::with< using application_type
stream_string_application>; = extend<string_application>::with<stream_string_application>;
using transport_type = stream_transport<application_type>; using transport_type = stream_transport<application_type>;
std::vector<byte> read_buf(1024); std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
...@@ -219,14 +212,12 @@ CAF_TEST(receive) { ...@@ -219,14 +212,12 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(read(sockets.second, read_buf), CAF_CHECK_EQUAL(read(sockets.second, read_buf),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers"); CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(mpx, sys, auto mgr1 = make_endpoint_manager(
transport_type{sockets.first, mpx, sys, transport_type{sockets.first, application_type{buf}});
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr1->init(), none); CAF_CHECK_EQUAL(mgr1->init(), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto mgr2 = make_endpoint_manager(mpx, sys, auto mgr2 = make_endpoint_manager(
transport_type{sockets.second, mpx, sys, transport_type{sockets.second, application_type{buf}});
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr2->init(), none); CAF_CHECK_EQUAL(mgr2->init(), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
CAF_MESSAGE("resolve actor-proxy"); CAF_MESSAGE("resolve actor-proxy");
......
...@@ -37,13 +37,11 @@ using namespace caf::net; ...@@ -37,13 +37,11 @@ using namespace caf::net;
namespace { namespace {
using buffer_type = std::vector<byte>;
constexpr string_view hello_test = "hello test!"; constexpr string_view hello_test = "hello test!";
struct application_result { struct application_result {
bool initialized; bool initialized;
std::vector<byte> data_buffer; byte_buffer data_buffer;
std::string resolve_path; std::string resolve_path;
actor resolve_listener; actor resolve_listener;
std::string timeout_value; std::string timeout_value;
...@@ -73,9 +71,13 @@ public: ...@@ -73,9 +71,13 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> ptr) {
auto header_buffer = parent.next_header_buffer(); auto payload_buf = parent.next_payload_buffer();
parent.write_packet(header_buffer, msg->payload); binary_serializer sink(parent.system(), payload_buf);
if (auto err = sink(ptr->msg->content()))
CAF_FAIL("serializing failed: " << err);
CAF_MESSAGE("before sending: " << CAF_ARG(ptr->msg->content()));
parent.write_packet(payload_buf);
} }
template <class Parent> template <class Parent>
...@@ -102,15 +104,6 @@ public: ...@@ -102,15 +104,6 @@ public:
res_->err = err; res_->err = err;
} }
static expected<std::vector<byte>> serialize(actor_system& sys,
const message& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
}
private: private:
std::shared_ptr<application_result> res_; std::shared_ptr<application_result> res_;
}; };
...@@ -121,12 +114,12 @@ public: ...@@ -121,12 +114,12 @@ public:
using application_type = dummy_application; using application_type = dummy_application;
dummy_transport(std::shared_ptr<transport_result> res) dummy_transport(actor_system& sys, std::shared_ptr<transport_result> res)
: res_(std::move(res)) { : sys_(sys), res_(std::move(res)) {
// nop // nop
} }
void write_packet(ip_endpoint ep, span<buffer_type*> buffers) { void write_packet(ip_endpoint ep, span<byte_buffer*> buffers) {
res_->ep = ep; res_->ep = ep;
auto& packet_buf = res_->packet_buffer; auto& packet_buf = res_->packet_buffer;
packet_buf.clear(); packet_buf.clear();
...@@ -134,19 +127,24 @@ public: ...@@ -134,19 +127,24 @@ public:
packet_buf.insert(packet_buf.end(), buf->begin(), buf->end()); packet_buf.insert(packet_buf.end(), buf->begin(), buf->end());
} }
actor_system& system() {
return sys_;
}
transport_type& transport() { transport_type& transport() {
return *this; return *this;
} }
std::vector<byte> next_header_buffer() { byte_buffer next_header_buffer() {
return {}; return {};
} }
std::vector<byte> next_payload_buffer() { byte_buffer next_payload_buffer() {
return {}; return {};
} }
private: private:
actor_system& sys_;
std::shared_ptr<transport_result> res_; std::shared_ptr<transport_result> res_;
}; };
...@@ -156,7 +154,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -156,7 +154,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() fixture()
: transport_results{std::make_shared<transport_result>()}, : transport_results{std::make_shared<transport_result>()},
application_results{std::make_shared<application_result>()}, application_results{std::make_shared<application_result>()},
transport(transport_results), transport(sys, transport_results),
worker{dummy_application{application_results}} { worker{dummy_application{application_results}} {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init()) if (auto err = mpx->init())
...@@ -196,20 +194,23 @@ CAF_TEST(handle_data) { ...@@ -196,20 +194,23 @@ CAF_TEST(handle_data) {
} }
CAF_TEST(write_message) { CAF_TEST(write_message) {
std::string hello_test{"hello world!"};
actor act; actor act;
auto strong_actor = actor_cast<strong_actor_ptr>(act); auto strong_actor = actor_cast<strong_actor_ptr>(act);
mailbox_element::forwarding_stack stack; mailbox_element::forwarding_stack stack;
auto msg = make_message(); auto msg = make_message(hello_test);
auto elem = make_mailbox_element(strong_actor, make_message_id(12345), stack, auto elem = make_mailbox_element(strong_actor, make_message_id(12345), stack,
msg); msg);
auto test_span = as_bytes(make_span(hello_test));
std::vector<byte> payload(test_span.begin(), test_span.end());
using message_type = endpoint_manager_queue::message; using message_type = endpoint_manager_queue::message;
auto message = detail::make_unique<message_type>(std::move(elem), nullptr, auto message = detail::make_unique<message_type>(std::move(elem), nullptr);
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()}; binary_deserializer source{sys, buf};
caf::message received_msg;
CAF_CHECK(!source(received_msg));
CAF_MESSAGE(CAF_ARG(received_msg));
auto received_str = received_msg.get_as<std::string>(0);
string_view result{received_str};
CAF_CHECK_EQUAL(result, hello_test); CAF_CHECK_EQUAL(result, hello_test);
CAF_CHECK_EQUAL(transport_results->ep, ep); CAF_CHECK_EQUAL(transport_results->ep, ep);
} }
......
...@@ -66,10 +66,10 @@ public: ...@@ -66,10 +66,10 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> ptr) {
rec_buf_->push_back(static_cast<byte>(id_)); rec_buf_->push_back(static_cast<byte>(id_));
auto header_buf = parent.next_header_buffer(); auto data = ptr->msg->content().get_as<std::vector<byte>>(0);
parent.write_packet(header_buf, msg->payload); parent.write_packet(data);
} }
template <class Parent> template <class Parent>
...@@ -92,10 +92,6 @@ public: ...@@ -92,10 +92,6 @@ public:
rec_buf_->push_back(static_cast<byte>(id_)); rec_buf_->push_back(static_cast<byte>(id_));
} }
static expected<buffer_type> serialize(actor_system&, const message&) {
return buffer_type{};
}
private: private:
std::shared_ptr<buffer_type> rec_buf_; std::shared_ptr<buffer_type> rec_buf_;
uint8_t id_; uint8_t id_;
...@@ -126,7 +122,8 @@ struct dummy_transport { ...@@ -126,7 +122,8 @@ struct dummy_transport {
using application_type = dummy_application; using application_type = dummy_application;
dummy_transport(std::shared_ptr<buffer_type> buf) : buf_(std::move(buf)) { dummy_transport(actor_system& sys, std::shared_ptr<buffer_type> buf)
: sys_(sys), buf_(std::move(buf)) {
// nop // nop
} }
...@@ -136,6 +133,10 @@ struct dummy_transport { ...@@ -136,6 +133,10 @@ struct dummy_transport {
buf_->insert(buf_->end(), buf->begin(), buf->end()); buf_->insert(buf_->end(), buf->begin(), buf->end());
} }
actor_system& system() {
return sys_;
}
transport_type& transport() { transport_type& transport() {
return *this; return *this;
} }
...@@ -149,6 +150,7 @@ struct dummy_transport { ...@@ -149,6 +150,7 @@ struct dummy_transport {
} }
private: private:
actor_system& sys_;
std::shared_ptr<buffer_type> buf_; std::shared_ptr<buffer_type> buf_;
}; };
...@@ -182,32 +184,31 @@ uri operator"" _u(const char* cstr, size_t cstr_len) { ...@@ -182,32 +184,31 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
} }
struct fixture : host_fixture { struct fixture : host_fixture {
using dispatcher_type = transport_worker_dispatcher<dummy_application_factory, using dispatcher_type
ip_endpoint>; = transport_worker_dispatcher<dummy_application_factory, ip_endpoint>;
fixture() fixture()
: buf{std::make_shared<buffer_type>()}, : buf{std::make_shared<buffer_type>()},
dispatcher{dummy_application_factory{buf}}, dispatcher{dummy_application_factory{buf}},
dummy{buf} { dummy{sys, buf} {
add_new_workers(); add_new_workers();
} }
std::unique_ptr<net::endpoint_manager_queue::message> std::unique_ptr<net::endpoint_manager_queue::message>
make_dummy_message(node_id nid) { make_dummy_message(node_id nid) {
actor_id aid = 42; actor_id aid = 42;
auto test_span = as_bytes(make_span(hello_test));
byte_buffer payload(test_span.begin(), test_span.end());
actor_config cfg; actor_config cfg;
auto p = make_actor<dummy_actor, strong_actor_ptr>(aid, nid, &sys, cfg); auto p = make_actor<dummy_actor, strong_actor_ptr>(aid, nid, &sys, cfg);
auto test_span = as_bytes(make_span(hello_test));
buffer_type payload(test_span.begin(), test_span.end());
auto receiver = actor_cast<strong_actor_ptr>(p); auto receiver = actor_cast<strong_actor_ptr>(p);
if (!receiver) if (!receiver)
CAF_FAIL("failed to cast receiver to a strong_actor_ptr"); CAF_FAIL("failed to cast receiver to a strong_actor_ptr");
mailbox_element::forwarding_stack stack; mailbox_element::forwarding_stack stack;
auto elem = make_mailbox_element(nullptr, make_message_id(12345), auto elem = make_mailbox_element(nullptr, make_message_id(12345),
std::move(stack), make_message()); std::move(stack), make_message(payload));
return detail::make_unique<endpoint_manager_queue::message>(std::move(elem), return detail::make_unique<endpoint_manager_queue::message>(std::move(elem),
receiver, receiver);
payload);
} }
bool contains(byte x) { bool contains(byte x) {
...@@ -227,6 +228,7 @@ struct fixture : host_fixture { ...@@ -227,6 +228,7 @@ struct fixture : host_fixture {
auto msg = make_dummy_message(testcase.nid); auto msg = make_dummy_message(testcase.nid);
if (!msg->receiver) if (!msg->receiver)
CAF_FAIL("receiver is null"); CAF_FAIL("receiver is null");
CAF_MESSAGE(CAF_ARG(msg));
dispatcher.write_message(dummy, std::move(msg)); dispatcher.write_message(dummy, std::move(msg));
} }
...@@ -255,9 +257,8 @@ struct fixture : host_fixture { ...@@ -255,9 +257,8 @@ struct fixture : host_fixture {
test_write_message(testcase); \ test_write_message(testcase); \
CAF_CHECK_EQUAL(buf->size(), hello_test.size() + 1u); \ CAF_CHECK_EQUAL(buf->size(), hello_test.size() + 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \ CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
CAF_CHECK_EQUAL(memcmp(buf->data() + 1, hello_test.data(), \ CAF_CHECK_EQUAL( \
hello_test.size()), \ memcmp(buf->data() + 1, hello_test.data(), hello_test.size()), 0); \
0); \
buf->clear(); buf->clear();
#define CHECK_TIMEOUT(testcase) \ #define CHECK_TIMEOUT(testcase) \
......
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