Commit 69c027dc authored by Joseph Noir's avatar Joseph Noir

Let acceptor decide if children start reading

parent 8c04c830
...@@ -172,9 +172,6 @@ struct newb : public network::newb_base { ...@@ -172,9 +172,6 @@ struct newb : public network::newb_base {
// as an event handler. // as an event handler.
if (!reading_ && !writing_) if (!reading_ && !writing_)
intrusive_ptr_add_ref(this->ctrl()); intrusive_ptr_add_ref(this->ctrl());
start_reading();
if (trans)
trans->prepare_next_read(this);
} }
void stop() override { void stop() override {
...@@ -219,6 +216,8 @@ struct newb : public network::newb_base { ...@@ -219,6 +216,8 @@ struct newb : public network::newb_base {
if (!reading_) { if (!reading_) {
event_handler::activate(); event_handler::activate();
reading_ = true; reading_ = true;
if (trans)
trans->prepare_next_read(this);
} }
} }
...@@ -379,7 +378,7 @@ template <class Protocol, spawn_options Os = no_spawn_options, class F, ...@@ -379,7 +378,7 @@ template <class Protocol, spawn_options Os = no_spawn_options, class F,
class... Ts> class... Ts>
typename infer_handle_from_fun<F>::type typename infer_handle_from_fun<F>::type
spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport, spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport,
network::native_socket sockfd, Ts&&... xs) { network::native_socket sockfd, bool start_reading, Ts&&... xs) {
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;
...@@ -400,6 +399,8 @@ spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport, ...@@ -400,6 +399,8 @@ spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport,
auto& ref = dynamic_cast<newb<message>&>(*ptr); auto& ref = dynamic_cast<newb<message>&>(*ptr);
// Start the event handler. // Start the event handler.
ref.start(); ref.start();
if (start_reading)
ref.start_reading();
return res; return res;
} }
...@@ -411,7 +412,7 @@ spawn_client(actor_system& sys, F fun, policy::transport_ptr transport, ...@@ -411,7 +412,7 @@ spawn_client(actor_system& sys, F fun, policy::transport_ptr transport,
expected<network::native_socket> esock = transport->connect(host, port); expected<network::native_socket> esock = transport->connect(host, port);
if (!esock) if (!esock)
return std::move(esock.error()); return std::move(esock.error());
return spawn_newb<Protocol>(sys, fun, std::move(transport), *esock, return spawn_newb<Protocol>(sys, fun, std::move(transport), *esock, true,
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
} }
...@@ -581,7 +582,8 @@ struct newb_acceptor : network::newb_base { ...@@ -581,7 +582,8 @@ struct newb_acceptor : network::newb_base {
io::spawn_newb<Protocol, no_spawn_options, Fun, Ts...>, io::spawn_newb<Protocol, no_spawn_options, Fun, Ts...>,
detail::get_indices(args_), detail::get_indices(args_),
args_, this->backend().system(), args_, this->backend().system(),
fun_, std::move(pol), sockfd fun_, std::move(pol), sockfd,
accept_pol->add_children_to_loop()
); );
link_to(n); link_to(n);
children_.push_back(n); children_.push_back(n);
......
...@@ -46,6 +46,10 @@ struct accept { ...@@ -46,6 +46,10 @@ struct accept {
// nop // nop
} }
virtual bool add_children_to_loop() {
return true;
}
virtual expected<io::network::native_socket> virtual expected<io::network::native_socket>
create_socket(uint16_t port, const char* host, bool reuse = false) = 0; create_socket(uint16_t port, const char* host, bool reuse = false) = 0;
......
...@@ -177,7 +177,7 @@ struct fixture { ...@@ -177,7 +177,7 @@ struct fixture {
// Create newb. // Create newb.
transport_ptr transport{new dummy_transport}; transport_ptr transport{new dummy_transport};
test_newb = spawn_newb<protocol_t>(sys, dummy_broker, std::move(transport), test_newb = spawn_newb<protocol_t>(sys, dummy_broker, std::move(transport),
esock->first); esock->first, true);
} }
~fixture() { ~fixture() {
...@@ -258,7 +258,7 @@ CAF_TEST(spawn newb) { ...@@ -258,7 +258,7 @@ CAF_TEST(spawn newb) {
CAF_REQUIRE(esock); CAF_REQUIRE(esock);
transport_ptr transport{new dummy_transport}; transport_ptr transport{new dummy_transport};
auto n = spawn_newb<protocol_t>(sys, my_newb, std::move(transport), auto n = spawn_newb<protocol_t>(sys, my_newb, std::move(transport),
esock->first); esock->first, true);
exec_all(); exec_all();
CAF_MESSAGE("send test message"); CAF_MESSAGE("send test message");
self->send(n, 3); self->send(n, 3);
...@@ -288,7 +288,7 @@ CAF_TEST(spawn stateful newb) { ...@@ -288,7 +288,7 @@ CAF_TEST(spawn stateful newb) {
CAF_REQUIRE(esock); CAF_REQUIRE(esock);
transport_ptr transport{new dummy_transport}; transport_ptr transport{new dummy_transport};
auto n = spawn_newb<protocol_t>(sys, my_newb, std::move(transport), auto n = spawn_newb<protocol_t>(sys, my_newb, std::move(transport),
esock->first); esock->first, true);
exec_all(); exec_all();
CAF_MESSAGE("set value in state"); CAF_MESSAGE("set value in state");
self->send(n, set_atom::value, 3); self->send(n, set_atom::value, 3);
......
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