Commit 98e60ce3 authored by Dominik Charousset's avatar Dominik Charousset

Replace sync_send with request

The streamlined request API offers a unified error handling, where an actor
either only provides a callback for the result or the callback and an error
handler. If no error handler is provided, the actor calls
`self->quit(exit_reason::unhandled_sync_failure)`. There is no replacement for
`on_sync_error`, since the unified error handling superseeds its former
purpose. Closes #365, closes #367, and relates #369.
parent da22be1b
......@@ -26,7 +26,7 @@ behavior mirror(event_based_actor* self) {
void hello_world(event_based_actor* self, const actor& buddy) {
// send "Hello World!" to our buddy ...
self->sync_send(buddy, "Hello World!").then(
self->request(buddy, "Hello World!").then(
// ... wait for a response ...
[=](const string& what) {
// ... and print it
......
......@@ -68,22 +68,22 @@ calculator_actor::behavior_type typed_calculator() {
template <class Handle>
void tester(event_based_actor* self, const Handle& testee, int x, int y) {
self->link_to(testee);
// will be invoked if we receive an unexpected response message
self->on_sync_failure([=] {
aout(self) << "AUT (actor under test) failed" << endl;
self->quit(exit_reason::user_shutdown);
});
// first test: 2 + 1 = 3
self->sync_send(testee, plus_atom::value, x, y).then(
self->request(testee, plus_atom::value, x, y).then(
[=](result_atom, int res1) {
aout(self) << x << " + " << y << " = " << res1 << endl;
self->sync_send(testee, minus_atom::value, x, y).then(
self->request(testee, minus_atom::value, x, y).then(
[=](result_atom, int res2) {
// both tests succeeded
aout(self) << x << " - " << y << " = " << res2 << endl;
self->quit(exit_reason::user_shutdown);
}
);
},
[=](const error& err) {
aout(self) << "AUT (actor under test) failed: "
<< self->system().render(err) << endl;
self->quit(exit_reason::user_shutdown);
}
);
}
......
......@@ -51,16 +51,11 @@ protected:
void tester(event_based_actor* self, const calculator_type& testee) {
self->link_to(testee);
// will be invoked if we receive an unexpected response message
self->on_sync_failure([=] {
aout(self) << "AUT (actor under test) failed" << endl;
self->quit(exit_reason::user_shutdown);
});
// first test: 2 + 1 = 3
self->sync_send(testee, plus_atom::value, 2, 1).then(
self->request(testee, plus_atom::value, 2, 1).then(
[=](result_atom, int r1) {
// second test: 2 - 1 = 1
self->sync_send(testee, minus_atom::value, 2, 1).then(
self->request(testee, minus_atom::value, 2, 1).then(
[=](result_atom, int r2) {
// both tests succeeded
if (r1 == 3 && r2 == 1) {
......@@ -70,6 +65,11 @@ void tester(event_based_actor* self, const calculator_type& testee) {
self->send_exit(testee, exit_reason::user_shutdown);
}
);
},
[=](const error& err) {
aout(self) << "AUT (actor under test) failed: "
<< self->system().render(err) << endl;
self->quit(exit_reason::user_shutdown);
}
);
}
......
......@@ -65,22 +65,24 @@ public:
}
private:
void sync_send_task(atom_value op, int lhs, int rhs) {
on_sync_failure([=] {
aout(this) << "*** sync_failure!" << endl;
});
sync_send(server_, op, lhs, rhs).then(
void request_task(atom_value op, int lhs, int rhs) {
request(server_, op, lhs, rhs).then(
[=](result_atom, int result) {
aout(this) << lhs << (op == plus_atom::value ? " + " : " - ")
<< rhs << " = " << result << endl;
},
[=](const sync_exited_msg& msg) {
aout(this) << "*** server down [" << msg.reason << "], "
<< "try to reconnect ..." << endl;
// try sync_sending this again after successful reconnect
become(keep_behavior,
reconnecting([=] { sync_send_task(op, lhs, rhs); }));
[=](const error& err) {
if (err == sec::request_receiver_down) {
aout(this) << "*** server down, try to reconnect ..." << endl;
// try requesting this again after successful reconnect
become(keep_behavior,
reconnecting([=] { request_task(op, lhs, rhs); }));
return;
}
aout(this) << "*** request resulted in error: "
<< system().render(err) << endl;
}
);
}
......@@ -90,7 +92,7 @@ private:
if (op != plus_atom::value && op != minus_atom::value) {
return;
}
sync_send_task(op, lhs, rhs);
request_task(op, lhs, rhs);
},
[=](rebind_atom, string& nhost, uint16_t nport) {
aout(this) << "*** rebind to " << nhost << ":" << nport << endl;
......
......@@ -162,6 +162,11 @@ protected:
public:
/// @cond PRIVATE
actor_system& home_system() {
CAF_ASSERT(home_system_ != nullptr);
return *home_system_;
}
enum linking_operation {
establish_link_op,
establish_backlink_op,
......
......@@ -29,13 +29,13 @@
#include "caf/behavior_policy.hpp"
#include "caf/response_handle.hpp"
#include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/requester.hpp"
namespace caf {
/// Base type for statically and dynamically typed event-based actors.
/// @tparam BehaviorType Denotes the expected type for become().
/// @tparam HasSyncSend Configures whether this base extends `sync_sender`.
/// @tparam HasSyncSend Configures whether this base extends `requester`.
/// @tparam Base Either `local_actor` (default) or a subtype thereof.
template <class BehaviorType, bool HasSyncSend, class Base = local_actor>
class abstract_event_based_actor : public Base,
......@@ -87,12 +87,12 @@ public:
template <class BehaviorType, class Base>
class abstract_event_based_actor<BehaviorType, true, Base>
: public extend<abstract_event_based_actor<BehaviorType, false, Base>>
::template with<mixin::sync_sender<nonblocking_response_handle_tag>
::template with<mixin::requester<nonblocking_response_handle_tag>
::template impl> {
public:
using super
= typename extend<abstract_event_based_actor<BehaviorType, false, Base>>
::template with<mixin::sync_sender<nonblocking_response_handle_tag>
::template with<mixin::requester<nonblocking_response_handle_tag>
::template impl>;
template <class... Ts>
......
......@@ -27,7 +27,7 @@
#include "caf/mailbox_element.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/shared_spinlock.hpp"
......
......@@ -36,7 +36,7 @@
#include "caf/detail/type_traits.hpp"
#include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/requester.hpp"
namespace caf {
......@@ -45,12 +45,12 @@ namespace caf {
/// @extends local_actor
class blocking_actor
: public extend<local_actor, blocking_actor>::
with<mixin::sync_sender<blocking_response_handle_tag>::impl> {
with<mixin::requester<blocking_response_handle_tag>::impl> {
public:
using behavior_type = behavior;
using super = extend<local_actor, blocking_actor>::
with<mixin::sync_sender<blocking_response_handle_tag>::impl>;
with<mixin::requester<blocking_response_handle_tag>::impl>;
blocking_actor(actor_config& sys);
......
......@@ -64,7 +64,6 @@ using sorted_builtin_types =
strmap, // @strmap
std::set<std::string>, // @strset
std::vector<std::string>, // @strvec
sync_exited_msg, // @sync_exited
sync_timeout_msg, // @sync_timeout
timeout_msg, // @timeout
uint16_t, // @u16
......
......@@ -104,6 +104,14 @@ struct type_checker {
}
};
template <class F>
struct type_checker<message, F> {
static void check() {
// nop
}
};
template <int X, int Pos, class A>
struct static_error_printer {
static_assert(X != Pos, "unexpected handler some position > 20");
......
......@@ -75,7 +75,6 @@ struct down_msg;
struct timeout_msg;
struct group_down_msg;
struct invalid_actor_t;
struct sync_exited_msg;
struct sync_timeout_msg;
struct invalid_actor_addr_t;
struct illegal_message_element;
......
......@@ -322,16 +322,6 @@ public:
/// Creates a `response_promise` to response to a request later on.
response_promise make_response_promise();
/// Sets the handler for unexpected synchronous response messages.
inline void on_sync_failure(std::function<void()> fun) {
sync_failure_handler_ = std::move(fun);
}
/// Checks wheter this actor has a user-defined sync failure handler.
inline bool has_sync_failure_handler() const {
return static_cast<bool>(sync_failure_handler_);
}
/// Sets a custom exception handler for this actor. If multiple handlers are
/// defined, only the functor that was added *last* is being executed.
template <class F>
......@@ -403,12 +393,10 @@ public:
return current_element_;
}
void handle_sync_failure();
template <class Handle, class... Ts>
message_id sync_send_impl(message_priority mp, const Handle& dh, Ts&&... xs) {
message_id request_impl(message_priority mp, const Handle& dh, Ts&&... xs) {
if (! dh) {
throw std::invalid_argument("cannot sync_send to invalid_actor");
throw std::invalid_argument("cannot request to invalid_actor");
}
auto req_id = new_request_id(mp);
send_impl(req_id, actor_cast<abstract_channel*>(dh),
......@@ -538,7 +526,9 @@ public:
behavior& fun,
message_id awaited_response);
using pending_response = std::pair<message_id, behavior>;
using error_handler = std::function<void (error&)>;
using pending_response = std::tuple<message_id, behavior, error_handler>;
message_id new_request_id(message_priority mp);
......@@ -548,9 +538,10 @@ public:
bool awaits(message_id response_id) const;
maybe<behavior&> find_pending_response(message_id mid);
maybe<pending_response&> 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,
error_handler f = nullptr);
behavior& awaited_response_handler();
......@@ -631,8 +622,6 @@ private:
void delayed_send_impl(message_id mid, const channel& whom,
const duration& rtime, message data);
std::function<void()> sync_failure_handler_;
};
/// A smart pointer to a {@link local_actor} instance.
......
......@@ -223,7 +223,9 @@ protected:
template <class F>
class catch_all_match_case : public match_case {
public:
using plain_result_type = typename detail::get_callable_trait<F>::result_type;
using ctrait = typename detail::get_callable_trait<F>::type;
using plain_result_type = typename ctrait::result_type;
using result_type =
typename std::conditional<
......@@ -232,7 +234,7 @@ public:
typename std::remove_const<plain_result_type>::type
>::type;
using arg_types = detail::type_list<>;
using arg_types = typename ctrait::arg_types;
catch_all_match_case(F f)
: match_case(true, detail::make_type_token<>()),
......@@ -240,15 +242,34 @@ public:
// nop
}
match_case::result invoke(maybe<message>& res, message&) override {
match_case::result invoke(maybe<message>& res, message& msg) override {
lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_};
auto fun_res = fun();
arg_types token;
auto fun_res = call_fun(fun, msg, token);
detail::optional_message_visitor omv;
res = omv(fun_res);
return match_case::match;
}
protected:
template <class T>
static auto call_fun(T& f, message&, detail::type_list<>) -> decltype(f()) {
return f();
}
template <class T>
static auto call_fun(T& f, message& x, detail::type_list<const message&>)
-> decltype(f(x)) {
return f(x);
}
template <class T>
static auto call_fun(T& f, message& x, detail::type_list<message&>)
-> decltype(f(x)) {
x.force_unshare();
return f(x);
}
F fun_;
};
......
......@@ -22,6 +22,7 @@
#include <string>
#include <cstdint>
#include <functional>
#include "caf/config.hpp"
#include "caf/message_priority.hpp"
......@@ -144,4 +145,16 @@ inline std::string to_string(const message_id& x) {
} // namespace caf
namespace std {
template <>
struct hash<caf::message_id> {
inline size_t operator()(caf::message_id x) const {
hash<uint64_t> f;
return f(x.integer_value());
}
};
} // namespace std
#endif // CAF_MESSAGE_ID_HPP
......@@ -33,17 +33,17 @@ namespace caf {
namespace mixin {
template <class Base, class Subtype, class HandleTag>
class sync_sender_impl : public Base {
class requester_impl : public Base {
public:
using response_handle_type = response_handle<Subtype, message, HandleTag>;
template <class... Ts>
sync_sender_impl(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
requester_impl(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
/****************************************************************************
* sync_send(...) *
* request(...) *
****************************************************************************/
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
......@@ -52,10 +52,10 @@ public:
/// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Ts>
response_handle_type sync_send(message_priority mp, const actor& dest,
Ts&&... xs) {
response_handle_type request(message_priority mp, const actor& dest,
Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return {dptr()->sync_send_impl(mp, dest, std::forward<Ts>(xs)...),
return {dptr()->request_impl(mp, dest, std::forward<Ts>(xs)...),
dptr()};
}
......@@ -65,8 +65,8 @@ public:
/// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Ts>
response_handle_type sync_send(const actor& dest, Ts&&... xs) {
return sync_send(message_priority::normal, dest, std::forward<Ts>(xs)...);
response_handle_type request(const actor& dest, Ts&&... xs) {
return request(message_priority::normal, dest, std::forward<Ts>(xs)...);
}
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
......@@ -84,7 +84,7 @@ public:
>::type...>
>::type,
HandleTag>
sync_send(message_priority mp, const typed_actor<Sigs...>& dest, Ts&&... xs) {
request(message_priority mp, const typed_actor<Sigs...>& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
detail::type_list<
......@@ -93,7 +93,7 @@ public:
>::type...>;
token tk;
check_typed_input(dest, tk);
return {dptr()->sync_send_impl(mp, dest, std::forward<Ts>(xs)...),
return {dptr()->request_impl(mp, dest, std::forward<Ts>(xs)...),
dptr()};
}
......@@ -112,7 +112,7 @@ public:
>::type...>
>::type,
HandleTag>
sync_send(const typed_actor<Sigs...>& dest, Ts&&... xs) {
request(const typed_actor<Sigs...>& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
detail::type_list<
......@@ -121,13 +121,13 @@ public:
>::type...>;
token tk;
check_typed_input(dest, tk);
return {dptr()->sync_send_impl(message_priority::normal,
return {dptr()->request_impl(message_priority::normal,
dest, std::forward<Ts>(xs)...),
dptr()};
}
/****************************************************************************
* timed_sync_send(...) *
* timed_request(...) *
****************************************************************************/
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`
......@@ -137,10 +137,10 @@ public:
/// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Ts>
response_handle_type timed_sync_send(message_priority mp, const actor& dest,
response_handle_type timed_request(message_priority mp, const actor& dest,
const duration& rtime, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return {dptr()->timed_sync_send_impl(mp, dest, rtime,
return {dptr()->timed_request_impl(mp, dest, rtime,
std::forward<Ts>(xs)...),
dptr()};
}
......@@ -152,9 +152,9 @@ public:
/// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Ts>
response_handle_type timed_sync_send(const actor& dest, const duration& rtime,
response_handle_type timed_request(const actor& dest, const duration& rtime,
Ts&&... xs) {
return timed_sync_send(message_priority::normal, dest, rtime,
return timed_request(message_priority::normal, dest, rtime,
std::forward<Ts>(xs)...);
}
......@@ -173,7 +173,7 @@ public:
>::type...
>::type,
HandleTag>
timed_sync_send(message_priority mp, const typed_actor<Sigs...>& dest,
timed_request(message_priority mp, const typed_actor<Sigs...>& dest,
const duration& rtime, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
......@@ -183,7 +183,7 @@ public:
>::type...>;
token tk;
check_typed_input(dest, tk);
return {dptr()->timed_sync_send_impl(mp, dest, rtime,
return {dptr()->timed_request_impl(mp, dest, rtime,
std::forward<Ts>(xs)...),
dptr()};
}
......@@ -203,9 +203,9 @@ public:
>::type...
>::type,
HandleTag>
timed_sync_send(const typed_actor<Sigs...>& dest, const duration& rtime,
timed_request(const typed_actor<Sigs...>& dest, const duration& rtime,
Ts&&... xs) {
return timed_sync_send(message_priority::normal, dest, rtime,
return timed_request(message_priority::normal, dest, rtime,
std::forward<Ts>(xs)...);
}
......@@ -220,10 +220,10 @@ private:
};
template <class ResponseHandleTag>
class sync_sender {
class requester {
public:
template <class Base, class Subtype>
using impl = sync_sender_impl<Base, Subtype, ResponseHandleTag>;
using impl = requester_impl<Base, Subtype, ResponseHandleTag>;
};
} // namespace mixin
......
......@@ -22,6 +22,7 @@
#include <type_traits>
#include "caf/sec.hpp"
#include "caf/message_id.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/continue_helper.hpp"
......@@ -44,15 +45,29 @@ struct nonblocking_response_handle_tag {};
struct blocking_response_handle_tag {};
/// This helper class identifies an expected response message
/// and enables `sync_send(...).then(...)`.
template <class Self, class ResultOptPairOrMessage, class Tag>
/// and enables `request(...).then(...)`.
template <class Self, class Output, class Tag>
class response_handle;
template <class Output, class F>
struct get_continue_helper {
using type = typed_continue_helper<
typename detail::lifted_result_type<
typename detail::get_callable_trait<F>::result_type
>::type
>;
};
template <class F>
struct get_continue_helper<message, F> {
using type = continue_helper;
};
/******************************************************************************
* nonblocking + untyped *
* nonblocking *
******************************************************************************/
template <class Self>
class response_handle<Self, message, nonblocking_response_handle_tag> {
template <class Self, class Output>
class response_handle<Self, Output, nonblocking_response_handle_tag> {
public:
response_handle() = delete;
response_handle(const response_handle&) = default;
......@@ -62,55 +77,57 @@ public:
// nop
}
template <class... Ts>
continue_helper then(Ts&&... xs) const {
self_->set_response_handler(mid_, behavior{std::forward<Ts>(xs)...});
return {mid_};
using error_handler = std::function<void (error&)>;
template <class F, class T>
typename get_continue_helper<Output, F>::type
then(F f, error_handler ef, timeout_definition<T> tdef) const {
return then_impl(f, ef, std::move(tdef));
}
private:
message_id mid_;
Self* self_;
};
template <class F>
typename get_continue_helper<Output, F>::type
then(F f, error_handler ef = nullptr) const {
return then_impl(f, ef);
}
/******************************************************************************
* nonblocking + typed *
******************************************************************************/
template <class Self, class TypedOutputPair>
class response_handle<Self, TypedOutputPair, nonblocking_response_handle_tag> {
public:
response_handle() = delete;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
template <class F, class T>
typename get_continue_helper<Output, F>::type
then(F f, timeout_definition<T> tdef) const {
return then(std::move(f), nullptr, std::move(tdef));
}
response_handle(message_id mid, Self* self) : mid_(mid), self_(self) {
// nop
void generic_then(std::function<void (message&)> f, error_handler ef) {
behavior tmp{
others >> f
};
self_->set_response_handler(mid_, behavior{std::move(tmp)}, std::move(ef));
}
template <class F>
typed_continue_helper<
typename detail::lifted_result_type<
typename detail::get_callable_trait<F>::result_type
>::type>
then(F fun) {
private:
template <class F, class... Ts>
typename get_continue_helper<Output, F>::type
then_impl(F& f, error_handler& ef, Ts&&... xs) const {
static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context");
detail::type_checker<TypedOutputPair, F>::check();
self_->set_response_handler(mid_, behavior{std::move(fun)});
detail::type_checker<Output, F>::check();
self_->set_response_handler(mid_,
behavior{std::move(f), std::forward<Ts>(xs)...},
std::move(ef));
return {mid_};
}
private:
message_id mid_;
Self* self_;
};
/******************************************************************************
* blocking + untyped *
* blocking *
******************************************************************************/
template <class Self>
class response_handle<Self, message, blocking_response_handle_tag> {
template <class Self, class Output>
class response_handle<Self, Output, blocking_response_handle_tag> {
public:
response_handle() = delete;
response_handle(const response_handle&) = default;
......@@ -120,61 +137,52 @@ public:
// nop
}
void await(behavior& bhvr) {
self_->dequeue(bhvr, mid_);
}
using error_handler = std::function<void (error&)>;
template <class... Ts>
void await(Ts&&... xs) const {
behavior bhvr{std::forward<Ts>(xs)...};
self_->dequeue(bhvr, mid_);
template <class F, class T>
void await(F f, error_handler ef, timeout_definition<T> tdef) {
await_impl(f, ef, std::move(tdef));
}
private:
message_id mid_;
Self* self_;
};
/******************************************************************************
* blocking + typed *
******************************************************************************/
template <class Self, class Output>
class response_handle<Self, Output, blocking_response_handle_tag> {
public:
response_handle() = delete;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
response_handle(message_id mid, Self* self) : mid_(mid), self_(self) {
// nop
template <class F>
void await(F f, error_handler ef = nullptr) {
await_impl(f, ef);
}
template <class F>
void await(F fun) {
static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context");
detail::type_checker<Output, F>::check();
behavior tmp{std::move(fun)};
self_->dequeue(tmp, mid_);
template <class F, class T>
void await(F f, timeout_definition<T> tdef) {
await(std::move(f), nullptr, std::move(tdef));
}
template <class F, class E>
void await(F fun, E error_handler) {
private:
template <class F, class... Ts>
void await_impl(F& f, error_handler& ef, Ts&&... xs) {
static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context");
static_assert(std::is_same<
decltype(error_handler(std::declval<error&>())),
void
>::value,
"error handler accepts no caf::error or returns not void");
detail::type_checker<Output, F>::check();
behavior tmp{std::move(fun), std::move(error_handler)};
behavior tmp;
if (! ef)
tmp.assign(
std::move(f),
others >> [=] {
self_->quit(exit_reason::unhandled_sync_failure);
},
std::forward<Ts>(xs)...
);
else
tmp.assign(
std::move(f),
ef,
others >> [ef] {
error err = sec::unexpected_response;
ef(err);
},
std::forward<Ts>(xs)...
);
self_->dequeue(tmp, mid_);
}
private:
message_id mid_;
Self* self_;
};
......
......@@ -29,8 +29,10 @@ namespace caf {
enum class sec : uint8_t {
/// Indicates that a dynamically typed actor dropped an unexpected message.
unexpected_message = 1,
/// Indicates that a typed actor did not return a promised result.
broken_response_promise,
/// Indicates that a response message did not match the provided handler.
unexpected_response,
/// Indicates that the receiver of a request is no longer alive.
request_receiver_down,
/// Unpublishing failed because the actor is `invalid_actor`.
no_actor_to_unpublish,
/// Unpublishing failed because the actor is not bound to given port.
......
......@@ -60,22 +60,10 @@ inline std::string to_string(const down_msg& x) {
return "down" + deep_to_string(std::tie(x.source, x.reason));
}
/// Sent whenever a terminated actor receives a synchronous request.
struct sync_exited_msg {
/// The source of this message, i.e., the terminated actor.
actor_addr source;
/// The exit reason of the terminated actor.
exit_reason reason;
};
inline std::string to_string(const sync_exited_msg& x) {
return "sync_exited" + deep_to_string(std::tie(x.source, x.reason));
}
template <class T>
typename std::enable_if<
detail::tl_exists<
detail::type_list<exit_msg, down_msg, sync_exited_msg>,
detail::type_list<exit_msg, down_msg>,
detail::tbind<std::is_same, T>::template type
>::value,
bool
......@@ -87,7 +75,7 @@ operator==(const T& lhs, const T& rhs) {
template <class T>
typename std::enable_if<
detail::tl_exists<
detail::type_list<exit_msg, down_msg, sync_exited_msg>,
detail::type_list<exit_msg, down_msg>,
detail::tbind<std::is_same, T>::template type
>::value,
bool
......@@ -99,7 +87,7 @@ operator!=(const T& lhs, const T& rhs) {
template <class IO, class T>
typename std::enable_if<
detail::tl_exists<
detail::type_list<exit_msg, down_msg, sync_exited_msg>,
detail::type_list<exit_msg, down_msg>,
detail::tbind<
std::is_same,
typename std::remove_const<T>::type
......
......@@ -27,7 +27,7 @@
#include "caf/typed_behavior.hpp"
#include "caf/abstract_event_based_actor.hpp"
#include "caf/mixin/sync_sender.hpp"
#include "caf/mixin/requester.hpp"
namespace caf {
......
......@@ -206,8 +206,8 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
binary_serializer bs{self->context(), std::back_inserter(buf)};
self->save_state(bs, 0);
auto sender = node.sender;
// sync_send(...)
auto req = self->sync_send_impl(message_priority::normal, mm,
// request(...)
auto req = self->request_impl(message_priority::normal, mm,
migrate_atom::value, self->name(),
std::move(buf));
self->set_response_handler(req, behavior{
......@@ -316,7 +316,7 @@ response_promise fetch_response_promise(local_actor*, response_promise& hdl) {
return std::move(hdl);
}
// enables `return sync_send(...).then(...)`
// enables `return request(...).then(...)`
bool handle_message_id_res(local_actor* self, message& res,
response_promise hdl) {
CAF_ASSERT(hdl);
......@@ -329,8 +329,8 @@ bool handle_message_id_res(local_actor* self, message& res,
// 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)};
ref_opt->assign(
behavior inner{std::move(std::get<1>(*ref_opt))};
std::get<1>(*ref_opt).assign(
others >> [=] {
// inner is const inside this lambda and mutable a C++14 feature
auto ires = const_cast<behavior&>(inner)(self->current_message());
......@@ -370,20 +370,12 @@ bool post_process_invoke_res(local_actor* self, bool is_sync_request,
} else if (is_sync_request) {
CAF_LOG_DEBUG("report error back to sync caller");
if (res.empty())
res = sec::broken_response_promise;
res = sec::unexpected_response;
rp.deliver(make_message(res.error()));
}
return false;
}
void local_actor::handle_sync_failure() {
CAF_LOG_TRACE("");
if (sync_failure_handler_)
sync_failure_handler_();
else
quit(exit_reason::unhandled_sync_failure);
}
invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
behavior& fun,
message_id awaited_id) {
......@@ -432,7 +424,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
if (! post_process_invoke_res(this, false,
fun(current_element_->msg))) {
CAF_LOG_WARNING("sync failure occured:" << CAF_ARG(id()));
handle_sync_failure();
quit(exit_reason::unhandled_sync_failure);
}
}
ptr.swap(current_element_);
......@@ -502,32 +494,34 @@ bool local_actor::awaits(message_id mid) const {
predicate);
}
maybe<behavior&> local_actor::find_pending_response(message_id mid) {
maybe<local_actor::pending_response&>
local_actor::find_pending_response(message_id mid) {
pending_response_predicate predicate{mid};
auto last = pending_responses_.end();
auto i = std::find_if(pending_responses_.begin(), last, predicate);
if (i != last)
return i->second;
return *i;
return none;
}
void local_actor::set_response_handler(message_id response_id, behavior bhvr) {
void local_actor::set_response_handler(message_id response_id, behavior bhvr,
error_handler f) {
auto opt_ref = find_pending_response(response_id);
if (opt_ref) {
if (bhvr.timeout().valid())
request_sync_timeout_msg(bhvr.timeout(), response_id);
*opt_ref = std::move(bhvr);
get<1>(*opt_ref) = std::move(bhvr);
get<2>(*opt_ref) = std::move(f);
}
}
behavior& local_actor::awaited_response_handler() {
return pending_responses_.front().second;
return get<1>(pending_responses_.front());
}
message_id local_actor::awaited_response_id() {
return pending_responses_.empty()
? message_id::make()
: pending_responses_.front().first;
: get<0>(pending_responses_.front());
}
void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
......@@ -946,7 +940,7 @@ void local_actor::load_state(deserializer&, const unsigned int) {
behavior& local_actor::get_behavior() {
return pending_responses_.empty() ? bhvr_stack_.back()
: pending_responses_.front().second;
: get<1>(pending_responses_.front());
}
bool local_actor::finished() {
......
......@@ -17,6 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/sec.hpp"
#include "caf/atom.hpp"
#include "caf/config.hpp"
#include "caf/actor_cast.hpp"
......@@ -41,7 +42,7 @@ void sync_request_bouncer::operator()(const actor_addr& sender,
if (sender && mid.is_request()) {
auto ptr = actor_cast<abstract_actor_ptr>(sender);
ptr->enqueue(invalid_actor_addr, mid.response_id(),
make_message(sync_exited_msg{sender, rsn}),
make_message(make_error(sec::request_receiver_down)),
// TODO: this breaks out of the execution unit
nullptr);
}
......
......@@ -79,7 +79,6 @@ const char* numbered_type_names[] = {
"@strmap",
"@strset",
"@strvec",
"@sync_exited",
"@sync_timeout",
"@timeout",
"@u16",
......
......@@ -84,7 +84,7 @@ CAF_TEST(round_robin_actor_pool) {
self->send(w, sys_atom::value, put_atom::value, spawn_worker());
std::vector<actor_addr> workers;
for (int i = 0; i < 6; ++i) {
self->sync_send(w, i, i).await(
self->request(w, i, i).await(
[&](int res) {
CAF_CHECK_EQUAL(res, i + i);
auto sender = self->current_sender();
......@@ -99,7 +99,7 @@ CAF_TEST(round_robin_actor_pool) {
return addr == invalid_actor_addr;
};
CAF_CHECK(std::none_of(workers.begin(), workers.end(), is_invalid));
self->sync_send(w, sys_atom::value, get_atom::value).await(
self->request(w, sys_atom::value, get_atom::value).await(
[&](std::vector<actor>& ws) {
std::sort(workers.begin(), workers.end());
std::sort(ws.begin(), ws.end());
......@@ -118,7 +118,7 @@ CAF_TEST(round_robin_actor_pool) {
CAF_CHECK(dm.source == workers.back());
workers.pop_back();
// check whether actor pool removed failed worker
self->sync_send(w, sys_atom::value, get_atom::value).await(
self->request(w, sys_atom::value, get_atom::value).await(
[&](std::vector<actor>& ws) {
std::sort(ws.begin(), ws.end());
CAF_CHECK(workers.size() == ws.size()
......@@ -177,7 +177,7 @@ CAF_TEST(random_actor_pool) {
scoped_actor self{system};
auto w = actor_pool::make(&context, 5, spawn_worker, actor_pool::random());
for (int i = 0; i < 5; ++i) {
self->sync_send(w, 1, 2).await(
self->request(w, 1, 2).await(
[&](int res) {
CAF_CHECK_EQUAL(res, 3);
},
......@@ -212,12 +212,12 @@ CAF_TEST(split_join_actor_pool) {
scoped_actor self{system};
auto w = actor_pool::make(&context, 5, spawn_split_worker,
actor_pool::split_join<int>(join_fun, split_fun));
self->sync_send(w, std::vector<int>{1, 2, 3, 4, 5}).await(
self->request(w, std::vector<int>{1, 2, 3, 4, 5}).await(
[&](int res) {
CAF_CHECK_EQUAL(res, 15);
}
);
self->sync_send(w, std::vector<int>{6, 7, 8, 9, 10}).await(
self->request(w, std::vector<int>{6, 7, 8, 9, 10}).await(
[&](int res) {
CAF_CHECK_EQUAL(res, 40);
}
......
......@@ -134,10 +134,10 @@ testee::behavior_type testee_impl(testee::pointer self) {
};
}
CAF_TEST(sync_send_atom_constants) {
CAF_TEST(request_atom_constants) {
scoped_actor self{system};
auto tst = system.spawn(testee_impl);
self->sync_send(tst, abc_atom::value).await(
self->request(tst, abc_atom::value).await(
[](int i) {
CAF_CHECK_EQUAL(i, 42);
}
......
......@@ -510,13 +510,13 @@ CAF_TEST(spawn_event_testee2_test) {
// exclude this tests; advanced match cases are currently not supported on MSVC
#ifndef CAF_WINDOWS
CAF_TEST(sync_sends) {
CAF_TEST(requests) {
scoped_actor self{system};
auto sync_testee = system.spawn<blocking_api>([](blocking_actor* s) {
s->receive (
on("hi", arg_match) >> [&](actor from) {
s->sync_send(from, "whassup?", s).await(
on_arg_match >> [&](const string& str) -> string {
s->request(from, "whassup?", s).await(
[&](const string& str) -> string {
CAF_CHECK(s->current_sender() != nullptr);
CAF_CHECK_EQUAL(str, "nothing");
return "goodbye!";
......@@ -553,12 +553,16 @@ CAF_TEST(sync_sends) {
}
);
self->await_all_other_actors_done();
self->sync_send(sync_testee, "!?").await(
on<sync_exited_msg>() >> [] {
CAF_MESSAGE("received `sync_exited_msg`");
self->request(sync_testee, "!?").await(
[] {
CAF_TEST_ERROR("Unexpected empty message");
},
others >> [&] {
CAF_TEST_ERROR("Unexpected message");
[&](error& err) {
if (err == sec::request_receiver_down)
CAF_MESSAGE("received `request_receiver_down`");
else
CAF_TEST_ERROR("received unexpected error: "
<< self->system().render(err));
},
after(chrono::microseconds(1)) >> [] {
CAF_TEST_ERROR("Unexpected timeout");
......@@ -608,7 +612,7 @@ typed_testee::behavior_type testee() {
CAF_TEST(typed_await) {
scoped_actor self{system};
auto x = system.spawn(testee);
self->sync_send(x, abc_atom::value).await(
self->request(x, abc_atom::value).await(
[](const std::string& str) {
CAF_CHECK_EQUAL(str, "abc");
}
......@@ -787,7 +791,7 @@ CAF_TEST(move_only_argument) {
};
auto testee = system.spawn(f, std::move(uptr));
scoped_actor self{system};
self->sync_send(testee, 1.f).await(
self->request(testee, 1.f).await(
[](int i) {
CAF_CHECK(i == 42);
}
......
......@@ -83,12 +83,12 @@ CAF_TEST(migrate_locally) {
scoped_actor self{system};
self->send(a, put_atom::value, 42);
// migrate from a to b
self->sync_send(a, sys_atom::value, migrate_atom::value, mm1).await(
self->request(a, sys_atom::value, migrate_atom::value, mm1).await(
[&](ok_atom, const actor_addr& dest) {
CAF_CHECK(dest == b);
}
);
self->sync_send(a, get_atom::value).await(
self->request(a, get_atom::value).await(
[&](int result) {
CAF_CHECK(result == 42);
CAF_CHECK(self->current_sender() == b.address());
......@@ -97,12 +97,12 @@ CAF_TEST(migrate_locally) {
auto mm2 = system.spawn(pseudo_mm, a);
self->send(b, put_atom::value, 23);
// migrate back from b to a
self->sync_send(b, sys_atom::value, migrate_atom::value, mm2).await(
self->request(b, sys_atom::value, migrate_atom::value, mm2).await(
[&](ok_atom, const actor_addr& dest) {
CAF_CHECK(dest == a);
}
);
self->sync_send(b, get_atom::value).await(
self->request(b, get_atom::value).await(
[&](int result) {
CAF_CHECK(result == 23);
CAF_CHECK(self->current_sender() == a.address());
......
......@@ -49,13 +49,13 @@ struct fixture {
void run_testee(actor testee) {
scoped_actor self{system};
self->sync_send(testee, a_atom::value).await([](int i) {
self->request(testee, a_atom::value).await([](int i) {
CAF_CHECK_EQUAL(i, 1);
});
self->sync_send(testee, b_atom::value).await([](int i) {
self->request(testee, b_atom::value).await([](int i) {
CAF_CHECK_EQUAL(i, 2);
});
self->sync_send(testee, c_atom::value).await([](int i) {
self->request(testee, c_atom::value).await([](int i) {
CAF_CHECK_EQUAL(i, 3);
});
self->send_exit(testee, exit_reason::user_shutdown);
......
......@@ -111,7 +111,7 @@ struct fixture {
CAF_REQUIRE(config_server != invalid_actor);
// clear config
scoped_actor self{system};
self->sync_send(config_server, get_atom::value, "*").await(
self->request(config_server, get_atom::value, "*").await(
[&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) {
for (auto& kvp : msgs)
self->send(config_server, put_atom::value, kvp.first, message{});
......@@ -151,7 +151,7 @@ struct fixture {
>::type;
bool result = false;
scoped_actor self{system};
self->sync_send(config_server, get_atom::value, key).await(
self->request(config_server, get_atom::value, key).await(
[&](ok_atom, std::string&, message& msg) {
msg.apply(
[&](type& val) {
......@@ -186,7 +186,7 @@ struct fixture {
if (config_server != invalid_actor) {
size_t result = 0;
scoped_actor self{system};
self->sync_send(config_server, get_atom::value, "*").await(
self->request(config_server, get_atom::value, "*").await(
[&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) {
for (auto& kvp : msgs)
if (! kvp.second.empty())
......
......@@ -19,7 +19,7 @@
#include "caf/config.hpp"
#define CAF_SUITE sync_send
#define CAF_SUITE request
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
......@@ -101,7 +101,7 @@ private:
* *
* A B C *
* | | | *
* | --(sync_send)--> | | *
* | ---(request)---> | | *
* | | --(forward)----> | *
* | X |---\ *
* | | | *
......@@ -120,7 +120,7 @@ public:
behavior make_behavior() override {
return {
[=](go_atom, const actor& next) {
sync_send(next, gogo_atom::value).then(
request(next, gogo_atom::value).then(
[=](atom_value) {
CAF_MESSAGE("send `ok_atom` to buddy");
send(buddy(), ok_atom::value);
......@@ -175,8 +175,8 @@ public:
* *
* A D C *
* | | | *
* | --(sync_send)--> | | *
* | | --(sync_send)--> | *
* | ---(request)---> | | *
* | | ---(request)---> | *
* | | |---\ *
* | | | | *
* | | |<--/ *
......@@ -195,10 +195,10 @@ public:
behavior make_behavior() override {
return {
others >> [=] {
return sync_send(buddy(), std::move(current_message())).then(
others >> [=]() -> message {
return request(buddy(), std::move(current_message())).then(
[=](gogogo_atom x) -> gogogo_atom {
quit();
return std::move(current_message());
return x;
}
);
}
......@@ -272,7 +272,7 @@ CAF_TEST(test_void_res) {
};
});
scoped_actor self{system};
self->sync_send(buddy, 1, 2).await(
self->request(buddy, 1, 2).await(
[] {
CAF_MESSAGE("received void res");
}
......@@ -290,21 +290,20 @@ CAF_TEST(pending_quit) {
};
});
system.spawn([mirror](event_based_actor* self) {
self->sync_send(mirror, 42).then(
others >> [] {
self->request(mirror, 42).then(
[](int) {
CAF_TEST_ERROR("received result, should've been terminated already");
},
[](const error& err) {
CAF_CHECK(err == sec::request_receiver_down);
}
);
self->quit();
});
}
CAF_TEST(sync_send) {
CAF_TEST(request) {
scoped_actor self{system};
self->on_sync_failure([&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message()));
});
self->spawn<monitored + blocking_api>([](blocking_actor* s) {
int invocations = 0;
auto foi = s->spawn<float_or_int, linked>();
......@@ -314,41 +313,39 @@ CAF_TEST(sync_send) {
CAF_CHECK_EQUAL(i, 0);
}
);
s->on_sync_failure([=] {
CAF_TEST_ERROR("sync failure");
});
s->sync_send(foi, i_atom::value).await(
s->request(foi, i_atom::value).await(
[&](int i) {
CAF_CHECK_EQUAL(i, 0);
++invocations;
},
[&](float) {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(s->current_message()));
[&](const error& err) {
CAF_TEST_ERROR("Error: " << s->system().render(err));
}
);
s->sync_send(foi, f_atom::value).await(
[&](int) {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(s->current_message()));
},
s->request(foi, f_atom::value).await(
[&](float f) {
CAF_CHECK_EQUAL(f, 0.f);
++invocations;
},
[&](const error& err) {
CAF_TEST_ERROR("Error: " << s->system().render(err));
}
);
CAF_CHECK_EQUAL(invocations, 2);
CAF_MESSAGE("trigger sync failure");
// provoke invocation of s->handle_sync_failure()
bool sync_failure_called = false;
bool error_handler_called = false;
bool int_handler_called = false;
s->on_sync_failure([&] { sync_failure_called = true; });
s->sync_send(foi, f_atom::value).await(
s->request(foi, f_atom::value).await(
[&](int) {
int_handler_called = true;
},
[&](const error&) {
CAF_MESSAGE("error received");
error_handler_called = true;
}
);
CAF_CHECK_EQUAL(sync_failure_called, true);
CAF_CHECK_EQUAL(error_handler_called, true);
CAF_CHECK_EQUAL(int_handler_called, false);
s->quit(exit_reason::user_shutdown);
});
......@@ -363,7 +360,7 @@ CAF_TEST(sync_send) {
);
auto mirror = system.spawn<sync_mirror>();
bool continuation_called = false;
self->sync_send(mirror, 42).await([&](int value) {
self->request(mirror, 42).await([&](int value) {
continuation_called = true;
CAF_CHECK_EQUAL(value, 42);
});
......@@ -403,10 +400,9 @@ CAF_TEST(sync_send) {
CAF_MESSAGE("block on `await_all_other_actors_done`");
self->await_all_other_actors_done();
CAF_MESSAGE("`await_all_other_actors_done` finished");
self->sync_send(self, no_way_atom::value).await(
others >> [&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message()));
self->request(self, no_way_atom::value).await(
[&](int) {
CAF_TEST_ERROR("Unexpected message");
},
after(milliseconds(50)) >> [] {
CAF_MESSAGE("Got timeout");
......@@ -445,28 +441,27 @@ CAF_TEST(sync_send) {
auto c = self->spawn<C>(); // replies only to 'gogo' messages
// first test: sync error must occur, continuation must not be called
bool timeout_occured = false;
self->on_sync_failure([&] {
CAF_LOG_TRACE("");
CAF_TEST_ERROR("Unexpected sync failure: "
<< to_string(self->current_message()));
});
self->sync_send(c, milliseconds(500), hi_there_atom::value).await(
self->request(c, milliseconds(500), hi_there_atom::value).await(
[&](hi_there_atom) {
CAF_TEST_ERROR("C did reply to 'HiThere'");
},
[&](const error& err) {
CAF_LOG_TRACE("");
CAF_TEST_ERROR("Error: " << self->system().render(err));
},
after(milliseconds(500)) >> [&] {
CAF_MESSAGE("timeout occured");
timeout_occured = true;
}
);
CAF_CHECK_EQUAL(timeout_occured, true);
self->on_sync_failure([&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message()));
});
self->sync_send(c, gogo_atom::value).await(
self->request(c, gogo_atom::value).await(
[](gogogo_atom) {
CAF_MESSAGE("received `gogogo_atom`");
},
[&](const error& err) {
CAF_LOG_TRACE("");
CAF_TEST_ERROR("Error: " << self->system().render(err));
}
);
self->send_exit(c, exit_reason::user_shutdown);
......@@ -485,26 +480,24 @@ CAF_TEST(sync_send) {
});
// first 'idle', then 'request'
anon_send(serv, idle_atom::value, work);
s->sync_send(serv, request_atom::value).await(
[=](response_atom) {
s->request(serv, request_atom::value).await(
[&](response_atom) {
CAF_MESSAGE("received `response_atom`");
CAF_CHECK(s->current_sender() == work);
},
others >> [&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(s->current_message()));
[&](const error& err) {
CAF_TEST_ERROR("Error: " << s->system().render(err));
}
);
// first 'request', then 'idle'
auto handle = s->sync_send(serv, request_atom::value);
auto handle = s->request(serv, request_atom::value);
send_as(work, serv, idle_atom::value, work);
handle.await(
[=](response_atom) {
[&](response_atom) {
CAF_CHECK(s->current_sender() == work);
},
others >> [&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(s->current_message()));
[&](const error& err) {
CAF_TEST_ERROR("Error: " << s->system().render(err));
}
);
s->quit(exit_reason::user_shutdown);
......@@ -530,13 +523,13 @@ behavior snyc_send_no_then_A(event_based_actor * self) {
behavior snyc_send_no_then_B(event_based_actor * self) {
return {
[=](int number) {
self->sync_send(self->spawn(snyc_send_no_then_A), number);
self->request(self->spawn(snyc_send_no_then_A), number);
self->quit();
}
};
}
CAF_TEST(sync_send_no_then) {
CAF_TEST(request_no_then) {
anon_send(system.spawn(snyc_send_no_then_B), 8);
}
......
......@@ -49,7 +49,7 @@ behavior ping1(event_based_actor* self, const actor& pong_actor) {
self->send(self, send_ping_atom::value);
return {
[=](send_ping_atom) {
self->sync_send(pong_actor, ping_atom::value).then(
self->request(pong_actor, ping_atom::value).then(
[=](pong_atom) {
CAF_TEST_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown);
......@@ -69,7 +69,7 @@ behavior ping2(event_based_actor* self, const actor& pong_actor) {
auto received_inner = std::make_shared<bool>(false);
return {
[=](send_ping_atom) {
self->sync_send(pong_actor, ping_atom::value).then(
self->request(pong_actor, ping_atom::value).then(
[=](pong_atom) {
CAF_TEST_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown);
......
......@@ -58,19 +58,19 @@ CAF_TEST(test_serial_reply) {
self->become (
[=](hi_atom) -> continue_helper {
CAF_MESSAGE("received 'hi there'");
return self->sync_send(c0, sub0_atom::value).then(
return self->request(c0, sub0_atom::value).then(
[=](sub0_atom) -> continue_helper {
CAF_MESSAGE("received 'sub0'");
return self->sync_send(c1, sub1_atom::value).then(
return self->request(c1, sub1_atom::value).then(
[=](sub1_atom) -> continue_helper {
CAF_MESSAGE("received 'sub1'");
return self->sync_send(c2, sub2_atom::value).then(
return self->request(c2, sub2_atom::value).then(
[=](sub2_atom) -> continue_helper {
CAF_MESSAGE("received 'sub2'");
return self->sync_send(c3, sub3_atom::value).then(
return self->request(c3, sub3_atom::value).then(
[=](sub3_atom) -> continue_helper {
CAF_MESSAGE("received 'sub3'");
return self->sync_send(c4, sub4_atom::value).then(
return self->request(c4, sub4_atom::value).then(
[=](sub4_atom) -> atom_value {
CAF_MESSAGE("received 'sub4'");
return ho_atom::value;
......@@ -90,12 +90,12 @@ CAF_TEST(test_serial_reply) {
{ // lifetime scope of self
scoped_actor self{system};
CAF_MESSAGE("ID of main: " << self->id());
self->sync_send(master, hi_atom::value).await(
self->request(master, hi_atom::value).await(
[](ho_atom) {
CAF_MESSAGE("received 'ho'");
},
others >> [&] {
CAF_TEST_ERROR("Unexpected message");
[&](const error& err) {
CAF_TEST_ERROR("Error: " << self->system().render(err));
}
);
self->send_exit(master, exit_reason::user_shutdown);
......
......@@ -93,7 +93,7 @@ struct fixture {
self->send(aut, add_atom::value, 7);
self->send(aut, add_atom::value, 4);
self->send(aut, add_atom::value, 9);
self->sync_send(aut, get_atom::value).await(
self->request(aut, get_atom::value).await(
[](int x) {
CAF_CHECK_EQUAL(x, 20);
}
......@@ -110,7 +110,7 @@ struct fixture {
};
});
scoped_actor self{system};
self->sync_send(aut, get_atom::value).await(
self->request(aut, get_atom::value).await(
[&](const string& str) {
CAF_CHECK_EQUAL(str, expected);
}
......
......@@ -87,7 +87,7 @@ CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture)
CAF_TEST(typed_response_promise) {
scoped_actor self{system};
auto foo = self->spawn<foo_actor_impl>();
self->sync_send(foo, get_atom::value, 42).await(
self->request(foo, get_atom::value, 42).await(
[](int x) {
CAF_CHECK_EQUAL(x, 84);
}
......
......@@ -118,10 +118,10 @@ public:
};
void client(event_based_actor* self, actor parent, server_type serv) {
self->sync_send(serv, my_request{0, 0}).then(
self->request(serv, my_request{0, 0}).then(
[=](bool val1) {
CAF_CHECK_EQUAL(val1, true);
self->sync_send(serv, my_request{10, 20}).then(
self->request(serv, my_request{10, 20}).then(
[=](bool val2) {
CAF_CHECK_EQUAL(val2, false);
self->send(parent, passed_atom::value);
......@@ -146,12 +146,12 @@ void test_typed_spawn(server_type ts) {
CAF_CHECK_EQUAL(value, true);
}
);
self->sync_send(ts, my_request{10, 20}).await(
self->request(ts, my_request{10, 20}).await(
[](bool value) {
CAF_CHECK_EQUAL(value, false);
}
);
self->sync_send(ts, my_request{0, 0}).await(
self->request(ts, my_request{0, 0}).await(
[](bool value) {
CAF_CHECK_EQUAL(value, true);
}
......@@ -232,14 +232,14 @@ string_actor::behavior_type string_reverter() {
};
}
// uses `return sync_send(...).then(...)`
// uses `return request(...).then(...)`
string_actor::behavior_type string_relay(string_actor::pointer self,
string_actor master, bool leaf) {
auto next = leaf ? self->spawn(string_relay, master, false) : master;
self->link_to(next);
return {
[=](const string& str) {
return self->sync_send(next, str).then(
return self->request(next, str).then(
[](string& answer) -> string {
return std::move(answer);
}
......@@ -301,7 +301,7 @@ int_actor::behavior_type int_fun() {
behavior foo(event_based_actor* self) {
return {
[=](int i, int_actor server) {
return self->sync_send(server, i).then([=](int result) -> int {
return self->request(server, i).then([=](int result) -> int {
self->quit(exit_reason::normal);
return result;
});
......@@ -329,7 +329,7 @@ int_actor::behavior_type int_fun2(int_actor::pointer self) {
behavior foo2(event_based_actor* self) {
return {
[=](int i, int_actor server) {
return self->sync_send(server, i).then([=](int result) -> int {
return self->request(server, i).then([=](int result) -> int {
self->quit(exit_reason::normal);
return result;
});
......@@ -424,7 +424,7 @@ CAF_TEST(reverter_relay_chain) {
true);
set<string> iface{"caf::replies_to<@str>::with<@str>"};
CAF_CHECK(aut->message_types() == iface);
self->sync_send(aut, "Hello World!").await(
self->request(aut, "Hello World!").await(
[](const string& answer) {
CAF_CHECK_EQUAL(answer, "!dlroW olleH");
}
......@@ -441,7 +441,7 @@ CAF_TEST(string_delegator_chain) {
true);
set<string> iface{"caf::replies_to<@str>::with<@str>"};
CAF_CHECK(aut->message_types() == iface);
self->sync_send(aut, "Hello World!").await(
self->request(aut, "Hello World!").await(
[](const string& answer) {
CAF_CHECK_EQUAL(answer, "!dlroW olleH");
}
......@@ -455,7 +455,7 @@ CAF_TEST(maybe_string_delegator_chain) {
auto aut = system.spawn(maybe_string_delegator,
system.spawn(maybe_string_reverter));
CAF_MESSAGE("send empty string, expect error");
self->sync_send(aut, "").await(
self->request(aut, "").await(
[](ok_atom, const string&) {
throw std::logic_error("unexpected result!");
},
......@@ -466,7 +466,7 @@ CAF_TEST(maybe_string_delegator_chain) {
}
);
CAF_MESSAGE("send abcd string, expect dcba");
self->sync_send(aut, "abcd").await(
self->request(aut, "abcd").await(
[](ok_atom, const string& str) {
CAF_CHECK_EQUAL(str, "dcba");
},
......@@ -528,6 +528,119 @@ CAF_TEST(check_signature) {
);
}
using caf::detail::type_list;
template <class X, class Y>
struct typed_actor_combine_one {
using type = void;
};
template <class... Xs, class... Ys, class... Zs>
struct typed_actor_combine_one<typed_mpi<type_list<Xs...>, type_list<Ys...>>,
typed_mpi<type_list<Ys...>, type_list<Zs...>>> {
using type = typed_mpi<type_list<Xs...>, type_list<Zs...>>;
};
template <class X, class Y>
struct typed_actor_combine_all;
template <class X, class... Ys>
struct typed_actor_combine_all<X, type_list<Ys...>> {
using type = type_list<typename typed_actor_combine_one<X, Ys>::type...>;
};
template <class X, class Y>
struct type_actor_combine;
template <class... Xs, class... Ys>
struct type_actor_combine<typed_actor<Xs...>, typed_actor<Ys...>> {
// store Ys in a packed list
using ys = type_list<Ys...>;
// combine each X with all Ys
using all =
typename detail::tl_concat<
typename typed_actor_combine_all<Xs, ys>::type...
>::type;
// drop all mismatches (void results)
using filtered = typename detail::tl_filter_not_type<all, void>::type;
// throw error if we don't have a single match
static_assert(detail::tl_size<filtered>::value > 0,
"Left-hand actor type does not produce a single result which "
"is valid as input to the right-hand actor type.");
// compute final actor type
using type = typename detail::tl_apply<filtered, typed_actor>::type;
};
template <class... Xs, class... Ys>
typename type_actor_combine<typed_actor<Xs...>, typed_actor<Ys...>>::type
operator*(const typed_actor<Xs...>& x, const typed_actor<Ys...>& y) {
using res_type = typename type_actor_combine<typed_actor<Xs...>,
typed_actor<Ys...>>::type;
if (! x || ! y)
return {};
auto f = [=](event_based_actor* self) -> behavior {
self->link_to(x);
self->link_to(y);
self->trap_exit(true);
auto x_ = actor_cast<actor>(x);
auto y_ = actor_cast<actor>(y);
return {
[=](const exit_msg& msg) {
// also terminate for normal exit reasons
if (msg.source == x || msg.source == y)
self->quit(msg.reason);
},
others >> [=] {
auto rp = self->make_response_promise();
self->request(x_, self->current_message()).generic_then(
[=](message& msg) {
self->request(y_, std::move(msg)).generic_then(
[=](message& msg) {
rp.deliver(std::move(msg));
},
[=](error& err) {
rp.deliver(std::move(err));
}
);
},
[=](error& err) {
rp.deliver(std::move(err));
}
);
}
};
};
return actor_cast<res_type>(x->home_system().spawn(f));
}
using first_stage = typed_actor<replies_to<int>::with<double>>;
using second_stage = typed_actor<replies_to<double>::with<string>>;
first_stage::behavior_type first_stage_impl() {
return [](int i) {
return static_cast<double>(i) * 2;
};
};
second_stage::behavior_type second_stage_impl() {
return [](double x) {
return std::to_string(x);
};
}
CAF_TEST(composition) {
actor_system system;
auto first = system.spawn(first_stage_impl);
auto second = system.spawn(second_stage_impl);
auto first_then_second = first * second;
scoped_actor self{system};
self->request(first_then_second, 42).await(
[](const string& str) {
CAF_MESSAGE("received: " << str);
}
);
}
CAF_TEST_FIXTURE_SCOPE_END()
#endif // CAF_WINDOWS
......@@ -222,8 +222,8 @@ uint16_t middleman::publish(const actor_addr& whom, std::set<std::string> sigs,
uint16_t result;
std::string error_msg;
try {
self->sync_send(mm, publish_atom::value, port,
std::move(whom), std::move(sigs), str, ru).await(
self->request(mm, publish_atom::value, port,
std::move(whom), std::move(sigs), str, ru).await(
[&](ok_atom, uint16_t res) {
result = res;
},
......@@ -269,7 +269,7 @@ maybe<uint16_t> middleman::publish_local_groups(uint16_t port, const char* in) {
void middleman::unpublish(const actor_addr& whom, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
scoped_actor self{system(), true};
self->sync_send(actor_handle(), unpublish_atom::value, whom, port).await(
self->request(actor_handle(), unpublish_atom::value, whom, port).await(
[] {
// ok, basp_broker is done
},
......@@ -285,7 +285,7 @@ actor_addr middleman::remote_actor(std::set<std::string> ifs,
auto mm = actor_handle();
actor_addr result;
scoped_actor self{system(), true};
self->sync_send(mm, connect_atom::value, std::move(host), port).await(
self->request(mm, connect_atom::value, std::move(host), port).await(
[&](ok_atom, const node_id&, actor_addr res, std::set<std::string>& xs) {
CAF_LOG_TRACE(CAF_ARG(res) << CAF_ARG(xs));
if (!res)
......
......@@ -154,13 +154,13 @@ void run_earth(bool use_asio, bool as_server, uint16_t pub_port) {
self->receive_while([&] { return mars_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->sync_send(hdl, sys_atom::value, get_atom::value, "info").await(
self->request(hdl, sys_atom::value, get_atom::value, "info").await(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
mars_addr = addr;
CAF_MESSAGE(CAF_ARG(mars_addr));
self->sync_send(actor_cast<actor>(mars_addr), get_atom::value).await(
self->request(actor_cast<actor>(mars_addr), get_atom::value).await(
[&](uint16_t mp) {
CAF_MESSAGE("mars published its actor at port " << mp);
mars_port = mp;
......@@ -185,7 +185,7 @@ void run_earth(bool use_asio, bool as_server, uint16_t pub_port) {
self->receive_while([&] { return jupiter_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->sync_send(hdl, sys_atom::value, get_atom::value, "info").await(
self->request(hdl, sys_atom::value, get_atom::value, "info").await(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
......
......@@ -538,7 +538,7 @@ CAF_TEST(remote_actor_and_send) {
CAF_REQUIRE(mpx()->pending_scribes().count(make_pair(lo, 4242)) == 1);
auto mm1 = system.middleman().actor_handle();
actor result;
auto f = self()->sync_send(mm1, connect_atom::value,
auto f = self()->request(mm1, connect_atom::value,
lo, uint16_t{4242});
// wait until BASP broker has received and processed the connect message
while (! aut()->valid(remote_hdl(0)))
......
......@@ -181,14 +181,13 @@ void run_server(int argc, char** argv) {
auto serv = system.middleman().spawn_broker(peer_acceptor_fun,
system.spawn(pong));
std::thread child;
self->sync_send(serv, publish_atom::value).await(
self->request(serv, publish_atom::value).await(
[&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port);
child = std::thread([=] { run_client(argc, argv, port); });
},
others >> [&] {
CAF_TEST_ERROR("unexpected message: "
<< to_string(self->current_message()));
[&](const error& err) {
CAF_TEST_ERROR("Error: " << self->system().render(err));
}
);
self->await_all_other_actors_done();
......
......@@ -134,7 +134,7 @@ void spawn5_server_impl(event_based_actor* self, actor client, group grp) {
CAF_MESSAGE("spawned local subscriber: "
<< self->spawn_in_group(grp, reflector)->id());
CAF_MESSAGE("send {'Spawn5'} and await {'ok', actor_vector}");
self->sync_send(client, spawn5_atom::value, grp).then(
self->request(client, spawn5_atom::value, grp).then(
[=](ok_atom, const actor_vector& vec) {
CAF_LOG_TRACE(CAF_ARG(vec));
CAF_MESSAGE("received vector with " << vec.size() << " elements");
......@@ -192,9 +192,8 @@ void spawn5_server_impl(event_based_actor* self, actor client, group grp) {
}
);
},
others >> [=] {
CAF_LOG_TRACE("");
CAF_TEST_ERROR("Unexpected message");
[=](const error& err) {
CAF_TEST_ERROR("Error: " << self->system().render(err));
self->quit(exit_reason::user_shutdown);
},
after(chrono::seconds(10)) >> [=] {
......@@ -215,7 +214,7 @@ void spawn5_server(event_based_actor* self, actor client, bool inverted) {
"foobar"));
} else {
CAF_MESSAGE("request group");
self->sync_send(client, get_group_atom::value).then(
self->request(client, get_group_atom::value).then(
[=](const group& remote_group) {
CAF_LOG_TRACE(CAF_ARG(remote_group));
CAF_REQUIRE(remote_group != invalid_group);
......@@ -309,7 +308,7 @@ private:
void send_sync_msg() {
CAF_LOG_TRACE("");
CAF_MESSAGE("sync send {'SyncMsg', 4.2f}");
sync_send(server_, sync_msg_atom::value, 4.2f).then(
request(server_, sync_msg_atom::value, 4.2f).then(
[=](ok_atom) {
CAF_LOG_TRACE("");
send_foobars();
......@@ -325,7 +324,7 @@ private:
if (i == 100)
test_group_comm();
else {
sync_send(server_, foo_atom::value, bar_atom::value, i).then(
request(server_, foo_atom::value, bar_atom::value, i).then(
[=](foo_atom, bar_atom, int res) {
CAF_LOG_TRACE(CAF_ARG(res));
CAF_CHECK_EQUAL(res, i);
......@@ -338,7 +337,7 @@ private:
void test_group_comm() {
CAF_LOG_TRACE("");
CAF_MESSAGE("test group communication via network");
sync_send(server_, gclient_atom::value).then(
request(server_, gclient_atom::value).then(
[=](gclient_atom, actor gclient) {
CAF_LOG_TRACE(CAF_ARG(gclient));
auto s5a = spawn<monitored>(spawn5_server, gclient, false);
......@@ -484,7 +483,7 @@ private:
void test_group_comm_inverted(actor cptr) {
CAF_LOG_TRACE("");
CAF_MESSAGE("test group communication via network (inverted setup)");
sync_send(cptr, gclient_atom::value).then(
request(cptr, gclient_atom::value).then(
[=](gclient_atom, actor gclient) {
CAF_LOG_TRACE(CAF_ARG(gclient));
await_down(this, spawn<monitored>(spawn5_server, gclient, true), [=] {
......
......@@ -60,10 +60,6 @@ struct server_state {
behavior server(stateful_actor<server_state>* self) {
CAF_LOG_TRACE("");
self->on_sync_failure([=] {
CAF_TEST_ERROR("Unexpected sync response: "
<< to_string(self->current_message()));
});
return {
[=](ok_atom) {
CAF_LOG_TRACE("");
......@@ -72,7 +68,7 @@ behavior server(stateful_actor<server_state>* self) {
CAF_REQUIRE(self->node() != s.node());
self->state.client = actor_cast<actor>(s);
auto mm = self->system().middleman().actor_handle();
self->sync_send(mm, spawn_atom::value,
self->request(mm, spawn_atom::value,
s.node(), "mirror", make_message()).then(
[=](ok_atom, const actor_addr& addr, const std::set<std::string>& ifs) {
CAF_LOG_TRACE(CAF_ARG(addr) << CAF_ARG(ifs));
......@@ -91,6 +87,10 @@ behavior server(stateful_actor<server_state>* self) {
);
}
);
},
[=](const error& err) {
CAF_LOG_TRACE("");
CAF_TEST_ERROR("Error: " << self->system().render(err));
}
};
}
......
......@@ -197,7 +197,7 @@ void run_server(int argc, char** argv) {
scoped_actor self{system};
auto serv = system.middleman().spawn_broker(acceptor_fun, system.spawn(pong));
std::thread child;
self->sync_send(serv, publish_atom::value).await(
self->request(serv, publish_atom::value).await(
[&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port);
child = std::thread([=] {
......
......@@ -94,7 +94,7 @@ void run_client(int argc, char** argv, uint16_t port) {
port);
CAF_REQUIRE(serv);
scoped_actor self{system};
self->sync_send(serv, ping{42})
self->request(serv, ping{42})
.await([](const pong& p) { CAF_CHECK_EQUAL(p.value, 42); });
anon_send_exit(serv, exit_reason::user_shutdown);
self->monitor(serv);
......
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