Commit db44c650 authored by Dominik Charousset's avatar Dominik Charousset

group subscriptions are no longer attachables

parent c40333d8
...@@ -55,18 +55,48 @@ namespace cppa { class self_type; } ...@@ -55,18 +55,48 @@ namespace cppa { class self_type; }
namespace cppa { namespace detail { namespace cppa { namespace detail {
/** template<class Base>
class abstract_actor_base : public Base {
protected:
template<typename... Args>
abstract_actor_base(Args&&... args) : Base(std::forward<Args>(args)...) { }
inline void base_cleanup(std::uint32_t) { }
};
template<>
class abstract_actor_base<local_actor> : public local_actor {
typedef local_actor super;
protected:
template<typename... Args>
abstract_actor_base(Args&&... args) : super(std::forward<Args>(args)...) { }
inline void base_cleanup(std::uint32_t) {
// leave groups
this->m_subscriptions.clear();
}
};
/*
* @brief Implements linking and monitoring for actors. * @brief Implements linking and monitoring for actors.
* @tparam Base Either {@link cppa::actor actor} * @tparam Base Either {@link cppa::actor actor}
* or {@link cppa::local_actor local_actor}. * or {@link cppa::local_actor local_actor}.
*/ */
template<class Base> template<class Base>
class abstract_actor : public Base { class abstract_actor : public abstract_actor_base<Base> {
friend class self_type; friend class self_type;
typedef std::unique_ptr<attachable> attachable_ptr; typedef abstract_actor_base<Base> super;
typedef std::lock_guard<std::mutex> guard_type; typedef std::lock_guard<std::mutex> guard_type;
typedef std::unique_ptr<attachable> attachable_ptr;
public: public:
...@@ -192,7 +222,7 @@ class abstract_actor : public Base { ...@@ -192,7 +222,7 @@ class abstract_actor : public Base {
template<typename... Args> template<typename... Args>
abstract_actor(Args&&... args) abstract_actor(Args&&... args)
: Base(std::forward<Args>(args)...), m_exit_reason(exit_reason::not_exited){ : super(std::forward<Args>(args)...),m_exit_reason(exit_reason::not_exited){
// pre-allocate some nodes // pre-allocate some nodes
for (size_t i = 0; i < m_nodes.max_size() / 2; ++i) { for (size_t i = 0; i < m_nodes.max_size() / 2; ++i) {
m_nodes.push_back(new mailbox_element); m_nodes.push_back(new mailbox_element);
...@@ -201,6 +231,7 @@ class abstract_actor : public Base { ...@@ -201,6 +231,7 @@ class abstract_actor : public Base {
void cleanup(std::uint32_t reason) { void cleanup(std::uint32_t reason) {
if (reason == exit_reason::not_exited) return; if (reason == exit_reason::not_exited) return;
this->base_cleanup(reason);
decltype(m_links) mlinks; decltype(m_links) mlinks;
decltype(m_attachables) mattachables; decltype(m_attachables) mattachables;
{ // lifetime scope of guard { // lifetime scope of guard
......
...@@ -62,33 +62,35 @@ class group : public channel { ...@@ -62,33 +62,35 @@ class group : public channel {
public: public:
class unsubscriber; class subscription;
// needs access to unsubscribe() // needs access to unsubscribe()
friend class unsubscriber; friend class subscription;
// unsubscribes its channel from the group on destruction // unsubscribes its channel from the group on destruction
class unsubscriber : public attachable { class subscription {
friend class group; friend class group;
channel_ptr m_self; subscription(const subscription&) = delete;
intrusive_ptr<group> m_group; subscription& operator=(const subscription&) = delete;
public: public:
unsubscriber(const channel_ptr& s, const intrusive_ptr<group>& g); subscription() = default;
subscription(subscription&&) = default;
subscription(const channel_ptr& s, const intrusive_ptr<group>& g);
~unsubscriber(); ~subscription();
void actor_exited(std::uint32_t); inline bool valid() const { return (m_subscriber) && (m_group); }
// matches on m_group private:
bool matches(const attachable::token& what);
}; channel_ptr m_subscriber;
intrusive_ptr<group> m_group;
typedef std::unique_ptr<unsubscriber> subscription; };
/** /**
* @brief Module interface. * @brief Module interface.
......
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#ifndef CPPA_CONTEXT_HPP #ifndef CPPA_CONTEXT_HPP
#define CPPA_CONTEXT_HPP #define CPPA_CONTEXT_HPP
#include "cppa/group.hpp"
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/behavior.hpp" #include "cppa/behavior.hpp"
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
...@@ -332,6 +333,7 @@ class local_actor : public actor { ...@@ -332,6 +333,7 @@ class local_actor : public actor {
actor_ptr m_chained_actor; actor_ptr m_chained_actor;
detail::recursive_queue_node m_dummy_node; detail::recursive_queue_node m_dummy_node;
detail::recursive_queue_node* m_current_node; detail::recursive_queue_node* m_current_node;
std::map<group_ptr, group::subscription> m_subscriptions;
# endif // CPPA_DOCUMENTATION # endif // CPPA_DOCUMENTATION
......
...@@ -51,24 +51,12 @@ void group::add_module(group::module* ptr) { ...@@ -51,24 +51,12 @@ void group::add_module(group::module* ptr) {
detail::singleton_manager::get_group_manager()->add_module(ptr); detail::singleton_manager::get_group_manager()->add_module(ptr);
} }
group::unsubscriber::unsubscriber(const channel_ptr& s, group::subscription::subscription(const channel_ptr& s,
const intrusive_ptr<group>& g) const intrusive_ptr<group>& g)
: m_self(s), m_group(g) { : m_subscriber(s), m_group(g) { }
}
group::unsubscriber::~unsubscriber() {
if (m_group && m_self) m_group->unsubscribe(m_self);
}
void group::unsubscriber::actor_exited(std::uint32_t) {
// unsubscription is done in destructor
}
bool group::unsubscriber::matches(const attachable::token& what) { group::subscription::~subscription() {
if (what.subtype == typeid(group::unsubscriber)) { if (valid()) m_group->unsubscribe(m_subscriber);
return m_group == reinterpret_cast<const group*>(what.ptr);
}
return false;
} }
group::module::module(std::string name) : m_name(std::move(name)) { } group::module::module(std::string name) : m_name(std::move(name)) { }
......
...@@ -69,12 +69,11 @@ class group_impl : public group { ...@@ -69,12 +69,11 @@ class group_impl : public group {
} }
group::subscription subscribe(const channel_ptr& who) /*override*/ { group::subscription subscribe(const channel_ptr& who) /*override*/ {
group::subscription result;
exclusive_guard guard(m_shared_mtx); exclusive_guard guard(m_shared_mtx);
if (m_subscribers.insert(who).second) { if (m_subscribers.insert(who).second) {
result.reset(new group::unsubscriber(who, this)); return {who, this};
} }
return result; return {};
} }
void unsubscribe(const channel_ptr& who) /*override*/ { void unsubscribe(const channel_ptr& who) /*override*/ {
......
...@@ -85,14 +85,13 @@ void local_actor::on_exit() { } ...@@ -85,14 +85,13 @@ void local_actor::on_exit() { }
void local_actor::init() { } void local_actor::init() { }
void local_actor::join(const group_ptr& what) { void local_actor::join(const group_ptr& what) {
if (what) attach(what->subscribe(this)); if (what && m_subscriptions.count(what) == 0) {
m_subscriptions.insert(std::make_pair(what, what->subscribe(this)));
}
} }
void local_actor::leave(const group_ptr& what) { void local_actor::leave(const group_ptr& what) {
if (what) { if (what) m_subscriptions.erase(what);
attachable::token group_token(typeid(group::unsubscriber), what.get());
detach(group_token);
}
} }
} // namespace cppa } // namespace cppa
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