Commit 5c95ac2a authored by Dominik Charousset's avatar Dominik Charousset

Allow group subscriptions only in untyped actors

parent 01bd0842
......@@ -104,15 +104,14 @@ public:
make_message(std::forward<Ts>(xs)...), ctx);
}
virtual void unsubscribe(const actor_control_block* who) = 0;
/// @endcond
protected:
abstract_group(actor_system& sys, module_ptr module,
std::string group_id, const node_id& nid);
// called by local_actor
virtual void unsubscribe(const actor_control_block* who) = 0;
actor_system& system_;
module_ptr module_;
std::string identifier_;
......
......@@ -33,6 +33,7 @@
#include "caf/group.hpp"
#include "caf/extend.hpp"
#include "caf/logger.hpp"
#include "caf/others.hpp"
#include "caf/result.hpp"
#include "caf/message.hpp"
#include "caf/node_id.hpp"
......
......@@ -43,6 +43,7 @@
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/subscriber.hpp"
namespace caf {
namespace mixin {
......@@ -60,15 +61,13 @@ namespace caf {
/// @extends local_actor
class blocking_actor
: public extend<local_actor, blocking_actor>::
with<mixin::requester, mixin::sender>,
with<mixin::requester,
mixin::sender,
mixin::subscriber>,
public dynamically_typed_actor_base {
public:
// -- member types -----------------------------------------------------------
/// Direct base type.
using super = extend<local_actor, blocking_actor>::
with<mixin::requester, mixin::sender>;
/// Absolute timeout type.
using timeout_type = std::chrono::high_resolution_clock::time_point;
......
......@@ -31,6 +31,7 @@
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/subscriber.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/logger.hpp"
......@@ -49,23 +50,31 @@ public:
class event_based_actor : public extend<scheduled_actor,
event_based_actor>::
with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
mixin::behavior_changer,
mixin::subscriber>,
public dynamically_typed_actor_base {
public:
using super = extend<scheduled_actor, event_based_actor>::
with<mixin::sender, mixin::requester, mixin::behavior_changer>;
// -- member types -----------------------------------------------------------
/// Required by `spawn` for type deduction.
using signatures = none_t;
/// Required by `spawn` for type deduction.
using behavior_type = behavior;
// -- constructors, destructors ----------------------------------------------
explicit event_based_actor(actor_config& cfg);
~event_based_actor();
// -- overridden functions of local_actor ------------------------------------
void initialize() override;
protected:
// -- behavior management ----------------------------------------------------
/// Returns the initial actor behavior.
virtual behavior make_behavior();
};
......
......@@ -21,6 +21,8 @@
#define CAF_GROUP_HPP
#include <string>
#include <utility>
#include <functional>
#include "caf/fwd.hpp"
#include "caf/none.hpp"
......@@ -114,4 +116,14 @@ std::string to_string(const group& x);
} // namespace caf
namespace std {
template <>
struct hash<caf::group> {
inline size_t operator()(const caf::group& x) const {
// groups are singleton objects, the address is thus the best possible hash
return ! x ? 0 : reinterpret_cast<size_t>(x.get());
}
};
} // namespace std
#endif // CAF_GROUP_HPP
......@@ -202,13 +202,6 @@ public:
return context_->system();
}
/// Causes this actor to subscribe to the group `what`.
/// The group will be unsubscribed if the actor finishes execution.
void join(const group& what);
/// Causes this actor to leave the group `what`.
void leave(const group& what);
/// @cond PRIVATE
void monitor(abstract_actor* whom);
......@@ -377,9 +370,6 @@ protected:
// last used request ID
message_id last_request_id_;
// used for group management
std::set<group> subscriptions_;
/// Factory function for returning initial behavior in function-based actors.
std::function<behavior (local_actor*)> initial_behavior_fac_;
};
......
......@@ -41,13 +41,21 @@ namespace mixin {
template <class Base, class Derived>
class behavior_changer : public Base {
public:
// -- member types -----------------------------------------------------------
using extended_base = behavior_changer;
using behavior_type = typename behavior_type_of<Derived>::type;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
behavior_changer(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
// -- behavior management ----------------------------------------------------
void become(behavior_type bhvr) {
this->do_become(std::move(bhvr.unbox()), true);
}
......
......@@ -42,12 +42,18 @@ struct is_blocking_requester : std::false_type { };
template <class Base, class Subtype>
class requester : public Base {
public:
// -- member types -----------------------------------------------------------
using extended_base = requester;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
requester(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
static constexpr bool is_blocking_subtype = is_blocking_requester<Subtype>::value;
// -- request ----------------------------------------------------------------
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
/// @returns A handle identifying a future-like handle to the response.
......@@ -63,7 +69,7 @@ public:
typename std::decay<Ts>::type
>::type...>
>::type,
is_blocking_subtype>
is_blocking_requester<Subtype>::value>
request(const Handle& dest, const duration& timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
......@@ -73,16 +79,12 @@ public:
>::type...>;
static_assert(actor_accepts_message<typename signatures_of<Handle>::type, token>::value,
"receiver does not accept given message");
auto req_id = dptr()->new_request_id(P);
dest->eq_impl(req_id, dptr()->ctrl(), dptr()->context(),
auto dptr = static_cast<Subtype*>(this);
auto req_id = dptr->new_request_id(P);
dest->eq_impl(req_id, dptr->ctrl(), dptr->context(),
std::forward<Ts>(xs)...);
dptr()->request_response_timeout(timeout, req_id);
return {req_id.response_id(), dptr()};
}
private:
Subtype* dptr() {
return static_cast<Subtype*>(this);
dptr->request_response_timeout(timeout, req_id);
return {req_id.response_id(), dptr};
}
};
......
......@@ -38,11 +38,19 @@ namespace mixin {
template <class Base, class Subtype>
class sender : public Base {
public:
// -- member types -----------------------------------------------------------
using extended_base = sender;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
sender(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
// -- send function family ---------------------------------------------------
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
/// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor.
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_MIXIN_SUBSCRIBER_HPP
#define CAF_MIXIN_SUBSCRIBER_HPP
#include <unordered_set>
#include "caf/fwd.hpp"
#include "caf/group.hpp"
namespace caf {
namespace mixin {
/// Marker for `subscriber`.
struct subscriber_base {};
/// A `subscriber` is an actor that can subscribe
/// to a `group` via `self->join(...)`.
template <class Base, class Subtype>
class subscriber : public Base, public subscriber_base {
public:
// -- member types -----------------------------------------------------------
/// Allows subtypes to refer mixed types with a simple name.
using extended_base = subscriber;
/// A container for storing subscribed groups.
using subscriptions = std::unordered_set<group>;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
subscriber(actor_config& cfg, Ts&&... xs)
: Base(cfg, std::forward<Ts>(xs)...) {
if (cfg.groups != nullptr)
for (auto& grp : *cfg.groups)
join(grp);
}
// -- overridden functions of monitorable_actor ------------------------------
bool cleanup(error&& fail_state, execution_unit* ptr) override {
auto me = this->ctrl();
for (auto& subscription : subscriptions_)
subscription->unsubscribe(me);
subscriptions_.clear();
return Base::cleanup(std::move(fail_state), ptr);
}
// -- group management -------------------------------------------------------
/// Causes this actor to subscribe to the group `what`.
/// The group will be unsubscribed if the actor finishes execution.
void join(const group& what) {
CAF_LOG_TRACE(CAF_ARG(what));
if (what == invalid_group)
return;
if (what->subscribe(this->ctrl()))
subscriptions_.emplace(what);
}
/// Causes this actor to leave the group `what`.
void leave(const group& what) {
CAF_LOG_TRACE(CAF_ARG(what));
if (subscriptions_.erase(what) > 0)
what->unsubscribe(this->ctrl());
}
/// Returns all subscribed groups.
const subscriptions& joined_groups() const {
return subscriptions_;
}
private:
// -- data members -----------------------------------------------------------
/// Stores all subscribed groups.
subscriptions subscriptions_;
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_SUBSCRIBER_HPP
......@@ -50,7 +50,7 @@ bool blocking_actor::accept_one_cond::post() {
}
blocking_actor::blocking_actor(actor_config& sys)
: super(sys.add_flag(local_actor::is_blocking_flag)) {
: extended_base(sys.add_flag(local_actor::is_blocking_flag)) {
// nop
}
......
......@@ -22,7 +22,7 @@
namespace caf {
event_based_actor::event_based_actor(actor_config& cfg) : super(cfg) {
event_based_actor::event_based_actor(actor_config& cfg) : extended_base(cfg) {
// nop
}
......
......@@ -47,9 +47,7 @@ local_actor::local_actor(actor_config& cfg)
: monitorable_actor(cfg),
context_(cfg.host),
initial_behavior_fac_(std::move(cfg.init_fun)) {
if (cfg.groups != nullptr)
for (auto& grp : *cfg.groups)
join(grp);
// nop
}
local_actor::~local_actor() {
......@@ -92,30 +90,10 @@ void local_actor::demonitor(const actor_addr& whom) {
ptr->get()->detach(tk);
}
void local_actor::join(const group& what) {
CAF_LOG_TRACE(CAF_ARG(what));
if (what == invalid_group)
return;
if (what->subscribe(ctrl()))
subscriptions_.emplace(what);
}
void local_actor::leave(const group& what) {
CAF_LOG_TRACE(CAF_ARG(what));
if (subscriptions_.erase(what) > 0)
what->unsubscribe(ctrl());
}
void local_actor::on_exit() {
// nop
}
std::vector<group> local_actor::joined_groups() const {
std::vector<group> result;
for (auto& x : subscriptions_)
result.emplace_back(x);
return result;
}
message_id local_actor::new_request_id(message_priority mp) {
auto result = ++last_request_id_;
......@@ -211,10 +189,6 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
detail::sync_request_bouncer f{fail_state};
mailbox_.close(f);
}
auto me = ctrl();
for (auto& subscription : subscriptions_)
subscription->unsubscribe(me);
subscriptions_.clear();
// tell registry we're done
is_registered(false);
monitorable_actor::cleanup(std::move(fail_state), host);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 blocking_actor
#include "caf/test/unit_test.hpp"
#include "caf/all.hpp"
using namespace caf;
namespace {
struct fixture {
actor_system_config cfg;
actor_system system;
scoped_actor self;
fixture() : system(cfg), self(system) {
// nop
}
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(blocking_actor_tests, fixture)
CAF_TEST(catch_all) {
self->send(self, 42);
self->receive(
[](float) {
CAF_FAIL("received unexpected float");
},
others >> [](message_view& x) -> result<message> {
CAF_CHECK_EQUAL(to_string(x.content()), "(42)");
return sec::unexpected_message;
}
);
self->receive(
[](const error& err) {
CAF_CHECK_EQUAL(err, sec::unexpected_message);
}
);
}
CAF_TEST_FIXTURE_SCOPE_END()
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