Commit c7f1f0f6 authored by Dominik Charousset's avatar Dominik Charousset

Better structuring of client/server examples

parent 5a4a1b7f
...@@ -164,17 +164,17 @@ public: ...@@ -164,17 +164,17 @@ public:
} }
}; };
void caf_main(actor_system& system, const config& cfg) { void run_server(actor_system& system, const config& cfg) {
if (cfg.server_mode) { cout << "run in server mode" << endl;
cout << "run in server mode" << endl; auto pong_actor = system.spawn(pong);
auto pong_actor = system.spawn(pong); auto server_actor = system.middleman().spawn_server(server, cfg.port,
auto server_actor = system.middleman().spawn_server(server, cfg.port, pong_actor);
pong_actor); if (!server_actor)
if (!server_actor) cerr << "unable to spawn server: "
cout << "unable to spawn server: " << system.render(server_actor.error()) << endl;
<< system.render(server_actor.error()) << endl; }
return;
} void run_client(actor_system& system, const config& cfg) {
cout << "run in client mode" << endl; cout << "run in client mode" << endl;
auto ping_actor = system.spawn(ping, 20u); auto ping_actor = system.spawn(ping, 20u);
auto io_actor = system.middleman().spawn_client(protobuf_io, cfg.host, auto io_actor = system.middleman().spawn_client(protobuf_io, cfg.host,
...@@ -187,6 +187,11 @@ void caf_main(actor_system& system, const config& cfg) { ...@@ -187,6 +187,11 @@ void caf_main(actor_system& system, const config& cfg) {
send_as(*io_actor, ping_actor, kickoff_atom::value, *io_actor); send_as(*io_actor, ping_actor, kickoff_atom::value, *io_actor);
} }
void caf_main(actor_system& system, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client;
f(system, cfg);
}
} // namespace <anonymous> } // namespace <anonymous>
CAF_MAIN(io::middleman) CAF_MAIN(io::middleman)
...@@ -189,21 +189,21 @@ public: ...@@ -189,21 +189,21 @@ public:
} }
}; };
void caf_main(actor_system& system, const config& cfg) { void run_server(actor_system& system, const config& cfg) {
if (cfg.server_mode) { cout << "run in server mode" << endl;
cout << "run in server mode" << endl; auto pong_actor = system.spawn(pong);
auto pong_actor = system.spawn(pong); auto server_actor = system.middleman().spawn_server(server, cfg.port,
auto server_actor = system.middleman().spawn_server(server, cfg.port, pong_actor);
pong_actor); if (!server_actor) {
if (!server_actor) { std::cerr << "failed to spawn server: "
std::cerr << "failed to spawn server: " << system.render(server_actor.error()) << endl;
<< system.render(server_actor.error()) << endl;
return;
}
print_on_exit(*server_actor, "server");
print_on_exit(pong_actor, "pong");
return; return;
} }
print_on_exit(*server_actor, "server");
print_on_exit(pong_actor, "pong");
}
void run_client(actor_system& system, const config& cfg) {
auto ping_actor = system.spawn(ping, size_t{20}); auto ping_actor = system.spawn(ping, size_t{20});
auto io_actor = system.middleman().spawn_client(broker_impl, cfg.host, auto io_actor = system.middleman().spawn_client(broker_impl, cfg.host,
cfg.port, ping_actor); cfg.port, ping_actor);
...@@ -217,6 +217,11 @@ void caf_main(actor_system& system, const config& cfg) { ...@@ -217,6 +217,11 @@ void caf_main(actor_system& system, const config& cfg) {
send_as(*io_actor, ping_actor, kickoff_atom::value, *io_actor); send_as(*io_actor, ping_actor, kickoff_atom::value, *io_actor);
} }
void caf_main(actor_system& system, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client;
f(system, cfg);
}
} // namespace <anonymous> } // namespace <anonymous>
CAF_MAIN(io::middleman) CAF_MAIN(io::middleman)
...@@ -83,7 +83,7 @@ void caf_main(actor_system& system, const config& cfg) { ...@@ -83,7 +83,7 @@ void caf_main(actor_system& system, const config& cfg) {
<< system.render(server_actor.error()) << endl; << system.render(server_actor.error()) << endl;
return; return;
} }
cout << "*** run in server mode listen on: " << cfg.port << endl; cout << "*** listening on port " << cfg.port << endl;
cout << "*** to quit the program, simply press <enter>" << endl; cout << "*** to quit the program, simply press <enter>" << endl;
// wait for any input // wait for any input
std::string dummy; std::string dummy;
......
...@@ -261,32 +261,37 @@ public: ...@@ -261,32 +261,37 @@ public:
} }
}; };
void caf_main(actor_system& system, const config& cfg) { void run_server(actor_system& system, const config& cfg) {
if (!cfg.server_mode && cfg.port == 0) { auto calc = system.spawn(calculator_fun);
cerr << "*** no port to server specified" << endl; // try to publish math actor at given port
cout << "*** try publish at port " << cfg.port << endl;
auto expected_port = system.middleman().publish(calc, cfg.port);
if (!expected_port) {
std::cerr << "*** publish failed: "
<< system.render(expected_port.error()) << endl;
return; return;
} }
if (cfg.server_mode) { cout << "*** server successfully published at port " << *expected_port << endl
auto calc = system.spawn(calculator_fun); << "*** press [enter] to quit" << endl;
// try to publish math actor at given port string dummy;
cout << "*** try publish at port " << cfg.port << endl; std::getline(std::cin, dummy);
auto expected_port = system.middleman().publish(calc, cfg.port); cout << "... cya" << endl;
if (!expected_port) { anon_send_exit(calc, exit_reason::user_shutdown);
std::cerr << "*** publish failed: " }
<< system.render(expected_port.error()) << endl;
} else { void run_client(actor_system& system, const config& cfg) {
cout << "*** server successfully published at port " << *expected_port if (cfg.port == 0) {
<< endl << "*** press [enter] to quit" << endl; cerr << "*** no port to server specified" << endl;
string dummy;
std::getline(std::cin, dummy);
cout << "... cya" << endl;
anon_send_exit(calc, exit_reason::user_shutdown);
}
return; return;
} }
client_repl(system, cfg.host, cfg.port); client_repl(system, cfg.host, cfg.port);
} }
void caf_main(actor_system& system, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client;
f(system, cfg);
}
} // namespace <anonymous> } // namespace <anonymous>
CAF_MAIN(io::middleman) CAF_MAIN(io::middleman)
...@@ -3,7 +3,7 @@ ...@@ -3,7 +3,7 @@
* based on group communication. * * based on group communication. *
* * * *
* Setup for a minimal chat between "alice" and "bob": * * Setup for a minimal chat between "alice" and "bob": *
* - ./build/bin/group_server -p 4242 * * - ./build/bin/group_chat -s -p 4242 *
* - ./build/bin/group_chat -g remote:chatroom@localhost:4242 -n alice * * - ./build/bin/group_chat -g remote:chatroom@localhost:4242 -n alice *
* - ./build/bin/group_chat -g remote:chatroom@localhost:4242 -n bob * * - ./build/bin/group_chat -g remote:chatroom@localhost:4242 -n bob *
\ ******************************************************************************/ \ ******************************************************************************/
...@@ -34,8 +34,8 @@ istream& operator>>(istream& is, line& l) { ...@@ -34,8 +34,8 @@ istream& operator>>(istream& is, line& l) {
return is; return is;
} }
void client(event_based_actor* self, const string& name) { behavior client(event_based_actor* self, const string& name) {
self->become ( return {
[=](broadcast_atom, const string& message) { [=](broadcast_atom, const string& message) {
for(auto& dest : self->joined_groups()) { for(auto& dest : self->joined_groups()) {
self->send(dest, name + ": " + message); self->send(dest, name + ": " + message);
...@@ -53,29 +53,46 @@ void client(event_based_actor* self, const string& name) { ...@@ -53,29 +53,46 @@ void client(event_based_actor* self, const string& name) {
}, },
[=](const string& txt) { [=](const string& txt) {
// don't print own messages // don't print own messages
if (self->current_sender() != self) { if (self->current_sender() != self)
cout << txt << endl; cout << txt << endl;
}
}, },
[=](const group_down_msg& g) { [=](const group_down_msg& g) {
cout << "*** chatroom offline: " << to_string(g.source) << endl; cout << "*** chatroom offline: " << to_string(g.source) << endl;
} }
); };
} }
class config : public actor_system_config { class config : public actor_system_config {
public: public:
std::string name; std::string name;
std::string group_id; std::vector<std::string> group_uris;
uint16_t port = 0;
bool server_mode = false;
config() { config() {
opt_group{custom_options_, "global"} opt_group{custom_options_, "global"}
.add(name, "name,n", "set name") .add(name, "name,n", "set name")
.add(group_id, "group,g", "join group"); .add(group_uris, "group,g", "join group")
.add(server_mode, "server,s", "run in server mode")
.add(port, "port,p", "set port (ignored in client mode)");
} }
}; };
void caf_main(actor_system& system, const config& cfg) { void run_server(actor_system& system, const config& cfg) {
auto res = system.middleman().publish_local_groups(cfg.port);
if (! res) {
std::cerr << "*** publishing local groups failed: "
<< system.render(res.error()) << endl;
return;
}
cout << "*** listening at port " << *res << endl
<< "*** press [enter] to quit" << endl;
string dummy;
std::getline(std::cin, dummy);
cout << "... cya" << endl;
}
void run_client(actor_system& system, const config& cfg) {
auto name = cfg.name; auto name = cfg.name;
while (name.empty()) { while (name.empty()) {
cout << "please enter your name: " << flush; cout << "please enter your name: " << flush;
...@@ -84,27 +101,16 @@ void caf_main(actor_system& system, const config& cfg) { ...@@ -84,27 +101,16 @@ void caf_main(actor_system& system, const config& cfg) {
return; return;
} }
} }
cout << "*** starting client, type '/help' for a list of commands" << endl;
auto client_actor = system.spawn(client, name); auto client_actor = system.spawn(client, name);
// evaluate group parameters for (auto& uri : cfg.group_uris) {
if (!cfg.group_id.empty()) { auto tmp = system.groups().get(uri);
auto p = cfg.group_id.find(':'); if (tmp)
if (p == std::string::npos) { anon_send(client_actor, join_atom::value, std::move(*tmp));
cerr << "*** error parsing argument " << cfg.group_id else
<< ", expected format: <module_name>:<group_id>"; cerr << "*** failed to parse \"" << uri << "\" as group URI: "
} else { << system.render(tmp.error()) << 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;
}
anon_send(client_actor, join_atom::value, *g);
}
} }
cout << "*** starting client, type '/help' for a list of commands" << endl;
istream_iterator<line> eof; istream_iterator<line> eof;
vector<string> words; vector<string> words;
for (istream_iterator<line> i(cin); i != eof; ++i) { for (istream_iterator<line> i(cin); i != eof; ++i) {
...@@ -147,6 +153,11 @@ void caf_main(actor_system& system, const config& cfg) { ...@@ -147,6 +153,11 @@ void caf_main(actor_system& system, const config& cfg) {
anon_send_exit(client_actor, exit_reason::user_shutdown); anon_send_exit(client_actor, exit_reason::user_shutdown);
} }
void caf_main(actor_system& system, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client;
f(system, cfg);
}
} // namespace <anonymous> } // namespace <anonymous>
CAF_MAIN(io::middleman) CAF_MAIN(io::middleman)
...@@ -7,8 +7,8 @@ ...@@ -7,8 +7,8 @@
// Run client at the same host: // Run client at the same host:
// - remote_spawn -n localhost -p 4242 // - remote_spawn -n localhost -p 4242
// Manual refs: 137-139, 144, 148 (ConfiguringActorApplications) // Manual refs: 33-39, 99-101,106,110 (ConfiguringActorApplications)
// 101-119 (RemoteSpawn) // 124-142 (RemoteSpawn)
#include <array> #include <array>
#include <vector> #include <vector>
...@@ -36,7 +36,6 @@ using sub_atom = atom_constant<atom("sub")>; ...@@ -36,7 +36,6 @@ using sub_atom = atom_constant<atom("sub")>;
using calculator = typed_actor<replies_to<add_atom, int, int>::with<int>, using calculator = typed_actor<replies_to<add_atom, int, int>::with<int>,
replies_to<sub_atom, int, int>::with<int>>; replies_to<sub_atom, int, int>::with<int>>;
calculator::behavior_type calculator_fun(calculator::pointer self) { calculator::behavior_type calculator_fun(calculator::pointer self) {
return { return {
[=](add_atom, int a, int b) -> int { [=](add_atom, int a, int b) -> int {
...@@ -95,11 +94,35 @@ void client_repl(function_view<calculator> f) { ...@@ -95,11 +94,35 @@ void client_repl(function_view<calculator> f) {
usage(); usage();
} }
} }
}
struct config : actor_system_config {
config() {
add_actor_type("calculator", calculator_fun);
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port")
.add(host, "host,H", "set node (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode");
}
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
};
void server(actor_system& system, const config& cfg) {
auto res = system.middleman().open(cfg.port);
if (!res) {
std::cerr << "*** cannot open port: "
<< system.render(res.error()) << std::endl;
}
std::cout << "*** running on port: "
<< *res << std::endl
<< "*** press <enter> to shutdown server" << std::endl;
getchar();
} }
void client(actor_system& system, const std::string& host, uint16_t port) { void client(actor_system& system, const config& cfg) {
auto node = system.middleman().connect(host, port); auto node = system.middleman().connect(cfg.host, cfg.port);
if (!node) { if (!node) {
std::cerr << "*** connect failed: " std::cerr << "*** connect failed: "
<< system.render(node.error()) << std::endl; << system.render(node.error()) << std::endl;
...@@ -121,37 +144,9 @@ void client(actor_system& system, const std::string& host, uint16_t port) { ...@@ -121,37 +144,9 @@ void client(actor_system& system, const std::string& host, uint16_t port) {
anon_send_exit(*worker, exit_reason::kill); anon_send_exit(*worker, exit_reason::kill);
} }
void server(actor_system& system, uint16_t port) {
auto res = system.middleman().open(port);
if (!res) {
std::cerr << "*** cannot open port: "
<< system.render(res.error()) << std::endl;
}
std::cout << "*** running on port: "
<< *res << std::endl
<< "*** press <enter> to shutdown server" << std::endl;
getchar();
}
struct config : actor_system_config {
config() {
add_actor_type("calculator", calculator_fun);
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port")
.add(host, "host,H", "set node (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode");
}
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
};
void caf_main(actor_system& system, const config& cfg) { void caf_main(actor_system& system, const config& cfg) {
if (cfg.server_mode) auto f = cfg.server_mode ? server : client;
server(system, cfg.port); f(system, cfg);
else
client(system, cfg.host, cfg.port);
} }
} // namespace <anonymous> } // namespace <anonymous>
......
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