Commit f5991184 authored by Dominik Charousset's avatar Dominik Charousset

Add receiver pointer to endpoint_manager::message

parent 51f336ad
......@@ -76,13 +76,29 @@ public:
}
template <class Parent>
void write_message(Parent& parent,
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> ptr) {
header hdr{message_type::actor_message,
static_cast<uint32_t>(ptr->payload.size()),
// TODO: avoid extra copy of the payload
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
// TODO: valid?
return none;
}
if (src != nullptr) {
if (auto err = sink(src->node(), src->id(), dst->id(), ptr->msg->stages))
return err;
} else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
buf_.insert(buf_.end(), ptr->payload.begin(), ptr->payload.end());
header hdr{message_type::actor_message, static_cast<uint32_t>(buf_.size()),
ptr->msg->mid.integer_value()};
auto bytes = to_bytes(hdr);
parent.write_packet(parent, make_span(bytes), ptr->payload, unit);
parent.write_packet(parent, make_span(bytes), make_span(buf_), unit);
}
template <class Parent>
......
......@@ -91,10 +91,14 @@ public:
/// Original message to a remote actor.
mailbox_element_ptr msg;
/// ID of the receiving actor.
strong_actor_ptr receiver;
/// Serialized representation of of `msg->content()`.
std::vector<byte> payload;
message(mailbox_element_ptr msg, std::vector<byte> payload);
message(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
};
struct message_policy {
......@@ -136,7 +140,8 @@ public:
void resolve(std::string path, actor listener);
/// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<byte> payload);
void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
/// Enqueues a timeout to the endpoint.
void enqueue(timeout_msg msg);
......
......@@ -27,18 +27,18 @@ namespace net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) {
// anon_send(broker_, monitor_atom::value, ctrl());
// nop
}
actor_proxy_impl::~actor_proxy_impl() {
// anon_send(broker_, make_message(delete_atom::value, node(), id()));
// nop
}
void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
CAF_PUSH_AID(0);
CAF_ASSERT(what != nullptr);
if (auto payload = sf_(home_system(), what->content()))
dst_->enqueue(std::move(what), std::move(*payload));
dst_->enqueue(std::move(what), ctrl(), std::move(*payload));
else
CAF_LOG_ERROR(
"unable to serialize payload: " << home_system().render(payload.error()));
......
......@@ -37,8 +37,11 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
}
endpoint_manager::message::message(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload)
: msg(std::move(msg)), payload(std::move(payload)) {
: msg(std::move(msg)),
receiver(std::move(receiver)),
payload(std::move(payload)) {
// nop
}
......@@ -87,8 +90,10 @@ void endpoint_manager::resolve(std::string path, actor listener) {
}
void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) {
auto ptr = new message(std::move(msg), std::move(payload));
auto ptr = new message(std::move(msg), std::move(receiver),
std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write);
}
......
......@@ -192,6 +192,7 @@ CAF_TEST(write_message) {
hello_test.size());
std::vector<byte> payload(test_span.begin(), test_span.end());
auto message = detail::make_unique<endpoint_manager::message>(std::move(elem),
nullptr,
payload);
worker.write_message(transport, std::move(message));
auto& buf = transport_results->packet_buffer;
......
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