Commit c2a60ac7 authored by Dominik Charousset's avatar Dominik Charousset

Implement missing autobahn driver scaffolding

parent 9912b5b4
...@@ -407,6 +407,29 @@ function(caf_add_component name) ...@@ -407,6 +407,29 @@ function(caf_add_component name)
caf_export_and_install_lib(${name}) caf_export_and_install_lib(${name})
endfunction() 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 ----------------------------------- # -- build all components the user asked for -----------------------------------
add_subdirectory(libcaf_core) add_subdirectory(libcaf_core)
......
...@@ -122,3 +122,10 @@ caf_add_component( ...@@ -122,3 +122,10 @@ caf_add_component(
net.web_socket.framing net.web_socket.framing
net.web_socket.handshake net.web_socket.handshake
net.web_socket.server) 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: ...@@ -94,9 +94,9 @@ private:
ws_res_type ws_resources; 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> template <class Transport, class Trait, class... Ts>
class ws_connection_factory class ws_flow_conn_factory
: public connection_factory<typename Transport::connection_handle> { : public connection_factory<typename Transport::connection_handle> {
public: public:
using ws_acceptor_t = net::web_socket::acceptor<Ts...>; using ws_acceptor_t = net::web_socket::acceptor<Ts...>;
...@@ -111,7 +111,7 @@ public: ...@@ -111,7 +111,7 @@ public:
using connection_handle = typename Transport::connection_handle; 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, shared_producer_type producer,
size_t max_consecutive_reads) size_t max_consecutive_reads)
: on_request_(std::move(on_request)), : on_request_(std::move(on_request)),
...@@ -144,6 +144,35 @@ private: ...@@ -144,6 +144,35 @@ private:
size_t max_consecutive_reads_; size_t max_consecutive_reads_;
}; };
/// Specializes @ref connection_factory for custom upper layer implementation.
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();
auto mgr = net::socket_manager::make(mpx, std::move(transport));
return mgr;
}
private:
size_t max_consecutive_reads_;
MakeApp app_factory_;
};
} // namespace caf::detail } // namespace caf::detail
namespace caf::net::web_socket { namespace caf::net::web_socket {
...@@ -189,20 +218,35 @@ public: ...@@ -189,20 +218,35 @@ public:
/// protocol. /// protocol.
template <class OnStart> template <class OnStart>
expected<disposable> start(OnStart on_start) { expected<disposable> start(OnStart on_start) {
static_assert(std::is_invocable_v<OnStart, acceptor_resource>);
auto& cfg = super::config(); auto& cfg = super::config();
if constexpr (std::is_invocable_v<OnStart, acceptor_resource>) {
return cfg.visit([this, &cfg, &on_start](auto& data) { 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, 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, custom_impl_token{})
.or_else([&cfg](const error& err) { cfg.call_on_error(err); }); .or_else([&cfg](const error& err) { cfg.call_on_error(err); });
}); });
} }
}
private: private:
struct flow_impl_token {};
struct custom_impl_token {};
template <class Acceptor, class OnStart> template <class Acceptor, class OnStart>
expected<disposable> expected<disposable> do_start_impl(config_type& cfg, Acceptor acc,
do_start_impl(config_type& cfg, Acceptor acc, OnStart& on_start) { OnStart& on_start, flow_impl_token) {
using transport_t = typename Acceptor::transport_type; 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 impl_t = detail::accept_handler<Acceptor>;
using producer_t = async::blocking_producer<accept_event>; using producer_t = async::blocking_producer<accept_event>;
auto [pull, push] = async::make_spsc_buffer_resource<accept_event>(); auto [pull, push] = async::make_spsc_buffer_resource<accept_event>();
...@@ -219,28 +263,45 @@ private: ...@@ -219,28 +263,45 @@ private:
return expected<disposable>{disposable{std::move(ptr)}}; 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, expected<disposable> do_start(config_type& cfg,
dsl::server_config::socket& data, dsl::server_config::socket& data,
OnStart& on_start) { OnStart& on_start, Token) {
return checked_socket(data.take_fd()) return checked_socket(data.take_fd())
.and_then(data.acceptor_with_ctx([this, &cfg, &on_start](auto& acc) { .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, expected<disposable> do_start(config_type& cfg,
dsl::server_config::lazy& data, 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) 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) { .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&, error& err, OnStart&) { expected<disposable> do_start(config_type&, error& err, OnStart&, Token) {
return expected<disposable>{std::move(err)}; return expected<disposable>{std::move(err)};
} }
......
// Simple WebSocket server that sends everything it receives back to the sender. // Simple WebSocket server that sends everything it receives back to the sender.
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/web_socket/acceptor.hpp"
#include "caf/net/web_socket/framing.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
#include "caf/net/web_socket/upper_layer.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.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp" #include "caf/caf_main.hpp"
#include <cstdio>
#include <iostream> #include <iostream>
#include <utility> #include <utility>
using namespace std::literals;
namespace ssl = caf::net::ssl;
namespace ws = caf::net::web_socket;
// -- constants ---------------------------------------------------------------- // -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 8080; static constexpr uint16_t default_port = 8080;
...@@ -44,7 +47,7 @@ public: ...@@ -44,7 +47,7 @@ public:
return caf::none; return caf::none;
} }
caf::error accept(const caf::net::http::request_header& hdr) override { caf::error accept(const caf::net::http::request_header&) override {
return caf::none; return caf::none;
} }
...@@ -80,23 +83,48 @@ public: ...@@ -80,23 +83,48 @@ public:
}; };
int caf_main(caf::actor_system& sys, const config& cfg) { int caf_main(caf::actor_system& sys, const config& cfg) {
namespace ws = caf::net::web_socket;
// Read the configuration. // Read the configuration.
auto interval = caf::get_or(cfg, "interval", caf::timespan{1s});
auto port = caf::get_or(cfg, "port", default_port); 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", auto max_connections = caf::get_or(cfg, "max-connections",
default_max_connections); default_max_connections);
if (!key_file != !cert_file) {
auto app_layer = web_socket_app::make(); std::cerr << "*** inconsistent TLS config: declare neither file or both\n";
auto app = app_layer.get(); return EXIT_FAILURE;
auto uut_layer }
= caf::net::web_socket::framing::make_server(std::move(app_layer)); // Open up a TCP port for incoming connections and start the server.
auto uut = uut_layer.get(); using trait = ws::default_trait;
auto server
auto transport = ws::with(sys)
= caf::net::octet_stream::transport::make(std::move(uut_layer)); // Optionally enable TLS.
.context(ssl::context::enable(key_file && cert_file)
// transport->configure_read(caf::net::receive_policy::up_to(2048u)); .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();
})
// When started, run our worker actor 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;
}
// Run until the user stops the server by pressing enter.
std::cout << "*** running on port " << port << '\n'
<< "*** press [enter] to quit" << std::endl;
getchar();
return EXIT_SUCCESS; return EXIT_SUCCESS;
} }
......
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