Commit 98dd5641 authored by Joseph Noir's avatar Joseph Noir

Implement consume for length prefix framing

parent 48cb551e
...@@ -24,7 +24,10 @@ ...@@ -24,7 +24,10 @@
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/network_order.hpp" #include "caf/detail/network_order.hpp"
#include "caf/error.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/tag/message_oriented.hpp" #include "caf/tag/message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp" #include "caf/tag/stream_oriented.hpp"
...@@ -38,13 +41,15 @@ namespace caf::net { ...@@ -38,13 +41,15 @@ namespace caf::net {
template <class UpperLayer> template <class UpperLayer>
class length_prefix_framing { class length_prefix_framing {
public: public:
using byte_span = span<const byte>;
using input_tag = tag::stream_oriented; using input_tag = tag::stream_oriented;
using output_tag = tag::message_oriented; using output_tag = tag::message_oriented;
using length_prefix_type = uint32_t; using length_prefix_type = uint32_t;
constexpr size_t max_message_length = INT32_MAX; static constexpr size_t max_message_length = INT32_MAX;
// -- interface for the upper layer ------------------------------------------ // -- interface for the upper layer ------------------------------------------
...@@ -52,7 +57,7 @@ public: ...@@ -52,7 +57,7 @@ public:
class access { class access {
public: public:
access(LowerLayer* lower_layer, length_prefix_framing* this_layer) access(LowerLayer* lower_layer, length_prefix_framing* this_layer)
: lower_layer_(lower_layer), this_layer(this_layer) { : lower_layer_(lower_layer), this_layer_(this_layer) {
// nop // nop
} }
...@@ -64,10 +69,10 @@ public: ...@@ -64,10 +69,10 @@ public:
} }
byte_buffer& message_buffer() { byte_buffer& message_buffer() {
return lower_layer_->output_buffer.size(); return lower_layer_->output_buffer();
} }
void end_message() { bool end_message() {
using detail::to_network_order; using detail::to_network_order;
auto& buf = message_buffer(); auto& buf = message_buffer();
auto msg_begin = buf.begin() + message_offset_; auto msg_begin = buf.begin() + message_offset_;
...@@ -93,11 +98,7 @@ public: ...@@ -93,11 +98,7 @@ public:
} }
void configure_read(receive_policy policy) { void configure_read(receive_policy policy) {
if (policy.max_size > 0 && transport_->max_read_size_ == 0) lower_layer_->configure_read(policy);
parent_->register_reading();
transport_->min_read_size_ = policy.min_size;
transport_->max_read_size_ = policy.max_size;
transport_->read_buf_.resize(policy.max_size);
} }
private: private:
...@@ -136,7 +137,17 @@ public: ...@@ -136,7 +137,17 @@ public:
} }
template <class LowerLayer> template <class LowerLayer>
ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span delta) { ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span) {
using detail::from_network_order;
if (buffer.size() < 4)
return 0;
auto u32_size = 0;
memcpy(&u32_size, buffer.data(), 4);
auto msg_size = static_cast<size_t>(from_network_order(u32_size));
if (buffer.size() < msg_size + 4)
return 0;
upper_layer_.consume(down, make_span(buffer.data() + 4, msg_size));
return msg_size + 4;
} }
private: private:
......
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