Commit 5c897919 authored by Dominik Charousset's avatar Dominik Charousset

Implement new factories for lenght-prefix framing

parent 39ad3c4f
......@@ -4,7 +4,7 @@
#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/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
......@@ -23,9 +23,6 @@
// binary default trait operates on binary::frame items.
using trait = caf::net::binary::default_trait;
// Takes care converting a byte stream into a sequence of messages on the wire.
using lpf = caf::net::length_prefix_framing::bind<trait>;
// An implicitly shared type for storing a binary frame.
using bin_frame = caf::net::binary::frame;
......@@ -53,7 +50,8 @@ struct config : caf::actor_system_config {
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
// Connect to the server.
// Read the configuration.
bool had_error = false;
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", "");
......@@ -61,54 +59,54 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
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";
// Connect to the server.
caf::net::lp::with(sys)
.connect(host, port)
.do_on_error([&](const caf::error& what) {
std::cerr << "*** unable to connect to " << host << ":" << port << ": "
<< to_string(what) << '\n';
had_error = true;
})
.start([&sys, name](auto pull, auto push) {
// Spin up a worker that prints received inputs.
sys.spawn([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";
}
});
});
// Spin up a second worker that reads from std::cin and sends each line to
// the server. Put that to its own thread since it's doing I/O.
sys.spawn<caf::detached>([push, name] {
auto lines = caf::async::make_blocking_producer(push);
if (!lines)
throw std::logic_error("failed to create blocking producer");
auto line = std::string{};
auto prefix = name + ": ";
while (std::getline(std::cin, line)) {
line.insert(line.begin(), prefix.begin(), prefix.end());
lines->push(bin_frame{caf::as_bytes(caf::make_span(line))});
line.clear();
}
});
});
// 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;
});
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return had_error ? EXIT_FAILURE : EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -4,7 +4,7 @@
#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/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
......@@ -21,9 +21,6 @@
// binary default trait operates on binary::frame items.
using trait = caf::net::binary::default_trait;
// Takes care converting a byte stream into a sequence of messages on the wire.
using lpf = caf::net::length_prefix_framing::bind<trait>;
// An implicitly shared type for storing a binary frame.
using bin_frame = caf::net::binary::frame;
......@@ -47,7 +44,7 @@ struct config : caf::actor_system_config {
// -- multiplexing logic -------------------------------------------------------
void worker_impl(caf::event_based_actor* self,
caf::async::consumer_resource<lpf::accept_event_t> events) {
trait::acceptor_resource 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.
......@@ -64,7 +61,7 @@ void worker_impl(caf::event_based_actor* self,
.observe_on(self) //
.for_each(
[self, messages, pub = std::move(msg_pub)] //
(const lpf::accept_event_t& event) mutable {
(const trait::accept_event& event) mutable {
// Each connection gets a unique ID.
auto conn = caf::uuid::random();
std::cout << "*** accepted new connection " << to_string(conn) << '\n';
......@@ -98,29 +95,22 @@ void worker_impl(caf::event_based_actor* self,
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
// Open up a TCP port for incoming connections.
// Open up a TCP port for incoming connections and start the server.
auto had_error = false;
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::net::lp::with(sys)
.accept(port)
.do_on_error([&](const caf::error& what) {
std::cerr << "*** unable to open port " << port << ": " << to_string(what)
<< '\n';
had_error = true;
})
.start([&sys](trait::acceptor_resource accept_events) {
sys.spawn(worker_impl, std::move(accept_events));
});
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return had_error ? EXIT_FAILURE : EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -18,7 +18,7 @@
#include <vector>
#include "caf/all.hpp"
#include "caf/net/length_prefix_framing.hpp"
#include "caf/net/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
......@@ -54,7 +54,7 @@ public:
// -- main ---------------------------------------------------------------------
int caf_main(actor_system& sys, const config& cfg) {
// Connect to the server.
// Read the configuration.
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", "");
......@@ -62,22 +62,6 @@ int caf_main(actor_system& sys, const config& cfg) {
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 bin_frame = caf::net::binary::frame;
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.
using trait = caf::net::binary::default_trait;
using lpf = caf::net::length_prefix_framing::bind<trait>;
auto conn = lpf::run(sys, *fd, std::move(lpf_pull), std::move(lpf_push));
// Spin up Qt.
auto [argc, argv] = cfg.c_args_remainder();
QApplication app{argc, argv};
......@@ -85,9 +69,25 @@ int caf_main(actor_system& sys, const config& cfg) {
QMainWindow mw;
Ui::ChatWindow helper;
helper.setupUi(&mw);
helper.chatwidget->init(sys, name, std::move(app_pull), std::move(app_push));
// Connect to the server.
auto had_error = false;
auto conn
= caf::net::lp::with(sys)
.connect(host, port)
.do_on_error([&](const caf::error& what) {
std::cerr << "*** unable to connect to " << host << ":" << port
<< ": " << to_string(what) << '\n';
had_error = true;
})
.start([&](auto pull, auto push) {
std::cout << "*** connected to " << host << ":" << port << '\n';
helper.chatwidget->init(sys, name, std::move(pull), std::move(push));
});
if (had_error) {
mw.close();
return app.exec();
}
// Setup and run.
auto client = helper.chatwidget->as_actor();
mw.show();
auto result = app.exec();
conn.dispose();
......
......@@ -461,6 +461,11 @@ public:
visit_family(f, ptr.get());
}
// -- static utility functions -----------------------------------------------
/// Returns a pointer to the metric registry from the actor system.
static metric_registry* from(actor_system& sys);
// -- modifiers --------------------------------------------------------------
/// Takes ownership of all metric families in `other`.
......
......@@ -47,6 +47,9 @@ public:
// nop
}
/// Returns the `host` as string.
std::string host_str() const;
/// Returns whether `host` is empty, i.e., the host is not an IP address
/// and the string is empty.
bool empty() const noexcept {
......
......@@ -4,6 +4,7 @@
#include "caf/telemetry/metric_registry.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/config.hpp"
#include "caf/raise_error.hpp"
......@@ -36,6 +37,10 @@ metric_registry::~metric_registry() {
// nop
}
metric_registry* metric_registry::from(actor_system& sys) {
return &sys.metrics();
}
void metric_registry::merge(metric_registry& other) {
if (this == &other)
return;
......
......@@ -27,6 +27,12 @@ caf::uri::impl_type default_instance;
namespace caf {
std::string uri::authority_type::host_str() const {
if (auto* str = std::get_if<std::string>(&host))
return *str;
return to_string(std::get<ip_address>(host));
}
uri::impl_type::impl_type() : rc_(1) {
// nop
}
......
......@@ -53,7 +53,7 @@ caf_add_component(
src/net/http/upper_layer.cpp
src/net/http/v1.cpp
src/net/ip.cpp
src/net/length_prefix_framing.cpp
src/net/lp/framing.cpp
src/net/middleman.cpp
src/net/multiplexer.cpp
src/net/network_socket.cpp
......
......@@ -8,26 +8,27 @@
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_bridge_base.hpp"
#include "caf/detail/flow_connector.hpp"
#include "caf/fwd.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace caf::net::binary {
namespace caf::detail {
/// Convenience alias for referring to the base type of @ref flow_bridge.
template <class Trait>
using flow_bridge_base_t
= detail::flow_bridge_base<upper_layer, lower_layer, Trait>;
using binary_flow_bridge_base_t
= detail::flow_bridge_base<net::binary::upper_layer, net::binary::lower_layer,
Trait>;
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public flow_bridge_base_t<Trait> {
class binary_flow_bridge : public binary_flow_bridge_base_t<Trait> {
public:
using super = flow_bridge_base_t<Trait>;
using super = binary_flow_bridge_base_t<Trait>;
using input_type = typename Trait::input_type;
......@@ -37,9 +38,10 @@ public:
using super::super;
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
static std::unique_ptr<binary_flow_bridge>
make(async::execution_context_ptr loop, connector_pointer conn) {
return std::make_unique<binary_flow_bridge>(std::move(loop),
std::move(conn));
}
bool write(const output_type& item) override {
......@@ -62,4 +64,4 @@ public:
}
};
} // namespace caf::net::binary
} // namespace caf::detail
......@@ -4,6 +4,7 @@
#pragma once
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
......
......@@ -7,8 +7,9 @@
#include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_connector.hpp"
#include "caf/fwd.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include <utility>
......@@ -29,7 +30,7 @@ public:
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using connector_pointer = net::flow_connector_ptr<Trait>;
using connector_pointer = flow_connector_ptr<Trait>;
flow_bridge_base(async::execution_context_ptr loop, connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
......
......@@ -12,7 +12,13 @@
#include <tuple>
namespace caf::net {
namespace caf::detail {
template <class Trait>
class flow_connector;
template <class Trait>
using flow_connector_ptr = std::shared_ptr<flow_connector<Trait>>;
/// Connects a flow bridge to input and output buffers.
template <class Trait>
......@@ -48,16 +54,12 @@ public:
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 net::flow_connector<Trait> {
class flow_connector_trivial_impl : public flow_connector<Trait> {
public:
using super = net::flow_connector<Trait>;
using super = flow_connector<Trait>;
using input_type = typename Trait::input_type;
......@@ -85,9 +87,9 @@ private:
/// 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> {
class flow_connector_basic_server_impl : public flow_connector<Trait> {
public:
using super = net::flow_connector<Trait>;
using super = flow_connector<Trait>;
using input_type = typename Trait::input_type;
......@@ -117,10 +119,6 @@ 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) {
......@@ -135,4 +133,4 @@ flow_connector<Trait>::make_basic_server(connect_event_buf buf) {
return std::make_shared<impl_t>(std::move(buf));
}
} // namespace caf::net
} // namespace caf::detail
......@@ -8,8 +8,8 @@
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_bridge_base.hpp"
#include "caf/detail/flow_connector.hpp"
#include "caf/fwd.hpp"
#include "caf/net/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"
......@@ -17,18 +17,19 @@
#include <utility>
namespace caf::net::web_socket {
namespace caf::detail {
/// Convenience alias for referring to the base type of @ref flow_bridge.
template <class Trait>
using flow_bridge_base_t
= detail::flow_bridge_base<upper_layer, lower_layer, Trait>;
using ws_flow_bridge_base_t
= detail::flow_bridge_base<net::web_socket::upper_layer,
net::web_socket::lower_layer, Trait>;
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public flow_bridge_base_t<Trait> {
class ws_flow_bridge : public ws_flow_bridge_base_t<Trait> {
public:
using super = flow_bridge_base_t<Trait>;
using super = ws_flow_bridge_base_t<Trait>;
using input_type = typename Trait::input_type;
......@@ -38,9 +39,9 @@ public:
using super::super;
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
static std::unique_ptr<ws_flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) {
return std::make_unique<ws_flow_bridge>(std::move(loop), std::move(conn));
}
bool write(const output_type& item) override {
......@@ -82,4 +83,4 @@ public:
}
};
} // namespace caf::net::web_socket
} // namespace caf::detail
......@@ -4,25 +4,28 @@
#pragma once
namespace caf::net::web_socket {
#include "caf/detail/flow_connector.hpp"
#include "caf/net/web_socket/request.hpp"
namespace caf::detail {
/// Calls an `OnRequest` handler with a @ref request object and passes the
/// generated buffers to the @ref flow_bridge.
template <class OnRequest, class Trait, class... Ts>
class flow_connector_request_impl : public flow_connector<Trait> {
class ws_flow_connector_request_impl : public flow_connector<Trait> {
public:
using super = flow_connector<Trait>;
using result_type = typename super::result_type;
using request_type = request<Trait, Ts...>;
using request_type = net::web_socket::request<Trait, Ts...>;
using app_res_type = typename request_type::app_res_type;
using producer_type = async::blocking_producer<app_res_type>;
template <class T>
flow_connector_request_impl(OnRequest&& on_request, T&& out)
ws_flow_connector_request_impl(OnRequest&& on_request, T&& out)
: on_request_(std::move(on_request)), out_(std::forward<T>(out)) {
// nop
}
......@@ -48,4 +51,4 @@ private:
producer_type out_;
};
} // namespace caf::net::web_socket
} // namespace caf::detail
......@@ -4,6 +4,7 @@
#pragma once
#include "caf/async/spsc_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/binary/fwd.hpp"
......@@ -13,20 +14,45 @@
namespace caf::net::binary {
/// A default trait type for binary protocols that uses @ref frame as both input
/// and output types and provides async::consumer_resource and
/// async::producer_resource as input_resource and output_resource types,
/// respectively.
class CAF_NET_EXPORT default_trait {
public:
/// The input type of the application, i.e., what that flows from the
/// WebSocket to the application layer.
/// The input type of the application, i.e., what that flows from the socket
/// to the application layer.
using input_type = frame;
/// The output type of the application, i.e., what flows from the application
/// layer to the WebSocket.
/// layer to the socket.
using output_type = frame;
/// A resource for consuming input_type elements.
using input_resource = async::consumer_resource<input_type>;
/// A resource for producing output_type elements.
using output_resource = async::producer_resource<output_type>;
/// An accept event from the server to transmit read and write handles.
using accept_event = cow_tuple<input_resource, output_resource>;
/// A resource for consuming accept events.
using acceptor_resource = async::consumer_resource<accept_event>;
/// Converts an output element to a byte buffer.
/// @param x The output element to convert.
/// @param bytes The output byte buffer.
/// @returns `true` on success, `false` otherwise.
bool convert(const output_type& x, byte_buffer& bytes);
/// Converts a byte buffer to an input element.
/// @param bytes The input byte buffer.
/// @param x The output input element.
/// @returns `true` on success, `false` otherwise.
bool convert(const_byte_span bytes, input_type& x);
/// Returns the last error that occurred.
error last_error();
};
......
......@@ -27,9 +27,6 @@ class typed_actor_shell;
template <class... Sigs>
class typed_actor_shell_ptr;
template <class Trait>
class flow_connector;
// -- classes ------------------------------------------------------------------
class actor_shell;
......@@ -56,9 +53,6 @@ struct udp_datagram_socket;
using multiplexer_ptr = intrusive_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
template <class Trait>
using flow_connector_ptr = std::shared_ptr<flow_connector<Trait>>;
// -- miscellaneous aliases ----------------------------------------------------
using text_buffer = std::vector<char>;
......
......@@ -6,6 +6,7 @@
#include "caf/byte_span.hpp"
#include "caf/detail/append_hex.hpp"
#include "caf/detail/message_flow_bridge.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
......@@ -16,7 +17,6 @@
#include "caf/net/http/status.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/net/http/v1.hpp"
#include "caf/net/message_flow_bridge.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket_manager.hpp"
......
// 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/defaults.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/binary_flow_bridge.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/detail/flow_connector.hpp"
#include "caf/detail/shared_ssl_acceptor.hpp"
#include "caf/fwd.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/lp/framing.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/prometheus/accept_factory.hpp"
#include "caf/net/prometheus/server.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/none.hpp"
#include <cstdint>
#include <functional>
#include <variant>
namespace caf::detail {
/// Specializes @ref connection_factory for the length-prefixing protocol.
template <class Trait, class Transport>
class lp_connection_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using connection_handle = typename Transport::connection_handle;
using connector_ptr = flow_connector_ptr<Trait>;
explicit lp_connection_factory(connector_ptr connector)
: connector_(std::move(connector)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto bridge = binary_flow_bridge<Trait>::make(mpx, connector_);
auto bridge_ptr = bridge.get();
auto impl = net::lp::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 = net::socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(mgr->as_disposable());
return mgr;
}
private:
connector_ptr connector_;
};
} // namespace caf::detail
namespace caf::net::lp {
template <class>
class with_t;
/// Factory for the `with(...).accept(...).start(...)` DSL.
template <class Trait>
class accept_factory {
public:
friend class with_t<Trait>;
accept_factory(accept_factory&&) = default;
accept_factory(const accept_factory&) = delete;
accept_factory& operator=(accept_factory&&) noexcept = default;
accept_factory& operator=(const accept_factory&) noexcept = delete;
~accept_factory() {
if (auto* fd = std::get_if<tcp_accept_socket>(&state_))
close(*fd);
}
/// Configures how many concurrent connections we are allowing.
accept_factory& max_connections(size_t value) {
max_connections_ = value;
return *this;
}
/// Sets the callback for errors.
template <class F>
accept_factory& do_on_error(F callback) {
do_on_error_ = std::move(callback);
return *this;
}
/// Starts a server that accepts incoming connections with the
/// length-prefixing protocol.
template <class OnStart>
disposable start(OnStart on_start) {
using acceptor_resource = typename Trait::acceptor_resource;
static_assert(std::is_invocable_v<OnStart, acceptor_resource>);
switch (state_.index()) {
case 1: {
auto& cfg = std::get<1>(state_);
auto fd = make_tcp_accept_socket(cfg.port, cfg.address, cfg.reuse_addr);
if (fd)
return do_start(*fd, on_start);
if (do_on_error_)
do_on_error_(fd.error());
return {};
}
case 2: {
// Pass ownership of the socket to the accept handler.
auto fd = std::get<2>(state_);
state_ = none;
return do_start(fd, on_start);
}
default:
return {};
}
}
private:
struct config {
uint16_t port;
std::string address;
bool reuse_addr;
};
explicit accept_factory(multiplexer* mpx) : mpx_(mpx) {
// nop
}
template <class Factory, class AcceptHandler, class Acceptor, class OnStart>
disposable do_start_impl(Acceptor&& acc, OnStart& on_start) {
using accept_event = typename Trait::accept_event;
using connector_t = detail::flow_connector<Trait>;
auto [pull, push] = async::make_spsc_buffer_resource<accept_event>();
auto serv = connector_t::make_basic_server(push.try_open());
auto factory = std::make_unique<Factory>(std::move(serv));
auto impl = AcceptHandler::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);
on_start(std::move(pull));
return disposable{std::move(ptr)};
}
template <class OnStart>
disposable do_start(tcp_accept_socket fd, OnStart& on_start) {
if (!ctx_) {
using factory_t = detail::lp_connection_factory<Trait, stream_transport>;
using impl_t = detail::accept_handler<tcp_accept_socket, stream_socket>;
return do_start_impl<factory_t, impl_t>(fd, on_start);
}
using factory_t = detail::lp_connection_factory<Trait, ssl::transport>;
using acc_t = detail::shared_ssl_acceptor;
using impl_t = detail::accept_handler<acc_t, ssl::connection>;
return do_start_impl<factory_t, impl_t>(acc_t{fd, ctx_}, on_start);
}
void set_ssl(ssl::context ctx) {
ctx_ = std::make_shared<ssl::context>(std::move(ctx));
}
void init(uint16_t port, std::string address, bool reuse_addr) {
state_ = config{port, std::move(address), reuse_addr};
}
void init(tcp_accept_socket fd) {
state_ = fd;
}
/// Pointer to the hosting actor system.
multiplexer* mpx_;
/// Callback for errors.
std::function<void(const error&)> do_on_error_;
/// Configures the maximum number of concurrent connections.
size_t max_connections_ = defaults::net::max_connections.fallback;
/// User-defined state for getting things up and running.
std::variant<none_t, config, tcp_accept_socket> state_;
/// Pointer to the (optional) SSL context.
std::shared_ptr<ssl::context> ctx_;
};
} // namespace caf::net::lp
// 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/spsc_buffer.hpp"
#include "caf/detail/binary_flow_bridge.hpp"
#include "caf/detail/flow_connector.hpp"
#include "caf/disposable.hpp"
#include "caf/net/lp/framing.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/timespan.hpp"
#include <chrono>
#include <cstdint>
#include <functional>
#include <type_traits>
#include <variant>
namespace caf::net::lp {
template <class>
class with_t;
/// Factory for the `with(...).connect(...).start(...)` DSL.
template <class Trait>
class connect_factory {
public:
friend class with_t<Trait>;
connect_factory(const connect_factory&) noexcept = delete;
connect_factory& operator=(const connect_factory&) noexcept = delete;
connect_factory(connect_factory&&) noexcept = default;
connect_factory& operator=(connect_factory&&) noexcept = default;
/// Starts a connection with the length-prefixing protocol.
template <class OnStart>
disposable start(OnStart on_start) {
using input_res_t = typename Trait::input_resource;
using output_res_t = typename Trait::output_resource;
static_assert(std::is_invocable_v<OnStart, input_res_t, output_res_t>);
switch (state_.index()) {
case 1: { // config
auto fd = try_connect(std::get<1>(state_));
if (fd) {
if (ctx_) {
auto conn = ctx_->new_connection(*fd);
if (conn)
return do_start(std::move(*conn), on_start);
if (do_on_error_)
do_on_error_(conn.error());
return {};
}
return do_start(*fd, on_start);
}
if (do_on_error_)
do_on_error_(fd.error());
return {};
}
case 2: { // stream_socket
// Pass ownership of the stream socket.
auto fd = std::get<2>(state_);
state_ = none;
return do_start(fd, on_start);
}
case 3: { // ssl::connection
// Pass ownership of the SSL connection.
auto conn = std::move(std::get<3>(state_));
state_ = none;
return do_start(std::move(conn), on_start);
}
case 4: // error
if (do_on_error_)
do_on_error_(std::get<4>(state_));
return {};
default:
return {};
}
}
/// Sets the retry delay for connection attempts.
///
/// @param value The new retry delay.
/// @returns a reference to this `connect_factory`.
connect_factory& retry_delay(timespan value) {
if (auto* cfg = std::get_if<config>(&state_))
cfg->retry_delay = value;
return *this;
}
/// Sets the connection timeout for connection attempts.
///
/// @param value The new connection timeout.
/// @returns a reference to this `connect_factory`.
connect_factory& connection_timeout(timespan value) {
if (auto* cfg = std::get_if<config>(&state_))
cfg->connection_timeout = value;
return *this;
}
/// Sets the maximum number of connection retry attempts.
///
/// @param value The new maximum retry count.
/// @returns a reference to this `connect_factory`.
connect_factory& max_retry_count(size_t value) {
if (auto* cfg = std::get_if<config>(&state_))
cfg->max_retry_count = value;
return *this;
}
/// Sets the callback for errors.
/// @returns a reference to this `connect_factory`.
template <class F>
connect_factory& do_on_error(F callback) {
do_on_error_ = std::move(callback);
return *this;
}
private:
struct config {
config(std::string address, uint16_t port)
: address(std::move(address)), port(port) {
// nop
}
std::string address;
uint16_t port;
timespan retry_delay = std::chrono::seconds{1};
timespan connection_timeout = infinite;
size_t max_retry_count = 0;
};
expected<tcp_stream_socket> try_connect(const config& cfg) {
auto result = make_connected_tcp_stream_socket(cfg.address, cfg.port,
cfg.connection_timeout);
if (result)
return result;
for (size_t i = 1; i <= cfg.max_retry_count; ++i) {
std::this_thread::sleep_for(cfg.retry_delay);
result = make_connected_tcp_stream_socket(cfg.address, cfg.port,
cfg.connection_timeout);
if (result)
return result;
}
return result;
}
template <class Conn, class OnStart>
disposable do_start(Conn conn, OnStart& on_start) {
// s2a: socket-to-application (and a2s is the inverse).
using input_t = typename Trait::input_type;
using output_t = typename Trait::output_type;
using transport_t = typename Conn::transport_type;
auto [s2a_pull, s2a_push] = async::make_spsc_buffer_resource<input_t>();
auto [a2s_pull, a2s_push] = async::make_spsc_buffer_resource<output_t>();
auto fc = detail::flow_connector<Trait>::make_trivial(std::move(a2s_pull),
std::move(s2a_push));
auto bridge = detail::binary_flow_bridge<Trait>::make(mpx_, std::move(fc));
auto bridge_ptr = bridge.get();
auto impl = framing::make(std::move(bridge));
auto transport = transport_t::make(std::move(conn), std::move(impl));
auto ptr = socket_manager::make(mpx_, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable());
mpx_->start(ptr);
on_start(std::move(s2a_pull), std::move(a2s_push));
return disposable{std::move(ptr)};
}
explicit connect_factory(multiplexer* mpx) : mpx_(mpx) {
// nop
}
connect_factory(multiplexer* mpx, error err)
: mpx_(mpx), state_(std::move(err)) {
// nop
}
/// Initializes the connect factory to connect to the given TCP `host` and
/// `port`.
///
/// @param host The hostname or IP address to connect to.
/// @param port The port number to connect to.
void init(std::string host, uint16_t port) {
state_ = config{std::move(host), port};
}
/// Initializes the connect factory to connect to the given TCP `socket`.
///
/// @param fd The TCP socket to connect.
void init(stream_socket fd) {
state_ = fd;
}
/// Initializes the connect factory to connect to the given TCP `socket`.
///
/// @param conn The SSL connection object.
void init(ssl::connection conn) {
state_ = std::move(conn);
}
/// Initializes the connect factory with an error.
///
/// @param err The error to be later forwarded to the `do_on_error_` handler.
void init(error err) {
state_ = std::move(err);
}
void set_ssl(ssl::context ctx) {
ctx_ = std::make_shared<ssl::context>(std::move(ctx));
}
/// Pointer to multiplexer that runs the protocol stack.
multiplexer* mpx_;
/// Callback for errors.
std::function<void(const error&)> do_on_error_;
/// Configures the maximum number of concurrent connections.
size_t max_connections_ = defaults::net::max_connections.fallback;
/// User-defined state for getting things up and running.
std::variant<none_t, config, stream_socket, ssl::connection, error> state_;
/// Pointer to the (optional) SSL context.
std::shared_ptr<ssl::context> ctx_;
};
} // namespace caf::net::lp
// 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/actor_system.hpp"
#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/binary_flow_bridge.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/frame.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_oriented.hpp"
#include <cstdint>
#include <cstring>
#include <memory>
#include <type_traits>
namespace caf::net::lp {
/// Implements length-prefix framing for discretizing a Byte stream into
/// messages of varying size. The framing uses 4 Bytes for the length prefix,
/// but messages (including the 4 Bytes for the length prefix) are limited to a
/// maximum size of INT32_MAX. This limitation comes from the POSIX API (recv)
/// on 32-bit platforms.
class CAF_NET_EXPORT framing : public stream_oriented::upper_layer,
public binary::lower_layer {
public:
// -- member types -----------------------------------------------------------
using upper_layer_ptr = std::unique_ptr<binary::upper_layer>;
// -- constants --------------------------------------------------------------
static constexpr size_t hdr_size = sizeof(uint32_t);
static constexpr size_t max_message_length = INT32_MAX - sizeof(uint32_t);
// -- constructors, destructors, and assignment operators --------------------
explicit framing(upper_layer_ptr up) : up_(std::move(up)) {
// nop
}
// -- factories --------------------------------------------------------------
static std::unique_ptr<framing> make(upper_layer_ptr up);
// -- implementation of stream_oriented::upper_layer -------------------------
error start(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void prepare_send() override;
bool done_sending() override;
// -- implementation of binary::lower_layer ----------------------------------
bool can_send_more() const noexcept override;
void request_messages() override;
void suspend_reading() override;
bool is_reading() const noexcept override;
void write_later() override;
void begin_message() override;
byte_buffer& message_buffer() override;
bool end_message() override;
void shutdown() override;
// -- utility functions ------------------------------------------------------
static std::pair<size_t, byte_span> split(byte_span buffer) noexcept;
private:
// -- member variables -------------------------------------------------------
stream_oriented::lower_layer* down_;
upper_layer_ptr up_;
size_t message_offset_ = 0;
};
} // namespace caf::net::lp
// 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/lp/accept_factory.hpp"
#include "caf/net/lp/connect_factory.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include <cstdint>
namespace caf::net::lp {
/// Entry point for the `with(...)` DSL.
template <class Trait>
class with_t {
public:
explicit with_t(multiplexer* mpx) : mpx_(mpx) {
// nop
}
with_t(const with_t&) noexcept = default;
with_t& operator=(const with_t&) noexcept = default;
/// Creates an `accept_factory` object for the given TCP `port` and
/// `bind_address`.
///
/// @param port Port number to bind to.
/// @param bind_address IP address to bind to. Default is an empty string.
/// @param reuse_addr Whether or not to set `SO_REUSEADDR`.
/// @returns an `accept_factory` object initialized with the given parameters.
accept_factory<Trait> accept(uint16_t port, std::string bind_address = "",
bool reuse_addr = true) {
accept_factory<Trait> factory{mpx_};
factory.init(port, std::move(bind_address), std::move(reuse_addr));
return factory;
}
/// Creates an `accept_factory` object for the given accept socket.
///
/// @param fd File descriptor for the accept socket.
/// @returns an `accept_factory` object that will start a Prometheus server on
/// the given socket.
accept_factory<Trait> accept(tcp_accept_socket fd) {
accept_factory<Trait> factory{mpx_};
factory.init(fd);
return factory;
}
/// Creates an `accept_factory` object for the given acceptor.
///
/// @param acc The SSL acceptor for incoming connections.
/// @returns an `accept_factory` object that will start a Prometheus server on
/// the given acceptor.
accept_factory<Trait> accept(ssl::acceptor acc) {
accept_factory<Trait> factory{mpx_};
factory.set_ssl(std::move(std::move(acc.ctx())));
factory.init(acc.fd());
return factory;
}
/// Creates an `accept_factory` object for the given TCP `port` and
/// `bind_address`.
///
/// @param ctx The SSL context for encryption.
/// @param port Port number to bind to.
/// @param bind_address IP address to bind to. Default is an empty string.
/// @param reuse_addr Whether or not to set `SO_REUSEADDR`.
/// @returns an `accept_factory` object initialized with the given parameters.
accept_factory<Trait> accept(ssl::context ctx, uint16_t port,
std::string bind_address = "",
bool reuse_addr = true) {
accept_factory<Trait> factory{mpx_};
factory.set_ssl(std::move(std::move(ctx)));
factory.init(port, std::move(bind_address), std::move(reuse_addr));
return factory;
}
/// Creates a `connect_factory` object for the given TCP `host` and `port`.
///
/// @param host The hostname or IP address to connect to.
/// @param port The port number to connect to.
/// @returns a `connect_factory` object initialized with the given parameters.
connect_factory<Trait> connect(std::string host, uint16_t port) {
connect_factory<Trait> factory{mpx_};
factory.init(std::move(host), port);
return factory;
}
/// Creates a `connect_factory` object for the given SSL `context`, TCP
/// `host`, and `port`.
///
/// @param ctx The SSL context for encryption.
/// @param host The hostname or IP address to connect to.
/// @param port The port number to connect to.
/// @returns a `connect_factory` object initialized with the given parameters.
connect_factory<Trait> connect(ssl::context ctx, std::string host,
uint16_t port) {
connect_factory<Trait> factory{mpx_};
factory.set_ssl(std::move(ctx));
factory.init(std::move(host), port);
return factory;
}
/// Creates a `connect_factory` object for the given TCP `endpoint`.
///
/// @param endpoint The endpoint of the TCP server to connect to.
/// @returns a `connect_factory` object initialized with the given parameters.
connect_factory<Trait> connect(const uri& endpoint) {
return connect_impl(nullptr, endpoint);
}
/// Creates a `connect_factory` object for the given SSL `context` and TCP
/// `endpoint`.
///
/// @param ctx The SSL context for encryption.
/// @param endpoint The endpoint of the TCP server to connect to.
/// @returns a `connect_factory` object initialized with the given parameters.
connect_factory<Trait> connect(ssl::context ctx, const uri& endpoint) {
return connect_impl(&ctx, endpoint);
}
/// Creates a `connect_factory` object for the given TCP `endpoint`.
///
/// @param endpoint The endpoint of the TCP server to connect to.
/// @returns a `connect_factory` object initialized with the given parameters.
connect_factory<Trait> connect(expected<uri> endpoint) {
if (endpoint)
return connect_impl(nullptr, std::move(*endpoint));
return connect_factory<Trait>{std::move(endpoint.error())};
}
/// Creates a `connect_factory` object for the given SSL `context` and TCP
/// `endpoint`.
///
/// @param ctx The SSL context for encryption.
/// @param endpoint The endpoint of the TCP server to connect to.
/// @returns a `connect_factory` object initialized with the given parameters.
connect_factory<Trait> connect(ssl::context ctx, expected<uri> endpoint) {
if (endpoint)
return connect_impl(&ctx, std::move(*endpoint));
return connect_factory<Trait>{std::move(endpoint.error())};
}
/// Creates a `connect_factory` object for the given stream `fd`.
///
/// @param fd The stream socket to use for the connection.
/// @returns a `connect_factory` object that will use the given socket.
connect_factory<Trait> connect(stream_socket fd) {
connect_factory<Trait> factory{mpx_};
factory.init(fd);
return factory;
}
/// Creates a `connect_factory` object for the given SSL `connection`.
///
/// @param conn The SSL connection to use.
/// @returns a `connect_factory` object that will use the given connection.
connect_factory<Trait> connect(ssl::connection conn) {
connect_factory<Trait> factory{mpx_};
factory.init(std::move(conn));
return factory;
}
private:
connect_factory<Trait> connect_impl(ssl::context* ctx, const uri& endpoint) {
if (endpoint.scheme() != "tcp" || endpoint.authority().empty()) {
auto err = make_error(sec::invalid_argument,
"lp::connect expects tcp://<host>:<port> URIs");
return connect_factory<Trait>{mpx_, std::move(err)};
}
if (endpoint.authority().port == 0) {
auto err = make_error(sec::invalid_argument,
"lp::connect expects URIs with a non-zero port");
return connect_factory<Trait>{mpx_, std::move(err)};
}
connect_factory<Trait> factory{mpx_};
if (ctx != nullptr) {
factory.set_ssl(std::move(*ctx));
}
factory.init(endpoint.authority().host_str(), endpoint.authority().port);
return factory;
}
/// Pointer to multiplexer that runs the protocol stack.
multiplexer* mpx_;
};
template <class Trait = binary::default_trait>
with_t<Trait> with(actor_system& sys) {
return with_t<Trait>{multiplexer::from(sys)};
}
template <class Trait = binary::default_trait>
with_t<Trait> with(multiplexer* mpx) {
return with_t<Trait>{mpx};
}
} // namespace caf::net::lp
......@@ -57,6 +57,9 @@ public:
/// windows. Has no effect when running on Windows.
static void block_sigpipe();
/// Returns a pointer to the multiplexer from the actor system.
static multiplexer* from(actor_system& sys);
// -- constructors, destructors, and assignment operators --------------------
~multiplexer();
......
......@@ -120,31 +120,29 @@ private:
// nop
}
template <class Factory, class AcceptHandler, class Acceptor>
disposable do_start_impl(Acceptor&& acc) {
auto mpx = multiplexer::from(*sys_);
auto registry = &sys_->metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = std::make_unique<Factory>(std::move(state));
auto impl = AcceptHandler::make(std::forward<Acceptor>(acc),
std::move(factory), max_connections_);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return mgr->as_disposable();
}
disposable do_start(tcp_accept_socket fd) {
if (!ctx_) {
using factory_t = detail::prometheus_conn_factory<stream_transport>;
using impl_t = detail::accept_handler<tcp_accept_socket, stream_socket>;
auto mpx = &sys_->network_manager().mpx();
auto registry = &sys_->metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = std::make_unique<factory_t>(std::move(state));
auto impl = impl_t::make(fd, std::move(factory), max_connections_);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return mgr->as_disposable();
return do_start_impl<factory_t, impl_t>(fd);
}
using factory_t = detail::prometheus_conn_factory<ssl::transport>;
using acc_t = detail::shared_ssl_acceptor;
using impl_t = detail::accept_handler<acc_t, ssl::connection>;
auto mpx = &sys_->network_manager().mpx();
auto registry = &sys_->metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = std::make_unique<factory_t>(std::move(state));
auto impl = impl_t::make(acc_t{fd, ctx_}, std::move(factory),
max_connections_);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return mgr->as_disposable();
return do_start_impl<factory_t, impl_t>(acc_t{fd, ctx_});
}
void set_ssl(ssl::context ctx) {
......
......@@ -8,13 +8,13 @@
#include "caf/cow_tuple.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/detail/flow_connector.hpp"
#include "caf/detail/ws_flow_bridge.hpp"
#include "caf/detail/ws_flow_connector_request_impl.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
#include "caf/net/web_socket/flow_connector_request_impl.hpp"
#include "caf/net/web_socket/frame.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/net/web_socket/server.hpp"
......@@ -58,7 +58,7 @@ class ws_conn_factory
public:
using connection_handle = typename Transport::connection_handle;
using connector_pointer = net::flow_connector_ptr<Trait>;
using connector_pointer = flow_connector_ptr<Trait>;
explicit ws_conn_factory(connector_pointer connector)
: connector_(std::move(connector)) {
......@@ -67,7 +67,7 @@ public:
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto app = net::web_socket::flow_bridge<Trait>::make(mpx, connector_);
auto app = detail::ws_flow_bridge<Trait>::make(mpx, connector_);
auto app_ptr = app.get();
auto ws = net::web_socket::server::make(std::move(app));
auto fd = conn.fd();
......@@ -107,7 +107,7 @@ disposable accept(actor_system& sys, Acceptor acc,
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...>;
= detail::ws_flow_connector_request_impl<OnRequest, trait_t, Ts...>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
......
......@@ -6,12 +6,12 @@
#include "caf/byte_span.hpp"
#include "caf/detail/base64.hpp"
#include "caf/detail/message_flow_bridge.hpp"
#include "caf/error.hpp"
#include "caf/hash/sha1.hpp"
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
#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"
......
......@@ -4,13 +4,14 @@
#pragma once
namespace caf::net::web_socket {
namespace caf::detail {
template <class, class, class...>
class ws_flow_connector_request_impl;
template <class OnRequest, class Trait, class... Ts>
class flow_connector_request_impl;
} // namespace caf::detail
template <class Trait>
class flow_bridge;
namespace caf::net::web_socket {
template <class Trait, class... Ts>
class request;
......
......@@ -8,6 +8,7 @@
#include "caf/cow_tuple.hpp"
#include "caf/error.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/web_socket/fwd.hpp"
#include <utility>
......@@ -19,7 +20,7 @@ template <class Trait, class... Ts>
class request {
public:
template <class, class, class...>
friend class flow_connector_request_impl;
friend class detail::ws_flow_connector_request_impl;
using input_type = typename Trait::input_type;
......
......@@ -5,6 +5,7 @@
#pragma once
#include "caf/byte_span.hpp"
#include "caf/detail/message_flow_bridge.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
......@@ -13,7 +14,6 @@
#include "caf/net/http/method.hpp"
#include "caf/net/http/status.hpp"
#include "caf/net/http/v1.hpp"
#include "caf/net/message_flow_bridge.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket_manager.hpp"
......
#include "caf/net/length_prefix_framing.hpp"
#include "caf/net/lp/framing.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/error.hpp"
......@@ -6,35 +6,26 @@
#include "caf/net/receive_policy.hpp"
#include "caf/sec.hpp"
namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
length_prefix_framing::length_prefix_framing(upper_layer_ptr up)
: up_(std::move(up)) {
// nop
}
namespace caf::net::lp {
// -- factories ----------------------------------------------------------------
std::unique_ptr<length_prefix_framing>
length_prefix_framing::make(upper_layer_ptr up) {
return std::make_unique<length_prefix_framing>(std::move(up));
std::unique_ptr<framing> framing::make(upper_layer_ptr up) {
return std::make_unique<framing>(std::move(up));
}
// -- implementation of stream_oriented::upper_layer ---------------------------
error length_prefix_framing::start(stream_oriented::lower_layer* down,
const settings& cfg) {
error framing::start(stream_oriented::lower_layer* down, const settings& cfg) {
down_ = down;
return up_->start(this, cfg);
}
void length_prefix_framing::abort(const error& reason) {
void framing::abort(const error& reason) {
up_->abort(reason);
}
ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
ptrdiff_t framing::consume(byte_span input, byte_span) {
CAF_LOG_TRACE("got" << input.size() << "bytes\n");
if (input.size() < sizeof(uint32_t)) {
CAF_LOG_ERROR("received too few bytes from underlying transport");
......@@ -80,49 +71,49 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
}
}
void length_prefix_framing::prepare_send() {
void framing::prepare_send() {
up_->prepare_send();
}
bool length_prefix_framing::done_sending() {
bool framing::done_sending() {
return up_->done_sending();
}
// -- implementation of binary::lower_layer ------------------------------------
bool length_prefix_framing::can_send_more() const noexcept {
bool framing::can_send_more() const noexcept {
return down_->can_send_more();
}
void length_prefix_framing::suspend_reading() {
void framing::suspend_reading() {
down_->configure_read(receive_policy::stop());
}
bool length_prefix_framing::is_reading() const noexcept {
bool framing::is_reading() const noexcept {
return down_->is_reading();
}
void length_prefix_framing::write_later() {
void framing::write_later() {
down_->write_later();
}
void length_prefix_framing::request_messages() {
void framing::request_messages() {
if (!down_->is_reading())
down_->configure_read(receive_policy::exactly(hdr_size));
}
void length_prefix_framing::begin_message() {
void framing::begin_message() {
down_->begin_output();
auto& buf = down_->output_buffer();
message_offset_ = buf.size();
buf.insert(buf.end(), 4, std::byte{0});
}
byte_buffer& length_prefix_framing::message_buffer() {
byte_buffer& framing::message_buffer() {
return down_->output_buffer();
}
bool length_prefix_framing::end_message() {
bool framing::end_message() {
using detail::to_network_order;
auto& buf = down_->output_buffer();
CAF_ASSERT(message_offset_ < buf.size());
......@@ -140,14 +131,13 @@ bool length_prefix_framing::end_message() {
}
}
void length_prefix_framing::shutdown() {
void framing::shutdown() {
down_->shutdown();
}
// -- utility functions ------------------------------------------------------
std::pair<size_t, byte_span>
length_prefix_framing::split(byte_span buffer) noexcept {
std::pair<size_t, byte_span> framing::split(byte_span buffer) noexcept {
CAF_ASSERT(buffer.size() >= sizeof(uint32_t));
auto u32_size = uint32_t{0};
memcpy(&u32_size, buffer.data(), sizeof(uint32_t));
......@@ -155,4 +145,4 @@ length_prefix_framing::split(byte_span buffer) noexcept {
return std::make_pair(msg_size, buffer.subspan(sizeof(uint32_t)));
}
} // namespace caf::net
} // namespace caf::net::lp
......@@ -5,6 +5,7 @@
#include "caf/net/multiplexer.hpp"
#include "caf/action.hpp"
#include "caf/actor_system.hpp"
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/error.hpp"
......@@ -93,6 +94,10 @@ void multiplexer::block_sigpipe() {
#endif
multiplexer* multiplexer::from(actor_system& sys) {
return sys.network_manager().mpx_ptr();
}
// -- constructors, destructors, and assignment operators ----------------------
multiplexer::multiplexer(middleman* owner) : owner_(owner) {
......
......@@ -6,12 +6,12 @@
#include "caf/byte_span.hpp"
#include "caf/detail/base64.hpp"
#include "caf/detail/message_flow_bridge.hpp"
#include "caf/error.hpp"
#include "caf/hash/sha1.hpp"
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
#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"
......
......@@ -4,7 +4,8 @@
#include "caf/net/web_socket/connect.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/detail/flow_connector.hpp"
#include "caf/detail/ws_flow_bridge.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/ssl/transport.hpp"
......@@ -12,7 +13,6 @@
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/client.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
namespace ws = caf::net::web_socket;
......@@ -28,7 +28,7 @@ ws_do_connect_impl(actor_system& sys, Connection conn, ws::handshake& hs) {
auto mpx = sys.network_manager().mpx_ptr();
auto connector = std::make_shared<connector_t>(std::move(ws_pull),
std::move(ws_push));
auto bridge = ws::flow_bridge<trait_t>::make(mpx, std::move(connector));
auto bridge = ws_flow_bridge<trait_t>::make(mpx, std::move(connector));
auto impl = ws::client::make(std::move(hs), std::move(bridge));
auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(impl));
......
......@@ -4,7 +4,7 @@
#define CAF_SUITE net.length_prefix_framing
#include "caf/net/length_prefix_framing.hpp"
#include "caf/net/lp/with.hpp"
#include "net-test.hpp"
......@@ -123,7 +123,7 @@ auto decode(byte_buffer& buf) {
string_list result;
auto input = make_span(buf);
while (!input.empty()) {
auto [msg_size, msg] = net::length_prefix_framing::split(input);
auto [msg_size, msg] = net::lp::framing::split(input);
if (msg_size > msg.size()) {
CAF_FAIL("cannot decode buffer: invalid message size");
} else if (!std::all_of(msg.begin(), msg.begin() + msg_size, printable)) {
......@@ -157,11 +157,11 @@ void run_writer(net::stream_socket fd) {
} // namespace
SCENARIO("length-prefix framing reads data with 32-bit size headers") {
GIVEN("a length_prefix_framing with an app that consumes strings") {
GIVEN("a framing object with an app that consumes strings") {
WHEN("pushing data into the unit-under-test") {
auto buf = std::make_shared<string_list>();
auto app = app_t<false>::make(nullptr, buf);
auto framing = net::length_prefix_framing::make(std::move(app));
auto framing = net::lp::framing::make(std::move(app));
auto uut = mock_stream_transport::make(std::move(framing));
CHECK_EQ(uut->start(), error{});
THEN("the app receives all strings as individual messages") {
......@@ -181,7 +181,7 @@ SCENARIO("length-prefix framing reads data with 32-bit size headers") {
SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
using namespace std::literals;
GIVEN("a length_prefix_framing with an app that consumes strings") {
GIVEN("a framing object with an app that consumes strings") {
auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
auto writer = std::thread{run_writer, fd1};
auto mpx = net::multiplexer::make(nullptr);
......@@ -195,7 +195,7 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
auto buf = std::make_shared<string_list>();
auto app = app_t<true>::make(mpx, buf);
auto app_ptr = app.get();
auto framing = net::length_prefix_framing::make(std::move(app));
auto framing = net::lp::framing::make(std::move(app));
auto transport = net::stream_transport::make(fd2, std::move(framing));
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
CHECK_EQ(mgr->start(settings{}), none);
......@@ -233,7 +233,7 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
}
}
SCENARIO("length_prefix_framing::run translates between flows and socket I/O") {
SCENARIO("lp::with(...).connect(...) translates between flows and socket I/O") {
using namespace std::literals;
GIVEN("a connected socket with a writer at the other end") {
auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
......@@ -247,11 +247,8 @@ SCENARIO("length_prefix_framing::run translates between flows and socket I/O") {
caf::actor_system sys{cfg};
auto buf = std::make_shared<std::vector<std::string>>();
caf::actor hdl;
using trait = net::binary::default_trait;
using lpf = net::length_prefix_framing::bind<trait>;
lpf::run(sys, fd2, [&](auto event) {
hdl = sys.spawn([event, buf](event_based_actor* self) {
auto [pull, push] = event.data();
net::lp::with(sys).connect(fd2).start([&](auto pull, auto push) {
hdl = sys.spawn([buf, pull, push](event_based_actor* self) {
pull.observe_on(self)
.do_on_error([](const error& what) { //
MESSAGE("flow aborted: " << what);
......
......@@ -40,6 +40,7 @@ Contents
net/Overview
net/Prometheus
net/LengthPrefixFraming
.. toctree::
:maxdepth: 2
......
.. _net_prometheus:
Length-prefix Framing
=====================
Length-prefix framing is a simple protocol for encoding variable-length messages
on the network. Each message is preceded by a fixed-length value (32 bit) that
indicates the length of the message, in bytes.
When a sender wants to transmit a message, it first encodes the message into a
sequence of bytes. It then calculates the length of the byte sequence and
prefixes the message with the length. The receiver then reads the length prefix
to determine the length of the message before reading the bytes for the message.
Starting a Server
-----------------
The simplest way to start a length-prefix framing server is by using the
high-level factory API.
First, include the necessary header files:
.. code-block:: cpp
#include <caf/net/lp/with.hpp>
Then, create a new length-prefix factory using the ``caf::net::lp::with``
function:
.. code-block:: cpp
caf::net::lp::with(sys)
Optionally, you can also provide a custom trait type by calling
``caf::net::lp::with<my_trait>(sys)`` instead.
Once you have set up the factory, you can call the ``accept`` function on it to
start accepting incoming connections. The ``accept`` function can be called
with:
- A port number, a bind address (which defaults to the empty string, allowing
connections from any host), and a boolean indicating whether to create the
socket with ``SO_REUSEADDR`` (which defaults to ``true``).
- An ``ssl::context`` object (if you want to use SSL/TLS encryption), a port
number, a bind address, and a boolean indicating whether to create the socket
with ``SO_REUSEADDR`` (with the same defaults as before).
- A ``tcp_accept_socket`` object for accepting incoming connections.
- An ``ssl::acceptor`` object for accepting incoming connections over SSL/TLS.
After setting up the factory and calling ``accept``, you can use the following
functions on the returned object:
- ``max_connections(size_t)`` to limit the maximum number of concurrent
connections on the server.
- ``do_on_error(function<void(const caf::error&)>)`` to set an error callback.
- ``start(OnStart)`` to initialize the server and run it in the background. The
``OnStart`` callback takes an argument of type ``trait::acceptor_resource``.
This is a consumer resource for receiving accept events. Each accept event
consists of an input resource and an output resource for reading from and
writing to the new connection.
Starting a Client
-----------------
Starting a client also uses the ``with`` factory. Calling ``connect`` instead of
``accept`` creates a factory for clients. The ``connect`` function may be called
with:
- A remote address (such as a hostname or IP address) plus a port number.
- An ``url`` object using the ``tcp`` schema, i.e., ``tcp://<host>:<port>``.
- A ``stream_socket`` object for running the length-prefix framing.
- An ``ssl::connection`` object for running the length-prefix framing.
After setting up the factory and calling ``connect``, you can use the following
functions on the returned object:
- ``do_on_error(function<void(const caf::error&)>)`` to set an error callback.
- ``start(OnStart)`` to initialize the server and run it in the background. The
- ``retry_delay(timespan)`` to set the delay between connection retries when a
connection attempt fails.
- ``connection_timeout(timespan)`` to set the maximum amount of time to wait for
a connection attempt to succeed before giving up.
- ``max_retry_count(size_t)`` to set the maximum number of times to retry a
connection attempt before giving up.
- ``OnStart`` callback takes two arguments: the input resource and the output
resource for reading from and writing to the new connection.
Note that ``retry_delay``, ``connection_timeout(timespan)`` and
``max_retry_count(size_t)`` have no effect when constructing the factory object
from a socket or SSL connection.
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