Commit 697dba79 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1357'

Close #1357.
parents c7edd72b eccd681f
......@@ -142,6 +142,9 @@ endif()
add_net_example(http secure-time-server)
add_net_example(http time-server)
add_net_example(length_prefix_framing chat-client)
add_net_example(length_prefix_framing chat-server)
add_net_example(web_socket echo)
add_net_example(web_socket hello-client)
add_net_example(web_socket quote-server)
......
// Simple chat server with a binary protocol.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/length_prefix_framing.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/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
#include "caf/uuid.hpp"
#include <cassert>
#include <iostream>
#include <utility>
// -- convenience type aliases -------------------------------------------------
// Takes care converting a byte stream into a sequence of messages on the wire.
using lpf = caf::net::length_prefix_framing;
// An implicitly shared type for storing a binary frame.
using bin_frame = caf::net::binary::frame;
// Each client gets a UUID for identifying it. While processing messages, we add
// this ID to the input to tag it.
using message_t = std::pair<caf::uuid, bin_frame>;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 7788;
static constexpr std::string_view default_host = "localhost";
// -- configuration setup ------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port of the server")
.add<std::string>("host,H", "host of the server")
.add<std::string>("name,n", "set name");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
// Connect to the server.
auto port = caf::get_or(cfg, "port", default_port);
auto host = caf::get_or(cfg, "host", default_host);
auto name = caf::get_or(cfg, "name", "");
if (name.empty()) {
std::cerr << "*** mandatory parameter 'name' missing or empty\n";
return EXIT_FAILURE;
}
auto fd = caf::net::make_connected_tcp_stream_socket(host, port);
if (!fd) {
std::cerr << "*** unable to connect to " << host << ":" << port << ": "
<< to_string(fd.error()) << '\n';
return EXIT_FAILURE;
}
std::cout << "*** connected to " << host << ":" << port << ": " << '\n';
// Create our buffers that connect the worker to the socket.
using caf::async::make_spsc_buffer_resource;
auto [lpf_pull, app_push] = make_spsc_buffer_resource<bin_frame>();
auto [app_pull, lpf_push] = make_spsc_buffer_resource<bin_frame>();
// Spin up the network backend.
lpf::run(sys, *fd, std::move(lpf_pull), std::move(lpf_push));
// Spin up a worker that simply prints received inputs.
sys.spawn([pull = std::move(app_pull)](caf::event_based_actor* self) {
pull
.observe_on(self) //
.do_finally([self] {
std::cout << "*** lost connection to server -> quit\n"
<< "*** use CTRL+D or CTRL+C to terminate\n";
self->quit();
})
.for_each([](const bin_frame& frame) {
// Interpret the bytes as ASCII characters.
auto bytes = frame.bytes();
auto str = std::string_view{reinterpret_cast<const char*>(bytes.data()),
bytes.size()};
if (std::all_of(str.begin(), str.end(), ::isprint)) {
std::cout << str << '\n';
} else {
std::cout << "<non-ascii-data of size " << bytes.size() << ">\n";
}
});
});
// Wait for user input on std::cin and send them to the server.
auto push_buf = app_push.try_open();
assert(push_buf != nullptr);
auto inputs = caf::async::blocking_producer<bin_frame>{std::move(push_buf)};
auto line = std::string{};
auto prefix = name + ": ";
while (std::getline(std::cin, line)) {
line.insert(line.begin(), prefix.begin(), prefix.end());
inputs.push(bin_frame{caf::as_bytes(caf::make_span(line))});
line.clear();
}
// Done. However, the actor system will keep the application running for as
// long as it has open ports or connections.
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
// Simple chat server with a binary protocol.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/length_prefix_framing.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/scheduled_actor/flow.hpp"
#include "caf/uuid.hpp"
#include <iostream>
#include <utility>
// -- convenience type aliases -------------------------------------------------
// Takes care converting a byte stream into a sequence of messages on the wire.
using lpf = caf::net::length_prefix_framing;
// An implicitly shared type for storing a binary frame.
using bin_frame = caf::net::binary::frame;
// Each client gets a UUID for identifying it. While processing messages, we add
// this ID to the input to tag it.
using message_t = std::pair<caf::uuid, bin_frame>;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 7788;
// -- configuration setup ------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
}
};
// -- multiplexing logic -------------------------------------------------------
void worker_impl(caf::event_based_actor* self,
caf::async::consumer_resource<lpf::accept_event_t> events) {
// Allows us to push new flows into the central merge point.
caf::flow::item_publisher<caf::flow::observable<message_t>> msg_pub{self};
// Our central merge point combines all inputs into a single, shared flow.
auto messages = msg_pub.as_observable().merge().share();
// Have one subscription for debug output. This also makes sure that the
// shared observable stays subscribed to the merger.
messages.for_each([](const message_t& msg) {
const auto& [conn, frame] = msg;
std::cout << "*** got message of size " << frame.size() << " from "
<< to_string(conn) << '\n';
});
// Connect the flows for each incoming connection.
events
.observe_on(self) //
.for_each(
[self, messages, pub = std::move(msg_pub)] //
(const lpf::accept_event_t& event) mutable {
// Each connection gets a unique ID.
auto conn = caf::uuid::random();
std::cout << "*** accepted new connection " << to_string(conn) << '\n';
auto& [pull, push] = event.data();
// Subscribe the `push` end to the central merge point.
messages
.filter([conn](const message_t& msg) {
// Drop all messages received by this conn.
return msg.first != conn;
})
.map([](const message_t& msg) {
// Remove the server-internal UUID.
return msg.second;
})
.subscribe(push);
// Feed messages from the `pull` end into the central merge point.
auto inputs = pull.observe_on(self)
.on_error_complete() // Cary on if a connection breaks.
.do_on_complete([conn] {
std::cout << "*** lost connection " << to_string(conn)
<< '\n';
})
.map([conn](const bin_frame& frame) {
return message_t{conn, frame};
})
.as_observable();
pub.push(inputs);
});
}
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
// Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port);
caf::net::tcp_accept_socket fd;
if (auto maybe_fd = caf::net::make_tcp_accept_socket(port)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
return EXIT_FAILURE;
}
// Create buffers to signal events from the caf.net backend to the worker.
auto [wres, sres] = lpf::make_accept_event_resources();
// Spin up a worker to multiplex the messages.
auto worker = sys.spawn(worker_impl, std::move(wres));
// Set everything in motion.
lpf::accept(sys, fd, std::move(sres));
// Done. However, the actor system will keep the application running for as
// long as actors are still alive and for as long as it has open ports or
// connections. Since we never close the accept socket, this means the server
// is running indefinitely until the process gets killed (e.g., via CTRL+C).
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -130,10 +130,12 @@ public:
void fwd_on_next(input_key key, const observable<T>& item) {
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(item));
if (auto ptr = get(key)) {
if (auto ptr = get(key))
subscribe_to(item);
// Note: we need to double-check that the key still exists here, because
// subscribe_on may result in an error (that nukes all inputs).
if (auto ptr = get(key))
ptr->sub.request(1);
}
}
// -- implementation of subscription_impl ------------------------------------
......
......@@ -5,6 +5,7 @@
#pragma once
#include "caf/async/blocking_producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/error.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/sec.hpp"
......@@ -27,6 +28,11 @@ public:
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() {
// nop
}
......@@ -36,14 +42,22 @@ public:
/// Returns a trivial implementation that simply returns @p pull and @p push
/// from @p on_request.
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
/// @ref flow_bridge.
template <class Trait>
class flow_connector_trivial_impl : public flow_connector<Trait> {
class flow_connector_trivial_impl : public net::flow_connector<Trait> {
public:
using super = flow_connector<Trait>;
using super = net::flow_connector<Trait>;
using input_type = typename Trait::input_type;
......@@ -69,11 +83,56 @@ private:
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>
flow_connector_ptr<Trait>
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));
}
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
......@@ -8,10 +8,13 @@
#include "caf/async/spsc_buffer.hpp"
#include "caf/byte_span.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/disposable.hpp"
#include "caf/net/binary/default_trait.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/upper_layer.hpp"
#include "caf/net/middleman.hpp"
......@@ -62,18 +65,16 @@ public:
/// @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) {
/// @param pull Source for pulling data to send.
/// @param push Source for pushing received data.
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 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 fc = flow_connector<trait_t>::make_trivial(std::move(fd_pull),
std::move(fd_push));
auto fc = flow_connector<trait_t>::make_trivial(std::move(pull),
std::move(push));
auto bridge = binary::flow_bridge<trait_t>::make(mpx, std::move(fc));
auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge));
......@@ -81,10 +82,27 @@ public:
auto ptr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable());
mpx->start(ptr);
init(connect_event_t{app_pull, app_push});
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`.
static constexpr size_t default_max_connections = 128;
......@@ -92,15 +110,47 @@ public:
/// represented by two buffers: one for input and one for output.
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.
/// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this
/// socket must already listen to a port.
/// @param acc A connection acceptor such as @ref tcp_accept_socket or
/// @ref ssl::acceptor.
/// @param cfg Configures the acceptor. Currently, the only supported
/// configuration parameter is `max-connections`.
template <class Transport = stream_transport, class Socket, class OnConnect>
static disposable accept(actor_system& sys, Socket fd,
template <class Acceptor>
static disposable accept(actor_system& sys, Acceptor acc,
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 -------------------------
......@@ -141,6 +191,38 @@ public:
static std::pair<size_t, byte_span> split(byte_span buffer) noexcept;
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 -------------------------------------------------------
stream_oriented::lower_layer* down_;
......
......@@ -15,6 +15,10 @@ namespace caf::net::ssl {
/// Wraps an accept socket and an SSL context.
class CAF_NET_EXPORT acceptor {
public:
// -- member types -----------------------------------------------------------
using transport_type = transport;
// -- constructors, destructors, and assignment operators --------------------
acceptor() = delete;
......
......@@ -18,6 +18,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
using super = network_socket;
using super::super;
using transport_type = stream_transport;
};
/// Creates a new TCP socket to accept connections on a given port.
......
......@@ -77,13 +77,29 @@ private:
connector_pointer connector_;
};
template <class Transport, class Acceptor, class... Ts, class OnRequest>
disposable ws_accept_impl(actor_system& sys, Acceptor acc,
net::web_socket::acceptor_resource_t<Ts...> out,
OnRequest on_request, const settings& cfg) {
} // namespace caf::detail
namespace caf::net::web_socket {
/// Listens for incoming WebSocket connections.
/// @param sys The host system.
/// @param acc A connection acceptor such as @ref tcp_accept_socket or
/// @ref ssl::acceptor.
/// @param out A buffer resource that connects the server to a listener that
/// processes the buffer pairs for each incoming connection.
/// @param on_request Function object for accepting incoming requests.
/// @param cfg Configuration parameters for the acceptor.
template <class Acceptor, class... Ts, class OnRequest>
disposable accept(actor_system& sys, Acceptor acc,
acceptor_resource_t<Ts...> out, OnRequest on_request,
const settings& cfg = {}) {
using transport_t = typename Acceptor::transport_type;
using request_t = request<default_trait, Ts...>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request");
using trait_t = net::web_socket::default_trait;
using factory_t = detail::ws_conn_factory<Transport, trait_t>;
using conn_t = typename Transport::connection_handle;
using factory_t = detail::ws_conn_factory<transport_t, trait_t>;
using conn_t = typename transport_t::connection_handle;
using impl_t = detail::accept_handler<Acceptor, conn_t>;
using connector_t
= net::web_socket::flow_connector_request_impl<OnRequest, trait_t, Ts...>;
......@@ -105,44 +121,4 @@ disposable ws_accept_impl(actor_system& sys, Acceptor acc,
}
}
} // namespace caf::detail
namespace caf::net::web_socket {
/// Listens for incoming WebSocket connections.
/// @param sys The host system.
/// @param fd An accept socket in listening mode, already bound to a port.
/// @param out A buffer resource that connects the server to a listener that
/// processes the buffer pairs for each incoming connection.
/// @param on_request Function object for accepting incoming requests.
/// @param cfg Configuration parameters for the acceptor.
template <class... Ts, class OnRequest>
disposable accept(actor_system& sys, tcp_accept_socket fd,
acceptor_resource_t<Ts...> out, OnRequest on_request,
const settings& cfg = {}) {
using request_t = request<default_trait, Ts...>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request");
return detail::ws_accept_impl<stream_transport>(sys, fd, out,
std::move(on_request), cfg);
}
/// Listens for incoming WebSocket connections over TLS.
/// @param sys The host system.
/// @param acc An SSL connection acceptor with a socket that in listening mode.
/// @param out A buffer resource that connects the server to a listener that
/// processes the buffer pairs for each incoming connection.
/// @param on_request Function object for accepting incoming requests.
/// @param cfg Configuration parameters for the acceptor.
template <class... Ts, class OnRequest>
disposable accept(actor_system& sys, ssl::acceptor acc,
acceptor_resource_t<Ts...> out, OnRequest on_request,
const settings& cfg = {}) {
using request_t = request<default_trait, Ts...>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request");
return detail::ws_accept_impl<ssl::transport>(sys, std::move(acc), out,
std::move(on_request), cfg);
}
} // namespace caf::net::web_socket
......@@ -22,7 +22,7 @@ template <class Transport, class Connection>
ws::connect_state
ws_do_connect_impl(actor_system& sys, Connection conn, ws::handshake& hs) {
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 [app_pull, ws_push] = async::make_spsc_buffer_resource<ws::frame>();
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