Commit 3ae1ac34 authored by Dominik Charousset's avatar Dominik Charousset

Implement the new DSL for WebSocket servers

parent 69587e6e
......@@ -116,7 +116,6 @@ add_net_example(length_prefix_framing chat-server)
add_net_example(web_socket echo)
add_net_example(web_socket hello-client)
add_net_example(web_socket quote-server)
add_net_example(web_socket secure-echo)
add_net_example(web_socket stock-ticker)
if(TARGET CAF::net AND CAF_ENABLE_QT6_EXAMPLES)
......
......@@ -117,7 +117,6 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
return EXIT_FAILURE;
}
// Open up a TCP port for incoming connections and start the server.
auto had_error = false;
auto server
= caf::net::lp::with(sys)
// Optionally enable TLS.
......@@ -133,6 +132,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
.start([&sys](trait::acceptor_resource accept_events) {
sys.spawn(worker_impl, std::move(accept_events));
});
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
......
......@@ -8,84 +8,115 @@
#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/net/web_socket/with.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <utility>
static constexpr uint16_t default_port = 8080;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 7788;
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<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");
}
};
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Open up a TCP port for incoming connections.
namespace ssl = caf::net::ssl;
namespace ws = caf::net::web_socket;
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port})) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
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;
}
// 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 "
// Open up a TCP port for incoming connections and start the server.
using trait = ws::default_trait;
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)
// Accept only requests for path "/".
.on_request([](const caf::settings& hdr, ws::acceptor<>& acc) {
// 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.
acc.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 '/'");
acc.reject(std::move(err));
}
// Note: calling nothing on `acc` also rejects the connection.
})
// When started, run our worker actor to handle incoming connections.
.start([&sys](trait::acceptor_resource<> events) {
using event_t = trait::accept_event<>;
sys.spawn([events](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ...
self->make_observable().from_resource(events).for_each(
[self](const event_t& ev) {
// ... that simply pushes data back to the sender.
auto [pull, push] = ev.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 "
} 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, fd, std::move(sres), on_request);
}
})
.subscribe(push);
});
});
});
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
......
......@@ -8,21 +8,24 @@
#include "caf/cow_tuple.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.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/net/ssl/context.hpp"
#include "caf/net/web_socket/with.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
#include <cassert>
#include <iostream>
#include <random>
#include <utility>
using namespace std::literals;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 8080;
static constexpr size_t default_max_connections = 128;
static constexpr std::string_view epictetus[] = {
"Wealth consists not in having great possessions, but in having few wants.",
"Don't explain your philosophy. Embody it.",
......@@ -62,6 +65,22 @@ static constexpr std::string_view plato[] = {
"Beauty lies in the eyes of the beholder.",
};
// -- 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");
}
};
// -- helper functions ---------------------------------------------------------
// Returns a list of philosopher quotes by path.
caf::span<const std::string_view> quotes_by_path(std::string_view path) {
if (path == "/epictetus")
return caf::make_span(epictetus);
......@@ -73,13 +92,7 @@ caf::span<const std::string_view> quotes_by_path(std::string_view path) {
return {};
}
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
}
};
// Chooses a random quote from a list of quotes.
struct pick_random {
public:
pick_random() : engine_(std::random_device{}()) {
......@@ -96,51 +109,73 @@ private:
std::minstd_rand engine_;
};
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Open up a TCP port for incoming connections.
namespace ssl = caf::net::ssl;
namespace ws = caf::net::web_socket;
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
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.
using trait = ws::default_trait;
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)
// Forward the path from the WebSocket request to the worker.
.on_request(
[](const caf::settings& hdr, ws::acceptor<caf::cow_string>& acc) {
// The hdr parameter is a dictionary with fields from the WebSocket
// handshake such as the path. This is only field we care about
// here. By passing the (copy-on-write) string to accept() here, we
// make it available to the worker through the acceptor_resource.
auto path = caf::get_or(hdr, "web-socket.path", "/");
acc.accept(caf::cow_string{std::move(path)});
})
// When started, run our worker actor to handle incoming connections.
.start([&sys](trait::acceptor_resource<caf::cow_string> events) {
using event_t = trait::accept_event<caf::cow_string>;
sys.spawn([events](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ...
self->make_observable()
.from_resource(events) //
.for_each([self, f = pick_random{}](const event_t& ev) mutable {
// ... that pushes one random quote to the client.
auto [pull, push, path] = ev.data();
auto quotes = quotes_by_path(path);
auto quote = quotes.empty()
? "Try /epictetus, /seneca or /plato."
: f(quotes);
self->make_observable().just(ws::frame{quote}).subscribe(push);
// We ignore whatever the client may send to us.
pull.observe_on(self).subscribe(std::ignore);
});
});
});
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Convenience type aliases.
using event_t = ws::accept_event_t<caf::cow_string>;
// Create buffers to signal events from the WebSocket server to the worker.
auto [wres, sres] = ws::make_accept_event_resources<caf::cow_string>();
// 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, f = pick_random{}](const event_t& event) mutable {
// ... that pushes one random quote to the client.
auto [pull, push, path] = event.data();
auto quotes = quotes_by_path(path);
auto quote = quotes.empty() ? "Try /epictetus, /seneca or /plato."
: f(quotes);
self->make_observable().just(ws::frame{quote}).subscribe(push);
// We ignore whatever the client may send to us.
pull.observe_on(self).subscribe(std::ignore);
});
});
// 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. This is only field we care about here.
auto path = caf::get_or(hdr, "web-socket.path", "/");
req.accept(caf::cow_string{std::move(path)});
};
// Set everything in motion.
ws::accept(sys, fd, std::move(sres), on_request);
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
......
// 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)
......@@ -10,9 +10,7 @@
#include "caf/json_writer.hpp"
#include "caf/net/middleman.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/net/web_socket/with.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
......@@ -24,8 +22,14 @@
using namespace std::literals;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 8080;
static constexpr size_t default_max_connections = 128;
// -- custom types -------------------------------------------------------------
namespace stock {
struct info {
......@@ -47,12 +51,15 @@ bool inspect(Inspector& f, info& x) {
} // namespace stock
// -- actor for generating a random feed ---------------------------------------
struct random_feed_state {
using trait = caf::net::web_socket::default_trait;
using frame = caf::net::web_socket::frame;
using accept_event = caf::net::web_socket::accept_event_t<>;
using accept_event = trait::accept_event<>;
random_feed_state(caf::event_based_actor* selfptr,
caf::async::consumer_resource<accept_event> events,
trait::acceptor_resource<> events,
caf::timespan update_interval)
: self(selfptr), val_dist(0, 100000), index_dist(0, 19) {
// Init random number generator.
......@@ -132,39 +139,68 @@ struct random_feed_state {
using random_feed_impl = caf::stateful_actor<random_feed_state>;
// -- 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")
.add<caf::timespan>("interval,i", "update interval");
;
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");
}
};
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Open up a TCP port for incoming connections.
namespace ssl = caf::net::ssl;
namespace ws = caf::net::web_socket;
// Read the configuration.
auto interval = caf::get_or(cfg, "interval", caf::timespan{1s});
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
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;
}
// 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 interval = caf::get_or(cfg, "interval", caf::timespan{1s});
auto worker = sys.spawn<random_feed_impl>(std::move(wres), interval);
// Callback for incoming WebSocket requests.
auto on_request = [](const caf::settings&, auto& req) { req.accept(); };
// Set everything in motion.
ws::accept(sys, fd, std::move(sres), on_request);
// Open up a TCP port for incoming connections and start the server.
using trait = ws::default_trait;
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)
// Accept only requests for path "/".
.on_request([](const caf::settings&, ws::acceptor<>& acc) {
// Ignore all header fields and accept the connection.
acc.accept();
})
// When started, run our worker actor to handle incoming connections.
.start([&sys, interval](trait::acceptor_resource<> events) {
sys.spawn<random_feed_impl>(std::move(events), interval);
});
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
......
......@@ -7,7 +7,6 @@
#include "caf/async/blocking_producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/error.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/sec.hpp"
#include <tuple>
......
......@@ -11,7 +11,6 @@
#include "caf/detail/flow_connector.hpp"
#include "caf/fwd.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/net/web_socket/upper_layer.hpp"
#include "caf/sec.hpp"
......
......@@ -5,40 +5,45 @@
#pragma once
#include "caf/detail/flow_connector.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/net/web_socket/acceptor.hpp"
namespace caf::detail {
/// Calls an `OnRequest` handler with a @ref request object and passes the
/// generated buffers to the @ref flow_bridge.
template <class OnRequest, class Trait, class... Ts>
template <class Trait, class... Ts>
class ws_flow_connector_request_impl : public flow_connector<Trait> {
public:
using super = flow_connector<Trait>;
using result_type = typename super::result_type;
using request_type = net::web_socket::request<Trait, Ts...>;
using acceptor_base_type = net::web_socket::acceptor<Ts...>;
using app_res_type = typename request_type::app_res_type;
using acceptor_type = net::web_socket::acceptor_impl<Trait, Ts...>;
using app_res_type = typename acceptor_type::app_res_type;
using producer_type = async::blocking_producer<app_res_type>;
using on_request_cb_type
= shared_callback_ptr<void(const settings&, acceptor_base_type&)>;
template <class T>
ws_flow_connector_request_impl(OnRequest&& on_request, T&& out)
ws_flow_connector_request_impl(on_request_cb_type on_request, T&& out)
: on_request_(std::move(on_request)), out_(std::forward<T>(out)) {
// nop
}
result_type on_request(const settings& cfg) override {
request_type req;
on_request_(cfg, req);
if (req.accepted()) {
out_.push(req.app_resources_);
auto& [pull, push] = req.ws_resources_;
acceptor_type acc;
(*on_request_)(cfg, acc);
if (acc.accepted()) {
out_.push(acc.app_resources);
auto& [pull, push] = acc.ws_resources;
return {error{}, std::move(pull), std::move(push)};
} else if (auto& err = req.reject_reason()) {
return {err, {}, {}};
} else if (auto&& err = std::move(acc).reject_reason()) {
return {std::move(err), {}, {}};
} else {
auto def_err = make_error(sec::runtime_error,
"WebSocket request rejected without reason");
......@@ -47,7 +52,7 @@ public:
}
private:
OnRequest on_request_;
on_request_cb_type on_request_;
producer_type out_;
};
......
......@@ -56,7 +56,7 @@ public:
return *cfg_;
}
private:
protected:
Derived& dref() {
return static_cast<Derived&>(*this);
}
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_system.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/detail/flow_connector.hpp"
#include "caf/detail/ws_flow_bridge.hpp"
#include "caf/detail/ws_flow_connector_request_impl.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/frame.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/net/web_socket/server.hpp"
#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...>>;
/// A consumer resource for processing accepted connections.
template <class... Ts>
using listener_resource_t = async::consumer_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 {
template <class Transport, class Trait>
class ws_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using connection_handle = typename Transport::connection_handle;
using connector_pointer = flow_connector_ptr<Trait>;
explicit ws_conn_factory(connector_pointer connector)
: connector_(std::move(connector)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto app = detail::ws_flow_bridge<Trait>::make(mpx, connector_);
auto app_ptr = app.get();
auto ws = net::web_socket::server::make(std::move(app));
auto fd = conn.fd();
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));
app_ptr->self_ref(mgr->as_disposable());
return mgr;
}
private:
connector_pointer connector_;
};
} // namespace caf::detail
namespace caf::net::web_socket {
/// Listens for incoming WebSocket connections.
/// @param sys The host system.
/// @param acc A connection acceptor such as @ref tcp_accept_socket or
/// @ref ssl::acceptor.
/// @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 Acceptor, class... Ts, class OnRequest>
disposable accept(actor_system& sys, Acceptor acc,
acceptor_resource_t<Ts...> out, OnRequest on_request,
const settings& cfg = {}) {
using transport_t = typename Acceptor::transport_type;
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 trait_t = net::web_socket::default_trait;
using factory_t = detail::ws_conn_factory<transport_t, trait_t>;
using conn_t = typename transport_t::connection_handle;
using impl_t = detail::accept_handler<Acceptor, conn_t>;
using connector_t
= detail::ws_flow_connector_request_impl<OnRequest, trait_t, Ts...>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto conn = std::make_shared<connector_t>(std::move(on_request),
std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(conn));
auto impl = impl_t::make(std::move(acc), std::move(factory),
max_connections);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(&mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
mpx.start(ptr);
return disposable{std::move(ptr)};
} else {
return {};
}
}
} // namespace caf::net::web_socket
......@@ -5,6 +5,7 @@
#pragma once
#include "caf/async/spsc_buffer.hpp"
#include "caf/callback.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/error.hpp"
#include "caf/net/fwd.hpp"
......@@ -14,33 +15,20 @@
namespace caf::net::web_socket {
/// Represents a WebSocket connection request.
/// @tparam Trait Converts between native and wire format.
template <class Trait, class... Ts>
class request {
/// Accepts or rejects incoming connection requests.
/// @tparam Ts Denotes the types for the worker handshake.
template <class... Ts>
class acceptor {
public:
template <class, class, class...>
friend class detail::ws_flow_connector_request_impl;
template <class Trait>
using impl_type = acceptor_impl<Trait, Ts...>;
using input_type = typename Trait::input_type;
template <class Trait>
using server_factory_type = server_factory<Trait, Ts...>;
using output_type = typename Trait::output_type;
using app_res_type = cow_tuple<async::consumer_resource<input_type>,
async::producer_resource<output_type>, Ts...>;
virtual ~acceptor() = default;
using ws_res_type = async::resource_pair<output_type, input_type>;
void accept(Ts... worker_args) {
if (accepted())
return;
auto [app_pull, ws_push] = async::make_spsc_buffer_resource<input_type>();
auto [ws_pull, app_push] = async::make_spsc_buffer_resource<output_type>();
ws_resources_ = ws_res_type{ws_pull, ws_push};
app_resources_ = make_cow_tuple(app_pull, app_push,
std::move(worker_args)...);
accepted_ = true;
}
virtual void accept(Ts... xs) = 0;
void reject(error reason) {
reject_reason_ = reason;
......@@ -52,15 +40,59 @@ public:
return accepted_;
}
const error& reject_reason() const noexcept {
error&& reject_reason() && noexcept {
return std::move(reject_reason_);
}
const error& reject_reason() const& noexcept {
return reject_reason_;
}
private:
protected:
bool accepted_ = false;
ws_res_type ws_resources_;
app_res_type app_resources_;
error reject_reason_;
};
template <class Trait, class... Ts>
class acceptor_impl : public acceptor<Ts...> {
public:
using super = acceptor<Ts...>;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using app_res_type = cow_tuple<async::consumer_resource<input_type>,
async::producer_resource<output_type>, Ts...>;
using ws_res_type = async::resource_pair<output_type, input_type>;
void accept(Ts... xs) override {
if (super::accepted_)
return;
auto [app_pull, ws_push] = async::make_spsc_buffer_resource<input_type>();
auto [ws_pull, app_push] = async::make_spsc_buffer_resource<output_type>();
ws_resources = ws_res_type{ws_pull, ws_push};
app_resources = make_cow_tuple(app_pull, app_push, std::move(xs)...);
super::accepted_ = true;
}
ws_res_type ws_resources;
app_res_type app_resources;
};
/// Type trait that determines if a type is an `acceptor`.
template <class T>
struct is_acceptor : std::false_type {};
/// Specialization of `is_acceptor` for `acceptor` types.
template <class... Ts>
struct is_acceptor<acceptor<Ts...>> : std::true_type {};
/// A `constexpr bool` variable that evaluates to `true` if the given type is an
/// `acceptor`, `false` otherwise.
template <class T>
inline constexpr bool is_acceptor_v = is_acceptor<T>::value;
} // namespace caf::net::web_socket
......@@ -4,9 +4,11 @@
#pragma once
#include "caf/async/spsc_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/web_socket/fwd.hpp"
#include "caf/net/web_socket/frame.hpp"
#include <string_view>
#include <vector>
......@@ -23,6 +25,20 @@ public:
/// layer to the WebSocket.
using output_type = frame;
/// A resource for consuming input_type elements.
using input_resource = async::consumer_resource<input_type>;
/// A resource for producing output_type elements.
using output_resource = async::producer_resource<output_type>;
/// An accept event from the server to transmit read and write handles.
template <class... Ts>
using accept_event = cow_tuple<input_resource, output_resource, Ts...>;
/// A resource for consuming accept events.
template <class... Ts>
using acceptor_resource = async::consumer_resource<accept_event<Ts...>>;
bool converts_to_binary(const output_type& x);
bool convert(const output_type& x, byte_buffer& bytes);
......
......@@ -6,15 +6,21 @@
namespace caf::detail {
template <class, class, class...>
template <class, class...>
class ws_flow_connector_request_impl;
} // namespace caf::detail
namespace caf::net::web_socket {
template <class... Ts>
class acceptor;
template <class Trait, class... Ts>
class acceptor_impl;
template <class Trait, class... Ts>
class request;
class server_factory;
class frame;
class framing;
......
// 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/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/net/dsl/server_factory_base.hpp"
#include "caf/net/web_socket/acceptor.hpp"
#include "caf/net/web_socket/server_factory.hpp"
#include <type_traits>
namespace caf::net::web_socket {
/// DSL entry point for creating a server.
template <class Trait>
class has_on_request
: public dsl::server_factory_base<Trait, has_on_request<Trait>> {
public:
using super = dsl::server_factory_base<Trait, has_on_request<Trait>>;
using super::super;
/// Adds the handler for accepting or rejecting incoming requests.
template <class OnRequest>
auto on_request(OnRequest on_request) {
// Type checking.
using fn_trait = detail::get_callable_trait_t<OnRequest>;
static_assert(fn_trait::num_args == 2,
"on_request must take exactly two arguments");
using arg_types = typename fn_trait::arg_types;
using arg1_t = detail::tl_at_t<arg_types, 0>;
using arg2_t = detail::tl_at_t<arg_types, 1>;
static_assert(std::is_same_v<arg1_t, const settings&>,
"on_request must take 'const settings&' as first argument");
using acceptor_t = std::decay_t<arg2_t>;
static_assert(is_acceptor_v<acceptor_t>,
"on_request must take an acceptor as second argument");
static_assert(std::is_same_v<arg2_t, acceptor_t&>,
"on_request must take the acceptor as mutable reference");
// Wrap the callback and return the factory object.
using factory_t = typename acceptor_t::template server_factory_type<Trait>;
auto callback = make_shared_type_erased_callback(std::move(on_request));
return factory_t{super::cfg_, std::move(callback)};
}
};
} // namespace caf::net::web_socket
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/defaults.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/binary_flow_bridge.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/detail/shared_ssl_acceptor.hpp"
#include "caf/detail/ws_flow_bridge.hpp"
#include "caf/detail/ws_flow_connector_request_impl.hpp"
#include "caf/fwd.hpp"
#include "caf/net/dsl/server_factory_base.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket/server.hpp"
#include <cstdint>
#include <functional>
#include <variant>
namespace caf::detail {
/// Specializes @ref connection_factory for the WebSocket protocol.
template <class Trait, class Transport>
class ws_connection_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using connection_handle = typename Transport::connection_handle;
using connector_pointer = flow_connector_ptr<Trait>;
explicit ws_connection_factory(connector_pointer connector)
: connector_(std::move(connector)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto app = ws_flow_bridge<Trait>::make(mpx, connector_);
auto app_ptr = app.get();
auto ws = net::web_socket::server::make(std::move(app));
auto fd = conn.fd();
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));
app_ptr->self_ref(mgr->as_disposable());
return mgr;
}
private:
connector_pointer connector_;
};
} // namespace caf::detail
namespace caf::net::web_socket {
/// Factory type for the `with(...).accept(...).start(...)` DSL.
template <class Trait, class... Ts>
class server_factory
: public dsl::server_factory_base<Trait, server_factory<Trait>> {
public:
using super = dsl::server_factory_base<Trait, server_factory<Trait>>;
using on_request_cb_type
= shared_callback_ptr<void(const settings&, acceptor<Ts...>&)>;
server_factory(dsl::server_config_ptr<Trait> cfg,
on_request_cb_type on_request)
: super(std::move(cfg)), on_request_(std::move(on_request)) {
// 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;
/// The output type of the application, i.e., what flows from the application
/// layer to the WebSocket.
using output_type = typename Trait::output_type;
/// A resource for consuming input_type elements.
using input_resource = async::consumer_resource<input_type>;
/// A resource for producing output_type elements.
using output_resource = async::producer_resource<output_type>;
/// An accept event from the server to transmit read and write handles.
using accept_event = cow_tuple<input_resource, output_resource, Ts...>;
/// A resource for consuming accept events.
using acceptor_resource = async::consumer_resource<accept_event>;
/// Starts a server that accepts incoming connections with the WebSocket
/// protocol.
template <class OnStart>
start_res_t 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());
}
private:
template <class Factory, class AcceptHandler, class Acceptor, class OnStart>
start_res_t do_start_impl(dsl::server_config<Trait>& cfg, Acceptor acc,
OnStart& on_start) {
using connector_t = detail::ws_flow_connector_request_impl<Trait, Ts...>;
auto [pull, push] = async::make_spsc_buffer_resource<accept_event>();
auto serv = std::make_shared<connector_t>(on_request_, push.try_open());
auto factory = std::make_unique<Factory>(std::move(serv));
auto impl = AcceptHandler::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(dsl::server_config<Trait>& cfg, tcp_accept_socket fd,
OnStart& on_start) {
if (!cfg.ctx) {
using factory_t = detail::ws_connection_factory<Trait, stream_transport>;
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::ws_connection_factory<Trait, ssl::transport>;
using acc_t = detail::shared_ssl_acceptor;
using impl_t = detail::accept_handler<acc_t, ssl::connection>;
return do_start_impl<factory_t, impl_t>(cfg, acc_t{fd, cfg.ctx}, on_start);
}
template <class OnStart>
start_res_t
do_start(typename dsl::server_config<Trait>::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);
}
template <class OnStart>
start_res_t
do_start(typename dsl::server_config<Trait>::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);
}
template <class OnStart>
start_res_t do_start(dsl::fail_server_config<Trait>& cfg, OnStart&) {
cfg.call_on_error(cfg.err);
return start_res_t{std::move(cfg.err)};
}
on_request_cb_type on_request_;
};
} // namespace caf::net::web_socket
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/dsl/has_accept.hpp"
#include "caf/net/dsl/has_connect.hpp"
#include "caf/net/dsl/has_context.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/has_on_request.hpp"
#include <cstdint>
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 //
with<dsl::has_accept, dsl::has_context> {
public:
template <class... Ts>
explicit with_t(multiplexer* mpx, Ts&&... xs)
: mpx_(mpx), trait_(std::forward<Ts>(xs)...), ctx_(error{}) {
// nop
}
with_t(const with_t&) noexcept = default;
with_t& operator=(const with_t&) noexcept = default;
multiplexer* mpx() const noexcept override {
return mpx_;
}
const Trait& trait() const noexcept override {
return trait_;
}
/// @private
has_on_request<Trait> lift(dsl::server_config_ptr<Trait> cfg) {
return has_on_request<Trait>{std::move(cfg)};
}
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_;
};
template <class Trait = default_trait>
with_t<Trait> with(actor_system& sys) {
return with_t<Trait>{multiplexer::from(sys)};
}
template <class Trait = default_trait>
with_t<Trait> with(multiplexer* mpx) {
return with_t<Trait>{mpx};
}
} // namespace caf::net::web_socket
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