public:

void send_all_subscribers(actor* sender, const any_tuple& msg) {

shared_guard guard(m_mtx);

for (auto& s : m_subscribers) {

s->enqueue(sender, msg);

}

}

void enqueue(actor* sender, any_tuple msg) {

send_all_subscribers(sender, msg);

m_broker->enqueue(sender, move(msg));

}

pair<bool, size_t> add_subscriber(const channel_ptr& who) {

exclusive_guard guard(m_mtx);

if (m_subscribers.insert(who).second) {

return {true, m_subscribers.size()};

}

return {false, m_subscribers.size()};

}

pair<bool, size_t> erase_subscriber(const channel_ptr& who) {

exclusive_guard guard(m_mtx);

auto erased_one = m_subscribers.erase(who) > 0;

return {erased_one, m_subscribers.size()};

}

group::subscription subscribe(const channel_ptr& who) {

if (add_subscriber(who).first) {

return {who, this};

}

return {};

}

void unsubscribe(const channel_ptr& who) {

erase_subscriber(who);

}

void serialize(serializer* sink);

const actor_ptr& broker() const {

return m_broker;

}

local_group(bool spawn_local_broker, local_group_module* mod, string id);

protected:

util::shared_spinlock m_mtx;

set<channel_ptr> m_subscribers;

actor_ptr m_broker;

public:

local_broker(local_group_ptr g) : m_group(move(g)) { }

void init() {

become (

on(atom("JOIN"), arg_match) >> [=](const actor_ptr& other) {

if (other && m_acquaintances.insert(other).second) {

monitor(other);

}

},

on(atom("LEAVE"), arg_match) >> [=](const actor_ptr& other) {

if (other && m_acquaintances.erase(other) > 0) {

demonitor(other);

}

},

on(atom("FORWARD"), arg_match) >> [=](const any_tuple& what) {

// local forwarding

m_group->send_all_subscribers(last_sender().get(), what);

// forward to all acquaintances

send_to_acquaintances(what);

},

on<atom("DOWN"), uint32_t>() >> [=] {

actor_ptr other = last_sender();

if (other) m_acquaintances.erase(other);

},

others() >> [=] {

send_to_acquaintances(last_dequeued());

}

);

}

private:

void send_to_acquaintances(const any_tuple& what) {

// send to all remote subscribers

auto sender = last_sender().get();

for (auto& acquaintance : m_acquaintances) {

acquaintance->enqueue(sender, what);

}

}

local_group_ptr m_group;

set<actor_ptr> m_acquaintances;

typedef local_group super;

public:

template<typename... Args>

local_group_proxy(actor_ptr remote_broker, Args&&... args)

: super(false, forward<Args>(args)...) {

CPPA_REQUIRE(m_broker == nullptr);

CPPA_REQUIRE(remote_broker != nullptr);

CPPA_REQUIRE(remote_broker->is_proxy());

m_broker = move(remote_broker);

m_proxy_broker = spawn_hidden<proxy_broker>(this);

}

group::subscription subscribe(const channel_ptr& who) {

auto res = add_subscriber(who);

if (res.first) {

if (res.second == 1) {

// join the remote source

m_broker->enqueue(nullptr,

make_any_tuple(atom("JOIN"), m_proxy_broker));

}

return {who, this};

}

return {};

}

void unsubscribe(const channel_ptr& who) {

auto res = erase_subscriber(who);

if (res.first && res.second == 0) {

// leave the remote source,

// because there's no more subscriber on this node

m_broker->enqueue(nullptr,

make_any_tuple(atom("LEAVE"), m_proxy_broker));

}

}

void enqueue(actor* sender, any_tuple msg) {

// forward message to the broker

m_broker->enqueue(sender,

make_any_tuple(atom("FORWARD"), move(msg)));

}

private:

actor_ptr m_proxy_broker;

public:

proxy_broker(local_group_proxy_ptr grp) : m_group(move(grp)) { }

void init() {

become (

others() >> [=] {

m_group->send_all_subscribers(last_sender().get(),

last_dequeued());

}

);

}

private:

local_group_proxy_ptr m_group;

typedef group::module super;

public:

local_group_module()

: super("local"), m_process(process_information::get())

, m_actor_utype(uniform_typeid<actor_ptr>()){ }

group_ptr get(const string& identifier) {

shared_guard guard(m_instances_mtx);

auto i = m_instances.find(identifier);

if (i != m_instances.end()) {

return i->second;

}

else {

auto tmp = make_counted<local_group>(true, this, identifier);

{ // lifetime scope of uguard

upgrade_guard uguard(guard);

auto p = m_instances.insert(make_pair(identifier, tmp));

// someone might preempt us

return p.first->second;

}

}

}

intrusive_ptr<group> deserialize(deserializer* source) {

// deserialize {identifier, process_id, node_id}

auto identifier = source->read<string>();

// deserialize broker

actor_ptr broker;

m_actor_utype->deserialize(&broker, source);

CPPA_REQUIRE(broker != nullptr);

if (!broker) return nullptr;

if (!broker->is_proxy()) {

return this->get(identifier);

}

else {

shared_guard guard(m_proxies_mtx);

auto i = m_proxies.find(broker);

if (i != m_proxies.end()) {

return i->second;

}

else {

local_group_ptr tmp(new local_group_proxy(broker, this,

identifier));

upgrade_guard uguard(guard);

auto p = m_proxies.insert(make_pair(broker, tmp));

// someone might preempt us

return p.first->second;

}

}

}

void serialize(local_group* ptr, serializer* sink) {

// serialize identifier & broker

sink->write_value(ptr->identifier());

CPPA_REQUIRE(ptr->broker() != nullptr);

m_actor_utype->serialize(&ptr->broker(), sink);

}

inline const process_information& process() const {

return *m_process;

}

private:

process_information_ptr m_process;

const uniform_type_info* m_actor_utype;

util::shared_spinlock m_instances_mtx;

map<string,local_group_ptr> m_instances;

util::shared_spinlock m_proxies_mtx;

map<actor_ptr,local_group_ptr> m_proxies;

typedef group super;

public:

remote_group(group::module_ptr parent, string id, local_group_ptr decorated)

: super(parent, move(id)), m_decorated(decorated) { }

group::subscription subscribe(const channel_ptr& who) {

return m_decorated->subscribe(who);

}

void unsubscribe(const channel_ptr&) { /* never called */ }

void enqueue(actor* sender, any_tuple msg) {

m_decorated->enqueue(sender, msg);

}

void serialize(serializer* sink);

void group_down() {

group_ptr _this(this);

m_decorated->send_all_subscribers(nullptr,

make_any_tuple(atom("GROUP_DOWN"),

_this));

}

private:

local_group_ptr m_decorated;

typedef unique_lock<mutex> lock_type;

public:

remote_group_ptr get(const string& key) {

remote_group_ptr result;

{ // lifetime scope of guard

lock_type guard(m_mtx);

auto i = m_instances.find(key);

if (i == m_instances.end()) {

m_worker->enqueue(nullptr, make_any_tuple(atom("FETCH"), key));

do {

m_cond.wait(guard);

} while ((i = m_instances.find(key)) == m_instances.end());

}

result = i->second;

}

if (!result) {

throw network_error("could not connect to remote group");

}

return result;

}

group_ptr peek(const string& key) {

lock_type guard(m_mtx);

auto i = m_instances.find(key);

if (i != m_instances.end()) {

return i->second;

}

return nullptr;

}

void put(const string& key, const remote_group_ptr& ptr) {

lock_type guard(m_mtx);

m_instances[key] = ptr;

m_cond.notify_all();

}

actor_ptr m_worker;

private:

mutex m_mtx;

condition_variable m_cond;

map<string, remote_group_ptr> m_instances;

typedef group::module super;

public:

remote_group_module() : super("remote") {

auto sm = make_counted<shared_map>();

group::module_ptr _this = this;

m_map = sm;

auto worker = spawn<detached_and_hidden>([_this, sm] {

typedef map<string, pair<actor_ptr, vector<pair<string, remote_group_ptr>>>>

peer_map;

peer_map peers;

receive_loop (

on(atom("FETCH"), arg_match) >> [_this, sm, &peers](const string& key) {

// format is group@host:port

auto pos1 = key.find('@');

auto pos2 = key.find(':');

auto last = string::npos;

string authority;

if (pos1 != last && pos2 != last && pos1 < pos2) {

auto name = key.substr(0, pos1);

authority = key.substr(pos1 + 1);

auto i = peers.find(authority);

actor_ptr nameserver;

if (i != peers.end()) {

nameserver = i->second.first;

}

else {

auto host = key.substr(pos1 + 1, pos2 - pos1 - 1);

auto pstr = key.substr(pos2 + 1);

istringstream iss(pstr);

uint16_t port;

if (iss >> port) {

try {

nameserver = remote_actor(host, port);

self->monitor(nameserver);

peers[authority].first = nameserver;

}

catch (exception&) {

sm->put(key, nullptr);

return;

}

}

}

sync_send(nameserver, atom("GET_GROUP"), name).await (

on(atom("GROUP"), arg_match) >> [&](const group_ptr& g) {

auto gg = dynamic_cast<local_group*>(g.get());

if (gg) {

auto rg = make_counted<remote_group>(_this, key, gg);

sm->put(key, rg);

peers[authority].second.push_back(make_pair(key, rg));

}

else {

cerr << "*** WARNING: received a non-local "

"group form nameserver for key "

<< key << " in file "

<< __FILE__

<< ", line " << __LINE__

<< endl;

sm->put(key, nullptr);

}

},

after(chrono::seconds(10)) >> [sm, &key] {

sm->put(key, nullptr);

}

);

}

},

on<atom("DOWN"), std::uint32_t>() >> [&] {

auto who = self->last_sender();

auto find_peer = [&] {

return find_if(begin(peers), end(peers), [&](const peer_map::value_type& kvp) {

return kvp.second.first == who;

});

};

for (auto i = find_peer(); i != peers.end(); i = find_peer()) {

for (auto& kvp: i->second.second) {

sm->put(kvp.first, nullptr);

kvp.second->group_down();

}

peers.erase(i);

}

},

others() >> [] { }

);

});

sm->m_worker = worker;

}

intrusive_ptr<group> get(const std::string& group_name) {

return m_map->get(group_name);

}

intrusive_ptr<group> deserialize(deserializer* source) {

return get(source->read<string>());

}

void serialize(group* ptr, serializer* sink) {

sink->write_value(ptr->identifier());

}

private:

shared_map_ptr m_map;

local_group_module* mod,

string id)

: group(mod, move(id)) {

if (spawn_local_broker) m_broker = spawn_hidden<local_broker>(this);

// 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);

group::unique_module_ptr ptr(new local_group_module);

m_mmap.insert(make_pair(string("local"), move(ptr)));

ptr.reset(new remote_group_module);

m_mmap.insert(make_pair(string("remote"), move(ptr)));

string id = "__#";

id += std::to_string(++m_ad_hoc_id);

return get_module("local")->get(id);

const string& group_identifier) {

auto mod = get_module(module_name);

if (mod) {

return mod->get(group_identifier);

}

string error_msg = "no module named \"";

error_msg += module_name;

error_msg += "\" found";

throw logic_error(error_msg);

if (!mptr) return;

const string& mname = mptr->name();

{ // lifetime scope of guard

lock_guard<mutex> guard(m_mmap_mtx);

if (m_mmap.insert(make_pair(mname, move(mptr))).second) {

return; // success; don't throw

}

}

string error_msg = "module name \"";

error_msg += mname;

error_msg += "\" already defined";

throw logic_error(error_msg);