Commit aa28e12b authored by Dominik Charousset's avatar Dominik Charousset

Add receiver pointer to endpoint_manager::message

parent 34e9e924
...@@ -76,13 +76,29 @@ public: ...@@ -76,13 +76,29 @@ public:
} }
template <class Parent> template <class Parent>
void write_message(Parent& parent, error write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> ptr) { std::unique_ptr<endpoint_manager::message> ptr) {
header hdr{message_type::actor_message, // TODO: avoid extra copy of the payload
static_cast<uint32_t>(ptr->payload.size()), 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()}; ptr->msg->mid.integer_value()};
auto bytes = to_bytes(hdr); 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> template <class Parent>
......
...@@ -91,10 +91,14 @@ public: ...@@ -91,10 +91,14 @@ public:
/// Original message to a remote actor. /// Original message to a remote actor.
mailbox_element_ptr msg; mailbox_element_ptr msg;
/// ID of the receiving actor.
strong_actor_ptr receiver;
/// Serialized representation of of `msg->content()`. /// Serialized representation of of `msg->content()`.
std::vector<byte> payload; 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 { struct message_policy {
...@@ -136,7 +140,8 @@ public: ...@@ -136,7 +140,8 @@ public:
void resolve(std::string path, actor listener); void resolve(std::string path, actor listener);
/// Enqueues a message to the endpoint. /// 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. /// Enqueues a timeout to the endpoint.
void enqueue(timeout_msg msg); void enqueue(timeout_msg msg);
......
...@@ -27,18 +27,18 @@ namespace net { ...@@ -27,18 +27,18 @@ namespace 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), sf_(dst->serialize_fun()), dst_(std::move(dst)) {
// anon_send(broker_, monitor_atom::value, ctrl()); // nop
} }
actor_proxy_impl::~actor_proxy_impl() { 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*) { void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
CAF_PUSH_AID(0); CAF_PUSH_AID(0);
CAF_ASSERT(what != nullptr); CAF_ASSERT(what != nullptr);
if (auto payload = sf_(home_system(), what->content())) 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 else
CAF_LOG_ERROR( CAF_LOG_ERROR(
"unable to serialize payload: " << home_system().render(payload.error())); "unable to serialize payload: " << home_system().render(payload.error()));
......
...@@ -37,8 +37,11 @@ endpoint_manager::event::event(atom_value type, uint64_t id) ...@@ -37,8 +37,11 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
} }
endpoint_manager::message::message(mailbox_element_ptr msg, endpoint_manager::message::message(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) 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 // nop
} }
...@@ -87,8 +90,10 @@ void endpoint_manager::resolve(std::string path, actor listener) { ...@@ -87,8 +90,10 @@ void endpoint_manager::resolve(std::string path, actor listener) {
} }
void endpoint_manager::enqueue(mailbox_element_ptr msg, void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) { 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) if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write); mask_add(operation::write);
} }
......
...@@ -192,6 +192,7 @@ CAF_TEST(write_message) { ...@@ -192,6 +192,7 @@ CAF_TEST(write_message) {
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 = detail::make_unique<endpoint_manager::message>(std::move(elem), auto message = detail::make_unique<endpoint_manager::message>(std::move(elem),
nullptr,
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;
......
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