Commit da22be1b authored by Dominik Charousset's avatar Dominik Charousset

Fix chaining response handlers

parent fbf9dcba
...@@ -71,6 +71,10 @@ public: ...@@ -71,6 +71,10 @@ public:
impl_ = detail::make_behavior(xs...); impl_ = detail::make_behavior(xs...);
} }
inline void swap(behavior& other) {
impl_.swap(other.impl_);
}
void assign(intrusive_ptr<detail::behavior_impl> ptr) { void assign(intrusive_ptr<detail::behavior_impl> ptr) {
impl_.swap(ptr); impl_.swap(ptr);
} }
......
...@@ -548,7 +548,7 @@ public: ...@@ -548,7 +548,7 @@ public:
bool awaits(message_id response_id) const; bool awaits(message_id response_id) const;
maybe<pending_response&> find_pending_response(message_id mid); maybe<behavior&> find_pending_response(message_id mid);
void set_response_handler(message_id response_id, behavior bhvr); void set_response_handler(message_id response_id, behavior bhvr);
......
...@@ -167,6 +167,10 @@ public: ...@@ -167,6 +167,10 @@ public:
set(detail::make_behavior(x, xs...)); set(detail::make_behavior(x, xs...));
} }
inline void swap(typed_behavior& other) {
bhvr_.swap(other.bhvr_);
}
explicit operator bool() const { explicit operator bool() const {
return static_cast<bool>(bhvr_); return static_cast<bool>(bhvr_);
} }
......
...@@ -318,25 +318,24 @@ response_promise fetch_response_promise(local_actor*, response_promise& hdl) { ...@@ -318,25 +318,24 @@ response_promise fetch_response_promise(local_actor*, response_promise& hdl) {
// enables `return sync_send(...).then(...)` // enables `return sync_send(...).then(...)`
bool handle_message_id_res(local_actor* self, message& res, bool handle_message_id_res(local_actor* self, message& res,
maybe<response_promise> hdl) { response_promise hdl) {
CAF_ASSERT(hdl);
CAF_LOG_TRACE(CAF_ARG(res));
if (res.match_elements<atom_value, uint64_t>() if (res.match_elements<atom_value, uint64_t>()
&& res.get_as<atom_value>(0) == atom("MESSAGE_ID")) { && 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.get_as<uint64_t>(1); auto msg_id = message_id::from_integer_value(res.get_as<uint64_t>(1));
auto msg_id = message_id::from_integer_value(id);
auto ref_opt = self->find_pending_response(msg_id); auto ref_opt = self->find_pending_response(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) {
response_promise fhdl = hdl ? *hdl : self->make_response_promise(); behavior inner{std::move(*ref_opt)};
behavior& ref = std::get<1>(*ref_opt); ref_opt->assign(
behavior inner = ref;
ref.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
auto ires = const_cast<behavior&>(inner)(self->current_message()); auto ires = const_cast<behavior&>(inner)(self->current_message());
if (ires && ! handle_message_id_res(self, *ires, fhdl)) if (ires && ! handle_message_id_res(self, *ires, hdl))
fhdl.deliver(*ires); hdl.deliver(*ires);
} }
); );
return true; return true;
...@@ -364,7 +363,7 @@ bool post_process_invoke_res(local_actor* self, bool is_sync_request, ...@@ -364,7 +363,7 @@ bool post_process_invoke_res(local_actor* self, bool is_sync_request,
if (res) { if (res) {
CAF_LOG_DEBUG("respond via response_promise"); CAF_LOG_DEBUG("respond via response_promise");
// deliver empty messages only for sync responses // deliver empty messages only for sync responses
if (! handle_message_id_res(self, *res, none) if (! handle_message_id_res(self, *res, rp)
&& (! res->empty() || is_sync_request)) && (! res->empty() || is_sync_request))
rp.deliver(std::move(*res)); rp.deliver(std::move(*res));
return true; return true;
...@@ -503,24 +502,21 @@ bool local_actor::awaits(message_id mid) const { ...@@ -503,24 +502,21 @@ bool local_actor::awaits(message_id mid) const {
predicate); predicate);
} }
maybe<local_actor::pending_response&> maybe<behavior&> local_actor::find_pending_response(message_id mid) {
local_actor::find_pending_response(message_id mid) {
pending_response_predicate predicate{mid}; pending_response_predicate predicate{mid};
auto last = pending_responses_.end(); auto last = pending_responses_.end();
auto i = std::find_if(pending_responses_.begin(), last, predicate); auto i = std::find_if(pending_responses_.begin(), last, predicate);
if (i == last) { if (i != last)
return none; return i->second;
} return none;
return *i;
} }
void local_actor::set_response_handler(message_id response_id, behavior bhvr) { void local_actor::set_response_handler(message_id response_id, behavior bhvr) {
auto pr = find_pending_response(response_id); auto opt_ref = find_pending_response(response_id);
if (pr) { if (opt_ref) {
if (bhvr.timeout().valid()) { if (bhvr.timeout().valid())
request_sync_timeout_msg(bhvr.timeout(), response_id); request_sync_timeout_msg(bhvr.timeout(), response_id);
} *opt_ref = std::move(bhvr);
pr->second = std::move(bhvr);
} }
} }
......
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