Commit a6a9d24e authored by Dominik Charousset's avatar Dominik Charousset

fixed protobuf example

parent 4fdb2cb4
...@@ -52,7 +52,7 @@ class actor_ostream { ...@@ -52,7 +52,7 @@ class actor_ostream {
actor_ostream& operator=(actor_ostream&&) = default; actor_ostream& operator=(actor_ostream&&) = default;
actor_ostream& operator=(const actor_ostream&) = default; actor_ostream& operator=(const actor_ostream&) = default;
explicit actor_ostream(local_actor* self); explicit actor_ostream(actor self);
actor_ostream& write(std::string arg); actor_ostream& write(std::string arg);
...@@ -87,17 +87,15 @@ class actor_ostream { ...@@ -87,17 +87,15 @@ class actor_ostream {
private: private:
local_actor* m_self; actor m_self;
actor m_printer; actor m_printer;
}; };
inline actor_ostream aout(local_actor* self) { inline actor_ostream aout(actor self) {
return actor_ostream{self}; return actor_ostream{self};
} }
actor_ostream aout(scoped_actor& self);
} // namespace cppa } // namespace cppa
namespace std { namespace std {
......
...@@ -120,6 +120,8 @@ class broker : public extend<local_actor>:: ...@@ -120,6 +120,8 @@ class broker : public extend<local_actor>::
static broker_ptr from(std::function<void (broker*)> fun, acceptor_uptr in); static broker_ptr from(std::function<void (broker*)> fun, acceptor_uptr in);
static broker_ptr from(std::function<behavior (broker*)> fun, acceptor_uptr in);
template<typename F, typename T0, typename... Ts> template<typename F, typename T0, typename... Ts>
static broker_ptr from(F fun, acceptor_uptr in, T0&& arg0, Ts&&... args) { static broker_ptr from(F fun, acceptor_uptr in, T0&& arg0, Ts&&... args) {
return from(std::bind(std::move(fun), return from(std::bind(std::move(fun),
...@@ -172,6 +174,10 @@ class broker : public extend<local_actor>:: ...@@ -172,6 +174,10 @@ class broker : public extend<local_actor>::
input_stream_ptr in, input_stream_ptr in,
output_stream_ptr out); output_stream_ptr out);
static broker_ptr from_impl(std::function<behavior (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out);
void invoke_message(msg_hdr_cref hdr, any_tuple msg); void invoke_message(msg_hdr_cref hdr, any_tuple msg);
bool invoke_message_from_cache(); bool invoke_message_from_cache();
...@@ -198,7 +204,7 @@ class default_broker : public broker { ...@@ -198,7 +204,7 @@ class default_broker : public broker {
public: public:
typedef std::function<void (broker*)> function_type; typedef std::function<behavior (broker*)> function_type;
default_broker(function_type f, input_stream_ptr in, output_stream_ptr out); default_broker(function_type f, input_stream_ptr in, output_stream_ptr out);
......
...@@ -29,6 +29,8 @@ if (NOT "${CPPA_NO_PROTOBUF_EXAMPLES}" STREQUAL "yes") ...@@ -29,6 +29,8 @@ if (NOT "${CPPA_NO_PROTOBUF_EXAMPLES}" STREQUAL "yes")
if (PROTOBUF_FOUND) if (PROTOBUF_FOUND)
PROTOBUF_GENERATE_CPP(ProtoSources ProtoHeaders "${CMAKE_CURRENT_SOURCE_DIR}/remote_actors/pingpong.proto") PROTOBUF_GENERATE_CPP(ProtoSources ProtoHeaders "${CMAKE_CURRENT_SOURCE_DIR}/remote_actors/pingpong.proto")
include_directories(${PROTOBUF_INCLUDE_DIR}) include_directories(${PROTOBUF_INCLUDE_DIR})
# add binary dir as include path as generated headers will be located there
include_directories(${CMAKE_CURRENT_BINARY_DIR})
add_executable(protobuf_broker remote_actors/protobuf_broker.cpp ${ProtoSources}) add_executable(protobuf_broker remote_actors/protobuf_broker.cpp ${ProtoSources})
target_link_libraries(protobuf_broker ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES} ${PROTOBUF_LIBRARIES}) target_link_libraries(protobuf_broker ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES} ${PROTOBUF_LIBRARIES})
add_dependencies(protobuf_broker all_examples) add_dependencies(protobuf_broker all_examples)
......
...@@ -45,129 +45,134 @@ ...@@ -45,129 +45,134 @@
#include "cppa/io/ipv4_acceptor.hpp" #include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp" #include "cppa/io/ipv4_io_stream.hpp"
CPPA_PUSH_WARNINGS
#include "pingpong.pb.h" #include "pingpong.pb.h"
CPPA_POP_WARNINGS
using namespace std; using namespace std;
using namespace cppa; using namespace cppa;
using namespace cppa::io; using namespace cppa::io;
void print_on_exit(const actor_ptr& ptr, const std::string& name) { void print_on_exit(const actor& ptr, const std::string& name) {
ptr->attach_functor([=](std::uint32_t reason) { ptr->attach_functor([=](std::uint32_t reason) {
aout << name << " exited with reason " << reason << endl; aout(ptr) << name << " exited with reason " << reason << endl;
}); });
} }
void ping(size_t num_pings) { behavior ping(event_based_actor* self, size_t num_pings) {
auto count = make_shared<size_t>(0); auto count = make_shared<size_t>(0);
become ( return {
on(atom("kickoff"), arg_match) >> [=](const actor_ptr& pong) { on(atom("kickoff"), arg_match) >> [=](const actor& pong) {
send(pong, atom("ping"), 1); self->send(pong, atom("ping"), 1);
become ( self->become (
on(atom("pong"), arg_match) >> [=](int value) -> any_tuple { on(atom("pong"), arg_match) >> [=](int value) -> any_tuple {
aout << "pong: " << value << endl;
if (++*count >= num_pings) self->quit(); if (++*count >= num_pings) self->quit();
return {atom("ping"), value + 1}; return make_any_tuple(atom("ping"), value + 1);
} }
); );
} }
); };
} }
void pong() { behavior pong() {
become ( return {
on(atom("ping"), arg_match) >> [](int value) { on(atom("ping"), arg_match) >> [](int value) {
aout << "ping: " << value << endl; return make_any_tuple(atom("pong"), value);
reply(atom("pong"), value);
} }
); };
} }
void protobuf_io(broker* thisptr, connection_handle hdl, const actor_ptr& buddy) { void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) {
self->monitor(buddy); self->monitor(buddy);
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()));
thisptr->write(hdl, sizeof(int32_t), &s); self->write(hdl, sizeof(int32_t), &s);
thisptr->write(hdl, buf.size(), buf.data()); self->write(hdl, buf.size(), buf.data());
}; };
auto default_bhvr = ( partial_function default_bhvr = {
on(atom("IO_closed"), hdl) >> [=] { [=](const connection_closed_msg&) {
cout << "IO_closed" << endl; aout(self) << "connection closed" << endl;
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) { on(atom("ping"), arg_match) >> [=](int i) {
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) { on(atom("pong"), arg_match) >> [=](int i) {
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);
}, },
on_arg_match >> [=](const down_msg& dm) { [=](const down_msg& dm) {
if (dm.source == buddy) self->quit(dm.reason); if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl;
self->quit(dm.reason);
}
}, },
others() >> [=] { others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl; cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
} }
); };
partial_function await_protobuf_data { partial_function await_protobuf_data {
on(atom("IO_read"), hdl, arg_match) >> [=](const util::buffer& buf) { [=](const new_data_msg& msg) {
org::libcppa::PingOrPong p; org::libcppa::PingOrPong p;
p.ParseFromArray(buf.data(), static_cast<int>(buf.size())); p.ParseFromArray(msg.buf.data(), static_cast<int>(msg.buf.size()));
if (p.has_ping()) { if (p.has_ping()) {
send(buddy, atom("ping"), p.ping().id()); self->send(buddy, atom("ping"), p.ping().id());
} }
else if (p.has_pong()) { else if (p.has_pong()) {
send(buddy, atom("pong"), p.pong().id()); self->send(buddy, atom("pong"), p.pong().id());
} }
else { else {
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
cerr << "neither Ping nor Pong!" << endl; cerr << "neither Ping nor Pong!" << endl;
} }
// receive next length prefix // receive next length prefix
thisptr->receive_policy(hdl, broker::exactly, 4); self->receive_policy(hdl, broker::exactly, sizeof(int32_t));
unbecome(); self->unbecome();
}, },
default_bhvr default_bhvr
}; };
partial_function await_length_prefix { partial_function await_length_prefix {
on(atom("IO_read"), hdl, arg_match) >> [=](const util::buffer& buf) { [=](const new_data_msg& msg) {
int32_t num_bytes; int32_t num_bytes;
memcpy(&num_bytes, buf.data(), 4); memcpy(&num_bytes, msg.buf.data(), sizeof(int32_t));
num_bytes = htonl(num_bytes); num_bytes = htonl(num_bytes);
if (num_bytes < 0 || num_bytes > (1024 * 1024)) { if (num_bytes < 0 || num_bytes > (1024 * 1024)) {
aout << "someone is trying something nasty" << endl; aout(self) << "someone is trying something nasty" << endl;
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
return; return;
} }
// receive protobuf data // receive protobuf data
thisptr->receive_policy(hdl, broker::exactly, static_cast<size_t>(num_bytes)); self->receive_policy(hdl, broker::exactly, static_cast<size_t>(num_bytes));
become(keep_behavior, await_protobuf_data); self->become(keep_behavior, await_protobuf_data);
}, },
default_bhvr default_bhvr
}; };
// initial setup // initial setup
thisptr->receive_policy(hdl, broker::exactly, 4); self->receive_policy(hdl, broker::exactly, sizeof(int32_t));
become(await_length_prefix); self->become(await_length_prefix);
} }
void server(broker* thisptr, actor_ptr buddy) { behavior server(broker* self, actor buddy) {
aout << "server is running" << endl; aout(self) << "server is running" << endl;
become ( return {
on(atom("IO_accept"), arg_match) >> [=](accept_handle, connection_handle hdl) { [=](const new_connection_msg& msg) {
aout << "server: IO_accept" << endl; aout(self) << "server accepted new connection" << endl;
auto io_actor = thisptr->fork(protobuf_io, hdl, buddy); auto io_actor = self->fork(protobuf_io, msg.handle, buddy);
print_on_exit(io_actor, "protobuf_io"); print_on_exit(io_actor, "protobuf_io");
// only accept 1 connection // only accept 1 connection in our example
thisptr->quit(); self->quit();
}, },
others() >> [=] { others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl; cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
} }
); };
} }
optional<uint16_t> as_u16(const std::string& str) { optional<uint16_t> as_u16(const std::string& str) {
...@@ -195,6 +200,6 @@ int main(int argc, char** argv) { ...@@ -195,6 +200,6 @@ int main(int argc, char** argv) {
<< endl; << endl;
} }
); );
await_all_others_done(); await_all_actors_done();
shutdown(); shutdown();
} }
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
\******************************************************************************/ \******************************************************************************/
#include "cppa/cppa.hpp"
#include "cppa/scheduler.hpp" #include "cppa/scheduler.hpp"
#include "cppa/singletons.hpp" #include "cppa/singletons.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
...@@ -36,24 +37,20 @@ ...@@ -36,24 +37,20 @@
namespace cppa { namespace cppa {
actor_ostream::actor_ostream(local_actor* self) : m_self(self) { actor_ostream::actor_ostream(actor self) : m_self(std::move(self)) {
m_printer = get_scheduling_coordinator()->printer(); m_printer = get_scheduling_coordinator()->printer();
} }
actor_ostream& actor_ostream::write(std::string arg) { actor_ostream& actor_ostream::write(std::string arg) {
m_self->send(m_printer, atom("add"), move(arg)); send_as(m_self, m_printer, atom("add"), move(arg));
return *this; return *this;
} }
actor_ostream& actor_ostream::flush() { actor_ostream& actor_ostream::flush() {
m_self->send(m_printer, atom("flush")); send_as(m_self, m_printer, atom("flush"));
return *this; return *this;
} }
actor_ostream aout(scoped_actor& self) {
return actor_ostream{self.get()};
}
} // namespace cppa } // namespace cppa
namespace std { namespace std {
......
...@@ -78,7 +78,8 @@ behavior default_broker::make_behavior() { ...@@ -78,7 +78,8 @@ behavior default_broker::make_behavior() {
return ( return (
on(atom("INITMSG")) >> [=] { on(atom("INITMSG")) >> [=] {
unbecome(); unbecome();
m_fun(this); auto bhvr = m_fun(this);
if (bhvr) become(std::move(bhvr));
} }
); );
} }
...@@ -468,16 +469,30 @@ broker_ptr init_and_launch(broker_ptr ptr) { ...@@ -468,16 +469,30 @@ broker_ptr init_and_launch(broker_ptr ptr) {
return ptr; return ptr;
} }
broker_ptr broker::from_impl(std::function<void (broker*)> fun, broker_ptr broker::from_impl(std::function<behavior (broker*)> fun,
input_stream_ptr in, input_stream_ptr in,
output_stream_ptr out) { output_stream_ptr out) {
return make_counted<default_broker>(move(fun), move(in), move(out)); return make_counted<default_broker>(move(fun), move(in), move(out));
} }
broker_ptr broker::from(std::function<void (broker*)> fun, acceptor_uptr in) {
broker_ptr broker::from_impl(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out) {
auto f = [=](broker* ptr) -> behavior { fun(ptr); return behavior{}; };
return make_counted<default_broker>(f, move(in), move(out));
}
broker_ptr broker::from(std::function<behavior (broker*)> fun, acceptor_uptr in) {
return make_counted<default_broker>(move(fun), move(in)); return make_counted<default_broker>(move(fun), move(in));
} }
broker_ptr broker::from(std::function<void (broker*)> fun, acceptor_uptr in) {
auto f = [=](broker* ptr) -> behavior { fun(ptr); return behavior{}; };
return make_counted<default_broker>(f, move(in));
}
void broker::erase_io(int id) { void broker::erase_io(int id) {
m_io.erase(connection_handle::from_int(id)); m_io.erase(connection_handle::from_int(id));
} }
...@@ -509,7 +524,8 @@ actor broker::fork_impl(std::function<void (broker*)> fun, ...@@ -509,7 +524,8 @@ actor broker::fork_impl(std::function<void (broker*)> fun,
throw std::invalid_argument("invalid handle"); throw std::invalid_argument("invalid handle");
} }
scribe* sptr = i->second.get(); // non-owning pointer scribe* sptr = i->second.get(); // non-owning pointer
auto result = make_counted<default_broker>(move(fun), move(i->second)); auto f = [=](broker* ptr) -> behavior { fun(ptr); return behavior{}; };
auto result = make_counted<default_broker>(f, move(i->second));
init_and_launch(result); init_and_launch(result);
sptr->set_broker(result); // set new broker sptr->set_broker(result); // set new broker
m_io.erase(i); m_io.erase(i);
......
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