Commit f4ddeb2b authored by Dominik Charousset's avatar Dominik Charousset

Fix `then` and `await`, close #143

Synchronous response message are handled properly and patterns are no longer
extended automatically. The extension caused CAF to assume `on(any_vals,
arg_match)` before any functor given to `then` or `await`. This feature is
neither documented nor is it the expected behavior from a users point of view.
This could potentially break code, but the broken code shouldn't be valid in
the first place...
parent 1b744245
...@@ -29,6 +29,7 @@ set (LIBCAF_CORE_SRCS ...@@ -29,6 +29,7 @@ set (LIBCAF_CORE_SRCS
src/attachable.cpp src/attachable.cpp
src/behavior.cpp src/behavior.cpp
src/behavior_stack.cpp src/behavior_stack.cpp
src/behavior_impl.cpp
src/binary_deserializer.cpp src/binary_deserializer.cpp
src/binary_serializer.cpp src/binary_serializer.cpp
src/blocking_actor.cpp src/blocking_actor.cpp
......
...@@ -60,20 +60,19 @@ struct is_message_id_wrapper { ...@@ -60,20 +60,19 @@ struct is_message_id_wrapper {
template <class U> template <class U>
static char (&test(...))[2]; static char (&test(...))[2];
static constexpr bool value = sizeof(test<T>(0)) == 1; static constexpr bool value = sizeof(test<T>(0)) == 1;
}; };
template <class T> template <class T>
struct optional_message_visitor_enable_tpl { struct optional_message_visitor_enable_tpl {
using type = typename std::remove_const<T>::type;
static constexpr bool value = static constexpr bool value =
!detail::is_one_of<type, !detail::is_one_of<
none_t, typename std::remove_const<T>::type,
unit_t, none_t,
skip_message_t, unit_t,
optional<skip_message_t>>::value && skip_message_t,
!is_message_id_wrapper<T>::value; optional<skip_message_t>
>::value
&& !is_message_id_wrapper<T>::value;
}; };
struct optional_message_visitor : static_visitor<bhvr_invoke_result> { struct optional_message_visitor : static_visitor<bhvr_invoke_result> {
...@@ -115,13 +114,13 @@ struct optional_message_visitor : static_visitor<bhvr_invoke_result> { ...@@ -115,13 +114,13 @@ struct optional_message_visitor : static_visitor<bhvr_invoke_result> {
} }
template <class... Ts> template <class... Ts>
inline bhvr_invoke_result operator()(std::tuple<Ts...>& value) const { bhvr_invoke_result operator()(std::tuple<Ts...>& value) const {
return detail::apply_args(*this, detail::get_indices(value), value); return detail::apply_args(*this, detail::get_indices(value), value);
} }
// <backward_compatibility version="0.9"> // <backward_compatibility version="0.9">
template <class... Ts> template <class... Ts>
inline bhvr_invoke_result operator()(cow_tuple<Ts...>& value) const { bhvr_invoke_result operator()(cow_tuple<Ts...>& value) const {
return static_cast<message>(std::move(value)); return static_cast<message>(std::move(value));
} }
// </backward_compatibility> // </backward_compatibility>
...@@ -132,89 +131,67 @@ template <class... Ts> ...@@ -132,89 +131,67 @@ template <class... Ts>
struct has_skip_message { struct has_skip_message {
static constexpr bool value = static constexpr bool value =
detail::disjunction<std::is_same<Ts, skip_message_t>::value...>::value; detail::disjunction<std::is_same<Ts, skip_message_t>::value...>::value;
}; };
class behavior_impl : public ref_counted { class behavior_impl : public ref_counted {
public: public:
using pointer = intrusive_ptr<behavior_impl>;
~behavior_impl();
behavior_impl() = default; behavior_impl() = default;
behavior_impl(duration tout);
inline behavior_impl(duration tout) : m_timeout(tout) {}
virtual bhvr_invoke_result invoke(message&) = 0; virtual bhvr_invoke_result invoke(message&) = 0;
virtual bhvr_invoke_result invoke(const message&) = 0; virtual bhvr_invoke_result invoke(const message&) = 0;
inline bhvr_invoke_result invoke(message&& arg) { inline bhvr_invoke_result invoke(message&& arg) {
message tmp(std::move(arg)); message tmp(std::move(arg));
return invoke(tmp); return invoke(tmp);
} }
virtual void handle_timeout(); virtual void handle_timeout();
inline const duration& timeout() const { return m_timeout; }
using pointer = intrusive_ptr<behavior_impl>; inline const duration& timeout() const {
return m_timeout;
}
virtual pointer copy(const generic_timeout_definition& tdef) const = 0; virtual pointer copy(const generic_timeout_definition& tdef) const = 0;
inline pointer or_else(const pointer& other) { pointer or_else(const pointer& other);
CAF_REQUIRE(other != nullptr);
struct combinator : behavior_impl {
pointer first;
pointer second;
bhvr_invoke_result invoke(message& arg) {
auto res = first->invoke(arg);
if (!res) return second->invoke(arg);
return res;
}
bhvr_invoke_result invoke(const message& arg) {
auto res = first->invoke(arg);
if (!res) return second->invoke(arg);
return res;
}
void handle_timeout() {
// the second behavior overrides the timeout handling of
// first behavior
return second->handle_timeout();
}
pointer copy(const generic_timeout_definition& tdef) const {
return new combinator(first, second->copy(tdef));
}
combinator(const pointer& p0, const pointer& p1)
: behavior_impl(p1->timeout()), first(p0), second(p1) {}
};
return new combinator(this, other);
}
private: private:
duration m_timeout; duration m_timeout;
}; };
struct dummy_match_expr { struct dummy_match_expr {
inline variant<none_t> invoke(const message&) const { return none; } inline variant<none_t> invoke(const message&) const {
inline bool can_invoke(const message&) const { return false; } return none;
inline variant<none_t> operator()(const message&) const { return none; } }
inline bool can_invoke(const message&) const {
return false;
}
inline variant<none_t> operator()(const message&) const {
return none;
}
}; };
template <class MatchExpr, typename F> template <class MatchExpr, typename F>
class default_behavior_impl : public behavior_impl { class default_behavior_impl : public behavior_impl {
using super = behavior_impl;
public: public:
template <class Expr> template <class Expr>
default_behavior_impl(Expr&& expr, const timeout_definition<F>& d) default_behavior_impl(Expr&& expr, const timeout_definition<F>& d)
: super(d.timeout) : behavior_impl(d.timeout)
, m_expr(std::forward<Expr>(expr)) , m_expr(std::forward<Expr>(expr))
, m_fun(d.handler) {} , m_fun(d.handler) {}
template <class Expr> template <class Expr>
default_behavior_impl(Expr&& expr, duration tout, F f) default_behavior_impl(Expr&& expr, duration tout, F f)
: super(tout), m_expr(std::forward<Expr>(expr)), m_fun(f) {} : behavior_impl(tout),
m_expr(std::forward<Expr>(expr)),
m_fun(f) {
// nop
}
bhvr_invoke_result invoke(message& tup) { bhvr_invoke_result invoke(message& tup) {
auto res = m_expr(tup); auto res = m_expr(tup);
...@@ -230,17 +207,17 @@ class default_behavior_impl : public behavior_impl { ...@@ -230,17 +207,17 @@ class default_behavior_impl : public behavior_impl {
typename behavior_impl::pointer typename behavior_impl::pointer
copy(const generic_timeout_definition& tdef) const { copy(const generic_timeout_definition& tdef) const {
return new default_behavior_impl<MatchExpr, std::function<void()>>( return new default_behavior_impl<MatchExpr, std::function<void()>>(m_expr,
m_expr, tdef); tdef);
} }
void handle_timeout() { m_fun(); } void handle_timeout() {
m_fun();
}
private: private:
MatchExpr m_expr; MatchExpr m_expr;
F m_fun; F m_fun;
}; };
template <class MatchExpr, typename F> template <class MatchExpr, typename F>
...@@ -250,12 +227,13 @@ new_default_behavior(const MatchExpr& mexpr, duration d, F f) { ...@@ -250,12 +227,13 @@ new_default_behavior(const MatchExpr& mexpr, duration d, F f) {
} }
template <class F> template <class F>
default_behavior_impl<dummy_match_expr, F>* new_default_behavior(duration d, behavior_impl* new_default_behavior(duration d, F f) {
F f) { dummy_match_expr nop;
return new default_behavior_impl<dummy_match_expr, F>(dummy_match_expr{}, d, return new default_behavior_impl<dummy_match_expr, F>(nop, d, std::move(f));
f);
} }
behavior_impl* new_default_behavior(duration d, std::function<void()> fun);
using behavior_impl_ptr = intrusive_ptr<behavior_impl>; using behavior_impl_ptr = intrusive_ptr<behavior_impl>;
// implemented in message_handler.cpp // implemented in message_handler.cpp
......
...@@ -33,7 +33,6 @@ ...@@ -33,7 +33,6 @@
#include "caf/mixin/behavior_stack_based.hpp" #include "caf/mixin/behavior_stack_based.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/response_handle_util.hpp"
namespace caf { namespace caf {
......
...@@ -25,12 +25,11 @@ ...@@ -25,12 +25,11 @@
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/typed_behavior.hpp" #include "caf/typed_behavior.hpp"
#include "caf/continue_helper.hpp" #include "caf/continue_helper.hpp"
#include "caf/system_messages.hpp"
#include "caf/typed_continue_helper.hpp" #include "caf/typed_continue_helper.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/detail/typed_actor_util.hpp" #include "caf/detail/typed_actor_util.hpp"
#include "caf/detail/response_handle_util.hpp"
namespace caf { namespace caf {
...@@ -69,27 +68,22 @@ class response_handle<Self, message, nonblocking_response_handle_tag> { ...@@ -69,27 +68,22 @@ class response_handle<Self, message, nonblocking_response_handle_tag> {
// nop // nop
} }
continue_helper then(behavior bhvr) const { template <class T, class... Ts>
continue_helper then(T arg, Ts&&... args) const {
auto selfptr = m_self;
behavior bhvr{
std::move(arg),
std::forward<Ts>(args)...,
on<sync_timeout_msg>() >> [selfptr]() -> skip_message_t {
selfptr->handle_sync_timeout();
return {};
}
};
m_self->bhvr_stack().push_back(std::move(bhvr), m_mid); m_self->bhvr_stack().push_back(std::move(bhvr), m_mid);
auto ptr = m_self; auto ptr = m_self;
return {m_mid, [ptr](message_id mid) { return ptr->sync_handler(mid); }}; return {m_mid, [ptr](message_id mid) { return ptr->sync_handler(mid); }};
} }
template <class... Cs, class... Ts>
continue_helper then(const match_expr<Cs...>& arg,
const Ts&... args) const {
return then(behavior{arg, args...});
}
template <class... Fs>
typename std::enable_if<
detail::all_callable<Fs...>::value,
continue_helper
>::type
then(Fs... fs) const {
return then(detail::fs2bhvr(m_self, fs...));
}
private: private:
message_id m_mid; message_id m_mid;
Self* m_self; Self* m_self;
...@@ -121,9 +115,17 @@ class response_handle<Self, detail::type_list<Ts...>, ...@@ -121,9 +115,17 @@ class response_handle<Self, detail::type_list<Ts...>,
>::type> >::type>
then(F fun) { then(F fun) {
detail::assert_types<detail::type_list<Ts...>, F>(); detail::assert_types<detail::type_list<Ts...>, F>();
m_self->bhvr_stack().push_back(behavior{on_arg_match >> fun}, m_mid); auto selfptr = m_self;
auto ptr = m_self; behavior tmp{
return {m_mid, [ptr](message_id mid) { return ptr->sync_handler(mid); }}; fun,
on<sync_timeout_msg>() >> [selfptr]() -> skip_message_t {
selfptr->handle_sync_timeout();
return {};
}
};
m_self->bhvr_stack().push_back(std::move(tmp), m_mid);
auto get = [selfptr](message_id mid) { return selfptr->sync_handler(mid); };
return {m_mid, get};
} }
private: private:
...@@ -145,25 +147,22 @@ class response_handle<Self, message, blocking_response_handle_tag> { ...@@ -145,25 +147,22 @@ class response_handle<Self, message, blocking_response_handle_tag> {
// nop // nop
} }
void await(behavior& pfun) { void await(behavior& bhvr) {
m_self->dequeue_response(pfun, m_mid); m_self->dequeue_response(bhvr, m_mid);
}
void await(behavior&& pfun) {
behavior tmp{std::move(pfun)};
await(tmp);
}
template <class... Cs, class... Ts>
void await(const match_expr<Cs...>& arg, const Ts&... args) {
behavior bhvr{arg, args...};
await(bhvr);
} }
template <class... Fs> template <class T, class... Ts>
typename std::enable_if<detail::all_callable<Fs...>::value>::type void await(T arg, Ts&&... args) const {
await(Fs... fs) { auto selfptr = m_self;
await(detail::fs2bhvr(m_self, fs...)); behavior bhvr{
std::move(arg),
std::forward<Ts>(args)...,
on<sync_timeout_msg>() >> [selfptr]() -> skip_message_t {
selfptr->handle_sync_timeout();
return {};
}
};
m_self->dequeue_response(bhvr, m_mid);
} }
private: private:
...@@ -197,15 +196,22 @@ class response_handle<Self, detail::type_list<Ts...>, ...@@ -197,15 +196,22 @@ class response_handle<Self, detail::type_list<Ts...>,
>::type; >::type;
static constexpr size_t fun_args = detail::tl_size<arg_types>::value; static constexpr size_t fun_args = detail::tl_size<arg_types>::value;
static_assert(fun_args <= detail::tl_size<result_types>::value, static_assert(fun_args <= detail::tl_size<result_types>::value,
"functor takes too much arguments"); "functor takes too much arguments");
using recv_types = using recv_types =
typename detail::tl_right< typename detail::tl_right<
result_types, result_types,
fun_args fun_args
>::type; >::type;
static_assert(std::is_same<arg_types, recv_types>::value, static_assert(std::is_same<arg_types, recv_types>::value,
"wrong functor signature"); "wrong functor signature");
behavior tmp = detail::fs2bhvr(m_self, fun); auto selfptr = m_self;
behavior tmp{
fun,
on<sync_timeout_msg>() >> [selfptr]() -> skip_message_t {
selfptr->handle_sync_timeout();
return {};
}
};
m_self->dequeue_response(tmp, m_mid); m_self->dequeue_response(tmp, m_mid);
} }
......
...@@ -17,8 +17,8 @@ ...@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef TIMEOUT_DEFINITION_HPP #ifndef CAF_TIMEOUT_DEFINITION_HPP
#define TIMEOUT_DEFINITION_HPP #define CAF_TIMEOUT_DEFINITION_HPP
#include <functional> #include <functional>
...@@ -27,20 +27,25 @@ ...@@ -27,20 +27,25 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
class behavior_impl; class behavior_impl;
}
behavior_impl* new_default_behavior(duration d, std::function<void()> fun);
} // namespace detail
template <class F> template <class F>
struct timeout_definition { struct timeout_definition {
static constexpr bool may_have_timeout = true; static constexpr bool may_have_timeout = true;
duration timeout; duration timeout;
F handler; F handler;
detail::behavior_impl* as_behavior_impl() const; detail::behavior_impl* as_behavior_impl() const {
return detail::new_default_behavior(timeout, handler);
}
}; };
using generic_timeout_definition = timeout_definition<std::function<void()>>; using generic_timeout_definition = timeout_definition<std::function<void()>>;
} // namespace caf } // namespace caf
#endif // TIMEOUT_DEFINITION_HPP #endif // CAF_TIMEOUT_DEFINITION_HPP
...@@ -17,33 +17,71 @@ ...@@ -17,33 +17,71 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_DETAIL_RESPONSE_FUTURE_DETAIL_HPP #include "caf/detail/behavior_impl.hpp"
#define CAF_DETAIL_RESPONSE_FUTURE_DETAIL_HPP
#include "caf/on.hpp" #include "caf/message_handler.hpp"
#include "caf/skip_message.hpp"
#include "caf/system_messages.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Actor, class... Fs> namespace {
behavior fs2bhvr(Actor* self, Fs... fs) {
auto handle_sync_timeout = [self]() -> skip_message_t { class combinator : public behavior_impl {
self->handle_sync_timeout(); public:
return {}; bhvr_invoke_result invoke(message& arg) {
}; auto res = first->invoke(arg);
return behavior{ if (!res) return second->invoke(arg);
on<sync_timeout_msg>() >> handle_sync_timeout, return res;
on<unit_t>() >> skip_message, }
on<sync_exited_msg>() >> skip_message,
(on(any_vals, arg_match) >> std::move(fs))... bhvr_invoke_result invoke(const message& arg) {
}; auto res = first->invoke(arg);
if (!res) return second->invoke(arg);
return res;
}
void handle_timeout() {
// the second behavior overrides the timeout handling of
// first behavior
return second->handle_timeout();
}
pointer copy(const generic_timeout_definition& tdef) const {
return new combinator(first, second->copy(tdef));
}
combinator(const pointer& p0, const pointer& p1)
: behavior_impl(p1->timeout()),
first(p0),
second(p1) {
// nop
}
private:
pointer first;
pointer second;
};
} // namespace <anonymous>
behavior_impl::~behavior_impl() {
// nop
}
behavior_impl::behavior_impl(duration tout) : m_timeout(tout) {
// nop
}
behavior_impl::pointer behavior_impl::or_else(const pointer& other) {
CAF_REQUIRE(other != nullptr);
return new combinator(this, other);
}
behavior_impl* new_default_behavior(duration d, std::function<void()> fun) {
using impl = default_behavior_impl<dummy_match_expr, std::function<void()>>;
dummy_match_expr nop;
return new impl(nop, d, std::move(fun));
} }
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_DETAIL_RESPONSE_FUTURE_DETAIL_HPP
...@@ -9,154 +9,175 @@ using namespace caf; ...@@ -9,154 +9,175 @@ using namespace caf;
namespace { namespace {
struct sync_mirror : sb_actor<sync_mirror> { struct sync_mirror : event_based_actor {
behavior make_behavior() override {
behavior init_state; return {
others() >> [=] {
sync_mirror() { return last_dequeued();
init_state = (others() >> [=] { return 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 : event_based_actor {
behavior init_state = (on(atom("f")) >> [] { return 0.0f; }, behavior make_behavior() override {
on(atom("i")) >> [] { return 0; }); return {
on(atom("f")) >> [] {
return 0.0f;
},
on(atom("i")) >> [] {
return 0;
}
};
}
}; };
struct popular_actor : event_based_actor { // popular actors have a buddy struct popular_actor : event_based_actor { // popular actors have a buddy
actor m_buddy; actor m_buddy;
popular_actor(const actor& buddy) : m_buddy(buddy) {} popular_actor(const actor& buddy) : m_buddy(buddy) {
inline const actor& buddy() const { return m_buddy; } // nop
}
inline const actor& buddy() const {
return m_buddy;
}
void report_failure() { void report_failure() {
send(buddy(), atom("failure")); send(buddy(), atom("failure"));
quit(); quit();
} }
}; };
/******************************************************************************\ /******************************************************************************\
* test case 1: * * test case 1: *
* * * *
* A B C * * A B C *
* | | | * * | | | *
* | --(sync_send)--> | | * * | --(sync_send)--> | | *
* | | --(forward)----> | * * | | --(forward)----> | *
* | X |---\ * * | X |---\ *
* | | | * * | | | *
* | |<--/ * * | |<--/ *
* | <-------------(reply)-------------- | * * | <-------------(reply)-------------- | *
* X X * * X X *
\ ******************************************************************************/ \ ******************************************************************************/
struct A : popular_actor { struct A : popular_actor {
A(const actor& buddy) : popular_actor(buddy) {} A(const actor& buddy) : popular_actor(buddy) {
// nop
}
behavior make_behavior() override { behavior make_behavior() override {
return (on(atom("go"), arg_match) >> [=](const actor& next) { return {
CAF_CHECKPOINT(); on(atom("go"), arg_match) >> [=](const actor& next) {
sync_send(next, atom("gogo")).then([=] { CAF_CHECKPOINT();
sync_send(next, atom("gogo")).then(
[=](atom_value) {
CAF_CHECKPOINT(); CAF_CHECKPOINT();
send(buddy(), atom("success")); send(buddy(), atom("success"));
quit(); quit();
}); }
}, );
others() >> [=] { report_failure(); }); },
others() >> [=] {
report_failure();
}
};
} }
}; };
struct B : popular_actor { struct B : popular_actor {
B(const actor& buddy) : popular_actor(buddy) {} B(const actor& buddy) : popular_actor(buddy) {
// nop
}
behavior make_behavior() override { behavior make_behavior() override {
return (others() >> [=] { return {
CAF_CHECKPOINT(); others() >> [=] {
forward_to(buddy()); CAF_CHECKPOINT();
quit(); forward_to(buddy());
}); quit();
}
};
} }
}; };
struct C : sb_actor<C> { struct C : event_based_actor {
behavior init_state; behavior make_behavior() override {
C() { return {
init_state = (on(atom("gogo")) >> [=]()->atom_value { on(atom("gogo")) >> [=]() -> atom_value {
CAF_CHECKPOINT(); CAF_CHECKPOINT();
quit(); quit();
return atom("gogogo"); return atom("gogogo");
}); }
};
} }
}; };
/******************************************************************************\ /******************************************************************************\
* test case 2: * * test case 2: *
* * * *
* A D C * * A D C *
* | | | * * | | | *
* | --(sync_send)--> | | * * | --(sync_send)--> | | *
* | | --(sync_send)--> | * * | | --(sync_send)--> | *
* | | |---\ * * | | |---\ *
* | | | | * * | | | | *
* | | |<--/ * * | | |<--/ *
* | | <---(reply)----- | * * | | <---(reply)----- | *
* | <---(reply)----- | * * | <---(reply)----- | *
* X X * * X X *
\ ******************************************************************************/ \ ******************************************************************************/
struct D : popular_actor { struct D : popular_actor {
D(const actor& buddy) : popular_actor(buddy) {} D(const actor& buddy) : popular_actor(buddy) {}
behavior make_behavior() override { behavior make_behavior() override {
return (others() >> [=] { return {
/* others() >> [=] {
response_promise handle = make_response_promise(); return sync_send_tuple(buddy(), last_dequeued()).then(
sync_send_tuple(buddy(), last_dequeued()).then([=] { others() >> [=]() -> message {
reply_to(handle, last_dequeued()); quit();
quit(); return last_dequeued();
}); }
//*/ );
return sync_send_tuple(buddy(), last_dequeued()) }
.then([=]()->message { };
quit();
return last_dequeued();
}); //*/
});
} }
}; };
/******************************************************************************\ /******************************************************************************\
* test case 3: * * test case 3: *
* * * *
* Client Server Worker * * Client Server Worker *
* | | | * * | | | *
* | | <---(idle)------ | * * | | <---(idle)------ | *
* | ---(request)---> | | * * | ---(request)---> | | *
* | | ---(request)---> | * * | | ---(request)---> | *
* | | |---\ * * | | |---\ *
* | X | | * * | X | | *
* | |<--/ * * | |<--/ *
* | <------------(response)------------ | * * | <------------(response)------------ | *
* X * * X *
\ ******************************************************************************/ \ ******************************************************************************/
struct server : event_based_actor { struct server : event_based_actor {
behavior make_behavior() override { behavior make_behavior() override {
auto die = [=] { quit(exit_reason::user_shutdown); }; auto die = [=] {
return (on(atom("idle"), arg_match) >> [=](actor worker) { quit(exit_reason::user_shutdown);
become(keep_behavior, };
on(atom("request")) >> [=] { return {
forward_to(worker); on(atom("idle"), arg_match) >> [=](actor worker) {
unbecome(); // await next idle message become(
}, keep_behavior,
on(atom("idle")) >> skip_message, others() >> die); on(atom("request")) >> [=] {
}, forward_to(worker);
on(atom("request")) >> skip_message, others() >> die); unbecome(); // await next idle message
},
on(atom("idle")) >> skip_message,
others() >> die
);
},
on(atom("request")) >> skip_message,
others() >> die
};
} }
}; };
void test_sync_send() { void test_sync_send() {
...@@ -177,17 +198,24 @@ void test_sync_send() { ...@@ -177,17 +198,24 @@ void test_sync_send() {
s->on_sync_failure([=] { s->on_sync_failure([=] {
CAF_FAILURE("received: " << to_string(s->last_dequeued())); CAF_FAILURE("received: " << to_string(s->last_dequeued()));
}); });
s->sync_send(foi, atom("i")) s->sync_send(foi, atom("i")).await(
.await([&](int i) { [&](int i) {
CAF_CHECK_EQUAL(i, 0); CAF_CHECK_EQUAL(i, 0);
++invocations; ++invocations;
}, },
[&](float) { CAF_UNEXPECTED_MSG(s); }); [&](float) {
s->sync_send(foi, atom("f")) CAF_UNEXPECTED_MSG(s);
.await([&](int) { CAF_UNEXPECTED_MSG(s); }, [&](float f) { }
CAF_CHECK_EQUAL(f, 0.f); );
++invocations; s->sync_send(foi, atom("f")).await(
}); [&](int) {
CAF_UNEXPECTED_MSG(s);
},
[&](float f) {
CAF_CHECK_EQUAL(f, 0.f);
++invocations;
}
);
CAF_CHECK_EQUAL(invocations, 2); CAF_CHECK_EQUAL(invocations, 2);
CAF_PRINT("trigger sync failure"); CAF_PRINT("trigger sync failure");
// provoke invocation of s->handle_sync_failure() // provoke invocation of s->handle_sync_failure()
...@@ -216,34 +244,35 @@ void test_sync_send() { ...@@ -216,34 +244,35 @@ void test_sync_send() {
self->send_exit(mirror, exit_reason::user_shutdown); self->send_exit(mirror, exit_reason::user_shutdown);
self->await_all_other_actors_done(); self->await_all_other_actors_done();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
auto non_normal_down_msg = [](down_msg dm)->optional<down_msg> { auto non_normal_down_msg = [](down_msg dm) -> optional<down_msg> {
if (dm.reason != exit_reason::normal) return dm; if (dm.reason != exit_reason::normal) {
return dm;
}
return none; return none;
}; };
auto await_success_message = [&] { auto await_success_message = [&] {
self->receive(on(atom("success")) >> CAF_CHECKPOINT_CB(), self->receive(
on(atom("failure")) >> on(atom("success")) >> CAF_CHECKPOINT_CB(),
CAF_FAILURE_CB("A didn't receive sync response"), on(atom("failure")) >> CAF_FAILURE_CB("A didn't receive sync response"),
on(non_normal_down_msg) >> [&](const down_msg& dm) { on(non_normal_down_msg) >> [&](const down_msg& dm) {
CAF_FAILURE("A exited for reason " << dm.reason); CAF_FAILURE("A exited for reason " << dm.reason);
}); }
);
}; };
self->send(self->spawn<A, monitored>(self), atom("go"), self->send(self->spawn<A, monitored>(self), atom("go"), spawn<B>(spawn<C>()));
spawn<B>(spawn<C>()));
await_success_message(); await_success_message();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
self->await_all_other_actors_done(); self->await_all_other_actors_done();
self->send(self->spawn<A, monitored>(self), atom("go"), self->send(self->spawn<A, monitored>(self), atom("go"), spawn<D>(spawn<C>()));
spawn<D>(spawn<C>()));
await_success_message(); await_success_message();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
self->await_all_other_actors_done(); self->await_all_other_actors_done();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
self->timed_sync_send(self, std::chrono::milliseconds(50), atom("NoWay")) self->timed_sync_send(self, std::chrono::milliseconds(50), atom("NoWay"))
.await(on<sync_timeout_msg>() >> CAF_CHECKPOINT_CB(), .await(
others() >> CAF_UNEXPECTED_MSG_CB_REF(self)); on<sync_timeout_msg>() >> CAF_CHECKPOINT_CB(),
others() >> CAF_UNEXPECTED_MSG_CB_REF(self)
);
// we should have received two DOWN messages with normal exit reason // we should have received two DOWN messages with normal exit reason
// plus 'NoWay' // plus 'NoWay'
int i = 0; int i = 0;
...@@ -253,9 +282,8 @@ void test_sync_send() { ...@@ -253,9 +282,8 @@ void test_sync_send() {
}, },
on(atom("NoWay")) >> [] { on(atom("NoWay")) >> [] {
CAF_CHECKPOINT(); CAF_CHECKPOINT();
CAF_PRINT( CAF_PRINT("trigger \"actor did not reply to a "
"trigger \"actor did not reply to a " "synchronous request message\"");
"synchronous request message\"");
}, },
others() >> CAF_UNEXPECTED_MSG_CB_REF(self), others() >> CAF_UNEXPECTED_MSG_CB_REF(self),
after(std::chrono::seconds(0)) >> CAF_UNEXPECTED_TOUT_CB()); after(std::chrono::seconds(0)) >> CAF_UNEXPECTED_TOUT_CB());
......
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