Commit 0504b91e authored by Dominik Charousset's avatar Dominik Charousset

Restructure response handling

Implement new API for behaviors that skips the intermediate maybe<message>
return type. Instead, behaviors accept visitors that allow fine-grained access
to the result of message handlers. Also remove the feature of returning
continue helpers, since this is obsoleted by delegate.
parent 71fde408
...@@ -55,6 +55,7 @@ set (LIBCAF_CORE_SRCS ...@@ -55,6 +55,7 @@ set (LIBCAF_CORE_SRCS
src/get_root_uuid.cpp src/get_root_uuid.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/invoke_result_visitor.cpp
src/match_case.cpp src/match_case.cpp
src/merged_tuple.cpp src/merged_tuple.cpp
src/monitorable_actor.cpp src/monitorable_actor.cpp
......
...@@ -32,6 +32,7 @@ ...@@ -32,6 +32,7 @@
#include "caf/group.hpp" #include "caf/group.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/result.hpp"
#include "caf/channel.hpp" #include "caf/channel.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
......
...@@ -97,8 +97,13 @@ public: ...@@ -97,8 +97,13 @@ public:
} }
/// Runs this handler and returns its (optional) result. /// Runs this handler and returns its (optional) result.
inline maybe<message> operator()(message& arg) { inline maybe<message> operator()(message& x) {
return (impl_) ? impl_->invoke(arg) : none; return impl_ ? impl_->invoke(x) : none;
}
/// Runs this handler with callback.
inline bool operator()(detail::invoke_result_visitor& f, message& x) {
return impl_ ? impl_->invoke(f, x) : false;
} }
/// Checks whether this behavior is not empty. /// Checks whether this behavior is not empty.
......
...@@ -20,22 +20,14 @@ ...@@ -20,22 +20,14 @@
#ifndef CAF_CONTINUE_HELPER_HPP #ifndef CAF_CONTINUE_HELPER_HPP
#define CAF_CONTINUE_HELPER_HPP #define CAF_CONTINUE_HELPER_HPP
#include <functional>
#include "caf/on.hpp"
#include "caf/behavior.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/message_handler.hpp"
namespace caf { namespace caf {
class local_actor;
/// Helper class to enable users to add continuations /// Helper class to enable users to add continuations
/// when dealing with synchronous sends. /// when dealing with synchronous sends.
class continue_helper { class continue_helper {
public: public:
using message_id_wrapper_tag = int;
continue_helper(message_id mid); continue_helper(message_id mid);
/// Returns the ID of the expected response message. /// Returns the ID of the expected response message.
......
...@@ -26,7 +26,7 @@ namespace caf { ...@@ -26,7 +26,7 @@ namespace caf {
/// Helper class to indicate that a request has been forwarded. /// Helper class to indicate that a request has been forwarded.
template <class... Ts> template <class... Ts>
struct delegated { class delegated {
// nop // nop
}; };
......
...@@ -42,12 +42,12 @@ ...@@ -42,12 +42,12 @@
#include "caf/detail/int_list.hpp" #include "caf/detail/int_list.hpp"
#include "caf/detail/apply_args.hpp" #include "caf/detail/apply_args.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/invoke_result_visitor.hpp"
#include "caf/detail/tail_argument_token.hpp" #include "caf/detail/tail_argument_token.hpp"
namespace caf { namespace caf {
class message_handler; class message_handler;
using bhvr_invoke_result = maybe<message>;
} // namespace caf } // namespace caf
...@@ -68,13 +68,15 @@ public: ...@@ -68,13 +68,15 @@ public:
behavior_impl(duration tout = duration{}); behavior_impl(duration tout = duration{});
virtual bhvr_invoke_result invoke(message&); virtual bool invoke(detail::invoke_result_visitor& f, message&);
inline bhvr_invoke_result invoke(message&& arg) { inline bool invoke(detail::invoke_result_visitor& f, message&& arg) {
message tmp(std::move(arg)); message tmp(std::move(arg));
return invoke(tmp); return invoke(f, tmp);
} }
maybe<message> invoke(message&);
virtual void handle_timeout(); virtual void handle_timeout();
inline const duration& timeout() const { inline const duration& timeout() const {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_INVOKE_VISITOR_HPP
#define CAF_DETAIL_INVOKE_VISITOR_HPP
#include <tuple>
#include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/unit.hpp"
#include "caf/maybe.hpp"
#include "caf/result.hpp"
#include "caf/message.hpp"
#include "caf/skip_message.hpp"
#include "caf/typed_continue_helper.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/apply_args.hpp"
namespace caf {
namespace detail {
class invoke_result_visitor {
public:
virtual ~invoke_result_visitor();
constexpr invoke_result_visitor() {
// nop
}
// severs as catch-all case for all values producing
// no result such as response_promise
virtual void operator()() = 0;
// called if the message handler explicitly returned an error
virtual void operator()(error&) = 0;
// called if the message handler returned any "ordinary" value
virtual void operator()(message&) = 0;
// called if the message handler returns explictly none or unit
virtual void operator()(const none_t&) = 0;
// map unit to an empty message
inline void operator()(const unit_t&) {
message empty_msg;
(*this)(empty_msg);
}
// unwrap maybes
template <class T>
void operator()(maybe<T>& x) {
if (x)
(*this)(*x);
else if (x.empty())
(*this)(none);
else {
auto tmp = x.error();
(*this)(tmp);
}
}
// convert values to messages
template <class... Ts>
void operator()(Ts&... xs) {
auto tmp = make_message(std::move(xs)...);
(*this)(tmp);
}
// unwrap tuples
template <class... Ts>
void operator()(std::tuple<Ts...>& xs) {
apply_args(*this, get_indices(xs), xs);
}
// disambiguations
inline void operator()(none_t& x) {
(*this)(const_cast<const none_t&>(x));
}
inline void operator()(unit_t& x) {
(*this)(const_cast<const unit_t&>(x));
}
// special purpose handler that don't procude results
inline void operator()(response_promise&) {
(*this)();
}
template <class... Ts>
void operator()(typed_response_promise<Ts...>&) {
(*this)();
}
template <class... Ts>
void operator()(delegated<Ts...>&) {
(*this)();
}
// visit API - returns true if T was visited, false if T was skipped
template <class T>
bool visit(T& x) {
(*this)(x);
return true;
}
inline bool visit(skip_message_t&) {
return false;
}
inline bool visit(const skip_message_t&) {
return false;
}
inline bool visit(maybe<skip_message_t>& x) {
if (x.valid())
return false;
(*this)(x);
return true;
}
template <class... Ts>
bool visit(result<Ts...>& x) {
switch (x.flag) {
case rt_value:
(*this)(x.value);
return true;
case rt_error:
(*this)(x.err);
return true;
case rt_delegated:
(*this)();
return true;
case rt_skip_message:
return false;
}
}
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_INVOKE_VISITOR_HPP
...@@ -25,14 +25,15 @@ ...@@ -25,14 +25,15 @@
namespace caf { namespace caf {
template <class> // templates
class maybe; template <class> class maybe;
template <class> class intrusive_ptr;
template <class> struct actor_cast_access;
template <class> class typed_continue_helper;
template <class> // variadic templates
class intrusive_ptr; template <class...> class delegated;
template <class...> class typed_response_promise;
template <class>
struct actor_cast_access;
// classes // classes
class actor; class actor;
...@@ -61,8 +62,10 @@ class actor_registry; ...@@ -61,8 +62,10 @@ class actor_registry;
class blocking_actor; class blocking_actor;
class execution_unit; class execution_unit;
class proxy_registry; class proxy_registry;
class continue_helper;
class mailbox_element; class mailbox_element;
class message_handler; class message_handler;
class response_promise;
class event_based_actor; class event_based_actor;
class binary_serializer; class binary_serializer;
class binary_deserializer; class binary_deserializer;
......
...@@ -34,16 +34,16 @@ ...@@ -34,16 +34,16 @@
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/pseudo_tuple.hpp" #include "caf/detail/pseudo_tuple.hpp"
#include "caf/detail/left_or_right.hpp" #include "caf/detail/left_or_right.hpp"
#include "caf/detail/optional_message_visitor.hpp" #include "caf/detail/invoke_result_visitor.hpp"
namespace caf { namespace caf {
class match_case { class match_case {
public: public:
enum result { enum result {
fall_through,
no_match, no_match,
match match,
skip
}; };
match_case(bool has_wildcard, uint32_t token); match_case(bool has_wildcard, uint32_t token);
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
virtual ~match_case(); virtual ~match_case();
virtual result invoke(maybe<message>&, message&) = 0; virtual result invoke(detail::invoke_result_visitor&, message&) = 0;
inline uint32_t type_token() const { inline uint32_t type_token() const {
return token_; return token_;
...@@ -192,7 +192,7 @@ public: ...@@ -192,7 +192,7 @@ public:
// nop // nop
} }
match_case::result invoke(maybe<message>& res, message& msg) override { match_case::result invoke(detail::invoke_result_visitor& f, message& msg) override {
intermediate_tuple it; intermediate_tuple it;
detail::meta_elements<pattern> ms; detail::meta_elements<pattern> ms;
// check if try_match() reports success // check if try_match() reports success
...@@ -210,10 +210,8 @@ public: ...@@ -210,10 +210,8 @@ public:
} }
} }
lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_}; lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_};
detail::optional_message_visitor omv; auto fun_res = apply_args(fun, detail::get_indices(it), it);
auto funres = apply_args(fun, detail::get_indices(it), it); return f.visit(fun_res) ? match_case::match : match_case::skip;
res = omv(funres);
return match_case::match;
} }
protected: protected:
...@@ -242,13 +240,11 @@ public: ...@@ -242,13 +240,11 @@ public:
// nop // nop
} }
match_case::result invoke(maybe<message>& res, message& msg) override { match_case::result invoke(detail::invoke_result_visitor& f, message& msg) override {
lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_}; lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_};
arg_types token; arg_types token;
auto fun_res = call_fun(fun, msg, token); auto fun_res = call_fun(fun, msg, token);
detail::optional_message_visitor omv; return f.visit(fun_res) ? match_case::match : match_case::skip;
res = omv(fun_res);
return match_case::match;
} }
protected: protected:
...@@ -294,7 +290,7 @@ public: ...@@ -294,7 +290,7 @@ public:
// however, dealing with all the template parameters in a debugger // however, dealing with all the template parameters in a debugger
// is just dreadful; this "hack" essentially hides all the ugly // is just dreadful; this "hack" essentially hides all the ugly
// template boilterplate types when debugging CAF applications // template boilterplate types when debugging CAF applications
match_case::result invoke(maybe<message>& res, message& msg) override { match_case::result invoke(detail::invoke_result_visitor& f, message& msg) override {
struct storage { struct storage {
storage() : valid(false) { storage() : valid(false) {
// nop // nop
...@@ -311,10 +307,8 @@ public: ...@@ -311,10 +307,8 @@ public:
if (prepare_invoke(msg, &st.data)) { if (prepare_invoke(msg, &st.data)) {
st.valid = true; st.valid = true;
lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_}; lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_};
detail::optional_message_visitor omv; auto fun_res = apply_args(fun, detail::get_indices(st.data), st.data);
auto funres = apply_args(fun, detail::get_indices(st.data), st.data); return f.visit(fun_res) ? match_case::match : match_case::skip;
res = omv(funres);
return match_case::match;
} }
return match_case::no_match; return match_case::no_match;
} }
......
...@@ -65,6 +65,10 @@ public: ...@@ -65,6 +65,10 @@ public:
return *this; return *this;
} }
inline bool is_async() const {
return value_ == 0;
}
inline bool is_response() const { inline bool is_response() const {
return (value_ & response_flag_mask) != 0; return (value_ & response_flag_mask) != 0;
} }
......
...@@ -136,6 +136,12 @@ private: ...@@ -136,6 +136,12 @@ private:
static_assert(detail::is_callable<F>::value, "argument is not callable"); static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value, static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context"); "match cases are not allowed in this context");
static_assert(std::is_same<
void,
typename detail::get_callable_trait<F>::result_type
>::value,
"response handlers are not allowed to have a return "
"type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_awaited_response_handler(mid_, self_->set_awaited_response_handler(mid_,
behavior{std::move(f), std::forward<Ts>(xs)...}, behavior{std::move(f), std::forward<Ts>(xs)...},
...@@ -149,6 +155,12 @@ private: ...@@ -149,6 +155,12 @@ private:
static_assert(detail::is_callable<F>::value, "argument is not callable"); static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value, static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context"); "match cases are not allowed in this context");
static_assert(std::is_same<
void,
typename detail::get_callable_trait<F>::result_type
>::value,
"response handlers are not allowed to have a return "
"type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_multiplexed_response_handler(mid_, self_->set_multiplexed_response_handler(mid_,
behavior{std::move(f), std::forward<Ts>(xs)...}, behavior{std::move(f), std::forward<Ts>(xs)...},
...@@ -197,6 +209,12 @@ private: ...@@ -197,6 +209,12 @@ private:
static_assert(detail::is_callable<F>::value, "argument is not callable"); static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value, static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context"); "match cases are not allowed in this context");
static_assert(std::is_same<
void,
typename detail::get_callable_trait<F>::result_type
>::value,
"response handlers are not allowed to have a return "
"type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
behavior tmp; behavior tmp;
if (! ef) if (! ef)
......
...@@ -57,11 +57,22 @@ public: ...@@ -57,11 +57,22 @@ public:
/// For non-requests, nothing is done. /// For non-requests, nothing is done.
void deliver(error x); void deliver(error x);
/// Returns whether this response promise replies to an asynchronous message.
bool async() const;
/// Queries whether this promise is a valid promise that is not satisfied yet. /// Queries whether this promise is a valid promise that is not satisfied yet.
inline bool pending() const { inline bool pending() const {
return ! stages_.empty() || source_; return ! stages_.empty() || source_;
} }
/// @cond PRIVATE
inline local_actor* self() {
return self_;
}
/// @endcond
private: private:
using forwarding_stack = std::vector<actor_addr>; using forwarding_stack = std::vector<actor_addr>;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_RESULT_HPP
#define CAF_RESULT_HPP
#include "caf/none.hpp"
#include "caf/error.hpp"
#include "caf/message.hpp"
#include "caf/skip_message.hpp"
namespace caf {
enum result_runtime_type {
rt_value,
rt_error,
rt_delegated,
rt_skip_message
};
template <class... Ts>
struct result {
public:
result(Ts... xs) : flag(rt_value), value(make_message(std::move(xs)...)) {
// nop
}
template <class E,
class = typename std::enable_if<
std::is_same<
decltype(make_error(std::declval<const E&>())),
error
>::value
>::type>
result(E x) : flag(rt_error), err(make_error(x)) {
// nop
}
result(error x) : flag(rt_error), err(std::move(x)) {
// nop
}
result(skip_message_t) : flag(rt_skip_message) {
// nop
}
result(delegated<Ts...>) : flag(rt_delegated) {
// nop
}
result_runtime_type flag;
message value;
error err;
};
} // namespace caf
#endif // CAF_RESULT_HPP
...@@ -22,17 +22,11 @@ ...@@ -22,17 +22,11 @@
#include "caf/continue_helper.hpp" #include "caf/continue_helper.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
namespace caf { namespace caf {
template <class OutputList> template <class OutputList>
class typed_continue_helper { class typed_continue_helper {
public: public:
using message_id_wrapper_tag = int;
typed_continue_helper(message_id mid) : ch_(mid) { typed_continue_helper(message_id mid) : ch_(mid) {
// nop // nop
} }
...@@ -45,6 +39,10 @@ public: ...@@ -45,6 +39,10 @@ public:
return ch_.get_message_id(); return ch_.get_message_id();
} }
continue_helper& unbox() {
return ch_;
}
private: private:
continue_helper ch_; continue_helper ch_;
}; };
......
...@@ -28,9 +28,8 @@ namespace { ...@@ -28,9 +28,8 @@ namespace {
class combinator final : public behavior_impl { class combinator final : public behavior_impl {
public: public:
bhvr_invoke_result invoke(message& arg) override { bool invoke(detail::invoke_result_visitor& f, message& arg) override {
auto res = first->invoke(arg); return first->invoke(f, arg) || second->invoke(f, arg);
return res ? res : second->invoke(arg);
} }
void handle_timeout() override { void handle_timeout() override {
...@@ -55,6 +54,27 @@ private: ...@@ -55,6 +54,27 @@ private:
pointer second; pointer second;
}; };
class maybe_message_visitor : public detail::invoke_result_visitor {
public:
maybe<message> value;
void operator()() override {
value = message{};
}
void operator()(error& x) override {
value = std::move(x);
}
void operator()(message& x) override {
value = std::move(x);
}
void operator()(const none_t&) override {
(*this)();
}
};
} // namespace <anonymous> } // namespace <anonymous>
behavior_impl::~behavior_impl() { behavior_impl::~behavior_impl() {
...@@ -68,16 +88,25 @@ behavior_impl::behavior_impl(duration tout) ...@@ -68,16 +88,25 @@ behavior_impl::behavior_impl(duration tout)
// nop // nop
} }
bhvr_invoke_result behavior_impl::invoke(message& msg) { bool behavior_impl::invoke(detail::invoke_result_visitor& f, message& msg) {
auto msg_token = msg.type_token(); auto msg_token = msg.type_token();
bhvr_invoke_result res; for (auto i = begin_; i != end_; ++i)
for (auto i = begin_; i != end_; ++i) { if (i->has_wildcard || i->type_token == msg_token)
if ((i->has_wildcard || i->type_token == msg_token) switch (i->ptr->invoke(f, msg)) {
&& i->ptr->invoke(res, msg) != match_case::no_match) { case match_case::match:
return res; return true;
} case match_case::skip:
} return false;
return none; case match_case::no_match:
static_cast<void>(0); // nop
};
return false;
}
maybe<message> behavior_impl::invoke(message& x) {
maybe_message_visitor f;
invoke(f, x);
return std::move(f.value);
} }
void behavior_impl::handle_timeout() { void behavior_impl::handle_timeout() {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/invoke_result_visitor.hpp"
namespace caf {
namespace detail {
invoke_result_visitor::~invoke_result_visitor() {
// nop
}
} // namespace detail
} // namespace caf
...@@ -322,6 +322,78 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) { ...@@ -322,6 +322,78 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
return msg_type::ordinary; return msg_type::ordinary;
} }
class invoke_result_visitor_helper {
public:
invoke_result_visitor_helper(response_promise x) : rp(x) {
// nop
}
void operator()(error& x) {
CAF_LOG_DEBUG("report error back to requesting actor");
rp.deliver(std::move(x));
}
void operator()(message& x) {
CAF_LOG_DEBUG("respond via response_promise");
// suppress empty messages for asynchronous messages
if (x.empty() && rp.async())
return;
rp.deliver(std::move(x));
}
void operator()(const none_t&) {
error err = sec::unexpected_response;
(*this)(err);
}
private:
response_promise rp;
};
class local_actor_invoke_result_visitor : public detail::invoke_result_visitor {
public:
local_actor_invoke_result_visitor(local_actor* ptr) : self_(ptr) {
// nop
}
void operator()() override {
// nop
}
template <class T>
void delegate(T& x) {
auto rp = self_->make_response_promise();
if (! rp.pending()) {
CAF_LOG_DEBUG("suppress response message due to invalid response promise");
return;
}
invoke_result_visitor_helper f{std::move(rp)};
f(x);
}
void operator()(error& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_LOG_DEBUG("report error back to requesting actor");
delegate(x);
}
void operator()(message& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_LOG_DEBUG("respond via response_promise");
delegate(x);
}
void operator()(const none_t& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_LOG_DEBUG("message handler returned none");
delegate(x);
}
private:
local_actor* self_;
};
/*
response_promise fetch_response_promise(local_actor* self, int) { response_promise fetch_response_promise(local_actor* self, int) {
return self->make_response_promise(); return self->make_response_promise();
} }
...@@ -338,25 +410,7 @@ bool handle_message_id_res(local_actor* self, message& res, ...@@ -338,25 +410,7 @@ bool handle_message_id_res(local_actor* self, message& 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 msg_id = message_id::from_integer_value(res.get_as<uint64_t>(1));
auto fun = [=](maybe<local_actor::pending_response&> ref_opt) mutable {
// install a behavior that calls the user-defined behavior
// and using the result of its inner behavior as response
if (ref_opt) {
behavior inner{std::move(ref_opt->second.first)};
ref_opt->second.first.assign(
others >> [=]() mutable {
auto ires = const_cast<behavior&>(inner)(self->current_message());
if (ires && ! handle_message_id_res(self, *ires, hdl))
hdl.deliver(*ires);
}
);
return true;
}
return false;
};
return fun(self->find_multiplexed_response(msg_id))
|| fun(self->find_awaited_response(msg_id));
} }
return false; return false;
} }
...@@ -392,12 +446,14 @@ bool post_process_invoke_res(local_actor* self, bool is_sync_request, ...@@ -392,12 +446,14 @@ bool post_process_invoke_res(local_actor* self, bool is_sync_request,
} }
return false; return false;
} }
*/
invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
behavior& fun, behavior& fun,
message_id awaited_id) { message_id awaited_id) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE(CAF_ARG(*ptr) << CAF_ARG(awaited_id)); CAF_LOG_TRACE(CAF_ARG(*ptr) << CAF_ARG(awaited_id));
local_actor_invoke_result_visitor visitor{this};
switch (filter_msg(this, *ptr)) { switch (filter_msg(this, *ptr)) {
case msg_type::normal_exit: case msg_type::normal_exit:
CAF_LOG_DEBUG("dropped normal exit signal"); CAF_LOG_DEBUG("dropped normal exit signal");
...@@ -438,8 +494,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -438,8 +494,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
if (ref_fun.timeout().valid()) { if (ref_fun.timeout().valid()) {
ref_fun.handle_timeout(); ref_fun.handle_timeout();
} }
} else if (! post_process_invoke_res(this, false, } else if (! ref_fun(visitor, current_element_->msg)) {
ref_fun(current_element_->msg))) { //} else if (! post_process_invoke_res(this, false,
// ref_fun(current_element_->msg))) {
CAF_LOG_WARNING("multiplexed response failure occured:" << CAF_ARG(id())); CAF_LOG_WARNING("multiplexed response failure occured:" << CAF_ARG(id()));
quit(exit_reason::unhandled_sync_failure); quit(exit_reason::unhandled_sync_failure);
} }
...@@ -458,8 +515,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -458,8 +515,9 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
fun.handle_timeout(); fun.handle_timeout();
} }
} else { } else {
if (! post_process_invoke_res(this, false, if (! fun(visitor, current_element_->msg)) {
fun(current_element_->msg))) { //if (! post_process_invoke_res(this, false,
// fun(current_element_->msg))) {
CAF_LOG_WARNING("sync response failure occured:" << CAF_ARG(id())); CAF_LOG_WARNING("sync response failure occured:" << CAF_ARG(id()));
quit(exit_reason::unhandled_sync_failure); quit(exit_reason::unhandled_sync_failure);
} }
...@@ -480,9 +538,10 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -480,9 +538,10 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
has_timeout(false); has_timeout(false);
} }
ptr.swap(current_element_); ptr.swap(current_element_);
auto is_req = current_element_->mid.is_request(); //auto is_req = current_element_->mid.is_request();
auto res = post_process_invoke_res(this, is_req, auto res = fun(visitor, current_element_->msg);
fun(current_element_->msg)); //auto res = post_process_invoke_res(this, is_req,
// fun(current_element_->msg));
ptr.swap(current_element_); ptr.swap(current_element_);
if (res) if (res)
return im_success; return im_success;
......
...@@ -19,10 +19,12 @@ ...@@ -19,10 +19,12 @@
#include "caf/monitorable_actor.hpp" #include "caf/monitorable_actor.hpp"
#include "caf/on.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/message_handler.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/default_attachable.hpp" #include "caf/default_attachable.hpp"
......
...@@ -32,15 +32,22 @@ response_promise::response_promise() ...@@ -32,15 +32,22 @@ response_promise::response_promise()
response_promise::response_promise(local_actor* self, mailbox_element& src) response_promise::response_promise(local_actor* self, mailbox_element& src)
: self_(self), : self_(self),
source_(std::move(src.sender)),
stages_(std::move(src.stages)),
id_(src.mid) { id_(src.mid) {
src.mid.mark_as_answered(); // form an invalid request promise when initialized from a
// response ID, since CAF always drops messages in this case
if (! src.mid.is_response()) {
source_ = std::move(src.sender);
stages_ = std::move(src.stages);
}
} }
void response_promise::deliver(error x) { void response_promise::deliver(error x) {
if (id_.valid()) //if (id_.valid())
deliver_impl(make_message(std::move(x))); deliver_impl(make_message(std::move(x)));
}
bool response_promise::async() const {
return id_.is_async();
} }
void response_promise::deliver_impl(message msg) { void response_promise::deliver_impl(message msg) {
......
...@@ -515,10 +515,10 @@ CAF_TEST(requests) { ...@@ -515,10 +515,10 @@ CAF_TEST(requests) {
s->receive ( s->receive (
on("hi", arg_match) >> [&](actor from) { on("hi", arg_match) >> [&](actor from) {
s->request(from, "whassup?", s).receive( s->request(from, "whassup?", s).receive(
[&](const string& str) -> string { [&](const string& str) {
CAF_CHECK(s->current_sender() != nullptr); CAF_CHECK(s->current_sender() != nullptr);
CAF_CHECK_EQUAL(str, "nothing"); CAF_CHECK_EQUAL(str, "nothing");
return "goodbye!"; s->send(from, "goodbye!");
}, },
after(chrono::minutes(1)) >> [] { after(chrono::minutes(1)) >> [] {
CAF_ERROR("Error in unit test."); CAF_ERROR("Error in unit test.");
......
...@@ -39,9 +39,8 @@ CAF_TEST(simple_ints) { ...@@ -39,9 +39,8 @@ CAF_TEST(simple_ints) {
auto two = on(2) >> [] { }; auto two = on(2) >> [] { };
auto three = on(3) >> [] { }; auto three = on(3) >> [] { };
auto skip_two = [](int i) -> maybe<skip_message_t> { auto skip_two = [](int i) -> maybe<skip_message_t> {
if (i == 2) { if (i == 2)
return skip_message(); return skip_message();
}
return none; return none;
}; };
CAF_CHECK_EQUAL(msg.extract(one), make_message(2, 3)); CAF_CHECK_EQUAL(msg.extract(one), make_message(2, 3));
......
...@@ -122,7 +122,7 @@ public: ...@@ -122,7 +122,7 @@ public:
[=](go_atom, const actor& next) { [=](go_atom, const actor& next) {
request(next, gogo_atom::value).then( request(next, gogo_atom::value).then(
[=](atom_value) { [=](atom_value) {
CAF_MESSAGE("send `ok_atom` to buddy"); CAF_MESSAGE("send 'ok' to buddy");
send(buddy(), ok_atom::value); send(buddy(), ok_atom::value);
quit(); quit();
} }
...@@ -194,13 +194,15 @@ public: ...@@ -194,13 +194,15 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [=] { others >> [=]() -> response_promise {
return request(buddy(), std::move(current_message())).then( auto rp = make_response_promise();
[=](gogogo_atom x) -> gogogo_atom { request(buddy(), std::move(current_message())).then(
[=](gogogo_atom x) mutable {
rp.deliver(x);
quit(); quit();
return x;
} }
); );
return rp;
} }
}; };
} }
...@@ -368,15 +370,14 @@ CAF_TEST(request) { ...@@ -368,15 +370,14 @@ CAF_TEST(request) {
auto await_ok_message = [&] { auto await_ok_message = [&] {
self->receive( self->receive(
[](ok_atom) { [](ok_atom) {
CAF_MESSAGE("received `ok_atom`"); CAF_MESSAGE("received 'ok'");
}, },
[](error_atom) { [](error_atom) {
CAF_ERROR("A didn't receive sync response"); CAF_ERROR("A didn't receive sync response");
}, },
[&](const down_msg& dm) -> maybe<skip_message_t> { [&](const down_msg& dm) -> maybe<skip_message_t> {
if (dm.reason == exit_reason::normal) { if (dm.reason == exit_reason::normal)
return skip_message(); return skip_message();
}
CAF_ERROR("A exited for reason " << to_string(dm.reason)); CAF_ERROR("A exited for reason " << to_string(dm.reason));
return none; return none;
} }
...@@ -398,14 +399,13 @@ CAF_TEST(request) { ...@@ -398,14 +399,13 @@ CAF_TEST(request) {
CAF_MESSAGE("`await_all_other_actors_done` finished"); CAF_MESSAGE("`await_all_other_actors_done` finished");
self->request(self, no_way_atom::value).receive( self->request(self, no_way_atom::value).receive(
[&](int) { [&](int) {
CAF_ERROR("Unexpected message"); CAF_ERROR("unexpected message of type int");
}, },
after(milliseconds(50)) >> [] { after(milliseconds(50)) >> [] {
CAF_MESSAGE("Got timeout"); CAF_MESSAGE("got timeout");
} }
); );
// we should have received two DOWN messages with normal exit reason CAF_MESSAGE("expect two DOWN messages and one 'NoWay'");
// plus 'NoWay'
int i = 0; int i = 0;
self->receive_for(i, 3)( self->receive_for(i, 3)(
[&](const down_msg& dm) { [&](const down_msg& dm) {
...@@ -413,17 +413,16 @@ CAF_TEST(request) { ...@@ -413,17 +413,16 @@ CAF_TEST(request) {
}, },
[](no_way_atom) { [](no_way_atom) {
CAF_MESSAGE("trigger \"actor did not reply to a " CAF_MESSAGE("trigger \"actor did not reply to a "
"synchronous request message\""); "synchronous request message\"");
}, },
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_ERROR("Unexpected timeout"); CAF_ERROR("unexpected timeout");
} }
); );
// mailbox should be empty now CAF_MESSAGE("mailbox should be empty now");
self->receive( self->receive(
others >> [&] { others >> [&] {
CAF_ERROR("Unexpected message: " CAF_ERROR("Unexpected message: "
...@@ -478,11 +477,11 @@ CAF_TEST(request) { ...@@ -478,11 +477,11 @@ CAF_TEST(request) {
anon_send(serv, idle_atom::value, work); anon_send(serv, idle_atom::value, work);
s->request(serv, request_atom::value).receive( s->request(serv, request_atom::value).receive(
[&](response_atom) { [&](response_atom) {
CAF_MESSAGE("received `response_atom`"); CAF_MESSAGE("received 'response'");
CAF_CHECK(s->current_sender() == work); CAF_CHECK(s->current_sender() == work);
}, },
[&](const error& err) { [&](const error& err) {
CAF_ERROR("Error: " << s->system().render(err)); CAF_ERROR("error: " << s->system().render(err));
} }
); );
// first 'request', then 'idle' // first 'request', then 'idle'
...@@ -493,7 +492,7 @@ CAF_TEST(request) { ...@@ -493,7 +492,7 @@ CAF_TEST(request) {
CAF_CHECK(s->current_sender() == work); CAF_CHECK(s->current_sender() == work);
}, },
[&](const error& err) { [&](const error& err) {
CAF_ERROR("Error: " << s->system().render(err)); CAF_ERROR("error: " << s->system().render(err));
} }
); );
s->quit(exit_reason::user_shutdown); s->quit(exit_reason::user_shutdown);
...@@ -503,7 +502,7 @@ CAF_TEST(request) { ...@@ -503,7 +502,7 @@ CAF_TEST(request) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown); CAF_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown);
}, },
others >> [&] { others >> [&] {
CAF_ERROR("Unexpected message: " CAF_ERROR("unexpected message: "
<< to_string(self->current_message())); << to_string(self->current_message()));
} }
); );
......
...@@ -56,24 +56,25 @@ CAF_TEST(test_serial_reply) { ...@@ -56,24 +56,25 @@ CAF_TEST(test_serial_reply) {
auto c3 = self->spawn<linked>(mirror_behavior); auto c3 = self->spawn<linked>(mirror_behavior);
auto c4 = self->spawn<linked>(mirror_behavior); auto c4 = self->spawn<linked>(mirror_behavior);
self->become ( self->become (
[=](hi_atom) -> continue_helper { [=](hi_atom) {
auto rp = self->make_response_promise();
CAF_MESSAGE("received 'hi there'"); CAF_MESSAGE("received 'hi there'");
return self->request(c0, sub0_atom::value).then( self->request(c0, sub0_atom::value).then(
[=](sub0_atom) -> continue_helper { [=](sub0_atom) {
CAF_MESSAGE("received 'sub0'"); CAF_MESSAGE("received 'sub0'");
return self->request(c1, sub1_atom::value).then( self->request(c1, sub1_atom::value).then(
[=](sub1_atom) -> continue_helper { [=](sub1_atom) {
CAF_MESSAGE("received 'sub1'"); CAF_MESSAGE("received 'sub1'");
return self->request(c2, sub2_atom::value).then( self->request(c2, sub2_atom::value).then(
[=](sub2_atom) -> continue_helper { [=](sub2_atom) {
CAF_MESSAGE("received 'sub2'"); CAF_MESSAGE("received 'sub2'");
return self->request(c3, sub3_atom::value).then( self->request(c3, sub3_atom::value).then(
[=](sub3_atom) -> continue_helper { [=](sub3_atom) {
CAF_MESSAGE("received 'sub3'"); CAF_MESSAGE("received 'sub3'");
return self->request(c4, sub4_atom::value).then( self->request(c4, sub4_atom::value).then(
[=](sub4_atom) -> atom_value { [=](sub4_atom) mutable {
CAF_MESSAGE("received 'sub4'"); CAF_MESSAGE("received 'sub4'");
return ho_atom::value; rp.deliver(ho_atom::value);
} }
); );
} }
...@@ -97,5 +98,6 @@ CAF_TEST(test_serial_reply) { ...@@ -97,5 +98,6 @@ CAF_TEST(test_serial_reply) {
CAF_ERROR("Error: " << self->system().render(err)); CAF_ERROR("Error: " << self->system().render(err));
} }
); );
CAF_REQUIRE(self->mailbox().count() == 0);
self->send_exit(master, exit_reason::user_shutdown); self->send_exit(master, exit_reason::user_shutdown);
} }
...@@ -50,6 +50,7 @@ ...@@ -50,6 +50,7 @@
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/message_handler.hpp"
#include "caf/primitive_variant.hpp" #include "caf/primitive_variant.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
......
This diff is collapsed.
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