Commit f2df192f authored by Jakob Otto's avatar Jakob Otto

Integrate review feedback

parent b068df71
......@@ -37,7 +37,6 @@ public:
void kill_proxy(execution_unit* ctx, error rsn) override;
private:
endpoint_manager::serialize_fun_type sf_;
endpoint_manager_ptr dst_;
};
......
......@@ -136,9 +136,6 @@ public:
// nop
}
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf);
// -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path);
......
......@@ -44,10 +44,6 @@ public:
using super = socket_manager;
/// A function type for serializing message payloads.
using serialize_fun_type = error_code<sec> (*)(actor_system&, const message&,
std::vector<byte>& buf);
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager(socket handle, const multiplexer_ptr& parent,
......@@ -82,9 +78,6 @@ public:
/// Initializes the manager before adding it to the multiplexer's event loop.
virtual error init() = 0;
/// @returns the protocol-specific function for serializing payloads.
virtual serialize_fun_type serialize_fun() const noexcept = 0;
protected:
bool enqueue(endpoint_manager_queue::element* ptr);
......
......@@ -62,8 +62,8 @@ public:
// -- timeout management -----------------------------------------------------
template <class... Ts>
uint64_t set_timeout(actor_clock::time_point tp, std::string type,
Ts&&... xs) {
uint64_t
set_timeout(actor_clock::time_point tp, std::string type, Ts&&... xs) {
auto act = actor_cast<abstract_actor*>(timeout_proxy_);
CAF_ASSERT(act != nullptr);
sys_.clock().set_multi_timeout(tp, act, std::move(type), next_timeout_id_);
......@@ -119,10 +119,6 @@ public:
transport_.handle_error(code);
}
serialize_fun_type serialize_fun() const noexcept override {
return application_type::serialize;
}
private:
transport_type transport_;
......
......@@ -112,7 +112,7 @@ public:
// nop
}
task_size_type task_size(const event&) const noexcept {
static constexpr task_size_type task_size(const event&) noexcept {
return 1;
}
};
......@@ -149,10 +149,8 @@ public:
// nop
}
static task_size_type task_size(const message&) noexcept {
// Return at least 1 if the payload is empty.
return static_cast<task_size_type>(1);
// was x.payload.size() + static_cast<task_size_type>(x.payload.empty());
static constexpr task_size_type task_size(const message&) noexcept {
return 1;
}
};
......
......@@ -25,7 +25,7 @@
namespace caf::net {
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);
dst_->enqueue_event(node(), id());
}
......
......@@ -52,8 +52,6 @@ error application::write_message(
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
auto payload_prefix_buf = writer.next_payload_buffer();
binary_serializer sink{system(), payload_prefix_buf};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
......@@ -61,8 +59,7 @@ error application::write_message(
return none;
}
auto payload_buf = writer.next_payload_buffer();
if (auto err = serialize(system(), ptr->msg->content(), payload_buf))
return err;
binary_serializer sink{system(), payload_buf};
if (src != nullptr) {
auto src_id = src->id();
system().registry().put(src_id, src);
......@@ -72,13 +69,14 @@ error application::write_message(
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
if (auto err = sink(ptr->msg->content()))
return err;
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_prefix_buf.size()
+ payload_buf.size()),
static_cast<uint32_t>(payload_buf.size()),
ptr->msg->mid.integer_value()},
hdr);
writer.write_packet(hdr, payload_prefix_buf, payload_buf);
writer.write_packet(hdr, payload_buf);
return none;
}
......@@ -121,12 +119,6 @@ void application::local_actor_down(packet_writer& writer, actor_id id,
writer.write_packet(hdr, payload);
}
error_code<sec> application::serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
strong_actor_ptr application::resolve_local_path(string_view path) {
CAF_LOG_TRACE(CAF_ARG(path));
// We currently support two path formats: `id/<actor_id>` and `name/<atom>`.
......
......@@ -123,7 +123,8 @@ public:
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto payload_buf = parent.next_payload_buffer();
if (auto err = serialize(parent.system(), msg->msg->payload, payload_buf))
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
}
......@@ -167,12 +168,6 @@ public:
CAF_FAIL("handle_error called: " << to_string(sec));
}
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
private:
buffer_ptr rec_buf_;
};
......
......@@ -60,12 +60,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application {
public:
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
template <class Parent>
error init(Parent&) {
return none;
......@@ -75,7 +69,8 @@ public:
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto payload_buf = parent.next_payload_buffer();
if (auto err = serialize(parent.system(), msg->msg->payload, payload_buf))
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
}
......@@ -116,11 +111,6 @@ class dummy_application_factory {
public:
using application_type = dummy_application;
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
return dummy_application::serialize(sys, x, buf);
}
template <class Parent>
error init(Parent&) {
return none;
......
......@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
......@@ -61,12 +62,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
};
class dummy_application {
public:
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
// nop
};
class dummy_transport {
......@@ -103,12 +99,10 @@ public:
template <class Manager>
bool handle_write_event(Manager& mgr) {
auto sf = mgr.serialize_fun();
for (auto x = mgr.next_message(); x != nullptr; x = mgr.next_message()) {
std::vector<byte> payload_buf;
if (auto err = sf(mgr.system(), x->msg->payload, payload_buf))
binary_serializer sink{mgr.system(), buf_};
if (auto err = sink(x->msg->payload))
CAF_FAIL("serializing failed: " << err);
buf_.insert(buf_.end(), payload_buf.begin(), payload_buf.end());
}
auto res = write(handle_, buf_);
if (auto num_bytes = get_if<size_t>(&res)) {
......
......@@ -94,7 +94,8 @@ public:
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto payload_buf = parent.next_payload_buffer();
if (auto err = serialize(parent.system(), msg->msg->payload, payload_buf))
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
}
......@@ -138,12 +139,6 @@ public:
// nop
}
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
private:
buffer_ptr rec_buf_;
};
......
......@@ -75,8 +75,8 @@ class string_application {
public:
using header_type = string_application_header;
string_application(actor_system& sys, std::shared_ptr<std::vector<byte>> buf)
: sys_(sys), buf_(std::move(buf)) {
string_application(std::shared_ptr<std::vector<byte>> buf)
: buf_(std::move(buf)) {
// nop
}
......@@ -86,8 +86,8 @@ public:
}
template <class Parent>
void handle_packet(Parent&, header_type&, span<const byte> payload) {
binary_deserializer source{sys_, payload};
void handle_packet(Parent& parent, header_type&, span<const byte> payload) {
binary_deserializer source{parent.system(), payload};
message msg;
if (auto err = msg.load(source))
CAF_FAIL("unable to deserialize message: " << err);
......@@ -106,23 +106,17 @@ public:
return;
auto header_buf = parent.next_header_buffer();
auto payload_buf = parent.next_payload_buffer();
if (auto err = serialize(parent.system(), ptr->msg->payload, payload_buf))
binary_serializer payload_sink{parent.system(), payload_buf};
if (auto err = payload_sink(ptr->msg->payload))
CAF_FAIL("serializing failed: " << err);
binary_serializer sink{sys_, header_buf};
header_type header{static_cast<uint32_t>(payload_buf.size())};
if (auto err = sink(header))
binary_serializer header_sink{parent.system(), header_buf};
if (auto err
= header_sink(header_type{static_cast<uint32_t>(payload_buf.size())}))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(header_buf, payload_buf);
}
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
private:
actor_system& sys_;
std::shared_ptr<std::vector<byte>> buf_;
};
......@@ -131,9 +125,8 @@ class stream_string_application : public Base {
public:
using header_type = typename Base::header_type;
stream_string_application(actor_system& sys,
std::shared_ptr<std::vector<byte>> buf)
: Base(sys, std::move(buf)), await_payload_(false) {
stream_string_application(std::shared_ptr<std::vector<byte>> buf)
: Base(std::move(buf)), await_payload_(false) {
// nop
}
......@@ -220,11 +213,11 @@ CAF_TEST(receive) {
sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(
mpx, sys, transport_type{sockets.first, application_type{sys, buf}});
mpx, sys, transport_type{sockets.first, application_type{buf}});
CAF_CHECK_EQUAL(mgr1->init(), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto mgr2 = make_endpoint_manager(
mpx, sys, transport_type{sockets.second, application_type{sys, buf}});
mpx, sys, transport_type{sockets.second, application_type{buf}});
CAF_CHECK_EQUAL(mgr2->init(), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
CAF_MESSAGE("resolve actor-proxy");
......
......@@ -37,13 +37,11 @@ using namespace caf::net;
namespace {
using buffer_type = std::vector<byte>;
constexpr string_view hello_test = "hello test!";
struct application_result {
bool initialized;
std::vector<byte> data_buffer;
byte_buffer data_buffer;
std::string resolve_path;
actor resolve_listener;
std::string timeout_value;
......@@ -74,10 +72,12 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
if (auto payload = serialize(parent.system(), ptr->msg->payload))
parent.write_packet(*payload);
else
CAF_FAIL("serializing failed: " << payload.error());
auto payload_buf = parent.next_payload_buffer();
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>
......@@ -104,15 +104,6 @@ public:
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:
std::shared_ptr<application_result> res_;
};
......@@ -128,7 +119,7 @@ public:
// nop
}
void write_packet(ip_endpoint ep, span<buffer_type*> buffers) {
void write_packet(ip_endpoint ep, span<byte_buffer*> buffers) {
res_->ep = ep;
auto& packet_buf = res_->packet_buffer;
packet_buf.clear();
......@@ -144,11 +135,11 @@ public:
return *this;
}
std::vector<byte> next_header_buffer() {
byte_buffer next_header_buffer() {
return {};
}
std::vector<byte> next_payload_buffer() {
byte_buffer next_payload_buffer() {
return {};
}
......@@ -202,25 +193,27 @@ CAF_TEST(handle_data) {
CAF_CHECK_EQUAL(result, hello_test);
}
/* TODO: Removed payload, send data in form of message
CAF_TEST(write_message) {
CAF_TEST(write_message) {
std::string hello_test{"hello world!"};
actor act;
auto strong_actor = actor_cast<strong_actor_ptr>(act);
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,
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;
auto message = detail::make_unique<message_type>(std::move(elem), nullptr,
payload);
auto message = detail::make_unique<message_type>(std::move(elem), nullptr);
worker.write_message(transport, std::move(message));
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(transport_results->ep, ep);
}*/
}
CAF_TEST(resolve) {
worker.resolve(transport, "foo", self);
......
......@@ -64,14 +64,12 @@ public:
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager_queue::message> msg) {
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
rec_buf_->push_back(static_cast<byte>(id_));
if (auto payload = serialize(transport.system(), msg->msg->payload))
transport.write_packet(*payload);
else
CAF_FAIL("serializing failed: " << payload.error());
auto data = ptr->msg->content().get_as<std::vector<byte>>(0);
parent.write_packet(data);
}
template <class Parent>
......@@ -94,10 +92,6 @@ public:
rec_buf_->push_back(static_cast<byte>(id_));
}
static expected<buffer_type> serialize(actor_system&, const message&) {
return buffer_type{};
}
private:
std::shared_ptr<buffer_type> rec_buf_;
uint8_t id_;
......@@ -203,16 +197,16 @@ struct fixture : host_fixture {
std::unique_ptr<net::endpoint_manager_queue::message>
make_dummy_message(node_id nid) {
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;
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);
if (!receiver)
CAF_FAIL("failed to cast receiver to a strong_actor_ptr");
mailbox_element::forwarding_stack stack;
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),
receiver);
}
......@@ -234,6 +228,7 @@ struct fixture : host_fixture {
auto msg = make_dummy_message(testcase.nid);
if (!msg->receiver)
CAF_FAIL("receiver is null");
CAF_MESSAGE(CAF_ARG(msg));
dispatcher.write_message(dummy, std::move(msg));
}
......@@ -289,13 +284,12 @@ CAF_TEST(handle_data) {
CHECK_HANDLE_DATA(test_data.at(3));
}
/*
CAF_TEST(write_message write_packet) {
CAF_TEST(write_message write_packet) {
CHECK_WRITE_MESSAGE(test_data.at(0));
CHECK_WRITE_MESSAGE(test_data.at(1));
CHECK_WRITE_MESSAGE(test_data.at(2));
CHECK_WRITE_MESSAGE(test_data.at(3));
}*/
}
CAF_TEST(resolve) {
// TODO think of a test for this
......
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