Commit d0d73a0f authored by Dominik Charousset's avatar Dominik Charousset

Reduce code duplication and add documentation

parent 991874c9
...@@ -28,6 +28,8 @@ caf_add_component( ...@@ -28,6 +28,8 @@ caf_add_component(
src/net/abstract_actor_shell.cpp src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp src/net/actor_shell.cpp
src/net/datagram_socket.cpp src/net/datagram_socket.cpp
src/net/generic_lower_layer.cpp
src/net/generic_upper_layer.cpp
src/net/host.cpp src/net/host.cpp
src/net/http/header.cpp src/net/http/header.cpp
src/net/http/lower_layer.cpp src/net/http/lower_layer.cpp
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
namespace caf::net {
/// Bundles protocol-independent, generic member functions for (almost all)
/// lower layers.
class CAF_NET_EXPORT generic_lower_layer {
public:
virtual ~generic_lower_layer();
/// Queries whether the output device can accept more data straight away.
[[nodiscard]] virtual bool can_send_more() const noexcept = 0;
/// Halt receiving of additional bytes or messages.
virtual void suspend_reading() = 0;
/// Queries whether the lower layer is currently configured to halt receiving
/// of additional bytes or messages.
[[nodiscard]] virtual bool stopped_reading() const noexcept = 0;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net {
/// Bundles protocol-independent, generic member functions for (almost all)
/// upper layers.
class CAF_NET_EXPORT generic_upper_layer {
public:
virtual ~generic_upper_layer();
/// Prepares any pending data for sending.
/// @returns `false` in case of an error to cause the lower layer to stop,
/// `true` otherwise.
[[nodiscard]] virtual bool prepare_send() = 0;
/// Queries whether all pending data has been sent. The lower calls this
/// function to decide whether it has to wait for write events on the socket.
[[nodiscard]] virtual bool done_sending() = 0;
/// Called by the lower layer after some event triggered re-registering the
/// socket manager for read operations after it has been stopped previously
/// by the read policy. May restart consumption of bytes or messages.
virtual void continue_reading() = 0;
/// Called by the lower layer for cleaning up any state in case of an error.
virtual void abort(const error& reason) = 0;
};
} // namespace caf::net
...@@ -4,25 +4,20 @@ ...@@ -4,25 +4,20 @@
#pragma once #pragma once
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/http/fwd.hpp" #include "caf/net/http/fwd.hpp"
namespace caf::net::http { namespace caf::net::http {
/// Parses HTTP requests and passes them to the upper layer. /// Parses HTTP requests and passes them to the upper layer.
class lower_layer { class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public: public:
virtual ~lower_layer(); virtual ~lower_layer();
/// Queries whether the output device can accept more data straight away.
virtual bool can_send_more() const noexcept = 0;
/// Stops reading until manually calling `continue_reading` on the socket
/// manager.
virtual void suspend_reading() = 0;
/// Sends the next header to the client. /// Sends the next header to the client.
virtual bool virtual bool
send_header(context ctx, status code, const header_fields_map& fields) send_header(context ctx, status code, const header_fields_map& fields)
......
...@@ -82,6 +82,8 @@ public: ...@@ -82,6 +82,8 @@ public:
void suspend_reading() override; void suspend_reading() override;
bool stopped_reading() const noexcept override;
bool send_header(context, status code, bool send_header(context, status code,
const header_fields_map& fields) override; const header_fields_map& fields) override;
......
...@@ -7,12 +7,13 @@ ...@@ -7,12 +7,13 @@
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/generic_upper_layer.hpp"
#include "caf/net/http/fwd.hpp" #include "caf/net/http/fwd.hpp"
namespace caf::net::http { namespace caf::net::http {
/// Operates on HTTP requests. /// Operates on HTTP requests.
class upper_layer { class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public: public:
virtual ~upper_layer(); virtual ~upper_layer();
...@@ -26,15 +27,6 @@ public: ...@@ -26,15 +27,6 @@ public:
init(socket_manager* owner, lower_layer* down, const settings& config) init(socket_manager* owner, lower_layer* down, const settings& config)
= 0; = 0;
/// Called by the lower layer for cleaning up any state in case of an error.
virtual void abort(const error& reason) = 0;
/// Called by the lower layer for preparing outgoing data.
virtual bool prepare_send() = 0;
/// Queries whether any pending sends have been completed.
virtual bool done_sending() = 0;
/// Consumes an HTTP message. /// Consumes an HTTP message.
/// @param ctx Identifies this request. The response message must pass this /// @param ctx Identifies this request. The response message must pass this
/// back to the lower layer. Allows clients to send multiple /// back to the lower layer. Allows clients to send multiple
...@@ -48,12 +40,6 @@ public: ...@@ -48,12 +40,6 @@ public:
virtual ptrdiff_t virtual ptrdiff_t
consume(context ctx, const header& hdr, const_byte_span payload) consume(context ctx, const header& hdr, const_byte_span payload)
= 0; = 0;
/// Called by the lower layer after some event triggered re-registering the
/// socket manager for read operations after it has been stopped previously
/// by the read policy. May restart consumption of bytes by setting a new
/// read policy.
virtual void continue_reading() = 0;
}; };
} // namespace caf::net::http } // namespace caf::net::http
...@@ -95,6 +95,8 @@ public: ...@@ -95,6 +95,8 @@ public:
void suspend_reading() override; void suspend_reading() override;
bool stopped_reading() const noexcept override;
void begin_message() override; void begin_message() override;
byte_buffer& message_buffer() override; byte_buffer& message_buffer() override;
......
...@@ -4,21 +4,20 @@ ...@@ -4,21 +4,20 @@
#pragma once #pragma once
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net { namespace caf::net::message_oriented {
/// A contract between two message-oriented layers. The lower layer provides class lower_layer;
/// access to a resource that operates on the granularity of messages, e.g., a
/// UDP socket. The upper layer requests bytes and consumes raw chunks of data.
class message_oriented {
public:
class lower_layer;
class upper_layer { /// Consumes binary messages from the lower layer.
public: class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer(); virtual ~upper_layer();
/// Initializes the upper layer. /// Initializes the upper layer.
...@@ -31,42 +30,23 @@ public: ...@@ -31,42 +30,23 @@ public:
init(socket_manager* owner, lower_layer* down, const settings& config) init(socket_manager* owner, lower_layer* down, const settings& config)
= 0; = 0;
/// Called by the lower layer for cleaning up any state in case of an error.
virtual void abort(const error& reason) = 0;
/// Called by the lower layer for preparing outgoing data.
virtual bool prepare_send() = 0;
/// Queries whether any pending sends have been completed.
virtual bool done_sending() = 0;
/// Consumes bytes from the lower layer. /// Consumes bytes from the lower layer.
/// @param payload Payload of the received message. /// @param payload Payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an /// @returns The number of consumed bytes or a negative value to signal an
/// error. /// error.
/// @note Discarded data is lost permanently. /// @note Discarded data is lost permanently.
virtual ptrdiff_t consume(byte_span payload) = 0; [[nodiscard]] virtual ptrdiff_t consume(byte_span payload) = 0;
};
/// Called by the lower layer after some event triggered re-registering the
/// socket manager for read operations after it has been stopped previously
/// by the read policy. May restart consumption of bytes by setting a new
/// read policy.
virtual void continue_reading() = 0;
};
class lower_layer { /// Provides access to a resource that operates on the granularity of messages,
public: /// e.g., a UDP socket.
class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer(); virtual ~lower_layer();
/// Queries whether the output device can accept more data straight away.
virtual bool can_send_more() const noexcept = 0;
/// Pulls messages from the transport until calling `suspend_reading`. /// Pulls messages from the transport until calling `suspend_reading`.
virtual void request_messages() = 0; virtual void request_messages() = 0;
/// Suspends reading until calling `continue_reading`.
virtual void suspend_reading() = 0;
/// Prepares the layer for an outgoing message, e.g., by allocating an /// Prepares the layer for an outgoing message, e.g., by allocating an
/// output buffer as necessary. /// output buffer as necessary.
virtual void begin_message() = 0; virtual void begin_message() = 0;
...@@ -84,12 +64,15 @@ public: ...@@ -84,12 +64,15 @@ public:
virtual bool end_message() = 0; virtual bool end_message() = 0;
/// Inform the remote endpoint that no more messages will arrive. /// Inform the remote endpoint that no more messages will arrive.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message() = 0; virtual void send_close_message() = 0;
/// Inform the remote endpoint that no more messages will arrive because of /// Inform the remote endpoint that no more messages will arrive because of
/// an error. /// an error.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message(const error& reason) = 0; virtual void send_close_message(const error& reason) = 0;
};
}; };
} // namespace caf::net } // namespace caf::net::message_oriented
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
#include "caf/byte_buffer.hpp" #include "caf/byte_buffer.hpp"
#include "caf/byte_span.hpp" #include "caf/byte_span.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
...@@ -268,7 +269,7 @@ namespace caf::net { ...@@ -268,7 +269,7 @@ namespace caf::net {
/// Implements a stream_transport that manages a stream socket with encrypted /// Implements a stream_transport that manages a stream socket with encrypted
/// communication over OpenSSL. /// communication over OpenSSL.
class openssl_transport : public stream_transport { class CAF_NET_EXPORT openssl_transport : public stream_transport {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
......
...@@ -4,20 +4,20 @@ ...@@ -4,20 +4,20 @@
#pragma once #pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net { namespace caf::net::stream_oriented {
/// A contract between two stream-oriented layers. The lower layer provides class lower_layer;
/// access to a resource that operates on a byte stream, e.g., a TCP socket. The
/// upper layer requests bytes and consumes raw chunks of data.
class stream_oriented {
public:
class lower_layer;
class upper_layer { /// The upper layer requests bytes from the lower layer and consumes raw chunks
public: /// of data.
class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer(); virtual ~upper_layer();
/// Initializes the upper layer. /// Initializes the upper layer.
...@@ -30,40 +30,21 @@ public: ...@@ -30,40 +30,21 @@ public:
init(socket_manager* owner, lower_layer* down, const settings& config) init(socket_manager* owner, lower_layer* down, const settings& config)
= 0; = 0;
/// Called by the lower layer for cleaning up any state in case of an error.
virtual void abort(const error& reason) = 0;
/// Consumes bytes from the lower layer. /// Consumes bytes from the lower layer.
/// @param buffer Available bytes to read. /// @param buffer Available bytes to read.
/// @param delta Bytes that arrived since last calling this function. /// @param delta Bytes that arrived since last calling this function.
/// @returns The number of consumed bytes. May be zero if waiting for more /// @returns The number of consumed bytes. May be zero if waiting for more
/// input or negative to signal an error. /// input or negative to signal an error.
virtual ptrdiff_t consume(byte_span buffer, byte_span delta) = 0; [[nodiscard]] virtual ptrdiff_t consume(byte_span buffer, byte_span delta)
= 0;
/// Called by the lower layer after some event triggered re-registering the };
/// socket manager for read operations after it has been stopped previously
/// by the read policy. May restart consumption of bytes by setting a new
/// read policy.
virtual void continue_reading() = 0;
/// Prepares any pending data for sending.
virtual bool prepare_send() = 0;
/// Queries whether all pending data has been sent.
virtual bool done_sending() = 0;
};
class lower_layer { /// Provides access to a resource that operates on a byte stream, e.g., a TCP
public: /// socket.
class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer(); virtual ~lower_layer();
/// Queries whether the output device can accept more data straight away.
virtual bool can_send_more() const noexcept = 0;
/// Queries whether the current read policy would cause the transport layer
/// to stop receiving additional bytes.
virtual bool stopped() const noexcept = 0;
/// Configures threshold for the next receive operations. Policies remain /// Configures threshold for the next receive operations. Policies remain
/// active until calling this function again. /// active until calling this function again.
/// @warning Calling this function in `consume` invalidates both byte spans. /// @warning Calling this function in `consume` invalidates both byte spans.
...@@ -81,7 +62,6 @@ public: ...@@ -81,7 +62,6 @@ public:
/// Prepares written data for transfer, e.g., by flushing buffers or /// Prepares written data for transfer, e.g., by flushing buffers or
/// registering sockets for write events. /// registering sockets for write events.
virtual bool end_output() = 0; virtual bool end_output() = 0;
};
}; };
} // namespace caf::net } // namespace caf::net::stream_oriented
...@@ -118,7 +118,9 @@ public: ...@@ -118,7 +118,9 @@ public:
bool end_output() override; bool end_output() override;
bool stopped() const noexcept override; void suspend_reading() override;
bool stopped_reading() const noexcept override;
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
......
...@@ -77,10 +77,14 @@ public: ...@@ -77,10 +77,14 @@ public:
// -- web_socket::lower_layer implementation --------------------------------- // -- web_socket::lower_layer implementation ---------------------------------
bool can_send_more() override; bool can_send_more() const noexcept override;
void suspend_reading() override; void suspend_reading() override;
bool stopped_reading() const noexcept override;
void request_messages() override;
void begin_binary_message() override; void begin_binary_message() override;
byte_buffer& binary_message_buffer() override; byte_buffer& binary_message_buffer() override;
......
...@@ -4,30 +4,51 @@ ...@@ -4,30 +4,51 @@
#pragma once #pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/web_socket/fwd.hpp" #include "caf/net/web_socket/fwd.hpp"
#include <string_view> #include <string_view>
namespace caf::net::web_socket { namespace caf::net::web_socket {
// TODO: Documentation. /// Produces text and binary messages for the upper layer.
class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
class lower_layer {
public: public:
virtual ~lower_layer(); virtual ~lower_layer();
virtual bool can_send_more() = 0; /// Pulls messages from the transport until calling `suspend_reading`.
virtual void suspend_reading() = 0; virtual void request_messages() = 0;
/// Begins transmission of a binary message.
virtual void begin_binary_message() = 0; virtual void begin_binary_message() = 0;
/// Returns the buffer for the current binary message. Must be called between
/// calling `begin_binary_message` and `end_binary_message`.
virtual byte_buffer& binary_message_buffer() = 0; virtual byte_buffer& binary_message_buffer() = 0;
/// Seals the current binary message buffer and ships a new WebSocket frame.
virtual bool end_binary_message() = 0; virtual bool end_binary_message() = 0;
/// Begins transmission of a text message.
virtual void begin_text_message() = 0; virtual void begin_text_message() = 0;
/// Returns the buffer for the current text message. Must be called between
/// calling `begin_text_message` and `end_text_message`.
virtual text_buffer& text_message_buffer() = 0; virtual text_buffer& text_message_buffer() = 0;
/// Seals the current text message buffer and ships a new WebSocket frame.
virtual bool end_text_message() = 0; virtual bool end_text_message() = 0;
/// Sends the close message with @ref status `normal_close`.
virtual void send_close_message() = 0; virtual void send_close_message() = 0;
/// Sends the close message with an custom @ref status.
virtual void send_close_message(status code, std::string_view desc) = 0; virtual void send_close_message(status code, std::string_view desc) = 0;
/// Sends the close message with @ref status `unexpected_condition`.
void send_close_message(const error& reason); void send_close_message(const error& reason);
}; };
......
...@@ -4,13 +4,14 @@ ...@@ -4,13 +4,14 @@
#pragma once #pragma once
#include "caf/detail/net_export.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/web_socket/fwd.hpp" #include "caf/net/web_socket/fwd.hpp"
namespace caf::net::web_socket { namespace caf::net::web_socket {
// TODO: Documentation. /// Consumes text and binary messages from the lower layer.
class upper_layer { class CAF_NET_EXPORT upper_layer {
public: public:
virtual ~upper_layer(); virtual ~upper_layer();
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/generic_lower_layer.hpp"
namespace caf::net {
generic_lower_layer::~generic_lower_layer() {
// nop
}
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net {
generic_upper_layer::~generic_upper_layer() {
// nop
}
} // namespace caf::net
...@@ -15,7 +15,11 @@ bool server::can_send_more() const noexcept { ...@@ -15,7 +15,11 @@ bool server::can_send_more() const noexcept {
} }
void server::suspend_reading() { void server::suspend_reading() {
down_->configure_read(receive_policy::stop()); return down_->suspend_reading();
}
bool server::stopped_reading() const noexcept {
return down_->stopped_reading();
} }
bool server::send_header(context, status code, bool server::send_header(context, status code,
......
...@@ -65,7 +65,7 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) { ...@@ -65,7 +65,7 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
if (msg_size == msg.size() && msg_size + hdr_size == input.size()) { if (msg_size == msg.size() && msg_size + hdr_size == input.size()) {
CAF_LOG_DEBUG("got message of size" << msg_size); CAF_LOG_DEBUG("got message of size" << msg_size);
if (up_->consume(msg) >= 0) { if (up_->consume(msg) >= 0) {
if (!down_->stopped()) if (!down_->stopped_reading())
down_->configure_read(receive_policy::exactly(hdr_size)); down_->configure_read(receive_policy::exactly(hdr_size));
return static_cast<ptrdiff_t>(input.size()); return static_cast<ptrdiff_t>(input.size());
} else { } else {
...@@ -97,13 +97,17 @@ bool length_prefix_framing::can_send_more() const noexcept { ...@@ -97,13 +97,17 @@ bool length_prefix_framing::can_send_more() const noexcept {
return down_->can_send_more(); return down_->can_send_more();
} }
void length_prefix_framing::request_messages() { void length_prefix_framing::suspend_reading() {
if (down_->stopped()) down_->suspend_reading();
down_->configure_read(receive_policy::exactly(hdr_size));
} }
void length_prefix_framing::suspend_reading() { bool length_prefix_framing::stopped_reading() const noexcept {
down_->configure_read(receive_policy::stop()); return down_->stopped_reading();
}
void length_prefix_framing::request_messages() {
if (down_->stopped_reading())
down_->configure_read(receive_policy::exactly(hdr_size));
} }
void length_prefix_framing::begin_message() { void length_prefix_framing::begin_message() {
......
...@@ -4,14 +4,14 @@ ...@@ -4,14 +4,14 @@
#include "caf/net/message_oriented.hpp" #include "caf/net/message_oriented.hpp"
namespace caf::net { namespace caf::net::message_oriented {
message_oriented::upper_layer::~upper_layer() { upper_layer::~upper_layer() {
// nop // nop
} }
message_oriented::lower_layer::~lower_layer() { lower_layer::~lower_layer() {
// nop // nop
} }
} // namespace caf::net } // namespace caf::net::message_oriented
...@@ -4,14 +4,14 @@ ...@@ -4,14 +4,14 @@
#include "caf/net/stream_oriented.hpp" #include "caf/net/stream_oriented.hpp"
namespace caf::net { namespace caf::net::stream_oriented {
stream_oriented::upper_layer::~upper_layer() { upper_layer::~upper_layer() {
// nop // nop
} }
stream_oriented::lower_layer::~lower_layer() { lower_layer::~lower_layer() {
// nop // nop
} }
} // namespace caf::net } // namespace caf::net::stream_oriented
...@@ -93,7 +93,11 @@ bool stream_transport::end_output() { ...@@ -93,7 +93,11 @@ bool stream_transport::end_output() {
return true; return true;
} }
bool stream_transport::stopped() const noexcept { void stream_transport::suspend_reading() {
configure_read(receive_policy::stop());
}
bool stream_transport::stopped_reading() const noexcept {
return max_read_size_ == 0; return max_read_size_ == 0;
} }
......
...@@ -18,12 +18,21 @@ void framing::init(socket_manager*, stream_oriented::lower_layer* down) { ...@@ -18,12 +18,21 @@ void framing::init(socket_manager*, stream_oriented::lower_layer* down) {
// -- web_socket::lower_layer implementation ----------------------------------- // -- web_socket::lower_layer implementation -----------------------------------
bool framing::can_send_more() { bool framing::can_send_more() const noexcept {
return down_->can_send_more(); return down_->can_send_more();
} }
void framing::suspend_reading() { void framing::suspend_reading() {
down_->configure_read(receive_policy::stop()); down_->suspend_reading();
}
bool framing::stopped_reading() const noexcept {
return down_->stopped_reading();
}
void framing::request_messages() {
if (down_->stopped_reading())
down_->configure_read(receive_policy::up_to(2048));
} }
void framing::begin_binary_message() { void framing::begin_binary_message() {
...@@ -91,7 +100,7 @@ ptrdiff_t framing::consume(byte_span input, byte_span) { ...@@ -91,7 +100,7 @@ ptrdiff_t framing::consume(byte_span input, byte_span) {
size_t frame_size = hdr_bytes + hdr.payload_len; size_t frame_size = hdr_bytes + hdr.payload_len;
if (buffer.size() < frame_size) { if (buffer.size() < frame_size) {
// Ask for more data unless the upper layer called suspend_reading. // Ask for more data unless the upper layer called suspend_reading.
if (!down_->stopped()) if (!down_->stopped_reading())
down_->configure_read(receive_policy::up_to(2048)); down_->configure_read(receive_policy::up_to(2048));
down_->configure_read(receive_policy::exactly(frame_size)); down_->configure_read(receive_policy::exactly(frame_size));
return consumed; return consumed;
...@@ -161,7 +170,7 @@ ptrdiff_t framing::consume(byte_span input, byte_span) { ...@@ -161,7 +170,7 @@ ptrdiff_t framing::consume(byte_span input, byte_span) {
buffer = buffer.subspan(frame_size); buffer = buffer.subspan(frame_size);
if (buffer.empty()) { if (buffer.empty()) {
// Ask for more data unless the upper layer called suspend_reading. // Ask for more data unless the upper layer called suspend_reading.
if (!down_->stopped()) if (!down_->stopped_reading())
down_->configure_read(receive_policy::up_to(2048)); down_->configure_read(receive_policy::up_to(2048));
return consumed + static_cast<ptrdiff_t>(frame_size); return consumed + static_cast<ptrdiff_t>(frame_size);
} }
......
...@@ -18,6 +18,10 @@ bool mock_stream_transport::can_send_more() const noexcept { ...@@ -18,6 +18,10 @@ bool mock_stream_transport::can_send_more() const noexcept {
return true; return true;
} }
void mock_stream_transport::suspend_reading() {
configure_read(net::receive_policy::stop());
}
void mock_stream_transport::configure_read(net::receive_policy policy) { void mock_stream_transport::configure_read(net::receive_policy policy) {
min_read_size = policy.min_size; min_read_size = policy.min_size;
max_read_size = policy.max_size; max_read_size = policy.max_size;
...@@ -35,7 +39,7 @@ bool mock_stream_transport::end_output() { ...@@ -35,7 +39,7 @@ bool mock_stream_transport::end_output() {
return true; return true;
} }
bool mock_stream_transport::stopped() const noexcept { bool mock_stream_transport::stopped_reading() const noexcept {
return max_read_size == 0; return max_read_size == 0;
} }
......
...@@ -32,6 +32,8 @@ public: ...@@ -32,6 +32,8 @@ public:
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
void suspend_reading() override;
void configure_read(caf::net::receive_policy policy) override; void configure_read(caf::net::receive_policy policy) override;
void begin_output() override; void begin_output() override;
...@@ -40,7 +42,7 @@ public: ...@@ -40,7 +42,7 @@ public:
bool end_output() override; bool end_output() override;
bool stopped() const noexcept override; bool stopped_reading() const noexcept override;
// -- initialization --------------------------------------------------------- // -- initialization ---------------------------------------------------------
......
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