Commit 063022f1 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1358'

Close #1358.
parents 343f3930 a554024e
...@@ -43,6 +43,16 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -43,6 +43,16 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The types `caf::byte`, `caf::optional` and `caf::string_view` became obsolete - The types `caf::byte`, `caf::optional` and `caf::string_view` became obsolete
after switching to C++17. Consequently, these types are now deprecated in after switching to C++17. Consequently, these types are now deprecated in
favor of their standard library counterpart. favor of their standard library counterpart.
- The group-based pub/sub mechanism never fit nicely into the typed messaging
API and the fact that group messages use the regular mailbox makes it hard to
separate regular communication from multi-cast messages. Hence, we decided to
drop the group API and instead focus on the new flows and streams that can
replace group-communication in many use cases.
- The "actor-composition operator" was added as a means to enable the first
experimental streaming API. With that gone, there's no justification to keep
this feature. While it has some neat niche-applications, the prevent some
optimization we would like to apply to the messaging layer. Hence, we will
remove this feature without a replacement.
### Removed ### Removed
......
...@@ -65,8 +65,6 @@ if(TARGET CAF::io) ...@@ -65,8 +65,6 @@ if(TARGET CAF::io)
endfunction() endfunction()
# basic remoting # basic remoting
add_io_example(remoting group_chat)
add_io_example(remoting group_server)
add_io_example(remoting remote_spawn) add_io_example(remoting remote_spawn)
add_io_example(remoting distributed_calculator) add_io_example(remoting distributed_calculator)
...@@ -88,36 +86,6 @@ if(TARGET CAF::io) ...@@ -88,36 +86,6 @@ if(TARGET CAF::io)
add_dependencies(protobuf_broker all_examples) add_dependencies(protobuf_broker all_examples)
endif() endif()
if(CAF_ENABLE_QT6_EXAMPLES)
find_package(Qt6 COMPONENTS Core Gui Widgets REQUIRED)
qt6_wrap_ui(GROUP_CHAT_UI_HDR qtsupport/chatwindow.ui)
qt6_wrap_cpp(GROUP_CHAT_MOC_SRC qtsupport/chatwidget.hpp)
# generated headers will be in cmake build directory
add_executable(qt_group_chat
qtsupport/qt_group_chat.cpp
qtsupport/chatwidget.cpp
${GROUP_CHAT_MOC_SRC}
${GROUP_CHAT_UI_HDR})
target_link_libraries(qt_group_chat
CAF::io
CAF::internal
Qt6::Core
Qt6::Gui
Qt6::Widgets)
target_include_directories(qt_group_chat PRIVATE
qtsupport
${CMAKE_CURRENT_BINARY_DIR}
${Qt6Core_INCLUDE_DIRS}
${Qt6Gui_INCLUDE_DIRS}
${Qt6Widgets_INCLUDE_DIRS})
if (CMAKE_VERSION VERSION_LESS 3.8)
message(STATUS "Note: build fails if Qt6 sets incompatible -std=ARG flag")
else()
set_property(TARGET qt_group_chat PROPERTY CXX_STANDARD 17)
endif()
add_dependencies(qt_group_chat all_examples)
endif()
if(CAF_ENABLE_CURL_EXAMPLES) if(CAF_ENABLE_CURL_EXAMPLES)
find_package(CURL REQUIRED) find_package(CURL REQUIRED)
add_executable(curl_fuse curl/curl_fuse.cpp) add_executable(curl_fuse curl/curl_fuse.cpp)
...@@ -150,3 +118,33 @@ add_net_example(web_socket hello-client) ...@@ -150,3 +118,33 @@ add_net_example(web_socket hello-client)
add_net_example(web_socket quote-server) add_net_example(web_socket quote-server)
add_net_example(web_socket secure-echo) add_net_example(web_socket secure-echo)
add_net_example(web_socket stock-ticker) add_net_example(web_socket stock-ticker)
if(TARGET CAF::net AND CAF_ENABLE_QT6_EXAMPLES)
find_package(Qt6 COMPONENTS Core Gui Widgets REQUIRED)
qt6_wrap_ui(GROUP_CHAT_UI_HDR qtsupport/chatwindow.ui)
qt6_wrap_cpp(GROUP_CHAT_MOC_SRC qtsupport/chatwidget.hpp)
# generated headers will be in cmake build directory
add_executable(qt_group_chat
qtsupport/qt_group_chat.cpp
qtsupport/chatwidget.cpp
${GROUP_CHAT_MOC_SRC}
${GROUP_CHAT_UI_HDR})
target_link_libraries(qt_group_chat
CAF::net
CAF::internal
Qt6::Core
Qt6::Gui
Qt6::Widgets)
target_include_directories(qt_group_chat PRIVATE
qtsupport
${CMAKE_CURRENT_BINARY_DIR}
${Qt6Core_INCLUDE_DIRS}
${Qt6Gui_INCLUDE_DIRS}
${Qt6Widgets_INCLUDE_DIRS})
if (CMAKE_VERSION VERSION_LESS 3.8)
message(STATUS "Note: build fails if Qt6 sets incompatible -std=ARG flag")
else()
set_property(TARGET qt_group_chat PROPERTY CXX_STANDARD 17)
endif()
add_dependencies(qt_group_chat all_examples)
endif()
...@@ -46,6 +46,8 @@ struct config : caf::actor_system_config { ...@@ -46,6 +46,8 @@ struct config : caf::actor_system_config {
} }
}; };
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) { int caf_main(caf::actor_system& sys, const config& cfg) {
// Connect to the server. // Connect to the server.
auto port = caf::get_or(cfg, "port", default_port); auto port = caf::get_or(cfg, "port", default_port);
......
...@@ -4,6 +4,7 @@ ...@@ -4,6 +4,7 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/scheduled_actor/flow.hpp"
CAF_PUSH_WARNINGS CAF_PUSH_WARNINGS
#include <QInputDialog> #include <QInputDialog>
...@@ -24,45 +25,43 @@ ChatWidget::~ChatWidget() { ...@@ -24,45 +25,43 @@ ChatWidget::~ChatWidget() {
// nop // nop
} }
void ChatWidget::init(actor_system& system) { void ChatWidget::init(actor_system& system, const std::string& name,
caf::async::consumer_resource<bin_frame> pull,
caf::async::producer_resource<bin_frame> push) {
name_ = QString::fromUtf8(name);
print("*** hello " + name_);
super::init(system); super::init(system);
set_message_handler([=](actor_companion* self) -> message_handler { self() //
->make_observable()
.from_resource(pull)
.do_finally([this] { //
print("*** chatroom offline: lost connection to the server");
})
.for_each([this](const bin_frame& frame) {
auto bytes = frame.bytes();
auto str = std::string_view{reinterpret_cast<const char*>(bytes.data()),
bytes.size()};
if (std::all_of(str.begin(), str.end(), ::isprint)) {
print(QString::fromUtf8(str.data(), str.size()));
} else {
QString msg = "<non-ascii-data of size ";
msg += QString::number(bytes.size());
msg += '>';
print(msg);
}
});
publisher_ = std::make_unique<publisher_type>(self());
publisher_->as_observable()
.map([](const QString& str) {
auto encoded = str.toUtf8();
auto bytes = caf::as_bytes(
caf::make_span(encoded.data(), static_cast<size_t>(encoded.size())));
return bin_frame{bytes};
})
.subscribe(push);
set_message_handler([=](actor_companion*) -> message_handler {
return { return {
[=](join_atom, const group& what) {
auto groups = self->joined_groups();
for (const auto& grp : groups) {
print(("*** leave " + to_string(grp)).c_str());
self->send(grp, name_ + " has left the chatroom");
self->leave(grp);
}
print(("*** join " + to_string(what)).c_str());
self->join(what);
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(); }, [=](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()));
},
[=](leave_atom) {
auto groups = self->joined_groups();
for (const auto& grp : groups) {
std::cout << "*** leave " << to_string(grp) << std::endl;
self->send(grp, name_ + " has left the chatroom");
self->leave(grp);
}
},
}; };
}); });
} }
...@@ -78,63 +77,29 @@ void ChatWidget::sendChatMessage() { ...@@ -78,63 +77,29 @@ void ChatWidget::sendChatMessage() {
auto sv = std::string_view{utf8.constData(), auto sv = std::string_view{utf8.constData(),
static_cast<size_t>(utf8.size())}; static_cast<size_t>(utf8.size())};
split(words, sv, is_any_of(" ")); split(words, sv, is_any_of(" "));
if (words.size() == 3 && words[0] == "join") { if (words.size() == 2 && words[0] == "setName") {
if (auto x = system().groups().get(words[1], words[2])) auto name = QString::fromUtf8(words[1]);
send_as(as_actor(), as_actor(), join_atom_v, std::move(*x)); if (!name.isEmpty())
else name_ = name;
print("*** error: " + QString::fromUtf8(to_string(x.error()).c_str())); } else {
} else if (words.size() == 2 && words[0] == "setName")
send_as(as_actor(), as_actor(), set_name_atom_v, std::move(words[1]));
else {
print("*** list of commands:\n" print("*** list of commands:\n"
"/join <module> <group id>\n"
"/setName <new name>\n"); "/setName <new name>\n");
return; return;
} }
} else { } else {
if (name_.empty()) { auto msg = name_;
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 += ": ";
msg += line.toUtf8().constData(); msg += line;
print("<you>: " + input()->text()); print("<you>: " + input()->text());
send_as(as_actor(), chatroom_, std::move(msg)); if (publisher_) {
} publisher_->push(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 = QStringView{gname}.mid(pos + 1).toUtf8().constData();
auto x = system().groups().get(mod, gid);
if (!x)
QMessageBox::critical(this, "Error", to_string(x.error()).c_str());
else
self()->send(self(), join_atom_v, std::move(*x));
} }
void ChatWidget::changeName() { void ChatWidget::changeName() {
auto name = QInputDialog::getText(this, "Change Name", auto name = QInputDialog::getText(this, "Change Name",
"Please enter a new name"); "Please enter a new name");
if (!name.isEmpty()) if (!name.isEmpty())
send_as(as_actor(), as_actor(), set_name_atom_v, name.toUtf8().constData()); name_ = name;
} }
...@@ -3,7 +3,9 @@ ...@@ -3,7 +3,9 @@
#include <exception> #include <exception>
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/mixin/actor_widget.hpp" #include "caf/mixin/actor_widget.hpp"
#include "caf/net/binary/frame.hpp"
CAF_PUSH_WARNINGS CAF_PUSH_WARNINGS
#include <QWidget> #include <QWidget>
...@@ -13,7 +15,6 @@ CAF_POP_WARNINGS ...@@ -13,7 +15,6 @@ CAF_POP_WARNINGS
CAF_BEGIN_TYPE_ID_BLOCK(qtsupport, first_custom_type_id) CAF_BEGIN_TYPE_ID_BLOCK(qtsupport, first_custom_type_id)
CAF_ADD_ATOM(qtsupport, set_name_atom)
CAF_ADD_ATOM(qtsupport, quit_atom) CAF_ADD_ATOM(qtsupport, quit_atom)
CAF_END_TYPE_ID_BLOCK(qtsupport) CAF_END_TYPE_ID_BLOCK(qtsupport)
...@@ -29,16 +30,21 @@ public: ...@@ -29,16 +30,21 @@ public:
using super = caf::mixin::actor_widget<QWidget>; using super = caf::mixin::actor_widget<QWidget>;
using bin_frame = caf::net::binary::frame;
using publisher_type = caf::flow::item_publisher<QString>;
ChatWidget(QWidget* parent = nullptr); ChatWidget(QWidget* parent = nullptr);
~ChatWidget(); ~ChatWidget();
void init(caf::actor_system& system); void init(caf::actor_system& system, const std::string& name,
caf::async::consumer_resource<bin_frame> pull,
caf::async::producer_resource<bin_frame> push);
public slots: public slots:
void sendChatMessage(); void sendChatMessage();
void joinGroup();
void changeName(); void changeName();
private: private:
...@@ -71,6 +77,6 @@ private: ...@@ -71,6 +77,6 @@ private:
QLineEdit* input_; QLineEdit* input_;
QTextEdit* output_; QTextEdit* output_;
std::string name_; QString name_;
caf::group chatroom_; std::unique_ptr<publisher_type> publisher_;
}; };
...@@ -101,16 +101,10 @@ ...@@ -101,16 +101,10 @@
<property name="title"> <property name="title">
<string>File</string> <string>File</string>
</property> </property>
<addaction name="actionJoin_Group"/>
<addaction name="actionSet_Name"/> <addaction name="actionSet_Name"/>
</widget> </widget>
<addaction name="menuFile"/> <addaction name="menuFile"/>
</widget> </widget>
<action name="actionJoin_Group">
<property name="text">
<string>Join Group</string>
</property>
</action>
<action name="actionSet_Name"> <action name="actionSet_Name">
<property name="text"> <property name="text">
<string>Set Name</string> <string>Set Name</string>
...@@ -126,28 +120,11 @@ ...@@ -126,28 +120,11 @@
<slots> <slots>
<slot>sendChatMessage()</slot> <slot>sendChatMessage()</slot>
<slot>changeName()</slot> <slot>changeName()</slot>
<slot>joinGroup()</slot>
</slots> </slots>
</customwidget> </customwidget>
</customwidgets> </customwidgets>
<resources/> <resources/>
<connections> <connections>
<connection>
<sender>actionJoin_Group</sender>
<signal>triggered()</signal>
<receiver>chatwidget</receiver>
<slot>joinGroup()</slot>
<hints>
<hint type="sourcelabel">
<x>-1</x>
<y>-1</y>
</hint>
<hint type="destinationlabel">
<x>399</x>
<y>300</y>
</hint>
</hints>
</connection>
<connection> <connection>
<sender>actionSet_Name</sender> <sender>actionSet_Name</sender>
<signal>triggered()</signal> <signal>triggered()</signal>
......
/******************************************************************************\ /******************************************************************************\
* This example program represents a minimal GUI chat program * * This example program represents a minimal GUI chat program *
* based on group communication. This chat program is compatible to the * * based on group communication. This chat program is compatible to the *
* terminal version in remote_actors/group_chat.cpp. * * terminal version in length_prefix_framing/chat-server.cpp. *
* * * *
* Setup for a minimal chat between "alice" and "bob": * * Setup for a minimal chat between "alice" and "bob": *
* - group_server -p 4242 * * - chat-server -p 4242 *
* - qt_group_chat -g remote:chatroom@localhost:4242 -n alice * * - qt_group_chat -H localhost -p 4242 -n alice *
* - qt_group_chat -g remote:chatroom@localhost:4242 -n bob * * - qt_group_chat -H localhost -p 4242 -n bob *
\******************************************************************************/ \******************************************************************************/
#include <cstdlib> #include <cstdlib>
...@@ -18,7 +18,10 @@ ...@@ -18,7 +18,10 @@
#include <vector> #include <vector>
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/all.hpp" #include "caf/net/length_prefix_framing.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
CAF_PUSH_WARNINGS CAF_PUSH_WARNINGS
#include "ui_chatwindow.h" // auto generated from chatwindow.ui #include "ui_chatwindow.h" // auto generated from chatwindow.ui
...@@ -30,49 +33,64 @@ CAF_POP_WARNINGS ...@@ -30,49 +33,64 @@ CAF_POP_WARNINGS
using namespace caf; using namespace caf;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 7788;
static constexpr std::string_view default_host = "localhost";
// -- configuration setup ------------------------------------------------------
class config : public actor_system_config { class config : public actor_system_config {
public: public:
config() { config() {
opt_group{custom_options_, "global"} opt_group{custom_options_, "global"}
.add<std::string>("name,n", "set name") .add<uint16_t>("port,p", "port of the server")
.add<std::string>("group,g", "join group"); .add<std::string>("host,H", "host of the server")
.add<std::string>("name,n", "set name");
} }
}; };
// -- main ---------------------------------------------------------------------
int caf_main(actor_system& sys, const config& cfg) { int caf_main(actor_system& sys, const config& cfg) {
std::string name; // Connect to the server.
if (auto config_name = get_if<std::string>(&cfg, "name")) auto port = caf::get_or(cfg, "port", default_port);
name = std::move(*config_name); auto host = caf::get_or(cfg, "host", default_host);
while (name.empty()) { auto name = caf::get_or(cfg, "name", "");
std::cout << "please enter your name: " << std::flush; if (name.empty()) {
if (!std::getline(std::cin, name)) { std::cerr << "*** mandatory parameter 'name' missing or empty\n";
std::cerr << "*** no name given... terminating" << std::endl; return EXIT_FAILURE;
return EXIT_FAILURE;
}
} }
group grp; auto fd = caf::net::make_connected_tcp_stream_socket(host, port);
// evaluate group parameters if (!fd) {
if (auto locator = get_if<std::string>(&cfg, "group")) { std::cerr << "*** unable to connect to " << host << ":" << port << ": "
if (auto maybe_grp = sys.groups().get(*locator)) { << to_string(fd.error()) << '\n';
grp = std::move(*maybe_grp); return EXIT_FAILURE;
} else {
std::cerr << R"(*** failed to parse ")" << *locator
<< R"(" as group locator: )" << to_string(maybe_grp.error())
<< std::endl;
}
} }
std::cout << "*** connected to " << host << ":" << port << ": " << '\n';
// Create our buffers that connect the worker to the socket.
using bin_frame = caf::net::binary::frame;
using caf::async::make_spsc_buffer_resource;
auto [lpf_pull, app_push] = make_spsc_buffer_resource<bin_frame>();
auto [app_pull, lpf_push] = make_spsc_buffer_resource<bin_frame>();
// Spin up the network backend.
using lpf = caf::net::length_prefix_framing;
auto conn = lpf::run(sys, *fd, std::move(lpf_pull), std::move(lpf_push));
// Spin up Qt.
auto [argc, argv] = cfg.c_args_remainder(); auto [argc, argv] = cfg.c_args_remainder();
QApplication app{argc, argv}; QApplication app{argc, argv};
app.setQuitOnLastWindowClosed(true); app.setQuitOnLastWindowClosed(true);
QMainWindow mw; QMainWindow mw;
Ui::ChatWindow helper; Ui::ChatWindow helper;
helper.setupUi(&mw); helper.setupUi(&mw);
helper.chatwidget->init(sys); helper.chatwidget->init(sys, name, std::move(app_pull), std::move(app_push));
// Setup and run.
auto client = helper.chatwidget->as_actor(); auto client = helper.chatwidget->as_actor();
anon_send(client, set_name_atom_v, move(name));
anon_send(client, join_atom_v, std::move(grp));
mw.show(); mw.show();
return app.exec(); auto result = app.exec();
conn.dispose();
return result;
} }
CAF_MAIN(id_block::qtsupport, io::middleman) CAF_MAIN(caf::id_block::qtsupport, caf::net::middleman)
/******************************************************************************\
* This example program represents a minimal terminal chat program *
* based on group communication. *
* *
* Setup for a minimal chat between "alice" and "bob": *
* - group_chat -s -p 4242 *
* - group_chat -g remote:chatroom@localhost:4242 -n alice *
* - group_chat -g remote:chatroom@localhost:4242 -n bob *
\******************************************************************************/
#include <cstdlib>
#include <iostream>
#include <map>
#include <set>
#include <sstream>
#include <vector>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/string_algorithms.hpp"
CAF_BEGIN_TYPE_ID_BLOCK(group_chat, first_custom_type_id)
CAF_ADD_ATOM(group_chat, broadcast_atom)
CAF_END_TYPE_ID_BLOCK(group_chat)
using namespace caf;
struct line {
std::string str;
};
std::istream& operator>>(std::istream& is, line& l) {
std::getline(is, l.str);
return is;
}
behavior client(event_based_actor* self, const std::string& name) {
return {
[=](broadcast_atom, const std::string& message) {
for (auto& dest : self->joined_groups()) {
self->send(dest, name + ": " + message);
}
},
[=](join_atom, const group& what) {
auto groups = self->joined_groups();
for (auto&& grp : groups) {
std::cout << "*** leave " << to_string(grp) << std::endl;
self->send(grp, name + " has left the chatroom");
self->leave(grp);
}
std::cout << "*** join " << to_string(what) << std::endl;
self->join(what);
self->send(what, name + " has entered the chatroom");
},
[=](const std::string& txt) {
// don't print own messages
if (self->current_sender() != self)
std::cout << txt << std::endl;
},
[=](const group_down_msg& g) {
std::cout << "*** chatroom offline: " << to_string(g.source) << std::endl;
},
[=](leave_atom) {
auto groups = self->joined_groups();
for (auto&& grp : groups) {
std::cout << "*** leave " << to_string(grp) << std::endl;
self->send(grp, name + " has left the chatroom");
self->leave(grp);
}
},
};
}
class config : public actor_system_config {
public:
config() {
opt_group{custom_options_, "global"}
.add<std::string>("name,n", "set name")
.add<std::string>("group,g", "join group")
.add<bool>("server,s", "run in server mode")
.add<uint16_t>("port,p", "set port (ignored in client mode)");
}
};
void run_server(actor_system& sys) {
auto port = get_or(sys.config(), "port", uint16_t{0});
auto res = sys.middleman().publish_local_groups(port);
if (!res) {
std::cerr << "*** publishing local groups failed: "
<< to_string(res.error()) << std::endl;
return;
}
std::cout << "*** listening at port " << *res << std::endl
<< "*** press [enter] to quit" << std::endl;
std::string dummy;
std::getline(std::cin, dummy);
std::cout << "... cya" << std::endl;
}
void run_client(actor_system& sys) {
std::string name;
if (auto config_name = get_if<std::string>(&sys.config(), "name"))
name = *config_name;
while (name.empty()) {
std::cout << "please enter your name: " << std::flush;
if (!std::getline(std::cin, name)) {
std::cerr << "*** no name given... terminating" << std::endl;
return;
}
}
auto client_actor = sys.spawn(client, name);
if (auto locator = get_if<std::string>(&sys.config(), "group")) {
if (auto grp = sys.groups().get(*locator)) {
anon_send(client_actor, join_atom_v, std::move(*grp));
} else {
std::cerr << R"(*** failed to parse ")" << *locator
<< R"(" as group locator: )" << to_string(grp.error())
<< std::endl;
}
}
std::cout << "*** starting client, type '/help' for a list of commands\n";
std::istream_iterator<line> eof;
std::vector<std::string> words;
for (std::istream_iterator<line> i{std::cin}; i != eof; ++i) {
if (i->str.empty()) {
// Ignore empty lines.
} else if (i->str[0] == '/') {
words.clear();
split(words, i->str, is_any_of(" "));
if (words.size() == 3 && words[0] == "/join") {
if (auto grp = sys.groups().get(words[1], words[2]))
anon_send(client_actor, join_atom_v, *grp);
else
std::cerr << "*** failed to join group: " << to_string(grp.error())
<< std::endl;
} else if (words.size() == 1 && words[0] == "/quit") {
std::cin.setstate(std::ios_base::eofbit);
} else {
std::cout << "*** available commands:\n"
" /join <module> <group> join a new chat channel\n"
" /quit quit the program\n"
" /help print this text\n";
}
} else {
anon_send(client_actor, broadcast_atom_v, i->str);
}
}
anon_send(client_actor, leave_atom_v);
anon_send_exit(client_actor, exit_reason::user_shutdown);
}
void caf_main(actor_system& sys, const config& cfg) {
auto f = get_or(cfg, "server", false) ? run_server : run_client;
f(sys);
}
CAF_MAIN(id_block::group_chat, io::middleman)
/******************************************************************************\
* This example program represents a minimal IRC-like group *
* communication server. *
* *
* Setup for a minimal chat between "alice" and "bob": *
* - group_server -p 4242 *
* - group_chat -g remote:chatroom@localhost:4242 -n alice *
* - group_chat -g remote:chatroom@localhost:4242 -n bob *
\******************************************************************************/
#include <string>
#include <cstdlib>
#include <iostream>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using namespace std;
using namespace caf;
namespace {
class config : public actor_system_config {
public:
uint16_t port = 0;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port");
}
};
void caf_main(actor_system& system, const config& cfg) {
system.middleman().publish_local_groups(cfg.port);
cout << "type 'quit' to shutdown the server" << endl;
string line;
while (getline(cin, line))
if (line == "quit")
return;
else
cerr << "illegal command" << endl;
}
} // namespace
CAF_MAIN(io::middleman)
...@@ -223,7 +223,6 @@ caf_add_component( ...@@ -223,7 +223,6 @@ caf_add_component(
actor_registry actor_registry
actor_system_config actor_system_config
actor_termination actor_termination
aout
async.blocking_consumer async.blocking_consumer
async.blocking_producer async.blocking_producer
async.consumer_adapter async.consumer_adapter
...@@ -234,7 +233,6 @@ caf_add_component( ...@@ -234,7 +233,6 @@ caf_add_component(
binary_deserializer binary_deserializer
binary_serializer binary_serializer
blocking_actor blocking_actor
composition
config_option config_option
config_option_set config_option_set
config_value config_value
......
...@@ -158,7 +158,7 @@ private: ...@@ -158,7 +158,7 @@ private:
}; };
/// Combine `f` and `g` so that `(f*g)(x) = f(g(x))`. /// Combine `f` and `g` so that `(f*g)(x) = f(g(x))`.
CAF_CORE_EXPORT actor operator*(actor f, actor g); [[deprecated]] CAF_CORE_EXPORT actor operator*(actor f, actor g);
/// @relates actor /// @relates actor
CAF_CORE_EXPORT bool operator==(const actor& lhs, abstract_actor* rhs); CAF_CORE_EXPORT bool operator==(const actor& lhs, abstract_actor* rhs);
......
...@@ -45,11 +45,13 @@ public: ...@@ -45,11 +45,13 @@ public:
actor_ostream& flush(); actor_ostream& flush();
/// Redirects all further output from `self` to `file_name`. /// Redirects all further output from `self` to `file_name`.
static void redirect(abstract_actor* self, std::string fn, int flags = 0); [[deprecated]] static void redirect(abstract_actor* self, std::string fn,
int flags = 0);
/// Redirects all further output from any actor that did not /// Redirects all further output from any actor that did not
/// redirect its output to `fname`. /// redirect its output to `fname`.
static void redirect_all(actor_system& sys, std::string fn, int flags = 0); [[deprecated]] static void redirect_all(actor_system& sys, std::string fn,
int flags = 0);
/// Writes `arg` to the buffer allocated for the calling actor. /// Writes `arg` to the buffer allocated for the calling actor.
actor_ostream& operator<<(const char* arg) { actor_ostream& operator<<(const char* arg) {
......
...@@ -259,27 +259,6 @@ public: ...@@ -259,27 +259,6 @@ public:
/// Renders `x` using the line format `lf` to `out`. /// Renders `x` using the line format `lf` to `out`.
void render(std::ostream& out, const line_format& lf, const event& x) const; void render(std::ostream& out, const line_format& lf, const event& x) const;
/// Returns a string representation of the joined groups of `x` if `x` is an
/// actor with the `subscriber` mixin.
template <class T>
static
typename std::enable_if<std::is_base_of<mixin::subscriber_base, T>::value,
std::string>::type
joined_groups_of(const T& x) {
return deep_to_string(x.joined_groups());
}
/// Returns a string representation of an empty list if `x` is not an actor
/// with the `subscriber` mixin.
template <class T>
static
typename std::enable_if<!std::is_base_of<mixin::subscriber_base, T>::value,
const char*>::type
joined_groups_of(const T& x) {
CAF_IGNORE_UNUSED(x);
return "[]";
}
// -- thread-local properties ------------------------------------------------ // -- thread-local properties ------------------------------------------------
/// Stores the actor system for the current thread. /// Stores the actor system for the current thread.
...@@ -531,8 +510,7 @@ CAF_CORE_EXPORT bool operator==(const logger::field& x, const logger::field& y); ...@@ -531,8 +510,7 @@ CAF_CORE_EXPORT bool operator==(const logger::field& x, const logger::field& y);
<< ref.id() << "; NAME =" << ref.name() << "; TYPE =" \ << ref.id() << "; NAME =" << ref.name() << "; TYPE =" \
<< ::caf::detail::pretty_type_name(typeid(ref)) \ << ::caf::detail::pretty_type_name(typeid(ref)) \
<< "; ARGS =" << ctor_data.c_str() \ << "; ARGS =" << ctor_data.c_str() \
<< "; NODE =" << ref.node() \ << "; NODE =" << ref.node())
<< "; GROUPS =" << ::caf::logger::joined_groups_of(ref))
# define CAF_LOG_SEND_EVENT(ptr) \ # define CAF_LOG_SEND_EVENT(ptr) \
CAF_LOG_IMPL(CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, \ CAF_LOG_IMPL(CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, \
......
...@@ -34,7 +34,7 @@ public: ...@@ -34,7 +34,7 @@ public:
: Base(cfg, std::forward<Ts>(xs)...) { : Base(cfg, std::forward<Ts>(xs)...) {
if (cfg.groups != nullptr) if (cfg.groups != nullptr)
for (auto& grp : *cfg.groups) for (auto& grp : *cfg.groups)
join(grp); join_impl(grp);
} }
// -- overridden functions of monitorable_actor ------------------------------ // -- overridden functions of monitorable_actor ------------------------------
...@@ -51,27 +51,31 @@ public: ...@@ -51,27 +51,31 @@ public:
/// Causes this actor to subscribe to the group `what`. /// Causes this actor to subscribe to the group `what`.
/// The group will be unsubscribed if the actor finishes execution. /// The group will be unsubscribed if the actor finishes execution.
void join(const group& what) { [[deprecated("use flows instead of groups")]] void join(const group& what) {
CAF_LOG_TRACE(CAF_ARG(what)); join_impl(what);
if (what == invalid_group)
return;
if (what->subscribe(this->ctrl()))
subscriptions_.emplace(what);
} }
/// Causes this actor to leave the group `what`. /// Causes this actor to leave the group `what`.
void leave(const group& what) { [[deprecated("use flows instead of groups")]] void leave(const group& what) {
CAF_LOG_TRACE(CAF_ARG(what)); CAF_LOG_TRACE(CAF_ARG(what));
if (subscriptions_.erase(what) > 0) if (subscriptions_.erase(what) > 0)
what->unsubscribe(this->ctrl()); what->unsubscribe(this->ctrl());
} }
/// Returns all subscribed groups. /// Returns all subscribed groups.
const subscriptions& joined_groups() const { [[deprecated("use flows instead of groups")]] const subscriptions&
joined_groups() const {
return subscriptions_; return subscriptions_;
} }
private: private:
void join_impl(const group& what) {
CAF_LOG_TRACE(CAF_ARG(what));
if (what == invalid_group)
return;
if (what->subscribe(this->ctrl()))
subscriptions_.emplace(what);
}
// -- data members ----------------------------------------------------------- // -- data members -----------------------------------------------------------
/// Stores all subscribed groups. /// Stores all subscribed groups.
......
...@@ -301,7 +301,8 @@ bool operator!=(std::nullptr_t, const typed_actor<Xs...>& x) noexcept { ...@@ -301,7 +301,8 @@ bool operator!=(std::nullptr_t, const typed_actor<Xs...>& x) noexcept {
/// Returns a new actor that implements the composition `f.g(x) = f(g(x))`. /// Returns a new actor that implements the composition `f.g(x) = f(g(x))`.
/// @relates typed_actor /// @relates typed_actor
template <class... Xs, class... Ys> template <class... Xs, class... Ys>
composed_type_t<detail::type_list<Xs...>, detail::type_list<Ys...>> [[deprecated]] composed_type_t<detail::type_list<Xs...>,
detail::type_list<Ys...>>
operator*(typed_actor<Xs...> f, typed_actor<Ys...> g) { operator*(typed_actor<Xs...> f, typed_actor<Ys...> g) {
using result using result
= composed_type_t<detail::type_list<Xs...>, detail::type_list<Ys...>>; = composed_type_t<detail::type_list<Xs...>, detail::type_list<Ys...>>;
......
...@@ -38,10 +38,6 @@ struct testee_state { ...@@ -38,10 +38,6 @@ struct testee_state {
pending = self->run_delayed(10s, [this] { run_delayed_called = true; }); pending = self->run_delayed(10s, [this] { run_delayed_called = true; });
}, },
[](const std::string&) { CAF_LOG_TRACE(""); }, [](const std::string&) { CAF_LOG_TRACE(""); },
[this](group& grp) {
CAF_LOG_TRACE("");
self->join(grp);
},
}; };
} }
}; };
...@@ -121,24 +117,4 @@ CAF_TEST(delay_actor_message) { ...@@ -121,24 +117,4 @@ CAF_TEST(delay_actor_message) {
expect((std::string), from(aut).to(aut).with("foo")); expect((std::string), from(aut).to(aut).with("foo"));
} }
CAF_TEST(delay_group_message) {
// Have AUT join the group.
auto grp = sys.groups().anonymous();
self->send(aut, grp);
expect((group), from(self).to(aut).with(_));
// Schedule a message for now + 10s.
auto n = t.now() + 10s;
auto autptr = actor_cast<strong_actor_ptr>(aut);
t.schedule_message(n, std::move(grp), autptr, make_message("foo"));
CHECK_EQ(t.actions.size(), 1u);
// Advance time to send the message.
t.advance_time(10s);
CHECK_EQ(t.actions.size(), 0u);
// Have AUT receive the message.
expect((std::string), from(aut).to(aut).with("foo"));
// Kill AUT (necessary because the group keeps a reference around).
self->send_exit(aut, exit_reason::kill);
expect((exit_msg), from(self).to(aut).with(_));
}
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE aout
#include "core-test.hpp"
#include "caf/all.hpp"
using namespace caf;
using std::endl;
namespace {
constexpr const char* global_redirect = ":test";
constexpr const char* local_redirect = ":test2";
constexpr const char* chatty_line = "hi there!:)";
constexpr const char* chattier_line = "hello there, fellow friend!:)";
void chatty_actor(event_based_actor* self) {
aout(self) << chatty_line << endl;
}
void chattier_actor(event_based_actor* self, const std::string& fn) {
aout(self) << chatty_line << endl;
actor_ostream::redirect(self, fn);
aout(self) << chattier_line << endl;
}
struct fixture {
fixture() : system(cfg) {
// nop
}
actor_system_config cfg;
actor_system system;
scoped_actor self{system, true};
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(redirect_aout_globally) {
self->join(system.groups().get_local(global_redirect));
actor_ostream::redirect_all(system, global_redirect);
system.spawn(chatty_actor);
self->receive([](const std::string& virtual_file, std::string& line) {
// drop trailing '\n'
if (!line.empty())
line.pop_back();
CHECK_EQ(virtual_file, ":test");
CHECK_EQ(line, chatty_line);
});
self->await_all_other_actors_done();
CHECK_EQ(self->mailbox().size(), 0u);
}
CAF_TEST(global_and_local_redirect) {
self->join(system.groups().get_local(global_redirect));
self->join(system.groups().get_local(local_redirect));
actor_ostream::redirect_all(system, global_redirect);
system.spawn(chatty_actor);
system.spawn(chattier_actor, local_redirect);
std::vector<std::pair<std::string, std::string>> expected{
{":test", chatty_line}, {":test", chatty_line}, {":test2", chattier_line}};
std::vector<std::pair<std::string, std::string>> lines;
int i = 0;
self->receive_for(i, 3)([&](std::string& virtual_file, std::string& line) {
// drop trailing '\n'
if (!line.empty())
line.pop_back();
lines.emplace_back(std::move(virtual_file), std::move(line));
});
CHECK(std::is_permutation(lines.begin(), lines.end(), expected.begin()));
self->await_all_other_actors_done();
CHECK_EQ(self->mailbox().size(), 0u);
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE composition
#include "caf/actor.hpp"
#include "core-test.hpp"
using namespace caf;
namespace {
behavior multiplier(int x) {
return {[=](int y) { return x * y; },
[=](int y1, int y2) { return x * y1 * y2; }};
}
behavior adder(int x) {
return {[=](int y) { return x + y; },
[=](int y1, int y2) { return x + y1 + y2; }};
}
behavior float_adder(float x) {
return {[=](float y) { return x + y; }};
}
using fixture = test_coordinator_fixture<>;
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(depth2) {
auto stage1 = sys.spawn(multiplier, 4);
auto stage2 = sys.spawn(adder, 10);
auto testee = stage2 * stage1;
self->send(testee, 1);
expect((int), from(self).to(stage1).with(1));
expect((int), from(self).to(stage2).with(4));
expect((int), from(stage2).to(self).with(14));
}
CAF_TEST(depth3) {
auto stage1 = sys.spawn(multiplier, 4);
auto stage2 = sys.spawn(adder, 10);
auto testee = stage1 * stage2 * stage1;
self->send(testee, 1);
expect((int), from(self).to(stage1).with(1));
expect((int), from(self).to(stage2).with(4));
expect((int), from(self).to(stage1).with(14));
expect((int), from(stage1).to(self).with(56));
}
CAF_TEST(depth2_type_mismatch) {
auto stage1 = sys.spawn(multiplier, 4);
auto stage2 = sys.spawn(float_adder, 10);
auto testee = stage2 * stage1;
self->send(testee, 1);
expect((int), from(self).to(stage1).with(1));
expect((int), from(self).to(stage2).with(4));
expect((error), from(stage2).to(self).with(sec::unexpected_message));
}
END_FIXTURE_SCOPE()
...@@ -12,6 +12,8 @@ ...@@ -12,6 +12,8 @@
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(err); } #define ERROR_HANDLER [&](error& err) { CAF_FAIL(err); }
CAF_PUSH_DEPRECATED_WARNING
using namespace caf; using namespace caf;
namespace { namespace {
...@@ -150,3 +152,5 @@ CAF_TEST(dot_composition_2) { ...@@ -150,3 +152,5 @@ CAF_TEST(dot_composition_2) {
} }
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
CAF_POP_WARNINGS
...@@ -67,18 +67,19 @@ public: ...@@ -67,18 +67,19 @@ public:
/// `Transport`. /// `Transport`.
/// @param pull Source for pulling data to send. /// @param pull Source for pulling data to send.
/// @param push Source for pushing received data. /// @param push Source for pushing received data.
template <class Transport = stream_transport, class Connection> template <class Connection>
static disposable run(actor_system& sys, Connection conn, static disposable run(actor_system& sys, Connection conn,
async::consumer_resource<binary::frame> pull, async::consumer_resource<binary::frame> pull,
async::producer_resource<binary::frame> push) { async::producer_resource<binary::frame> push) {
using trait_t = binary::default_trait; using trait_t = binary::default_trait;
using transport_t = typename Connection::transport_type;
auto mpx = sys.network_manager().mpx_ptr(); auto mpx = sys.network_manager().mpx_ptr();
auto fc = flow_connector<trait_t>::make_trivial(std::move(pull), auto fc = flow_connector<trait_t>::make_trivial(std::move(pull),
std::move(push)); std::move(push));
auto bridge = binary::flow_bridge<trait_t>::make(mpx, std::move(fc)); auto bridge = binary::flow_bridge<trait_t>::make(mpx, std::move(fc));
auto bridge_ptr = bridge.get(); auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge)); auto impl = length_prefix_framing::make(std::move(bridge));
auto transport = Transport::make(std::move(conn), std::move(impl)); auto transport = transport_t::make(std::move(conn), std::move(impl));
auto ptr = socket_manager::make(mpx, std::move(transport)); auto ptr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable()); bridge_ptr->self_ref(ptr->as_disposable());
mpx->start(ptr); mpx->start(ptr);
...@@ -90,7 +91,7 @@ public: ...@@ -90,7 +91,7 @@ public:
/// @param conn A connected stream socket or SSL connection, depending on the /// @param conn A connected stream socket or SSL connection, depending on the
/// `Transport`. /// `Transport`.
/// @param init Function object for setting up the created flows. /// @param init Function object for setting up the created flows.
template <class Transport = stream_transport, class Connection, class Init> template <class Connection, class Init>
static disposable run(actor_system& sys, Connection conn, Init init) { static disposable run(actor_system& sys, Connection conn, Init init) {
static_assert(std::is_invocable_v<Init, connect_event_t&&>, static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init"); "invalid signature found for init");
......
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
#include "caf/byte_span.hpp" #include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/net/ssl/errc.hpp" #include "caf/net/ssl/errc.hpp"
#include "caf/net/ssl/fwd.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include <cstddef> #include <cstddef>
...@@ -18,6 +19,9 @@ class CAF_NET_EXPORT connection { ...@@ -18,6 +19,9 @@ class CAF_NET_EXPORT connection {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// The default transport for exchanging raw bytes over an SSL connection.
using transport_type = transport;
/// The opaque implementation type. /// The opaque implementation type.
struct impl; struct impl;
......
...@@ -9,6 +9,7 @@ ...@@ -9,6 +9,7 @@
#include "caf/byte_span.hpp" #include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
// Note: This API mostly wraps platform-specific functions that return ssize_t. // Note: This API mostly wraps platform-specific functions that return ssize_t.
...@@ -20,8 +21,12 @@ namespace caf::net { ...@@ -20,8 +21,12 @@ namespace caf::net {
/// A connection-oriented network communication endpoint for bidirectional byte /// A connection-oriented network communication endpoint for bidirectional byte
/// streams. /// streams.
struct CAF_NET_EXPORT stream_socket : network_socket { struct CAF_NET_EXPORT stream_socket : network_socket {
/// The parent type.
using super = network_socket; using super = network_socket;
/// The default transport for exchanging raw bytes over a stream socket.
using transport_type = stream_transport;
using super::super; using super::super;
constexpr stream_socket fd() const noexcept { constexpr stream_socket fd() const noexcept {
......
.. _groups:
Group Communication
===================
CAF supports publish/subscribe-based group communication. Dynamically typed
actors can join and leave groups and send messages to groups. The following
example showcases the basic API for retrieving a group from a module by its
name, joining, and leaving.
.. code-block:: C++
auto expected_grp = system.groups().get("local", "foo");
if (!expected_grp) {
std::cerr << "*** cannot load group: " << to_string(expected_grp.error())
<< std::endl;
return;
}
auto grp = std::move(*expected_grp);
scoped_actor self{system};
self->join(grp);
self->send(grp, "test");
self->receive(
[](const std::string& str) {
assert(str == "test");
}
);
self->leave(grp);
It is worth mentioning that the module ``"local"`` is guaranteed to
never return an error. The example above uses the general API for retrieving
the group. However, local modules can be easier accessed by calling
``system.groups().get_local(id)``, which returns ``group``
instead of ``expected<group>``.
.. _anonymous-group:
Anonymous Groups
----------------
Groups created on-the-fly with ``system.groups().anonymous()`` can be
used to coordinate a set of workers. Each call to this function returns a new,
unique group instance.
.. _local-group:
Local Groups
------------
The ``"local"`` group module creates groups for in-process
communication. For example, a group for GUI related events could be identified
by ``system.groups().get_local("GUI events")``. The group ID
``"GUI events"`` uniquely identifies a singleton group instance of the
module ``"local"``.
.. _remote-group:
Remote Groups
-------------
Calling``system.middleman().publish_local_groups(port, addr)`` makes
all local groups available to other nodes in the network. The first argument
denotes the port, while the second (optional) parameter can be used to
whitelist IP addresses.
After publishing the group at one node (the server), other nodes (the clients)
can get a handle for that group by using the ``remote`` module:
``system.groups().get("remote", "<group>@<host>:<port>")``. This implementation
uses N-times unicast underneath and the group is only available as long as the
hosting server is alive.
...@@ -21,7 +21,6 @@ Contents ...@@ -21,7 +21,6 @@ Contents
ReferenceCounting ReferenceCounting
Error Error
ConfiguringActorApplications ConfiguringActorApplications
GroupCommunication
ManagingGroupsOfWorkers ManagingGroupsOfWorkers
DataFlows DataFlows
Testing Testing
......
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