Commit d9a8f6d3 authored by Dominik Charousset's avatar Dominik Charousset

Re-implement web_socket::accept

parent a2abe9a3
......@@ -127,3 +127,16 @@ if(TARGET CAF::io)
endif()
endif()
if(TARGET CAF::net)
function(add_net_example folder name)
add_example(${folder} ${name} ${ARGN})
target_link_libraries(${name} PRIVATE CAF::internal CAF::net)
endfunction()
else()
function(add_net_example)
# nop
endfunction()
endif()
add_net_example(web_socket echo)
#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/tcp_accept_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <utility>
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");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
// 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 frame = cn::web_socket::frame;
using frame_res_pair = caf::async::resource_pair<frame>;
// Create a buffer of frame buffer pairs. Each buffer pair represents the
// input and output channel for a single WebSocket connection. The outer
// 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) {
// For each buffer pair, we create a new flow ...
self->make_observable()
.from_resource(worker_res)
.for_each([self](const frame_res_pair& frp) {
// ... that simply pushes data back to the sender.
auto [pull, push] = frp;
self->make_observable()
.from_resource(pull)
.do_on_next([](const 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);
});
});
// Handler for
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 refused.
auto err = caf::make_error(caf::sec::invalid_argument,
"unrecognized path, try '/'");
req.reject(std::move(err));
}
};
// Set everything in motion.
cn::web_socket::accept(sys, fd, std::move(sres), on_request);
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -124,6 +124,10 @@ public:
// nop
}
explicit blocking_producer(spsc_buffer_ptr<T> buf) {
impl_.emplace(std::move(buf));
}
~blocking_producer() {
if (impl_)
impl_->close();
......
......@@ -462,19 +462,23 @@ private:
intrusive_ptr<resource_ctrl<T, true>> ctrl_;
};
/// Creates spsc buffer and returns two resources connected by that buffer.
template <class T1, class T2 = T1>
using resource_pair = std::pair<consumer_resource<T1>, producer_resource<T2>>;
/// Creates an @ref spsc_buffer and returns two resources connected by that
/// buffer.
template <class T>
std::pair<consumer_resource<T>, producer_resource<T>>
resource_pair<T>
make_spsc_buffer_resource(size_t buffer_size, size_t min_request_size) {
using buffer_type = spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
return {async::consumer_resource<T>{buf}, async::producer_resource<T>{buf}};
}
/// Creates spsc buffer and returns two resources connected by that buffer.
/// Creates an @ref spsc_buffer and returns two resources connected by that
/// buffer.
template <class T>
std::pair<consumer_resource<T>, producer_resource<T>>
make_spsc_buffer_resource() {
resource_pair<T> make_spsc_buffer_resource() {
return make_spsc_buffer_resource<T>(defaults::flow::buffer_size,
defaults::flow::min_demand);
}
......
......@@ -16,15 +16,12 @@ class do_on_next {
public:
using trait = detail::get_callable_trait_t<F>;
static_assert(!std::is_same_v<typename trait::result_type, void>,
"do_on_next functions may not return void");
static_assert(trait::num_args == 1,
"do_on_next functions must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = std::decay_t<typename trait::result_type>;
using output_type = input_type;
explicit do_on_next(F fn) : fn_(std::move(fn)) {
// nop
......
......@@ -34,6 +34,8 @@ caf_add_component(
src/net/http/status.cpp
src/net/http/v1.cpp
src/net/middleman.cpp
src/net/web_socket/default_trait.cpp
src/net/web_socket/frame.cpp
src/net/web_socket/handshake.cpp
src/network_socket.cpp
src/pipe_socket.cpp
......
// 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/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
#include "caf/net/web_socket/flow_connector.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::detail {
template <template <class> class Transport, class Trait>
class ws_acceptor_factory {
public:
using connector_pointer = net::web_socket::flow_connector_ptr<Trait>;
explicit ws_acceptor_factory(connector_pointer connector)
: connector_(std::move(connector)) {
// nop
}
error init(net::socket_manager*, const settings&) {
return none;
}
template <class Socket>
net::socket_manager_ptr make(Socket fd, net::multiplexer* mpx) {
using app_t = net::web_socket::flow_bridge<Trait>;
using stack_t = Transport<net::web_socket::server<app_t>>;
return net::make_socket_manager<stack_t>(fd, mpx, connector_);
}
void abort(const error&) {
// nop
}
private:
connector_pointer connector_;
};
} // namespace caf::detail
namespace caf::net::web_socket {
/// 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 res 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 limit The maximum amount of connections before closing @p fd. Passing
/// 0 means "no limit".
template <template <class> class Transport = stream_transport, class Socket,
class OnRequest>
void accept(actor_system& sys, Socket fd,
async::producer_resource<async::resource_pair<frame>> res,
OnRequest on_request, size_t limit = 0) {
using trait_t = default_trait;
using request_t = request<default_trait>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request");
using factory_t = detail::ws_acceptor_factory<Transport, trait_t>;
using impl_t = connection_acceptor<Socket, factory_t>;
using conn_t = flow_connector_impl<OnRequest, trait_t>;
if (auto buf = res.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto conn = std::make_shared<conn_t>(std::move(buf), std::move(on_request));
auto factory = factory_t{std::move(conn)};
auto ptr = make_socket_manager<impl_t>(std::move(fd), &mpx, limit,
std::move(factory));
mpx.init(ptr);
}
// TODO: accept() should return a disposable to stop the WebSocket server.
}
} // 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/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/web_socket/fwd.hpp"
#include <string_view>
#include <vector>
namespace caf::net::web_socket {
class CAF_NET_EXPORT default_trait {
public:
/// The input type of the application, i.e., what that flows from the
/// WebSocket to the application layer.
using input_type = frame;
/// The output type of the application, i.e., what flows from the application
/// layer to the WebSocket.
using output_type = frame;
bool converts_to_binary(const output_type& x);
bool convert(const output_type& x, byte_buffer& bytes);
bool convert(const output_type& x, std::vector<char>& text);
bool convert(const_byte_span bytes, input_type& x);
bool convert(std::string_view text, input_type& x);
};
} // 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/async/spsc_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/net/consumer_adapter.hpp"
#include "caf/net/producer_adapter.hpp"
#include "caf/net/web_socket/flow_connector.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/sec.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/no_auto_reading.hpp"
#include <utility>
namespace caf::net::web_socket {
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public tag::no_auto_reading {
public:
using input_tag = tag::mixed_message_oriented;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = consumer_adapter<async::spsc_buffer<output_type>>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = producer_adapter<async::spsc_buffer<input_type>>;
using request_type = request<Trait>;
using connector_pointer = flow_connector_ptr<Trait>;
explicit flow_bridge(connector_pointer conn) : conn_(std::move(conn)) {
// nop
}
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr, const settings& cfg) {
request_type req;
if (auto err = conn_->on_request(cfg, req); !err) {
auto& [pull, push] = req.ws_resources_;
in_ = consumer_type::try_open(mgr, pull);
out_ = producer_type::try_open(mgr, push);
CAF_ASSERT(in_ != nullptr);
CAF_ASSERT(out_ != nullptr);
conn_ = nullptr;
return none;
} else {
conn_ = nullptr;
return err;
}
}
template <class LowerLayerPtr>
bool write(LowerLayerPtr down, const output_type& item) {
if (trait_.converts_to_binary(item)) {
down->begin_binary_message();
auto& bytes = down->binary_message_buffer();
return trait_.convert(item, bytes) && down->end_binary_message();
} else {
down->begin_text_message();
auto& text = down->text_message_buffer();
return trait_.convert(item, text) && down->end_text_message();
}
}
template <class LowerLayerPtr>
struct write_helper {
using bridge_type = flow_bridge;
bridge_type* bridge;
LowerLayerPtr down;
bool aborted = false;
size_t consumed = 0;
error err;
write_helper(bridge_type* bridge, LowerLayerPtr down)
: bridge(bridge), down(down) {
// nop
}
void on_next(const output_type& item) {
if (!bridge->write(down, item)) {
aborted = true;
return;
}
}
void on_complete() {
// nop
}
void on_error(const error& x) {
err = x;
}
};
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
write_helper<LowerLayerPtr> helper{this, down};
while (down->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down->send_close_message(helper.err);
} else {
down->send_close_message();
}
in_ = nullptr;
} else if (helper.aborted) {
down->abort_reason(make_error(sec::conversion_failed));
in_->cancel();
in_ = nullptr;
return false;
} else if (consumed == 0) {
return true;
}
}
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
return !in_ || !in_->has_data();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::socket_disconnected || reason == sec::disposed)
out_->close();
else
out_->abort(reason);
out_ = nullptr;
}
if (in_) {
in_->cancel();
in_ = nullptr;
}
}
template <class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr down, byte_span buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
return -1;
}
input_type val;
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
return -1;
}
if (out_->push(std::move(val)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
template <class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, std::string_view buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
return -1;
}
input_type val;
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
return -1;
}
if (out_->push(std::move(val)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
private:
/// The output of the application. Serialized to the socket.
intrusive_ptr<consumer_type> in_;
/// The input to the application. Deserialized from the socket.
intrusive_ptr<producer_type> out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
};
} // 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/async/blocking_producer.hpp"
#include "caf/error.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/sec.hpp"
namespace caf::net::web_socket {
template <class Trait>
class flow_connector {
public:
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using app_res_pair_type = async::resource_pair<input_type, 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() {
// nop
}
error on_request(const settings& cfg, request<Trait>& req) {
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>
using flow_connector_ptr = std::shared_ptr<flow_connector<Trait>>;
template <class OnRequest, class Trait>
class flow_connector_impl : public flow_connector<Trait> {
public:
using super = flow_connector<Trait>;
using producer_type = typename super::producer_type;
template <class T>
flow_connector_impl(T&& out, OnRequest on_request)
: super(std::forward<T>(out)), on_request_(std::move(on_request)) {
// nop
}
private:
void do_on_request(const settings& cfg, request<Trait>& req) {
on_request_(cfg, req);
}
OnRequest 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/async/fwd.hpp"
#include "caf/byte_span.hpp"
#include "caf/config.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/raise_error.hpp"
#include "caf/type_id.hpp"
#include <cstddef>
#include <string_view>
#ifdef CAF_CLANG
# pragma clang diagnostic push
# pragma clang diagnostic ignored "-Wc99-extensions"
#elif defined(CAF_GCC)
# pragma GCC diagnostic push
# pragma GCC diagnostic ignored "-Wpedantic"
#elif defined(CAF_MSVC)
# pragma warning(push)
# pragma warning(disable : 4200)
#endif
namespace caf::net::web_socket {
/// An implicitly shared type for passing along WebSocket data frames, i.e.,
/// text or binary frames.
class CAF_NET_EXPORT frame {
public:
frame() = default;
frame(frame&&) = default;
frame(const frame&) = default;
frame& operator=(frame&&) = default;
frame& operator=(const frame&) = default;
explicit frame(const_byte_span bytes);
explicit frame(std::string_view text);
explicit operator bool() const noexcept {
return static_cast<bool>(data_);
}
size_t size() const noexcept {
return data_ ? data_->size() : 0;
}
bool empty() const noexcept {
return data_ ? data_->size() == 0 : true;
}
void swap(frame& other) {
data_.swap(other.data_);
}
bool is_binary() const noexcept {
return data_ && data_->is_binary();
}
bool is_text() const noexcept {
return data_ && !data_->is_binary();
}
const_byte_span as_binary() const noexcept;
std::string_view as_text() const noexcept;
private:
class data {
public:
data() = delete;
data(const data&) = delete;
data& operator=(const data&) = delete;
explicit data(bool bin, size_t size) : rc_(1), bin_(bin), size_(size) {
// nop
}
// -- reference counting ---------------------------------------------------
bool unique() const noexcept {
return rc_ == 1;
}
void ref() const noexcept {
rc_.fetch_add(1, std::memory_order_relaxed);
}
void deref() noexcept;
friend void intrusive_ptr_add_ref(const data* ptr) {
ptr->ref();
}
friend void intrusive_ptr_release(data* ptr) {
ptr->deref();
}
// -- properties -----------------------------------------------------------
size_t size() noexcept {
return size_;
}
bool is_binary() const noexcept {
return bin_;
}
std::byte* storage() noexcept {
return storage_;
}
const std::byte* storage() const noexcept {
return storage_;
}
private:
static constexpr size_t padding_size = CAF_CACHE_LINE_SIZE
- sizeof(std::atomic<size_t>);
mutable std::atomic<size_t> rc_;
[[maybe_unused]] std::byte padding_[padding_size];
bool bin_;
size_t size_;
std::byte storage_[];
};
explicit frame(intrusive_ptr<data> ptr) : data_(std::move(ptr)) {
// nop
}
void alloc(bool is_binary, size_t num_bytes);
intrusive_ptr<data> data_;
};
} // namespace caf::net::web_socket
#ifdef CAF_CLANG
# pragma clang diagnostic pop
#elif defined(CAF_GCC)
# pragma GCC diagnostic pop
#elif defined(CAF_MSVC)
# pragma warning(pop)
#endif
// 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
namespace caf::net::web_socket {
template <class Trait>
class flow_connector;
template <class Trait>
class flow_bridge;
template <class Trait>
class request;
class CAF_CORE_EXPORT frame;
} // 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/async/spsc_buffer.hpp"
#include "caf/error.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net::web_socket {
/// Represents a WebSocket connection request.
/// @tparam Trait Converts between native and wire format.
template <class Trait>
class request {
public:
template <class>
friend class flow_connector;
template <class>
friend class flow_bridge;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using app_res_pair_type = async::resource_pair<input_type, output_type>;
using ws_res_pair_type = async::resource_pair<output_type, input_type>;
void accept() {
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_ = std::tie(ws_pull, ws_push);
app_resources_ = std::tie(app_pull, app_push);
accepted_ = true;
}
void reject(error reason) {
reject_reason_ = reason;
if (accepted_)
accepted_ = false;
}
bool accepted() const noexcept {
return accepted_;
}
const error& reject_reason() const noexcept {
return reject_reason_;
}
private:
bool accepted_ = false;
ws_res_pair_type ws_resources_;
app_res_pair_type app_resources_;
error reject_reason_;
};
} // namespace caf::net::web_socket
......@@ -267,159 +267,3 @@ socket_manager_ptr make_server(multiplexer& mpx, Socket fd,
}
} // namespace caf::net::web_socket
namespace caf::detail {
template <class T, class Trait>
using on_request_result = expected<
std::tuple<async::consumer_resource<T>, // For the connection to read from.
async::producer_resource<T>, // For the connection to write to.
Trait>>; // For translating between native and wire format.
template <class T>
struct is_on_request_result : std::false_type {};
template <class T, class Trait>
struct is_on_request_result<on_request_result<T, Trait>> : std::true_type {};
template <class T>
struct on_request_trait;
template <class T, class ServerTrait>
struct on_request_trait<on_request_result<T, ServerTrait>> {
using value_type = T;
using trait_type = ServerTrait;
};
template <class OnRequest>
class ws_accept_trait {
public:
using on_request_r
= decltype(std::declval<OnRequest&>()(std::declval<const settings&>()));
static_assert(is_on_request_result<on_request_r>::value,
"OnRequest must return an on_request_result");
using on_request_t = on_request_trait<on_request_r>;
using value_type = typename on_request_t::value_type;
using decorated_trait = typename on_request_t::trait_type;
using consumer_resource_t = async::consumer_resource<value_type>;
using producer_resource_t = async::producer_resource<value_type>;
using in_out_tuple = std::tuple<consumer_resource_t, producer_resource_t>;
using init_res_t = expected<in_out_tuple>;
ws_accept_trait() = delete;
explicit ws_accept_trait(OnRequest on_request) : state_(on_request) {
// nop
}
ws_accept_trait(ws_accept_trait&&) = default;
ws_accept_trait& operator=(ws_accept_trait&&) = default;
init_res_t init(const settings& cfg) {
auto f = std::move(std::get<OnRequest>(state_));
if (auto res = f(cfg)) {
auto& [in, out, trait] = *res;
if (auto err = trait.init(cfg)) {
state_ = none;
return std::move(res.error());
} else {
state_ = std::move(trait);
return std::make_tuple(std::move(in), std::move(out));
}
} else {
state_ = none;
return std::move(res.error());
}
}
bool converts_to_binary(const value_type& x) {
return decorated().converts_to_binary(x);
}
bool convert(const value_type& x, byte_buffer& bytes) {
return decorated().convert(x, bytes);
}
bool convert(const value_type& x, std::vector<char>& text) {
return decorated().convert(x, text);
}
bool convert(const_byte_span bytes, value_type& x) {
return decorated().convert(bytes, x);
}
bool convert(std::string_view text, value_type& x) {
return decorated().convert(text, x);
}
private:
decorated_trait& decorated() {
return std::get<decorated_trait>(state_);
}
std::variant<none_t, OnRequest, decorated_trait> state_;
};
template <template <class> class Transport, class OnRequest>
class ws_acceptor_factory {
public:
explicit ws_acceptor_factory(OnRequest on_request)
: on_request_(std::move(on_request)) {
// nop
}
error init(net::socket_manager*, const settings&) {
return none;
}
template <class Socket>
net::socket_manager_ptr make(Socket fd, net::multiplexer* mpx) {
using trait_t = ws_accept_trait<OnRequest>;
using value_type = typename trait_t::value_type;
using app_t = net::message_flow_bridge<value_type, trait_t,
tag::mixed_message_oriented>;
using stack_t = Transport<net::web_socket::server<app_t>>;
return net::make_socket_manager<stack_t>(fd, mpx, trait_t{on_request_});
}
void abort(const error&) {
// nop
}
private:
OnRequest on_request_;
};
} // namespace caf::detail
namespace caf::net::web_socket {
/// Creates a WebSocket server on the connected socket `fd`.
/// @param mpx The multiplexer that takes ownership of the socket.
/// @param fd An accept socket in listening mode. For a TCP socket, this socket
/// must already listen to an address plus port.
/// @param on_request Function object for turning requests into a tuple
/// consisting of a consumer resource, a producer resource,
/// and a trait. These arguments get forwarded to
/// @ref make_server internally.
template <template <class> class Transport = stream_transport, class Socket,
class OnRequest>
void accept(multiplexer& mpx, Socket fd, OnRequest on_request,
size_t limit = 0) {
using factory = detail::ws_acceptor_factory<Transport, OnRequest>;
using impl = connection_acceptor<Socket, factory>;
auto ptr = make_socket_manager<impl>(std::move(fd), &mpx, limit,
factory{std::move(on_request)});
mpx.init(ptr);
}
} // namespace caf::net::web_socket
......@@ -123,7 +123,7 @@ std::pair<status, std::string_view> header::parse(std::string_view raw) {
}
// The path must form a valid URI when prefixing a scheme. We don't actually
// care about the scheme, so just use "foo" here for the validation step.
if (auto res = make_uri("nil://host" + std::string{request_uri_str})) {
if (auto res = make_uri("nil:" + std::string{request_uri_str})) {
uri_ = std::move(*res);
} else {
CAF_LOG_DEBUG("Failed to parse URI" << request_uri_str << "->"
......
// 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.
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/frame.hpp"
namespace caf::net::web_socket {
bool default_trait::converts_to_binary(const output_type& x) {
return x.is_binary();
}
bool default_trait::convert(const output_type& x, byte_buffer& bytes) {
auto src = x.as_binary();
bytes.insert(bytes.end(), src.begin(), src.end());
return true;
}
bool default_trait::convert(const output_type& x, std::vector<char>& text) {
auto src = x.as_text();
text.insert(text.end(), src.begin(), src.end());
return true;
}
bool default_trait::convert(const_byte_span bytes, input_type& x) {
x = frame{bytes};
return true;
}
bool default_trait::convert(std::string_view text, input_type& x) {
x = frame{text};
return true;
}
} // namespace caf::net::web_socket
#include "caf/net/web_socket/frame.hpp"
#include <cstring>
#include <new>
namespace caf::net::web_socket {
frame::frame(const_byte_span bytes) {
alloc(true, bytes.size());
memcpy(data_->storage(), bytes.data(), bytes.size());
}
frame::frame(std::string_view text) {
alloc(false, text.size());
memcpy(data_->storage(), text.data(), text.size());
}
const_byte_span frame::as_binary() const noexcept {
return {data_->storage(), data_->size()};
}
std::string_view frame::as_text() const noexcept {
return {reinterpret_cast<const char*>(data_->storage()), data_->size()};
}
void frame::data::deref() noexcept {
if (unique() || rc_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
this->~data();
free(this);
}
}
void frame::alloc(bool is_binary, size_t num_bytes) {
auto total_size = sizeof(data) + num_bytes;
auto vptr = malloc(total_size);
data_.reset(new (vptr) data(is_binary, num_bytes), false);
}
} // 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