Commit aee566e5 authored by Dominik Charousset's avatar Dominik Charousset

Return expected<group> from modules, fix #482

parent 8d6d3be3
......@@ -12,6 +12,14 @@ if(WIN32 AND NOT CAF_BUILD_STATIC_ONLY)
set(CAF_BUILD_STATIC_ONLY yes)
endif()
################################################################################
# set prefix paths if available #
################################################################################
if(DEFINED CAF_QT_PREFIX_PATH)
set(CMAKE_PREFIX_PATH "${CAF_QT_PREFIX_PATH}")
endif()
################################################################################
# make sure all variables are set to "no" if undefined for summary output #
################################################################################
......
......@@ -42,6 +42,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--lib-dir=DIR library directory [build/lib]
--with-clang=FILE path to clang++ executable
--with-gcc=FILE path to g++ executable
--with-qt-prefix=PATH prefix path for Qt5 cmake modules
--dual-build build using both gcc and clang
--build-static build as static and shared library
--build-static-only build as static library only
......@@ -288,6 +289,9 @@ while [ $# -ne 0 ]; do
--with-gcc=*)
gcc=$optarg
;;
--with-qt-prefix=*)
append_cache_entry CAF_QT_PREFIX_PATH STRING "$optarg"
;;
--build-type=*)
append_cache_entry CMAKE_BUILD_TYPE STRING $optarg
;;
......
......@@ -73,7 +73,7 @@ if(NOT CAF_NO_PROTOBUF_EXAMPLES)
endif()
if(NOT CAF_NO_QT_EXAMPLES)
find_package(Qt5 COMPONENTS Core Gui Widgets)
find_package(Qt5 COMPONENTS Core Gui Widgets QUIET)
if(Qt5_FOUND)
message(STATUS "Found Qt5")
#include(${QT_USE_FILE})
......
......@@ -15,118 +15,127 @@ using namespace std;
using namespace caf;
ChatWidget::ChatWidget(QWidget* parent, Qt::WindowFlags f)
: super(parent, f), input_(nullptr), output_(nullptr) {
set_message_handler ([=](local_actor* self) -> message_handler {
return {
on(atom("join"), arg_match) >> [=](const group& what) {
if (chatroom_) {
self->send(chatroom_, name_ + " has left the chatroom");
self->leave(chatroom_);
}
self->join(what);
print(("*** joined " + to_string(what)).c_str());
chatroom_ = what;
self->send(what, name_ + " has entered the chatroom");
},
on(atom("setName"), arg_match) >> [=](string& name) {
self->send(chatroom_, name_ + " is now known as " + name);
name_ = std::move(name);
print("*** changed name to "
+ QString::fromUtf8(name_.c_str()));
},
on(atom("quit")) >> [=] {
close(); // close widget
},
[=](const string& txt) {
// don't print own messages
if (self != self->current_sender()) {
print(QString::fromUtf8(txt.c_str()));
}
},
[=](const group_down_msg& gdm) {
print("*** chatroom offline: "
+ QString::fromUtf8(to_string(gdm.source).c_str()));
}
};
});
: super(parent, f),
input_(nullptr),
output_(nullptr) {
// nop
}
ChatWidget::~ChatWidget() {
// nop
}
void ChatWidget::init(actor_system& system) {
super::init(system);
set_message_handler ([=](actor_companion* self) -> message_handler {
return {
[=](join_atom, const group& what) {
if (chatroom_) {
self->send(chatroom_, name_ + " has left the chatroom");
self->leave(chatroom_);
}
self->join(what);
print(("*** joined " + to_string(what)).c_str());
chatroom_ = what;
self->send(what, name_ + " has entered the chatroom");
},
[=](set_name_atom, string& name) {
self->send(chatroom_, name_ + " is now known as " + name);
name_ = std::move(name);
print("*** changed name to " + QString::fromUtf8(name_.c_str()));
},
[=](quit_atom) {
quit_and_close();
},
[=](const string& txt) {
// don't print own messages
if (self != self->current_sender())
print(QString::fromUtf8(txt.c_str()));
},
[=](const group_down_msg& gdm) {
print("*** chatroom offline: "
+ QString::fromUtf8(to_string(gdm.source).c_str()));
}
};
});
}
void ChatWidget::sendChatMessage() {
auto cleanup = detail::make_scope_guard([=] {
input()->setText(QString());
auto cleanup = detail::make_scope_guard([=] {
input()->setText(QString());
});
QString line = input()->text();
if (line.startsWith('/')) {
vector<string> words;
split(words, line.midRef(1).toUtf8().constData(), is_any_of(" "));
message_builder mb;
if (words.size() > 1) {
// convert first word to an atom
mb.append(atom_from_string(words.front()))
.append(words.begin() + 1, words.end());
};
auto res = mb.apply({
[=](join_atom, const string& mod, const string& g) {
auto x = system().groups().get(mod, g);
if (! x)
print("*** error: "
+ QString::fromUtf8(system().render(x.error()).c_str()));
else
self()->send(self(), atom("join"), std::move(*x));
},
[=](set_name_atom atm, string& name) {
send_as(as_actor(), as_actor(), atm, std::move(name));
}
});
QString line = input()->text();
if (line.startsWith('/')) {
vector<string> words;
split(words, line.midRef(1).toUtf8().constData(), is_any_of(" "));
message_builder(words.begin(), words.end()).apply({
on("join", arg_match) >> [=](const string& mod, const string& g) {
group gptr;
try { gptr = group::get(mod, g); }
catch (exception& e) {
print("*** exception: " + QString::fromUtf8((e.what())));
}
if (gptr) {
send_as(as_actor(), as_actor(), atom("join"), gptr);
}
},
on("setName", arg_match) >> [=](const string& str) {
send_as(as_actor(), as_actor(), atom("setName"), str);
},
others() >> [=] {
print("*** list of commands:\n"
"/join <module> <group id>\n"
"/setName <new name>\n");
}
});
return;
}
if (name_.empty()) {
print("*** please set a name before sending messages");
return;
}
if (! chatroom_) {
print("*** no one is listening... please join a group");
return;
}
string msg = name_;
msg += ": ";
msg += line.toUtf8().constData();
print("<you>: " + input()->text());
send_as(as_actor(), chatroom_, std::move(msg));
if (! res)
print("*** list of commands:\n"
"/join <module> <group id>\n"
"/setName <new name>\n");
return;
}
if (name_.empty()) {
print("*** please set a name before sending messages");
return;
}
if (! chatroom_) {
print("*** no one is listening... please join a group");
return;
}
string msg = name_;
msg += ": ";
msg += line.toUtf8().constData();
print("<you>: " + input()->text());
send_as(as_actor(), chatroom_, std::move(msg));
}
void ChatWidget::joinGroup() {
if (name_.empty()) {
QMessageBox::information(this,
"No Name, No Chat",
"Please set a name first.");
return;
}
auto gname = QInputDialog::getText(this,
"Join Group",
"Please enter a group as <module>:<id>",
QLineEdit::Normal,
"remote:chatroom@localhost:4242");
int pos = gname.indexOf(':');
if (pos < 0) {
QMessageBox::warning(this, "Not a Group", "Invalid format");
return;
}
string mod = gname.left(pos).toUtf8().constData();
string gid = gname.midRef(pos+1).toUtf8().constData();
try {
auto gptr = group::get(mod, gid);
send_as(as_actor(), as_actor(), atom("join"), gptr);
}
catch (exception& e) {
QMessageBox::critical(this, "Exception", e.what());
}
if (name_.empty()) {
QMessageBox::information(this, "No Name, No Chat",
"Please set a name first.");
return;
}
auto gname = QInputDialog::getText(this,
"Join Group",
"Please enter a group as <module>:<id>",
QLineEdit::Normal,
"remote:chatroom@localhost:4242");
int pos = gname.indexOf(':');
if (pos < 0) {
QMessageBox::warning(this, "Not a Group", "Invalid format");
return;
}
string mod = gname.left(pos).toUtf8().constData();
string gid = gname.midRef(pos+1).toUtf8().constData();
auto x = system().groups().get(mod, gid);
if (! x)
QMessageBox::critical(this, "Error", system().render(x.error()).c_str());
else
self()->send(self(), join_atom::value, std::move(*x));
}
void ChatWidget::changeName() {
auto name = QInputDialog::getText(this, "Change Name", "Please enter a new name");
if (! name.isEmpty()) {
send_as(as_actor(), as_actor(), atom("setName"), name.toUtf8().constData());
}
auto name = QInputDialog::getText(this, "Change Name",
"Please enter a new name");
if (! name.isEmpty())
send_as(as_actor(), as_actor(), atom("setName"), name.toUtf8().constData());
}
......@@ -10,49 +10,62 @@ CAF_PUSH_WARNINGS
CAF_POP_WARNINGS
class ChatWidget : public caf::mixin::actor_widget<QWidget> {
private:
// -- Qt boilerplate code ----------------------------------------------------
Q_OBJECT
typedef caf::mixin::actor_widget<QWidget> super;
Q_OBJECT
public:
// -- member types -----------------------------------------------------------
using super = caf::mixin::actor_widget<QWidget>;
using set_name_atom = caf::atom_constant<caf::atom("setName")>;
using quit_atom = caf::atom_constant<caf::atom("quit")>;
ChatWidget(QWidget* parent = nullptr, Qt::WindowFlags f = 0);
ChatWidget(QWidget* parent = nullptr, Qt::WindowFlags f = 0);
public slots:
~ChatWidget();
void sendChatMessage();
void joinGroup();
void changeName();
void init(caf::actor_system& system);
public slots:
void sendChatMessage();
void joinGroup();
void changeName();
private:
template<typename T>
T* get(T*& member, const char* name) {
if (member == nullptr) {
member = findChild<T*>(name);
if (member == nullptr)
throw std::runtime_error("unable to find child: "
+ std::string(name));
}
return member;
template<typename T>
T* get(T*& member, const char* name) {
if (member == nullptr) {
member = findChild<T*>(name);
if (member == nullptr)
throw std::runtime_error("unable to find child: " + std::string(name));
}
return member;
}
inline QLineEdit* input() {
return get(input_, "input");
}
inline QLineEdit* input() {
return get(input_, "input");
}
inline QTextEdit* output() {
return get(output_, "output");
}
inline QTextEdit* output() {
return get(output_, "output");
}
inline void print(const QString& what) {
output()->append(what);
}
inline void print(const QString& what) {
output()->append(what);
}
QLineEdit* input_;
QTextEdit* output_;
std::string name_;
caf::group chatroom_;
caf::actor_system& system() {
return self()->home_system();
}
QLineEdit* input_;
QTextEdit* output_;
std::string name_;
caf::group chatroom_;
};
......@@ -18,7 +18,16 @@
<property name="spacing">
<number>0</number>
</property>
<property name="margin">
<property name="leftMargin">
<number>0</number>
</property>
<property name="topMargin">
<number>0</number>
</property>
<property name="rightMargin">
<number>0</number>
</property>
<property name="bottomMargin">
<number>0</number>
</property>
<item>
......@@ -62,7 +71,16 @@
<property name="spacing">
<number>0</number>
</property>
<property name="margin">
<property name="leftMargin">
<number>0</number>
</property>
<property name="topMargin">
<number>0</number>
</property>
<property name="rightMargin">
<number>0</number>
</property>
<property name="bottomMargin">
<number>0</number>
</property>
</layout>
......@@ -116,7 +134,7 @@
<connections>
<connection>
<sender>actionJoin_Group</sender>
<signal>activated()</signal>
<signal>triggered()</signal>
<receiver>chatwidget</receiver>
<slot>joinGroup()</slot>
<hints>
......@@ -132,7 +150,7 @@
</connection>
<connection>
<sender>actionSet_Name</sender>
<signal>activated()</signal>
<signal>triggered()</signal>
<receiver>chatwidget</receiver>
<slot>changeName()</slot>
<hints>
......
......@@ -18,6 +18,7 @@
#include <cstdlib>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
CAF_PUSH_WARNINGS
#include <QMainWindow>
......@@ -28,57 +29,55 @@ CAF_POP_WARNINGS
using namespace std;
using namespace caf;
int main(int argc, char** argv) {
string name;
string group_id;
auto res = message_builder(argv + 1, argv + argc).extract_opts({
{"name,n", "set chat name", name},
{"group,g", "join chat group", group_id}
});
if (! res.error.empty()) {
cerr << res.error << endl;
return 1;
}
if (! res.remainder.empty()) {
std::cerr << res.helptext << std::endl;
return 1;
}
if (res.opts.count("help") > 0) {
cout << res.helptext << endl;
return 0;
class config : public actor_system_config {
public:
std::string name;
std::string group_id;
config(int argc, char** argv) {
opt_group{custom_options_, "global"}
.add(name, "name,n", "set name")
.add(group_id, "group,g", "join group (format: <module>:<id>");
parse(argc, argv);
load<io::middleman>();
}
group gptr;
// evaluate group parameter
if (! group_id.empty()) {
auto p = group_id.find(':');
};
int main(int argc, char** argv) {
config cfg{argc, argv};
actor_system system{cfg};
auto name = cfg.name;
group grp;
// evaluate group parameters
if (! cfg.group_id.empty()) {
auto p = cfg.group_id.find(':');
if (p == std::string::npos) {
cerr << "*** error parsing argument " << group_id
<< ", expected format: <module_name>:<group_id>";
cerr << "*** error parsing argument " << cfg.group_id
<< ", expected format: <module_name>:<group_id>";
} else {
try {
gptr = group::get(group_id.substr(0, p), group_id.substr(p + 1));
} catch (exception& e) {
cerr << "*** exception: group::get(\"" << group_id.substr(0, p)
<< "\", \"" << group_id.substr(p + 1) << "\") failed; "
<< to_verbose_string(e) << endl;
auto module = cfg.group_id.substr(0, p);
auto group_uri = cfg.group_id.substr(p + 1);
auto g = system.groups().get(module, group_uri);
if (! g) {
cerr << "*** unable to get group " << group_uri
<< " from module " << module << ": "
<< system.render(g.error()) << endl;
return -1;
}
grp = std::move(*g);
}
}
QApplication app(argc, argv);
QApplication app{argc, argv};
app.setQuitOnLastWindowClosed(true);
QMainWindow mw;
Ui::ChatWindow helper;
helper.setupUi(&mw);
helper.chatwidget->init(system);
auto client = helper.chatwidget->as_actor();
if (! name.empty()) {
if (! name.empty())
send_as(client, client, atom("setName"), move(name));
}
if (gptr) {
send_as(client, client, atom("join"), gptr);
}
if (grp)
send_as(client, client, atom("join"), std::move(grp));
mw.show();
auto app_res = app.exec();
await_all_actors_done();
shutdown();
return app_res;
return app.exec();
}
......@@ -94,9 +94,7 @@ void caf_main(actor_system& system, const config& cfg) {
} else {
auto module = cfg.group_id.substr(0, p);
auto group_uri = cfg.group_id.substr(p + 1);
auto g = (module == "remote")
? system.middleman().remote_group(group_uri)
: system.groups().get(module, group_uri);
auto g = system.groups().get(module, group_uri);
if (! g) {
cerr << "*** unable to get group " << group_uri
<< " from module " << module << ": "
......@@ -119,8 +117,7 @@ void caf_main(actor_system& system, const config& cfg) {
auto res = message_builder(words.begin(), words.end()).apply({
[&](const string& cmd, const string& mod, const string& id) {
if (cmd == "/join") {
auto grp = (mod == "remote") ? system.middleman().remote_group(id)
: system.groups().get(mod, id);
auto grp = system.groups().get(mod, id);
if (grp)
anon_send(client_actor, join_atom::value, *grp);
}
......
......@@ -77,18 +77,6 @@ public:
return identifier_;
}
/// @cond PRIVATE
template <class... Ts>
void eq_impl(message_id mid, strong_actor_ptr sender,
execution_unit* ctx, Ts&&... xs) {
CAF_ASSERT(! mid.is_request());
enqueue(std::move(sender), mid,
make_message(std::forward<Ts>(xs)...), ctx);
}
/// @endcond
protected:
abstract_group(group_module& parent, std::string id, node_id origin);
......
......@@ -23,11 +23,13 @@
#include <memory>
#include <functional>
#include "caf/fwd.hpp"
#include "caf/extend.hpp"
#include "caf/local_actor.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/subscriber.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/detail/disposer.hpp"
......@@ -35,42 +37,79 @@
namespace caf {
/// An co-existing forwarding all messages through a user-defined
template <>
class behavior_type_of<actor_companion> {
public:
using type = behavior;
};
/// An co-existing actor forwarding all messages through a user-defined
/// callback to another object, thus serving as gateway to
/// allow any object to interact with other actors.
/// @extends local_actor
class actor_companion : public extend<local_actor, actor_companion>::
with<mixin::sender> {
class actor_companion : public extend<scheduled_actor, actor_companion>::
with<mixin::sender,
mixin::subscriber,
mixin::behavior_changer> {
public:
// -- member types -----------------------------------------------------------
/// Required by `spawn` for type deduction.
using signatures = none_t;
/// Required by `spawn` for type deduction.
using behavior_type = behavior;
/// A shared lockable.
using lock_type = detail::shared_spinlock;
using message_pointer = std::unique_ptr<mailbox_element, detail::disposer>;
using enqueue_handler = std::function<void (message_pointer)>;
/// Delegates incoming messages to user-defined event loop.
using enqueue_handler = std::function<void (mailbox_element_ptr)>;
/// Callback for actor termination.
using on_exit_handler = std::function<void ()>;
// -- constructors, destructors ----------------------------------------------
actor_companion(actor_config& cfg);
~actor_companion();
// -- overridden functions ---------------------------------------------------
void enqueue(mailbox_element_ptr what, execution_unit* host) override;
void enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host) override;
void launch(execution_unit* eu, bool lazy, bool hide) override;
void on_exit() override;
// -- modifiers --------------------------------------------------------------
/// Removes the handler for incoming messages and terminates
/// the companion for exit reason @ rsn.
/// the companion for exit reason `rsn`.
void disconnect(exit_reason rsn = exit_reason::normal);
/// Sets the handler for incoming messages.
/// @warning `handler` needs to be thread-safe
void on_enqueue(enqueue_handler handler);
void enqueue(mailbox_element_ptr what, execution_unit* host) override;
void enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host) override;
/// Sets the handler for incoming messages.
void on_exit(on_exit_handler handler);
private:
// set by parent to define custom enqueue action
enqueue_handler on_enqueue_;
// custom code for on_exit()
on_exit_handler on_exit_;
// guards access to handler_
lock_type lock_;
};
/// A pointer to a co-existing (actor) object.
/// @relates actor_companion
using actor_companion_ptr = intrusive_ptr<actor_companion>;
} // namespace caf
#endif // CAF_ACTOR_COMPANION_HPP
......@@ -37,6 +37,8 @@ enum class atom_value : uint64_t {
std::string to_string(const atom_value& x);
atom_value atom_from_string(const std::string& x);
/// Creates an atom from given string literal.
template <size_t Size>
constexpr atom_value atom(char const (&str)[Size]) {
......
......@@ -49,9 +49,10 @@ public:
/// @extends local_actor
class event_based_actor : public extend<scheduled_actor,
event_based_actor>::
with<mixin::sender, mixin::requester,
mixin::behavior_changer,
mixin::subscriber>,
with<mixin::sender,
mixin::requester,
mixin::subscriber,
mixin::behavior_changer>,
public dynamically_typed_actor_base {
public:
// -- member types -----------------------------------------------------------
......
......@@ -76,6 +76,7 @@ class actor_registry;
class blocking_actor;
class execution_unit;
class proxy_registry;
class actor_companion;
class continue_helper;
class mailbox_element;
class message_handler;
......
......@@ -76,15 +76,6 @@ public:
return ! ptr_;
}
/// Returns a handle that grants access to actor operations such as enqueue.
inline abstract_group* operator->() const noexcept {
return get();
}
inline abstract_group& operator*() const noexcept {
return *get();
}
static intptr_t compare(const abstract_group* lhs, const abstract_group* rhs);
intptr_t compare(const group& other) const noexcept;
......@@ -101,6 +92,36 @@ public:
return ptr_.get();
}
/// @cond PRIVATE
template <class... Ts>
void eq_impl(message_id mid, strong_actor_ptr sender,
execution_unit* ctx, Ts&&... xs) const {
CAF_ASSERT(! mid.is_request());
if (ptr_)
ptr_->enqueue(std::move(sender), mid,
make_message(std::forward<Ts>(xs)...), ctx);
}
inline bool subscribe(strong_actor_ptr who) const {
if (! ptr_)
return false;
return ptr_->subscribe(std::move(who));
}
inline void unsubscribe(const actor_control_block* who) const {
if (ptr_)
ptr_->unsubscribe(who);
}
/// CAF's messaging primitives assume a non-null guarantee. A group
/// object indirects pointer-like access to a group to prevent UB.
inline const group* operator->() const noexcept {
return this;
}
/// @endcond
private:
inline abstract_group* release() noexcept {
return ptr_.release();
......
......@@ -47,7 +47,7 @@ public:
/// Returns a pointer to the group associated with the name `group_name`.
/// @threadsafe
virtual group get(const std::string& group_name) = 0;
virtual expected<group> get(const std::string& group_name) = 0;
/// Loads a group of this module from `source` and stores it in `storage`.
virtual error load(deserializer& source, group& storage) = 0;
......
......@@ -21,10 +21,12 @@
#define CAF_MIXIN_ACTOR_WIDGET_HPP
#include "caf/config.hpp"
#include "caf/make_counted.hpp"
#include "caf/make_actor.hpp"
#include "caf/actor_companion.hpp"
#include "caf/message_handler.hpp"
#include "caf/scoped_execution_unit.hpp"
CAF_PUSH_WARNINGS
#include <QEvent>
#include <QApplication>
......@@ -36,36 +38,56 @@ namespace mixin {
template<typename Base, int EventId = static_cast<int>(QEvent::User + 31337)>
class actor_widget : public Base {
public:
typedef typename actor_companion::message_pointer message_pointer;
struct event_type : public QEvent {
message_pointer mptr;
event_type(message_pointer ptr)
mailbox_element_ptr mptr;
event_type(mailbox_element_ptr ptr)
: QEvent(static_cast<QEvent::Type>(EventId)), mptr(std::move(ptr)) {
// nop
}
};
template <typename... Ts>
actor_widget(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
companion_ = make_counted<actor_companion>();
companion_->on_enqueue([=](message_pointer ptr) {
actor_widget(Ts&&... xs) : Base(std::forward<Ts>(xs)...), alive_(false) {
// nop
}
~actor_widget() {
if (companion_)
self()->cleanup(error{}, &dummy_);
}
void init(actor_system& system) {
alive_ = true;
companion_ = actor_cast<strong_actor_ptr>(system.spawn<actor_companion>());
self()->on_enqueue([=](mailbox_element_ptr ptr) {
qApp->postEvent(this, new event_type(std::move(ptr)));
});
self()->on_exit([=] {
// close widget if actor companion dies
this->close();
});
}
template <class F>
void set_message_handler(F pfun) {
self()->become(pfun(self()));
}
template <typename T>
void set_message_handler(T pfun) {
companion_->become(pfun(companion_.get()));
/// Terminates the actor companion and closes this widget.
void quit_and_close(error exit_state = error{}) {
self()->quit(std::move(exit_state));
this->close();
}
bool event(QEvent* event) override {
if (event->type() == static_cast<QEvent::Type>(EventId)) {
auto ptr = dynamic_cast<event_type*>(event);
if (ptr) {
companion_->invoke_message(ptr->mptr,
companion_->get_behavior(),
companion_->awaited_response_id());
if (ptr && alive_) {
switch (self()->activate(&dummy_, *(ptr->mptr))) {
default:
break;
};
return true;
}
}
......@@ -73,11 +95,22 @@ public:
}
actor as_actor() const {
return companion_;
CAF_ASSERT(companion_);
return actor_cast<actor>(companion_);
}
actor_companion* self() {
using bptr = abstract_actor*; // base pointer
using dptr = actor_companion*; // derived pointer
return companion_ ? static_cast<dptr>(actor_cast<bptr>(companion_))
: nullptr;
}
private:
actor_companion_ptr companion_;
scoped_execution_unit dummy_;
strong_actor_ptr companion_;
bool alive_;
};
} // namespace mixin
......
......@@ -22,13 +22,12 @@
namespace caf {
void actor_companion::disconnect(exit_reason rsn) {
enqueue_handler tmp;
{ // lifetime scope of guard
std::lock_guard<lock_type> guard(lock_);
on_enqueue_.swap(tmp);
}
cleanup(rsn, context());
actor_companion::actor_companion(actor_config& cfg) : extended_base(cfg) {
// nop
}
actor_companion::~actor_companion() {
// nop
}
void actor_companion::on_enqueue(enqueue_handler handler) {
......@@ -36,6 +35,10 @@ void actor_companion::on_enqueue(enqueue_handler handler) {
on_enqueue_ = std::move(handler);
}
void actor_companion::on_exit(on_exit_handler handler) {
on_exit_ = std::move(handler);
}
void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr);
shared_lock<lock_type> guard(lock_);
......@@ -49,4 +52,18 @@ void actor_companion::enqueue(strong_actor_ptr src, message_id mid,
enqueue(std::move(ptr), eu);
}
void actor_companion::launch(execution_unit*, bool, bool hide) {
is_registered(! hide);
}
void actor_companion::on_exit() {
enqueue_handler tmp;
{ // lifetime scope of guard
std::lock_guard<lock_type> guard(lock_);
on_enqueue_.swap(tmp);
}
if (on_exit_)
on_exit_();
}
} // namespace caf
......@@ -21,6 +21,15 @@
namespace caf {
atom_value atom_from_string(const std::string& x) {
if (x.size() > 10)
return atom("");
char buf[11];
memcpy(buf, x.c_str(), x.size());
buf[x.size()] = '\0';
return atom(buf);
}
std::string to_string(const atom_value& what) {
auto x = static_cast<uint64_t>(what);
std::string result;
......
......@@ -84,11 +84,11 @@ void serialize(deserializer& source, group& x, const unsigned int) {
}
std::string to_string(const group& x) {
if (x == invalid_group)
if (! x)
return "<invalid-group>";
std::string result = x->module().name();
std::string result = x.get()->module().name();
result += "/";
result += x->identifier();
result += x.get()->identifier();
return result;
}
......
......@@ -327,12 +327,12 @@ public:
CAF_LOG_TRACE("");
}
group get(const std::string& identifier) override {
expected<group> get(const std::string& identifier) override {
CAF_LOG_TRACE(CAF_ARG(identifier));
upgrade_guard guard(instances_mtx_);
auto i = instances_.find(identifier);
if (i != instances_.end())
return {i->second};
return group{i->second};
auto tmp = make_counted<local_group>(*this, identifier,
system().node(), none);
upgrade_to_unique_guard uguard(guard);
......@@ -342,7 +342,7 @@ public:
// someone might preempt us
if (result != tmp)
tmp->stop();
return {result};
return group{result};
}
error load(deserializer& source, group& storage) override {
......@@ -358,7 +358,7 @@ public:
}
auto broker = actor_cast<actor>(broker_ptr);
if (broker->node() == system().node()) {
storage = this->get(identifier);
storage = *this->get(identifier);
return {};
}
upgrade_guard guard(proxies_mtx_);
......@@ -465,7 +465,8 @@ group group_manager::anonymous() const {
CAF_LOG_TRACE("");
std::string id = "__#";
id += std::to_string(++s_ad_hoc_id);
return get_module("local")->get(id);
// local module is guaranteed to not return an error
return *get_module("local")->get(id);
}
expected<group> group_manager::get(const std::string& module_name,
......
......@@ -319,6 +319,33 @@ void middleman::stop() {
}
void middleman::init(actor_system_config& cfg) {
// add remote group module to config
struct remote_groups : group_module {
public:
remote_groups(middleman& parent)
: group_module(parent.system(), "remote"),
parent_(parent) {
// nop
}
void stop() {
// nop
}
expected<group> get(const std::string& group_name) {
return parent_.remote_group(group_name);
}
error load(deserializer&, group&) {
// never called, because we hand out group instances of the local module
return sec::no_such_group_module;
}
private:
middleman& parent_;
};
auto gfactory = [=]() -> group_module* { return new remote_groups(*this); };
cfg.group_module_factories.emplace_back(gfactory);
// logging not available at this stage
// add I/O-related types to config
cfg.add_message_type<network::protocol>("@protocol")
......
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