Commit 6e9f806d authored by Dominik Charousset's avatar Dominik Charousset

Remove messaging side effect from make_proxy

parent 822dc4d0
...@@ -149,6 +149,12 @@ using link_atom = atom_constant<atom("link")>; ...@@ -149,6 +149,12 @@ using link_atom = atom_constant<atom("link")>;
/// Used for removing networked links. /// Used for removing networked links.
using unlink_atom = atom_constant<atom("unlink")>; using unlink_atom = atom_constant<atom("unlink")>;
/// Used for monitor requests over network.
using monitor_atom = atom_constant<atom("monitor")>;
/// Used for removing networked monitors.
using demonitor_atom = atom_constant<atom("demonitor")>;
/// Used for publishing actors at a given port. /// Used for publishing actors at a given port.
using publish_atom = atom_constant<atom("publish")>; using publish_atom = atom_constant<atom("publish")>;
......
...@@ -30,7 +30,7 @@ namespace caf { ...@@ -30,7 +30,7 @@ namespace caf {
forwarding_actor_proxy::forwarding_actor_proxy(actor_config& cfg, actor dest) forwarding_actor_proxy::forwarding_actor_proxy(actor_config& cfg, actor dest)
: actor_proxy(cfg), : actor_proxy(cfg),
broker_(std::move(dest)) { broker_(std::move(dest)) {
// nop anon_send(broker_, monitor_atom::value, ctrl());
} }
forwarding_actor_proxy::~forwarding_actor_proxy() { forwarding_actor_proxy::~forwarding_actor_proxy() {
......
...@@ -89,15 +89,6 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -89,15 +89,6 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
bptr->state.proxies().erase(nid, res->id(), rsn); bptr->state.proxies().erase(nid, res->id(), rsn);
}); });
}); });
CAF_LOG_DEBUG("successfully created proxy instance, "
"write announce_proxy_instance:"
<< CAF_ARG(nid) << CAF_ARG(aid)
<< CAF_ARG2("hdl", this_context->hdl));
// tell remote side we are monitoring this actor now
instance.write_monitor_message(self->context(), get_buffer(this_context->hdl),
nid, aid);
flush(this_context->hdl);
mm->notify<hook::new_remote_actor>(res);
return res; return res;
} }
...@@ -509,6 +500,27 @@ behavior basp_broker::make_behavior() { ...@@ -509,6 +500,27 @@ behavior basp_broker::make_behavior() {
} }
return delegated<message>(); return delegated<message>();
}, },
// received from proxy instances to signal that we need to send a BASP
// monitor_message to the origin node
[=](monitor_atom, const strong_actor_ptr& proxy) {
if (proxy == nullptr) {
CAF_LOG_WARNING("received a monitor message from an invalid proxy");
return;
}
auto route = state.instance.tbl().lookup(proxy->node());
if (route == none) {
CAF_LOG_DEBUG("connection to origin already lost, kill proxy");
state.instance.proxies().erase(proxy->node(), proxy->id());
return;
}
CAF_LOG_DEBUG("write monitor_message:" << CAF_ARG(proxy));
// tell remote side we are monitoring this actor now
auto hdl = route->hdl;
state.instance.write_monitor_message(context(), state.get_buffer(hdl),
proxy->node(), proxy->id());
flush(hdl);
system().middleman().notify<hook::new_remote_actor>(proxy);
},
// received from underlying broker implementation // received from underlying broker implementation
[=](const new_connection_msg& msg) { [=](const new_connection_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
......
...@@ -313,7 +313,7 @@ public: ...@@ -313,7 +313,7 @@ public:
buffer buf; buffer buf;
std::tie(hdr, buf) = read_from_out_buf(hdl); std::tie(hdr, buf) = read_from_out_buf(hdl);
CAF_MESSAGE("dispatch output buffer for connection " << hdl.id()); CAF_MESSAGE("dispatch output buffer for connection " << hdl.id());
CAF_REQUIRE(hdr.operation == basp::message_type::direct_message); CAF_REQUIRE_EQUAL(hdr.operation, basp::message_type::direct_message);
binary_deserializer source{mpx_, buf}; binary_deserializer source{mpx_, buf};
std::vector<strong_actor_ptr> stages; std::vector<strong_actor_ptr> stages;
message msg; message msg;
...@@ -354,8 +354,9 @@ public: ...@@ -354,8 +354,9 @@ public:
buffer buf; buffer buf;
this_->to_payload(buf, xs...); this_->to_payload(buf, xs...);
buffer& ob = this_->mpx()->output_buffer(hdl); buffer& ob = this_->mpx()->output_buffer(hdl);
while (ob.size() < basp::header_size) while (this_->mpx()->try_exec_runnable()) {
this_->mpx()->exec_runnable(); // repeat
}
CAF_MESSAGE("output buffer has " << ob.size() << " bytes"); CAF_MESSAGE("output buffer has " << ob.size() << " bytes");
basp::header hdr; basp::header hdr;
{ // lifetime scope of source { // lifetime scope of source
...@@ -376,12 +377,6 @@ public: ...@@ -376,12 +377,6 @@ public:
ob.erase(ob.begin(), ob.begin() + basp::header_size); ob.erase(ob.begin(), ob.begin() + basp::header_size);
} }
CAF_CHECK_EQUAL(operation, hdr.operation); CAF_CHECK_EQUAL(operation, hdr.operation);
if (hdr.operation == basp::message_type::direct_message) {
binary_deserializer source{this_->mpx(), payload};
std::vector<strong_actor_ptr> fwd_stack;
message msg;
source(fwd_stack, msg);
}
CAF_CHECK_EQUAL(flags, static_cast<uint8_t>(hdr.flags)); CAF_CHECK_EQUAL(flags, static_cast<uint8_t>(hdr.flags));
CAF_CHECK_EQUAL(payload_len, hdr.payload_len); CAF_CHECK_EQUAL(payload_len, hdr.payload_len);
CAF_CHECK_EQUAL(operation_data, hdr.operation_data); CAF_CHECK_EQUAL(operation_data, hdr.operation_data);
......
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