Commit 39657215 authored by Dominik Charousset's avatar Dominik Charousset

Further simplify caf-net DSL scaffolding

parent c0663ea7
......@@ -13,21 +13,23 @@ namespace caf::detail {
/// Accepts incoming clients with an Acceptor and handles them via a connection
/// factory.
template <class Acceptor, class ConnectionHandle>
template <class Acceptor,
class ConnHandle = typename Acceptor::accept_result_type>
class accept_handler : public net::socket_event_layer {
public:
// -- member types -----------------------------------------------------------
using socket_type = net::socket;
using connection_handle = ConnectionHandle;
using connection_handle = ConnHandle;
using factory_type = connection_factory<connection_handle>;
using factory_ptr = detail::connection_factory_ptr<connection_handle>;
// -- constructors, destructors, and assignment operators --------------------
template <class FactoryPtr, class... Ts>
accept_handler(Acceptor acc, FactoryPtr fptr, size_t max_connections)
accept_handler(Acceptor acc, factory_ptr fptr, size_t max_connections)
: acc_(std::move(acc)),
factory_(std::move(fptr)),
max_connections_(max_connections) {
......@@ -42,9 +44,8 @@ public:
// -- factories --------------------------------------------------------------
template <class FactoryPtr, class... Ts>
static std::unique_ptr<accept_handler>
make(Acceptor acc, FactoryPtr fptr, size_t max_connections) {
static std::unique_ptr<accept_handler> make(Acceptor acc, factory_ptr fptr,
size_t max_connections) {
return std::make_unique<accept_handler>(std::move(acc), std::move(fptr),
max_connections);
}
......@@ -119,7 +120,7 @@ private:
Acceptor acc_;
detail::connection_factory_ptr<connection_handle> factory_;
factory_ptr factory_;
size_t max_connections_;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/socket.hpp"
#include "caf/sec.hpp"
namespace caf::net {
/// Lifts `Socket` to a `expected<Socket>` and sets an error if `fd` is invalid.
template <class Socket>
expected<Socket> checked_socket(Socket fd) {
using res_t = expected<Socket>;
if (fd.id != invalid_socket_id)
return res_t{fd};
else
return res_t{make_error(sec::runtime_error, "invalid socket handle")};
}
/// A function object that calls `checked_socket`.
static constexpr auto check_socket = [](auto fd) { return checked_socket(fd); };
} // namespace caf::net
......@@ -10,48 +10,7 @@
namespace caf::net::dsl {
/// Base type for our DSL classes to configure a factory object..
template <class Trait>
class base {
public:
using trait_type = Trait;
virtual ~base() {
// nop
}
/// @returns the pointer to the @ref multiplexer.
virtual multiplexer* mpx() const noexcept = 0;
/// @returns the trait object.
virtual const Trait& trait() const noexcept = 0;
/// @returns the optional SSL context, whereas an object with
/// default-constructed error is treated as "no SSL".
expected<ssl::context>& get_context() {
return get_context_impl();
}
/// @private
template <class ConfigType>
auto with_context(intrusive_ptr<ConfigType> ptr) {
// Move the context into the config if present.
auto& ctx = get_context();
if (ctx) {
ptr->ctx = std::make_shared<ssl::context>(std::move(*ctx));
return as_base_ptr(ptr);
}
// Default-constructed error just means "no SSL".
if (!ctx.error())
return as_base_ptr(ptr);
// We actually have an error: replace `ptr` with a fail config. Need to cast
// to the base type for to_fail_config to pick up the right overload.
auto fptr = to_fail_config(as_base_ptr(ptr), std::move(ctx.error()));
return as_base_ptr(fptr);
}
private:
virtual expected<ssl::context>& get_context_impl() noexcept = 0;
};
/// Base type for our DSL classes to configure a factory object.
struct base {};
} // namespace caf::net::dsl
......@@ -23,7 +23,9 @@ namespace caf::net::dsl {
/// Meta programming utility.
template <class T>
struct client_config_tag {};
struct client_config_tag {
using type = T;
};
/// Simple type for storing host and port information for reaching a server.
struct server_address {
......@@ -34,28 +36,28 @@ struct server_address {
uint16_t port;
};
/// Wraps configuration parameters for starting clients.
class client_config {
public:
/// Configuration for a client that creates the socket on demand.
class lazy {
class CAF_NET_EXPORT lazy : public has_ctx {
public:
/// Type for holding a client address.
using server_t = std::variant<server_address, uri>;
lazy(std::string host, uint16_t port) {
server = server_address{std::move(host), port};
static constexpr std::string_view name = "lazy";
lazy(std::string host, uint16_t port)
: server(server_address{std::move(host), port}) {
}
explicit lazy(uri addr) {
server = addr;
explicit lazy(const uri& addr) : server(addr) {
// nop
}
/// The address for reaching the server or an error.
server_t server;
/// SSL context for secure servers.
std::shared_ptr<ssl::context> ctx;
/// The delay between connection attempts.
timespan retry_delay = std::chrono::seconds{1};
......@@ -64,31 +66,15 @@ public:
/// The maximum amount of retries.
size_t max_retry_count = 0;
/// Returns a function that, when called with a @ref stream_socket, calls
/// `f` either with a new SSL connection from `ctx` or with the file the
/// file descriptor if no SSL context is defined.
template <class F>
auto with_ctx(F&& f) {
return [this, g = std::forward<F>(f)](stream_socket fd) mutable {
using res_t = decltype(g(fd));
if (ctx) {
auto conn = ctx->new_connection(fd);
if (conn)
return g(*conn);
else
return res_t{std::move(conn.error())};
} else
return g(fd);
};
}
};
static constexpr auto lazy_v = client_config_tag<lazy>{};
/// Configuration for a client that uses a user-provided socket.
class socket {
class CAF_NET_EXPORT socket : public has_ctx {
public:
static constexpr std::string_view name = "socket";
explicit socket(stream_socket fd) : fd(fd) {
// nop
}
......@@ -117,9 +103,6 @@ public:
/// The socket file descriptor to use.
stream_socket fd;
/// SSL context for secure servers.
std::shared_ptr<ssl::context> ctx;
/// Returns the file descriptor and setting the `fd` member variable to the
/// invalid socket.
stream_socket take_fd() noexcept {
......@@ -133,8 +116,10 @@ public:
/// Configuration for a client that uses an already established SSL
/// connection.
class conn {
class CAF_NET_EXPORT conn {
public:
static constexpr std::string_view name = "conn";
explicit conn(ssl::connection st) : state(std::move(st)) {
// nop
}
......@@ -165,36 +150,19 @@ public:
static constexpr auto fail_v = client_config_tag<error>{};
template <class Base>
class value : public Base {
class value : public config_impl<Base, lazy, socket, conn> {
public:
using super = Base;
using super = config_impl<Base, lazy, socket, conn>;
template <class Trait, class... Data>
value(net::multiplexer* mpx, Trait trait, Data&&... arg)
: super(mpx, std::move(trait)), data(std::forward<Data>(arg)...) {
// nop
}
template <class... Data>
explicit value(const value& other, Data&&... arg)
: super(other), data(std::forward<Data>(arg)...) {
// nop
}
std::variant<error, lazy, socket, conn> data;
using super::super;
template <class T, class Trait, class... Args>
static intrusive_ptr<value> make(client_config_tag<T>,
net::multiplexer* mpx, Trait trait,
Args&&... args) {
return make_counted<value>(mpx, std::move(trait), std::in_place_type<T>,
template <class From, class T, class... Args>
static auto make(client_config_tag<T>, From&& from, Args&&... args) {
static_assert(std::is_constructible_v<T, Args...>);
return make_counted<value>(std::forward<From>(from),
std::in_place_type<T>,
std::forward<Args>(args)...);
}
template <class F>
auto visit(F&& f) {
return std::visit([&](auto& arg) { return f(arg); }, data);
}
};
};
......@@ -204,11 +172,4 @@ using client_config_value = client_config::value<Base>;
template <class Base>
using client_config_ptr = intrusive_ptr<client_config_value<Base>>;
/// Creates a `fail_client_config` from another configuration object plus error.
template <class Base>
client_config_ptr<Base> to_fail_config(client_config_ptr<Base> ptr, error err) {
using val_t = typename client_config::template value<Base>;
return make_counted<val_t>(*ptr, std::move(err));
}
} // namespace caf::net::dsl
......@@ -7,13 +7,18 @@
#include "caf/callback.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/net/checked_socket.hpp"
#include "caf/net/dsl/base.hpp"
#include "caf/net/dsl/get_name.hpp"
#include "caf/net/dsl/has_ctx.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/ref_counted.hpp"
#include "caf/sec.hpp"
#include "caf/uri.hpp"
#include <cassert>
......@@ -35,6 +40,34 @@ public:
virtual ~config_base();
virtual std::string_view name() const noexcept = 0;
virtual void fail(error err) noexcept = 0;
virtual error fail_reason() const = 0;
/// Convenience function for setting a default error if `as_has_ctx` returns
/// `nullptr` while trying to set an SSL context.
error cannot_add_ctx() {
return make_error(sec::logic_error,
"cannot add an SSL context to a config of type "
+ std::string{name()});
}
virtual has_ctx* as_has_ctx() noexcept = 0;
bool failed() const noexcept {
return name() == get_name<error>::value;
}
explicit operator bool() const noexcept {
return !failed();
}
bool operator!() const noexcept {
return failed();
}
/// The pointer to the parent @ref multiplexer.
multiplexer* mpx;
......@@ -54,8 +87,7 @@ class config_with_trait : public config_base {
public:
using trait_type = Trait;
explicit config_with_trait(multiplexer* mpx, Trait trait)
: config_base(mpx), trait(std::move(trait)) {
explicit config_with_trait(multiplexer* mpx) : config_base(mpx) {
// nop
}
......@@ -68,4 +100,53 @@ public:
Trait trait;
};
template <class Base, class... Data>
class config_impl : public Base {
public:
using super = Base;
static_assert(std::is_base_of_v<config_base, super>);
template <class From, class... Args>
explicit config_impl(From&& from, Args&&... args)
: super(std::forward<From>(from)), data(std::forward<Args>(args)...) {
if constexpr (std::is_base_of_v<config_base, std::decay_t<From>>) {
if (!from)
data = from.fail_reason();
}
}
std::variant<error, Data...> data;
template <class F>
auto visit(F&& f) {
return std::visit([&](auto& arg) { return f(arg); }, data);
}
/// Returns the name of the configuration type.
std::string_view name() const noexcept override {
auto f = [](const auto& val) {
using val_t = std::decay_t<decltype(val)>;
return get_name<val_t>::value;
};
return std::visit(f, data);
}
void fail(error err) noexcept override {
if (!std::holds_alternative<error>(data))
data = std::move(err);
}
error fail_reason() const override {
if (auto* val = std::get_if<error>(&data))
return *val;
else
return {};
}
has_ctx* as_has_ctx() noexcept override {
return has_ctx::from(data);
}
};
} // namespace caf::net::dsl
// 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/intrusive_ptr.hpp"
#include "caf/net/dsl/base.hpp"
#include "caf/net/dsl/config_base.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/ssl/context.hpp"
#include <string_view>
namespace caf::net::dsl {
/// Meta programming utility.
template <class T>
struct generic_config_tag {
using type = T;
};
/// Wraps configuration of some base parameters before we know whether the user
/// is starting a client or a server.
class generic_config {
public:
/// Configuration for a client that creates the socket on demand.
class lazy : public has_ctx {
public:
static constexpr std::string_view name = "lazy";
};
static constexpr auto lazy_v = generic_config_tag<lazy>{};
static constexpr auto fail_v = generic_config_tag<error>{};
template <class Base>
class value : public config_impl<Base, lazy> {
public:
using super = config_impl<Base, lazy>;
using super::super;
static auto make(multiplexer* mpx) {
return make_counted<value>(mpx, std::in_place_type<lazy>);
}
};
};
template <class Base>
using generic_config_value = typename generic_config::value<Base>;
template <class Base>
using generic_config_ptr = intrusive_ptr<generic_config_value<Base>>;
} // namespace caf::net::dsl
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
namespace caf::net::dsl {
/// Meta programming utility for accessing the name of a type.
template <class T>
struct get_name {
static constexpr std::string_view value = T::name;
};
template <>
struct get_name<error> {
static constexpr std::string_view value = "fail";
};
} // namespace caf::net::dsl
......@@ -20,8 +20,6 @@ namespace caf::net::dsl {
template <class Base, class Subtype>
class has_accept : public Base {
public:
using trait_type = typename Base::trait_type;
/// Creates an `accept_factory` object for the given TCP `port` and
/// `bind_address`.
///
......@@ -30,8 +28,7 @@ public:
/// @returns an `accept_factory` object initialized with the given parameters.
auto accept(uint16_t port, std::string bind_address = "") {
auto& dref = static_cast<Subtype&>(*this);
auto cfg = make_lazy_config(port, std::move(bind_address));
return dref.lift(dref.with_context(std::move(cfg)));
return dref.make(server_config::lazy_v, port, std::move(bind_address));
}
/// Creates an `accept_factory` object for the given accept socket.
......@@ -40,7 +37,7 @@ public:
/// @returns an `accept_factory` object that will start a server on `fd`.
auto accept(tcp_accept_socket fd) {
auto& dref = static_cast<Subtype&>(*this);
return dref.lift(dref.with_context(make_socket_config(fd)));
return dref.make(server_config::socket_v, fd);
}
/// Creates an `accept_factory` object for the given acceptor.
......@@ -49,45 +46,20 @@ public:
/// @returns an `accept_factory` object that will start a server on `acc`.
auto accept(ssl::acceptor acc) {
auto& dref = static_cast<Subtype&>(*this);
auto& cfg = dref.config();
auto ptr = cfg.as_has_ctx();
// The SSL acceptor has its own context, we cannot have two.
auto& ctx = dref().context();
if (ctx.has_value()) {
auto err = make_error(
sec::logic_error,
"passed an ssl::acceptor to a factory with a valid SSL context");
return dref.lift(make_fail_config(std::move(err)));
}
// Forward an already existing error.
if (ctx.error()) {
return dref.lift(make_fail_config(std::move(ctx.error())));
if (!ptr) {
return dref.make(server_config::fail_v, cfg, cfg.cannot_add_ctx());
} else if (ptr->ctx) {
auto err = make_error(sec::logic_error,
"passed an ssl::acceptor to a factory "
"with a valid SSL context");
return dref.make(server_config::fail_v, std::move(err));
} else {
ptr->ctx = std::make_shared<ssl::context>(std::move(acc.ctx()));
return accept(acc.fd());
}
// Default-constructed error means: "no SSL". Use he one from the acceptor.
ctx = std::move(acc.ctx());
return accept(acc.fd());
}
protected:
using config_base_type = dsl::config_with_trait<trait_type>;
template <class... Ts>
auto make_lazy_config(Ts&&... xs) {
using impl_t = lazy_server_config<config_base_type>;
return make_counted<impl_t>(std::forward<Ts>(xs)..., this->mpx(),
this->trait());
}
template <class... Ts>
auto make_socket_config(Ts&&... xs) {
using impl_t = socket_server_config<config_base_type>;
return make_counted<impl_t>(std::forward<Ts>(xs)..., this->mpx(),
this->trait());
}
template <class... Ts>
auto make_fail_config(Ts&&... xs) {
using impl_t = fail_server_config<config_base_type>;
return make_counted<impl_t>(std::forward<Ts>(xs)..., this->mpx(),
this->trait());
}
};
......
......@@ -19,16 +19,13 @@ namespace caf::net::dsl {
template <class Base, class Subtype>
class has_connect : public Base {
public:
using trait_type = typename Base::trait_type;
/// 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.
auto connect(std::string host, uint16_t port) {
auto& dref = static_cast<Subtype&>(*this);
return dref.make(client_config::lazy_v, this->mpx(), this->trait(),
std::move(host), port);
return dref.make(client_config::lazy_v, std::move(host), port);
}
/// Creates a `connect_factory` object for the given stream `fd`.
......@@ -36,7 +33,7 @@ public:
/// @returns a `connect_factory` object that will use the given socket.
auto connect(stream_socket fd) {
auto& dref = static_cast<Subtype&>(*this);
return dref.make(client_config::socket_v, this->mpx(), this->trait(), fd);
return dref.make(client_config::socket_v, fd);
}
/// Creates a `connect_factory` object for the given SSL `connection`.
......@@ -44,8 +41,7 @@ public:
/// @returns a `connect_factory` object that will use the given connection.
auto connect(ssl::connection conn) {
auto& dref = static_cast<Subtype&>(*this);
return dref.make(client_config::conn_v, this->mpx(), this->trait(),
std::move(conn));
return dref.make(client_config::conn_v, std::move(conn));
}
};
......
......@@ -16,18 +16,25 @@ public:
/// Sets the optional SSL context.
/// @param ctx The SSL context for encryption.
/// @returns a reference to `*this`.
Subtype& context(expected<ssl::context> ctx) {
Subtype& context(ssl::context ctx) {
auto& dref = static_cast<Subtype&>(*this);
dref.get_context() = std::move(ctx);
auto& cfg = dref.config();
if (auto* ptr = cfg.as_has_ctx())
ptr->ctx = std::make_shared<ssl::context>(std::move(ctx));
else if (cfg)
cfg.fail(cfg.cannot_add_ctx());
return dref;
}
/// Sets the optional SSL context.
/// @param ctx The SSL context for encryption.
/// @returns a reference to `*this`.
Subtype& context(ssl::context ctx) {
Subtype& context(expected<ssl::context> ctx) {
auto& dref = static_cast<Subtype&>(*this);
dref.get_context() = std::move(ctx);
if (ctx)
context(std::move(*ctx));
else if (ctx.error())
dref.config().fail(std::move(ctx).error());
return dref;
}
};
......
// 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/expected.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/stream_socket.hpp"
#include <memory>
#include <variant>
namespace caf::net::dsl {
/// Configuration for a client that uses a user-provided socket.
class has_ctx {
public:
/// SSL context for secure servers.
std::shared_ptr<ssl::context> ctx;
/// Returns a function that, when called with a @ref stream_socket, calls
/// `f` either with a new SSL connection from `ctx` or with the file the
/// file descriptor if no SSL context is defined.
template <class F>
auto connection_with_ctx(F&& f) {
return [this, g = std::forward<F>(f)](stream_socket fd) mutable {
using res_t = decltype(g(fd));
if (ctx) {
auto conn = ctx->new_connection(fd);
if (conn)
return g(*conn);
close(fd);
return res_t{std::move(conn.error())};
} else
return g(fd);
};
}
/// Returns a function that, when called with an accept socket, calls `f`
/// either with a new SSL acceptor from `ctx` or with the file the file
/// descriptor if no SSL context is defined.
template <class F>
auto acceptor_with_ctx(F&& f) {
return [this, g = std::forward<F>(f)](auto fd) mutable {
if (ctx) {
auto acc = ssl::acceptor{fd, std::move(*ctx)};
return g(acc);
} else
return g(fd);
};
}
template <class SumType>
static has_ctx* from(SumType& data) noexcept {
auto get_ptr = [](auto& val) -> has_ctx* {
using val_t = std::decay_t<decltype(val)>;
if constexpr (std::is_base_of_v<has_ctx, val_t>)
return &val;
else
return nullptr;
};
return std::visit(get_ptr, data);
}
};
} // namespace caf::net::dsl
......@@ -31,8 +31,7 @@ public:
/// @returns a `connect_factory` object initialized with the given parameters.
auto connect(const uri& endpoint) {
auto& dref = static_cast<Subtype&>(*this);
return dref.make(client_config::lazy_v, this->mpx(), this->trait(),
endpoint);
return dref.make(client_config::lazy_v, endpoint);
}
/// Creates a `connect_factory` object for the given TCP `endpoint`.
......@@ -43,8 +42,7 @@ public:
if (endpoint)
return connect(*endpoint);
auto& dref = static_cast<Subtype&>(*this);
return dref.make(client_config::fail_v, this->mpx(), this->trait(),
std::move(endpoint.error()));
return dref.make(client_config::fail_v, std::move(endpoint.error()));
}
};
......
......@@ -19,225 +19,94 @@
namespace caf::net::dsl {
/// The server config type enum class.
enum class server_config_type { lazy, socket, fail };
/// Empty tag type for meta programming.
struct server_config_tag {};
/// Base class for server configuration objects.
template <class Base>
class server_config : public Base, public server_config_tag {
public:
using trait_type = typename Base::trait_type;
class lazy;
class socket;
class fail;
friend class lazy;
friend class socket;
friend class fail;
/// Anchor type for meta programming.
using base_type = server_config;
server_config(server_config&&) = default;
server_config(const server_config&) = default;
/// Returns the server configuration type.
virtual server_config_type type() const noexcept = 0;
/// SSL context for secure servers.
std::shared_ptr<ssl::context> ctx;
/// Limits how many connections the server allows concurrently.
size_t max_connections = defaults::net::max_connections.fallback;
/// Limits how many times the server may read from the socket before allowing
/// others to read.
size_t max_consecutive_reads = defaults::middleman::max_consecutive_reads;
private:
/// Private constructor to enforce sealing.
template <class... Ts>
explicit server_config(multiplexer* mpx, Ts&&... xs)
: Base(mpx, std::forward<Ts>(xs)...) {
// nop
}
};
/// Intrusive pointer type for server configurations.
template <class Base>
using server_config_ptr = intrusive_ptr<server_config<Base>>;
/// Configuration for a server that creates the socket on demand.
template <class Base>
class server_config<Base>::lazy final : public server_config<Base> {
public:
static constexpr auto type_token = server_config_type::lazy;
using super = server_config;
template <class... Ts>
lazy(uint16_t port, std::string bind_address, multiplexer* mpx, Ts&&... xs)
: super(mpx, std::forward<Ts>(xs)...),
port(port),
bind_address(std::move(bind_address)) {
// nop
}
/// Returns the server configuration type.
server_config_type type() const noexcept override {
return type_token;
}
/// The port number to bind to.
uint16_t port = 0;
/// The address to bind to.
std::string bind_address;
/// Whether to set `SO_REUSEADDR` on the socket.
bool reuse_addr = true;
/// Meta programming utility.
template <class T>
struct server_config_tag {
using type = T;
};
/// Configuration for a server that uses a user-provided socket.
template <class Base>
class server_config<Base>::socket final : public server_config<Base> {
/// Wraps configuration parameters for starting clients.
class server_config {
public:
static constexpr auto type_token = server_config_type::socket;
using super = server_config;
template <class... Ts>
socket(tcp_accept_socket fd, multiplexer* mpx, Ts&&... xs)
: super(mpx, std::forward<Ts>(xs)...), fd(fd) {
// nop
}
~socket() override {
if (fd != invalid_socket)
close(fd);
}
/// Returns the server configuration type.
server_config_type type() const noexcept override {
return type_token;
}
/// The socket file descriptor to use.
tcp_accept_socket fd;
/// Returns the file descriptor and setting the `fd` member variable to the
/// invalid socket.
tcp_accept_socket take_fd() noexcept {
auto result = fd;
fd.id = invalid_socket_id;
return result;
}
/// Configuration for a server that creates the socket on demand.
class lazy : public has_ctx {
public:
static constexpr std::string_view name = "lazy";
lazy(uint16_t port, std::string bind_address)
: port(port), bind_address(std::move(bind_address)) {
// nop
}
/// The port number to bind to.
uint16_t port = 0;
/// The address to bind to.
std::string bind_address;
/// Whether to set `SO_REUSEADDR` on the socket.
bool reuse_addr = true;
};
static constexpr auto lazy_v = server_config_tag<lazy>{};
/// Configuration for a server that uses a user-provided socket.
class socket : public has_ctx {
public:
static constexpr std::string_view name = "socket";
explicit socket(tcp_accept_socket fd) : fd(fd) {
// nop
}
~socket() {
if (fd != invalid_socket)
close(fd);
}
/// The socket file descriptor to use.
tcp_accept_socket fd;
/// Returns the file descriptor and setting the `fd` member variable to the
/// invalid socket.
tcp_accept_socket take_fd() noexcept {
auto result = fd;
fd.id = invalid_socket_id;
return result;
}
};
static constexpr auto socket_v = server_config_tag<error>{};
static constexpr auto fail_v = server_config_tag<error>{};
template <class Base>
class value : public config_impl<Base, lazy, socket> {
public:
using super = config_impl<Base, lazy, socket>;
using super::super;
template <class T, class... Args>
static auto make(server_config_tag<T>, const Base& other, Args&&... args) {
return make_counted<value>(other, std::in_place_type<T>,
std::forward<Args>(args)...);
}
/// Configures how many reads we allow on a socket before returning to the
/// event loop.
size_t max_consecutive_reads
= caf::defaults::middleman::max_consecutive_reads;
/// Configures how many concurrent connections the server allows.
size_t max_connections = defaults::net::max_connections.fallback;
};
};
/// Wraps an error that occurred earlier in the setup phase.
template <class Base>
class server_config<Base>::fail final : public server_config<Base> {
public:
static constexpr auto type_token = server_config_type::fail;
using super = server_config;
template <class... Ts>
fail(error err, multiplexer* mpx, Ts&&... xs)
: super(mpx, std::forward<Ts>(xs)...), err(std::move(err)) {
// nop
}
fail(error err, const super& other) : super(other), err(std::move(err)) {
// nop
}
using server_config_value = server_config::value<Base>;
/// Returns the server configuration type.
server_config_type type() const noexcept override {
return type_token;
}
/// The forwarded error.
error err;
};
/// Convenience alias for the `lazy` sub-type of @ref server_config.
template <class Base>
using lazy_server_config = typename server_config<Base>::lazy;
/// Convenience alias for the `socket` sub-type of @ref server_config.
template <class Base>
using socket_server_config = typename server_config<Base>::socket;
/// Convenience alias for the `fail` sub-type of @ref server_config.
template <class Base>
using fail_server_config = typename server_config<Base>::fail;
/// Calls a function object with the actual subtype of a server configuration
/// and returns its result.
template <class F, class Base>
decltype(auto) visit(F&& f, server_config<Base>& cfg) {
auto type = cfg.type();
switch (cfg.type()) {
case server_config_type::lazy:
return f(static_cast<lazy_server_config<Base>&>(cfg));
case server_config_type::socket:
return f(static_cast<socket_server_config<Base>&>(cfg));
default:
assert(type == server_config_type::fail);
return f(static_cast<fail_server_config<Base>&>(cfg));
}
}
/// Calls a function object with the actual subtype of a server configuration.
template <class F, class Base>
decltype(auto) visit(F&& f, const server_config<Base>& cfg) {
auto type = cfg.type();
switch (cfg.type()) {
case server_config_type::lazy:
return f(static_cast<const lazy_server_config<Base>&>(cfg));
case server_config_type::socket:
return f(static_cast<const socket_server_config<Base>&>(cfg));
default:
assert(type == server_config_type::fail);
return f(static_cast<const fail_server_config<Base>&>(cfg));
}
}
/// Gets a pointer to a specific subtype of a server configuration.
template <class T, class Base>
T* get_if(server_config<Base>* cfg) {
if (T::type_token == cfg->type())
return static_cast<T*>(cfg);
return nullptr;
}
/// Gets a pointer to a specific subtype of a server configuration.
template <class T, class Base>
const T* get_if(const server_config<Base>* cfg) {
if (T::type_token == cfg->type())
return static_cast<const T*>(cfg);
return nullptr;
}
/// Creates a `fail_server_config` from another configuration object plus error.
template <class Base>
auto to_fail_config(server_config_ptr<Base> ptr, error err) {
using impl_t = fail_server_config<Base>;
return make_counted<impl_t>(std::move(err), ptr->mpx, ptr->trait);
}
/// Returns the pointer as a pointer to the `server_config` base type.
template <class T>
auto as_base_ptr(
intrusive_ptr<T> ptr,
std::enable_if_t<std::is_base_of_v<server_config_tag, T>>* = nullptr) {
return std::move(ptr).template upcast<typename T::base_type>();
}
using server_config_ptr = intrusive_ptr<server_config_value<Base>>;
} // namespace caf::net::dsl
......@@ -20,20 +20,27 @@ namespace caf::net::dsl {
template <class ConfigBase, class Derived>
class server_factory_base {
public:
using config_type = server_config<ConfigBase>;
using trait_type = typename config_type::trait_type;
using config_type = server_config_value<ConfigBase>;
using config_pointer = intrusive_ptr<config_type>;
explicit server_factory_base(config_pointer cfg) : cfg_(std::move(cfg)) {
// nop
}
server_factory_base(server_factory_base&&) = default;
server_factory_base(const server_factory_base&) = default;
server_factory_base& operator=(server_factory_base&&) = default;
server_factory_base& operator=(const server_factory_base&) = default;
explicit server_factory_base(config_pointer cfg) : cfg_(std::move(cfg)) {
// nop
}
template <class T, class... Ts>
explicit server_factory_base(dsl::server_config_tag<T> token, Ts&&... xs) {
cfg_ = config_type::make(token, std::forward<Ts>(xs)...);
}
/// Sets the callback for errors.
template <class F>
Derived& do_on_error(F callback) {
......@@ -49,9 +56,9 @@ public:
}
/// Configures whether the server creates its socket with `SO_REUSEADDR`.
Derived& reuse_addr(bool value) {
if (auto* cfg = get_if<lazy_server_config<ConfigBase>>(cfg_.get()))
cfg->reuse_addr = value;
Derived& reuse_address(bool value) {
if (auto* lazy = get_if<server_config::lazy>(&cfg_->data))
lazy->reuse_addr = value;
return dref();
}
......
......@@ -7,6 +7,7 @@
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/binary_flow_bridge.hpp"
#include "caf/disposable.hpp"
#include "caf/net/checked_socket.hpp"
#include "caf/net/dsl/client_factory_base.hpp"
#include "caf/net/lp/framing.hpp"
#include "caf/net/ssl/connection.hpp"
......@@ -65,9 +66,6 @@ private:
namespace caf::net::lp {
template <class>
class with_t;
/// Factory for the `with(...).connect(...).start(...)` DSL.
template <class Trait>
class client_factory
......@@ -130,7 +128,7 @@ private:
return detail::tcp_try_connect(std::move(addr.host), addr.port,
data.connection_timeout,
data.max_retry_count, data.retry_delay)
.and_then(data.with_ctx([this, &cfg, &on_start](auto& conn) {
.and_then(data.connection_with_ctx([this, &cfg, &on_start](auto& conn) {
return this->do_start_impl(cfg, std::move(conn), on_start);
}));
}
......@@ -139,7 +137,10 @@ private:
expected<disposable> do_start(config_type& cfg,
dsl::client_config::socket& data,
OnStart& on_start) {
return do_start_impl(cfg, data.take_fd(), on_start);
return checked_socket(data.take_fd())
.and_then(data.connection_with_ctx([this, &cfg, &on_start](auto& conn) {
return this->do_start_impl(cfg, std::move(conn), on_start);
}));
}
template <class OnStart>
......
......@@ -11,6 +11,7 @@
#include "caf/detail/connection_factory.hpp"
#include "caf/detail/shared_ssl_acceptor.hpp"
#include "caf/fwd.hpp"
#include "caf/net/checked_socket.hpp"
#include "caf/net/dsl/server_factory_base.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/lp/framing.hpp"
......@@ -74,12 +75,9 @@ private:
};
/// Specializes @ref connection_factory for the length-prefixing protocol.
template <class Transport, class Trait>
class lp_connection_factory
: public connection_factory<typename Transport::connection_handle> {
template <class Transport, class ConnHandle, class Trait>
class lp_connection_factory : public connection_factory<ConnHandle> {
public:
using connection_handle = typename Transport::connection_handle;
using accept_event = typename Trait::accept_event;
using producer_type = async::blocking_producer<accept_event>;
......@@ -93,7 +91,7 @@ public:
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
ConnHandle conn) override {
using bridge_t = lp_server_flow_bridge<Trait>;
auto bridge = bridge_t::make(mpx, producer_);
auto bridge_ptr = bridge.get();
......@@ -129,79 +127,65 @@ public:
using super::super;
using start_res_t = expected<disposable>;
/// Starts a server that accepts incoming connections with the
/// length-prefixing protocol.
template <class OnStart>
start_res_t start(OnStart on_start) {
[[nodiscard]] expected<disposable> start(OnStart on_start) {
using acceptor_resource = typename Trait::acceptor_resource;
static_assert(std::is_invocable_v<OnStart, acceptor_resource>);
auto f = [this, &on_start](auto& cfg) {
return this->do_start(cfg, on_start);
};
return visit(f, this->config());
auto& cfg = super::config();
return cfg.visit([this, &cfg, &on_start](auto& data) {
return this->do_start(cfg, data, on_start)
.or_else([&cfg](const error& err) { cfg.call_on_error(err); });
});
}
private:
template <class Factory, class AcceptHandler, class Acceptor, class OnStart>
start_res_t do_start_impl(config_type& cfg, Acceptor acc, OnStart& on_start) {
template <class Acceptor, class OnStart>
expected<disposable>
do_start_impl(config_type& cfg, Acceptor acc, OnStart& on_start) {
using conn_t = typename Acceptor::accept_result_type;
using transport_t = typename Acceptor::transport_type;
using factory_t = detail::lp_connection_factory<transport_t, conn_t, Trait>;
using impl_t = detail::accept_handler<Acceptor>;
using accept_event = typename Trait::accept_event;
auto [pull, push] = async::make_spsc_buffer_resource<accept_event>();
auto producer = async::make_blocking_producer(push.try_open());
auto factory = std::make_unique<Factory>(std::move(producer),
cfg.max_consecutive_reads);
auto impl = AcceptHandler::make(std::move(acc), std::move(factory),
cfg.max_connections);
auto factory = std::make_unique<factory_t>(std::move(producer),
cfg.max_consecutive_reads);
auto impl = impl_t::make(std::move(acc), std::move(factory),
cfg.max_connections);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(cfg.mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
cfg.mpx->start(ptr);
on_start(std::move(pull));
return start_res_t{disposable{std::move(ptr)}};
}
template <class OnStart>
start_res_t
do_start(config_type& cfg, tcp_accept_socket fd, OnStart& on_start) {
if (!cfg.ctx) {
using factory_t = detail::lp_connection_factory<stream_transport, Trait>;
using impl_t = detail::accept_handler<tcp_accept_socket, stream_socket>;
return do_start_impl<factory_t, impl_t>(cfg, fd, on_start);
}
using factory_t = detail::lp_connection_factory<ssl::transport, Trait>;
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>(cfg, acc_t{fd, cfg.ctx}, on_start);
return expected<disposable>{disposable{std::move(ptr)}};
}
template <class OnStart>
start_res_t do_start(typename config_type::socket& cfg, OnStart& on_start) {
if (cfg.fd == invalid_socket) {
auto err = make_error(
sec::runtime_error,
"server factory cannot create a server on an invalid socket");
cfg.call_on_error(err);
return start_res_t{std::move(err)};
}
return do_start(cfg, cfg.take_fd(), on_start);
expected<disposable> do_start(config_type& cfg,
dsl::server_config::socket& data,
OnStart& on_start) {
return checked_socket(data.take_fd())
.and_then(data.acceptor_with_ctx([this, &cfg, &on_start](auto& acc) {
return this->do_start_impl(cfg, std::move(acc), on_start);
}));
}
template <class OnStart>
start_res_t do_start(typename config_type::lazy& cfg, OnStart& on_start) {
auto fd = make_tcp_accept_socket(cfg.port, cfg.bind_address,
cfg.reuse_addr);
if (!fd) {
cfg.call_on_error(fd.error());
return start_res_t{std::move(fd.error())};
}
return do_start(cfg, *fd, on_start);
expected<disposable> do_start(config_type& cfg,
dsl::server_config::lazy& data,
OnStart& on_start) {
return make_tcp_accept_socket(data.port, data.bind_address, data.reuse_addr)
.and_then(data.acceptor_with_ctx([this, &cfg, &on_start](auto& acc) {
return this->do_start_impl(cfg, std::move(acc), on_start);
}));
}
template <class OnStart>
start_res_t do_start(typename config_type::fail& cfg, OnStart&) {
cfg.call_on_error(cfg.err);
return start_res_t{std::move(cfg.err)};
expected<disposable> do_start(config_type&, error& err, OnStart&) {
return expected<disposable>{std::move(err)};
}
};
......
......@@ -5,6 +5,8 @@
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/dsl/base.hpp"
#include "caf/net/dsl/generic_config.hpp"
#include "caf/net/dsl/has_accept.hpp"
#include "caf/net/dsl/has_connect.hpp"
#include "caf/net/dsl/has_context.hpp"
......@@ -21,54 +23,43 @@ namespace caf::net::lp {
/// Entry point for the `with(...)` DSL.
template <class Trait>
class with_t : public extend<dsl::base<Trait>, with_t<Trait>>::template //
class with_t : public extend<dsl::base, with_t<Trait>>::template //
with<dsl::has_accept, dsl::has_connect, dsl::has_context> {
public:
using config_type = dsl::generic_config_value<dsl::config_with_trait<Trait>>;
template <class... Ts>
explicit with_t(multiplexer* mpx, Ts&&... xs)
: mpx_(mpx), trait_(std::forward<Ts>(xs)...), ctx_(error{}) {
explicit with_t(multiplexer* mpx) : config_(config_type::make(mpx)) {
// nop
}
with_t(with_t&&) noexcept = default;
with_t(const with_t&) noexcept = default;
with_t& operator=(const with_t&) noexcept = default;
with_t& operator=(with_t&&) noexcept = default;
multiplexer* mpx() const noexcept override {
return mpx_;
}
const Trait& trait() const noexcept override {
return trait_;
}
with_t& operator=(const with_t&) noexcept = default;
/// @private
using config_base_type = dsl::config_with_trait<Trait>;
config_type& config() {
return *config_;
}
/// @private
server_factory<Trait> lift(dsl::server_config_ptr<config_base_type> cfg) {
return server_factory<Trait>{std::move(cfg)};
template <class T, class... Ts>
auto make(dsl::server_config_tag<T> token, Ts&&... xs) {
return server_factory<Trait>{token, *config_, std::forward<Ts>(xs)...};
}
/// @private
template <class T, class... Ts>
auto make(dsl::client_config_tag<T> tag, Ts&&... xs) {
return client_factory<Trait>{tag, std::forward<Ts>(xs)...};
auto make(dsl::client_config_tag<T> token, Ts&&... xs) {
return client_factory<Trait>{token, *config_, std::forward<Ts>(xs)...};
}
private:
expected<ssl::context>& get_context_impl() noexcept override {
return ctx_;
}
/// Pointer to multiplexer that runs the protocol stack.
multiplexer* mpx_;
/// User-defined trait for configuring serialization.
Trait trait_;
/// The optional SSL context.
expected<ssl::context> ctx_;
intrusive_ptr<config_type> config_;
};
template <class Trait = binary::default_trait>
......
......@@ -10,9 +10,11 @@
#include "caf/detail/connection_factory.hpp"
#include "caf/detail/shared_ssl_acceptor.hpp"
#include "caf/fwd.hpp"
#include "caf/net/checked_socket.hpp"
#include "caf/net/dsl/server_config.hpp"
#include "caf/net/dsl/server_factory_base.hpp"
#include "caf/net/http/server.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"
......@@ -25,20 +27,17 @@
namespace caf::detail {
template <class Transport>
class prometheus_conn_factory
: public connection_factory<typename Transport::connection_handle> {
template <class Transport, class ConnHandle>
class prom_conn_factory : public connection_factory<ConnHandle> {
public:
using state_ptr = net::prometheus::server::scrape_state_ptr;
using connection_handle = typename Transport::connection_handle;
explicit prometheus_conn_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
explicit prom_conn_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
ConnHandle conn) override {
auto prom_serv = net::prometheus::server::make(ptr_);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto transport = Transport::make(std::move(conn), std::move(http_serv));
......@@ -53,123 +52,63 @@ private:
namespace caf::net::prometheus {
class with_t;
/// Entry point for the `with(...).accept(...).start()` DSL.
class accept_factory {
class server_factory
: public dsl::server_factory_base<dsl::config_base, server_factory> {
public:
friend class with_t;
accept_factory(accept_factory&&) = default;
using super = dsl::server_factory_base<dsl::config_base, server_factory>;
accept_factory(const accept_factory&) = delete;
using config_type = typename super::config_type;
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;
}
using super::super;
/// Starts the Prometheus service in the background.
disposable start() {
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);
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);
}
default:
return {};
}
[[nodiscard]] expected<disposable> start() {
auto& cfg = super::config();
return cfg.visit([this, &cfg](auto& data) {
return do_start(cfg, data).or_else([&cfg](const error& err) { //
cfg.call_on_error(err);
});
});
}
private:
struct config {
uint16_t port;
std::string address;
bool reuse_addr;
};
explicit accept_factory(actor_system* sys) : sys_(sys) {
// nop
}
template <class Factory, class AcceptHandler, class Acceptor>
disposable do_start_impl(Acceptor&& acc) {
auto mpx = multiplexer::from(*sys_);
auto registry = &sys_->metrics();
template <class Acceptor>
expected<disposable> do_start_impl(config_type& cfg, Acceptor acc) {
using conn_t = typename Acceptor::accept_result_type;
using transport_t = typename Acceptor::transport_type;
using factory_t = detail::prom_conn_factory<transport_t, conn_t>;
using impl_t = detail::accept_handler<Acceptor>;
auto* mpx = cfg.mpx;
auto* registry = &mpx->system().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 factory = std::make_unique<factory_t>(std::move(state));
auto impl = impl_t::make(std::forward<Acceptor>(acc), std::move(factory),
cfg.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>;
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>;
return do_start_impl<factory_t, impl_t>(acc_t{fd, ctx_});
return expected<disposable>{mgr->as_disposable()};
}
void set_ssl(ssl::context ctx) {
ctx_ = std::make_shared<ssl::context>(std::move(ctx));
expected<disposable> do_start(config_type& cfg,
dsl::server_config::socket& data) {
return checked_socket(data.take_fd())
.and_then(data.acceptor_with_ctx([this, &cfg](auto& acc) {
return this->do_start_impl(cfg, std::move(acc));
}));
}
void init(uint16_t port, std::string address, bool reuse_addr) {
state_ = config{port, std::move(address), reuse_addr};
expected<disposable> do_start(config_type& cfg,
dsl::server_config::lazy& data) {
return make_tcp_accept_socket(data.port, data.bind_address, data.reuse_addr)
.and_then(data.acceptor_with_ctx([this, &cfg](auto& acc) {
return this->do_start_impl(cfg, std::move(acc));
}));
}
void init(tcp_accept_socket fd) {
state_ = fd;
expected<disposable> do_start(config_type&, error& err) {
return expected<disposable>{std::move(err)};
}
/// Pointer to the hosting actor system.
actor_system* sys_;
/// 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_;
std::shared_ptr<ssl::context> ctx_;
};
} // namespace caf::net::prometheus
......@@ -5,7 +5,10 @@
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/prometheus/accept_factory.hpp"
#include "caf/net/dsl/generic_config.hpp"
#include "caf/net/dsl/has_accept.hpp"
#include "caf/net/dsl/has_context.hpp"
#include "caf/net/prometheus/server_factory.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/tcp_accept_socket.hpp"
......@@ -15,9 +18,13 @@
namespace caf::net::prometheus {
/// Entry point for the `with(...).accept(...).start()` DSL.
class with_t {
class with_t : public extend<dsl::base, with_t>::template //
with<dsl::has_accept, dsl::has_context> {
public:
explicit with_t(actor_system* sys) : sys_(sys) {
using config_type = dsl::generic_config_value<dsl::config_base>;
template <class... Ts>
explicit with_t(multiplexer* mpx) : config_(config_type::make(mpx)) {
// nop
}
......@@ -25,66 +32,27 @@ public:
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 accept(uint16_t port, std::string bind_address = "",
bool reuse_addr = true) {
accept_factory factory{sys_};
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 accept(tcp_accept_socket fd) {
accept_factory factory{sys_};
factory.init(fd);
return factory;
/// @private
config_type& config() {
return *config_;
}
/// 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 accept(ssl::acceptor acc) {
accept_factory factory{sys_};
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 accept(ssl::context ctx, uint16_t port,
std::string bind_address = "", bool reuse_addr = true) {
accept_factory factory{sys_};
factory.set_ssl(std::move(std::move(ctx)));
factory.init(port, std::move(bind_address), std::move(reuse_addr));
return factory;
/// @private
template <class T, class... Ts>
auto make(dsl::server_config_tag<T> token, Ts&&... xs) {
return server_factory{token, *config_, std::forward<Ts>(xs)...};
}
private:
/// Pointer to context.
actor_system* sys_;
intrusive_ptr<config_type> config_;
};
inline with_t with(actor_system& sys) {
return with_t{&sys};
return with_t{multiplexer::from(sys)};
}
inline with_t with(multiplexer* mpx) {
return with_t{mpx};
}
} // namespace caf::net::prometheus
......@@ -21,6 +21,8 @@ public:
using transport_type = transport;
using accept_result_type = connection;
// -- constructors, destructors, and assignment operators --------------------
acceptor() = delete;
......
......@@ -20,6 +20,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
using super::super;
using transport_type = stream_transport;
using accept_result_type = tcp_stream_socket;
};
/// Creates a new TCP socket to accept connections on a given port.
......
......@@ -7,6 +7,7 @@
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/ws_flow_bridge.hpp"
#include "caf/disposable.hpp"
#include "caf/net/checked_socket.hpp"
#include "caf/net/dsl/client_factory_base.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/tcp_stream_socket.hpp"
......@@ -68,8 +69,11 @@ class client_factory_config : public dsl::config_with_trait<Trait> {
public:
using super = dsl::config_with_trait<Trait>;
client_factory_config(multiplexer* mpx, Trait trait)
: super(mpx, std::move(trait)) {
explicit client_factory_config(multiplexer* mpx) : super(mpx) {
hs.endpoint("/");
}
explicit client_factory_config(const super& other) : super(other) {
hs.endpoint("/");
}
......@@ -146,7 +150,7 @@ private:
return detail::tcp_try_connect(std::move(addr.host), addr.port,
data.connection_timeout,
data.max_retry_count, data.retry_delay)
.and_then(data.with_ctx([this, &cfg, &on_start](auto& conn) {
.and_then(data.connection_with_ctx([this, &cfg, &on_start](auto& conn) {
return this->do_start_impl(cfg, std::move(conn), on_start);
}));
}
......@@ -193,7 +197,7 @@ private:
return detail::tcp_try_connect(std::move(host), port,
data.connection_timeout,
data.max_retry_count, data.retry_delay)
.and_then(data.with_ctx([this, &cfg, &on_start](auto& conn) {
.and_then(data.connection_with_ctx([this, &cfg, &on_start](auto& conn) {
return this->do_start_impl(cfg, std::move(conn), on_start);
}));
}
......@@ -212,9 +216,12 @@ private:
expected<disposable> do_start(config_type& cfg,
dsl::client_config::socket& data,
OnStart& on_start) {
return sanity_check(cfg).and_then([&] { //
return do_start_impl(cfg, data.take_fd(), on_start);
});
return sanity_check(cfg)
.transform([&data] { return data.take_fd(); })
.and_then(check_socket)
.and_then(data.connection_with_ctx([this, &cfg, &on_start](auto& conn) {
return this->do_start_impl(cfg, std::move(conn), on_start);
}));
}
template <class OnStart>
......
......@@ -90,12 +90,9 @@ private:
};
/// Specializes @ref connection_factory for the WebSocket protocol.
template <class Transport, class Trait, class... Ts>
class ws_connection_factory
: public connection_factory<typename Transport::connection_handle> {
template <class Transport, class ConnHandle, class Trait, class... Ts>
class ws_connection_factory : public connection_factory<ConnHandle> {
public:
using connection_handle = typename Transport::connection_handle;
using ws_acceptor_t = net::web_socket::acceptor<Ts...>;
using on_request_cb_type
......@@ -117,7 +114,7 @@ public:
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
ConnHandle conn) override {
if (producer_->canceled()) {
// TODO: stop the caller?
return nullptr;
......@@ -164,8 +161,6 @@ public:
// nop
}
using start_res_t = expected<disposable>;
/// The input type of the application, i.e., what that flows from the
/// WebSocket to the application layer.
using input_type = typename Trait::input_type;
......@@ -189,74 +184,62 @@ public:
/// Starts a server that accepts incoming connections with the WebSocket
/// protocol.
template <class OnStart>
start_res_t start(OnStart on_start) {
expected<disposable> start(OnStart on_start) {
static_assert(std::is_invocable_v<OnStart, acceptor_resource>);
auto f = [this, &on_start](auto& cfg) {
return this->do_start(cfg, on_start);
};
return visit(f, this->config());
auto& cfg = super::config();
return cfg.visit([this, &cfg, &on_start](auto& data) {
return this->do_start(cfg, data, on_start)
.or_else([&cfg](const error& err) { cfg.call_on_error(err); });
});
}
private:
template <class Factory, class AcceptHandler, class Acceptor, class OnStart>
start_res_t do_start_impl(config_type& cfg, Acceptor acc, OnStart& on_start) {
template <class Acceptor, class OnStart>
expected<disposable>
do_start_impl(config_type& cfg, Acceptor acc, OnStart& on_start) {
using conn_t = typename Acceptor::accept_result_type;
using transport_t = typename Acceptor::transport_type;
using factory_t
= detail::ws_connection_factory<transport_t, conn_t, Trait, Ts...>;
using impl_t = detail::accept_handler<Acceptor>;
using producer_t = async::blocking_producer<accept_event>;
auto [pull, push] = async::make_spsc_buffer_resource<accept_event>();
auto producer = std::make_shared<producer_t>(producer_t{push.try_open()});
auto factory = std::make_unique<Factory>(on_request_, std::move(producer),
cfg.max_consecutive_reads);
auto impl = AcceptHandler::make(std::move(acc), std::move(factory),
cfg.max_connections);
auto factory = std::make_unique<factory_t>(on_request_, std::move(producer),
cfg.max_consecutive_reads);
auto impl = impl_t::make(std::move(acc), std::move(factory),
cfg.max_connections);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(cfg.mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
cfg.mpx->start(ptr);
on_start(std::move(pull));
return start_res_t{disposable{std::move(ptr)}};
}
template <class OnStart>
start_res_t
do_start(config_type& cfg, tcp_accept_socket fd, OnStart& on_start) {
using detail::ws_connection_factory;
if (!cfg.ctx) {
using factory_t = ws_connection_factory<stream_transport, Trait, Ts...>;
using impl_t = detail::accept_handler<tcp_accept_socket, stream_socket>;
return do_start_impl<factory_t, impl_t>(cfg, fd, on_start);
}
using factory_t = ws_connection_factory<ssl::transport, Trait, Ts...>;
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>(cfg, acc_t{fd, cfg.ctx}, on_start);
return expected<disposable>{disposable{std::move(ptr)}};
}
template <class OnStart>
start_res_t do_start(typename config_type::socket& cfg, OnStart& on_start) {
if (cfg.fd == invalid_socket) {
auto err = make_error(
sec::runtime_error,
"server factory cannot create a server on an invalid socket");
cfg.call_on_error(err);
return start_res_t{std::move(err)};
}
return do_start(cfg, cfg.take_fd(), on_start);
expected<disposable> do_start(config_type& cfg,
dsl::server_config::socket& data,
OnStart& on_start) {
return checked_socket(data.take_fd())
.and_then(data.acceptor_with_ctx([this, &cfg, &on_start](auto& acc) {
return this->do_start_impl(cfg, std::move(acc), on_start);
}));
}
template <class OnStart>
start_res_t do_start(typename config_type::lazy& cfg, OnStart& on_start) {
auto fd = make_tcp_accept_socket(cfg.port, cfg.bind_address,
cfg.reuse_addr);
if (!fd) {
cfg.call_on_error(fd.error());
return start_res_t{std::move(fd.error())};
}
return do_start(cfg, *fd, on_start);
expected<disposable> do_start(config_type& cfg,
dsl::server_config::lazy& data,
OnStart& on_start) {
return make_tcp_accept_socket(data.port, data.bind_address, data.reuse_addr)
.and_then(data.acceptor_with_ctx([this, &cfg, &on_start](auto& acc) {
return this->do_start_impl(cfg, std::move(acc), on_start);
}));
}
template <class OnStart>
start_res_t do_start(typename config_type::fail& cfg, OnStart&) {
cfg.call_on_error(cfg.err);
return start_res_t{std::move(cfg.err)};
expected<disposable> do_start(config_type&, error& err, OnStart&) {
return expected<disposable>{std::move(err)};
}
on_request_cb_type on_request_;
......
......@@ -5,6 +5,7 @@
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/dsl/generic_config.hpp"
#include "caf/net/dsl/has_accept.hpp"
#include "caf/net/dsl/has_context.hpp"
#include "caf/net/dsl/has_uri_connect.hpp"
......@@ -22,12 +23,13 @@ namespace caf::net::web_socket {
/// Entry point for the `with(...)` DSL.
template <class Trait>
class with_t : public extend<dsl::base<Trait>, with_t<Trait>>::template //
class with_t : public extend<dsl::base, with_t<Trait>>::template //
with<dsl::has_accept, dsl::has_uri_connect, dsl::has_context> {
public:
using config_type = dsl::generic_config_value<dsl::config_with_trait<Trait>>;
template <class... Ts>
explicit with_t(multiplexer* mpx, Ts&&... xs)
: mpx_(mpx), trait_(std::forward<Ts>(xs)...), ctx_(error{}) {
explicit with_t(multiplexer* mpx) : config_(config_type::make(mpx)) {
// nop
}
......@@ -35,39 +37,25 @@ public:
with_t& operator=(const with_t&) noexcept = default;
multiplexer* mpx() const noexcept override {
return mpx_;
}
const Trait& trait() const noexcept override {
return trait_;
/// @private
config_type& config() {
return *config_;
}
/// @private
template <class T>
auto lift(intrusive_ptr<T> cfg) {
return has_on_request<Trait>{std::move(cfg)};
template <class T, class... Ts>
auto make(dsl::server_config_tag<T> token, Ts&&... xs) {
return has_on_request<Trait>{token, *config_, std::forward<Ts>(xs)...};
}
/// @private
template <class T, class... Ts>
auto make(dsl::client_config_tag<T> token, Ts&&... xs) {
return client_factory<Trait>{token, std::forward<Ts>(xs)...};
return client_factory<Trait>{token, *config_, std::forward<Ts>(xs)...};
}
private:
expected<ssl::context>& get_context_impl() noexcept override {
return ctx_;
}
/// Pointer to multiplexer that runs the protocol stack.
multiplexer* mpx_;
/// User-defined trait for configuring serialization.
Trait trait_;
/// The optional SSL context.
expected<ssl::context> ctx_;
intrusive_ptr<config_type> config_;
};
template <class Trait = default_trait>
......
......@@ -34,12 +34,12 @@ bool inspect(Inspector& f, prom_config& x) {
}
void launch_prom_server(actor_system& sys, const prom_config& cfg) {
prometheus::with(sys)
.accept(cfg.port, cfg.address, cfg.reuse_address)
.do_on_error([](const error& err) {
CAF_LOG_WARNING("failed to start Prometheus server: " << err);
})
.start();
auto server = prometheus::with(sys)
.accept(cfg.port, cfg.address)
.reuse_address(cfg.reuse_address)
.start();
if (!server)
CAF_LOG_WARNING("failed to start Prometheus server: " << server.error());
}
void launch_background_tasks(actor_system& sys) {
......
......@@ -271,7 +271,7 @@ SCENARIO("lp::with(...).connect(...) translates between flows and socket I/O") {
.subscribe(push);
});
});
REQUIRE(conn);
conn.or_else([](const error& err) { FAIL("connect failed:" << err); });
scoped_actor self{sys};
self->wait_for(hdl);
if (CHECK_EQ(buf->size(), 5u)) {
......
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