Commit ec01dc81 authored by Dominik Charousset's avatar Dominik Charousset

Add new WebSocket example with path handling

parent 57c4685f
...@@ -139,4 +139,5 @@ else() ...@@ -139,4 +139,5 @@ else()
endfunction() endfunction()
endif() endif()
add_net_example(web_socket quote-server)
add_net_example(web_socket echo) add_net_example(web_socket echo)
...@@ -23,6 +23,7 @@ struct config : caf::actor_system_config { ...@@ -23,6 +23,7 @@ struct config : caf::actor_system_config {
int caf_main(caf::actor_system& sys, const config& cfg) { int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net; namespace cn = caf::net;
namespace ws = cn::web_socket;
// 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);
cn::tcp_accept_socket fd; cn::tcp_accept_socket fd;
...@@ -36,22 +37,20 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -36,22 +37,20 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
return EXIT_FAILURE; return EXIT_FAILURE;
} }
// Convenience type aliases. // Convenience type aliases.
using frame = cn::web_socket::frame; using event_t = ws::accept_event_t<>;
using frame_res_pair = caf::async::resource_pair<frame>; // Create buffers to signal events from the WebSocket server to the worker.
// Create a buffer of frame buffer pairs. Each buffer pair represents the auto [wres, sres] = ws::make_accept_event_resources();
// input and output channel for a single WebSocket connection. The outer // Spin up a worker to handle the events.
// buffer transfers buffers pairs from the WebSocket server to our worker.
auto [wres, sres] = caf::async::make_spsc_buffer_resource<frame_res_pair>();
auto worker = sys.spawn([worker_res = wres](caf::event_based_actor* self) { auto worker = sys.spawn([worker_res = wres](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ... // For each buffer pair, we create a new flow ...
self->make_observable() self->make_observable()
.from_resource(worker_res) .from_resource(worker_res)
.for_each([self](const frame_res_pair& frp) { .for_each([self](const event_t& event) {
// ... that simply pushes data back to the sender. // ... that simply pushes data back to the sender.
auto [pull, push] = frp; auto [pull, push] = event.data();
self->make_observable() self->make_observable()
.from_resource(pull) .from_resource(pull)
.do_on_next([](const frame& x) { .do_on_next([](const ws::frame& x) {
if (x.is_binary()) { if (x.is_binary()) {
std::cout << "*** received a binary WebSocket frame of size " std::cout << "*** received a binary WebSocket frame of size "
<< x.size() << '\n'; << x.size() << '\n';
...@@ -85,7 +84,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -85,7 +84,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
// Note: calling neither accept nor reject also rejects the connection. // Note: calling neither accept nor reject also rejects the connection.
}; };
// Set everything in motion. // Set everything in motion.
cn::web_socket::accept(sys, fd, std::move(sres), on_request); ws::accept(sys, fd, std::move(sres), on_request);
return EXIT_SUCCESS; return EXIT_SUCCESS;
} }
......
// Simple WebSocket server that sends local files from a working directory to
// the clients.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/cow_string.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
#include <iostream>
#include <utility>
using namespace std::literals;
static constexpr uint16_t default_port = 8080;
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.",
"First say to yourself what you would be; and then do what you have to do.",
"It's not what happens to you, but how you react to it that matters.",
"If you want to improve, be content to be thought foolish and stupid.",
"He who laughs at himself never runs out of things to laugh at.",
"It is impossible for a man to learn what he thinks he already knows.",
"Circumstances don't make the man, they only reveal him to himself.",
"People are not disturbed by things, but by the views they take of them.",
"Only the educated are free.",
};
static constexpr std::string_view seneca[] = {
"Luck is what happens when preparation meets opportunity.",
"All cruelty springs from weakness.",
"We suffer more often in imagination than in reality.",
"Difficulties strengthen the mind, as labor does the body.",
"If a man knows not to which port he sails, no wind is favorable.",
"It is the power of the mind to be unconquerable.",
"No man was ever wise by chance.",
"He suffers more than necessary, who suffers before it is necessary.",
"I shall never be ashamed of citing a bad author if the line is good.",
"Only time can heal what reason cannot.",
};
static constexpr std::string_view plato[] = {
"Love is a serious mental disease.",
"The measure of a man is what he does with power.",
"Ignorance, the root and stem of every evil.",
"Those who tell the stories rule society.",
"You should not honor men more than truth.",
"When men speak ill of thee, live so as nobody may believe them.",
"The beginning is the most important part of the work.",
"Necessity is the mother of invention.",
"The greatest wealth is to live content with little.",
"Beauty lies in the eyes of the beholder.",
};
caf::span<const std::string_view> quotes_by_path(std::string_view path) {
if (path == "/epictetus")
return caf::make_span(epictetus);
else if (path == "/seneca")
return caf::make_span(seneca);
else if (path == "/plato")
return caf::make_span(plato);
else
return {};
}
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
}
};
struct pick_random {
public:
pick_random() : engine_(std::random_device{}()) {
// nop
}
std::string_view operator()(caf::span<const std::string_view> quotes) {
assert(!quotes.empty());
auto dis = std::uniform_int_distribution<size_t>{0, quotes.size() - 1};
return quotes[dis(engine_)];
}
private:
std::minstd_rand engine_;
};
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.
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';
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);
// Note: we simply drop `pull` here, which will close the buffer.
});
});
// 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);
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#pragma once #pragma once
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/cow_tuple.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.hpp" #include "caf/net/web_socket/flow_connector.hpp"
...@@ -49,30 +50,50 @@ private: ...@@ -49,30 +50,50 @@ private:
namespace caf::net::web_socket { 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...>>();
}
/// Listens for incoming WebSocket connection on @p fd. /// Listens for incoming WebSocket connection on @p fd.
/// @param sys The host system. /// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this socket /// @param fd An accept socket in listening mode. For a TCP socket, this socket
/// must already listen to a port. /// must already listen to a port.
/// @param res A buffer resource that connects the server to a listener that /// @param out A buffer resource that connects the server to a listener that
/// processes the buffer pairs for each incoming connection. /// processes the buffer pairs for each incoming connection.
/// @param on_request Function object for accepting incoming requests. /// @param on_request Function object for accepting incoming requests.
/// @param limit The maximum amount of connections before closing @p fd. Passing /// @param limit The maximum amount of connections before closing @p fd. Passing
/// 0 means "no limit". /// 0 means "no limit".
template <template <class> class Transport = stream_transport, class Socket, template <template <class> class Transport = stream_transport, class Socket,
class OnRequest> class... Ts, class OnRequest>
void accept(actor_system& sys, Socket fd, void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
async::producer_resource<async::resource_pair<frame>> res,
OnRequest on_request, size_t limit = 0) { OnRequest on_request, size_t limit = 0) {
using trait_t = default_trait; using trait_t = default_trait;
using request_t = request<default_trait>; using request_t = request<default_trait, Ts...>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>, static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request"); "invalid signature found for on_request");
using factory_t = detail::ws_acceptor_factory<Transport, trait_t>; using factory_t = detail::ws_acceptor_factory<Transport, trait_t>;
using impl_t = connection_acceptor<Socket, factory_t>; using impl_t = connection_acceptor<Socket, factory_t>;
using conn_t = flow_connector_impl<OnRequest, trait_t>; using conn_t = flow_connector_impl<OnRequest, trait_t, Ts...>;
if (auto buf = res.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<conn_t>(std::move(buf), std::move(on_request)); auto conn = std::make_shared<conn_t>(std::move(on_request), std::move(buf));
auto factory = factory_t{std::move(conn)}; auto factory = factory_t{std::move(conn)};
auto ptr = make_socket_manager<impl_t>(std::move(fd), &mpx, limit, auto ptr = make_socket_manager<impl_t>(std::move(fd), &mpx, limit,
std::move(factory)); std::move(factory));
......
...@@ -44,9 +44,8 @@ public: ...@@ -44,9 +44,8 @@ public:
template <class LowerLayerPtr> template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr, const settings& cfg) { error init(net::socket_manager* mgr, LowerLayerPtr, const settings& cfg) {
request_type req; auto [err, pull, push] = conn_->on_request(cfg);
if (auto err = conn_->on_request(cfg, req); !err) { if (!err) {
auto& [pull, push] = req.ws_resources_;
in_ = consumer_type::try_open(mgr, pull); in_ = consumer_type::try_open(mgr, pull);
out_ = producer_type::try_open(mgr, push); out_ = producer_type::try_open(mgr, push);
CAF_ASSERT(in_ != nullptr); CAF_ASSERT(in_ != nullptr);
......
...@@ -9,6 +9,8 @@ ...@@ -9,6 +9,8 @@
#include "caf/net/web_socket/request.hpp" #include "caf/net/web_socket/request.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include <tuple>
namespace caf::net::web_socket { namespace caf::net::web_socket {
template <class Trait> template <class Trait>
...@@ -18,63 +20,57 @@ public: ...@@ -18,63 +20,57 @@ public:
using output_type = typename Trait::output_type; using output_type = typename Trait::output_type;
using app_res_pair_type = async::resource_pair<input_type, output_type>; using result_type = std::tuple<error, async::consumer_resource<input_type>,
async::producer_resource<output_type>>;
using ws_res_pair_type = async::resource_pair<output_type, input_type>;
using producer_type = async::blocking_producer<app_res_pair_type>;
virtual ~flow_connector() { virtual ~flow_connector() {
// nop // nop
} }
error on_request(const settings& cfg, request<Trait>& req) { virtual result_type on_request(const settings& cfg) = 0;
do_on_request(cfg, req);
if (req.accepted()) {
out_.push(req.app_resources_);
return none;
} else if (auto& err = req.reject_reason()) {
return err;
} else {
return make_error(sec::runtime_error,
"WebSocket request rejected without reason");
}
}
protected:
template <class T>
explicit flow_connector(T&& out) : out_(std::forward<T>(out)) {
// nop
}
private:
virtual void do_on_request(const settings& cfg, request<Trait>& req) = 0;
producer_type out_;
}; };
template <class Trait> template <class Trait>
using flow_connector_ptr = std::shared_ptr<flow_connector<Trait>>; using flow_connector_ptr = std::shared_ptr<flow_connector<Trait>>;
template <class OnRequest, class Trait> template <class OnRequest, class Trait, class... Ts>
class flow_connector_impl : public flow_connector<Trait> { class flow_connector_impl : public flow_connector<Trait> {
public: public:
using super = flow_connector<Trait>; using super = flow_connector<Trait>;
using producer_type = typename super::producer_type; using result_type = typename super::result_type;
using request_type = request<Trait, Ts...>;
using app_res_type = typename request_type::app_res_type;
using producer_type = async::blocking_producer<app_res_type>;
template <class T> template <class T>
flow_connector_impl(T&& out, OnRequest on_request) flow_connector_impl(OnRequest&& on_request, T&& out)
: super(std::forward<T>(out)), on_request_(std::move(on_request)) { : on_request_(std::move(on_request)), out_(std::forward<T>(out)) {
// nop // nop
} }
private: result_type on_request(const settings& cfg) override {
void do_on_request(const settings& cfg, request<Trait>& req) { request_type req;
on_request_(cfg, req); on_request_(cfg, req);
if (req.accepted()) {
out_.push(req.app_resources_);
auto& [pull, push] = req.ws_resources_;
return {error{}, std::move(pull), std::move(push)};
} else if (auto& err = req.reject_reason()) {
return {err, {}, {}};
} else {
auto def_err = make_error(sec::runtime_error,
"WebSocket request rejected without reason");
return {std::move(def_err), {}, {}};
}
} }
private:
OnRequest on_request_; OnRequest on_request_;
producer_type out_;
}; };
} // namespace caf::net::web_socket } // namespace caf::net::web_socket
...@@ -9,10 +9,13 @@ namespace caf::net::web_socket { ...@@ -9,10 +9,13 @@ namespace caf::net::web_socket {
template <class Trait> template <class Trait>
class flow_connector; class flow_connector;
template <class OnRequest, class Trait, class... Ts>
class flow_connector_impl;
template <class Trait> template <class Trait>
class flow_bridge; class flow_bridge;
template <class Trait> template <class Trait, class... Ts>
class request; class request;
class CAF_CORE_EXPORT frame; class CAF_CORE_EXPORT frame;
......
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#pragma once #pragma once
#include "caf/async/spsc_buffer.hpp" #include "caf/async/spsc_buffer.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
...@@ -12,30 +13,29 @@ namespace caf::net::web_socket { ...@@ -12,30 +13,29 @@ namespace caf::net::web_socket {
/// Represents a WebSocket connection request. /// Represents a WebSocket connection request.
/// @tparam Trait Converts between native and wire format. /// @tparam Trait Converts between native and wire format.
template <class Trait> template <class Trait, class... Ts>
class request { class request {
public: public:
template <class> template <class, class, class...>
friend class flow_connector; friend class flow_connector_impl;
template <class>
friend class flow_bridge;
using input_type = typename Trait::input_type; using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type; using output_type = typename Trait::output_type;
using app_res_pair_type = async::resource_pair<input_type, output_type>; using app_res_type = cow_tuple<async::consumer_resource<input_type>,
async::producer_resource<output_type>, Ts...>;
using ws_res_pair_type = async::resource_pair<output_type, input_type>; using ws_res_type = async::resource_pair<output_type, input_type>;
void accept() { void accept(Ts... worker_args) {
if (accepted()) if (accepted())
return; return;
auto [app_pull, ws_push] = async::make_spsc_buffer_resource<input_type>(); auto [app_pull, ws_push] = async::make_spsc_buffer_resource<input_type>();
auto [ws_pull, app_push] = async::make_spsc_buffer_resource<output_type>(); auto [ws_pull, app_push] = async::make_spsc_buffer_resource<output_type>();
ws_resources_ = std::tie(ws_pull, ws_push); ws_resources_ = std::tie(ws_pull, ws_push);
app_resources_ = std::tie(app_pull, app_push); app_resources_ = make_cow_tuple(app_pull, app_push,
std::move(worker_args)...);
accepted_ = true; accepted_ = true;
} }
...@@ -55,8 +55,8 @@ public: ...@@ -55,8 +55,8 @@ public:
private: private:
bool accepted_ = false; bool accepted_ = false;
ws_res_pair_type ws_resources_; ws_res_type ws_resources_;
app_res_pair_type app_resources_; app_res_type app_resources_;
error reject_reason_; error reject_reason_;
}; };
......
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