Commit eebfb6de authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'ufownl-topic/group-down-msg' into develop

parents db05a3f6 2f4e4bb6
...@@ -232,6 +232,7 @@ class local_group_proxy : public local_group { ...@@ -232,6 +232,7 @@ class local_group_proxy : public local_group {
CAF_ASSERT(remote_broker != invalid_actor); CAF_ASSERT(remote_broker != invalid_actor);
m_broker = std::move(remote_broker); m_broker = std::move(remote_broker);
m_proxy_broker = spawn<proxy_broker, hidden>(this); m_proxy_broker = spawn<proxy_broker, hidden>(this);
m_monitor = spawn(broker_monitor_actor, this);
} }
attachable_ptr subscribe(const actor_addr& who) override { attachable_ptr subscribe(const actor_addr& who) override {
...@@ -268,13 +269,28 @@ class local_group_proxy : public local_group { ...@@ -268,13 +269,28 @@ class local_group_proxy : public local_group {
void stop() override { void stop() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
await_all_locals_down({m_proxy_broker, m_broker}); await_all_locals_down({m_monitor, m_proxy_broker, m_broker});
m_monitor = invalid_actor;
m_proxy_broker = invalid_actor; m_proxy_broker = invalid_actor;
m_broker = invalid_actor; m_broker = invalid_actor;
} }
private: private:
static behavior broker_monitor_actor(event_based_actor* self,
local_group_proxy* grp) {
self->monitor(grp->m_broker);
return {
[=](const down_msg& down) {
auto msg = make_message(group_down_msg{group(grp)});
grp->send_all_subscribers(self->address(), std::move(msg),
self->host());
self->quit(down.reason);
}
};
}
actor m_proxy_broker; actor m_proxy_broker;
actor m_monitor;
}; };
using local_group_proxy_ptr = intrusive_ptr<local_group_proxy>; using local_group_proxy_ptr = intrusive_ptr<local_group_proxy>;
......
...@@ -501,7 +501,7 @@ CAF_TEST(test_remote_actor) { ...@@ -501,7 +501,7 @@ CAF_TEST(test_remote_actor) {
CAF_CHECK(serv2 == io::remote_actor("127.0.0.1", p2)); CAF_CHECK(serv2 == io::remote_actor("127.0.0.1", p2));
} }
// connect to published groups // connect to published groups
io::remote_group("whatever", "127.0.0.1", gport); auto grp = io::remote_group("whatever", "127.0.0.1", gport);
auto c = self->spawn<client, monitored>(serv); auto c = self->spawn<client, monitored>(serv);
self->receive( self->receive(
[&](const down_msg& dm) { [&](const down_msg& dm) {
...@@ -509,6 +509,7 @@ CAF_TEST(test_remote_actor) { ...@@ -509,6 +509,7 @@ CAF_TEST(test_remote_actor) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal); CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
} }
); );
grp->stop();
}, },
on("-s") >> [&] { on("-s") >> [&] {
CAF_MESSAGE("don't run remote actor (server mode)"); CAF_MESSAGE("don't run remote actor (server mode)");
......
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