Commit d4365f40 authored by Dominik Charousset's avatar Dominik Charousset

allow actors to respond by returning values

parent e4b22d2e
...@@ -55,6 +55,9 @@ struct optional_any_tuple_visitor { ...@@ -55,6 +55,9 @@ struct optional_any_tuple_visitor {
inline result_type operator()(cow_tuple<Ts...>& value) const { inline result_type operator()(cow_tuple<Ts...>& value) const {
return any_tuple{std::move(value)}; return any_tuple{std::move(value)};
} }
inline result_type operator()(any_tuple& value) const {
return std::move(value);
}
}; };
template<typename... Ts> template<typename... Ts>
......
...@@ -406,17 +406,24 @@ class receive_policy { ...@@ -406,17 +406,24 @@ class receive_policy {
case ordinary_message: { case ordinary_message: {
if (!awaited_response.valid()) { if (!awaited_response.valid()) {
auto previous_node = hm_begin(client, node, policy); auto previous_node = hm_begin(client, node, policy);
if (fun(node->msg)) { auto res = fun(node->msg);
// make sure synchronous request if (res) {
// always receive a response auto mid = node->mid;
auto id = node->mid;
auto sender = node->sender; auto sender = node->sender;
if (id.valid() && !id.is_answered() && sender) { message_header hdr{client, sender, mid.response_id()};
CPPA_LOGMF(CPPA_WARNING, client, if (res->empty()) {
"actor did not reply to a " // make sure synchronous requests
"synchronous request message"); // always receive a response
sender->enqueue({client, sender, id.response_id()}, if (mid.valid() && !mid.is_answered() && sender) {
make_any_tuple(atom("VOID"))); CPPA_LOGMF(CPPA_WARNING, client,
"actor did not reply to a "
"synchronous request message");
sender->enqueue(std::move(hdr),
make_any_tuple(atom("VOID")));
}
} else {
// respond by using the result of 'fun'
sender->enqueue(std::move(hdr), std::move(*res));
} }
hm_cleanup(client, previous_node, policy); hm_cleanup(client, previous_node, policy);
return hm_msg_handled; return hm_msg_handled;
......
...@@ -34,29 +34,29 @@ class event_testee : public sb_actor<event_testee> { ...@@ -34,29 +34,29 @@ class event_testee : public sb_actor<event_testee> {
event_testee() { event_testee() {
wait4string = ( wait4string = (
on<string>() >> [=]() { on<string>() >> [=] {
become(wait4int); become(wait4int);
}, },
on<atom("get_state")>() >> [=]() { on<atom("get_state")>() >> [=] {
reply("wait4string"); return "wait4string";
} }
); );
wait4float = ( wait4float = (
on<float>() >> [=]() { on<float>() >> [=] {
become(wait4string); become(wait4string);
}, },
on<atom("get_state")>() >> [=]() { on<atom("get_state")>() >> [=] {
reply("wait4float"); return "wait4float";
} }
); );
wait4int = ( wait4int = (
on<int>() >> [=]() { on<int>() >> [=] {
become(wait4float); become(wait4float);
}, },
on<atom("get_state")>() >> [=]() { on<atom("get_state")>() >> [=] {
reply("wait4int"); return "wait4int";
} }
); );
} }
...@@ -86,10 +86,10 @@ struct chopstick : public sb_actor<chopstick> { ...@@ -86,10 +86,10 @@ struct chopstick : public sb_actor<chopstick> {
behavior taken_by(actor_ptr whom) { behavior taken_by(actor_ptr whom) {
return ( return (
on<atom("take")>() >> [=]() { on<atom("take")>() >> [=] {
reply(atom("busy")); return atom("busy");
}, },
on(atom("put"), whom) >> [=]() { on(atom("put"), whom) >> [=] {
become(available); become(available);
}, },
on(atom("break")) >> [=]() { on(atom("break")) >> [=]() {
...@@ -104,11 +104,11 @@ struct chopstick : public sb_actor<chopstick> { ...@@ -104,11 +104,11 @@ struct chopstick : public sb_actor<chopstick> {
chopstick() { chopstick() {
available = ( available = (
on(atom("take"), arg_match) >> [=](actor_ptr whom) { on(atom("take"), arg_match) >> [=](actor_ptr whom) -> atom_value {
become(taken_by(whom)); become(taken_by(whom));
reply(atom("taken")); return atom("taken");
}, },
on(atom("break")) >> [=]() { on(atom("break")) >> [=] {
quit(); quit();
} }
); );
...@@ -125,7 +125,7 @@ class testee_actor { ...@@ -125,7 +125,7 @@ class testee_actor {
string_received = true; string_received = true;
}, },
on<atom("get_state")>() >> [&] { on<atom("get_state")>() >> [&] {
reply("wait4string"); return "wait4string";
} }
) )
.until(gref(string_received)); .until(gref(string_received));
...@@ -138,7 +138,7 @@ class testee_actor { ...@@ -138,7 +138,7 @@ class testee_actor {
float_received = true; float_received = true;
}, },
on<atom("get_state")>() >> [&] { on<atom("get_state")>() >> [&] {
reply("wait4float"); return "wait4float";
} }
) )
.until(gref(float_received)); .until(gref(float_received));
...@@ -153,7 +153,7 @@ class testee_actor { ...@@ -153,7 +153,7 @@ class testee_actor {
wait4float(); wait4float();
}, },
on<atom("get_state")>() >> [&] { on<atom("get_state")>() >> [&] {
reply("wait4int"); return "wait4int";
} }
); );
} }
...@@ -253,8 +253,8 @@ class fixed_stack : public sb_actor<fixed_stack> { ...@@ -253,8 +253,8 @@ class fixed_stack : public sb_actor<fixed_stack> {
data.push_back(what); data.push_back(what);
become(filled); become(filled);
}, },
on(atom("pop")) >> [=]() { on(atom("pop")) >> [=] {
reply(atom("failure")); return atom("failure");
} }
); );
...@@ -276,7 +276,7 @@ struct simple_mirror : sb_actor<simple_mirror> { ...@@ -276,7 +276,7 @@ struct simple_mirror : sb_actor<simple_mirror> {
behavior init_state = ( behavior init_state = (
others() >> [] { others() >> [] {
reply_tuple(self->last_dequeued()); return self->last_dequeued();
} }
); );
...@@ -556,8 +556,8 @@ int main() { ...@@ -556,8 +556,8 @@ int main() {
auto factory = factory::event_based([&](int* i, float*, string*) { auto factory = factory::event_based([&](int* i, float*, string*) {
become ( become (
on(atom("get_int")) >> [i]() { on(atom("get_int")) >> [i] {
reply(*i); return *i;
}, },
on(atom("set_int"), arg_match) >> [i](int new_value) { on(atom("set_int"), arg_match) >> [i](int new_value) {
*i = new_value; *i = new_value;
...@@ -618,7 +618,7 @@ int main() { ...@@ -618,7 +618,7 @@ int main() {
auto sync_testee1 = spawn<blocking_api>([] { auto sync_testee1 = spawn<blocking_api>([] {
receive ( receive (
on(atom("get")) >> [] { on(atom("get")) >> [] {
reply(42, 2); return make_cow_tuple(42, 2);
} }
); );
}); });
...@@ -681,11 +681,11 @@ int main() { ...@@ -681,11 +681,11 @@ int main() {
self->monitor(sync_testee); self->monitor(sync_testee);
send(sync_testee, "hi"); send(sync_testee, "hi");
receive ( receive (
on("whassup?") >> [&] { on("whassup?") >> [&]() -> std::string {
CPPA_CHECK(true); CPPA_CHECK(true);
// this is NOT a reply, it's just an asynchronous message // this is NOT a reply, it's just an asynchronous message
send(self->last_sender(), "a lot!"); send(self->last_sender(), "a lot!");
reply("nothing"); return "nothing";
} }
); );
receive ( receive (
...@@ -788,8 +788,8 @@ int main() { ...@@ -788,8 +788,8 @@ int main() {
auto f = factory::event_based([](string* name) { auto f = factory::event_based([](string* name) {
become ( become (
on(atom("get_name")) >> [name]() { on(atom("get_name")) >> [name] {
reply(atom("name"), *name); return make_cow_tuple(atom("name"), *name);
} }
); );
}); });
......
...@@ -7,15 +7,15 @@ using namespace cppa::placeholders; ...@@ -7,15 +7,15 @@ using namespace cppa::placeholders;
struct sync_mirror : sb_actor<sync_mirror> { struct sync_mirror : sb_actor<sync_mirror> {
behavior init_state = ( behavior init_state = (
others() >> [] { reply_tuple(self->last_dequeued()); } others() >> [] { return self->last_dequeued(); }
); );
}; };
// replies to 'f' with 0.0f and to 'i' with 0 // replies to 'f' with 0.0f and to 'i' with 0
struct float_or_int : sb_actor<float_or_int> { struct float_or_int : sb_actor<float_or_int> {
behavior init_state = ( behavior init_state = (
on(atom("f")) >> [] { reply(0.0f); }, on(atom("f")) >> [] { return 0.0f; },
on(atom("i")) >> [] { reply(0); } on(atom("i")) >> [] { return 0; }
); );
}; };
...@@ -273,7 +273,7 @@ int main() { ...@@ -273,7 +273,7 @@ int main() {
spawn<monitored + blocking_api>([] { // client spawn<monitored + blocking_api>([] { // client
auto s = spawn<server, linked>(); // server auto s = spawn<server, linked>(); // server
auto w = spawn<linked>([] { // worker auto w = spawn<linked>([] { // worker
become(on(atom("request")) >> []{ reply(atom("response")); }); become(on(atom("request")) >> []{ return atom("response"); });
}); });
// first 'idle', then 'request' // first 'idle', then 'request'
send_as(w, s, atom("idle")); send_as(w, s, atom("idle"));
......
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