Commit 0e8eabbd authored by Dominik Charousset's avatar Dominik Charousset

streamlined broker implementation

parent dd5c4195
...@@ -99,17 +99,15 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -99,17 +99,15 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
void write(const connection_handle& hdl, util::buffer&& buf); void write(const connection_handle& hdl, util::buffer&& buf);
static broker_ptr from(std::function<void (broker*, connection_handle)> fun, template<typename F, typename... Ts>
static broker_ptr from(F fun,
input_stream_ptr in, input_stream_ptr in,
output_stream_ptr out); output_stream_ptr out,
Ts&&... args) {
template<typename F, typename T0, typename... Ts> auto hdl = connection_handle::from_int(in->read_handle());
static broker_ptr from(F fun, input_stream_ptr in, output_stream_ptr out, return from_impl(std::bind(std::move(fun),
T0&& arg0, Ts&&... args) {
return from(std::bind(std::move(fun),
std::placeholders::_1, std::placeholders::_1,
std::placeholders::_2, hdl,
detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...), detail::fwd<Ts>(args)...),
std::move(in), std::move(in),
std::move(out)); std::move(out));
...@@ -126,18 +124,13 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -126,18 +124,13 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
std::move(in)); std::move(in));
} }
actor_ptr fork(std::function<void (broker*, connection_handle)> fun, template<typename F, typename... Ts>
connection_handle hdl);
template<typename F, typename T0, typename... Ts>
actor_ptr fork(F fun, actor_ptr fork(F fun,
connection_handle hdl, connection_handle hdl,
T0&& arg0,
Ts&&... args) { Ts&&... args) {
return this->fork(std::bind(std::move(fun), return this->fork_impl(std::bind(std::move(fun),
std::placeholders::_1, std::placeholders::_1,
std::placeholders::_2, hdl,
detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...), detail::fwd<Ts>(args)...),
hdl); hdl);
} }
...@@ -167,6 +160,13 @@ class broker : public extend<local_actor>::with<threadless, stackless> { ...@@ -167,6 +160,13 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
private: private:
actor_ptr fork_impl(std::function<void (broker*)> fun,
connection_handle hdl);
static broker_ptr from_impl(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out);
void invoke_message(const message_header& hdr, any_tuple msg); void invoke_message(const message_header& hdr, any_tuple msg);
void erase_io(int id); void erase_io(int id);
......
...@@ -63,21 +63,10 @@ class default_broker : public broker { ...@@ -63,21 +63,10 @@ class default_broker : public broker {
typedef std::function<void (broker*)> function_type; typedef std::function<void (broker*)> function_type;
struct fork_flag { };
template<typename... Ts> template<typename... Ts>
default_broker(function_type&& fun, Ts&&... args) default_broker(function_type&& fun, Ts&&... args)
: broker{std::forward<Ts>(args)...}, m_fun{move(fun)} { } : broker{std::forward<Ts>(args)...}, m_fun{move(fun)} { }
template<typename... Ts>
default_broker(fork_flag,
std::function<void (broker*, connection_handle)>&& fun,
connection_handle hdl,
Ts&&... args)
: broker{std::forward<Ts>(args)...}
, m_fun{std::bind(move(fun), std::placeholders::_1, hdl)} { }
void init() override { void init() override {
enqueue(nullptr, make_any_tuple(atom("INITMSG"))); enqueue(nullptr, make_any_tuple(atom("INITMSG")));
become( become(
...@@ -265,7 +254,6 @@ class broker::doorman : public broker::servant { ...@@ -265,7 +254,6 @@ class broker::doorman : public broker::servant {
} }
continue_reading_result continue_reading() override { continue_reading_result continue_reading() override {
CPPA_REQUIRE(parent() != nullptr);
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
for (;;) { for (;;) {
optional<stream_ptr_pair> opt{none}; optional<stream_ptr_pair> opt{none};
...@@ -437,15 +425,10 @@ local_actor_ptr init_and_launch(broker_ptr ptr) { ...@@ -437,15 +425,10 @@ local_actor_ptr init_and_launch(broker_ptr ptr) {
return move(ptr); return move(ptr);
} }
broker_ptr broker::from(std::function<void (broker*, connection_handle)> fun, broker_ptr broker::from_impl(std::function<void (broker*)> fun,
input_stream_ptr in, input_stream_ptr in,
output_stream_ptr out) { output_stream_ptr out) {
auto hdl = connection_handle::from_int(in->read_handle()); return make_counted<default_broker>(move(fun), move(in), move(out));
return make_counted<default_broker>(std::bind(move(fun),
std::placeholders::_1,
hdl),
move(in),
move(out));
} }
broker_ptr broker::from(std::function<void (broker*)> fun, acceptor_uptr in) { broker_ptr broker::from(std::function<void (broker*)> fun, acceptor_uptr in) {
...@@ -475,15 +458,12 @@ accept_handle broker::add_doorman(acceptor_uptr ptr) { ...@@ -475,15 +458,12 @@ accept_handle broker::add_doorman(acceptor_uptr ptr) {
return id; return id;
} }
actor_ptr broker::fork(std::function<void (broker*, connection_handle)> fun, actor_ptr broker::fork_impl(std::function<void (broker*)> fun,
connection_handle hdl) { connection_handle hdl) {
auto i = m_io.find(hdl); auto i = m_io.find(hdl);
if (i == m_io.end()) throw std::invalid_argument("invalid handle"); if (i == m_io.end()) 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>(default_broker::fork_flag{}, auto result = make_counted<default_broker>(move(fun), move(i->second));
move(fun),
hdl,
move(i->second));
init_and_launch(result); init_and_launch(result);
sptr->set_parent(result); // set new parent sptr->set_parent(result); // set new parent
m_io.erase(i); m_io.erase(i);
......
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
#include <memory> #include <memory>
#include<iostream> #include <iostream>
#include "test.hpp" #include "test.hpp"
#include "cppa/cppa.hpp" #include "cppa/cppa.hpp"
...@@ -39,11 +39,6 @@ using namespace cppa; ...@@ -39,11 +39,6 @@ using namespace cppa;
namespace { constexpr size_t message_size = sizeof(atom_value) + sizeof(int); } namespace { constexpr size_t message_size = sizeof(atom_value) + sizeof(int); }
void report_unexpected() {
std::cerr << "unexpected message: " << to_string(self->last_dequeued())
<< std::endl;
}
void ping(size_t num_pings) { void ping(size_t num_pings) {
auto count = std::make_shared<size_t>(0); auto count = std::make_shared<size_t>(0);
become ( become (
...@@ -54,10 +49,10 @@ void ping(size_t num_pings) { ...@@ -54,10 +49,10 @@ void ping(size_t num_pings) {
if (++*count >= num_pings) self->quit(); if (++*count >= num_pings) self->quit();
else reply(atom("ping"), value + 1); else reply(atom("ping"), value + 1);
}, },
others() >> report_unexpected others() >> CPPA_UNEXPECTED_MSG_CB()
); );
}, },
others() >> report_unexpected others() >> CPPA_UNEXPECTED_MSG_CB()
); );
} }
...@@ -73,10 +68,10 @@ void pong() { ...@@ -73,10 +68,10 @@ void pong() {
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) { on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
self->quit(reason); self->quit(reason);
}, },
others() >> report_unexpected others() >> CPPA_UNEXPECTED_MSG_CB()
); );
}, },
others() >> report_unexpected others() >> CPPA_UNEXPECTED_MSG_CB()
); );
} }
...@@ -86,14 +81,9 @@ void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy ...@@ -86,14 +81,9 @@ void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy
cout << "num_connections() != 1" << endl; cout << "num_connections() != 1" << endl;
throw std::logic_error("num_connections() != 1"); throw std::logic_error("num_connections() != 1");
} }
//thisptr->for_each_connection([=](io::connection_handle hdl) {
// thisptr->receive_policy(hdl, io::broker::exactly, message_size);
//});
auto write = [=](atom_value type, int value) { auto write = [=](atom_value type, int value) {
//thisptr->for_each_connection([=](io::connection_handle hdl) {
thisptr->write(hdl, sizeof(type), &type); thisptr->write(hdl, sizeof(type), &type);
thisptr->write(hdl, sizeof(value), &value); thisptr->write(hdl, sizeof(value), &value);
//});
}; };
become ( become (
on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) { on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) {
...@@ -115,7 +105,7 @@ void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy ...@@ -115,7 +105,7 @@ void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) { on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
if (thisptr->last_sender() == buddy) self->quit(reason); if (thisptr->last_sender() == buddy) self->quit(reason);
}, },
others() >> report_unexpected others() >> CPPA_UNEXPECTED_MSG_CB()
); );
} }
...@@ -127,7 +117,7 @@ void peer_acceptor(io::broker* thisptr, const actor_ptr& buddy) { ...@@ -127,7 +117,7 @@ void peer_acceptor(io::broker* thisptr, const actor_ptr& buddy) {
thisptr->fork(peer, hdl, buddy); thisptr->fork(peer, hdl, buddy);
self->quit(); self->quit();
}, },
others() >> report_unexpected others() >> CPPA_UNEXPECTED_MSG_CB()
); );
} }
......
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