Commit 6c85cdf1 authored by Dominik Charousset's avatar Dominik Charousset

refactored group interface to enable user-defined serialization (see Issue 53)

parent 07f00977
...@@ -51,11 +51,13 @@ class group_manager { ...@@ -51,11 +51,13 @@ class group_manager {
intrusive_ptr<group> anonymous(); intrusive_ptr<group> anonymous();
void add_module(group::module*); void add_module(group::unique_module_ptr);
group::module_ptr get_module(const std::string& module_name);
private: private:
typedef std::map< std::string, std::unique_ptr<group::module> > modules_map; typedef std::map<std::string, group::unique_module_ptr> modules_map;
modules_map m_mmap; modules_map m_mmap;
std::mutex m_mmap_mtx; std::mutex m_mmap_mtx;
......
...@@ -38,27 +38,25 @@ ...@@ -38,27 +38,25 @@
#include "cppa/attachable.hpp" #include "cppa/attachable.hpp"
#include "cppa/ref_counted.hpp" #include "cppa/ref_counted.hpp"
namespace cppa { namespace detail { class group_manager; } } namespace cppa { namespace detail {
class group_manager;
class peer_connection;
} } // namespace cppa::detail
namespace cppa { namespace cppa {
class serializer;
class deserializer;
/** /**
* @brief A multicast group. * @brief A multicast group.
*/ */
class group : public channel { class group : public channel {
friend class detail::group_manager; friend class detail::group_manager;
friend class detail::peer_connection; // needs access to remote_enqueue
std::string m_identifier;
std::string m_module_name;
protected:
group(std::string&& id, std::string&& mod_name);
group(const std::string& id, const std::string& mod_name);
virtual void unsubscribe(const channel_ptr& who) = 0;
public: public:
...@@ -119,8 +117,15 @@ class group : public channel { ...@@ -119,8 +117,15 @@ class group : public channel {
*/ */
virtual intrusive_ptr<group> get(const std::string& group_name) = 0; virtual intrusive_ptr<group> get(const std::string& group_name) = 0;
virtual intrusive_ptr<group> deserialize(deserializer* source) = 0;
}; };
typedef module* module_ptr;
typedef std::unique_ptr<module> unique_module_ptr;
virtual void serialize(serializer* sink) = 0;
/** /**
* @brief A string representation of the group identifier. * @brief A string representation of the group identifier.
* @returns The group identifier as string (e.g. "224.0.0.1" for IPv4 * @returns The group identifier as string (e.g. "224.0.0.1" for IPv4
...@@ -128,6 +133,8 @@ class group : public channel { ...@@ -128,6 +133,8 @@ class group : public channel {
*/ */
const std::string& identifier() const; const std::string& identifier() const;
module_ptr get_module() const;
/** /**
* @brief The name of the module. * @brief The name of the module.
* @returns The module name of this group (e.g. "local"). * @returns The module name of this group (e.g. "local").
...@@ -152,9 +159,9 @@ class group : public channel { ...@@ -152,9 +159,9 @@ class group : public channel {
/** /**
* @brief Returns an anonymous group. * @brief Returns an anonymous group.
* *
* Each calls to this member function returns a new instances * Each calls to this member function returns a new instance
* of an anonymous group. Anonymous groups can be used whenever * of an anonymous group. Anonymous groups can be used whenever
* a set of actors wants to communicate over an exclusive channel. * a set of actors wants to communicate using an exclusive channel.
*/ */
static intrusive_ptr<group> anonymous(); static intrusive_ptr<group> anonymous();
...@@ -162,7 +169,31 @@ class group : public channel { ...@@ -162,7 +169,31 @@ class group : public channel {
* @brief Add a new group module to the libcppa group management. * @brief Add a new group module to the libcppa group management.
* @threadsafe * @threadsafe
*/ */
static void add_module(module*); static void add_module(unique_module_ptr);
/**
* @brief Returns the module associated with @p module_name.
* @threadsafe
*/
static module_ptr get_module(const std::string& module_name);
protected:
group(module_ptr module, std::string group_id);
virtual void unsubscribe(const channel_ptr& who) = 0;
/**
* @brief Called whenever a message was received via network. If @p this
* is a proxy, it should not send the message
* back to the original group but forward the message to its local
* subscribers. This member function should call @p enqueue for
* all non-proxy instances.
*/
virtual void remote_enqueue(actor* sender, any_tuple msg);
module_ptr m_module;
std::string m_identifier;
}; };
......
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "cppa/group.hpp" #include "cppa/group.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/util/shared_spinlock.hpp" #include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp" #include "cppa/util/shared_lock_guard.hpp"
#include "cppa/util/upgrade_lock_guard.hpp" #include "cppa/util/upgrade_lock_guard.hpp"
...@@ -38,17 +39,29 @@ ...@@ -38,17 +39,29 @@
namespace cppa { namespace cppa {
namespace {
inline detail::group_manager* manager() {
return detail::singleton_manager::get_group_manager();
}
} // <anonymous>
intrusive_ptr<group> group::get(const std::string& arg0, intrusive_ptr<group> group::get(const std::string& arg0,
const std::string& arg1) { const std::string& arg1) {
return detail::singleton_manager::get_group_manager()->get(arg0, arg1); return manager()->get(arg0, arg1);
} }
intrusive_ptr<group> group::anonymous() { intrusive_ptr<group> group::anonymous() {
return detail::singleton_manager::get_group_manager()->anonymous(); return manager()->anonymous();
} }
void group::add_module(group::module* ptr) { void group::add_module(group::unique_module_ptr ptr) {
detail::singleton_manager::get_group_manager()->add_module(ptr); manager()->add_module(std::move(ptr));
}
group::module_ptr group::get_module(const std::string& module_name) {
return manager()->get_module(module_name);
} }
group::subscription::subscription(const channel_ptr& s, group::subscription::subscription(const channel_ptr& s,
...@@ -65,20 +78,23 @@ const std::string& group::module::name() { ...@@ -65,20 +78,23 @@ const std::string& group::module::name() {
return m_name; return m_name;
} }
group::group(std::string&& id, std::string&& mod_name) group::group(group::module_ptr mod, std::string id)
: m_identifier(std::move(id)), m_module_name(std::move(mod_name)) { : m_module(mod), m_identifier(std::move(id)) { }
}
group::group(const std::string& id, const std::string& mod_name)
: m_identifier(id), m_module_name(mod_name) {
}
const std::string& group::identifier() const { const std::string& group::identifier() const {
return m_identifier; return m_identifier;
} }
group::module_ptr group::get_module() const {
return m_module;
}
const std::string& group::module_name() const { const std::string& group::module_name() const {
return m_module_name; return get_module()->name();
}
void group::remote_enqueue(actor* sender, any_tuple msg) {
enqueue(sender, std::move(msg));
} }
} // namespace cppa } // namespace cppa
...@@ -32,6 +32,8 @@ ...@@ -32,6 +32,8 @@
#include <stdexcept> #include <stdexcept>
#include "cppa/any_tuple.hpp" #include "cppa/any_tuple.hpp"
#include "cppa/serializer.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/detail/group_manager.hpp" #include "cppa/detail/group_manager.hpp"
#include "cppa/util/shared_spinlock.hpp" #include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp" #include "cppa/util/shared_lock_guard.hpp"
...@@ -47,24 +49,14 @@ typedef util::upgrade_lock_guard<util::shared_spinlock> upgrade_guard; ...@@ -47,24 +49,14 @@ typedef util::upgrade_lock_guard<util::shared_spinlock> upgrade_guard;
class local_group_module; class local_group_module;
class group_impl : public group { class local_group : public group {
util::shared_spinlock m_shared_mtx;
std::set<channel_ptr> m_subscribers;
protected:
template<typename F, typename S>
group_impl(F&& f, S&& s) : group(std::forward<F>(f), std::forward<S>(s)) { }
public: public:
void enqueue(actor* sender, any_tuple msg) /*override*/ { void enqueue(actor* sender, any_tuple msg) /*override*/ {
shared_guard guard(m_shared_mtx); shared_guard guard(m_shared_mtx);
for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i) { for (auto& s : m_subscribers) {
// this cast is safe because we don't affect the "value" s->enqueue(sender, msg);
// of *i, thus, the set remains in a consistent state
const_cast<channel_ptr&>(*i)->enqueue(sender, msg);
} }
} }
...@@ -80,17 +72,15 @@ class group_impl : public group { ...@@ -80,17 +72,15 @@ class group_impl : public group {
exclusive_guard guard(m_shared_mtx); exclusive_guard guard(m_shared_mtx);
m_subscribers.erase(who); m_subscribers.erase(who);
} }
};
struct anonymous_group : group_impl { void serialize(serializer* sink);
anonymous_group() : group_impl("anonymous", "anonymous") { }
};
class local_group : public group_impl { local_group(local_group_module* mod, std::string id);
friend class local_group_module; private:
local_group(const std::string& gname) : group_impl(gname, "local") { } util::shared_spinlock m_shared_mtx;
std::set<channel_ptr> m_subscribers;
}; };
...@@ -98,53 +88,74 @@ class local_group_module : public group::module { ...@@ -98,53 +88,74 @@ class local_group_module : public group::module {
typedef group::module super; typedef group::module super;
util::shared_spinlock m_mtx;
std::map<std::string, group_ptr> m_instances;
public: public:
local_group_module() : super("local") { } local_group_module() : super("local") { }
group_ptr get(const std::string& group_name) { group_ptr get(const std::string& identifier) {
shared_guard guard(m_mtx); shared_guard guard(m_mtx);
auto i = m_instances.find(group_name); auto i = m_instances.find(identifier);
if (i != m_instances.end()) { if (i != m_instances.end()) {
return i->second; return i->second;
} }
else { else {
group_ptr tmp(new local_group(group_name)); group_ptr tmp(new local_group(this, identifier));
{ // lifetime scope of uguard { // lifetime scope of uguard
upgrade_guard uguard(guard); upgrade_guard uguard(guard);
auto p = m_instances.insert(std::make_pair(group_name, tmp)); auto p = m_instances.insert(std::make_pair(identifier, tmp));
// someone might preempt us // someone might preempt us
return p.first->second; return p.first->second;
} }
} }
} }
intrusive_ptr<group> deserialize(deserializer* source) {
auto gname = source->read_value(pt_u8string);
return this->get(cppa::get<std::string>(gname));
}
void serialize(local_group* ptr, serializer* sink) {
sink->write_value(ptr->identifier());
}
private:
util::shared_spinlock m_mtx;
std::map<std::string, group_ptr> m_instances;
}; };
local_group::local_group(local_group_module* mod, std::string id)
: group(mod, std::move(id)) { }
void local_group::serialize(serializer* sink) {
// this cast is safe, because the only available constructor accepts
// local_group_module* as module pointer
static_cast<local_group_module*>(m_module)->serialize(this, sink);
}
std::atomic<size_t> m_ad_hoc_id;
} // namespace <anonymous> } // namespace <anonymous>
namespace cppa { namespace detail { namespace cppa { namespace detail {
group_manager::group_manager() { group_manager::group_manager() {
std::unique_ptr<group::module> ptr(new local_group_module); group::unique_module_ptr ptr(new local_group_module);
m_mmap.insert(std::make_pair(std::string("local"), std::move(ptr))); m_mmap.insert(std::make_pair(std::string("local"), std::move(ptr)));
} }
intrusive_ptr<group> group_manager::anonymous() { intrusive_ptr<group> group_manager::anonymous() {
return new anonymous_group; std::string id = "__#";
id += std::to_string(++m_ad_hoc_id);
return get_module("local")->get(id);
} }
intrusive_ptr<group> group_manager::get(const std::string& module_name, intrusive_ptr<group> group_manager::get(const std::string& module_name,
const std::string& group_identifier) { const std::string& group_identifier) {
{ // lifetime scope of guard auto mod = get_module(module_name);
std::lock_guard<std::mutex> guard(m_mmap_mtx); if (mod) {
auto i = m_mmap.find(module_name); return mod->get(group_identifier);
if (i != m_mmap.end()) {
return (i->second)->get(group_identifier);
}
} }
std::string error_msg = "no module named \""; std::string error_msg = "no module named \"";
error_msg += module_name; error_msg += module_name;
...@@ -152,9 +163,9 @@ intrusive_ptr<group> group_manager::get(const std::string& module_name, ...@@ -152,9 +163,9 @@ intrusive_ptr<group> group_manager::get(const std::string& module_name,
throw std::logic_error(error_msg); throw std::logic_error(error_msg);
} }
void group_manager::add_module(group::module* mod) { void group_manager::add_module(std::unique_ptr<group::module> mptr) {
const std::string& mname = mod->name(); if (!mptr) return;
std::unique_ptr<group::module> mptr(mod); const std::string& mname = mptr->name();
{ // lifetime scope of guard { // lifetime scope of guard
std::lock_guard<std::mutex> guard(m_mmap_mtx); std::lock_guard<std::mutex> guard(m_mmap_mtx);
if (m_mmap.insert(std::make_pair(mname, std::move(mptr))).second) { if (m_mmap.insert(std::make_pair(mname, std::move(mptr))).second) {
...@@ -167,4 +178,10 @@ void group_manager::add_module(group::module* mod) { ...@@ -167,4 +178,10 @@ void group_manager::add_module(group::module* mod) {
throw std::logic_error(error_msg); throw std::logic_error(error_msg);
} }
group::module* group_manager::get_module(const std::string& module_name) {
std::lock_guard<std::mutex> guard(m_mmap_mtx);
auto i = m_mmap.find(module_name);
return (i != m_mmap.end()) ? i->second.get() : nullptr;
}
} } // namespace cppa::detail } } // namespace cppa::detail
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