Commit cf8d4947 authored by Dominik Charousset's avatar Dominik Charousset

added new `group` handle

parent 9386fc7e
......@@ -169,6 +169,7 @@ set(LIBCPPA_SRC
src/functor_based_blocking_actor.cpp
src/get_root_uuid.cpp
src/get_mac_addresses.cpp
src/group.cpp
src/group_manager.cpp
src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp
......
......@@ -82,6 +82,7 @@ cppa/exit_reason.hpp
cppa/extend.hpp
cppa/from_string.hpp
cppa/get.hpp
cppa/group.hpp
cppa/guard_expr.hpp
cppa/intrusive/blocking_single_reader_queue.hpp
cppa/intrusive/single_reader_queue.hpp
......@@ -267,6 +268,7 @@ src/functor_based_actor.cpp
src/functor_based_blocking_actor.cpp
src/get_mac_addresses.cpp
src/get_root_uuid.cpp
src/group.cpp
src/group_manager.cpp
src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp
......
......@@ -28,8 +28,8 @@
\******************************************************************************/
#ifndef CPPA_GROUP_HPP
#define CPPA_GROUP_HPP
#ifndef CPPA_ABSTRACT_GROUP_HPP
#define CPPA_ABSTRACT_GROUP_HPP
#include <string>
#include <memory>
......@@ -48,6 +48,7 @@ class peer_connection;
namespace cppa {
class group;
class serializer;
class deserializer;
......@@ -116,9 +117,9 @@ class abstract_group : public abstract_channel {
* the name @p group_name.
* @threadsafe
*/
virtual intrusive_ptr<abstract_group> get(const std::string& group_name) = 0;
virtual group get(const std::string& group_name) = 0;
virtual intrusive_ptr<abstract_group> deserialize(deserializer* source) = 0;
virtual group deserialize(deserializer* source) = 0;
};
......@@ -149,35 +150,6 @@ class abstract_group : public abstract_channel {
*/
virtual subscription subscribe(const channel& who) = 0;
/**
* @brief Get a pointer to the group associated with
* @p group_identifier from the module @p module_name.
* @threadsafe
*/
static intrusive_ptr<abstract_group> get(const std::string& module_name,
const std::string& group_identifier);
/**
* @brief Returns an anonymous group.
*
* Each calls to this member function returns a new instance
* of an anonymous group. Anonymous groups can be used whenever
* a set of actors wants to communicate using an exclusive channel.
*/
static intrusive_ptr<abstract_group> anonymous();
/**
* @brief Add a new group module to the libcppa group management.
* @threadsafe
*/
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:
abstract_group(module_ptr module, std::string group_id);
......@@ -193,7 +165,7 @@ class abstract_group : public abstract_channel {
* @brief A smart pointer type that manages instances of {@link group}.
* @relates group
*/
typedef intrusive_ptr<abstract_group> group_ptr;
typedef intrusive_ptr<abstract_group> abstract_group_ptr;
/**
* @brief Makes *all* local groups accessible via network on address @p addr
......@@ -205,4 +177,4 @@ void publish_local_groups(std::uint16_t port, const char* addr = nullptr);
} // namespace cppa
#endif // CPPA_GROUP_HPP
#endif // CPPA_ABSTRACT_GROUP_HPP
......@@ -179,15 +179,7 @@ class actor : util::comparable<actor>
}
inline bool operator!() const {
return !static_cast<bool>(m_ops.m_ptr);
}
/**
* @brief Queries whether this handle is valid, i.e., points
* to an instance of an untyped actor.
*/
inline bool valid() const {
return static_cast<bool>(m_ops.m_ptr);
return !m_ops.m_ptr;
}
/**
......@@ -204,10 +196,12 @@ class actor : util::comparable<actor>
intptr_t compare(const actor& other) const;
intptr_t compare(const invalid_actor_t&) const;
intptr_t compare(const actor_addr&) const;
inline intptr_t compare(const invalid_actor_t&) const {
return m_ops.m_ptr ? 1 : 0;
}
inline intptr_t compare(const invalid_actor_addr_t&) const {
return compare(invalid_actor);
}
......
......@@ -42,7 +42,10 @@
namespace cppa {
class actor;
class group;
struct invalid_actor_t;
struct invalid_group_t;
namespace detail { class raw_access; }
......@@ -61,8 +64,12 @@ class channel : util::comparable<channel>
channel(const actor&);
channel(const group&);
channel(const invalid_actor_t&);
channel(const invalid_group_t&);
template<typename T>
channel(intrusive_ptr<T> ptr,
typename std::enable_if<
......@@ -72,9 +79,14 @@ class channel : util::comparable<channel>
channel(abstract_channel* ptr);
explicit operator bool() const;
inline explicit operator bool() const {
return static_cast<bool>(m_ptr);
}
inline bool operator!() const {
return !m_ptr;
}
bool operator!() const;
void enqueue(const message_header& hdr, any_tuple msg) const;
......
......@@ -37,7 +37,7 @@ namespace cppa {
// classes
class actor;
class abstract_group;
class group;
class channel;
class node_id;
class behavior;
......@@ -46,6 +46,7 @@ class actor_addr;
class actor_proxy;
class scoped_actor;
class abstract_actor;
class abstract_group;
class blocking_actor;
class message_header;
class partial_function;
......@@ -66,7 +67,7 @@ template<typename> class intrusive_ptr;
template<typename> class weak_intrusive_ptr;
// intrusive pointer typedefs
typedef intrusive_ptr<abstract_group> group_ptr;
typedef intrusive_ptr<abstract_group> abstract_group_ptr;
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
typedef intrusive_ptr<node_id> node_id_ptr;
......
......@@ -48,10 +48,10 @@ class group_manager : public singleton_mixin<group_manager> {
public:
intrusive_ptr<abstract_group> get(const std::string& module_name,
const std::string& group_identifier);
group get(const std::string& module_name,
const std::string& group_identifier);
intrusive_ptr<abstract_group> anonymous();
group anonymous();
void add_module(abstract_group::unique_module_ptr);
......
......@@ -32,9 +32,11 @@
#define CPPA_RAW_ACCESS_HPP
#include "cppa/actor.hpp"
#include "cppa/group.hpp"
#include "cppa/channel.hpp"
#include "cppa/actor_addr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/abstract_group.hpp"
#include "cppa/abstract_channel.hpp"
namespace cppa { namespace detail {
......@@ -55,6 +57,10 @@ class raw_access {
return hdl.m_ptr.get();
}
static abstract_group* get(const group& hdl) {
return hdl.m_ptr.get();
}
static actor unsafe_cast(abstract_actor* ptr) {
return {ptr};
}
......
......@@ -68,7 +68,7 @@ using mapped_type_list = util::type_list<
channel,
down_msg,
exit_msg,
group_ptr,
group,
node_id_ptr,
io::accept_handle,
io::connection_handle,
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_GROUP_HPP
#define CPPA_GROUP_HPP
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_group.hpp"
#include "cppa/util/comparable.hpp"
#include "cppa/util/type_traits.hpp"
namespace cppa {
class channel;
class any_tuple;
class message_header;
struct invalid_group_t { constexpr invalid_group_t() { } };
namespace detail {
class raw_access;
} // namespace detail
/**
* @brief Identifies an invalid {@link group}.
* @relates group
*/
constexpr invalid_group_t invalid_group = invalid_group_t{};
class group : util::comparable<group>
, util::comparable<group, invalid_group_t> {
friend class detail::raw_access;
public:
group() = default;
group(group&&) = default;
group(const group&) = default;
group(const invalid_group_t&);
group& operator=(group&&) = default;
group& operator=(const group&) = default;
group& operator=(const invalid_group_t&);
group(intrusive_ptr<abstract_group> ptr);
inline explicit operator bool() const {
return static_cast<bool>(m_ptr);
}
inline bool operator!() const {
return !static_cast<bool>(m_ptr);
}
/**
* @brief Returns a handle that grants access to
* actor operations such as enqueue.
*/
inline abstract_group* operator->() const {
return m_ptr.get();
}
inline abstract_group& operator*() const {
return *m_ptr;
}
intptr_t compare(const group& other) const;
inline intptr_t compare(const invalid_actor_t&) const {
return m_ptr ? 1 : 0;
}
/**
* @brief Get a pointer to the group associated with
* @p group_identifier from the module @p module_name.
* @threadsafe
*/
static group get(const std::string& module_name,
const std::string& group_identifier);
/**
* @brief Returns an anonymous group.
*
* Each calls to this member function returns a new instance
* of an anonymous group. Anonymous groups can be used whenever
* a set of actors wants to communicate using an exclusive channel.
*/
static group anonymous();
/**
* @brief Add a new group module to the libcppa group management.
* @threadsafe
*/
static void add_module(abstract_group::unique_module_ptr);
/**
* @brief Returns the module associated with @p module_name.
* @threadsafe
*/
static abstract_group::module_ptr get_module(const std::string& module_name);
private:
abstract_group_ptr m_ptr;
};
} // namespace cppa
#endif // CPPA_GROUP_HPP
......@@ -84,10 +84,10 @@ template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn(Ts&&... args);
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_in_group(const group_ptr&, Ts&&... args);
actor spawn_in_group(const group&, Ts&&... args);
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_in_group(const group_ptr&, Ts&&... args);
actor spawn_in_group(const group&, Ts&&... args);
/**
* @brief Base class for local running Actors.
......@@ -129,13 +129,13 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
}
template<spawn_options Options = no_spawn_options, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
actor spawn_in_group(const group& grp, Ts&&... args) {
auto res = cppa::spawn_in_group<make_unbound(Options)>(grp, std::forward<Ts>(args)...);
return eval_opts(Options, std::move(res));
}
template<class Impl, spawn_options Options, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
actor spawn_in_group(const group& grp, Ts&&... args) {
auto res = cppa::spawn_in_group<Impl, make_unbound(Options)>(grp, std::forward<Ts>(args)...);
return eval_opts(Options, std::move(res));
}
......@@ -197,7 +197,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
* The group will be unsubscribed if the actor finishes execution.
* @param what Group instance that should be joined.
*/
void join(const group_ptr& what);
void join(const group& what);
/**
* @brief Causes this actor to leave the group @p what.
......@@ -205,7 +205,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
* @note Groups are leaved automatically if the Actor finishes
* execution.
*/
void leave(const group_ptr& what);
void leave(const group& what);
/**
* @brief Finishes execution of this actor after any currently running
......@@ -290,7 +290,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
/**
* @brief Returns all joined groups of this actor.
*/
std::vector<group_ptr> joined_groups() const;
std::vector<group> joined_groups() const;
/**
* @brief Creates a {@link response_promise} to allow actors to response
......@@ -423,7 +423,7 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
mailbox_element* m_current_node;
// {group => subscription} map of all joined groups
std::map<group_ptr, abstract_group::subscription> m_subscriptions;
std::map<group, abstract_group::subscription> m_subscriptions;
// set by quit
std::uint32_t m_planned_exit_reason;
......
......@@ -187,7 +187,7 @@ actor spawn(Ts&&... args) {
* @note The spawned has joined the group before this function returns.
*/
template<spawn_options Opts, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
actor spawn_in_group(const group& grp, Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "too few arguments provided");
using base_class = typename std::conditional<
has_blocking_api_flag(Opts),
......@@ -208,7 +208,7 @@ actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
* @note The spawned has joined the group before this function returns.
*/
template<class Impl, spawn_options Opts, typename... Ts>
actor spawn_in_group(const group_ptr& grp, Ts&&... args) {
actor spawn_in_group(const group& grp, Ts&&... args) {
return detail::spawn_fwd_args<Impl, Opts>(
[&](local_actor* ptr) { ptr->join(grp); },
std::forward<Ts>(args)...);
......
......@@ -35,15 +35,16 @@
#include "cppa/atom.hpp" // included for to_string(atom_value)
#include "cppa/actor.hpp"
#include "cppa/abstract_group.hpp"
#include "cppa/group.hpp"
#include "cppa/object.hpp"
#include "cppa/channel.hpp"
#include "cppa/node_id.hpp"
#include "cppa/anything.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/abstract_group.hpp"
#include "cppa/message_header.hpp"
#include "cppa/uniform_type_info.hpp"
namespace std { class exception; }
......@@ -76,7 +77,7 @@ inline std::string to_string(const actor_addr& what) {
return detail::to_string_impl(what);
}
inline std::string to_string(const group_ptr& what) {
inline std::string to_string(const group& what) {
return detail::to_string_impl(what);
}
......
......@@ -165,7 +165,7 @@ struct is_builtin {
any_tuple,
message_header,
actor,
group_ptr,
group,
channel,
node_id_ptr
>::value;
......
......@@ -15,7 +15,7 @@ using namespace cppa;
ChatWidget::ChatWidget(QWidget* parent, Qt::WindowFlags f)
: super(parent, f), m_input(nullptr), m_output(nullptr) {
set_message_handler (
on(atom("join"), arg_match) >> [=](const group_ptr& what) {
on(atom("join"), arg_match) >> [=](const group& what) {
if (m_chatroom) {
send(m_chatroom, m_name + " has left the chatroom");
self->leave(m_chatroom);
......@@ -51,7 +51,7 @@ void ChatWidget::sendChatMessage() {
if (line.startsWith('/')) {
match_split(line.midRef(1).toUtf8().constData(), ' ') (
on("join", arg_match) >> [=](const string& mod, const string& g) {
group_ptr gptr;
group gptr;
try { gptr = abstract_group::get(mod, g); }
catch (exception& e) {
print("*** exception: " + QString::fromUtf8((e.what())));
......@@ -108,7 +108,7 @@ void ChatWidget::joinGroup() {
}
string mod = gname.left(pos).toUtf8().constData();
string gid = gname.midRef(pos+1).toUtf8().constData();
group_ptr gptr;
group gptr;
try {
auto gptr = abstract_group::get(mod, gid);
send_as(as_actor(), as_actor(), atom("join"), gptr);
......
......@@ -51,6 +51,6 @@ class ChatWidget : public cppa::actor_widget_mixin<QWidget> {
QLineEdit* m_input;
QTextEdit* m_output;
std::string m_name;
cppa::group_ptr m_chatroom;
cppa::group m_chatroom;
};
......@@ -60,7 +60,7 @@ int main(int argc, char** argv) {
on_opt0('h', "help", &desc, "print help") >> print_desc_and_exit(&desc)
);
group_ptr gptr;
group gptr;
// evaluate group parameter
if (!group_id.empty()) {
auto p = group_id.find(':');
......
......@@ -49,7 +49,7 @@ void client(event_based_actor* self, const string& name) {
self->send(dest, name + ": " + message);
}
},
on(atom("join"), arg_match) >> [=](const group_ptr& what) {
on(atom("join"), arg_match) >> [=](const group& what) {
for (auto g : self->joined_groups()) {
cout << "*** leave " << to_string(g) << endl;
self->send(self, g, name + " has left the chatroom");
......@@ -102,7 +102,7 @@ int main(int argc, char** argv) {
}
else {
try {
auto g = abstract_group::get(group_id.substr(0, p),
auto g = group::get(group_id.substr(0, p),
group_id.substr(p + 1));
anon_send(client_actor, atom("join"), g);
}
......@@ -120,7 +120,7 @@ int main(int argc, char** argv) {
match_each (lines, eof, split_line) (
on("/join", arg_match) >> [&](const string& mod, const string& id) {
try {
anon_send(client_actor, atom("join"), abstract_group::get(mod, id));
anon_send(client_actor, atom("join"), group::get(mod, id));
}
catch (exception& e) {
cerr << "*** exception: " << to_verbose_string(e) << endl;
......
......@@ -41,23 +41,6 @@
namespace cppa {
intrusive_ptr<abstract_group> abstract_group::get(const std::string& arg0,
const std::string& arg1) {
return get_group_manager()->get(arg0, arg1);
}
intrusive_ptr<abstract_group> abstract_group::anonymous() {
return get_group_manager()->anonymous();
}
void abstract_group::add_module(abstract_group::unique_module_ptr ptr) {
get_group_manager()->add_module(std::move(ptr));
}
abstract_group::module_ptr abstract_group::get_module(const std::string& module_name) {
return get_group_manager()->get_module(module_name);
}
abstract_group::subscription::subscription(const channel& s,
const intrusive_ptr<abstract_group>& g)
: m_subscriber(s), m_group(g) { }
......@@ -91,7 +74,7 @@ struct group_nameserver : event_based_actor {
behavior make_behavior() override {
return (
on(atom("GET_GROUP"), arg_match) >> [](const std::string& name) {
return make_cow_tuple(atom("GROUP"), abstract_group::get("local", name));
return make_cow_tuple(atom("GROUP"), group::get("local", name));
},
on(atom("SHUTDOWN")) >> [=] {
quit();
......
......@@ -57,10 +57,6 @@ intptr_t actor::compare(const actor& other) const {
return channel::compare(m_ops.m_ptr.get(), other.m_ops.m_ptr.get());
}
intptr_t actor::compare(const invalid_actor_t&) const {
return valid() ? 1 : 0;
}
intptr_t actor::compare(const actor_addr& other) const {
return m_ops.compare(*other);
}
......
......@@ -36,9 +36,13 @@
namespace cppa {
channel::channel(const actor& other) : m_ptr(detail::raw_access::get(other)) { }
channel::channel(const group& other) : m_ptr(detail::raw_access::get(other)) { }
channel::channel(const invalid_actor_t&) : m_ptr(nullptr) { }
channel::channel(const actor& other) : m_ptr(detail::raw_access::get(other)) { }
channel::channel(const invalid_group_t&) : m_ptr(nullptr) { }
intptr_t channel::compare(const abstract_channel* lhs, const abstract_channel* rhs) {
return reinterpret_cast<intptr_t>(lhs) - reinterpret_cast<intptr_t>(rhs);
......@@ -46,14 +50,6 @@ intptr_t channel::compare(const abstract_channel* lhs, const abstract_channel* r
channel::channel(abstract_channel* ptr) : m_ptr(ptr) { }
channel::operator bool() const {
return static_cast<bool>(m_ptr);
}
bool channel::operator!() const {
return !m_ptr;
}
void channel::enqueue(const message_header& hdr, any_tuple msg) const {
if (m_ptr) m_ptr->enqueue(hdr, std::move(msg));
}
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/group.hpp"
#include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/singletons.hpp"
#include "cppa/detail/group_manager.hpp"
namespace cppa {
group::group(const invalid_group_t&) : m_ptr(nullptr) { }
group::group(abstract_group_ptr ptr) : m_ptr(std::move(ptr)) { }
group& group::operator=(const invalid_group_t&) {
m_ptr.reset();
return *this;
}
intptr_t group::compare(const group& other) const {
return channel::compare(m_ptr.get(), other.m_ptr.get());
}
group group::get(const std::string& arg0, const std::string& arg1) {
return get_group_manager()->get(arg0, arg1);
}
group group::anonymous() {
return get_group_manager()->anonymous();
}
void group::add_module(abstract_group::unique_module_ptr ptr) {
get_group_manager()->add_module(std::move(ptr));
}
abstract_group::module_ptr group::get_module(const std::string& module_name) {
return get_group_manager()->get_module(module_name);
}
} // namespace cppa
......@@ -36,6 +36,7 @@
#include <condition_variable>
#include "cppa/cppa.hpp"
#include "cppa/group.hpp"
#include "cppa/to_string.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/serializer.hpp"
......@@ -292,11 +293,11 @@ class local_group_module : public abstract_group::module {
: super("local"), m_process(node_id::get())
, m_actor_utype(uniform_typeid<actor>()){ }
group_ptr get(const string& identifier) {
group get(const string& identifier) override {
shared_guard guard(m_instances_mtx);
auto i = m_instances.find(identifier);
if (i != m_instances.end()) {
return i->second;
return {i->second};
}
else {
auto tmp = make_counted<local_group>(true, this, identifier);
......@@ -304,18 +305,18 @@ class local_group_module : public abstract_group::module {
upgrade_guard uguard(guard);
auto p = m_instances.insert(make_pair(identifier, tmp));
// someone might preempt us
return p.first->second;
return {p.first->second};
}
}
}
intrusive_ptr<abstract_group> deserialize(deserializer* source) {
group deserialize(deserializer* source) override {
// deserialize {identifier, process_id, node_id}
auto identifier = source->read<string>();
// deserialize broker
actor broker;
m_actor_utype->deserialize(&broker, source);
if (!broker) return nullptr;
if (!broker) return invalid_group;
if (!broker->is_remote()) {
return this->get(identifier);
}
......@@ -323,15 +324,15 @@ class local_group_module : public abstract_group::module {
shared_guard guard(m_proxies_mtx);
auto i = m_proxies.find(broker);
if (i != m_proxies.end()) {
return i->second;
return {i->second};
}
else {
local_group_ptr tmp(new local_group_proxy(broker, this,
identifier));
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;
return {p.first->second};
}
}
}
......@@ -385,10 +386,10 @@ class remote_group : public abstract_group {
void group_down() {
CPPA_LOG_TRACE("");
group_ptr _this{this};
group this_group{this};
m_decorated->send_all_subscribers({invalid_actor_addr, nullptr},
make_any_tuple(atom("GROUP_DOWN"),
_this));
this_group));
}
private:
......@@ -425,13 +426,13 @@ class shared_map : public ref_counted {
return result;
}
group_ptr peek(const string& key) {
group peek(const string& key) {
lock_type guard(m_mtx);
auto i = m_instances.find(key);
if (i != m_instances.end()) {
return i->second;
return {i->second};
}
return nullptr;
return invalid_group;
}
void put(const string& key, const remote_group_ptr& ptr) {
......@@ -460,10 +461,10 @@ class remote_group_module : public abstract_group::module {
remote_group_module() : super("remote") {
auto sm = make_counted<shared_map>();
abstract_group::module_ptr _this = this;
abstract_group::module_ptr this_group{this};
m_map = sm;
m_map->m_worker = spawn<hidden>([=](event_based_actor* self) -> behavior {
CPPA_LOGC_TRACE(detail::demangle(typeid(*_this)),
CPPA_LOGC_TRACE(detail::demangle(typeid(*this_group)),
"remote_group_module$worker",
"");
typedef map<string, pair<actor, vector<pair<string, remote_group_ptr>>>>
......@@ -502,10 +503,10 @@ class remote_group_module : public abstract_group::module {
}
}
self->timed_sync_send(nameserver, chrono::seconds(10), atom("GET_GROUP"), name).then (
on(atom("GROUP"), arg_match) >> [&](const group_ptr& g) {
auto gg = dynamic_cast<local_group*>(g.get());
on(atom("GROUP"), arg_match) >> [&](const group& g) {
auto gg = dynamic_cast<local_group*>(detail::raw_access::get(g));
if (gg) {
auto rg = make_counted<remote_group>(_this, key, gg);
auto rg = make_counted<remote_group>(this_group, key, gg);
sm->put(key, rg);
(*peers)[authority].second.push_back(make_pair(key, rg));
}
......@@ -545,11 +546,11 @@ class remote_group_module : public abstract_group::module {
});
}
intrusive_ptr<abstract_group> get(const std::string& group_name) {
return m_map->get(group_name);
group get(const std::string& group_name) {
return {m_map->get(group_name)};
}
intrusive_ptr<abstract_group> deserialize(deserializer* source) {
group deserialize(deserializer* source) {
return get(source->read<string>());
}
......@@ -593,14 +594,14 @@ group_manager::group_manager() {
m_mmap.insert(make_pair(string("remote"), move(ptr)));
}
intrusive_ptr<abstract_group> group_manager::anonymous() {
group group_manager::anonymous() {
string id = "__#";
id += std::to_string(++m_ad_hoc_id);
return get_module("local")->get(id);
}
intrusive_ptr<abstract_group> group_manager::get(const string& module_name,
const string& group_identifier) {
group group_manager::get(const string& module_name,
const string& group_identifier) {
auto mod = get_module(module_name);
if (mod) {
return mod->get(group_identifier);
......
......@@ -97,18 +97,20 @@ void local_actor::demonitor(const actor_addr& whom) {
void local_actor::on_exit() { }
void local_actor::join(const group_ptr& what) {
void local_actor::join(const group& what) {
CPPA_LOG_TRACE(CPPA_TSARG(what));
if (what && m_subscriptions.count(what) == 0) {
CPPA_LOG_DEBUG("join group: " << to_string(what));
m_subscriptions.insert(std::make_pair(what, what->subscribe(this)));
}
}
void local_actor::leave(const group_ptr& what) {
void local_actor::leave(const group& what) {
if (what) m_subscriptions.erase(what);
}
std::vector<group_ptr> local_actor::joined_groups() const {
std::vector<group_ptr> result;
std::vector<group> local_actor::joined_groups() const {
std::vector<group> result;
for (auto& kvp : m_subscriptions) {
result.emplace_back(kvp.first);
}
......
......@@ -35,11 +35,12 @@
#include <algorithm>
#include <type_traits>
#include "cppa/abstract_group.hpp"
#include "cppa/group.hpp"
#include "cppa/logging.hpp"
#include "cppa/announce.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_header.hpp"
#include "cppa/abstract_group.hpp"
#include "cppa/actor_namespace.hpp"
#include "cppa/util/duration.hpp"
......@@ -67,7 +68,7 @@ namespace cppa { namespace detail {
{ "cppa::channel", "@channel" },
{ "cppa::down_msg", "@down" },
{ "cppa::exit_msg", "@exit" },
{ "cppa::intrusive_ptr<cppa::group>", "@group" },
{ "cppa::group", "@group" },
{ "cppa::intrusive_ptr<cppa::node_id>", "@proc" },
{ "cppa::io::accept_handle", "@ac_hdl" },
{ "cppa::io::connection_handle", "@cn_hdl" },
......@@ -193,25 +194,26 @@ void deserialize_impl(actor& ptr, deserializer* source) {
ptr = detail::raw_access::unsafe_cast(addr);
}
void serialize_impl(const group_ptr& ptr, serializer* sink) {
if (ptr == nullptr) {
void serialize_impl(const group& gref, serializer* sink) {
if (!gref) {
CPPA_LOGF_DEBUG("serialized an invalid group");
// write an empty string as module name
std::string empty_string;
sink->write_value(empty_string);
}
else {
sink->write_value(ptr->module_name());
ptr->serialize(sink);
sink->write_value(gref->module_name());
gref->serialize(sink);
}
}
void deserialize_impl(group_ptr& ptrref, deserializer* source) {
void deserialize_impl(group& gref, deserializer* source) {
auto modname = source->read<std::string>();
if (modname.empty()) ptrref.reset();
else ptrref = abstract_group::get_module(modname)->deserialize(source);
if (modname.empty()) gref = invalid_group;
else gref = group::get_module(modname)->deserialize(source);
}
void serialize_impl(const channel& ptr, serializer* sink) {
void serialize_impl(const channel& chref, serializer* sink) {
// channel is an abstract base class that's either an actor or a group
// to indicate that, we write a flag first, that is
// 0 if ptr == nullptr
......@@ -221,9 +223,14 @@ void serialize_impl(const channel& ptr, serializer* sink) {
auto wr_nullptr = [&] {
sink->write_value(flag);
};
if (ptr == nullptr) wr_nullptr();
if (!chref) {
// invalid channel
wr_nullptr();
}
else {
auto rptr = detail::raw_access::get(ptr);
// raw pointer
auto rptr = detail::raw_access::get(chref);
// raw actor pointer
auto aptr = dynamic_cast<abstract_actor*>(rptr);
if (aptr != nullptr) {
flag = 1;
......@@ -231,11 +238,12 @@ void serialize_impl(const channel& ptr, serializer* sink) {
serialize_impl(detail::raw_access::unsafe_cast(aptr), sink);
}
else {
auto gptr = group_ptr{dynamic_cast<abstract_group*>(rptr)};
if (gptr != nullptr) {
// get raw group pointer and store it inside a group handle
group tmp{dynamic_cast<abstract_group*>(rptr)};
if (tmp) {
flag = 2;
sink->write_value(flag);
serialize_impl(gptr, sink);
serialize_impl(tmp, sink);
}
else {
CPPA_LOGF_ERROR("ptr is neither an actor nor a group");
......@@ -259,7 +267,7 @@ void deserialize_impl(channel& ptrref, deserializer* source) {
break;
}
case 2: {
group_ptr tmp;
group tmp;
deserialize_impl(tmp, source);
ptrref = tmp;
break;
......@@ -890,7 +898,7 @@ class utim_impl : public uniform_type_info_map {
buffer_type_info_impl m_type_buffer;
uti_impl<actor> m_type_actor;
uti_impl<actor_addr> m_type_actor_addr;
uti_impl<group_ptr> m_type_group;
uti_impl<group> m_type_group;
// 10-19
uti_impl<any_tuple> m_type_tuple;
......
......@@ -32,9 +32,10 @@ void reflector(event_based_actor* self) {
);
}
void spawn5_server_impl(event_based_actor* self, actor client, group_ptr grp) {
void spawn5_server_impl(event_based_actor* self, actor client, group grp) {
CPPA_LOGF_TRACE(CPPA_TARG(client, to_string)
<< ", " << CPPA_TARG(grp, to_string));
CPPA_CHECK(grp != invalid_group);
self->spawn_in_group(grp, reflector);
self->spawn_in_group(grp, reflector);
CPPA_LOGF_INFO("send {'Spawn5'} and await {'ok', actor_vector}");
......@@ -104,11 +105,11 @@ void spawn5_server_impl(event_based_actor* self, actor client, group_ptr grp) {
// receive seven reply messages (2 local, 5 remote)
void spawn5_server(event_based_actor* self, actor client, bool inverted) {
if (!inverted) spawn5_server_impl(self, client, abstract_group::get("local", "foobar"));
if (!inverted) spawn5_server_impl(self, client, group::get("local", "foobar"));
else {
CPPA_LOGF_INFO("request group");
self->sync_send(client, atom("GetGroup")).then (
[=](const group_ptr& remote_group) {
[=](const group& remote_group) {
spawn5_server_impl(self, client, remote_group);
}
);
......@@ -117,11 +118,11 @@ void spawn5_server(event_based_actor* self, actor client, bool inverted) {
void spawn5_client(event_based_actor* self) {
self->become (
on(atom("GetGroup")) >> []() -> group_ptr {
on(atom("GetGroup")) >> []() -> group {
CPPA_LOGF_INFO("received {'GetGroup'}");
return abstract_group::get("local", "foobar");
return group::get("local", "foobar");
},
on(atom("Spawn5"), arg_match) >> [=](const group_ptr& grp) -> any_tuple {
on(atom("Spawn5"), arg_match) >> [=](const group& grp) -> any_tuple {
CPPA_LOGF_INFO("received {'Spawn5'}");
actor_vector vec;
for (int i = 0; i < 5; ++i) {
......@@ -205,6 +206,7 @@ class client : public event_based_actor {
CPPA_PRINT("test group communication via network");
sync_send(m_server, atom("GClient")).then(
on(atom("GClient"), arg_match) >> [=](actor gclient) {
CPPA_CHECKPOINT();
auto s5a = spawn<monitored>(spawn5_server, gclient, false);
await_down(this, s5a, [=]{
test_group_comm_inverted();
......@@ -217,6 +219,7 @@ class client : public event_based_actor {
CPPA_PRINT("test group communication via network (inverted setup)");
become (
on(atom("GClient")) >> [=]() -> any_tuple {
CPPA_CHECKPOINT();
auto cptr = last_sender();
auto s5c = spawn<monitored>(spawn5_client);
// set next behavior
......
......@@ -88,7 +88,7 @@ int main() {
"@tuple", // any_tuple
"@header", // message_header
"@actor", // actor_ptr
"@group", // group_ptr
"@group", // group
"@channel", // channel
"@proc", // intrusive_ptr<node_id>
"@duration", // util::duration
......@@ -151,7 +151,7 @@ int main() {
std::string, std::u16string, std::u32string,
float, double,
atom_value, any_tuple, message_header,
actor, group_ptr,
actor, group,
channel, node_id_ptr
>::arr;
......@@ -175,7 +175,7 @@ int main() {
uniform_typeid<any_tuple>(),
uniform_typeid<message_header>(),
uniform_typeid<actor>(),
uniform_typeid<group_ptr>(),
uniform_typeid<group>(),
uniform_typeid<channel>(),
uniform_typeid<node_id_ptr>()
};
......
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