Commit 9909c679 authored by Dominik Charousset's avatar Dominik Charousset

Fix protobuf example, close #491

parent 50526ff3
...@@ -67,7 +67,7 @@ if(NOT CAF_NO_PROTOBUF_EXAMPLES) ...@@ -67,7 +67,7 @@ if(NOT CAF_NO_PROTOBUF_EXAMPLES)
include_directories(${PROTOBUF_INCLUDE_DIR}) include_directories(${PROTOBUF_INCLUDE_DIR})
# add binary dir as include path as generated headers will be located there # add binary dir as include path as generated headers will be located there
include_directories(${CMAKE_CURRENT_BINARY_DIR}) include_directories(${CMAKE_CURRENT_BINARY_DIR})
add_executable(protobuf_broker brokers/protobuf_broker.cpp ${ProtoSources}) add_executable(protobuf_broker broker/protobuf_broker.cpp ${ProtoSources})
target_link_libraries(protobuf_broker ${CMAKE_DL_LIBS} ${CAF_LIBRARIES} ${PTHREAD_LIBRARIES} ${PROTOBUF_LIBRARIES}) target_link_libraries(protobuf_broker ${CMAKE_DL_LIBS} ${CAF_LIBRARIES} ${PTHREAD_LIBRARIES} ${PROTOBUF_LIBRARIES})
add_dependencies(protobuf_broker all_examples) add_dependencies(protobuf_broker all_examples)
endif(PROTOBUF_FOUND AND PROTOBUF_PROTOC_EXECUTABLE) endif(PROTOBUF_FOUND AND PROTOBUF_PROTOC_EXECUTABLE)
......
...@@ -13,41 +13,64 @@ CAF_PUSH_WARNINGS ...@@ -13,41 +13,64 @@ CAF_PUSH_WARNINGS
#include "pingpong.pb.h" #include "pingpong.pb.h"
CAF_POP_WARNINGS CAF_POP_WARNINGS
namespace {
using namespace std; using namespace std;
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
void print_on_exit(const actor& hdl, const std::string& name) { using ping_atom = atom_constant<atom("ping")>;
hdl->attach_functor([=](abstract_actor* ptr, exit_reason reason) { using pong_atom = atom_constant<atom("pong")>;
aout(ptr) << name << " exited with reason " << reason << endl; using kickoff_atom = atom_constant<atom("kickoff")>;
// utility function to print an exit message with custom name
void print_on_exit(scheduled_actor* self, const std::string& name) {
self->attach_functor([=](const error& reason) {
aout(self) << name << " exited: " << self->home_system().render(reason)
<< endl;
}); });
} }
behavior ping(event_based_actor* self, size_t num_pings) { struct ping_state {
auto count = make_shared<size_t>(0); size_t count = 0;
};
behavior ping(stateful_actor<ping_state>* self, size_t num_pings) {
print_on_exit(self, "ping");
return { return {
on(atom("kickoff"), arg_match) >> [=](const actor& pong) { [=](kickoff_atom, const actor& pong) {
self->send(pong, atom("ping"), 1); self->send(pong, ping_atom::value, 1);
self->become ( self->become (
on(atom("pong"), arg_match) >> [=](int value) -> message { [=](pong_atom, int value) -> message {
if (++*count >= num_pings) self->quit(); if (++(self->state.count) >= num_pings)
return make_message(atom("ping"), value + 1); self->quit();
return make_message(ping_atom::value, value + 1);
} }
); );
} }
}; };
} }
behavior pong() { behavior pong(event_based_actor* self) {
print_on_exit(self, "pong");
return { return {
on(atom("ping"), arg_match) >> [](int value) { [=](ping_atom, int value) {
return make_message(atom("pong"), value); return make_message(pong_atom::value, value);
} }
}; };
} }
void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) { void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) {
print_on_exit(self, "protobuf_io");
aout(self) << "protobuf broker started" << endl;
self->monitor(buddy); self->monitor(buddy);
self->set_down_handler(
[=](const down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl;
self->quit(dm.reason);
}
});
auto write = [=](const org::libcppa::PingOrPong& p) { auto write = [=](const org::libcppa::PingOrPong& p) {
string buf = p.SerializeAsString(); string buf = p.SerializeAsString();
int32_t s = htonl(static_cast<int32_t>(buf.size())); int32_t s = htonl(static_cast<int32_t>(buf.size()));
...@@ -61,26 +84,17 @@ void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -61,26 +84,17 @@ void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) {
self->send_exit(buddy, exit_reason::remote_link_unreachable); self->send_exit(buddy, exit_reason::remote_link_unreachable);
self->quit(exit_reason::remote_link_unreachable); self->quit(exit_reason::remote_link_unreachable);
}, },
on(atom("ping"), arg_match) >> [=](int i) { [=](ping_atom, int i) {
aout(self) << "'ping' " << i << endl; aout(self) << "'ping' " << i << endl;
org::libcppa::PingOrPong p; org::libcppa::PingOrPong p;
p.mutable_ping()->set_id(i); p.mutable_ping()->set_id(i);
write(p); write(p);
}, },
on(atom("pong"), arg_match) >> [=](int i) { [=](pong_atom, int i) {
aout(self) << "'pong' " << i << endl; aout(self) << "'pong' " << i << endl;
org::libcppa::PingOrPong p; org::libcppa::PingOrPong p;
p.mutable_pong()->set_id(i); p.mutable_pong()->set_id(i);
write(p); write(p);
},
[=](const down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl;
self->quit(dm.reason);
}
},
others >> [=] {
cout << "unexpected: " << to_string(self->current_message()) << endl;
} }
}; };
auto await_protobuf_data = message_handler { auto await_protobuf_data = message_handler {
...@@ -88,10 +102,10 @@ void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -88,10 +102,10 @@ void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) {
org::libcppa::PingOrPong p; org::libcppa::PingOrPong p;
p.ParseFromArray(msg.buf.data(), static_cast<int>(msg.buf.size())); p.ParseFromArray(msg.buf.data(), static_cast<int>(msg.buf.size()));
if (p.has_ping()) { if (p.has_ping()) {
self->send(buddy, atom("ping"), p.ping().id()); self->send(buddy, ping_atom::value, p.ping().id());
} }
else if (p.has_pong()) { else if (p.has_pong()) {
self->send(buddy, atom("pong"), p.pong().id()); self->send(buddy, pong_atom::value, p.pong().id());
} }
else { else {
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
...@@ -124,54 +138,55 @@ void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -124,54 +138,55 @@ void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) {
} }
behavior server(broker* self, actor buddy) { behavior server(broker* self, actor buddy) {
print_on_exit(self, "server");
aout(self) << "server is running" << endl; aout(self) << "server is running" << endl;
return { return {
[=](const new_connection_msg& msg) { [=](const new_connection_msg& msg) {
aout(self) << "server accepted new connection" << endl; aout(self) << "server accepted new connection" << endl;
auto io_actor = self->fork(protobuf_io, msg.handle, buddy); auto io_actor = self->fork(protobuf_io, msg.handle, buddy);
print_on_exit(io_actor, "protobuf_io");
// only accept 1 connection in our example // only accept 1 connection in our example
self->quit(); self->quit();
},
others >> [=] {
cout << "unexpected: " << to_string(self->current_message()) << endl;
} }
}; };
} }
maybe<uint16_t> as_u16(const std::string& str) { class config : public actor_system_config {
return static_cast<uint16_t>(stoul(str)); public:
} uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
int main(int argc, char** argv) { config() {
actor_system_config cfg; opt_group{custom_options_, "global"}
cfg.load<io::middleman>(); .add(port, "port,p", "set port")
actor_system system{cfg}; .add(host, "host,H", "set host (ignored in server mode)")
message_builder{argv + 1, argv + argc}.apply({ .add(server_mode, "server-mode,s", "enable server mode");
on("-s", as_u16) >> [&](uint16_t port) { }
cout << "run in server mode" << endl; };
auto pong_actor = system.spawn(pong);
auto server_actor = system.middleman().spawn_server(server, port, void caf_main(actor_system& system, const config& cfg) {
pong_actor); if (cfg.server_mode) {
if (server_actor) { cout << "run in server mode" << endl;
print_on_exit(*server_actor, "server"); auto pong_actor = system.spawn(pong);
print_on_exit(pong_actor, "pong"); auto server_actor = system.middleman().spawn_server(server, cfg.port,
} pong_actor);
}, if (!server_actor)
on("-c", val<string>, as_u16) >> [&](const string& host, uint16_t port) { cout << "unable to spawn server: "
auto ping_actor = system.spawn(ping, 20); << system.render(server_actor.error()) << endl;
auto io_actor = system.middleman().spawn_client(protobuf_io, host, return;
port, ping_actor); }
if (io_actor) { cout << "run in client mode" << endl;
print_on_exit(ping_actor, "ping"); auto ping_actor = system.spawn(ping, 20u);
print_on_exit(*io_actor, "protobuf_io"); auto io_actor = system.middleman().spawn_client(protobuf_io, cfg.host,
send_as(*io_actor, ping_actor, atom("kickoff"), *io_actor); cfg.port, ping_actor);
} if (!io_actor) {
}, cout << "cannot connect to " << cfg.host << " at port " << cfg.port
others >> [] { << ": " << system.render(io_actor.error()) << endl;
cerr << "use with eihter '-s PORT' as server or " return;
"'-c HOST PORT' as client" }
<< endl; send_as(*io_actor, ping_actor, kickoff_atom::value, *io_actor);
}
});
} }
} // namespace <anonymous>
CAF_MAIN(io::middleman)
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