Commit e45262b4 authored by Dominik Charousset's avatar Dominik Charousset

fixed group chat example

parent b1035b56
...@@ -69,6 +69,7 @@ using mapped_type_list = util::type_list< ...@@ -69,6 +69,7 @@ using mapped_type_list = util::type_list<
down_msg, down_msg,
exit_msg, exit_msg,
group, group,
group_down_msg,
node_id_ptr, node_id_ptr,
io::accept_handle, io::accept_handle,
io::connection_handle, io::connection_handle,
......
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
#include <cstdint> #include <cstdint>
#include "cppa/group.hpp"
#include "cppa/actor_addr.hpp" #include "cppa/actor_addr.hpp"
namespace cppa { namespace cppa {
...@@ -68,6 +69,16 @@ struct down_msg { ...@@ -68,6 +69,16 @@ struct down_msg {
std::uint32_t reason; std::uint32_t reason;
}; };
/**
* @brief Sent to all members of a group when it goes offline.
*/
struct group_down_msg {
/**
* @brief The source of this message, i.e., the now unreachable group.
*/
group source;
};
/** /**
* @brief Sent whenever a timeout occurs during a synchronous send. * @brief Sent whenever a timeout occurs during a synchronous send.
* *
......
...@@ -57,12 +57,15 @@ void client(event_based_actor* self, const string& name) { ...@@ -57,12 +57,15 @@ void client(event_based_actor* self, const string& name) {
} }
cout << "*** join " << to_string(what) << endl; cout << "*** join " << to_string(what) << endl;
self->join(what); self->join(what);
self->send(self, what, name + " has entered the chatroom"); self->send(what, name + " has entered the chatroom");
}, },
on<string>() >> [=](const string& txt) { [=](const string& txt) {
// don't print own messages // don't print own messages
if (self->last_sender() != self) cout << txt << endl; if (self->last_sender() != self) cout << txt << endl;
}, },
[=](const group_down_msg& g) {
cout << "*** chatroom offline: " << to_string(g.source) << endl;
},
others() >> [=]() { others() >> [=]() {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl; cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
} }
...@@ -90,7 +93,6 @@ int main(int argc, char** argv) { ...@@ -90,7 +93,6 @@ int main(int argc, char** argv) {
} }
} }
cout << "*** starting client, type '/help' for a list of commands" << endl;
auto client_actor = spawn(client, name); auto client_actor = spawn(client, name);
// evaluate group parameters // evaluate group parameters
...@@ -115,6 +117,7 @@ int main(int argc, char** argv) { ...@@ -115,6 +117,7 @@ int main(int argc, char** argv) {
} }
} }
cout << "*** starting client, type '/help' for a list of commands" << endl;
istream_iterator<line> lines(cin); istream_iterator<line> lines(cin);
istream_iterator<line> eof; istream_iterator<line> eof;
match_each (lines, eof, split_line) ( match_each (lines, eof, split_line) (
......
...@@ -389,8 +389,7 @@ class remote_group : public abstract_group { ...@@ -389,8 +389,7 @@ class remote_group : public abstract_group {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
group this_group{this}; group this_group{this};
m_decorated->send_all_subscribers({invalid_actor_addr, nullptr}, m_decorated->send_all_subscribers({invalid_actor_addr, nullptr},
make_any_tuple(atom("GROUP_DOWN"), make_any_tuple(group_down_msg{this_group}));
this_group));
} }
private: private:
...@@ -464,15 +463,16 @@ class remote_group_module : public abstract_group::module { ...@@ -464,15 +463,16 @@ class remote_group_module : public abstract_group::module {
auto sm = make_counted<shared_map>(); auto sm = make_counted<shared_map>();
abstract_group::module_ptr this_group{this}; abstract_group::module_ptr this_group{this};
m_map = sm; m_map = sm;
typedef map<string, pair<actor, vector<pair<string, remote_group_ptr>>>>
peer_map;
auto peers = std::make_shared<peer_map>();
m_map->m_worker = spawn<hidden>([=](event_based_actor* self) -> behavior { m_map->m_worker = spawn<hidden>([=](event_based_actor* self) -> behavior {
CPPA_LOGC_TRACE(detail::demangle(typeid(*this_group)), CPPA_LOGC_TRACE(detail::demangle(typeid(*this_group)),
"remote_group_module$worker", "remote_group_module$worker",
""); "");
typedef map<string, pair<actor, vector<pair<string, remote_group_ptr>>>> return {
peer_map;
auto peers = std::make_shared<peer_map>();
return (
on(atom("FETCH"), arg_match) >> [=](const string& key) { on(atom("FETCH"), arg_match) >> [=](const string& key) {
CPPA_LOGF_TRACE("");
// format is group@host:port // format is group@host:port
auto pos1 = key.find('@'); auto pos1 = key.find('@');
auto pos2 = key.find(':'); auto pos2 = key.find(':');
...@@ -489,22 +489,19 @@ class remote_group_module : public abstract_group::module { ...@@ -489,22 +489,19 @@ class remote_group_module : public abstract_group::module {
else { else {
auto host = key.substr(pos1 + 1, pos2 - pos1 - 1); auto host = key.substr(pos1 + 1, pos2 - pos1 - 1);
auto pstr = key.substr(pos2 + 1); auto pstr = key.substr(pos2 + 1);
istringstream iss(pstr); try {
uint16_t port; auto port = static_cast<uint16_t>(std::stoi(pstr));
if (iss >> port) { nameserver = remote_actor(host, port);
try { self->monitor(nameserver);
nameserver = remote_actor(host, port); (*peers)[authority].first = nameserver;
self->monitor(nameserver); }
(*peers)[authority].first = nameserver; catch (exception&) {
} sm->put(key, nullptr);
catch (exception&) { return;
sm->put(key, nullptr);
return;
}
} }
} }
self->timed_sync_send(nameserver, chrono::seconds(10), atom("GET_GROUP"), name).then ( self->timed_sync_send(nameserver, chrono::seconds(10), atom("GET_GROUP"), name).then (
on(atom("GROUP"), arg_match) >> [&](const group& g) { on(atom("GROUP"), arg_match) >> [=](const group& g) {
auto gg = dynamic_cast<local_group*>(detail::raw_access::get(g)); auto gg = dynamic_cast<local_group*>(detail::raw_access::get(g));
if (gg) { if (gg) {
auto rg = make_counted<remote_group>(this_group, key, gg); auto rg = make_counted<remote_group>(this_group, key, gg);
...@@ -521,29 +518,34 @@ class remote_group_module : public abstract_group::module { ...@@ -521,29 +518,34 @@ class remote_group_module : public abstract_group::module {
sm->put(key, nullptr); sm->put(key, nullptr);
} }
}, },
on<sync_timeout_msg>() >> [sm, &key] { on<sync_timeout_msg>() >> [sm, key] {
CPPA_LOGF_WARNING("'GET_GROUP' timed out");
sm->put(key, nullptr); sm->put(key, nullptr);
} }
); );
} }
}, },
on_arg_match >> [&](const down_msg&) { on_arg_match >> [=](const down_msg&) {
auto who = self->last_sender(); auto who = self->last_sender();
auto find_peer = [&] { auto i = peers->begin();
return find_if(begin(*peers), end(*peers), [&](const peer_map::value_type& kvp) { auto last = peers->end();
return kvp.second.first == who; while (i != last) {
}); auto& e = i->second;
}; if (e.first == who) {
for (auto i = find_peer(); i != peers->end(); i = find_peer()) { for (auto& kvp: e.second) {
for (auto& kvp: i->second.second) { sm->put(kvp.first, nullptr);
sm->put(kvp.first, nullptr); kvp.second->group_down();
kvp.second->group_down(); }
i = peers->erase(i);
} }
peers->erase(i); else ++i;
} }
}, },
others() >> [] { } others() >> [=] {
); CPPA_LOGF_ERROR("unexpected message: "
<< to_string(self->last_dequeued()));
}
};
}); });
} }
......
...@@ -117,6 +117,7 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) { ...@@ -117,6 +117,7 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) {
} }
size_t ipv4_io_stream::write_some(const void* buf, size_t len) { size_t ipv4_io_stream::write_some(const void* buf, size_t len) {
CPPA_LOG_TRACE(CPPA_ARG(buf) << ", " << CPPA_ARG(len));
auto send_result = ::send(m_fd, reinterpret_cast<const char*>(buf), len, 0); auto send_result = ::send(m_fd, reinterpret_cast<const char*>(buf), len, 0);
handle_write_result(send_result, true); handle_write_result(send_result, true);
return static_cast<size_t>(send_result); return static_cast<size_t>(send_result);
......
...@@ -202,8 +202,10 @@ class middleman_impl : public middleman { ...@@ -202,8 +202,10 @@ class middleman_impl : public middleman {
peer* get_peer(const node_id& node) override { peer* get_peer(const node_id& node) override {
CPPA_LOG_TRACE(CPPA_TARG(node, to_string)); CPPA_LOG_TRACE(CPPA_TARG(node, to_string));
auto i = m_peers.find(node); auto i = m_peers.find(node);
if (i != m_peers.end()) { // future work (?): we *could* try to be smart here and try to
CPPA_REQUIRE(i->second.impl != nullptr); // route all messages to node via other known peers in the network
// if i->second.impl == nullptr
if (i != m_peers.end() && i->second.impl != nullptr) {
CPPA_LOG_DEBUG("result = " << i->second.impl); CPPA_LOG_DEBUG("result = " << i->second.impl);
return i->second.impl; return i->second.impl;
} }
......
...@@ -69,6 +69,7 @@ namespace cppa { namespace detail { ...@@ -69,6 +69,7 @@ namespace cppa { namespace detail {
{ "cppa::down_msg", "@down" }, { "cppa::down_msg", "@down" },
{ "cppa::exit_msg", "@exit" }, { "cppa::exit_msg", "@exit" },
{ "cppa::group", "@group" }, { "cppa::group", "@group" },
{ "cppa::group_down_msg", "@group_down" },
{ "cppa::intrusive_ptr<cppa::node_id>", "@proc" }, { "cppa::intrusive_ptr<cppa::node_id>", "@proc" },
{ "cppa::io::accept_handle", "@ac_hdl" }, { "cppa::io::accept_handle", "@ac_hdl" },
{ "cppa::io::connection_handle", "@cn_hdl" }, { "cppa::io::connection_handle", "@cn_hdl" },
...@@ -413,6 +414,14 @@ deserialize_impl(T& dm, deserializer* source) { ...@@ -413,6 +414,14 @@ deserialize_impl(T& dm, deserializer* source) {
dm.reason = source->read<std::uint32_t>(); dm.reason = source->read<std::uint32_t>();
} }
inline void serialize_impl(const group_down_msg& dm, serializer* sink) {
serialize_impl(dm.source, sink);
}
inline void deserialize_impl(group_down_msg& dm, deserializer* source) {
deserialize_impl(dm.source, source);
}
inline void serialize_impl(const timeout_msg& tm, serializer* sink) { inline void serialize_impl(const timeout_msg& tm, serializer* sink) {
sink->write_value(tm.timeout_id); sink->write_value(tm.timeout_id);
} }
...@@ -777,6 +786,7 @@ class utim_impl : public uniform_type_info_map { ...@@ -777,6 +786,7 @@ class utim_impl : public uniform_type_info_map {
*i++ = &m_type_duration; // @duration *i++ = &m_type_duration; // @duration
*i++ = &m_type_exit_msg; // @exit *i++ = &m_type_exit_msg; // @exit
*i++ = &m_type_group; // @group *i++ = &m_type_group; // @group
*i++ = &m_type_group_down; // @group_down
*i++ = &m_type_header; // @header *i++ = &m_type_header; // @header
*i++ = &m_type_i16; // @i16 *i++ = &m_type_i16; // @i16
*i++ = &m_type_i32; // @i32 *i++ = &m_type_i32; // @i32
...@@ -906,6 +916,7 @@ class utim_impl : public uniform_type_info_map { ...@@ -906,6 +916,7 @@ class utim_impl : public uniform_type_info_map {
uti_impl<group> m_type_group; uti_impl<group> m_type_group;
// 10-19 // 10-19
uti_impl<group_down_msg> m_type_group_down;
uti_impl<any_tuple> m_type_tuple; uti_impl<any_tuple> m_type_tuple;
uti_impl<util::duration> m_type_duration; uti_impl<util::duration> m_type_duration;
uti_impl<sync_exited_msg> m_type_sync_exited; uti_impl<sync_exited_msg> m_type_sync_exited;
...@@ -915,9 +926,9 @@ class utim_impl : public uniform_type_info_map { ...@@ -915,9 +926,9 @@ class utim_impl : public uniform_type_info_map {
uti_impl<unit_t> m_type_unit; uti_impl<unit_t> m_type_unit;
uti_impl<atom_value> m_type_atom; uti_impl<atom_value> m_type_atom;
uti_impl<std::string> m_type_str; uti_impl<std::string> m_type_str;
uti_impl<std::u16string> m_type_u16str;
// 20-29 // 20-29
uti_impl<std::u16string> m_type_u16str;
uti_impl<std::u32string> m_type_u32str; uti_impl<std::u32string> m_type_u32str;
default_uniform_type_info<strmap> m_type_strmap; default_uniform_type_info<strmap> m_type_strmap;
uti_impl<bool> m_type_bool; uti_impl<bool> m_type_bool;
...@@ -927,16 +938,16 @@ class utim_impl : public uniform_type_info_map { ...@@ -927,16 +938,16 @@ class utim_impl : public uniform_type_info_map {
int_tinfo<std::int8_t> m_type_i8; int_tinfo<std::int8_t> m_type_i8;
int_tinfo<std::uint8_t> m_type_u8; int_tinfo<std::uint8_t> m_type_u8;
int_tinfo<std::int16_t> m_type_i16; int_tinfo<std::int16_t> m_type_i16;
int_tinfo<std::uint16_t> m_type_u16;
// 30-33 // 30-34
int_tinfo<std::uint16_t> m_type_u16;
int_tinfo<std::int32_t> m_type_i32; int_tinfo<std::int32_t> m_type_i32;
int_tinfo<std::uint32_t> m_type_u32; int_tinfo<std::uint32_t> m_type_u32;
int_tinfo<std::int64_t> m_type_i64; int_tinfo<std::int64_t> m_type_i64;
int_tinfo<std::uint64_t> m_type_u64; int_tinfo<std::uint64_t> m_type_u64;
// both containers are sorted by uniform name // both containers are sorted by uniform name
std::array<pointer, 34> m_builtin_types; std::array<pointer, 35> m_builtin_types;
std::vector<uniform_type_info*> m_user_types; std::vector<uniform_type_info*> m_user_types;
mutable util::shared_spinlock m_lock; mutable util::shared_spinlock m_lock;
......
...@@ -89,6 +89,7 @@ int main() { ...@@ -89,6 +89,7 @@ int main() {
"@header", // message_header "@header", // message_header
"@actor", // actor_ptr "@actor", // actor_ptr
"@group", // group "@group", // group
"@group_down", // group_down_msg
"@channel", // channel "@channel", // channel
"@proc", // intrusive_ptr<node_id> "@proc", // intrusive_ptr<node_id>
"@duration", // util::duration "@duration", // util::duration
......
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