Commit 8cd93e91 authored by Dominik Charousset's avatar Dominik Charousset

Avoid using current_message()

parent ac1ce5f2
...@@ -153,9 +153,8 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -153,9 +153,8 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
// send composed message to our buddy // send composed message to our buddy
self->send(buddy, atm, ival); self->send(buddy, atm, ival);
}, },
others >> [=] { others >> [=](const message& msg) {
aout(self) << "unexpected: " aout(self) << "unexpected: " << to_string(msg) << endl;
<< to_string(self->current_message()) << endl;
} }
}; };
} }
...@@ -172,9 +171,8 @@ behavior server(broker* self, const actor& buddy) { ...@@ -172,9 +171,8 @@ behavior server(broker* self, const actor& buddy) {
aout(self) << "quit server (only accept 1 connection)" << endl; aout(self) << "quit server (only accept 1 connection)" << endl;
self->quit(); self->quit();
}, },
others >> [=] { others >> [=](const message& msg) {
aout(self) << "unexpected: " aout(self) << "unexpected: " << to_string(msg) << endl;
<< to_string(self->current_message()) << endl;
} }
}; };
} }
......
...@@ -61,8 +61,8 @@ behavior server(broker* self) { ...@@ -61,8 +61,8 @@ behavior server(broker* self) {
*counter = 0; *counter = 0;
self->delayed_send(self, std::chrono::seconds(1), tick_atom::value); self->delayed_send(self, std::chrono::seconds(1), tick_atom::value);
}, },
others >> [=] { others >> [=](const message& msg) {
aout(self) << "unexpected: " << to_string(self->current_message()) << endl; aout(self) << "unexpected: " << to_string(msg) << endl;
} }
}; };
} }
......
...@@ -37,8 +37,8 @@ behavior calculator_fun(event_based_actor* self) { ...@@ -37,8 +37,8 @@ behavior calculator_fun(event_based_actor* self) {
[](sub_atom, int a, int b) { [](sub_atom, int a, int b) {
return a - b; return a - b;
}, },
others >> [=] { others >> [=](const message& msg) {
aout(self) << "received: " << to_string(self->current_message()) << endl; aout(self) << "received: " << to_string(msg) << endl;
} }
}; };
} }
...@@ -52,8 +52,8 @@ void blocking_calculator_fun(blocking_actor* self) { ...@@ -52,8 +52,8 @@ void blocking_calculator_fun(blocking_actor* self) {
[](sub_atom, int a, int b) { [](sub_atom, int a, int b) {
return a - b; return a - b;
}, },
others >> [=] { others >> [=](const message& msg) {
aout(self) << "received: " << to_string(self->current_message()) << endl; aout(self) << "received: " << to_string(msg) << endl;
} }
); );
} }
......
...@@ -58,8 +58,8 @@ void client(event_based_actor* self, const string& name) { ...@@ -58,8 +58,8 @@ void client(event_based_actor* self, const string& name) {
[=](const group_down_msg& g) { [=](const group_down_msg& g) {
cout << "*** chatroom offline: " << to_string(g.source) << endl; cout << "*** chatroom offline: " << to_string(g.source) << endl;
}, },
others >> [=]() { others >> [=](const message& msg) {
cout << "unexpected: " << to_string(self->current_message()) << endl; cout << "unexpected: " << to_string(msg) << endl;
} }
); );
} }
......
...@@ -51,15 +51,15 @@ public: ...@@ -51,15 +51,15 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
// first message is the forwarded request // first message is the forwarded request
others >> [=] { others >> [=](message& msg) {
auto rp = this->make_response_promise(); auto rp = this->make_response_promise();
split_(workset_, this->current_message()); split_(workset_, msg);
for (auto& x : workset_) for (auto& x : workset_)
this->send(x.first, std::move(x.second)); this->send(x.first, std::move(x.second));
this->become( this->become(
// collect results // collect results
others >> [=]() mutable { others >> [=](message& res) mutable {
join_(value_, this->current_message()); join_(value_, res);
if (--awaited_results_ == 0) { if (--awaited_results_ == 0) {
rp.deliver(make_message(value_)); rp.deliver(make_message(value_));
quit(); quit();
......
...@@ -313,6 +313,7 @@ public: ...@@ -313,6 +313,7 @@ public:
set_flag(value, trap_exit_flag); set_flag(value, trap_exit_flag);
} }
/// @cond PRIVATE
/// Returns the currently processed message. /// Returns the currently processed message.
/// @warning Only set during callback invocation. Calling this member function /// @warning Only set during callback invocation. Calling this member function
/// is undefined behavior (dereferencing a `nullptr`) when not in a /// is undefined behavior (dereferencing a `nullptr`) when not in a
...@@ -320,6 +321,7 @@ public: ...@@ -320,6 +321,7 @@ public:
inline message& current_message() { inline message& current_message() {
return current_element_->msg; return current_element_->msg;
} }
/// @endcond
/// Returns the address of the sender of the current message. /// Returns the address of the sender of the current message.
/// @warning Only set during callback invocation. Calling this member function /// @warning Only set during callback invocation. Calling this member function
......
...@@ -253,16 +253,13 @@ protected: ...@@ -253,16 +253,13 @@ protected:
return f(); return f();
} }
template <class T> template <class T, class U>
static auto call_fun(T& f, message& x, detail::type_list<const message&>) static auto call_fun(T& f, message& x, detail::type_list<U>)
-> decltype(f(x)) {
return f(x);
}
template <class T>
static auto call_fun(T& f, message& x, detail::type_list<message&>)
-> decltype(f(x)) { -> decltype(f(x)) {
x.force_unshare(); static_assert(std::is_same<typename std::decay<U>::type, message>::value,
"catch-all match case callback must take either no "
"arguments or one argument of type `message`, "
"`const message&`, or `message&`");
return f(x); return f(x);
} }
......
...@@ -135,9 +135,11 @@ public: ...@@ -135,9 +135,11 @@ public:
* miscellaneous actor operations * * miscellaneous actor operations *
****************************************************************************/ ****************************************************************************/
/// @cond PRIVATE
message& current_message() { message& current_message() {
return self_->current_message(); return self_->current_message();
} }
/// @endcond
actor_system& system() const { actor_system& system() const {
return self_->system(); return self_->system();
......
...@@ -246,8 +246,8 @@ void actor_registry::start() { ...@@ -246,8 +246,8 @@ void actor_registry::start() {
CAF_LOG_TRACE(CAF_ARG(dm)); CAF_LOG_TRACE(CAF_ARG(dm));
unsubscribe_all(actor_cast<actor>(dm.source)); unsubscribe_all(actor_cast<actor>(dm.source));
}, },
others >> [=]() -> error { others >> [=](const message& msg) -> error {
CAF_LOG_WARNING("unexpected:" << CAF_ARG(self->current_message())); CAF_LOG_WARNING("unexpected:" << CAF_ARG(msg));
return sec::unexpected_message; return sec::unexpected_message;
} }
}; };
...@@ -264,8 +264,8 @@ void actor_registry::start() { ...@@ -264,8 +264,8 @@ void actor_registry::start() {
return sec::cannot_spawn_actor_from_arguments; return sec::cannot_spawn_actor_from_arguments;
return {ok_atom::value, res.first, res.second}; return {ok_atom::value, res.first, res.second};
}, },
others >> [=] { others >> [=](const message& msg) {
CAF_LOG_WARNING("unexpected:" << CAF_ARG(self->current_message())); CAF_LOG_WARNING("unexpected:" << CAF_ARG(msg));
} }
}; };
}; };
......
...@@ -189,8 +189,7 @@ public: ...@@ -189,8 +189,7 @@ public:
acquaintances_.erase(i); acquaintances_.erase(i);
} }
}, },
others >> [=] { others >> [=](const message& msg) {
auto msg = current_message();
CAF_LOG_TRACE(CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(msg));
send_to_acquaintances(msg); send_to_acquaintances(msg);
} }
...@@ -314,8 +313,7 @@ private: ...@@ -314,8 +313,7 @@ private:
behavior proxy_broker::make_behavior() { behavior proxy_broker::make_behavior() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return { return {
others >> [=] { others >> [=](const message& msg) {
auto msg = current_message();
CAF_LOG_TRACE(CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(msg));
group_->send_all_subscribers(current_sender(), msg, context()); group_->send_all_subscribers(current_sender(), msg, context());
} }
......
...@@ -42,12 +42,12 @@ struct splitter_state { ...@@ -42,12 +42,12 @@ struct splitter_state {
behavior fan_out_fan_in(stateful_actor<splitter_state>* self, behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
const std::vector<actor_addr>& workers) { const std::vector<actor_addr>& workers) {
return { return {
others >> [=] { others >> [=](const message& msg) {
self->state.rp = self->make_response_promise(); self->state.rp = self->make_response_promise();
self->state.pending = workers.size(); self->state.pending = workers.size();
// request().await() has LIFO ordering // request().await() has LIFO ordering
for (auto i = workers.rbegin(); i != workers.rend(); ++i) for (auto i = workers.rbegin(); i != workers.rend(); ++i)
self->request(actor_cast<actor>(*i), self->current_message()).generic_await( self->request(actor_cast<actor>(*i), msg).generic_await(
[=](const message& tmp) { [=](const message& tmp) {
self->state.result += tmp; self->state.result += tmp;
if (--self->state.pending == 0) if (--self->state.pending == 0)
......
...@@ -52,8 +52,8 @@ public: ...@@ -52,8 +52,8 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [=] { others >> [=](const message& msg) {
return current_message(); return msg;
} }
}; };
} }
......
...@@ -212,9 +212,9 @@ public: ...@@ -212,9 +212,9 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [=]() -> message { others >> [=](const message& msg) -> message {
quit(exit_reason::normal); quit(exit_reason::normal);
return current_message(); return msg;
} }
}; };
} }
...@@ -232,9 +232,9 @@ public: ...@@ -232,9 +232,9 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [=] { others >> [=](const message& msg) {
CAF_MESSAGE("simple_mirror: return current message"); CAF_MESSAGE("simple_mirror: return current message");
return current_message(); return msg;
} }
}; };
} }
......
...@@ -43,10 +43,10 @@ public: ...@@ -43,10 +43,10 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [=] { others >> [=](message& msg) {
CAF_CHECK_EQUAL(current_message().cvals()->get_reference_count(), 2u); CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 2u);
quit(); quit();
return std::move(current_message()); return std::move(msg);
} }
}; };
} }
......
...@@ -50,8 +50,8 @@ struct sync_mirror : event_based_actor { ...@@ -50,8 +50,8 @@ struct sync_mirror : event_based_actor {
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [=] { others >> [=](const message& msg) {
return current_message(); return msg;
} }
}; };
} }
...@@ -194,9 +194,9 @@ public: ...@@ -194,9 +194,9 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [=]() -> response_promise { others >> [=](message& msg) -> response_promise {
auto rp = make_response_promise(); auto rp = make_response_promise();
request(buddy(), std::move(current_message())).then( request(buddy(), std::move(msg)).then(
[=](gogogo_atom x) mutable { [=](gogogo_atom x) mutable {
rp.deliver(x); rp.deliver(x);
quit(); quit();
...@@ -238,8 +238,8 @@ behavior server(event_based_actor* self) { ...@@ -238,8 +238,8 @@ behavior server(event_based_actor* self) {
[](idle_atom) { [](idle_atom) {
return skip_message(); return skip_message();
}, },
others >> [=] { others >> [=](const message& msg) {
CAF_ERROR("Unexpected message:" << to_string(self->current_message())); CAF_ERROR("Unexpected message:" << to_string(msg));
die(); die();
} }
); );
...@@ -247,8 +247,8 @@ behavior server(event_based_actor* self) { ...@@ -247,8 +247,8 @@ behavior server(event_based_actor* self) {
[](request_atom) { [](request_atom) {
return skip_message(); return skip_message();
}, },
others >> [=] { others >> [=](const message& msg) {
CAF_ERROR("Unexpected message:" << to_string(self->current_message())); CAF_ERROR("Unexpected message:" << to_string(msg));
die(); die();
} }
}; };
...@@ -281,9 +281,9 @@ CAF_TEST(test_void_res) { ...@@ -281,9 +281,9 @@ CAF_TEST(test_void_res) {
CAF_TEST(pending_quit) { CAF_TEST(pending_quit) {
auto mirror = system.spawn([](event_based_actor* self) -> behavior { auto mirror = system.spawn([](event_based_actor* self) -> behavior {
return { return {
others >> [=] { others >> [=](message& msg) {
self->quit(); self->quit();
return std::move(self->current_message()); return std::move(msg);
} }
}; };
}); });
...@@ -351,9 +351,8 @@ CAF_TEST(request) { ...@@ -351,9 +351,8 @@ CAF_TEST(request) {
[&](const down_msg& dm) { [&](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown); CAF_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown);
}, },
others >> [&] { others >> [&](const message& msg) {
CAF_ERROR("Unexpected message: " CAF_ERROR("Unexpected message: " << to_string(msg));
<< to_string(self->current_message()));
} }
); );
auto mirror = system.spawn<sync_mirror>(); auto mirror = system.spawn<sync_mirror>();
...@@ -424,9 +423,8 @@ CAF_TEST(request) { ...@@ -424,9 +423,8 @@ CAF_TEST(request) {
); );
CAF_MESSAGE("mailbox should be empty now"); CAF_MESSAGE("mailbox should be empty now");
self->receive( self->receive(
others >> [&] { others >> [&](const message& msg) {
CAF_ERROR("Unexpected message: " CAF_ERROR("Unexpected message: " << to_string(msg));
<< to_string(self->current_message()));
}, },
after(milliseconds(0)) >> [] { after(milliseconds(0)) >> [] {
CAF_MESSAGE("Mailbox is empty, all good"); CAF_MESSAGE("Mailbox is empty, all good");
...@@ -501,9 +499,8 @@ CAF_TEST(request) { ...@@ -501,9 +499,8 @@ CAF_TEST(request) {
[&](const down_msg& dm) { [&](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown); CAF_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown);
}, },
others >> [&] { others >> [&](const message& msg) {
CAF_ERROR("unexpected message: " CAF_ERROR("unexpected message: " << to_string(msg));
<< to_string(self->current_message()));
} }
); );
} }
......
...@@ -42,9 +42,9 @@ using sub4_atom = atom_constant<atom("sub4")>; ...@@ -42,9 +42,9 @@ using sub4_atom = atom_constant<atom("sub4")>;
CAF_TEST(test_serial_reply) { CAF_TEST(test_serial_reply) {
actor_system system; actor_system system;
auto mirror_behavior = [=](event_based_actor* self) { auto mirror_behavior = [=](event_based_actor* self) {
self->become(others >> [=]() -> message { self->become(others >> [=](const message& msg) -> message {
CAF_MESSAGE("return self->current_message()"); CAF_MESSAGE("return msg: " << to_string(msg));
return self->current_message(); return msg;
}); });
}; };
auto master = system.spawn([=](event_based_actor* self) { auto master = system.spawn([=](event_based_actor* self) {
......
...@@ -30,18 +30,18 @@ CAF_TEST(simple_reply_response) { ...@@ -30,18 +30,18 @@ CAF_TEST(simple_reply_response) {
actor_system system; actor_system system;
auto s = system.spawn([](event_based_actor* self) -> behavior { auto s = system.spawn([](event_based_actor* self) -> behavior {
return ( return (
others >> [=]() -> message { others >> [=](const message& msg) -> message {
CAF_CHECK(self->current_message() == make_message(ok_atom::value)); CAF_CHECK(msg == make_message(ok_atom::value));
self->quit(); self->quit();
return self->current_message(); return msg;
} }
); );
}); });
scoped_actor self{system}; scoped_actor self{system};
self->send(s, ok_atom::value); self->send(s, ok_atom::value);
self->receive( self->receive(
others >> [&] { others >> [&](const message& msg) {
CAF_CHECK(self->current_message() == make_message(ok_atom::value)); CAF_CHECK(msg == make_message(ok_atom::value));
} }
); );
} }
...@@ -286,9 +286,9 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -286,9 +286,9 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(dm)); CAF_LOG_TRACE(CAF_ARG(dm));
this_actor->quit(dm.reason); this_actor->quit(dm.reason);
}, },
others >> [=] { others >> [=](const message& msg) {
CAF_LOG_ERROR("spawn server has received unexpected message:" CAF_LOG_ERROR("unexpected:" << CAF_ARG(msg));
<< CAF_ARG(this_actor->current_message())); static_cast<void>(msg); // suppress unused warning
} }
); );
}, },
...@@ -664,8 +664,9 @@ behavior basp_broker::make_behavior() { ...@@ -664,8 +664,9 @@ behavior basp_broker::make_behavior() {
tick_atom::value, interval); tick_atom::value, interval);
}, },
// catch-all error handler // catch-all error handler
others >> [=] { others >> [=](const message& msg) {
CAF_LOG_ERROR("unexpected message:" << CAF_ARG(current_message())); CAF_LOG_ERROR("unexpected message: " << CAF_ARG(msg));
static_cast<void>(msg); // suppress unused warning
}}; }};
} }
......
...@@ -80,15 +80,6 @@ using namespace caf::io; ...@@ -80,15 +80,6 @@ using namespace caf::io;
namespace { namespace {
#define THROW_ON_UNEXPECTED(selfref) \
others >> [&] { \
throw std::logic_error("unexpected message: " \
+ to_string(selfref->current_message())); \
}, \
after(std::chrono::seconds(0)) >> [&] { \
throw std::logic_error("unexpected timeout"); \
}
static constexpr uint32_t num_remote_nodes = 2; static constexpr uint32_t num_remote_nodes = 2;
using buffer = std::vector<char>; using buffer = std::vector<char>;
...@@ -514,7 +505,6 @@ CAF_TEST(client_handshake_and_dispatch) { ...@@ -514,7 +505,6 @@ CAF_TEST(client_handshake_and_dispatch) {
CAF_CHECK(c == 3); CAF_CHECK(c == 3);
return a + b + c; return a + b + c;
} }
// , THROW_ON_UNEXPECTED(self())
); );
CAF_MESSAGE("exec message of forwarding proxy"); CAF_MESSAGE("exec message of forwarding proxy");
mpx()->exec_runnable(); mpx()->exec_runnable();
...@@ -523,7 +513,6 @@ CAF_TEST(client_handshake_and_dispatch) { ...@@ -523,7 +513,6 @@ CAF_TEST(client_handshake_and_dispatch) {
[](int i) { [](int i) {
CAF_CHECK(i == 6); CAF_CHECK(i == 6);
} }
// , THROW_ON_UNEXPECTED(pseudo_remote(0))
); );
} }
...@@ -634,7 +623,6 @@ CAF_TEST(remote_actor_and_send) { ...@@ -634,7 +623,6 @@ CAF_TEST(remote_actor_and_send) {
CAF_CHECK(self()->current_sender() == result); CAF_CHECK(self()->current_sender() == result);
CAF_CHECK(str == "hi there!"); CAF_CHECK(str == "hi there!");
} }
// , THROW_ON_UNEXPECTED(self())
); );
} }
...@@ -716,7 +704,6 @@ CAF_TEST(indirect_connections) { ...@@ -716,7 +704,6 @@ CAF_TEST(indirect_connections) {
CAF_CHECK(str == "hello from jupiter!"); CAF_CHECK(str == "hello from jupiter!");
return "hello from earth!"; return "hello from earth!";
} }
// , THROW_ON_UNEXPECTED(self())
); );
mpx()->exec_runnable(); // process forwarded message in basp_broker mpx()->exec_runnable(); // process forwarded message in basp_broker
mock() mock()
...@@ -815,7 +802,6 @@ CAF_TEST(automatic_connection) { ...@@ -815,7 +802,6 @@ CAF_TEST(automatic_connection) {
CAF_CHECK(str == "hello from jupiter!"); CAF_CHECK(str == "hello from jupiter!");
return "hello from earth!"; return "hello from earth!";
} }
// , THROW_ON_UNEXPECTED(self())
); );
mpx()->exec_runnable(); // process forwarded message in basp_broker mpx()->exec_runnable(); // process forwarded message in basp_broker
CAF_MESSAGE("response message must take direct route now"); CAF_MESSAGE("response message must take direct route now");
......
...@@ -127,20 +127,20 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -127,20 +127,20 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
self->send(buddy, type, value); self->send(buddy, type, value);
}, },
[=](ping_atom, int value) { [=](ping_atom, int value) {
CAF_MESSAGE("received: " << to_string(self->current_message())); CAF_MESSAGE("received: ping " << value);
write(ping_atom::value, value); write(ping_atom::value, value);
}, },
[=](pong_atom, int value) { [=](pong_atom, int value) {
CAF_MESSAGE("received: " << to_string(self->current_message())); CAF_MESSAGE("received: pong " << value);
write(pong_atom::value, value); write(pong_atom::value, value);
}, },
[=](const down_msg& dm) { [=](const down_msg& dm) {
CAF_MESSAGE("received: " << to_string(self->current_message())); CAF_MESSAGE("received: " << to_string(dm));
if (dm.source == buddy) if (dm.source == buddy)
self->quit(dm.reason); self->quit(dm.reason);
}, },
others >> [=] { others >> [=](const message& msg) {
CAF_MESSAGE("unexpected: " << to_string(self->current_message())); CAF_MESSAGE("unexpected: " << to_string(msg));
} }
); );
} }
......
...@@ -157,8 +157,8 @@ behavior http_worker(http_broker* self, connection_handle hdl) { ...@@ -157,8 +157,8 @@ behavior http_worker(http_broker* self, connection_handle hdl) {
[=](const connection_closed_msg&) { [=](const connection_closed_msg&) {
self->quit(); self->quit();
}, },
others >> [=] { others >> [=](const message& msg) {
aout(self) << "unexpected: " << self->current_message() << endl; aout(self) << "unexpected message: " << msg << endl;
} }
}; };
} }
...@@ -170,8 +170,8 @@ behavior server(broker* self) { ...@@ -170,8 +170,8 @@ behavior server(broker* self) {
aout(self) << "fork on new connection" << endl; aout(self) << "fork on new connection" << endl;
auto worker = self->fork(http_worker, ncm.handle); auto worker = self->fork(http_worker, ncm.handle);
}, },
others >> [=] { others >> [=](const message& msg) {
aout(self) << "Unexpected: " << self->current_message() << endl; aout(self) << "Unexpected message: " << msg << endl;
} }
}; };
} }
......
...@@ -28,49 +28,50 @@ ...@@ -28,49 +28,50 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/all.hpp" #include "caf/io/all.hpp"
using namespace caf;
namespace { namespace {
constexpr char local_host[] = "127.0.0.1"; constexpr char local_host[] = "127.0.0.1";
caf::actor_system_config make_actor_system_config() { actor_system_config make_actor_system_config() {
caf::actor_system_config cfg(caf::test::engine::argc(), actor_system_config cfg(test::engine::argc(), test::engine::argv());
caf::test::engine::argv()); cfg.load<io::middleman>();
cfg.load<caf::io::middleman>(); cfg.add_message_type<std::vector<actor>>("std::vector<actor>");
cfg.add_message_type<std::vector<caf::actor>>("std::vector<caf::actor>");
return cfg; return cfg;
} }
struct fixture { struct fixture {
caf::actor_system server_side{make_actor_system_config()}; actor_system server_side{make_actor_system_config()};
caf::actor_system client_side{make_actor_system_config()}; actor_system client_side{make_actor_system_config()};
caf::io::middleman& server_side_mm = server_side.middleman(); io::middleman& server_side_mm = server_side.middleman();
caf::io::middleman& client_side_mm = client_side.middleman(); io::middleman& client_side_mm = client_side.middleman();
}; };
caf::behavior make_reflector_behavior(caf::event_based_actor* self) { behavior make_reflector_behavior(event_based_actor* self) {
return { return {
caf::others >> [=] { others >> [=](const message& msg) {
self->quit(); self->quit();
return self->current_message(); return msg;
} }
}; };
} }
using spawn_atom = caf::atom_constant<caf::atom("Spawn")>; using spawn_atom = atom_constant<atom("Spawn")>;
using get_group_atom = caf::atom_constant<caf::atom("GetGroup")>; using get_group_atom = atom_constant<atom("GetGroup")>;
struct await_reflector_down_behavior { struct await_reflector_down_behavior {
caf::event_based_actor* self; event_based_actor* self;
int cnt; int cnt;
void operator()(const caf::down_msg&) { void operator()(const down_msg&) {
if (++cnt == 5) if (++cnt == 5)
self->quit(); self->quit();
} }
}; };
struct await_reflector_reply_behavior { struct await_reflector_reply_behavior {
caf::event_based_actor* self; event_based_actor* self;
int cnt; int cnt;
void operator()(const std::string& str, double val) { void operator()(const std::string& str, double val) {
...@@ -82,13 +83,13 @@ struct await_reflector_reply_behavior { ...@@ -82,13 +83,13 @@ struct await_reflector_reply_behavior {
}; };
// `grp` may be either local or remote // `grp` may be either local or remote
void make_client_behavior(caf::event_based_actor* self, void make_client_behavior(event_based_actor* self,
caf::actor server, caf::group grp) { actor server, group grp) {
self->spawn_in_group(grp, make_reflector_behavior); self->spawn_in_group(grp, make_reflector_behavior);
self->spawn_in_group(grp, make_reflector_behavior); self->spawn_in_group(grp, make_reflector_behavior);
self->request(server, spawn_atom::value, grp).then( self->request(server, spawn_atom::value, grp).then(
[=](const std::vector<caf::actor>& vec) { [=](const std::vector<actor>& vec) {
auto is_remote = [=](caf::actor actor) { auto is_remote = [=](actor actor) {
return actor->node() != self->node(); return actor->node() != self->node();
}; };
CAF_CHECK(std::all_of(vec.begin(), vec.end(), is_remote)); CAF_CHECK(std::all_of(vec.begin(), vec.end(), is_remote));
...@@ -101,13 +102,13 @@ void make_client_behavior(caf::event_based_actor* self, ...@@ -101,13 +102,13 @@ void make_client_behavior(caf::event_based_actor* self,
); );
} }
caf::behavior make_server_behavior(caf::event_based_actor* self) { behavior make_server_behavior(event_based_actor* self) {
return { return {
[=](get_group_atom) { [=](get_group_atom) {
return self->system().groups().get("local", "foobar"); return self->system().groups().get("local", "foobar");
}, },
[=](spawn_atom, caf::group group) -> std::vector<caf::actor> { [=](spawn_atom, group group) -> std::vector<actor> {
std::vector<caf::actor> vec; std::vector<actor> vec;
for (auto i = 0; i < 5; ++i) { for (auto i = 0; i < 5; ++i) {
vec.push_back(self->spawn_in_group(group, make_reflector_behavior)); vec.push_back(self->spawn_in_group(group, make_reflector_behavior));
} }
...@@ -137,14 +138,14 @@ CAF_TEST(server_side_group_comm) { ...@@ -137,14 +138,14 @@ CAF_TEST(server_side_group_comm) {
// client side // client side
auto server = client_side_mm.remote_actor(local_host, *port); auto server = client_side_mm.remote_actor(local_host, *port);
CAF_REQUIRE(server); CAF_REQUIRE(server);
caf::scoped_actor group_resolver(client_side, true); scoped_actor group_resolver(client_side, true);
caf::group group; group grp;
group_resolver->request(server, get_group_atom::value).receive( group_resolver->request(server, get_group_atom::value).receive(
[&](caf::group grp) { [&](const group& x) {
group = grp; grp = x;
} }
); );
client_side.spawn(make_client_behavior, server, group); client_side.spawn(make_client_behavior, server, grp);
} }
CAF_TEST(client_side_group_comm) { CAF_TEST(client_side_group_comm) {
......
...@@ -36,10 +36,10 @@ using namespace caf; ...@@ -36,10 +36,10 @@ using namespace caf;
namespace { namespace {
behavior mirror(event_based_actor* self) { behavior mirror() {
return { return {
others >> [=] { others >> [=](message& msg) {
return self->current_message(); return std::move(msg);
} }
}; };
} }
......
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