Commit a000469e authored by Lingxi-Li's avatar Lingxi-Li

Redesign actor decorators

1) Give decorators their own identities
2) Fix dot composition behavior to match `f.g(x)=f(g(x))`
3) Make decorators depend on decorated actors, but not vice versa
4) Factor decorator unit tests
parent 60c07ea1
......@@ -23,25 +23,29 @@
#include "caf/message.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/monitorable_actor.hpp"
namespace caf {
/// An actor decorator implementing `std::bind`-like compositions.
class bound_actor : public abstract_actor {
/// Bound actors are hidden actors. A bound actor exits when its
/// decorated actor exits. The decorated actor has no dependency
/// on the bound actor by default, and exit of a bound actor has
/// no effect on the decorated actor. Bound actors are hosted on
/// the same actor system and node as decorated actors.
class bound_actor : public monitorable_actor {
public:
bound_actor(actor_system* sys, actor_addr decorated, message msg);
void attach(attachable_ptr ptr) override;
size_t detach(const attachable::token& what) override;
bound_actor(actor_addr decorated, message msg);
// non-system messages are processed and then forwarded;
// system messages are handled and consumed on the spot;
// in either case, the processing is done synchronously
void enqueue(mailbox_element_ptr what, execution_unit* host) override;
protected:
bool link_impl(linking_operation op, const actor_addr& other) override;
private:
void handle_system_message(const message& msg, execution_unit* host);
static bool is_system_message(const message& msg);
actor_addr decorated_;
message merger_;
};
......
......@@ -20,25 +20,39 @@
#ifndef CAF_COMPOSED_ACTOR_HPP
#define CAF_COMPOSED_ACTOR_HPP
#include "caf/local_actor.hpp"
#include "caf/actor_addr.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/monitorable_actor.hpp"
namespace caf {
/// An actor decorator implementing "dot operator"-like compositions,
/// i.e., `f.g(x) = f(g(x))`.
class composed_actor : public local_actor {
/// i.e., `f.g(x) = f(g(x))`. Composed actors are hidden actors.
/// A composed actor exits when either of its constituent actors exits;
/// Constituent actors have no dependency on the composed actor
/// by default, and exit of a composed actor has no effect on its
/// constituent actors. A composed actor is hosted on the same actor
/// system and node as `g`, the first actor on the forwarding chain.
class composed_actor : public monitorable_actor {
public:
composed_actor(actor_system* sys, actor_addr first, actor_addr second);
using message_types_set = std::set<std::string>;
void initialize() override;
composed_actor(actor_addr f, actor_addr g, message_types_set msg_types);
// non-system messages are processed and then forwarded;
// system messages are handled and consumed on the spot;
// in either case, the processing is done synchronously
void enqueue(mailbox_element_ptr what, execution_unit* host) override;
message_types_set message_types() const override;
private:
bool is_system_message(const message& msg);
void handle_system_message(const message& msg, execution_unit* host);
static bool is_system_message(const message& msg);
actor_addr first_;
actor_addr second_;
actor_addr f_;
actor_addr g_;
message_types_set msg_types_;
};
} // namespace caf
......
......@@ -27,6 +27,7 @@
#include "caf/replies_to.hpp"
#include "caf/bound_actor.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/composed_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/typed_response_promise.hpp"
......@@ -214,7 +215,7 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
bind(Ts&&... xs) const {
if (! ptr_)
return invalid_actor;
auto ptr = make_counted<bound_actor>(&ptr_->home_system(), ptr_->address(),
auto ptr = make_counted<bound_actor>(ptr_->address(),
make_message(xs...));
return {ptr.release(), false};
}
......@@ -287,18 +288,21 @@ bool operator!=(const typed_actor<Xs...>& x,
template <class... Xs, class... Ys>
typename detail::mpi_composition<
typed_actor,
detail::type_list<Ys...>,
Xs...
detail::type_list<Xs...>,
Ys...
>::type
operator*(typed_actor<Xs...> f, typed_actor<Ys...> g) {
using result =
typename detail::mpi_composition<
typed_actor,
detail::type_list<Ys...>,
Xs...
detail::type_list<Xs...>,
Ys...
>::type;
return actor_cast<result>(actor_cast<actor>(std::move(f))
* actor_cast<actor>(std::move(g)));
auto ptr = make_counted<composed_actor>(f.address(),
g.address(),
g->home_system().message_types(
result{}));
return actor_cast<result>(std::move(ptr));
}
/// @relates typed_actor
......
......@@ -78,16 +78,17 @@ actor_id actor::id() const noexcept {
actor actor::bind_impl(message msg) const {
if (! ptr_)
return invalid_actor;
return actor_cast<actor>(make_counted<bound_actor>(&ptr_->home_system(),
address(),
return actor_cast<actor>(make_counted<bound_actor>(address(),
std::move(msg)));
}
actor operator*(actor f, actor g) {
if (! f || ! g)
return invalid_actor;
auto ptr = make_counted<composed_actor>(&f->home_system(),
f.address(), g.address());
auto ptr = make_counted<composed_actor>(f.address(),
g.address(),
g->home_system().message_types(
actor{}));
return actor_cast<actor>(std::move(ptr));
}
......
......@@ -20,47 +20,95 @@
#include "caf/bound_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/default_attachable.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/system_messages.hpp"
#include "caf/detail/disposer.hpp"
#include "caf/detail/merged_tuple.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf {
bound_actor::bound_actor(actor_system* sys, actor_addr decorated, message msg)
: abstract_actor(sys, decorated->id(), decorated->node(),
is_abstract_actor_flag | is_actor_bind_decorator_flag),
decorated_(std::move(decorated)),
merger_(std::move(msg)) {
// nop
bound_actor::bound_actor(actor_addr decorated, message msg)
: monitorable_actor{ &decorated->home_system(),
decorated->home_system().next_actor_id(),
decorated->node(),
is_abstract_actor_flag
| is_actor_bind_decorator_flag },
decorated_{ std::move(decorated) },
merger_{ std::move(msg) } {
// bound actor has dependency on the decorated actor by default;
// if the decorated actor is already dead upon establishing the
// dependency, the actor is spawned dead
decorated_->attach(default_attachable::make_monitor(address()));
}
void bound_actor::attach(attachable_ptr ptr) {
decorated_->attach(std::move(ptr));
}
size_t bound_actor::detach(const attachable::token& what) {
return decorated_->detach(what);
void bound_actor::enqueue(mailbox_element_ptr what,
execution_unit* host) {
if (! what)
return; // not even an empty message
auto reason = exit_reason_.load();
if (reason != exit_reason::not_exited) {
// actor has exited
auto& mid = what->mid;
if (mid.is_request()) {
// make sure that a request always gets a response;
// the exit reason reflects the first actor on the
// forwarding chain that is out of service
detail::sync_request_bouncer rb{reason};
rb(what->sender, mid);
}
return;
}
if (is_system_message(what->msg)) {
// handle and consume the system message;
// the only effect that MAY result from handling a system message
// is to exit the actor if it hasn't exited already;
// `handle_system_message()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done
handle_system_message(what->msg, host);
} else {
// process and forward the non-system message
auto& msg = what->msg;
message tmp{ detail::merged_tuple::make(merger_, std::move(msg)) };
msg.swap(tmp);
decorated_->enqueue(std::move(what), host);
}
}
void bound_actor::enqueue(mailbox_element_ptr what, execution_unit* host) {
// ignore system messages
auto& msg = what->msg;
if (! ((msg.size() == 1 && (msg.match_element<exit_msg>(0)
|| msg.match_element<down_msg>(0)))
|| (msg.size() > 1 && msg.match_element<sys_atom>(0)))) {
// merge non-system messages
message tmp{detail::merged_tuple::make(merger_, std::move(msg))};
msg.swap(tmp);
void bound_actor::handle_system_message(const message& msg,
execution_unit* host) {
// `monitorable_actor::cleanup()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done;
// handles only `down_msg` from the decorated actor and `exit_msg` from anyone
if (msg.size() != 1)
return; // neither case
if (msg.match_element<down_msg>(0)) {
auto& dm = msg.get_as<down_msg>(0);
CAF_ASSERT(dm.reason != exit_reason::not_exited);
if (dm.source == decorated_) {
// decorated actor has exited, so exits
// the bound actor with the same reason
monitorable_actor::cleanup(dm.reason, host);
}
return;
}
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
CAF_ASSERT(em.reason != exit_reason::not_exited);
// exit message received, so exits with the same reason
monitorable_actor::cleanup(em.reason, host);
return;
}
decorated_->enqueue(std::move(what), host);
// drop other cases
}
bool bound_actor::link_impl(linking_operation op, const actor_addr& other) {
if (actor_cast<abstract_actor*>(other) == this || other == decorated_)
return false;
return decorated_->link_impl(op, other);
bool bound_actor::is_system_message(const message& msg) {
return (msg.size() == 1 && (msg.match_element<exit_msg>(0)
|| msg.match_element<down_msg>(0)))
|| (msg.size() > 1 && msg.match_element<sys_atom>(0));
}
} // namespace caf
......@@ -19,53 +19,99 @@
#include "caf/composed_actor.hpp"
namespace caf {
#include "caf/actor_system.hpp"
#include "caf/default_attachable.hpp"
composed_actor::composed_actor(actor_system* sys,
actor_addr first, actor_addr second)
: local_actor(sys, first->id(), first->node(),
is_abstract_actor_flag | is_actor_dot_decorator_flag),
first_(first),
second_(second) {
// nop
}
#include "caf/detail/disposer.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf {
void composed_actor::initialize() {
trap_exit(true);
link_to(first_);
link_to(second_);
do_become(
{
[=](const exit_msg& em) {
if (em.source == first_ || em.source == second_)
quit(em.reason);
else if (em.reason != exit_reason::normal)
quit(em.reason);
},
others >> [] {
// drop
}
},
true
);
composed_actor::composed_actor(actor_addr f, actor_addr g,
message_types_set msg_types)
: monitorable_actor{ &g->home_system(),
g->home_system().next_actor_id(),
g->node(),
is_abstract_actor_flag
| is_actor_dot_decorator_flag },
f_{ std::move(f) },
g_{ std::move(g) },
msg_types_{ std::move(msg_types) } {
// composed actor has dependency on constituent actors by default;
// if either constituent actor is already dead upon establishing
// the dependency, the actor is spawned dead
f_->attach(default_attachable::make_monitor(address()));
if (g_ != f_)
g_->attach(default_attachable::make_monitor(address()));
}
void composed_actor::enqueue(mailbox_element_ptr what, execution_unit* host) {
void composed_actor::enqueue(mailbox_element_ptr what,
execution_unit* host) {
if (! what)
return; // not even an empty message
auto reason = exit_reason_.load();
if (reason != exit_reason::not_exited) {
// actor has exited
auto& mid = what->mid;
if (mid.is_request()) {
// make sure that a request always gets a response;
// the exit reason reflects the first actor on the
// forwarding chain that is out of service
detail::sync_request_bouncer rb{reason};
rb(what->sender, mid);
}
return;
}
if (is_system_message(what->msg)) {
local_actor::enqueue(std::move(what), host);
// handle and consume the system message;
// the only effect that MAY result from handling a system message
// is to exit the actor if it hasn't exited already;
// `handle_system_message()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done
handle_system_message(what->msg, host);
} else {
// process and forward the non-system message;
// store `f` as the next stage in the forwarding chain
what->stages.push_back(f_);
// forward modified message to `g`
g_->enqueue(std::move(what), host);
}
}
composed_actor::message_types_set composed_actor::message_types() const {
return msg_types_;
}
void composed_actor::handle_system_message(const message& msg,
execution_unit* host) {
// `monitorable_actor::cleanup()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done;
// handles only `down_msg` from constituent actors and `exit_msg` from anyone
if (msg.size() != 1)
return; // neither case
if (msg.match_element<down_msg>(0)) {
auto& dm = msg.get_as<down_msg>(0);
CAF_ASSERT(dm.reason != exit_reason::not_exited);
if (dm.source == f_ || dm.source == g_) {
// one of the constituent actors has exited, so
// exits the composed actor with the same reason
monitorable_actor::cleanup(dm.reason, host);
}
return;
}
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
CAF_ASSERT(em.reason != exit_reason::not_exited);
// exit message received; exits with the same reason
monitorable_actor::cleanup(em.reason, host);
return;
}
// store second_ as the next stage in the forwarding chain
what->stages.push_back(second_);
// forward modified message to first_
first_->enqueue(std::move(what), host);
// drop other cases
}
bool composed_actor::is_system_message(const message& msg) {
return (msg.size() == 1 && (msg.match_element<exit_msg>(0)
|| msg.match_element<down_msg>(0)))
|| msg.match_element<down_msg>(0)))
|| (msg.size() > 1 && msg.match_element<sys_atom>(0));
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/config.hpp"
#define CAF_SUITE bound_actor
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
using namespace caf;
namespace {
behavior dbl_bhvr(event_based_actor* self) {
return {
[](int v) {
return 2 * v;
},
[=] {
self->quit();
}
};
}
struct fixture {
void wait_until_exited() {
self->receive(
[](const down_msg&) {
CAF_CHECK(true);
}
);
}
template <class Actor>
static bool exited(const Actor& handle) {
auto ptr = actor_cast<abstract_actor*>(handle);
auto dptr = dynamic_cast<monitorable_actor*>(ptr);
CAF_REQUIRE(dptr != nullptr);
return dptr->exited();
}
actor_system system;
scoped_actor self{ system, true };
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(bound_actor_tests, fixture)
CAF_TEST(identity) {
auto dbl = system.spawn(dbl_bhvr);
CAF_CHECK(system.registry().running() == 1);
auto bound = dbl.bind(1);
CAF_CHECK(system.registry().running() == 1);
CAF_CHECK(&bound->home_system() == &dbl->home_system());
CAF_CHECK(bound->node() == dbl->node());
CAF_CHECK(bound != dbl);
CAF_CHECK(bound->id() != dbl->id());
anon_send_exit(bound, exit_reason::kill);
anon_send_exit(dbl, exit_reason::kill);
}
// bound actor spawned dead if decorated
// actor is already dead upon spawning
CAF_TEST(lifetime_1) {
auto dbl = system.spawn(dbl_bhvr);
self->monitor(dbl);
anon_send_exit(dbl, exit_reason::kill);
wait_until_exited();
auto bound = dbl.bind(1);
CAF_CHECK(exited(bound));
}
// bound actor exits when decorated actor exits
CAF_TEST(lifetime_2) {
auto dbl = system.spawn(dbl_bhvr);
auto bound = dbl.bind(1);
self->monitor(bound);
anon_send(dbl, message{});
wait_until_exited();
}
// 1) ignores down message not from the decorated actor
// 2) exits by receiving an exit message
// 3) exit has no effect on decorated actor
CAF_TEST(lifetime_3) {
auto dbl = system.spawn(dbl_bhvr);
auto bound = dbl.bind(1);
anon_send(bound, down_msg{ self->address(),
exit_reason::kill });
CAF_CHECK(! exited(bound));
self->monitor(bound);
auto em_sender = system.spawn(dbl_bhvr);
em_sender->link_to(bound->address());
anon_send_exit(em_sender, exit_reason::kill);
wait_until_exited();
self->request(dbl, 1).receive(
[](int v) {
CAF_CHECK(v == 2);
},
[](error) {
CAF_CHECK(false);
}
);
anon_send_exit(dbl, exit_reason::kill);
}
CAF_TEST(request_response_promise) {
auto dbl = system.spawn(dbl_bhvr);
auto bound = dbl.bind(1);
anon_send_exit(bound, exit_reason::kill);
CAF_CHECK(exited(bound));
self->request(bound, message{}).receive(
[](int) {
CAF_CHECK(false);
},
[](error err) {
CAF_CHECK(err.code() == sec::request_receiver_down);
}
);
anon_send_exit(dbl, exit_reason::kill);
}
CAF_TEST(partial_currying) {
using namespace std::placeholders;
auto impl = []() -> behavior {
return {
[](ok_atom, int x) {
return x;
},
[](ok_atom, double x) {
return x;
}
};
};
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
auto bound = aut.bind(ok_atom::value, _1);
CAF_CHECK(aut.id() != bound.id());
CAF_CHECK(aut.node() == bound.node());
CAF_CHECK(aut != bound);
CAF_CHECK(system.registry().running() == 1);
self->request(bound, 2.0).receive(
[](double y) {
CAF_CHECK(y == 2.0);
}
);
self->request(bound, 10).receive(
[](int y) {
CAF_CHECK(y == 10);
}
);
self->send_exit(aut, exit_reason::kill);
}
CAF_TEST(full_currying) {
auto dbl_actor = system.spawn(dbl_bhvr);
auto bound = dbl_actor.bind(1);
self->request(bound, message{}).receive(
[](int v) {
CAF_CHECK(v == 2);
},
[](error err) {
CAF_CHECK(false);
}
);
anon_send_exit(bound, exit_reason::kill);
anon_send_exit(dbl_actor, exit_reason::kill);
}
CAF_TEST(type_safe_currying) {
using namespace std::placeholders;
using testee = typed_actor<replies_to<ok_atom, int>::with<int>,
replies_to<ok_atom, double>::with<double>>;
auto impl = []() -> testee::behavior_type {
return {
[](ok_atom, int x) {
return x;
},
[](ok_atom, double x) {
return x;
}
};
};
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using curried_signature = typed_actor<replies_to<int>::with<int>,
replies_to<double>::with<double>>;
auto bound = aut.bind(ok_atom::value, _1);
CAF_CHECK(aut != bound);
CAF_CHECK(system.registry().running() == 1);
static_assert(std::is_same<decltype(bound), curried_signature>::value,
"bind returned wrong actor handle");
self->request(bound, 2.0).receive(
[](double y) {
CAF_CHECK(y == 2.0);
}
);
self->request(bound, 10).receive(
[](int y) {
CAF_CHECK(y == 10);
}
);
self->send_exit(aut, exit_reason::kill);
}
CAF_TEST(reordering) {
auto impl = []() -> behavior {
return {
[](int x, double y) {
return x * y;
}
};
};
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using namespace std::placeholders;
auto bound = aut.bind(_2, _1);
CAF_CHECK(aut != bound);
CAF_CHECK(system.registry().running() == 1);
self->request(bound, 2.0, 10).receive(
[](double y) {
CAF_CHECK(y == 20.0);
}
);
self->send_exit(aut, exit_reason::kill);
}
CAF_TEST(type_safe_reordering) {
using testee = typed_actor<replies_to<int, double>::with<double>>;
auto impl = []() -> testee::behavior_type {
return {
[](int x, double y) {
return x * y;
}
};
};
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using namespace std::placeholders;
using swapped_signature = typed_actor<replies_to<double, int>::with<double>>;
auto bound = aut.bind(_2, _1);
CAF_CHECK(aut != bound);
CAF_CHECK(system.registry().running() == 1);
static_assert(std::is_same<decltype(bound), swapped_signature>::value,
"bind returned wrong actor handle");
self->request(bound, 2.0, 10).receive(
[](double y) {
CAF_CHECK(y == 20.0);
}
);
self->send_exit(aut, exit_reason::kill);
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/config.hpp"
#define CAF_SUITE composed_actor
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
using namespace caf;
namespace {
behavior dbl_bhvr(event_based_actor* self) {
return {
[](int v) {
return 2 * v;
},
[=] {
self->quit();
}
};
}
using first_stage = typed_actor<replies_to<int>::with<double, double>>;
using second_stage = typed_actor<replies_to<double, double>::with<double>>;
first_stage::behavior_type first_stage_impl() {
return [](int i) {
return std::make_tuple(i * 2.0, i * 4.0);
};
};
second_stage::behavior_type second_stage_impl() {
return [](double x, double y) {
return x * y;
};
}
struct fixture {
void wait_until_exited() {
self->receive(
[](const down_msg&) {
CAF_CHECK(true);
}
);
}
template <class Actor>
static bool exited(const Actor& handle) {
auto ptr = actor_cast<abstract_actor*>(handle);
auto dptr = dynamic_cast<monitorable_actor*>(ptr);
CAF_REQUIRE(dptr != nullptr);
return dptr->exited();
}
actor_system system;
scoped_actor self{ system, true };
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(bound_actor_tests, fixture)
CAF_TEST(identity) {
actor_system system_of_g;
actor_system system_of_f;
auto g = system_of_g.spawn(first_stage_impl);
auto f = system_of_f.spawn(second_stage_impl);
CAF_CHECK(system_of_g.registry().running() == 1);
auto composed = f * g;
CAF_CHECK(system_of_g.registry().running() == 1);
CAF_CHECK(&composed->home_system() == &g->home_system());
CAF_CHECK(composed->node() == g->node());
CAF_CHECK(composed->id() != g->id());
CAF_CHECK(composed != g);
CAF_CHECK(composed->message_types()
== g->home_system().message_types(composed));
anon_send_exit(composed, exit_reason::kill);
anon_send_exit(f, exit_reason::kill);
anon_send_exit(g, exit_reason::kill);
}
// spawned dead if `g` is already dead upon spawning
CAF_TEST(lifetime_1a) {
auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr);
self->monitor(g);
anon_send_exit(g, exit_reason::kill);
wait_until_exited();
auto fg = f * g;
CAF_CHECK(exited(fg));
anon_send_exit(f, exit_reason::kill);
}
// spawned dead if `f` is already dead upon spawning
CAF_TEST(lifetime_1b) {
auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr);
self->monitor(f);
anon_send_exit(f, exit_reason::kill);
wait_until_exited();
auto fg = f * g;
CAF_CHECK(exited(fg));
anon_send_exit(g, exit_reason::kill);
}
// `f.g` exits when `g` exits
CAF_TEST(lifetime_2a) {
auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr);
auto fg = f * g;
self->monitor(fg);
anon_send(g, message{});
wait_until_exited();
anon_send_exit(f, exit_reason::kill);
}
// `f.g` exits when `f` exits
CAF_TEST(lifetime_2b) {
auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr);
auto fg = f * g;
self->monitor(fg);
anon_send(f, message{});
wait_until_exited();
anon_send_exit(g, exit_reason::kill);
}
// 1) ignores down message not from constituent actors
// 2) exits by receiving an exit message
// 3) exit has no effect on constituent actors
CAF_TEST(lifetime_3) {
auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr);
auto fg = f * g;
self->monitor(fg);
anon_send(fg, down_msg{ self->address(),
exit_reason::kill });
CAF_CHECK(! exited(fg));
auto em_sender = system.spawn(dbl_bhvr);
em_sender->link_to(fg.address());
anon_send_exit(em_sender, exit_reason::kill);
wait_until_exited();
self->request(f, 1).receive(
[](int v) {
CAF_CHECK(v == 2);
},
[](error) {
CAF_CHECK(false);
}
);
self->request(g, 1).receive(
[](int v) {
CAF_CHECK(v == 2);
},
[](error) {
CAF_CHECK(false);
}
);
anon_send_exit(f, exit_reason::kill);
anon_send_exit(g, exit_reason::kill);
}
CAF_TEST(request_response_promise) {
auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr);
auto fg = f * g;
anon_send_exit(fg, exit_reason::kill);
CAF_CHECK(exited(fg));
self->request(fg, 1).receive(
[](int) {
CAF_CHECK(false);
},
[](error err) {
CAF_CHECK(err.code() == sec::request_receiver_down);
}
);
anon_send_exit(f, exit_reason::kill);
anon_send_exit(g, exit_reason::kill);
}
// single composition of distinct actors
CAF_TEST(dot_composition_1) {
auto first = system.spawn(first_stage_impl);
auto second = system.spawn(second_stage_impl);
auto first_then_second = second * first;
self->request(first_then_second, 42).receive(
[](double res) {
CAF_CHECK(res == (42 * 2.0) * (42 * 4.0));
}
);
anon_send_exit(first, exit_reason::kill);
anon_send_exit(second, exit_reason::kill);
}
// multiple self composition
CAF_TEST(dot_composition_2) {
auto dbl_actor = system.spawn(dbl_bhvr);
auto dbl_x4_actor = dbl_actor * dbl_actor
* dbl_actor * dbl_actor;
self->request(dbl_x4_actor, 1).receive(
[](int v) {
CAF_CHECK(v == 16);
},
[](error) {
CAF_CHECK(false);
}
);
anon_send_exit(dbl_actor, exit_reason::kill);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -521,173 +521,6 @@ CAF_TEST(check_signature) {
);
}
namespace {
using first_stage = typed_actor<replies_to<int>::with<double, double>>;
using second_stage = typed_actor<replies_to<double, double>::with<double>>;
first_stage::behavior_type first_stage_impl() {
return [](int i) {
return std::make_tuple(i * 2.0, i * 4.0);
};
};
second_stage::behavior_type second_stage_impl() {
return [](double x, double y) {
return x * y;
};
}
} // namespace <anonymous>
CAF_TEST(dot_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).receive(
[](double res) {
CAF_CHECK(res == (42 * 2.0) * (42 * 4.0));
}
);
anon_send_exit(first, exit_reason::user_shutdown);
anon_send_exit(second, exit_reason::user_shutdown);
}
CAF_TEST(currying) {
using namespace std::placeholders;
auto impl = []() -> behavior {
return {
[](ok_atom, int x) {
return x;
},
[](ok_atom, double x) {
return x;
}
};
};
actor_system system;
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
auto bound = aut.bind(ok_atom::value, _1);
CAF_CHECK(aut.id() == bound.id());
CAF_CHECK(aut.node() == bound.node());
CAF_CHECK(aut == bound);
CAF_CHECK(system.registry().running() == 1);
scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0).receive(
[](double y) {
CAF_CHECK(y == 2.0);
}
);
self->request(bound, 10).receive(
[](int y) {
CAF_CHECK(y == 10);
}
);
self->send_exit(bound, exit_reason::kill);
self->await_all_other_actors_done();
}
CAF_TEST(type_safe_currying) {
using namespace std::placeholders;
using testee = typed_actor<replies_to<ok_atom, int>::with<int>,
replies_to<ok_atom, double>::with<double>>;
auto impl = []() -> testee::behavior_type {
return {
[](ok_atom, int x) {
return x;
},
[](ok_atom, double x) {
return x;
}
};
};
actor_system system;
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using curried_signature = typed_actor<replies_to<int>::with<int>,
replies_to<double>::with<double>>;
//auto bound = actor_bind(aut, ok_atom::value, _1);
auto bound = aut.bind(ok_atom::value, _1);
CAF_CHECK(aut == bound);
CAF_CHECK(system.registry().running() == 1);
static_assert(std::is_same<decltype(bound), curried_signature>::value,
"bind returned wrong actor handle");
scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0).receive(
[](double y) {
CAF_CHECK(y == 2.0);
}
);
self->request(bound, 10).receive(
[](int y) {
CAF_CHECK(y == 10);
}
);
self->send_exit(bound, exit_reason::kill);
self->await_all_other_actors_done();
}
CAF_TEST(reordering) {
auto impl = []() -> behavior {
return {
[](int x, double y) {
return x * y;
}
};
};
actor_system system;
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using namespace std::placeholders;
auto bound = aut.bind(_2, _1);
CAF_CHECK(aut == bound);
CAF_CHECK(system.registry().running() == 1);
scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0, 10).receive(
[](double y) {
CAF_CHECK(y == 20.0);
}
);
self->send_exit(bound, exit_reason::kill);
self->await_all_other_actors_done();
}
CAF_TEST(type_safe_reordering) {
using testee = typed_actor<replies_to<int, double>::with<double>>;
auto impl = []() -> testee::behavior_type {
return {
[](int x, double y) {
return x * y;
}
};
};
actor_system system;
auto aut = system.spawn(impl);
CAF_CHECK(system.registry().running() == 1);
using namespace std::placeholders;
using swapped_signature = typed_actor<replies_to<double, int>::with<double>>;
auto bound = aut.bind(_2, _1);
CAF_CHECK(aut == bound);
CAF_CHECK(system.registry().running() == 1);
static_assert(std::is_same<decltype(bound), swapped_signature>::value,
"bind returned wrong actor handle");
scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0, 10).receive(
[](double y) {
CAF_CHECK(y == 20.0);
}
);
self->send_exit(bound, exit_reason::kill);
self->await_all_other_actors_done();
}
CAF_TEST_FIXTURE_SCOPE_END()
#endif // CAF_WINDOWS
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