Commit de01b1e6 authored by Joseph Noir's avatar Joseph Noir

Enable arguments for acceptor childs

parent 69c027dc
...@@ -370,8 +370,8 @@ struct function_traits<R (T::*)(Args...) const> { ...@@ -370,8 +370,8 @@ struct function_traits<R (T::*)(Args...) const> {
}; };
template<class T> template<class T>
using first_argument_type using first_argument_type =
= typename std::tuple_element<0, typename function_traits<T>::argument_types>::type; typename std::tuple_element<0, typename function_traits<T>::argument_types>::type;
/// Spawns a new "newb" broker. /// Spawns a new "newb" broker.
template <class Protocol, spawn_options Os = no_spawn_options, class F, template <class Protocol, spawn_options Os = no_spawn_options, class F,
...@@ -382,7 +382,8 @@ spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport, ...@@ -382,7 +382,8 @@ spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport,
using impl = typename infer_handle_from_fun<F>::impl; using impl = typename infer_handle_from_fun<F>::impl;
using first = first_argument_type<F>; using first = first_argument_type<F>;
using message = typename std::remove_pointer<first>::type::message_type; using message = typename std::remove_pointer<first>::type::message_type;
auto& dm = dynamic_cast<network::default_multiplexer&>(sys.middleman().backend()); auto& dm =
dynamic_cast<network::default_multiplexer&>(sys.middleman().backend());
// Setup the config. // Setup the config.
actor_config cfg(&dm); actor_config cfg(&dm);
detail::init_fun_factory<impl, F> fac; detail::init_fun_factory<impl, F> fac;
...@@ -405,7 +406,8 @@ spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport, ...@@ -405,7 +406,8 @@ spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport,
} }
/// Spawn a new "newb" broker client to connect to `host`:`port`. /// Spawn a new "newb" broker client to connect to `host`:`port`.
template <class Protocol, spawn_options Os = no_spawn_options, class F, class... Ts> template <class Protocol, spawn_options Os = no_spawn_options,
class F, class... Ts>
expected<typename infer_handle_from_fun<F>::type> expected<typename infer_handle_from_fun<F>::type>
spawn_client(actor_system& sys, F fun, policy::transport_ptr transport, spawn_client(actor_system& sys, F fun, policy::transport_ptr transport,
std::string host, uint16_t port, Ts&&... xs) { std::string host, uint16_t port, Ts&&... xs) {
...@@ -427,13 +429,14 @@ struct newb_acceptor : network::newb_base { ...@@ -427,13 +429,14 @@ struct newb_acceptor : network::newb_base {
newb_acceptor(actor_config& cfg, network::default_multiplexer& dm, newb_acceptor(actor_config& cfg, network::default_multiplexer& dm,
network::native_socket sockfd, Fun f, network::native_socket sockfd, Fun f,
policy::accept_ptr<message_type> pol, Ts&&... xs) policy::accept_ptr<message_type> pol,
std::tuple<Ts...> args)
: newb_base(cfg, dm, sockfd), : newb_base(cfg, dm, sockfd),
accept_pol(std::move(pol)), accept_pol(std::move(pol)),
fun_(std::move(f)), fun_(std::move(f)),
reading_(false), reading_(false),
writing_(false), writing_(false),
args_(std::forward<Ts>(xs)...) { args_(std::move(args)) {
// nop // nop
if (sockfd == io::network::invalid_native_socket) if (sockfd == io::network::invalid_native_socket)
CAF_LOG_ERROR("Creating newb with invalid socket"); CAF_LOG_ERROR("Creating newb with invalid socket");
...@@ -643,21 +646,19 @@ spawn_acceptor(actor_system& sys, Fun fun, policy::accept_ptr<Message> pol, ...@@ -643,21 +646,19 @@ spawn_acceptor(actor_system& sys, Fun fun, policy::accept_ptr<Message> pol,
network::native_socket sockfd, Ts&&... xs) { network::native_socket sockfd, Ts&&... xs) {
using first = first_argument_type<Fun>; using first = first_argument_type<Fun>;
using message = typename std::remove_pointer<first>::type::message_type; using message = typename std::remove_pointer<first>::type::message_type;
using impl = newb_acceptor<Protocol, Fun, Ts...>;
static_assert(std::is_same<message, Message>::value, static_assert(std::is_same<message, Message>::value,
"Fun must accept a message type matching the protocol"); "Fun must accept a message type matching the protocol");
// TODO: Can we also check that the signature of Fun ends in Ts...?
using acceptor_type = newb_acceptor<Protocol, Fun, Ts...>;
auto& dm = auto& dm =
dynamic_cast<network::default_multiplexer&>(sys.middleman().backend()); dynamic_cast<network::default_multiplexer&>(sys.middleman().backend());
actor_config cfg(&dm); actor_config cfg(&dm);
auto res = sys.spawn_class<acceptor_type, Os>(cfg, dm, sockfd, auto res = sys.spawn_class<impl, Os>(cfg, dm, sockfd,
std::move(fun), std::move(fun), std::move(pol),
std::move(pol), std::tuple<Ts...>(std::forward<Ts>(xs)...));
std::forward<Ts>(xs)...);
// Get a reference to the newb type. // Get a reference to the newb type.
auto ptr = caf::actor_cast<caf::abstract_actor*>(res); auto ptr = caf::actor_cast<caf::abstract_actor*>(res);
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
auto& ref = dynamic_cast<acceptor_type&>(*ptr); auto& ref = dynamic_cast<impl&>(*ptr);
// Start the event handler. // Start the event handler.
ref.start(); ref.start();
return res; return res;
...@@ -684,45 +685,5 @@ spawn_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol, ...@@ -684,45 +685,5 @@ spawn_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol,
nullptr, false); nullptr, false);
} }
/*
template <class P, class F, class... Ts>
using acceptor_ptr = caf::intrusive_ptr<newb_acceptor<P, F, Ts...>>;
template <class Protocol, class Fun, class Message, class... Ts>
acceptor_ptr<Protocol, Fun, Ts...>
make_acceptor(actor_system& sys, Fun fun, policy::accept_ptr<Message> pol,
network::native_socket sockfd, Ts&&... xs) {
auto& dm =
dynamic_cast<network::default_multiplexer&>(sys.middleman().backend());
auto res = make_counted<newb_acceptor<Protocol, Fun, Ts...>>(dm, sockfd,
std::move(fun),
std::move(pol),
std::forward<Ts>(xs)...);
res->start();
return res;
}
template <class Protocol, class F, class Message, class... Ts>
expected<caf::intrusive_ptr<newb_acceptor<Protocol, F, Ts...>>>
make_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol,
uint16_t port, const char* addr, bool reuse, Ts&&... xs) {
auto esock = pol->create_socket(port, addr, reuse);
if (!esock) {
CAF_LOG_ERROR("Could not open " << CAF_ARG(port) << CAF_ARG(addr));
return sec::cannot_open_port;
}
return make_acceptor<Protocol, F>(sys, std::move(fun), std::move(pol), *esock,
std::forward<Ts>(xs)...);
}
template <class Protocol, class F, class Message>
expected<caf::intrusive_ptr<newb_acceptor<Protocol, F>>>
make_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol,
uint16_t port, const char* addr = nullptr, bool reuse = false) {
return make_server<Protocol, F>(sys, std::move(fun), std::move(pol), port,
addr, reuse);
}
*/
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -68,7 +68,7 @@ constexpr auto host = "127.0.0.1"; ...@@ -68,7 +68,7 @@ constexpr auto host = "127.0.0.1";
// }; // };
// } // }
behavior tcp_server(newb<new_raw_msg>* self) { behavior tcp_server(newb<new_raw_msg>* self, actor responder) {
self->configure_read(io::receive_policy::exactly(sizeof(uint32_t))); self->configure_read(io::receive_policy::exactly(sizeof(uint32_t)));
return { return {
[=](new_raw_msg& msg) { [=](new_raw_msg& msg) {
...@@ -82,11 +82,11 @@ behavior tcp_server(newb<new_raw_msg>* self) { ...@@ -82,11 +82,11 @@ behavior tcp_server(newb<new_raw_msg>* self) {
binary_serializer bs(&self->backend(), *whdl.buf); binary_serializer bs(&self->backend(), *whdl.buf);
bs(data + 1); bs(data + 1);
}, },
[=](io_error_msg& msg) { [=](io_error_msg&) {
//CAF_FAIL("server got io error: " << to_string(msg.op));
CAF_MESSAGE("server: connection lost"); CAF_MESSAGE("server: connection lost");
self->quit(); self->quit();
self->stop(); self->stop();
self->send(responder, shutdown_atom::value);
}, },
[=](caf::exit_msg&) { [=](caf::exit_msg&) {
CAF_MESSAGE("parent shut down, doing the same"); CAF_MESSAGE("parent shut down, doing the same");
...@@ -111,12 +111,9 @@ behavior tcp_client(newb<new_raw_msg>* self, uint32_t value) { ...@@ -111,12 +111,9 @@ behavior tcp_client(newb<new_raw_msg>* self, uint32_t value) {
self->stop(); self->stop();
self->quit(); self->quit();
}, },
[=](io_error_msg& msg) { [=](io_error_msg&) {
// CAF_FAIL("client got io error: " << to_string(msg.op));
// self->stop();
// self->quit();
// self->send(responder, quit_atom::value);
CAF_MESSAGE("client: connection lost"); CAF_MESSAGE("client: connection lost");
// self->send(responder, quit_atom::value);
self->stop(); self->stop();
self->quit(); self->quit();
} }
...@@ -135,7 +132,7 @@ CAF_TEST(newb tcp communication) { ...@@ -135,7 +132,7 @@ CAF_TEST(newb tcp communication) {
accept_ptr<policy::new_raw_msg> pol{new accept_tcp<policy::new_raw_msg>}; accept_ptr<policy::new_raw_msg> pol{new accept_tcp<policy::new_raw_msg>};
auto eserver = io::spawn_server<tcp_protocol<raw>>(system, tcp_server, auto eserver = io::spawn_server<tcp_protocol<raw>>(system, tcp_server,
std::move(pol), 0, std::move(pol), 0,
nullptr, true); nullptr, true, self);
if (!eserver) if (!eserver)
CAF_FAIL("failed to start server: " << system.render(eserver.error())); CAF_FAIL("failed to start server: " << system.render(eserver.error()));
uint16_t port; uint16_t port;
......
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