Commit 991874c9 authored by Dominik Charousset's avatar Dominik Charousset

Turn layering contracts into actual interfaces

Drastically reduce template usage in CAF::net by implementing layers as
interfaces. This gives us better compiler diagnostics and stronger
isolation between the layers.
parent 0407221b
......@@ -5,7 +5,9 @@
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
......
......@@ -8,7 +8,9 @@
#include "caf/cow_tuple.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
......
......@@ -268,7 +268,6 @@ caf_add_component(
detail.serialized_size
detail.type_id_list_builder
detail.unique_function
detail.unordered_flat_map
dictionary
dsl
dynamic_spawn
......@@ -358,6 +357,7 @@ caf_add_component(
typed_response_promise
typed_spawn
unit
unordered_flat_map
uri
uuid)
......
......@@ -20,6 +20,7 @@ enum class hex_format {
template <hex_format format = hex_format::uppercase, class Buf = std::string>
void append_hex(Buf& result, const void* vptr, size_t n) {
using value_type = typename Buf::value_type;
if (n == 0)
return;
auto xs = reinterpret_cast<const uint8_t*>(vptr);
......@@ -30,8 +31,8 @@ void append_hex(Buf& result, const void* vptr, size_t n) {
tbl = "0123456789abcdef";
for (size_t i = 0; i < n; ++i) {
auto c = xs[i];
result.push_back(tbl[c >> 4]);
result.push_back(tbl[c & 0x0F]);
result.push_back(static_cast<value_type>(tbl[c >> 4]));
result.push_back(static_cast<value_type>(tbl[c & 0x0F]));
}
}
......
......@@ -11,7 +11,6 @@
#include "caf/cow_vector.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/disposable.hpp"
#include "caf/flow/coordinated.hpp"
#include "caf/flow/coordinator.hpp"
......@@ -33,6 +32,7 @@
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include "caf/sec.hpp"
#include "caf/unordered_flat_map.hpp"
#include <cstddef>
#include <functional>
......
......@@ -40,7 +40,7 @@ public:
using input_ptr = std::unique_ptr<input_t>;
using input_map = detail::unordered_flat_map<input_key, input_ptr>;
using input_map = unordered_flat_map<input_key, input_ptr>;
// -- constructors, destructors, and assignment operators --------------------
......
......@@ -8,6 +8,7 @@
#include <cstdint>
#include <memory>
#include <string_view>
#include <utility>
#include <variant>
#include <vector>
......@@ -48,6 +49,9 @@ template <class Iterator, class Sentinel = Iterator> struct parser_state;
template <class, class, int> class actor_cast_access;
template <class K, class V, class = std::allocator<std::pair<K, V>>>
class unordered_flat_map;
// -- variadic templates -------------------------------------------------------
template <class...> class const_typed_message_view;
......
......@@ -19,7 +19,6 @@
#include "caf/actor_traits.hpp"
#include "caf/detail/behavior_stack.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
#include "caf/extend.hpp"
......@@ -45,6 +44,7 @@
#include "caf/response_handle.hpp"
#include "caf/sec.hpp"
#include "caf/telemetry/timer.hpp"
#include "caf/unordered_flat_map.hpp"
namespace caf {
......@@ -546,7 +546,7 @@ protected:
std::forward_list<pending_response> awaited_responses_;
/// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
unordered_flat_map<message_id, behavior> multiplexed_responses_;
/// Customization point for setting a default `message` callback.
default_handler default_handler_;
......
......@@ -175,6 +175,10 @@ enum class sec : uint8_t {
disposed,
/// Failed to open a resource.
cannot_open_resource,
/// Received malformed data on a network socket.
protocol_error,
/// Encountered faulty logic in the program.
logic_error,
};
// --(rst-sec-end)--
......
......@@ -10,13 +10,13 @@
#include "caf/detail/comparable.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include "caf/raise_error.hpp"
namespace caf::detail {
namespace caf {
/// A map abstraction with an unsorted `std::vector` providing `O(n)` lookup.
template <class Key, class T,
class Allocator = std::allocator<std::pair<Key, T>>>
template <class Key, class T, class Allocator>
class unordered_flat_map {
public:
// -- member types ----------------------------------------------------------
......@@ -209,7 +209,7 @@ public:
auto i = find(key);
if (i == end())
CAF_RAISE_ERROR(std::out_of_range,
"caf::detail::unordered_flat_map::at out of range");
"caf::unordered_flat_map::at out of range");
return i->second;
}
......@@ -278,4 +278,4 @@ bool operator>=(const unordered_flat_map<K, T, A>& xs,
return !(xs < ys);
}
} // namespace caf::detail
} // namespace caf
......@@ -10,7 +10,6 @@
#include "caf/detail/comparable.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/fwd.hpp"
#include "caf/hash/fnv.hpp"
#include "caf/inspector_access.hpp"
......@@ -18,6 +17,7 @@
#include "caf/intrusive_ptr.hpp"
#include "caf/ip_address.hpp"
#include "caf/make_counted.hpp"
#include "caf/unordered_flat_map.hpp"
namespace caf {
......@@ -59,7 +59,7 @@ public:
using path_list = std::vector<std::string_view>;
/// Separates the query component into key-value pairs.
using query_map = detail::unordered_flat_map<std::string, std::string>;
using query_map = unordered_flat_map<std::string, std::string>;
class CAF_CORE_EXPORT impl_type {
public:
......
......@@ -2,9 +2,9 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE detail.unordered_flat_map
#define CAF_SUITE unordered_flat_map
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/unordered_flat_map.hpp"
#include "core-test.hpp"
......@@ -12,7 +12,7 @@
#include <utility>
#include <vector>
namespace caf::detail {
namespace caf {
template <class T>
bool operator==(const unordered_flat_map<int, T>& xs,
......@@ -26,14 +26,14 @@ bool operator==(const std::vector<std::pair<int, T>>& xs,
return ys == xs;
}
} // namespace caf::detail
} // namespace caf
using std::make_pair;
using std::pair;
using std::string;
using std::vector;
using caf::detail::unordered_flat_map;
using caf::unordered_flat_map;
using namespace caf;
......
......@@ -23,31 +23,44 @@ caf_add_component(
HEADERS
${CAF_NET_HEADERS}
SOURCES
src/datagram_socket.cpp
src/detail/convert_ip_endpoint.cpp
src/detail/rfc6455.cpp
src/host.cpp
src/ip.cpp
src/multiplexer.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/datagram_socket.cpp
src/net/host.cpp
src/net/http/header.cpp
src/net/http/lower_layer.cpp
src/net/http/method.cpp
src/net/http/server.cpp
src/net/http/status.cpp
src/net/http/upper_layer.cpp
src/net/http/v1.cpp
src/net/ip.cpp
src/net/length_prefix_framing.cpp
src/net/message_oriented.cpp
src/net/middleman.cpp
src/net/multiplexer.cpp
src/net/network_socket.cpp
src/net/pipe_socket.cpp
src/net/pollset_updater.cpp
src/net/socket.cpp
src/net/socket_event_layer.cpp
src/net/socket_manager.cpp
src/net/stream_oriented.cpp
src/net/stream_socket.cpp
src/net/stream_transport.cpp
src/net/tcp_accept_socket.cpp
src/net/tcp_stream_socket.cpp
src/net/udp_datagram_socket.cpp
src/net/web_socket/client.cpp
src/net/web_socket/default_trait.cpp
src/net/web_socket/frame.cpp
src/net/web_socket/framing.cpp
src/net/web_socket/handshake.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
src/net/web_socket/lower_layer.cpp
src/net/web_socket/server.cpp
src/net/web_socket/upper_layer.cpp
TEST_SOURCES
test/net-test.cpp
test/pem.cpp
......
......@@ -7,7 +7,6 @@
#include "caf/actor_traits.hpp"
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
......@@ -18,6 +17,7 @@
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/unordered_flat_map.hpp"
namespace caf::net {
......@@ -40,7 +40,9 @@ public:
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
using fallback_handler_sig = result<message>(abstract_actor_shell*, message&);
using fallback_handler = unique_callback_ptr<fallback_handler_sig>;
// -- constructors, destructors, and assignment operators --------------------
......@@ -48,6 +50,10 @@ public:
~abstract_actor_shell() override;
// -- properties -------------------------------------------------------------
bool terminated() const noexcept;
// -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox.
......@@ -56,7 +62,20 @@ public:
/// Overrides the default handler for unexpected messages.
template <class F>
void set_fallback(F f) {
fallback_ = make_type_erased_callback(std::move(f));
using msg_res_t = result<message>;
using self_ptr_t = abstract_actor_shell*;
if constexpr (std::is_invocable_r_v<msg_res_t, F, self_ptr_t, message&>) {
fallback_ = make_type_erased_callback(std::move(f));
} else {
static_assert(std::is_invocable_r_v<msg_res_t, F, message&>,
"illegal signature for the fallback handler: must be "
"'result<message>(message&)' or "
"'result<message>(abstract_actor_shell*, message&)'");
auto g = [f = std::move(f)](self_ptr_t, message& msg) mutable {
return f(msg);
};
fallback_ = make_type_erased_callback(std::move(g));
}
}
// -- mailbox access ---------------------------------------------------------
......@@ -114,7 +133,7 @@ protected:
fallback_handler fallback_;
// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
unordered_flat_map<message_id, behavior> multiplexed_responses_;
};
} // namespace caf::net
......@@ -4,68 +4,67 @@
#pragma once
#include "caf/logger.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/send.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/settings.hpp"
namespace caf::net {
/// A connection_acceptor accepts connections from an accept socket and creates
/// socket managers to handle them via its factory.
template <class Socket, class Factory>
class connection_acceptor {
class connection_acceptor : public socket_event_layer {
public:
// -- member types -----------------------------------------------------------
using input_tag = tag::io_event_oriented;
using socket_type = Socket;
using factory_type = Factory;
using read_result = typename socket_manager::read_result;
using read_result = typename socket_event_layer::read_result;
using write_result = typename socket_manager::write_result;
using write_result = typename socket_event_layer::write_result;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
explicit connection_acceptor(size_t limit, Ts&&... xs)
: factory_(std::forward<Ts>(xs)...), limit_(limit) {
explicit connection_acceptor(Socket fd, size_t limit, Ts&&... xs)
: fd_(fd), limit_(limit), factory_(std::forward<Ts>(xs)...) {
// nop
}
// -- interface functions ----------------------------------------------------
// -- factories --------------------------------------------------------------
template <class... Ts>
static std::unique_ptr<connection_acceptor>
make(Socket fd, size_t limit, Ts&&... xs) {
return std::make_unique<connection_acceptor>(fd, limit,
std::forward<Ts>(xs)...);
}
template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
// -- implementation of socket_event_layer -----------------------------------
error init(socket_manager* owner, const settings& cfg) override {
CAF_LOG_TRACE("");
owner_ = owner;
cfg_ = cfg;
if (auto err = factory_.init(owner, cfg))
return err;
down->register_reading();
owner->register_reading();
return none;
}
template <class LowerLayerPtr>
read_result handle_read_event(LowerLayerPtr down) {
read_result handle_read_event() override {
CAF_LOG_TRACE("");
if (auto x = accept(down->handle())) {
socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr());
if (auto x = accept(fd_)) {
socket_manager_ptr child = factory_.make(owner_->mpx_ptr(), *x);
if (!child) {
CAF_LOG_ERROR("factory failed to create a new child");
down->abort_reason(sec::runtime_error);
return read_result::stop;
return read_result::abort;
}
if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
down->abort_reason(std::move(err));
return read_result::stop;
return read_result::abort;
}
if (limit_ == 0) {
return read_result::again;
......@@ -78,74 +77,41 @@ public:
}
}
template <class LowerLayerPtr>
static read_result handle_buffered_data(LowerLayerPtr) {
read_result handle_buffered_data() override {
return read_result::again;
}
template <class LowerLayerPtr>
static read_result handle_continue_reading(LowerLayerPtr) {
read_result handle_continue_reading() override {
return read_result::again;
}
template <class LowerLayerPtr>
write_result handle_write_event(LowerLayerPtr) {
write_result handle_write_event() override {
CAF_LOG_ERROR("connection_acceptor received write event");
return write_result::stop;
}
template <class LowerLayerPtr>
static write_result handle_continue_writing(LowerLayerPtr) {
write_result handle_continue_writing() override {
CAF_LOG_ERROR("connection_acceptor received continue writing event");
return write_result::stop;
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
void abort(const error& reason) override {
CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
factory_.abort(reason);
}
private:
factory_type factory_;
socket_manager* owner_;
Socket fd_;
size_t limit_;
size_t accepted_ = 0;
settings cfg_;
};
/// Converts a function object into a factory object for a
/// @ref connection_acceptor.
template <class FunctionObject>
class connection_acceptor_factory_adapter {
public:
explicit connection_acceptor_factory_adapter(FunctionObject f)
: f_(std::move(f)) {
// nop
}
connection_acceptor_factory_adapter(connection_acceptor_factory_adapter&&)
= default;
factory_type factory_;
error init(socket_manager*, const settings&) {
return none;
}
socket_manager* owner_ = nullptr;
template <class Socket>
socket_manager_ptr make(Socket connected_socket, multiplexer* mpx) {
return f_(std::move(connected_socket), mpx);
}
void abort(const error&) {
// nop
}
size_t accepted_ = 0;
private:
FunctionObject f_;
settings cfg_;
};
} // 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 <cstddef>
#include <cstdint>
// -- hard-coded default values for various CAF options ------------------------
namespace caf::defaults::middleman {
/// Maximum number of cached buffers for sending payloads.
constexpr size_t max_payload_buffers = 100;
/// Maximum number of cached buffers for sending headers.
constexpr size_t max_header_buffers = 10;
/// Port to listen on for tcp.
constexpr uint16_t tcp_port = 0;
} // namespace caf::defaults::middleman
......@@ -4,27 +4,21 @@
#pragma once
#include <memory>
#include "caf/intrusive_ptr.hpp"
#include "caf/type_id.hpp"
#include <memory>
#include <vector>
namespace caf::net {
// -- templates ----------------------------------------------------------------
template <class UpperLayer>
class stream_transport;
template <class Factory>
class datagram_transport;
template <class Application, class IdType = unit_t>
class transport_worker;
template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher;
template <class... Sigs>
class typed_actor_shell;
......@@ -38,16 +32,18 @@ class actor_shell_ptr;
class middleman;
class multiplexer;
class socket_manager;
class this_host;
// -- structs ------------------------------------------------------------------
struct datagram_socket;
struct network_socket;
struct pipe_socket;
struct receive_policy;
struct socket;
struct stream_socket;
struct tcp_accept_socket;
struct tcp_stream_socket;
struct datagram_socket;
struct udp_datagram_socket;
// -- smart pointer aliases ----------------------------------------------------
......@@ -56,4 +52,8 @@ using multiplexer_ptr = std::shared_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
// -- miscellaneous aliases ----------------------------------------------------
using text_buffer = std::vector<char>;
} // namespace caf::net
......@@ -9,7 +9,9 @@
namespace caf::net {
struct CAF_NET_EXPORT this_host {
/// Groups functions for managing the host system.
class CAF_NET_EXPORT this_host {
public:
/// Initializes the network subsystem.
static error startup();
......
......@@ -7,6 +7,6 @@
namespace caf::net::http {
/// Stores context information for a request. For HTTP/2, this is the stream ID.
struct context {};
class context {};
} // namespace caf::net::http
......@@ -4,8 +4,23 @@
#pragma once
namespace caf::tag {
#include "caf/fwd.hpp"
struct io_event_oriented {};
#include <string_view>
} // namespace caf::tag
namespace caf::net::http {
class context;
class header;
class lower_layer;
class server;
class upper_layer;
using header_fields_map
= unordered_flat_map<std::string_view, std::string_view>;
enum class method : uint8_t;
enum class status : uint16_t;
} // namespace caf::net::http
......@@ -4,7 +4,7 @@
#pragma once
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/unordered_flat_map.hpp"
#include <string_view>
......@@ -12,6 +12,6 @@ namespace caf::net::http {
/// Convenience type alias for a key-value map storing header fields.
using header_fields_map
= detail::unordered_flat_map<std::string_view, std::string_view>;
= unordered_flat_map<std::string_view, std::string_view>;
} // namespace caf::net::http
// 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/fwd.hpp"
namespace caf::net::http {
/// Parses HTTP requests and passes them to the upper layer.
class lower_layer {
public:
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.
virtual bool
send_header(context ctx, status code, const header_fields_map& fields)
= 0;
/// Sends the payload after the header.
virtual bool send_payload(context, const_byte_span bytes) = 0;
/// Sends a chunk of data if the full payload is unknown when starting to
/// send.
virtual bool send_chunk(context, const_byte_span bytes) = 0;
/// Sends the last chunk, completing a chunked payload.
virtual bool send_end_of_chunks() = 0;
/// Terminates an HTTP context.
virtual void fin(context) = 0;
/// Convenience function for sending header and payload. Automatically sets
/// the header fields `Content-Type` and `Content-Length`.
bool send_response(context ctx, status codek, std::string_view content_type,
const_byte_span content);
};
} // namespace caf::net::http
......@@ -6,6 +6,7 @@
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/fwd.hpp"
#include <cstdint>
#include <string>
......
This diff is collapsed.
......@@ -6,6 +6,7 @@
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/fwd.hpp"
#include "caf/string_view.hpp"
#include <cstdint>
......
// 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/fwd.hpp"
namespace caf::net::http {
/// Operates on HTTP requests.
class upper_layer {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 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.
/// @param ctx Identifies this request. The response message must pass this
/// back to the lower layer. Allows clients to send multiple
/// requests in "parallel" (i.e., send multiple requests before
/// receiving the response on the first one).
/// @param hdr The header fields for the received message.
/// @param payload The payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an
/// error.
/// @note Discarded data is lost permanently.
virtual ptrdiff_t
consume(context ctx, const header& hdr, const_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;
};
} // namespace caf::net::http
// 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/byte_span.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a hypertext-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// hypertext-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class hypertext_oriented_layer_ptr {
public:
using context_type = typename Layer::context_type;
using status_code_type = typename Layer::status_code_type;
using header_fields_type = typename Layer::header_fields_type;
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
/// Queries whether the underlying transport can send additional data.
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
/// Asks the underlying transport to stop receiving additional data until
/// resumed.
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
/// Returns the socket handle.
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
/// Sends a response header for answering the request identified by the
/// context.
/// @param context Identifies which request this response belongs to.
/// @param code Indicates either success or failure to the client.
/// @param fields Various informational fields for the client. When sending
/// a payload afterwards, the fields should at least include
/// the content length.
bool send_header(context_type context, status_code_type code,
const header_fields_type& fields) {
return lptr_->send_header(llptr_, context, code, fields);
}
/// Sends a payload to the client. Must follow a header.
/// @param context Identifies which request this response belongs to.
/// @param bytes Arbitrary data for the client.
/// @pre `bytes.size() > 0`
[[nodiscard]] bool send_payload(context_type context,
const_byte_span bytes) {
return lptr_->send_payload(llptr_, context, bytes);
}
/// Sends a single chunk of arbitrary data. The chunks must follow a header.
/// @pre `bytes.size() > 0`
[[nodiscard]] bool send_chunk(context_type context, const_byte_span bytes) {
return lptr_->send_chunk(llptr_, context, bytes);
}
/// Informs the client that the transfer completed, i.e., that the server
/// will not send additional chunks.
[[nodiscard]] bool send_end_of_chunks(context_type context) {
return lptr_->send_end_of_chunks(llptr_, context);
}
/// Convenience function for completing a request 'raw' (without adding
/// additional header fields) in a single function call. Calls
/// `send_header`, `send_payload` and `fin`.
bool send_raw_response(context_type context, status_code_type code,
const header_fields_type& fields,
const_byte_span content) {
if (lptr_->send_header(llptr_, context, code, fields)
&& (content.empty()
|| lptr_->send_payload(llptr_, context, content))) {
lptr_->fin(llptr_, context);
return true;
} else {
return false;
}
}
/// Convenience function for completing a request in a single function call.
/// Automatically sets the header fields 'Content-Type' and
/// 'Content-Length'. Calls `send_header`, `send_payload` and `fin`.
bool send_response(context_type context, status_code_type code,
header_fields_type fields, std::string_view content_type,
const_byte_span content) {
std::string len;
if (!content.empty()) {
auto len = std::to_string(content.size());
fields.emplace("Content-Length", len);
fields.emplace("Content-Type", content_type);
}
return send_raw_response(context, code, fields, content);
}
/// Convenience function for completing a request in a single function call.
/// Automatically sets the header fields 'Content-Type' and
/// 'Content-Length'. Calls `send_header`, `send_payload` and `fin`.
bool send_response(context_type context, status_code_type code,
std::string_view content_type, const_byte_span content) {
std::string len;
header_fields_type fields;
if (!content.empty()) {
len = std::to_string(content.size());
fields.emplace("Content-Type", content_type);
fields.emplace("Content-Length", len);
}
return send_raw_response(context, code, fields, content);
}
bool send_response(context_type context, status_code_type code,
header_fields_type fields, std::string_view content_type,
std::string_view content) {
return send_response(context, code, content_type, std::move(fields),
as_bytes(make_span(content)));
}
bool send_response(context_type context, status_code_type code,
std::string_view content_type,
std::string_view content) {
return send_response(context, code, content_type,
as_bytes(make_span(content)));
}
void fin(context_type context) {
lptr_->fin(llptr_, context);
}
/// Sets an abort reason on the transport.
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
/// Returns the current abort reason on the transport or a
/// default-constructed error is no error occurred yet.
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
hypertext_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
hypertext_oriented_layer_ptr(const hypertext_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_hypertext_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
using result_t = hypertext_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
......@@ -7,14 +7,12 @@
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/net/consumer_adapter.hpp"
#include "caf/net/message_oriented.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/producer_adapter.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/settings.hpp"
#include "caf/tag/message_oriented.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/no_auto_reading.hpp"
#include <utility>
......@@ -22,26 +20,39 @@ namespace caf::net {
/// Translates between a message-oriented transport and data flows.
///
/// The trait class converts between the native and the wire format:
/// The `Trait` provides a customization point that converts between native and
/// wire format.
///
/// ~~~
/// struct my_trait {
/// bool convert(const T& value, byte_buffer& bytes);
/// bool convert(const_byte_span bytes, T& value);
/// class my_trait {
/// using input_type = ...;
/// using output_type = ...;
/// bool convert(const_byte_span bytes, input_type& value);
/// bool convert(const output_type& value, byte_buffer& bytes);
/// };
/// ~~~
template <class T, class Trait, class Tag = tag::message_oriented>
class message_flow_bridge : public tag::no_auto_reading {
template <class Trait>
class message_flow_bridge : public message_oriented::upper_layer {
public:
using input_tag = Tag;
/// The input type for the application.
using input_type = typename Trait::input_type;
using buffer_type = async::spsc_buffer<T>;
/// The output type for of application.
using output_type = typename Trait::output_type;
using consumer_resource_t = async::consumer_resource<T>;
/// The resource type we pull from. We consume the output of the application.
using pull_resource_t = async::consumer_resource<output_type>;
using producer_resource_t = async::producer_resource<T>;
/// The buffer type from the @ref pull_resource_t.
using pull_buffer_t = async::spsc_buffer<output_type>;
message_flow_bridge(consumer_resource_t in_res, producer_resource_t out_res,
/// Type for the producer adapter. We produce the input of the application.
using push_resource_t = async::producer_resource<input_type>;
/// The buffer type from the @ref push_resource_t.
using push_buffer_t = async::spsc_buffer<input_type>;
message_flow_bridge(pull_resource_t in_res, push_resource_t out_res,
Trait trait)
: in_res_(std::move(in_res)),
trait_(std::move(trait)),
......@@ -53,66 +64,41 @@ public:
// nop
}
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr, const settings& cfg) {
mgr_ = mgr;
if constexpr (caf::detail::has_init_v<Trait>) {
if (auto err = init_res(trait_.init(cfg)))
return err;
}
error init(net::socket_manager* mgr, message_oriented::lower_layer* down,
const settings&) {
down_ = down;
if (in_res_) {
in_ = consumer_adapter<buffer_type>::try_open(mgr, in_res_);
in_ = consumer_adapter<pull_buffer_t>::try_open(mgr, in_res_);
in_res_ = nullptr;
}
if (out_res_) {
out_ = producer_adapter<buffer_type>::try_open(mgr, out_res_);
out_ = producer_adapter<push_buffer_t>::try_open(mgr, out_res_);
out_res_ = nullptr;
}
if (!in_ && !out_)
return make_error(sec::cannot_open_resource,
"flow bridge cannot run without at least one resource");
"a flow bridge needs at least one valid resource");
return none;
}
template <class LowerLayerPtr>
bool write(LowerLayerPtr down, const T& item) {
if constexpr (std::is_same_v<Tag, tag::message_oriented>) {
down->begin_message();
auto& buf = down->message_buffer();
return trait_.convert(item, buf) && down->end_message();
} else {
static_assert(std::is_same_v<Tag, tag::mixed_message_oriented>);
if (trait_.converts_to_binary(item)) {
down->begin_binary_message();
auto& bytes = down->binary_message_buffer();
return trait_.convert(item, bytes) && down->end_binary_message();
} else {
down->begin_text_message();
auto& text = down->text_message_buffer();
return trait_.convert(item, text) && down->end_text_message();
}
}
bool write(const input_type& item) {
down_->begin_message();
auto& buf = down_->message_buffer();
return trait_.convert(item, buf) && down_->end_message();
}
template <class LowerLayerPtr>
struct write_helper {
using bridge_type = message_flow_bridge;
bridge_type* bridge;
LowerLayerPtr down;
message_flow_bridge* bridge;
bool aborted = false;
size_t consumed = 0;
error err;
write_helper(bridge_type* bridge, LowerLayerPtr down)
: bridge(bridge), down(down) {
write_helper(message_flow_bridge* bridge) : bridge(bridge) {
// nop
}
void on_next(const T& item) {
if (!bridge->write(down, item)) {
void on_next(const input_type& item) {
if (!bridge->write(item))
aborted = true;
return;
}
}
void on_complete() {
......@@ -124,20 +110,18 @@ public:
}
};
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
write_helper<LowerLayerPtr> helper{this, down};
while (down->can_send_more() && in_) {
bool prepare_send() override {
write_helper helper{this};
while (down_->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down->send_close_message(helper.err);
down_->send_close_message(helper.err);
} else {
down->send_close_message();
down_->send_close_message();
}
in_ = nullptr;
} else if (helper.aborted) {
down->abort_reason(make_error(sec::conversion_failed));
in_->cancel();
in_ = nullptr;
return false;
......@@ -148,13 +132,11 @@ public:
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
bool done_sending() override {
return !in_ || !in_->has_data();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::socket_disconnected || reason == sec::disposed)
......@@ -169,89 +151,35 @@ public:
}
}
template <class U = Tag, class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
return -1;
}
T val;
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
return -1;
}
if (out_->push(std::move(val)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
template <class U = Tag, class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr down, byte_span buf) {
return consume(down, buf);
}
template <class U = Tag, class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, std::string_view buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
ptrdiff_t consume(byte_span buf) {
if (!out_)
return -1;
}
T val;
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
input_type val;
if (!trait_.convert(buf, val))
return -1;
}
if (out_->push(std::move(val)) == 0)
down->suspend_reading();
down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
private:
error init_res(error err) {
return err;
}
error init_res(consumer_resource_t in, producer_resource_t out) {
in_res_ = std::move(in);
out_res_ = std::move(out);
return caf::none;
}
error init_res(std::tuple<consumer_resource_t, producer_resource_t> in_out) {
auto& [in, out] = in_out;
return init_res(std::move(in), std::move(out));
}
error init_res(std::pair<consumer_resource_t, producer_resource_t> in_out) {
auto& [in, out] = in_out;
return init_res(std::move(in), std::move(out));
}
template <class R>
error init_res(expected<R> res) {
if (res)
return init_res(*res);
else
return std::move(res.error());
}
/// Points to the manager that runs this protocol stack.
net::socket_manager* mgr_ = nullptr;
/// Points to the next layer down the protocol stack.
message_oriented::lower_layer* down_ = nullptr;
/// Incoming messages, serialized to the socket.
consumer_adapter_ptr<buffer_type> in_;
consumer_adapter_ptr<pull_buffer_t> in_;
/// Outgoing messages, deserialized from the socket.
producer_adapter_ptr<buffer_type> out_;
producer_adapter_ptr<push_buffer_t> out_;
/// Converts between raw bytes and items.
Trait trait_;
/// Discarded after initialization.
consumer_resource_t in_res_;
pull_resource_t in_res_;
/// Discarded after initialization.
producer_resource_t out_res_;
push_resource_t out_res_;
};
} // 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// A contract between two message-oriented layers. The lower layer provides
/// 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 {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 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.
/// @param payload Payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an
/// error.
/// @note Discarded data is lost permanently.
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 {
public:
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`.
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
/// output buffer as necessary.
virtual void begin_message() = 0;
/// Returns a reference to the buffer for assembling the current message.
/// Users may only call this function and write to the buffer between
/// calling `begin_message()` and `end_message()`.
/// @note The lower layers may pre-fill the buffer, e.g., to prefix custom
/// headers.
virtual byte_buffer& message_buffer() = 0;
/// Seals and prepares a message for transfer.
/// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code.
virtual bool end_message() = 0;
/// Inform the remote endpoint that no more messages will arrive.
virtual void send_close_message() = 0;
/// Inform the remote endpoint that no more messages will arrive because of
/// an error.
virtual void send_close_message(const error& reason) = 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a message-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_message() {
lptr_->begin_message(llptr_);
}
[[nodiscard]] auto& message_buffer() {
return lptr_->message_buffer(llptr_);
}
[[nodiscard]] bool end_message() {
return lptr_->end_message(llptr_);
}
template <class... Ts>
bool send_close_message(Ts&&... xs) {
return lptr_->send_close_message(llptr_, std::forward<Ts>(xs)...);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
message_oriented_layer_ptr(const message_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_message_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
using result_t = message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
......@@ -39,32 +39,6 @@ public:
~middleman() override;
// -- socket manager functions -----------------------------------------------
/// Creates a new acceptor that accepts incoming connections from @p sock and
/// creates socket managers using @p factory.
/// @param sock An accept socket in listening mode. For a TCP socket, this
/// socket must already listen to an address plus port.
/// @param factory A function object for creating socket managers that take
/// ownership of incoming connections.
/// @param limit The maximum number of connections that this acceptor should
/// establish or 0 for 'no limit'.
template <class Socket, class Factory>
auto make_acceptor(Socket sock, Factory factory, size_t limit = 0) {
using connected_socket_type = typename Socket::connected_socket_type;
if constexpr (detail::is_callable_with<Factory, connected_socket_type,
multiplexer*>::value) {
connection_acceptor_factory_adapter<Factory> adapter{std::move(factory)};
return make_acceptor(std::move(sock), std::move(adapter), limit);
} else {
using impl = connection_acceptor<Socket, Factory>;
auto ptr = make_socket_manager<impl>(std::move(sock), &mpx_, limit,
std::move(factory));
mpx_.init(ptr);
return ptr;
}
}
// -- interface functions ----------------------------------------------------
void start() override;
......
// 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a mixed-message-oriented layer with a pointer to its
/// lower layer. Both pointers are then used to implement the interface required
/// for a mixed-message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class mixed_message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
void begin_binary_message() {
lptr_->begin_binary_message(llptr_);
}
auto& binary_message_buffer() {
return lptr_->binary_message_buffer(llptr_);
}
bool end_binary_message() {
return lptr_->end_binary_message(llptr_);
}
void begin_text_message() {
lptr_->begin_text_message(llptr_);
}
auto& text_message_buffer() {
return lptr_->text_message_buffer(llptr_);
}
bool end_text_message() {
return lptr_->end_text_message(llptr_);
}
template <class... Ts>
bool send_close_message(Ts&&... xs) {
return lptr_->send_close_message(llptr_, std::forward<Ts>(xs)...);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
mixed_message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
mixed_message_oriented_layer_ptr(const mixed_message_oriented_layer_ptr&)
= default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_mixed_message_oriented_layer_ptr(Layer* this_layer,
LowerLayerPtr down) {
using result_t = mixed_message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
......@@ -10,12 +10,12 @@
#include "caf/action.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
#include "caf/unordered_flat_map.hpp"
extern "C" {
......@@ -37,7 +37,7 @@ public:
socket_manager_ptr mgr;
};
using poll_update_map = detail::unordered_flat_map<socket, poll_update>;
using poll_update_map = unordered_flat_map<socket, poll_update>;
using pollfd_list = std::vector<pollfd>;
......
......@@ -7,7 +7,6 @@
#include "caf/byte_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/has_after_reading.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
......@@ -123,7 +122,7 @@ inline conn_ptr make_conn(const ctx_ptr& ctx, stream_socket fd) {
}
/// Manages an OpenSSL connection.
class policy {
class policy : public stream_transport::policy {
public:
// -- constructors, destructors, and assignment operators --------------------
......@@ -182,65 +181,29 @@ public:
}
}
// -- interface functions for the stream transport ---------------------------
/// Graceful shutdown.
void notify_close() {
SSL_shutdown(conn_.get());
}
// -- implementation of stream_transport::policy -----------------------------
/// Fetches a string representation for the last error.
std::string fetch_error_str() {
return openssl::fetch_error_str();
}
/// Reads data from the SSL connection into the buffer.
ptrdiff_t read(stream_socket, byte_span buf) {
return SSL_read(conn_.get(), buf.data(), static_cast<int>(buf.size()));
}
ptrdiff_t read(stream_socket, byte_span) override;
/// Writes data from the buffer to the SSL connection.
ptrdiff_t write(stream_socket, const_byte_span buf) {
return SSL_write(conn_.get(), buf.data(), static_cast<int>(buf.size()));
}
ptrdiff_t write(stream_socket, const_byte_span) override;
/// Performs a TLS/SSL handshake with the server.
ptrdiff_t connect(stream_socket) {
return SSL_connect(conn_.get());
}
stream_transport_error last_error(stream_socket, ptrdiff_t) override;
/// Waits for the client to performs a TLS/SSL handshake.
ptrdiff_t accept(stream_socket) {
return SSL_accept(conn_.get());
}
/// Returns the last SSL error.
stream_transport_error last_error(stream_socket fd, ptrdiff_t ret) {
switch (SSL_get_error(conn_.get(), static_cast<int>(ret))) {
case SSL_ERROR_NONE:
case SSL_ERROR_WANT_ACCEPT:
case SSL_ERROR_WANT_CONNECT:
// For all of these, OpenSSL docs say to do the operation again later.
return stream_transport_error::temporary;
case SSL_ERROR_SYSCALL:
// Need to consult errno, which we just leave to the default policy.
return default_stream_transport_policy::last_error(fd, ret);
case SSL_ERROR_WANT_READ:
return stream_transport_error::want_read;
case SSL_ERROR_WANT_WRITE:
return stream_transport_error::want_write;
default:
// Errors like SSL_ERROR_WANT_X509_LOOKUP are technically temporary, but
// we do not configure any callbacks. So seeing this is a red flag.
return stream_transport_error::permanent;
}
}
ptrdiff_t connect(stream_socket) override;
/// Graceful shutdown.
void notify_close() {
SSL_shutdown(conn_.get());
}
ptrdiff_t accept(stream_socket) override;
/// Returns the number of bytes that are buffered internally and that are
/// available for immediate read.
size_t buffered() {
return static_cast<size_t>(SSL_pending(conn_.get()));
}
size_t buffered() override;
private:
/// Our SSL connection data.
......@@ -305,27 +268,19 @@ namespace caf::net {
/// Implements a stream_transport that manages a stream socket with encrypted
/// communication over OpenSSL.
template <class UpperLayer>
class openssl_transport
: public stream_transport_base<openssl::policy, UpperLayer> {
class openssl_transport : public stream_transport {
public:
// -- member types -----------------------------------------------------------
using super = stream_transport_base<openssl::policy, UpperLayer>;
using super = stream_transport;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
explicit openssl_transport(openssl::conn_ptr conn, Ts&&... xs)
: super(openssl::policy{std::move(conn)}, std::forward<Ts>(xs)...) {
// nop
}
openssl_transport(stream_socket fd, openssl::conn_ptr conn,
upper_layer_ptr up);
template <class... Ts>
openssl_transport(openssl::policy policy, Ts&&... xs)
: super(std::move(policy), std::forward<Ts>(xs)...) {
// nop
}
private:
openssl::policy ssl_policy_;
};
} // namespace caf::net
......@@ -4,7 +4,9 @@
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp"
#include <array>
......@@ -14,7 +16,7 @@
namespace caf::net {
class pollset_updater : public socket_manager {
class CAF_NET_EXPORT pollset_updater : public socket_event_layer {
public:
// -- member types -----------------------------------------------------------
......@@ -37,20 +39,15 @@ public:
// -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer* parent);
explicit pollset_updater(pipe_socket fd);
~pollset_updater() override;
// -- factories --------------------------------------------------------------
// -- properties -------------------------------------------------------------
static std::unique_ptr<pollset_updater> make(pipe_socket fd);
/// Returns the managed socket.
pipe_socket handle() const noexcept {
return socket_cast<pipe_socket>(handle_);
}
// -- implementation of socket_event_layer -----------------------------------
// -- interface functions ----------------------------------------------------
error init(const settings& config) override;
error init(socket_manager* owner, const settings& cfg) override;
read_result handle_read_event() override;
......@@ -62,9 +59,11 @@ public:
write_result handle_continue_writing() override;
void handle_error(sec code) override;
void abort(const error& reason) override;
private:
pipe_socket fd_;
multiplexer* mpx_ = nullptr;
msg_buf buf_;
size_t buf_size_ = 0;
};
......
......@@ -8,8 +8,15 @@
namespace caf::net {
/// Configures how many bytes a @ref stream_transport receives before calling
/// `consume` on its upper layer.
struct receive_policy {
/// Configures how many bytes @ref stream_transport must read before it may
/// call `consume` on its upper layer.
uint32_t min_size;
/// Configures how many bytes @ref stream_transport may read at most before
/// it calls `consume` on its upper layer.
uint32_t max_size;
/// @pre `min_size > 0`
......@@ -24,6 +31,7 @@ struct receive_policy {
return {size, size};
}
/// @pre `max_size >= 1`
static constexpr receive_policy up_to(uint32_t max_size) {
return {1, max_size};
}
......
// 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"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// The lowest layer in a protocol stack. Called by a @ref socket_manager
/// directly.
class CAF_NET_EXPORT socket_event_layer {
public:
virtual ~socket_event_layer();
/// Encodes how to proceed after a read operation.
enum class read_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to write to the socket instead of reading
/// from the socket.
want_write,
/// Indicates that the manager no longer reads from the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Encodes how to proceed after a write operation.
enum class write_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to read from the socket instead of
/// writing to the socket.
want_read,
/// Indicates that the manager no longer writes to the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
virtual error init(socket_manager* owner, const settings& cfg) = 0;
virtual read_result handle_read_event() = 0;
virtual read_result handle_buffered_data() = 0;
virtual read_result handle_continue_reading() = 0;
virtual write_result handle_write_event() = 0;
virtual write_result handle_continue_writing() = 0;
virtual void abort(const error& reason) = 0;
};
} // namespace caf::net
......@@ -12,15 +12,15 @@ namespace caf::net {
template <class Socket>
class socket_guard {
public:
socket_guard() noexcept : sock_(invalid_socket_id) {
socket_guard() noexcept : fd_(invalid_socket_id) {
// nop
}
explicit socket_guard(Socket sock) noexcept : sock_(sock) {
explicit socket_guard(Socket fd) noexcept : fd_(fd) {
// nop
}
socket_guard(socket_guard&& other) noexcept : sock_(other.release()) {
socket_guard(socket_guard&& other) noexcept : fd_(other.release()) {
// nop
}
......@@ -34,28 +34,32 @@ public:
socket_guard& operator=(const socket_guard&) = delete;
~socket_guard() {
if (sock_.id != invalid_socket_id)
close(sock_);
if (fd_.id != invalid_socket_id)
close(fd_);
}
void reset(Socket x) noexcept {
if (sock_.id != invalid_socket_id)
close(sock_);
sock_ = x;
if (fd_.id != invalid_socket_id)
close(fd_);
fd_ = x;
}
Socket release() noexcept {
auto sock = sock_;
sock_.id = invalid_socket_id;
auto sock = fd_;
fd_.id = invalid_socket_id;
return sock;
}
Socket get() noexcept {
return fd_;
}
Socket socket() const noexcept {
return sock_;
return fd_;
}
private:
Socket sock_;
Socket fd_;
};
template <class Socket>
......
This diff is collapsed.
// 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/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// A contract between two stream-oriented layers. The lower layer provides
/// 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 {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 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.
/// @param buffer Available bytes to read.
/// @param delta Bytes that arrived since last calling this function.
/// @returns The number of consumed bytes. May be zero if waiting for more
/// input or negative to signal an error.
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 {
public:
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
/// active until calling this function again.
/// @warning Calling this function in `consume` invalidates both byte spans.
virtual void configure_read(receive_policy policy) = 0;
/// Prepares the layer for outgoing traffic, e.g., by allocating an output
/// buffer as necessary.
virtual void begin_output() = 0;
/// Returns a reference to the output buffer. Users may only call this
/// function and write to the buffer between calling `begin_output()` and
/// `end_output()`.
virtual byte_buffer& output_buffer() = 0;
/// Prepares written data for transfer, e.g., by flushing buffers or
/// registering sockets for write events.
virtual bool end_output() = 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Wraps a pointer to a stream-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// stream-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class stream_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_output() {
lptr_->begin_output(llptr_);
}
byte_buffer& output_buffer() {
return lptr_->output_buffer(llptr_);
}
void end_output() {
lptr_->end_output(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
void configure_read(receive_policy policy) {
lptr_->configure_read(llptr_, policy);
}
bool stopped() const noexcept {
return lptr_->stopped(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
stream_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
stream_oriented_layer_ptr(const stream_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_stream_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
return stream_oriented_layer_ptr<Layer, LowerLayerPtr>{this_layer, down};
}
} // namespace caf::net
This diff is collapsed.
......@@ -17,7 +17,7 @@
namespace caf::detail {
template <template <class> class Transport, class Trait>
template <class Transport, class Trait>
class ws_acceptor_factory {
public:
using connector_pointer = net::web_socket::flow_connector_ptr<Trait>;
......@@ -32,10 +32,11 @@ public:
}
template <class Socket>
net::socket_manager_ptr make(Socket fd, net::multiplexer* mpx) {
using app_t = net::web_socket::flow_bridge<Trait>;
using stack_t = Transport<net::web_socket::server<app_t>>;
return net::make_socket_manager<stack_t>(fd, mpx, connector_);
net::socket_manager_ptr make(net::multiplexer* mpx, Socket fd) {
auto app = net::web_socket::flow_bridge<Trait>::make(connector_);
auto ws = net::web_socket::server::make(std::move(app));
auto transport = Transport::make(fd, std::move(ws));
return net::socket_manager::make(mpx, fd, std::move(transport));
}
void abort(const error&) {
......@@ -80,8 +81,8 @@ auto make_accept_event_resources() {
/// @param on_request Function object for accepting incoming requests.
/// @param limit The maximum amount of connections before closing @p fd. Passing
/// 0 means "no limit".
template <template <class> class Transport = stream_transport, class Socket,
class... Ts, class OnRequest>
template <class Transport = stream_transport, class Socket, class... Ts,
class OnRequest>
void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
OnRequest on_request, size_t limit = 0) {
using trait_t = default_trait;
......@@ -95,8 +96,8 @@ void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
auto& mpx = sys.network_manager().mpx();
auto conn = std::make_shared<conn_t>(std::move(on_request), std::move(buf));
auto factory = factory_t{std::move(conn)};
auto ptr = make_socket_manager<impl_t>(std::move(fd), &mpx, limit,
std::move(factory));
auto impl = impl_t::make(fd, limit, std::move(factory));
auto ptr = socket_manager::make(&mpx, fd, std::move(impl));
mpx.init(ptr);
}
// TODO: accept() should return a disposable to stop the WebSocket server.
......
......@@ -13,12 +13,11 @@
#include "caf/net/http/v1.hpp"
#include "caf/net/message_flow_bridge.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/net/web_socket/framing.hpp"
#include "caf/net/web_socket/handshake.hpp"
#include "caf/pec.hpp"
#include "caf/settings.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
#include <algorithm>
......@@ -28,150 +27,71 @@ namespace caf::net::web_socket {
/// 6455. Initially, the layer performs the WebSocket handshake. Once completed,
/// this layer decodes RFC 6455 frames and forwards binary and text messages to
/// `UpperLayer`.
template <class UpperLayer>
class client {
class client : public stream_oriented::upper_layer {
public:
// -- member types -----------------------------------------------------------
using input_tag = tag::stream_oriented;
using handshake_ptr = std::unique_ptr<handshake>;
using output_tag = tag::mixed_message_oriented;
using upper_layer_ptr = std::unique_ptr<web_socket::upper_layer>;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
explicit client(handshake hs, Ts&&... xs)
: upper_layer_(std::forward<Ts>(xs)...) {
handshake_.reset(new handshake(std::move(hs)));
}
client(handshake_ptr hs, upper_layer_ptr up);
// -- initialization ---------------------------------------------------------
template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
CAF_ASSERT(handshake_ != nullptr);
owner_ = owner;
if (!handshake_->has_mandatory_fields())
return make_error(sec::runtime_error,
"handshake data lacks mandatory fields");
if (!handshake_->has_valid_key())
handshake_->randomize_key();
cfg_ = cfg;
down->begin_output();
handshake_->write_http_1_request(down->output_buffer());
down->end_output();
down->configure_read(receive_policy::up_to(handshake::max_http_size));
return none;
}
// -- factories --------------------------------------------------------------
static std::unique_ptr<client> make(handshake_ptr hs, upper_layer_ptr up);
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_.upper_layer();
web_socket::upper_layer& upper_layer() noexcept {
return framing_.upper_layer();
}
const auto& upper_layer() const noexcept {
return upper_layer_.upper_layer();
const web_socket::upper_layer& upper_layer() const noexcept {
return framing_.upper_layer();
}
// -- role: upper layer ------------------------------------------------------
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
return handshake_complete() ? upper_layer_.prepare_send(down) : true;
stream_oriented::lower_layer& lower_layer() noexcept {
return framing_.lower_layer();
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr down) {
return handshake_complete() ? upper_layer_.done_sending(down) : true;
const stream_oriented::lower_layer& lower_layer() const noexcept {
return framing_.lower_layer();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const error& reason) {
if (handshake_complete())
upper_layer_.abort(down, reason);
bool handshake_completed() const noexcept {
return hs_ == nullptr;
}
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr down) {
if (handshake_complete())
upper_layer_.continue_reading(down);
}
// -- implementation of stream_oriented::upper_layer -------------------------
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span input, byte_span delta) {
CAF_LOG_TRACE(CAF_ARG2("socket", down->handle().id)
<< CAF_ARG2("bytes", input.size()));
// Short circuit to the framing layer after the handshake completed.
if (handshake_complete())
return upper_layer_.consume(down, input, delta);
// Check whether received a HTTP header or else wait for more data or abort
// when exceeding the maximum size.
auto [hdr, remainder] = http::v1::split_header(input);
if (hdr.empty()) {
if (input.size() >= handshake::max_http_size) {
CAF_LOG_ERROR("server response exceeded maximum header size");
auto err = make_error(pec::too_many_characters,
"exceeded maximum header size");
down->abort_reason(std::move(err));
return -1;
} else {
return 0;
}
} else if (!handle_header(down, hdr)) {
return -1;
} else if (remainder.empty()) {
CAF_ASSERT(hdr.size() == input.size());
return hdr.size();
} else {
CAF_LOG_DEBUG(CAF_ARG2("socket", down->handle().id)
<< CAF_ARG2("remainder", remainder.size()));
if (auto sub_result = upper_layer_.consume(down, remainder, remainder);
sub_result >= 0) {
return hdr.size() + sub_result;
} else {
return sub_result;
}
}
}
error init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& config) override;
bool handshake_complete() const noexcept {
return handshake_ == nullptr;
}
void abort(const error& reason) override;
ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void continue_reading() override;
bool prepare_send() override;
bool done_sending() override;
private:
// -- HTTP response processing -----------------------------------------------
template <class LowerLayerPtr>
bool handle_header(LowerLayerPtr down, std::string_view http) {
CAF_ASSERT(handshake_ != nullptr);
auto http_ok = handshake_->is_valid_http_1_response(http);
handshake_.reset();
if (http_ok) {
if (auto err = upper_layer_.init(owner_, down, cfg_)) {
CAF_LOG_DEBUG("failed to initialize WebSocket framing layer");
down->abort_reason(std::move(err));
return false;
} else {
CAF_LOG_DEBUG("completed WebSocket handshake");
return true;
}
} else {
CAF_LOG_DEBUG("received invalid WebSocket handshake");
down->abort_reason(make_error(
sec::runtime_error,
"received invalid WebSocket handshake response from server"));
return false;
}
}
bool handle_header(std::string_view http);
// -- member variables -------------------------------------------------------
/// Stores the WebSocket handshake data until the handshake completed.
std::unique_ptr<handshake> handshake_;
handshake_ptr hs_;
/// Stores the upper layer.
framing<UpperLayer> upper_layer_;
framing framing_;
/// Stores a pointer to the owning manager for the delayed initialization.
socket_manager* owner_ = nullptr;
......@@ -179,17 +99,4 @@ private:
settings cfg_;
};
template <template <class> class Transport = stream_transport, class Socket,
class T, class Trait>
void run_client(multiplexer& mpx, Socket fd, handshake hs,
async::consumer_resource<T> in, async::producer_resource<T> out,
Trait trait) {
using app_t = message_flow_bridge<T, Trait, tag::mixed_message_oriented>;
using stack_t = Transport<client<app_t>>;
auto mgr = make_socket_manager<stack_t>(fd, &mpx, std::move(hs),
std::move(in), std::move(out),
std::move(trait));
mpx.init(mgr);
}
} // namespace caf::net::web_socket
// 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/async/publisher.hpp"
#include "caf/net/observer_adapter.hpp"
#include "caf/net/publisher_adapter.hpp"
namespace caf::net::web_socket::internal {
/// Implements a WebSocket application that uses two flows for bidirectional
/// communication: one input flow and one output flow.
template <class Reader, class Writer>
class bidir_app {
public:
using input_tag = tag::message_oriented;
using reader_output_type = typename Reader::value_type;
using writer_input_type = typename Writer::value_type;
bidir_app(Reader&& reader, Writer&& writer)
: reader_(std::move(reader)), writer_(std::move(writer)) {
// nop
}
async::publisher<reader_input_type>
connect_flows(net::socket_manager* mgr,
async::publisher<writer_input_type> in) {
using make_counted;
// Connect the writer adapter.
using writer_input_t = net::observer_adapter<writer_input_type>;
writer_input_ = make_counted<writer_input_t>(mgr);
in.subscribe(writer_input_->as_observer());
// Create the reader adapter.
using reader_output_t = net::publisher_adapter<node_message>;
reader_output_ = make_counted<reader_output_t>(mgr, reader_.buffer_size(),
reader_.batch_size());
return reader_output_->as_publisher();
}
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr&&, const settings& cfg) {
if (auto err = reader_.init(cfg))
return err;
if (auto err = writer_.init(cfg))
return err;
return none;
}
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
while (down->can_send_more()) {
auto [val, done, err] = writer_input_->poll();
if (val) {
if (!write(down, *val)) {
down->abort_reason(make_error(ec::invalid_message));
return false;
}
} else if (done) {
if (err) {
down->abort_reason(*err);
return false;
}
} else {
break;
}
}
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
return !writer_input_->has_data();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
reader_output_->flush();
if (reason == sec::socket_disconnected || reason == sec::disposed)
reader_output_->on_complete();
else
reader_output_->on_error(reason);
}
template <class LowerLayerPtr>
void after_reading(LowerLayerPtr) {
reader_output_->flush();
}
template <class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, std::string_view text) {
reader_msgput_type msg;
if (reader_.deserialize_text(text, msg)) {
if (reader_output_->push(std::move(msg)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(text.size());
} else {
down->abort_reason(make_error(ec::invalid_message));
return -1;
}
}
template <class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr, caf::byte_span bytes) {
reader_msgput_type msg;
if (reader_.deserialize_binary(bytes, msg)) {
if (reader_output_->push(std::move(msg)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(text.size());
} else {
down->abort_reason(make_error(ec::invalid_message));
return -1;
}
}
private:
template <class LowerLayerPtr>
bool write(LowerLayerPtr down, const writer_input_type& msg) {
if (writer_.is_text_message(msg)) {
down->begin_text_message();
if (writer_.serialize_text(msg, down->text_message_buffer())) {
down->end_text_message();
return true;
} else {
return false;
}
} else {
down->begin_binary_message();
if (writer_.serialize_binary(msg, down->binary_message_buffer())) {
down->end_binary_message();
return true;
} else {
return false;
}
}
}
/// Deserializes text or binary messages from the socket.
Reader reader_;
/// Serializes text or binary messages to the socket.
Writer writer_;
/// Forwards outgoing messages to the peer. We write whatever we receive from
/// this channel to the socket.
net::observer_adapter_ptr<node_message> writer_input_;
/// After receiving messages from the socket, we publish to this adapter for
/// downstream consumers.
net::publisher_adapter_ptr<node_message> reader_output_;
};
} // namespace caf::net::web_socket::internal
namespace caf::net::web_socket {
/// Connects to a WebSocket server for bidirectional communication.
/// @param sys The enclosing actor system.
/// @param cfg Provides optional configuration parameters such as WebSocket
/// protocols and extensions for the handshake.
/// @param locator Identifies the WebSocket server.
/// @param writer_input Publisher of events that go out to the server.
/// @param reader Reads messages from the server and publishes them locally.
/// @param writer Writes messages from the @p writer_input to text or binary
/// messages for sending them to the server.
/// @returns a publisher that makes messages from the server accessible on
/// success, an error otherwise.
template <template <class> class Transport = stream_transport, class Reader,
class Writer>
expected<async::publisher<typename Reader::value_type>>
flow_connect_bidir(actor_system& sys, const settings& cfg, uri locator,
async::publisher writer_input, Reader reader,
Writer writer) {
using stack_t
= Transport<web_socket::client<internal::bidir_app<Reader, Writer>>>;
using impl = socket_manager_impl<stack_t>;
if (locator.empty()) {
return make_error(sec::invalid_argument, __func__,
"cannot connect to empty URI");
} else if (locator.scheme() != "ws") {
return make_error(sec::invalid_argument, __func__,
"malformed URI, expected format 'ws://<authority>'");
} else if (!locator.fragment().empty()) {
return make_error(sec::invalid_argument, __func__,
"query and fragment components are not supported");
} else if (locator.authority().empty()) {
return make_error(sec::invalid_argument, __func__,
"malformed URI, expected format 'ws://<authority>'");
} else if (auto sock = impl::connect_to(locator.authority())) {
web_socket::handshake hs;
hs.host(to_string(locator.authority()));
if (locator.path().empty())
hs.endpoint("/");
else
hs.endpoint(to_string(locator.path()));
if (auto protocols = get_as<std::string>(cfg, "protocols"))
hs.protocols(std::move(*protocols));
if (auto extensions = get_as<std::string>(cfg, "extensions"))
hs.extensions(std::move(*extensions));
auto mgr = make_counted<impl>(*sock, sys.network_manager().mpx_ptr(),
std::move(hs), std::move(reader),
std::move(writer));
auto out = mgr->upper_layer().connect_flows(std::move(writer_input));
if (auto err = mgr->init(cfg); !err) {
return {std::move(out)};
} else {
return {std::move(err)};
}
} else {
return {std::move(sock.error())};
}
}
} // namespace caf::net::web_socket
......@@ -9,10 +9,10 @@
#include "caf/net/consumer_adapter.hpp"
#include "caf/net/producer_adapter.hpp"
#include "caf/net/web_socket/flow_connector.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/net/web_socket/upper_layer.hpp"
#include "caf/sec.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/no_auto_reading.hpp"
#include <utility>
......@@ -20,10 +20,8 @@ namespace caf::net::web_socket {
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public tag::no_auto_reading {
class flow_bridge : public web_socket::upper_layer {
public:
using input_tag = tag::mixed_message_oriented;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
......@@ -42,54 +40,34 @@ public:
// nop
}
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr, const settings& cfg) {
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
in_ = consumer_type::try_open(mgr, pull);
out_ = producer_type::try_open(mgr, push);
CAF_ASSERT(in_ != nullptr);
CAF_ASSERT(out_ != nullptr);
conn_ = nullptr;
return none;
} else {
conn_ = nullptr;
return err;
}
static std::unique_ptr<flow_bridge> make(connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(conn));
}
template <class LowerLayerPtr>
bool write(LowerLayerPtr down, const output_type& item) {
bool write(const output_type& item) {
if (trait_.converts_to_binary(item)) {
down->begin_binary_message();
auto& bytes = down->binary_message_buffer();
return trait_.convert(item, bytes) && down->end_binary_message();
down_->begin_binary_message();
auto& bytes = down_->binary_message_buffer();
return trait_.convert(item, bytes) && down_->end_binary_message();
} else {
down->begin_text_message();
auto& text = down->text_message_buffer();
return trait_.convert(item, text) && down->end_text_message();
down_->begin_text_message();
auto& text = down_->text_message_buffer();
return trait_.convert(item, text) && down_->end_text_message();
}
}
template <class LowerLayerPtr>
struct write_helper {
using bridge_type = flow_bridge;
bridge_type* bridge;
LowerLayerPtr down;
flow_bridge* thisptr;
bool aborted = false;
size_t consumed = 0;
error err;
write_helper(bridge_type* bridge, LowerLayerPtr down)
: bridge(bridge), down(down) {
explicit write_helper(flow_bridge* thisptr) : thisptr(thisptr) {
// nop
}
void on_next(const output_type& item) {
if (!bridge->write(down, item)) {
if (!thisptr->write(item))
aborted = true;
return;
}
}
void on_complete() {
......@@ -101,20 +79,42 @@ public:
}
};
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
write_helper<LowerLayerPtr> helper{this, down};
while (down->can_send_more() && in_) {
// -- implementation of web_socket::lower_layer ------------------------------
error init(net::socket_manager* mgr, web_socket::lower_layer* down,
const settings& cfg) override {
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
in_ = consumer_type::try_open(mgr, pull);
out_ = producer_type::try_open(mgr, push);
CAF_ASSERT(in_ != nullptr);
CAF_ASSERT(out_ != nullptr);
conn_ = nullptr;
return none;
} else {
conn_ = nullptr;
return err;
}
}
void continue_reading() override {
// nop
}
bool prepare_send() override {
write_helper helper{this};
while (down_->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down->send_close_message(helper.err);
down_->send_close_message(helper.err);
} else {
down->send_close_message();
down_->send_close_message();
}
in_ = nullptr;
return false;
} else if (helper.aborted) {
down->abort_reason(make_error(sec::conversion_failed));
in_->cancel();
in_ = nullptr;
return false;
......@@ -125,13 +125,11 @@ public:
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
bool done_sending() override {
return !in_ || !in_->has_data();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::socket_disconnected || reason == sec::disposed)
......@@ -146,39 +144,31 @@ public:
}
}
template <class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr down, byte_span buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
ptrdiff_t consume_binary(byte_span buf) override {
if (!out_)
return -1;
}
input_type val;
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
if (!trait_.convert(buf, val))
return -1;
}
if (out_->push(std::move(val)) == 0)
down->suspend_reading();
down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
template <class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, std::string_view buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
ptrdiff_t consume_text(std::string_view buf) override {
if (!out_)
return -1;
}
input_type val;
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
if (!trait_.convert(buf, val))
return -1;
}
if (out_->push(std::move(val)) == 0)
down->suspend_reading();
down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
private:
web_socket::lower_layer* down_;
/// The output of the application. Serialized to the socket.
intrusive_ptr<consumer_type> in_;
......
This diff is collapsed.
......@@ -18,6 +18,12 @@ class flow_bridge;
template <class Trait, class... Ts>
class request;
class CAF_CORE_EXPORT frame;
class frame;
class framing;
class lower_layer;
class server;
class upper_layer;
enum class status : uint16_t;
} // namespace caf::net::web_socket
// 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/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/web_socket/fwd.hpp"
#include <string_view>
namespace caf::net::web_socket {
// TODO: Documentation.
class lower_layer {
public:
virtual ~lower_layer();
virtual bool can_send_more() = 0;
virtual void suspend_reading() = 0;
virtual void begin_binary_message() = 0;
virtual byte_buffer& binary_message_buffer() = 0;
virtual bool end_binary_message() = 0;
virtual void begin_text_message() = 0;
virtual text_buffer& text_message_buffer() = 0;
virtual bool end_text_message() = 0;
virtual void send_close_message() = 0;
virtual void send_close_message(status code, std::string_view desc) = 0;
void send_close_message(const error& reason);
};
} // namespace caf::net::web_socket
This diff is collapsed.
......@@ -4,26 +4,25 @@
#pragma once
namespace caf::detail {
#include "caf/net/fwd.hpp"
#include "caf/net/web_socket/fwd.hpp"
template <class T, class LowerLayerPtr>
class has_after_reading {
private:
template <class A, class B>
static auto sfinae(A& up, B& ptr)
-> decltype(up.after_reading(ptr), std::true_type{});
template <class A>
static std::false_type sfinae(A&, ...);
using sfinae_result
= decltype(sfinae(std::declval<T&>(), std::declval<LowerLayerPtr&>()));
namespace caf::net::web_socket {
// TODO: Documentation.
class upper_layer {
public:
static constexpr bool value = sfinae_result::value;
virtual ~upper_layer();
virtual error
init(net::socket_manager* mgr, lower_layer* down, const settings& cfg)
= 0;
virtual bool prepare_send() = 0;
virtual bool done_sending() = 0;
virtual void continue_reading() = 0;
virtual void abort(const error& reason) = 0;
virtual ptrdiff_t consume_binary(byte_span buf) = 0;
virtual ptrdiff_t consume_text(std::string_view buf) = 0;
};
template <class T, class LowerLayerPtr>
constexpr bool has_after_reading_v = has_after_reading<T, LowerLayerPtr>::value;
} // namespace caf::detail
} // namespace caf::net::web_socket
// 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
namespace caf::tag {
/// Tells message-oriented transports to *not* implicitly register an
/// application for reading as part of the initialization when used a base type.
struct no_auto_reading {};
} // namespace caf::tag
// 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
namespace caf::tag {
struct stream_oriented {};
} // namespace caf::tag
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -2,10 +2,12 @@
// 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/net/socket_event_layer.hpp"
namespace caf::tag {
namespace caf::net {
struct message_oriented {};
socket_event_layer::~socket_event_layer() {
// nop
}
} // namespace caf::tag
} // namespace caf::net
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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