Commit 1e7adca3 authored by Dominik Charousset's avatar Dominik Charousset

Implement length_prefix_framing::accept

parent a5136c64
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#pragma once #pragma once
#include "caf/async/blocking_producer.hpp" #include "caf/async/blocking_producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/net/web_socket/request.hpp" #include "caf/net/web_socket/request.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
...@@ -27,6 +28,11 @@ public: ...@@ -27,6 +28,11 @@ public:
using result_type = std::tuple<error, pull_t, push_t>; using result_type = std::tuple<error, pull_t, push_t>;
using connect_event_type = cow_tuple<async::consumer_resource<output_type>,
async::producer_resource<input_type>>;
using connect_event_buf = async::spsc_buffer_ptr<connect_event_type>;
virtual ~flow_connector() { virtual ~flow_connector() {
// nop // nop
} }
...@@ -36,14 +42,22 @@ public: ...@@ -36,14 +42,22 @@ public:
/// Returns a trivial implementation that simply returns @p pull and @p push /// Returns a trivial implementation that simply returns @p pull and @p push
/// from @p on_request. /// from @p on_request.
static flow_connector_ptr<Trait> make_trivial(pull_t pull, push_t push); static flow_connector_ptr<Trait> make_trivial(pull_t pull, push_t push);
/// Returns an implementation for a basic server that simply creates connected
/// buffer pairs.
static flow_connector_ptr<Trait> make_basic_server(connect_event_buf buf);
}; };
} // namespace caf::net
namespace caf::detail {
/// Trivial flow connector that passes its constructor arguments to the /// Trivial flow connector that passes its constructor arguments to the
/// @ref flow_bridge. /// @ref flow_bridge.
template <class Trait> template <class Trait>
class flow_connector_trivial_impl : public flow_connector<Trait> { class flow_connector_trivial_impl : public net::flow_connector<Trait> {
public: public:
using super = flow_connector<Trait>; using super = net::flow_connector<Trait>;
using input_type = typename Trait::input_type; using input_type = typename Trait::input_type;
...@@ -69,11 +83,56 @@ private: ...@@ -69,11 +83,56 @@ private:
push_t push_; push_t push_;
}; };
/// A flow connector for basic servers that have no custom handshake logic.
template <class Trait>
class flow_connector_basic_server_impl : public net::flow_connector<Trait> {
public:
using super = net::flow_connector<Trait>;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using result_type = typename super::result_type;
using connect_event_buf = typename super::connect_event_buf;
using connect_event = typename super::connect_event_type;
using producer_type = async::blocking_producer<connect_event>;
explicit flow_connector_basic_server_impl(connect_event_buf buf)
: prod_(std::move(buf)) {
// nop
}
result_type on_request(const settings&) override {
auto [app_pull, srv_push] = async::make_spsc_buffer_resource<input_type>();
auto [srv_pull, app_push] = async::make_spsc_buffer_resource<output_type>();
prod_.push(connect_event{std::move(srv_pull), std::move(srv_push)});
return {{}, std::move(app_pull), std::move(app_push)};
}
private:
producer_type prod_;
};
} // namespace caf::detail
namespace caf::net {
template <class Trait> template <class Trait>
flow_connector_ptr<Trait> flow_connector_ptr<Trait>
flow_connector<Trait>::make_trivial(pull_t pull, push_t push) { flow_connector<Trait>::make_trivial(pull_t pull, push_t push) {
using impl_t = flow_connector_trivial_impl<Trait>; using impl_t = detail::flow_connector_trivial_impl<Trait>;
return std::make_shared<impl_t>(std::move(pull), std::move(push)); return std::make_shared<impl_t>(std::move(pull), std::move(push));
} }
template <class Trait>
flow_connector_ptr<Trait>
flow_connector<Trait>::make_basic_server(connect_event_buf buf) {
using impl_t = detail::flow_connector_basic_server_impl<Trait>;
return std::make_shared<impl_t>(std::move(buf));
}
} // namespace caf::net } // namespace caf::net
...@@ -8,10 +8,13 @@ ...@@ -8,10 +8,13 @@
#include "caf/async/spsc_buffer.hpp" #include "caf/async/spsc_buffer.hpp"
#include "caf/byte_span.hpp" #include "caf/byte_span.hpp"
#include "caf/cow_tuple.hpp" #include "caf/cow_tuple.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/net/binary/default_trait.hpp" #include "caf/net/binary/default_trait.hpp"
#include "caf/net/binary/flow_bridge.hpp" #include "caf/net/binary/flow_bridge.hpp"
#include "caf/net/binary/frame.hpp"
#include "caf/net/binary/lower_layer.hpp" #include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp" #include "caf/net/binary/upper_layer.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
...@@ -62,18 +65,16 @@ public: ...@@ -62,18 +65,16 @@ public:
/// @param sys The host system. /// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on the /// @param conn A connected stream socket or SSL connection, depending on the
/// `Transport`. /// `Transport`.
/// @param init Function object for setting up the created flows. /// @param pull Source for pulling data to send.
template <class Transport = stream_transport, class Connection, class Init> /// @param push Source for pushing received data.
static disposable run(actor_system& sys, Connection conn, Init init) { template <class Transport = stream_transport, class Connection>
static disposable run(actor_system& sys, Connection conn,
async::consumer_resource<binary::frame> pull,
async::producer_resource<binary::frame> push) {
using trait_t = binary::default_trait; using trait_t = binary::default_trait;
using frame_t = binary::frame;
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
auto [fd_pull, app_push] = async::make_spsc_buffer_resource<frame_t>();
auto [app_pull, fd_push] = async::make_spsc_buffer_resource<frame_t>();
auto mpx = sys.network_manager().mpx_ptr(); auto mpx = sys.network_manager().mpx_ptr();
auto fc = flow_connector<trait_t>::make_trivial(std::move(fd_pull), auto fc = flow_connector<trait_t>::make_trivial(std::move(pull),
std::move(fd_push)); std::move(push));
auto bridge = binary::flow_bridge<trait_t>::make(mpx, std::move(fc)); auto bridge = binary::flow_bridge<trait_t>::make(mpx, std::move(fc));
auto bridge_ptr = bridge.get(); auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge)); auto impl = length_prefix_framing::make(std::move(bridge));
...@@ -81,10 +82,27 @@ public: ...@@ -81,10 +82,27 @@ public:
auto ptr = socket_manager::make(mpx, std::move(transport)); auto ptr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable()); bridge_ptr->self_ref(ptr->as_disposable());
mpx->start(ptr); mpx->start(ptr);
init(connect_event_t{app_pull, app_push});
return disposable{std::move(ptr)}; return disposable{std::move(ptr)};
} }
/// Runs length-prefix framing on given connection.
/// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on the
/// `Transport`.
/// @param init Function object for setting up the created flows.
template <class Transport = stream_transport, class Connection, class Init>
static disposable run(actor_system& sys, Connection conn, Init init) {
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
using frame_t = binary::frame;
auto [fd_pull, app_push] = async::make_spsc_buffer_resource<frame_t>();
auto [app_pull, fd_push] = async::make_spsc_buffer_resource<frame_t>();
auto result = run(sys, std::move(conn), std::move(fd_pull),
std::move(fd_push));
init(connect_event_t{std::move(app_pull), std::move(app_push)});
return result;
}
/// The default number of concurrently open connections when using `accept`. /// The default number of concurrently open connections when using `accept`.
static constexpr size_t default_max_connections = 128; static constexpr size_t default_max_connections = 128;
...@@ -92,15 +110,47 @@ public: ...@@ -92,15 +110,47 @@ public:
/// represented by two buffers: one for input and one for output. /// represented by two buffers: one for input and one for output.
using acceptor_resource_t = async::producer_resource<connect_event_t>; using acceptor_resource_t = async::producer_resource<connect_event_t>;
/// Describes the per-connection event.
using accept_event_t
= cow_tuple<async::consumer_resource<binary::frame>, // Socket to App.
async::producer_resource<binary::frame>>; // App to Socket.
/// Convenience function for creating an event listener resource and an
/// @ref acceptor_resource_t via @ref async::make_spsc_buffer_resource.
template <class... Ts>
static auto make_accept_event_resources() {
return async::make_spsc_buffer_resource<accept_event_t>();
}
/// Listens for incoming connection on @p fd. /// Listens for incoming connection on @p fd.
/// @param sys The host system. /// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this /// @param acc A connection acceptor such as @ref tcp_accept_socket or
/// socket must already listen to a port. /// @ref ssl::acceptor.
/// @param cfg Configures the acceptor. Currently, the only supported /// @param cfg Configures the acceptor. Currently, the only supported
/// configuration parameter is `max-connections`. /// configuration parameter is `max-connections`.
template <class Transport = stream_transport, class Socket, class OnConnect> template <class Acceptor>
static disposable accept(actor_system& sys, Socket fd, static disposable accept(actor_system& sys, Acceptor acc,
acceptor_resource_t out, const settings& cfg = {}) { acceptor_resource_t out, const settings& cfg = {}) {
using transport_t = typename Acceptor::transport_type;
using trait_t = binary::default_trait;
using factory_t = cf_impl<transport_t>;
using conn_t = typename transport_t::connection_handle;
using impl_t = detail::accept_handler<Acceptor, conn_t>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto conn = flow_connector<trait_t>::make_basic_server(std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(conn));
auto impl = impl_t::make(std::move(acc), std::move(factory),
max_connections);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(&mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
mpx.start(ptr);
return disposable{std::move(ptr)};
} else {
return {};
}
} }
// -- implementation of stream_oriented::upper_layer ------------------------- // -- implementation of stream_oriented::upper_layer -------------------------
...@@ -141,6 +191,38 @@ public: ...@@ -141,6 +191,38 @@ public:
static std::pair<size_t, byte_span> split(byte_span buffer) noexcept; static std::pair<size_t, byte_span> split(byte_span buffer) noexcept;
private: private:
// -- helper classes ---------------------------------------------------------
template <class Transport>
class cf_impl
: public detail::connection_factory<typename Transport::connection_handle> {
public:
using connection_handle = typename Transport::connection_handle;
using connector_ptr = flow_connector_ptr<binary::default_trait>;
explicit cf_impl(connector_ptr connector) : conn_(std::move(connector)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
using trait_t = binary::default_trait;
auto bridge = binary::flow_bridge<trait_t>::make(mpx, conn_);
auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge));
auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(impl));
transport->active_policy().accept(fd);
auto mgr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(mgr->as_disposable());
return mgr;
}
private:
connector_ptr conn_;
};
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
stream_oriented::lower_layer* down_; stream_oriented::lower_layer* down_;
......
...@@ -22,7 +22,7 @@ template <class Transport, class Connection> ...@@ -22,7 +22,7 @@ template <class Transport, class Connection>
ws::connect_state ws::connect_state
ws_do_connect_impl(actor_system& sys, Connection conn, ws::handshake& hs) { ws_do_connect_impl(actor_system& sys, Connection conn, ws::handshake& hs) {
using trait_t = ws::default_trait; using trait_t = ws::default_trait;
using connector_t = net::flow_connector_trivial_impl<trait_t>; using connector_t = flow_connector_trivial_impl<trait_t>;
auto [ws_pull, app_push] = async::make_spsc_buffer_resource<ws::frame>(); auto [ws_pull, app_push] = async::make_spsc_buffer_resource<ws::frame>();
auto [app_pull, ws_push] = async::make_spsc_buffer_resource<ws::frame>(); auto [app_pull, ws_push] = async::make_spsc_buffer_resource<ws::frame>();
auto mpx = sys.network_manager().mpx_ptr(); auto mpx = sys.network_manager().mpx_ptr();
......
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