Commit 0900b3d1 authored by Joseph Noir's avatar Joseph Noir

Adapt dgram handling to one BASP message per dgram

parent 30cd1164
...@@ -864,7 +864,7 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, ...@@ -864,7 +864,7 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self,
} }
expected<dgram_scribe_handle> expected<dgram_scribe_handle>
default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd, default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
const std::string& host, uint16_t port) { const std::string& host, uint16_t port) {
auto res = interfaces::native_address(host, none, port); auto res = interfaces::native_address(host, none, port);
if (!res) { if (!res) {
......
...@@ -120,7 +120,7 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -120,7 +120,7 @@ connection_state instance::handle(execution_unit* ctx,
bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
using itr_t = std::vector<char>::iterator; using itr_t = std::vector<char>::iterator;
// Call in case of an error // function object providing cleanup code on errors
auto err = [&]() -> bool { auto err = [&]() -> bool {
auto cb = make_callback([&](const node_id& nid) -> error { auto cb = make_callback([&](const node_id& nid) -> error {
callee_.purge_state(nid); callee_.purge_state(nid);
...@@ -129,43 +129,31 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -129,43 +129,31 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
tbl_.erase(dm.handle, cb); tbl_.erase(dm.handle, cb);
return false; return false;
}; };
// Split message into hdr and payload // extract payload
// Loop until all messages in the datagram have been processed. std::vector<char> pl_buf{std::move_iterator<itr_t>(std::begin(dm.buf) +
auto dgram_itr = dm.buf.begin(); basp::header_size),
do { std::move_iterator<itr_t>(std::end(dm.buf))};
// extract header // resize header
std::vector<char> hdr_buf{std::move_iterator<itr_t>(dgram_itr), dm.buf.resize(basp::header_size);
std::move_iterator<itr_t>(dgram_itr + // extract header
basp::header_size)}; binary_deserializer bd{ctx, dm.buf};
dgram_itr += basp::header_size; auto e = bd(hdr);
// deserialize header if (e || !valid(hdr)) {
binary_deserializer bd{ctx, hdr_buf}; CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
auto e = bd(hdr); std::cerr << "Received invalid header!" << std::endl;
if (e || !valid(hdr)) { return err();
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr)); }
std::cerr << "[I] Received invalid header of type: " CAF_LOG_DEBUG(CAF_ARG(hdr));
<< to_string(hdr.operation) << std::endl; std::vector<char>* payload = nullptr;
return err(); if (hdr.payload_len > 0) {
} payload = &pl_buf;
CAF_LOG_DEBUG(CAF_ARG(hdr)); if (payload->size() != hdr.payload_len) {
// extract payload CAF_LOG_WARNING("received invalid payload");
std::vector<char> pl_buf{std::move_iterator<itr_t>(dgram_itr),
std::move_iterator<itr_t>(dgram_itr +
hdr.payload_len)};
dgram_itr += hdr.payload_len;
std::vector<char>* payload = nullptr;
if (hdr.payload_len > 0) {
payload = &pl_buf;
}
// needs forwarding?
if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
std::cerr << "[I] Needs forwarding?" << std::endl;
// TODO: anything to do here? (i.e., do we still need forwarding?)
return err(); return err();
} }
if (!handle_msg(ctx, dm.handle, hdr, payload, false, none)) }
return err(); if (!handle_msg(ctx, dm.handle, hdr, payload, false, none))
} while (dgram_itr != dm.buf.end()); return err();
return true; return true;
}; };
......
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