Unverified Commit 428eaaa7 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1524

Add autobahn driver
parents cd8363aa c64964ea
......@@ -407,6 +407,29 @@ function(caf_add_component name)
caf_export_and_install_lib(${name})
endfunction()
# -- convenience function for adding executable targets of tests ---------------
# Usage:
# caf_add_test_executable(
# foo
# DEPENDENCIES
# ...
# SOURCES
# ...
# )
function(caf_add_test_executable name)
if(NOT CAF_ENABLE_TESTING)
return()
endif()
set(varargs DEPENDENCIES SOURCES)
cmake_parse_arguments(args "" "" "${varargs}" ${ARGN})
if(NOT args_SOURCES)
message(FATAL_ERROR "Cannot add a CAF test executable without sources.")
endif()
add_executable(${name} ${args_SOURCES})
target_link_libraries(${name} PRIVATE CAF::internal ${args_DEPENDENCIES})
endfunction()
# -- build all components the user asked for -----------------------------------
add_subdirectory(libcaf_core)
......
......@@ -3,9 +3,9 @@ ignore:
- examples
- libcaf_core/caf/decorator
- libcaf_core/src/decorator
- libcaf_core/test
- libcaf_io/test
- libcaf_net/test
- libcaf_core/tests
- libcaf_io/tests
- libcaf_net/tests
- libcaf_openssl
- libcaf_test
- tools
......
......@@ -122,3 +122,10 @@ caf_add_component(
net.web_socket.framing
net.web_socket.handshake
net.web_socket.server)
caf_add_test_executable(
caf-net-autobahn-driver
SOURCES
tests/drivers/autobahn.cpp
DEPENDENCIES
CAF::net)
......@@ -94,9 +94,9 @@ private:
ws_res_type ws_resources;
};
/// Specializes @ref connection_factory for the WebSocket protocol.
/// Specializes @ref connection_factory for flows over the WebSocket protocol.
template <class Transport, class Trait, class... Ts>
class ws_connection_factory
class ws_flow_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using ws_acceptor_t = net::web_socket::acceptor<Ts...>;
......@@ -111,7 +111,7 @@ public:
using connection_handle = typename Transport::connection_handle;
ws_connection_factory(on_request_cb_type on_request,
ws_flow_conn_factory(on_request_cb_type on_request,
shared_producer_type producer,
size_t max_consecutive_reads)
: on_request_(std::move(on_request)),
......@@ -144,6 +144,34 @@ private:
size_t max_consecutive_reads_;
};
/// Specializes @ref connection_factory for custom upper layer implementations.
template <class Transport, class MakeApp>
class ws_simple_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using connection_handle = typename Transport::connection_handle;
ws_simple_conn_factory(MakeApp app_factory, size_t max_consecutive_reads)
: max_consecutive_reads_(max_consecutive_reads),
app_factory_(std::move(app_factory)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto app = app_factory_();
auto ws = net::web_socket::server::make(std::move(app));
auto transport = Transport::make(std::move(conn), std::move(ws));
transport->max_consecutive_reads(max_consecutive_reads_);
transport->active_policy().accept();
return net::socket_manager::make(mpx, std::move(transport));
}
private:
size_t max_consecutive_reads_;
MakeApp app_factory_;
};
} // namespace caf::detail
namespace caf::net::web_socket {
......@@ -189,20 +217,35 @@ public:
/// protocol.
template <class OnStart>
expected<disposable> start(OnStart on_start) {
static_assert(std::is_invocable_v<OnStart, acceptor_resource>);
auto& cfg = super::config();
if constexpr (std::is_invocable_v<OnStart, acceptor_resource>) {
return cfg.visit([this, &cfg, &on_start](auto& data) {
return this->do_start(cfg, data, on_start, flow_impl_token{})
.or_else([&cfg](const error& err) { cfg.call_on_error(err); });
});
} else {
using server_t = web_socket::upper_layer::server;
using server_ptr_t = std::unique_ptr<server_t>;
static_assert(std::is_invocable_r_v<server_ptr_t, OnStart>,
"OnStart must have signature 'void(acceptor_resource)' or "
"unique_ptr<web_socket::upper_layer::server>()'");
return cfg.visit([this, &cfg, &on_start](auto& data) {
return this->do_start(cfg, data, on_start)
return this->do_start(cfg, data, on_start, custom_impl_token{})
.or_else([&cfg](const error& err) { cfg.call_on_error(err); });
});
}
}
private:
struct flow_impl_token {};
struct custom_impl_token {};
template <class Acceptor, class OnStart>
expected<disposable>
do_start_impl(config_type& cfg, Acceptor acc, OnStart& on_start) {
expected<disposable> do_start_impl(config_type& cfg, Acceptor acc,
OnStart& on_start, flow_impl_token) {
using transport_t = typename Acceptor::transport_type;
using factory_t = detail::ws_connection_factory<transport_t, Trait, Ts...>;
using factory_t = detail::ws_flow_conn_factory<transport_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>();
......@@ -219,28 +262,45 @@ private:
return expected<disposable>{disposable{std::move(ptr)}};
}
template <class OnStart>
template <class Acceptor, class MakeApp>
expected<disposable> do_start_impl(config_type& cfg, Acceptor acc,
MakeApp& make_app, custom_impl_token) {
using transport_t = typename Acceptor::transport_type;
using factory_t = detail::ws_simple_conn_factory<transport_t, MakeApp>;
using impl_t = detail::accept_handler<Acceptor>;
auto factory = std::make_unique<factory_t>(std::move(make_app),
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);
return expected<disposable>{disposable{std::move(ptr)}};
}
template <class OnStart, class Token>
expected<disposable> do_start(config_type& cfg,
dsl::server_config::socket& data,
OnStart& on_start) {
OnStart& on_start, Token) {
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);
return this->do_start_impl(cfg, std::move(acc), on_start, Token{});
}));
}
template <class OnStart>
template <class OnStart, class Token>
expected<disposable> do_start(config_type& cfg,
dsl::server_config::lazy& data,
OnStart& on_start) {
OnStart& on_start, Token) {
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);
return this->do_start_impl(cfg, std::move(acc), on_start, Token{});
}));
}
template <class OnStart>
expected<disposable> do_start(config_type&, error& err, OnStart&) {
template <class OnStart, class Token>
expected<disposable> do_start(config_type&, error& err, OnStart&, Token) {
return expected<disposable>{std::move(err)};
}
......
// Simple WebSocket server that sends everything it receives back to the sender.
#include "caf/net/middleman.hpp"
#include "caf/net/web_socket/upper_layer.hpp"
#include "caf/net/web_socket/with.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include <iostream>
#include <utility>
using namespace std::literals;
namespace ssl = caf::net::ssl;
namespace ws = caf::net::web_socket;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 8080;
static constexpr size_t default_max_connections = 128;
// -- configuration setup ------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<size_t>("max-connections,m", "limit for concurrent clients");
opt_group{custom_options_, "tls"} //
.add<std::string>("key-file,k", "path to the private key file")
.add<std::string>("cert-file,c", "path to the certificate file");
}
};
// -- synchronous web server implementation ------------------------------------
class web_socket_app : public caf::net::web_socket::upper_layer::server {
public:
static auto make() {
return std::make_unique<web_socket_app>();
}
caf::error start(caf::net::web_socket::lower_layer* ll) override {
down = ll;
down->request_messages();
return caf::none;
}
caf::error accept(const caf::net::http::request_header&) override {
return caf::none;
}
void prepare_send() override {
// nop
}
bool done_sending() override {
return true;
}
void abort(const caf::error& reason) override {
CAF_LOG_ERROR(reason);
}
ptrdiff_t consume_text(std::string_view text) override {
down->begin_text_message();
auto& buffer = down->text_message_buffer();
buffer.insert(buffer.end(), text.begin(), text.end());
down->end_text_message();
return static_cast<ptrdiff_t>(text.size());
}
ptrdiff_t consume_binary(caf::byte_span bytes) override {
down->begin_binary_message();
auto& buffer = down->binary_message_buffer();
buffer.insert(buffer.end(), bytes.begin(), bytes.end());
down->end_binary_message();
return static_cast<ptrdiff_t>(bytes.size());
}
caf::net::web_socket::lower_layer* down = nullptr;
};
int caf_main(caf::actor_system& sys, const config& cfg) {
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
auto pem = ssl::format::pem;
auto key_file = caf::get_as<std::string>(cfg, "tls.key-file");
auto cert_file = caf::get_as<std::string>(cfg, "tls.cert-file");
auto max_connections = caf::get_or(cfg, "max-connections",
default_max_connections);
if (!key_file != !cert_file) {
std::cerr << "*** inconsistent TLS config: declare neither file or both\n";
return EXIT_FAILURE;
}
// Open up a TCP port for incoming connections and start the server.
auto server
= ws::with(sys)
// Optionally enable TLS.
.context(ssl::context::enable(key_file && cert_file)
.and_then(ssl::emplace_server(ssl::tls::v1_2))
.and_then(ssl::use_private_key_file(key_file, pem))
.and_then(ssl::use_certificate_file(cert_file, pem)))
// Bind to the user-defined port.
.accept(port)
// Limit how many clients may be connected at any given time.
.max_connections(max_connections)
// Add handler for incoming connections.
.on_request([](ws::acceptor<>& acc) {
// Ignore all header fields and accept the connection.
acc.accept();
})
// Create instances of our app to handle incoming connections.
.start([] { return web_socket_app::make(); });
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Wait until the server shuts down.
while (server->valid()) {
std::this_thread::sleep_for(1s);
}
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
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