Commit 7ad651b1 authored by Dominik Charousset's avatar Dominik Charousset

Fix nested `return sync_send(...).then(...)`

parent 6884f0c2
...@@ -189,63 +189,70 @@ class invoke_policy { ...@@ -189,63 +189,70 @@ class invoke_policy {
return res; return res;
} }
CAF_LOGF_DEBUG("res = " << to_string(*res)); CAF_LOGF_DEBUG("res = " << to_string(*res));
if (res->template match_elements<atom_value, uint64_t>() if (handle_message_id_res(self, *res, none)) {
&& res->template get_as<atom_value>(0) == atom("MESSAGE_ID")) { return message{};
}
// respond by using the result of 'fun'
CAF_LOG_DEBUG("respond via response_promise");
auto fhdl = fetch_response_promise(self, hdl);
if (fhdl) {
fhdl.deliver(std::move(*res));
// inform caller about success by returning not none
return message{};
}
return res;
}
private:
// enables `return sync_send(...).then(...)`
template <class Actor>
bool handle_message_id_res(Actor* self, message& res,
optional<response_promise> hdl) {
if (res.match_elements<atom_value, uint64_t>()
&& res.get_as<atom_value>(0) == atom("MESSAGE_ID")) {
CAF_LOG_DEBUG("message handler returned a message id wrapper"); CAF_LOG_DEBUG("message handler returned a message id wrapper");
auto id = res->template get_as<uint64_t>(1); auto id = res.get_as<uint64_t>(1);
auto msg_id = message_id::from_integer_value(id); auto msg_id = message_id::from_integer_value(id);
auto ref_opt = self->sync_handler(msg_id); auto ref_opt = self->sync_handler(msg_id);
// install a behavior that calls the user-defined behavior // install a behavior that calls the user-defined behavior
// and using the result of its inner behavior as response // and using the result of its inner behavior as response
if (ref_opt) { if (ref_opt) {
auto fhdl = fetch_response_promise(self, hdl); response_promise fhdl = hdl ? *hdl : self->make_response_promise();
behavior inner = *ref_opt; behavior inner = *ref_opt;
ref_opt->assign( ref_opt->assign(
others() >> [=] { others() >> [=] {
// inner is const inside this lambda and mutable a C++14 feature // inner is const inside this lambda and mutable a C++14 feature
behavior cpy = inner; behavior cpy = inner;
auto inner_res = cpy(self->last_dequeued()); auto inner_res = cpy(self->last_dequeued());
if (inner_res) { if (inner_res && !handle_message_id_res(self, *inner_res, fhdl)) {
fhdl.deliver(*inner_res); fhdl.deliver(*inner_res);
} }
} }
); );
return true;
} }
return res;
}
// respond by using the result of 'fun'
CAF_LOG_DEBUG("respond via response_promise");
auto fhdl = fetch_response_promise(self, hdl);
if (fhdl) {
fhdl.deliver(std::move(*res));
// inform caller about success by returning not none
return message{};
} }
return res; return false;
} }
private:
// identifies 'special' messages that should not be processed normally: // identifies 'special' messages that should not be processed normally:
// - system messages such as EXIT (if self doesn't trap exits) and TIMEOUT // - system messages such as exit_msg and timeout_msg
// - expired synchronous response messages // - expired synchronous response messages
template <class Actor> template <class Actor>
msg_type filter_msg(Actor* self, mailbox_element& node) { msg_type filter_msg(Actor* self, mailbox_element& node) {
const message& msg = node.msg; const message& msg = node.msg;
auto mid = node.mid; auto mid = node.mid;
if (msg.size() == 1) { if (mid.is_response()) {
if (msg.match_element<exit_msg>(0)) { if (msg.match_elements<sync_timeout_msg>()) {
auto& em = msg.get_as<exit_msg>(0); return msg_type::timeout_response;
CAF_REQUIRE(!mid.valid());
// make sure to get rid of attachables if they're no longer needed
self->unlink_from(em.source);
if (self->trap_exit() == false) {
if (em.reason != exit_reason::normal) {
self->quit(em.reason);
return msg_type::non_normal_exit;
} }
return msg_type::normal_exit; return self->awaits(mid) ? msg_type::sync_response
: msg_type::expired_sync_response;
} }
} else if (msg.match_element<timeout_msg>(0)) { if (msg.size() != 1) {
return msg_type::ordinary;
}
if (msg.match_element<timeout_msg>(0)) {
auto& tm = msg.get_as<timeout_msg>(0); auto& tm = msg.get_as<timeout_msg>(0);
auto tid = tm.timeout_id; auto tid = tm.timeout_id;
CAF_REQUIRE(!mid.valid()); CAF_REQUIRE(!mid.valid());
...@@ -254,13 +261,19 @@ class invoke_policy { ...@@ -254,13 +261,19 @@ class invoke_policy {
} }
return self->waits_for_timeout(tid) ? msg_type::inactive_timeout return self->waits_for_timeout(tid) ? msg_type::inactive_timeout
: msg_type::expired_timeout; : msg_type::expired_timeout;
} else if (mid.is_response() && msg.match_element<sync_timeout_msg>(0)) {
return msg_type::timeout_response;
} }
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
CAF_REQUIRE(!mid.valid());
// make sure to get rid of attachables if they're no longer needed
self->unlink_from(em.source);
if (self->trap_exit() == false) {
if (em.reason != exit_reason::normal) {
self->quit(em.reason);
return msg_type::non_normal_exit;
}
return msg_type::normal_exit;
} }
if (mid.is_response()) {
return self->awaits(mid) ? msg_type::sync_response
: msg_type::expired_sync_response;
} }
return msg_type::ordinary; return msg_type::ordinary;
} }
......
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