Commit 888058a9 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/wss'

parents 49851795 dcebc03e
...@@ -145,4 +145,5 @@ add_net_example(http time-server) ...@@ -145,4 +145,5 @@ add_net_example(http time-server)
add_net_example(web_socket echo) add_net_example(web_socket echo)
add_net_example(web_socket hello-client) add_net_example(web_socket hello-client)
add_net_example(web_socket quote-server) add_net_example(web_socket quote-server)
add_net_example(web_socket secure-echo)
add_net_example(web_socket stock-ticker) add_net_example(web_socket stock-ticker)
...@@ -15,6 +15,8 @@ ...@@ -15,6 +15,8 @@
#include "caf/scheduled_actor/flow.hpp" #include "caf/scheduled_actor/flow.hpp"
#include <iostream> #include <iostream>
#include <string>
#include <string_view>
#include <utility> #include <utility>
using namespace std::literals; using namespace std::literals;
...@@ -45,41 +47,18 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -45,41 +47,18 @@ 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.
auto port = caf::get_or(cfg, "port", default_port); auto port = caf::get_or(cfg, "port", default_port);
net::tcp_accept_socket fd; auto acc = net::ssl::acceptor::make_with_cert_file(port, cert_file, key_file);
if (auto maybe_fd = net::make_tcp_accept_socket({ipv4_address{}, port})) { if (!acc) {
std::cout << "*** started listening for incoming connections on port " std::cerr << "*** unable to initialize TLS: " << to_string(acc.error())
<< 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 the OpenSSL context and set key and certificate.
auto ctx = net::ssl::context::make_server(net::ssl::tls::any);
if (!ctx) {
std::cerr << "*** unable to create SSL context: " << to_string(ctx.error())
<< '\n'; << '\n';
return EXIT_FAILURE; return EXIT_FAILURE;
} }
if (!ctx->use_certificate_from_file(cert_file.c_str(), std::cout << "*** started listening for incoming connections on port " << port
net::ssl::format::pem)) { << '\n';
std::cerr << "*** unable to load certificate file: "
<< ctx->last_error_string() << '\n';
return EXIT_FAILURE;
}
if (!ctx->use_private_key_from_file(key_file.c_str(),
net::ssl::format::pem)) {
std::cerr << "*** unable to load private key file: "
<< ctx->last_error_string() << '\n';
return EXIT_FAILURE;
}
// Tie context and socket up into an acceptor for the http::serve API.
auto acc = net::ssl::acceptor{fd, std::move(*ctx)};
// Create buffers to signal events from the WebSocket server to the worker. // Create buffers to signal events from the WebSocket server to the worker.
auto [worker_pull, server_push] = net::http::make_request_resource(); auto [worker_pull, server_push] = net::http::make_request_resource();
// Spin up the HTTP server. // Spin up the HTTP server.
net::http::serve(sys, std::move(acc), std::move(server_push)); net::http::serve(sys, std::move(*acc), std::move(server_push));
// Spin up a worker to handle the HTTP requests. // Spin up a worker to handle the HTTP requests.
auto worker = sys.spawn([wres = worker_pull](event_based_actor* self) { auto worker = sys.spawn([wres = worker_pull](event_based_actor* self) {
// For each incoming request ... // For each incoming request ...
......
...@@ -13,6 +13,7 @@ ...@@ -13,6 +13,7 @@
#include "caf/scheduled_actor/flow.hpp" #include "caf/scheduled_actor/flow.hpp"
#include <iostream> #include <iostream>
#include <string>
#include <utility> #include <utility>
static constexpr uint16_t default_port = 8080; static constexpr uint16_t default_port = 8080;
......
// Simple WebSocket server with TLS that sends everything it receives back to
// the sender.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <string>
#include <string_view>
#include <utility>
using namespace std::literals;
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<std::string>("cert-file", "PEM server certificate file")
.add<std::string>("key-file", "PEM key file for the certificate");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Sanity checking.
auto cert_file = caf::get_or(cfg, "cert-file", ""sv);
auto key_file = caf::get_or(cfg, "key-file", ""sv);
if (cert_file.empty()) {
std::cerr << "*** mandatory parameter 'cert-file' missing or empty\n";
return EXIT_FAILURE;
}
if (key_file.empty()) {
std::cerr << "*** mandatory parameter 'key-file' missing or empty\n";
return EXIT_FAILURE;
}
// Create the OpenSSL context and set key and certificate.
auto port = caf::get_or(cfg, "port", default_port);
auto acc = cn::ssl::acceptor::make_with_cert_file(port, cert_file, key_file);
if (!acc) {
std::cerr << "*** unable to initialize TLS: " << to_string(acc.error())
<< '\n';
return EXIT_FAILURE;
}
std::cout << "*** started listening for incoming connections on port " << port
<< '\n';
// Convenience type aliases.
using event_t = ws::accept_event_t<>;
// Create buffers to signal events from the WebSocket server to the worker.
auto [wres, sres] = ws::make_accept_event_resources();
// Spin up a worker to handle the events.
auto worker = sys.spawn([worker_res = wres](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ...
self->make_observable()
.from_resource(worker_res)
.for_each([self](const event_t& event) {
// ... that simply pushes data back to the sender.
auto [pull, push] = event.data();
pull.observe_on(self)
.do_on_next([](const ws::frame& x) {
if (x.is_binary()) {
std::cout << "*** received a binary WebSocket frame of size "
<< x.size() << '\n';
} else {
std::cout << "*** received a text WebSocket frame of size "
<< x.size() << '\n';
}
})
.subscribe(push);
});
});
// Callback for incoming WebSocket requests.
auto on_request = [](const caf::settings& hdr, auto& req) {
// The hdr parameter is a dictionary with fields from the WebSocket
// handshake such as the path.
auto path = caf::get_or(hdr, "web-socket.path", "/");
std::cout << "*** new client request for path " << path << '\n';
// Accept the WebSocket connection only if the path is "/".
if (path == "/") {
// Calling `accept` causes the server to acknowledge the client and
// creates input and output buffers that go to the worker actor.
req.accept();
} else {
// Calling `reject` aborts the connection with HTTP status code 400 (Bad
// Request). The error gets converted to a string and send to the client
// to give some indication to the client why the request was rejected.
auto err = caf::make_error(caf::sec::invalid_argument,
"unrecognized path, try '/'");
req.reject(std::move(err));
}
// Note: calling neither accept nor reject also rejects the connection.
};
// Set everything in motion.
ws::accept(sys, std::move(*acc), std::move(sres), on_request);
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
...@@ -22,7 +22,7 @@ inline auto make_request_resource() { ...@@ -22,7 +22,7 @@ inline auto make_request_resource() {
return async::make_spsc_buffer_resource<request>(); return async::make_spsc_buffer_resource<request>();
} }
/// Listens for incoming HTTP requests on @p fd. /// Listens for incoming HTTP requests.
/// @param sys The host system. /// @param sys The host system.
/// @param fd An accept socket in listening mode, already bound to a port. /// @param fd An accept socket in listening mode, already bound to a port.
/// @param out A buffer resource that connects the server to a listener that /// @param out A buffer resource that connects the server to a listener that
...@@ -32,7 +32,7 @@ disposable CAF_NET_EXPORT serve(actor_system& sys, tcp_accept_socket fd, ...@@ -32,7 +32,7 @@ disposable CAF_NET_EXPORT serve(actor_system& sys, tcp_accept_socket fd,
async::producer_resource<request> out, async::producer_resource<request> out,
const settings& cfg = {}); const settings& cfg = {});
/// Listens for incoming HTTP requests on @p fd. /// Listens for incoming HTTPS requests.
/// @param sys The host system. /// @param sys The host system.
/// @param acc An SSL connection acceptor with a socket that in listening mode. /// @param acc An SSL connection acceptor with a socket that in listening mode.
/// @param out A buffer resource that connects the server to a listener that /// @param out A buffer resource that connects the server to a listener that
......
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#pragma once #pragma once
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/expected.hpp"
#include "caf/net/ssl/context.hpp" #include "caf/net/ssl/context.hpp"
#include "caf/net/ssl/fwd.hpp" #include "caf/net/ssl/fwd.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
...@@ -30,6 +31,34 @@ public: ...@@ -30,6 +31,34 @@ public:
// nop // nop
} }
// -- factories --------------------------------------------------------------
static expected<acceptor>
make_with_cert_file(tcp_accept_socket fd, const char* cert_file_path,
const char* key_file_path,
format file_format = format::pem);
static expected<acceptor>
make_with_cert_file(uint16_t port, const char* cert_file_path,
const char* key_file_path,
format file_format = format::pem);
static expected<acceptor>
make_with_cert_file(tcp_accept_socket fd, const std::string& cert_file_path,
const std::string& key_file_path,
format file_format = format::pem) {
return make_with_cert_file(fd, cert_file_path.c_str(),
key_file_path.c_str(), file_format);
}
static expected<acceptor>
make_with_cert_file(uint16_t port, const std::string& cert_file_path,
const std::string& key_file_path,
format file_format = format::pem) {
return make_with_cert_file(port, cert_file_path.c_str(),
key_file_path.c_str(), file_format);
}
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
tcp_accept_socket fd() const noexcept { tcp_accept_socket fd() const noexcept {
...@@ -49,6 +78,8 @@ private: ...@@ -49,6 +78,8 @@ private:
context ctx_; context ctx_;
}; };
// -- free functions -----------------------------------------------------------
/// Checks whether `acc` has a valid socket descriptor. /// Checks whether `acc` has a valid socket descriptor.
bool CAF_NET_EXPORT valid(const acceptor& acc); bool CAF_NET_EXPORT valid(const acceptor& acc);
......
...@@ -9,6 +9,8 @@ ...@@ -9,6 +9,8 @@
#include "caf/detail/accept_handler.hpp" #include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp" #include "caf/detail/connection_factory.hpp"
#include "caf/net/flow_connector.hpp" #include "caf/net/flow_connector.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/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp" #include "caf/net/web_socket/flow_bridge.hpp"
#include "caf/net/web_socket/flow_connector_request_impl.hpp" #include "caf/net/web_socket/flow_connector_request_impl.hpp"
...@@ -18,13 +20,38 @@ ...@@ -18,13 +20,38 @@
#include <type_traits> #include <type_traits>
namespace caf::net::web_socket {
/// Describes the per-connection event.
template <class... Ts>
using accept_event_t
= cow_tuple<async::consumer_resource<frame>, // Socket to App.
async::producer_resource<frame>, // App to Socket.
Ts...>; // Handshake data.
/// A producer resource for the acceptor. Any accepted WebSocket connection is
/// represented by two buffers. The user-defined types `Ts...` allow the
/// @ref request to transfer additional context for the connection to the
/// listener (usually extracted from WebSocket handshake fields).
template <class... Ts>
using acceptor_resource_t = async::producer_resource<accept_event_t<Ts...>>;
/// Convenience function for creating an event listener resource and an
/// @ref acceptor_resource_t via @ref async::make_spsc_buffer_resource.
template <class... Ts>
auto make_accept_event_resources() {
return async::make_spsc_buffer_resource<accept_event_t<Ts...>>();
}
} // namespace caf::net::web_socket
namespace caf::detail { namespace caf::detail {
template <class Transport, class Trait> template <class Transport, class Trait>
class ws_conn_factory class ws_conn_factory
: public connection_factory<typename Transport::connection_handle> { : public connection_factory<typename Transport::connection_handle> {
public: public:
using socket_type = typename Transport::socket_type; using connection_handle = typename Transport::connection_handle;
using connector_pointer = net::flow_connector_ptr<Trait>; using connector_pointer = net::flow_connector_ptr<Trait>;
...@@ -33,11 +60,14 @@ public: ...@@ -33,11 +60,14 @@ public:
// nop // nop
} }
net::socket_manager_ptr make(net::multiplexer* mpx, socket_type fd) override { 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 = net::web_socket::flow_bridge<Trait>::make(mpx, connector_);
auto app_ptr = app.get(); auto app_ptr = app.get();
auto ws = net::web_socket::server::make(std::move(app)); auto ws = net::web_socket::server::make(std::move(app));
auto transport = Transport::make(fd, std::move(ws)); auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(ws));
transport->active_policy().accept(fd);
auto mgr = net::socket_manager::make(mpx, std::move(transport)); auto mgr = net::socket_manager::make(mpx, std::move(transport));
app_ptr->self_ref(mgr->as_disposable()); app_ptr->self_ref(mgr->as_disposable());
return mgr; return mgr;
...@@ -47,60 +77,26 @@ private: ...@@ -47,60 +77,26 @@ private:
connector_pointer connector_; connector_pointer connector_;
}; };
} // namespace caf::detail template <class Transport, class Acceptor, class... Ts, class OnRequest>
disposable ws_accept_impl(actor_system& sys, Acceptor acc,
namespace caf::net::web_socket { net::web_socket::acceptor_resource_t<Ts...> out,
OnRequest on_request, const settings& cfg) {
/// Describes the per-connection event. using trait_t = net::web_socket::default_trait;
template <class... Ts>
using accept_event_t
= cow_tuple<async::consumer_resource<frame>, // Socket to App.
async::producer_resource<frame>, // App to Socket.
Ts...>; // Handshake data.
/// A producer resource for the acceptor. Any accepted WebSocket connection is
/// represented by two buffers. The user-defined types `Ts...` allow the
/// @ref request to transfer additional context for the connection to the
/// listener (usually extracted from WebSocket handshake fields).
template <class... Ts>
using acceptor_resource_t = async::producer_resource<accept_event_t<Ts...>>;
/// Convenience function for creating an event listener resource and an
/// @ref acceptor_resource_t via @ref async::make_spsc_buffer_resource.
template <class... Ts>
auto make_accept_event_resources() {
return async::make_spsc_buffer_resource<accept_event_t<Ts...>>();
}
/// Listens for incoming WebSocket connection on @p fd.
/// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this socket
/// must already listen to a port.
/// @param out A buffer resource that connects the server to a listener that
/// processes the buffer pairs for each incoming connection.
/// @param on_request Function object for accepting incoming requests.
/// @param cfg Configuration parameters for the acceptor.
template <class Transport = stream_transport, class Socket, class... Ts,
class OnRequest>
disposable accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
OnRequest on_request, const settings& cfg = {}) {
using trait_t = default_trait;
using request_t = request<default_trait, Ts...>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request");
using factory_t = detail::ws_conn_factory<Transport, trait_t>; using factory_t = detail::ws_conn_factory<Transport, trait_t>;
using conn_t = typename Transport::connection_handle; using conn_t = typename Transport::connection_handle;
using impl_t = detail::accept_handler<Socket, conn_t>; using impl_t = detail::accept_handler<Acceptor, conn_t>;
using connector_t = flow_connector_request_impl<OnRequest, trait_t, Ts...>; using connector_t
= net::web_socket::flow_connector_request_impl<OnRequest, trait_t, Ts...>;
auto max_connections = get_or(cfg, defaults::net::max_connections); auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) { if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx(); auto& mpx = sys.network_manager().mpx();
auto conn = std::make_shared<connector_t>(std::move(on_request), auto conn = std::make_shared<connector_t>(std::move(on_request),
std::move(buf)); std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(conn)); auto factory = std::make_unique<factory_t>(std::move(conn));
auto impl = impl_t::make(fd, std::move(factory), max_connections); auto impl = impl_t::make(std::move(acc), std::move(factory),
max_connections);
auto impl_ptr = impl.get(); auto impl_ptr = impl.get();
auto ptr = socket_manager::make(&mpx, std::move(impl)); auto ptr = net::socket_manager::make(&mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable()); impl_ptr->self_ref(ptr->as_disposable());
mpx.start(ptr); mpx.start(ptr);
return disposable{std::move(ptr)}; return disposable{std::move(ptr)};
...@@ -109,4 +105,44 @@ disposable accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out, ...@@ -109,4 +105,44 @@ disposable accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
} }
} }
} // namespace caf::detail
namespace caf::net::web_socket {
/// Listens for incoming WebSocket connections.
/// @param sys The host system.
/// @param fd An accept socket in listening mode, already bound to a port.
/// @param out A buffer resource that connects the server to a listener that
/// processes the buffer pairs for each incoming connection.
/// @param on_request Function object for accepting incoming requests.
/// @param cfg Configuration parameters for the acceptor.
template <class... Ts, class OnRequest>
disposable accept(actor_system& sys, tcp_accept_socket fd,
acceptor_resource_t<Ts...> out, OnRequest on_request,
const settings& cfg = {}) {
using request_t = request<default_trait, Ts...>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request");
return detail::ws_accept_impl<stream_transport>(sys, fd, out,
std::move(on_request), cfg);
}
/// Listens for incoming WebSocket connections over TLS.
/// @param sys The host system.
/// @param acc An SSL connection acceptor with a socket that in listening mode.
/// @param out A buffer resource that connects the server to a listener that
/// processes the buffer pairs for each incoming connection.
/// @param on_request Function object for accepting incoming requests.
/// @param cfg Configuration parameters for the acceptor.
template <class... Ts, class OnRequest>
disposable accept(actor_system& sys, ssl::acceptor acc,
acceptor_resource_t<Ts...> out, OnRequest on_request,
const settings& cfg = {}) {
using request_t = request<default_trait, Ts...>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request");
return detail::ws_accept_impl<ssl::transport>(sys, std::move(acc), out,
std::move(on_request), cfg);
}
} // namespace caf::net::web_socket } // namespace caf::net::web_socket
...@@ -24,6 +24,39 @@ acceptor& acceptor::operator=(acceptor&& other) { ...@@ -24,6 +24,39 @@ acceptor& acceptor::operator=(acceptor&& other) {
return *this; return *this;
} }
// -- factories ----------------------------------------------------------------
expected<acceptor>
acceptor::make_with_cert_file(tcp_accept_socket fd, const char* cert_file_path,
const char* key_file_path, format file_format) {
auto ctx = context::make_server(tls::any);
if (!ctx) {
return {make_error(sec::runtime_error, "unable to create SSL context")};
}
if (!ctx->use_certificate_from_file(cert_file_path, file_format)) {
return {make_error(sec::runtime_error, "unable to load certificate file",
ctx->last_error_string())};
}
if (!ctx->use_private_key_from_file(key_file_path, file_format)) {
return {make_error(sec::runtime_error, "unable to load private key file",
ctx->last_error_string())};
}
return {acceptor{fd, std::move(*ctx)}};
}
expected<acceptor>
acceptor::make_with_cert_file(uint16_t port, const char* cert_file_path,
const char* key_file_path, format file_format) {
if (auto fd = make_tcp_accept_socket(port)) {
return acceptor::make_with_cert_file(*fd, cert_file_path, key_file_path,
file_format);
} else {
return {make_error(sec::cannot_open_port)};
}
}
// -- free functions -----------------------------------------------------------
bool valid(const acceptor& acc) { bool valid(const acceptor& acc) {
return valid(acc.fd()); return valid(acc.fd());
} }
......
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