Commit c151505b authored by Joseph Noir's avatar Joseph Noir

Adapt to timeout and message handler changes

parent f4f473d7
...@@ -131,22 +131,15 @@ struct basp_newb : public io::network::newb<new_basp_message> { ...@@ -131,22 +131,15 @@ struct basp_newb : public io::network::newb<new_basp_message> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
} }
void handle(new_basp_message& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
std::string res;
binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}
behavior make_behavior() override { behavior make_behavior() override {
set_default_handler(print_and_drop); set_default_handler(print_and_drop);
return { return {
// Must be implemented at the moment, will be cought by the broker in a [=](new_basp_message& msg) {
// later implementation. CAF_LOG_TRACE("");
[=](atom_value atm, uint32_t id) { std::string res;
protocol->timeout(atm, id); binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}, },
[=](send_atom, actor_id sender, actor_id receiver, std::string payload) { [=](send_atom, actor_id sender, actor_id receiver, std::string payload) {
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error { auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
......
...@@ -245,22 +245,15 @@ struct basp_newb : public io::network::newb<new_basp_message> { ...@@ -245,22 +245,15 @@ struct basp_newb : public io::network::newb<new_basp_message> {
// nop // nop
} }
void handle(new_basp_message& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
std::string res;
binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}
behavior make_behavior() override { behavior make_behavior() override {
set_default_handler(print_and_drop); set_default_handler(print_and_drop);
return { return {
// Must be implemented at the moment, will be cought by the broker in a [=](new_basp_message& msg) {
// later implementation. CAF_LOG_TRACE("");
[=](atom_value atm, uint32_t id) { std::string res;
protocol->timeout(atm, id); binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}, },
[=](send_atom, actor_id sender, actor_id receiver, std::string payload) { [=](send_atom, actor_id sender, actor_id receiver, std::string payload) {
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error { auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
......
...@@ -79,20 +79,13 @@ struct raw_newb : public io::network::newb<new_data> { ...@@ -79,20 +79,13 @@ struct raw_newb : public io::network::newb<new_data> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
} }
void handle(message_type& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
char res = *msg.payload;
send(responder, res);
}
behavior make_behavior() override { behavior make_behavior() override {
set_default_handler(print_and_drop); set_default_handler(print_and_drop);
return { return {
// Must be implemented at the moment, will be cought by the broker in a [=](message_type& msg) {
// later implementation. CAF_LOG_TRACE("");
[=](atom_value atm, uint32_t id) { char res = *msg.payload;
protocol->timeout(atm, id); send(responder, res);
}, },
[=](send_atom, char c) { [=](send_atom, char c) {
auto whdl = wr_buf(nullptr); auto whdl = wr_buf(nullptr);
......
...@@ -77,20 +77,13 @@ struct raw_newb : public io::network::newb<new_data> { ...@@ -77,20 +77,13 @@ struct raw_newb : public io::network::newb<new_data> {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
} }
void handle(message_type& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
char res = *msg.payload;
send(responder, res);
}
behavior make_behavior() override { behavior make_behavior() override {
set_default_handler(print_and_drop); set_default_handler(print_and_drop);
return { return {
// Must be implemented at the moment, will be cought by the broker in a [=](message_type& msg) {
// later implementation. CAF_LOG_TRACE("");
[=](atom_value atm, uint32_t id) { char res = *msg.payload;
protocol->timeout(atm, id); send(responder, res);
}, },
[=](send_atom, char c) { [=](send_atom, char c) {
auto whdl = wr_buf(nullptr); auto whdl = wr_buf(nullptr);
......
...@@ -242,32 +242,24 @@ struct dummy_basp_newb : network::newb<new_basp_message> { ...@@ -242,32 +242,24 @@ struct dummy_basp_newb : network::newb<new_basp_message> {
// nop // nop
} }
void handle(new_basp_message& msg) override {
CAF_MESSAGE("handling new basp message = " << to_string(msg));
CAF_ASSERT(!expected.empty());
auto& e = expected.front();
CAF_CHECK_EQUAL(msg.header.from, e.first.from);
CAF_CHECK_EQUAL(msg.header.to, e.first.to);
int pl;
binary_deserializer bd(&backend(), msg.payload, msg.payload_size);
bd(pl);
CAF_CHECK_EQUAL(pl, e.second);
std::vector<char> payload{msg.payload, msg.payload + msg.payload_size};
messages.emplace_back(msg, payload);
messages.back().first.payload = messages.back().second.data();
transport->receive_buffer.clear();
expected.pop_front();
}
behavior make_behavior() override { behavior make_behavior() override {
set_default_handler(print_and_drop); set_default_handler(print_and_drop);
return { return {
// Must be implemented at the moment, will be cought by the broker in a [=](new_basp_message& msg) {
// later implementation. CAF_MESSAGE("handling new basp message = " << to_string(msg));
[=](atom_value atm, uint32_t id) { CAF_ASSERT(!expected.empty());
CAF_MESSAGE("timeout returned"); auto& e = expected.front();
timeout_messages.emplace_back(atm, id); CAF_CHECK_EQUAL(msg.header.from, e.first.from);
protocol->timeout(atm, id); CAF_CHECK_EQUAL(msg.header.to, e.first.to);
int pl;
binary_deserializer bd(&backend(), msg.payload, msg.payload_size);
bd(pl);
CAF_CHECK_EQUAL(pl, e.second);
std::vector<char> payload{msg.payload, msg.payload + msg.payload_size};
messages.emplace_back(msg, payload);
messages.back().first.payload = messages.back().second.data();
transport->receive_buffer.clear();
expected.pop_front();
}, },
[=](send_atom, actor_id sender, actor_id receiver, int payload) { [=](send_atom, actor_id sender, actor_id receiver, int payload) {
CAF_MESSAGE("send: from = " << sender << " to = " << receiver CAF_MESSAGE("send: from = " << sender << " to = " << receiver
...@@ -485,12 +477,14 @@ CAF_TEST(timeouts) { ...@@ -485,12 +477,14 @@ CAF_TEST(timeouts) {
dummy.transport->max_consecutive_reads = 1; dummy.transport->max_consecutive_reads = 1;
auto err = dummy.read_event(); auto err = dummy.read_event();
CAF_REQUIRE(!err); CAF_REQUIRE(!err);
CAF_CHECK(!dummy.expected.empty());
CAF_MESSAGE("trigger waiting timeouts"); CAF_MESSAGE("trigger waiting timeouts");
// Trigger timeout. // Trigger timeout.
sched.trigger_timeout(); sched.trigger_timeout();
// Handle received message. // Handle received message.
exec_all(); exec_all();
// Message handler will check if the expected message was received. // Message handler will check if the expected message was received.
CAF_CHECK(dummy.expected.empty());
} }
CAF_TEST(message ordering) { CAF_TEST(message ordering) {
......
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