Commit dcf5127b authored by Jakob Otto's avatar Jakob Otto

Add configurable base-ptr to newb events

parent fa755f75
...@@ -109,6 +109,10 @@ struct newb_base : public extend<scheduled_actor, newb_base>::template ...@@ -109,6 +109,10 @@ struct newb_base : public extend<scheduled_actor, newb_base>::template
virtual void stop_writing() = 0; virtual void stop_writing() = 0;
virtual void read_event(newb_base* = nullptr) = 0;
virtual void write_event(newb_base* = nullptr) = 0;
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
void init_newb(); void init_newb();
......
...@@ -127,10 +127,10 @@ struct newb : public network::newb_base { ...@@ -127,10 +127,10 @@ struct newb : public network::newb_base {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
switch (op) { switch (op) {
case io::network::operation::read: case io::network::operation::read:
read_event(); read_event(this);
break; break;
case io::network::operation::write: case io::network::operation::write:
write_event(); write_event(this);
break; break;
case io::network::operation::propagate_error: case io::network::operation::propagate_error:
handle_error(); handle_error();
...@@ -256,14 +256,14 @@ struct newb : public network::newb_base { ...@@ -256,14 +256,14 @@ struct newb : public network::newb_base {
trans->flush(this); trans->flush(this);
} }
void read_event() { void read_event(network::newb_base* base) override {
auto err = trans->read_some(this, *proto); auto err = trans->read_some(base, *proto);
if (err) if (err)
io_error(network::operation::read, std::move(err)); io_error(network::operation::read, std::move(err));
} }
void write_event() { void write_event(network::newb_base* base) override {
if (trans->write_some(this) == network::rw_state::failure) if (trans->write_some(base) == network::rw_state::failure)
io_error(network::operation::write, sec::runtime_error); io_error(network::operation::write, sec::runtime_error);
} }
...@@ -457,11 +457,11 @@ struct newb_acceptor : network::acceptor_base { ...@@ -457,11 +457,11 @@ struct newb_acceptor : network::acceptor_base {
CAF_LOG_DEBUG("new event: " << to_string(op)); CAF_LOG_DEBUG("new event: " << to_string(op));
switch (op) { switch (op) {
case network::operation::read: case network::operation::read:
read_event(); read_event(this);
break; break;
case network::operation::write: case network::operation::write:
// Required to multiplex over a single socket. // Required to multiplex over a single socket.
write_event(); write_event(this);
break; break;
case network::operation::propagate_error: case network::operation::propagate_error:
CAF_LOG_DEBUG("acceptor got error operation"); CAF_LOG_DEBUG("acceptor got error operation");
...@@ -549,7 +549,7 @@ struct newb_acceptor : network::acceptor_base { ...@@ -549,7 +549,7 @@ struct newb_acceptor : network::acceptor_base {
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
void read_event() { void read_event(network::newb_base*) override {
if (accept_pol->manual_read) { if (accept_pol->manual_read) {
accept_pol->read_event(this); accept_pol->read_event(this);
} else { } else {
...@@ -575,7 +575,7 @@ struct newb_acceptor : network::acceptor_base { ...@@ -575,7 +575,7 @@ struct newb_acceptor : network::acceptor_base {
} }
} }
void write_event() { void write_event(network::newb_base* base) override {
accept_pol->write_event(this); accept_pol->write_event(this);
} }
......
...@@ -325,7 +325,7 @@ CAF_TEST(read event) { ...@@ -325,7 +325,7 @@ CAF_TEST(read event) {
write_packet(buf, ohdr, bhdr, payload); write_packet(buf, ohdr, bhdr, payload);
dummy.trans->received_bytes = buf.size(); dummy.trans->received_bytes = buf.size();
CAF_MESSAGE("trigger a read event"); CAF_MESSAGE("trigger a read event");
dummy.read_event(); dummy.read_event(&dummy);
CAF_MESSAGE("check the basp header and payload"); CAF_MESSAGE("check the basp header and payload");
CAF_REQUIRE(!dummy.state.messages.empty()); CAF_REQUIRE(!dummy.state.messages.empty());
auto& msg = dummy.state.messages.front().first; auto& msg = dummy.state.messages.front().first;
...@@ -370,7 +370,7 @@ CAF_TEST(timeouts) { ...@@ -370,7 +370,7 @@ CAF_TEST(timeouts) {
exec_all(); exec_all();
CAF_MESSAGE("trigger read event"); CAF_MESSAGE("trigger read event");
auto& dummy = deref<newb_t>(test_newb); auto& dummy = deref<newb_t>(test_newb);
dummy.read_event(); dummy.read_event(&dummy);
CAF_CHECK(!dummy.state.expected.empty()); CAF_CHECK(!dummy.state.expected.empty());
CAF_MESSAGE("trigger waiting timeouts"); CAF_MESSAGE("trigger waiting timeouts");
// Trigger timeout. // Trigger timeout.
...@@ -400,12 +400,12 @@ CAF_TEST(message ordering) { ...@@ -400,12 +400,12 @@ CAF_TEST(message ordering) {
CAF_MESSAGE("read second message first"); CAF_MESSAGE("read second message first");
write_packet(buf, ohdr_second, bhdr_second, payload_second); write_packet(buf, ohdr_second, bhdr_second, payload_second);
dummy.trans->received_bytes = buf.size(); dummy.trans->received_bytes = buf.size();
dummy.read_event(); dummy.read_event(&dummy);
CAF_MESSAGE("followed by first message"); CAF_MESSAGE("followed by first message");
buf.clear(); buf.clear();
write_packet(buf, ohdr_first, bhdr_first, payload_first); write_packet(buf, ohdr_first, bhdr_first, payload_first);
dummy.trans->received_bytes = buf.size(); dummy.trans->received_bytes = buf.size();
dummy.read_event(); dummy.read_event(&dummy);
} }
CAF_TEST(write buf) { CAF_TEST(write buf) {
......
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