Commit fb6e8bef authored by Dominik Charousset's avatar Dominik Charousset

Implement WebSocket framing

parent 43f526d1
......@@ -34,10 +34,12 @@ endfunction()
# -- add library targets -------------------------------------------------------
add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
#src/net/backend/tcp.cpp
#src/net/backend/test.cpp
#src/actor_proxy_impl.cpp
#src/basp/application.cpp
#src/endpoint_manager.cpp
#src/net/backend/tcp.cpp
#src/net/backend/test.cpp
#src/net/endpoint_manager_queue.cpp
src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp
......@@ -45,14 +47,13 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp
src/defaults.cpp
#src/endpoint_manager.cpp
src/detail/rfc6455.cpp
src/header.cpp
src/host.cpp
src/ip.cpp
src/message_queue.cpp
src/multiplexer.cpp
src/net/actor_shell.cpp
#src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
src/net/packet_writer.cpp
......@@ -80,7 +81,8 @@ set_property(TARGET libcaf_net_obj PROPERTY POSITION_INDEPENDENT_CODE ON)
caf_net_set_default_properties(libcaf_net_obj libcaf_net)
target_include_directories(libcaf_net INTERFACE
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>)
$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>
$<BUILD_INTERFACE:${CMAKE_BINARY_DIR}>)
add_library(CAF::net ALIAS libcaf_net)
......@@ -127,6 +129,7 @@ caf_incubator_add_test_suites(caf-net-test
#application
convert_ip_endpoint
datagram_socket
detail.rfc6455
#datagram_transport
#doorman
#endpoint_manager
......@@ -139,7 +142,7 @@ caf_incubator_add_test_suites(caf-net-test
#net.basp.ping_pong
#net.basp.worker
net.length_prefix_framing
net.web_socket
net.web_socket_server
network_socket
pipe_socket
socket
......
......@@ -69,6 +69,9 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
interface base [role: lower layer] {
/// Returns whether the layer has output slots available.
bool can_send_more() const noexcept;
/// Returns the managed socket.
auto handle() const noexcept;
}
interface stream_oriented [role: upper layer] {
......@@ -179,3 +182,66 @@ stack *up*. Outgoing data always travels the protocol stack *down*.
bool end_message();
}
interface mixed_message_oriented [role: upper layer] {
/// Consumes a binary message from the lower layer.
/// @param down Reference to the lower layer that received the message.
/// @param buffer Payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an
/// error.
/// @note Discarded data is lost permanently.
/// @note When returning a negative value for the number of consumed bytes,
/// clients must also call `down.set_read_error(...)` with an
/// appropriate error code.
template <class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr down, byte_span buffer);
/// Consumes a text message from the lower layer.
/// @param down Reference to the lower layer that received the message.
/// @param text Payload of the received message. The encoding depends on the
/// application.
/// @returns The number of consumed characters or a negative value to signal
/// an error.
/// @note Discarded data is lost permanently.
/// @note When returning a negative value for the number of consumed
/// characters, clients must also call `down.set_read_error(...)` with
/// an appropriate error code.
template <class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, string_view text);
}
interface mixed_message_oriented [role: lower layer] {
/// Prepares the layer for an outgoing binary message, e.g., by allocating
/// an output buffer as necessary.
void begin_binary_message();
/// Returns a reference to the buffer for assembling the current message.
/// Users may only call this function and write to the buffer between
/// calling `begin_binary_message()` and `end_binary_message()`.
/// @note Lower layers may pre-fill the buffer, e.g., to prefix custom
/// headers.
byte_buffer& binary_message_buffer();
/// Seals and prepares a binary message for transfer.
/// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code.
template <class LowerLayerPtr>
bool end_binary_message();
/// Prepares the layer for an outgoing text message, e.g., by allocating
/// an output buffer as necessary.
void begin_text_message();
/// Returns a reference to the buffer for assembling the current message.
/// Users may only call this function and write to the buffer between
/// calling `begin_text_message()` and `end_text_message()`.
/// @note Lower layers may pre-fill the buffer, e.g., to prefix custom
/// headers.
std::vector<char>& text_message_buffer();
/// Seals and prepares a text message for transfer.
/// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code.
template <class LowerLayerPtr>
bool end_text_message();
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/byte.hpp"
#include "caf/detail/net_export.hpp"
#include <cstdint>
#include <vector>
namespace caf::detail {
struct CAF_NET_EXPORT rfc6455 {
// -- member types -----------------------------------------------------------
using binary_buffer = std::vector<byte>;
struct header {
bool fin;
uint8_t opcode;
uint32_t mask_key;
uint64_t payload_len;
};
// -- constants --------------------------------------------------------------
static constexpr uint8_t continuation_frame = 0x00;
static constexpr uint8_t text_frame = 0x01;
static constexpr uint8_t binary_frame = 0x02;
static constexpr uint8_t connection_close = 0x08;
static constexpr uint8_t ping = 0x09;
static constexpr uint8_t pong = 0x0A;
// -- utility functions ------------------------------------------------------
static void mask_data(uint32_t key, span<char> data);
static void mask_data(uint32_t key, span<byte> data);
static void assemble_frame(uint32_t mask_key, span<const char> data,
binary_buffer& out);
static void assemble_frame(uint32_t mask_key, span<const byte> data,
binary_buffer& out);
static void assemble_frame(uint8_t opcode, uint32_t mask_key,
span<const byte> data, binary_buffer& out);
static ptrdiff_t decode_header(span<const byte> data, header& hdr);
};
} // namespace caf::detail
......@@ -19,6 +19,8 @@
#pragma once
#include "caf/actor_traits.hpp"
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
......@@ -34,7 +36,7 @@
namespace caf::net {
///
class actor_shell
class CAF_NET_EXPORT actor_shell
: public extend<local_actor, actor_shell>::with<mixin::sender,
mixin::requester>,
public dynamically_typed_actor_base,
......@@ -65,9 +67,11 @@ public:
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators --------------------
explicit actor_shell(actor_config& cfg, socket_manager* owner);
actor_shell(actor_config& cfg, socket_manager* owner);
~actor_shell() override;
......@@ -76,6 +80,18 @@ public:
/// Detaches the shell from its owner and closes the mailbox.
void quit(error reason);
/// Overrides the callbacks for incoming messages.
template <class... Fs>
void set_behavior(Fs... fs) {
bhvr_ = behavior{std::move(fs)...};
}
/// Overrides the default handler for unexpected messages.
template <class F>
void set_fallback(F f) {
fallback_ = make_type_erased_callback(std::move(f));
}
// -- mailbox access ---------------------------------------------------------
auto& mailbox() noexcept {
......@@ -93,13 +109,9 @@ public:
// -- message processing -----------------------------------------------------
/// Dequeues and processes the next message from the mailbox.
/// @param bhvr Available message handlers.
/// @param fallback Callback for processing message that failed to match
/// `bhvr`.
/// @returns `true` if a message was dequeued and process, `false` if the
/// mailbox was empty.
bool consume_message(behavior& bhvr,
callback<result<message>(message&)>& fallback);
bool consume_message();
/// Adds a callback for a multiplexed response.
void add_multiplexed_response_handler(message_id response_id, behavior bhvr);
......@@ -130,13 +142,19 @@ private:
// Points to the owning manager (nullptr after quit was called).
socket_manager* owner_;
// Handler for consuming messages from the mailbox.
behavior bhvr_;
// Handler for unexpected messages.
fallback_handler fallback_;
// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
};
/// An "owning" pointer to an actor shell in the sense that it calls `quit()` on
/// the shell when going out of scope.
class actor_shell_ptr {
class CAF_NET_EXPORT actor_shell_ptr {
public:
friend class socket_manager;
......
......@@ -57,7 +57,7 @@ class CAF_NET_EXPORT application {
public:
// -- member types -----------------------------------------------------------
using byte_span = span<const byte>;
using byte_span = span<byte>;
using hub_type = detail::worker_hub<worker>;
......
......@@ -29,102 +29,104 @@
namespace caf::net {
/// A doorman accepts TCP connections and creates stream_transports to handle
/// them.
template <class Factory>
class doorman {
/// A connection_acceptor accepts connections from an accept socket and creates
/// socket managers to handle them via its factory.
template <class Socket, class Factory>
class connection_acceptor {
public:
// -- member types -----------------------------------------------------------
using factory_type = Factory;
using input_tag = tag::io_event_oriented;
using socket_type = Socket;
using application_type = typename Factory::application_type;
using factory_type = Factory;
// -- constructors, destructors, and assignment operators --------------------
explicit doorman(net::tcp_accept_socket acceptor, factory_type factory)
: acceptor_(acceptor), factory_(std::move(factory)) {
template <class... Ts>
explicit connection_acceptor(Ts&&... xs) : factory_(std::forward<Ts>(xs)...) {
// nop
}
// -- properties -------------------------------------------------------------
net::tcp_accept_socket handle() {
return acceptor_;
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent& parent) {
// TODO: is initializing application factory nessecary?
if (auto err = factory_.init(parent))
template <class ParentPtr>
error init(socket_manager* owner, ParentPtr parent, const settings& config) {
CAF_LOG_TRACE("");
owner_ = owner;
if (auto err = factory_.init(owner, config))
return err;
parent->register_reading();
return none;
}
template <class Parent>
bool handle_read_event(Parent& parent) {
auto x = net::accept(acceptor_);
if (!x) {
template <class ParentPtr>
bool handle_read_event(ParentPtr parent) {
CAF_LOG_TRACE("");
if (auto x = accept(parent->handle())) {
socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr());
CAF_ASSERT(child != nullptr);
if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
parent->abort_reason(std::move(err));
return false;
}
return true;
} else {
CAF_LOG_ERROR("accept failed:" << x.error());
return false;
}
auto mpx = parent.multiplexer();
if (!mpx) {
CAF_LOG_DEBUG("unable to get multiplexer from parent");
return false;
}
auto child = make_endpoint_manager(
mpx, parent.system(),
stream_transport<application_type>{*x, factory_.make()});
if (auto err = child->init())
return false;
return true;
}
template <class Parent>
bool handle_write_event(Parent&) {
CAF_LOG_ERROR("doorman received write event");
template <class ParentPtr>
bool handle_write_event(ParentPtr) {
CAF_LOG_ERROR("connection_acceptor received write event");
return false;
}
template <class Parent>
void
resolve(Parent&, [[maybe_unused]] const uri& locator, const actor& listener) {
CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(locator));
anon_send(listener, resolve_atom_v, "doormen cannot resolve paths");
template <class ParentPtr>
void abort(ParentPtr, const error& reason) {
CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
factory_.abort(reason);
}
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) {
CAF_LOG_ERROR("doorman received new_proxy" << CAF_ARG(peer) << CAF_ARG(id));
CAF_IGNORE_UNUSED(peer);
CAF_IGNORE_UNUSED(id);
private:
factory_type factory_;
socket_manager* owner_;
settings cfg_;
};
/// Converts a function object into a factory object for a
/// @ref connection_acceptor.
template <class FunctionObject>
class connection_acceptor_factory_adapter {
public:
explicit connection_acceptor_factory_adapter(FunctionObject f)
: f_(std::move(f)) {
// nop
}
void local_actor_down(endpoint_manager&, const node_id& peer, actor_id id,
error reason) {
CAF_LOG_ERROR("doorman received local_actor_down"
<< CAF_ARG(peer) << CAF_ARG(id) << CAF_ARG(reason));
CAF_IGNORE_UNUSED(peer);
CAF_IGNORE_UNUSED(id);
CAF_IGNORE_UNUSED(reason);
connection_acceptor_factory_adapter(connection_acceptor_factory_adapter&&)
= default;
error init(socket_manager*, const settings&) {
return none;
}
template <class Parent>
void timeout(Parent&, [[maybe_unused]] const std::string& tag,
[[maybe_unused]] uint64_t id) {
CAF_LOG_ERROR("doorman received timeout" << CAF_ARG(tag) << CAF_ARG(id));
template <class Socket>
socket_manager_ptr make(Socket connected_socket, multiplexer* mpx) {
return f_(std::move(connected_socket), mpx);
}
void handle_error([[maybe_unused]] sec err) {
CAF_LOG_ERROR("doorman encounterd error: " << err);
void abort(const error&) {
// nop
}
private:
net::tcp_accept_socket acceptor_;
factory_type factory_;
FunctionObject f_;
};
} // namespace caf::net
......@@ -41,7 +41,7 @@ namespace caf::net {
template <class UpperLayer>
class length_prefix_framing {
public:
using byte_span = span<const byte>;
using byte_span = span<byte>;
using input_tag = tag::stream_oriented;
......
......@@ -27,9 +27,11 @@
#include "caf/detail/net_export.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/scoped_actor.hpp"
namespace caf::net {
......@@ -54,6 +56,28 @@ public:
~middleman() override;
// -- socket manager functions -----------------------------------------------
///
/// @param sock An accept socket in listening mode. For a TCP socket, this
/// socket must already listen to an address plus port.
/// @param factory
template <class Socket, class Factory>
auto make_acceptor(Socket sock, Factory factory) {
using connected_socket_type = typename Socket::connected_socket_type;
if constexpr (detail::is_callable_with<Factory, connected_socket_type,
multiplexer*>::value) {
connection_acceptor_factory_adapter<Factory> adapter{std::move(factory)};
return make_acceptor(std::move(sock), std::move(adapter));
} else {
using impl = connection_acceptor<Socket, Factory>;
auto ptr = make_socket_manager<impl>(std::move(sock), &mpx_,
std::move(factory));
mpx_.init(ptr);
return ptr;
}
}
// -- interface functions ----------------------------------------------------
void start() override;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Wraps a pointer to a stream-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// mixed-message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class mixed_message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_binary_message() {
lptr_->begin_binary_message(llptr_);
}
auto& binary_message_buffer() {
return lptr_->binary_message_buffer(llptr_);
}
void end_binary_message() {
lptr_->end_binary_message(llptr_);
}
void begin_text_message() {
lptr_->begin_text_message(llptr_);
}
auto& text_message_buffer() {
return lptr_->text_message_buffer(llptr_);
}
void end_text_message() {
lptr_->end_text_message(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
mixed_message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
mixed_message_oriented_layer_ptr(const mixed_message_oriented_layer_ptr&)
= default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_mixed_message_oriented_layer_ptr(Layer* this_layer,
LowerLayerPtr down) {
using result_t = mixed_message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
......@@ -84,6 +84,10 @@ public:
/// @thread-safe
void register_writing(const socket_manager_ptr& mgr);
/// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe
void init(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect.
/// @thread-safe
......
......@@ -36,6 +36,16 @@ public:
using msg_buf = std::array<byte, sizeof(intptr_t) + 1>;
// -- constants --------------------------------------------------------------
static constexpr uint8_t register_reading_code = 0x00;
static constexpr uint8_t register_writing_code = 0x01;
static constexpr uint8_t init_manager_code = 0x02;
static constexpr uint8_t shutdown_code = 0x04;
// -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer* parent);
......
......@@ -18,20 +18,27 @@
#pragma once
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/make_counted.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
#include "caf/tag/io_event_oriented.hpp"
namespace caf::net {
/// Manages the lifetime of a single socket and handles any I/O events on it.
class CAF_NET_EXPORT socket_manager : public ref_counted {
public:
// -- member types -----------------------------------------------------------
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators --------------------
/// @pre `handle != invalid_socket`
......@@ -47,9 +54,8 @@ public:
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
template <class Socket = socket>
Socket handle() const {
return socket_cast<Socket>(handle_);
socket handle() const {
return handle_;
}
/// Returns the owning @ref multiplexer instance.
......@@ -62,6 +68,16 @@ public:
return *parent_;
}
/// Returns a pointer to the owning @ref multiplexer instance.
multiplexer* mpx_ptr() noexcept {
return parent_;
}
/// Returns a pointer to the owning @ref multiplexer instance.
const multiplexer* mpx_ptr() const noexcept {
return parent_;
}
/// Returns registered operations (read, write, or both).
operation mask() const noexcept {
return mask_;
......@@ -94,7 +110,20 @@ public:
// -- factories --------------------------------------------------------------
actor_shell_ptr make_actor_shell();
template <class LowerLayerPtr, class FallbackHandler>
actor_shell_ptr make_actor_shell(LowerLayerPtr, FallbackHandler f) {
auto ptr = make_actor_shell_impl();
ptr->set_fallback(std::move(f));
return ptr;
}
template <class LowerLayerPtr>
actor_shell_ptr make_actor_shell(LowerLayerPtr down) {
return make_actor_shell(down, [down](message& msg) -> result<message> {
down->abort_reason(make_error(sec::unexpected_message, std::move(msg)));
return make_error(sec::unexpected_message);
});
}
// -- event loop management --------------------------------------------------
......@@ -126,24 +155,46 @@ protected:
multiplexer* parent_;
error abort_reason_;
private:
actor_shell_ptr make_actor_shell_impl();
};
template <class Protocol>
class socket_manager_impl : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using output_tag = tag::io_event_oriented;
using socket_type = typename Protocol::socket_type;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
socket_manager_impl(typename Protocol::socket_type handle, multiplexer* mpx,
Ts&&... xs)
: socket_manager{handle, mpx}, protocol_(std::forward<Ts>(xs)...) {
socket_manager_impl(socket_type handle, multiplexer* mpx, Ts&&... xs)
: socket_manager(handle, mpx), protocol_(std::forward<Ts>(xs)...) {
// nop
}
// -- initialization ---------------------------------------------------------
error init(const settings& config) override {
CAF_LOG_TRACE("");
if (auto err = nonblocking(handle(), true)) {
CAF_LOG_ERROR("failed to set nonblocking flag in socket:" << err);
return err;
}
return protocol_.init(static_cast<socket_manager*>(this), this, config);
}
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
socket_type handle() const {
return socket_cast<socket_type>(this->handle_);
}
// -- event callbacks --------------------------------------------------------
bool handle_read_event() override {
......@@ -202,26 +253,6 @@ private:
/// @relates socket_manager
using socket_manager_ptr = intrusive_ptr<socket_manager>;
template <class B, template <class> class... Layers>
struct socket_type_helper;
template <class B>
struct socket_type_helper<B> {
using type = typename B::socket_type;
};
template <class B, template <class> class Layer,
template <class> class... Layers>
struct socket_type_helper<B, Layer, Layers...>
: socket_type_helper<Layer<B>, Layers...> {
using upper_input = typename B::input_tag;
using lower_output = typename Layer<B>::output_tag;
static_assert(std::is_same<upper_input, lower_output>::value);
};
template <class B, template <class> class... Layers>
using socket_type_t = typename socket_type_helper<B, Layers...>::type;
template <class B, template <class> class... Layers>
struct make_socket_manager_helper;
......@@ -234,19 +265,26 @@ template <class B, template <class> class Layer,
template <class> class... Layers>
struct make_socket_manager_helper<B, Layer, Layers...>
: make_socket_manager_helper<Layer<B>, Layers...> {
// no content
using upper_input = typename B::input_tag;
using lower_output = typename Layer<B>::output_tag;
static_assert(std::is_same<upper_input, lower_output>::value);
};
template <class B, template <class> class... Layers>
using make_socket_manager_helper_t =
typename make_socket_manager_helper<B, Layers...>::type;
template <class B, template <class> class... Layers>
using socket_type_t =
typename make_socket_manager_helper_t<B, Layers...,
socket_manager_impl>::socket_type;
template <class App, template <class> class... Layers, class... Ts>
auto make_socket_manager(socket_type_t<App, Layers...> handle, multiplexer* mpx,
Ts&&... xs) {
static_assert(std::is_base_of<socket, socket_type_t<App, Layers...>>::value);
using impl
= make_socket_manager_helper_t<App, Layers..., socket_manager_impl>;
static_assert(std::is_base_of<socket, typename impl::socket_type>::value);
return make_counted<impl>(std::move(handle), mpx, std::forward<Ts>(xs)...);
}
......
......@@ -41,6 +41,10 @@ public:
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_output() {
lptr_->begin_output(llptr_);
}
......
......@@ -31,6 +31,7 @@
#include "caf/sec.hpp"
#include "caf/settings.hpp"
#include "caf/span.hpp"
#include "caf/tag/io_event_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace caf::net {
......@@ -41,6 +42,8 @@ class stream_transport {
public:
// -- member types -----------------------------------------------------------
using input_tag = tag::io_event_oriented;
using output_tag = tag::stream_oriented;
using socket_type = stream_socket;
......@@ -48,7 +51,8 @@ public:
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
stream_transport(Ts&&... xs) : upper_layer_(std::forward<Ts>(xs)...) {
explicit stream_transport(Ts&&... xs)
: upper_layer_(std::forward<Ts>(xs)...) {
// nop
}
......@@ -63,6 +67,11 @@ public:
return write_buf_.size() < max_write_buf_size_;
}
template <class ParentPtr>
static socket_type handle(ParentPtr parent) noexcept {
return parent->handle();
}
template <class ParentPtr>
void begin_output(ParentPtr parent) {
if (write_buf_.empty())
......@@ -75,17 +84,17 @@ public:
}
template <class ParentPtr>
constexpr void end_output(ParentPtr) {
static constexpr void end_output(ParentPtr) {
// nop
}
template <class ParentPtr>
void abort_reason(ParentPtr parent, error reason) {
static void abort_reason(ParentPtr parent, error reason) {
return parent->abort_reason(std::move(reason));
}
template <class ParentPtr>
const error& abort_reason(ParentPtr parent) {
static const error& abort_reason(ParentPtr parent) {
return parent->abort_reason();
}
......@@ -95,7 +104,6 @@ public:
parent->register_reading();
min_read_size_ = policy.min_size;
max_read_size_ = policy.max_size;
read_buf_.resize(policy.max_size);
}
// -- properties -------------------------------------------------------------
......@@ -132,15 +140,14 @@ public:
auto default_max_reads = static_cast<uint32_t>(mm::max_consecutive_reads);
max_consecutive_reads_ = get_or(
config, "caf.middleman.max-consecutive-reads", default_max_reads);
// TODO: Tests fail, since nodelay can not be set on unix domain sockets.
// Maybe we can check for test runs using `if constexpr`?
/* if (auto err = nodelay(socket_cast<stream_socket>(parent.handle()),
true)) {
CAF_LOG_ERROR("nodelay failed: " << err);
return err;
}*/
if (auto socket_buf_size
= send_buffer_size(parent->template handle<network_socket>())) {
auto sock = parent->handle();
if constexpr (std::is_base_of<tcp_stream_socket, decltype(sock)>::value) {
if (auto err = nodelay(sock, true)) {
CAF_LOG_ERROR("nodelay failed: " << err);
return err;
}
}
if (auto socket_buf_size = send_buffer_size(parent->handle())) {
max_write_buf_size_ = *socket_buf_size;
CAF_ASSERT(max_write_buf_size_ > 0);
write_buf_.reserve(max_write_buf_size_ * 2);
......@@ -164,35 +171,72 @@ public:
upper_layer_.abort(this_layer_ptr, parent->abort_reason());
return false;
};
if (read_buf_.size() < max_read_size_)
read_buf_.resize(max_read_size_);
auto this_layer_ptr = make_stream_oriented_layer_ptr(this, parent);
for (size_t i = 0; max_read_size_ > 0 && i < max_consecutive_reads_; ++i) {
CAF_ASSERT(max_read_size_ > read_size_);
auto buf = read_buf_.data() + read_size_;
size_t len = max_read_size_ - read_size_;
CAF_LOG_DEBUG(CAF_ARG2("missing", len));
auto num_bytes = read(parent->template handle<socket_type>(),
make_span(buf, len));
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG2("handle", parent->handle().id)
<< CAF_ARG(num_bytes));
// Calling configure_read(read_policy::stop()) halts receive events.
if (max_read_size_ == 0) {
return false;
} else if (offset_ >= max_read_size_) {
auto old_max = max_read_size_;
// This may happen if the upper layer changes it receive policy to a
// smaller max. size than what was already available. In this case, the
// upper layer must consume bytes before we can receive new data.
auto bytes = make_span(read_buf_.data(), max_read_size_);
ptrdiff_t consumed = upper_layer_.consume(this_layer_ptr, bytes, {});
CAF_LOG_DEBUG(CAF_ARG2("socket", parent->handle().id)
<< CAF_ARG(consumed));
if (consumed < 0) {
upper_layer_.abort(this_layer_ptr,
parent->abort_reason_or(caf::sec::runtime_error));
return false;
} else if (consumed == 0) {
// At the very least, the upper layer must accept more data next time.
if (old_max >= max_read_size_) {
upper_layer_.abort(this_layer_ptr, parent->abort_reason_or(
caf::sec::runtime_error,
"unable to make progress"));
return false;
}
}
// Try again.
continue;
}
auto rd_buf = make_span(read_buf_.data() + offset_,
max_read_size_ - static_cast<size_t>(offset_));
auto read_res = read(parent->handle(), rd_buf);
CAF_LOG_DEBUG(CAF_ARG2("socket", parent->handle().id) //
<< CAF_ARG(max_read_size_) << CAF_ARG(offset_)
<< CAF_ARG(read_res));
// Update state.
if (num_bytes > 0) {
read_size_ += num_bytes;
if (read_size_ < min_read_size_)
if (read_res > 0) {
offset_ += read_res;
if (offset_ < min_read_size_)
continue;
auto delta = make_span(read_buf_.data() + delta_offset_,
read_size_ - delta_offset_);
auto consumed = upper_layer_.consume(this_layer_ptr,
make_span(read_buf_), delta);
auto bytes = make_span(read_buf_.data(), offset_);
auto delta = bytes.subspan(delta_offset_);
ptrdiff_t consumed = upper_layer_.consume(this_layer_ptr, bytes, delta);
CAF_LOG_DEBUG(CAF_ARG2("socket", parent->handle().id)
<< CAF_ARG(consumed));
if (consumed > 0) {
read_buf_.erase(read_buf_.begin(), read_buf_.begin() + consumed);
read_size_ -= consumed;
// Shift unconsumed bytes to the beginning of the buffer.
if (consumed < offset_)
std::copy(read_buf_.begin() + consumed, read_buf_.begin() + offset_,
read_buf_.begin());
offset_ -= consumed;
delta_offset_ = offset_;
} else if (consumed < 0) {
upper_layer_.abort(this_layer_ptr,
parent->abort_reason_or(caf::sec::runtime_error));
return false;
}
delta_offset_ = read_size_;
} else if (num_bytes < 0) {
// Our thresholds may have changed if the upper layer called
// configure_read. Shrink/grow buffer as necessary.
if (read_buf_.size() != max_read_size_)
if (offset_ < max_read_size_)
read_buf_.resize(max_read_size_);
} else if (read_res < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and
// stop reading on the socket on hard errors.
return last_socket_error_is_temporary()
......@@ -227,7 +271,7 @@ public:
}
if (write_buf_.empty())
return !upper_layer_.done_sending(this_layer_ptr);
auto written = write(parent->template handle<socket_type>(), write_buf_);
auto written = write(parent->handle(), write_buf_);
if (written > 0) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + written);
return !write_buf_.empty() || !upper_layer_.done_sending(this_layer_ptr);
......@@ -251,14 +295,31 @@ public:
}
private:
// Caches the config parameter for limiting max. socket operations.
uint32_t max_consecutive_reads_ = 0;
// Caches the write buffer size of the socket.
uint32_t max_write_buf_size_ = 0;
// Stores what the user has configured as read threshold.
uint32_t min_read_size_ = 0;
// Stores what the user has configured as max. number of bytes to receive.
uint32_t max_read_size_ = 0;
ptrdiff_t read_size_ = 0;
// Stores the current offset in `read_buf_`.
ptrdiff_t offset_ = 0;
// Stores the offset in `read_buf_` since last calling `upper_layer_.consume`.
ptrdiff_t delta_offset_ = 0;
// Caches incoming data.
byte_buffer read_buf_;
// Caches outgoing data.
byte_buffer write_buf_;
// Processes incoming data and generates outgoing data.
UpperLayer upper_layer_;
};
......
......@@ -20,6 +20,7 @@
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/uri.hpp"
......@@ -31,6 +32,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
using super = network_socket;
using super::super;
using connected_socket_type = tcp_stream_socket;
};
/// Creates a new TCP socket to accept connections on a given port.
......
This diff is collapsed.
......@@ -23,27 +23,30 @@
#include "caf/detail/encode_base64.hpp"
#include "caf/error.hpp"
#include "caf/hash/sha1.hpp"
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/web_socket_framing.hpp"
#include "caf/pec.hpp"
#include "caf/settings.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace caf::net {
/// Implements the WebSocket Protocol as defined in RFC 6455. Initially, the
/// layer performs the WebSocket handshake. Once completed, this layer becomes
/// fully transparent and forwards any data to the `UpperLayer`.
/// layer performs the WebSocket handshake. Once completed, this layer decodes
/// RFC 6455 frames and forwards binary and text messages to `UpperLayer`.
template <class UpperLayer>
class web_socket {
class web_socket_server {
public:
// -- member types -----------------------------------------------------------
using byte_span = span<const byte>;
using byte_span = span<byte>;
using input_tag = tag::stream_oriented;
using output_tag = tag::stream_oriented;
using output_tag = tag::mixed_message_oriented;
// -- constants --------------------------------------------------------------
......@@ -66,33 +69,31 @@ public:
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
web_socket(Ts&&... xs) : upper_layer_(std::forward<Ts>(xs)...) {
// nop
explicit web_socket_server(Ts&&... xs)
: upper_layer_(std::forward<Ts>(xs)...) {
// > A server MUST NOT mask any frames that it sends to the client.
// See RFC 6455, Section 5.1.
upper_layer_.mask_outgoing_frames = false;
}
virtual ~web_socket() {
// nop
// -- initialization ---------------------------------------------------------
template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
owner_ = owner;
cfg_ = cfg;
down->configure_read(net::receive_policy::up_to(max_header_size));
return none;
}
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_;
return upper_layer_.upper_layer();
}
const auto& upper_layer() const noexcept {
return upper_layer_;
}
// -- initialization ---------------------------------------------------------
template <class LowerLayerPtr>
error
init(socket_manager* owner, LowerLayerPtr down, const settings& config) {
owner_ = owner;
cfg_ = config;
down->configure_read(net::receive_policy::up_to(max_header_size));
return none;
return upper_layer_.upper_layer();
}
// -- role: upper layer ------------------------------------------------------
......@@ -115,6 +116,9 @@ public:
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buffer, byte_span delta) {
CAF_LOG_TRACE(CAF_ARG2("socket", down->handle().id)
<< CAF_ARG2("bytes", buffer.size())
<< CAF_ARG2("delta", delta.size()));
if (handshake_complete_)
return upper_layer_.consume(down, buffer, delta);
// TODO: we could avoid repeated scans by using the delta parameter.
......@@ -140,7 +144,10 @@ public:
return -1;
ptrdiff_t sub_result = 0;
if (offset < buffer.size()) {
sub_result = upper_layer_.consume(down, buffer.subspan(offset), {});
auto remainder = buffer.subspan(offset);
CAF_LOG_TRACE(CAF_ARG2("socket", down->handle().id)
<< CAF_ARG2("remainder", remainder.size()));
sub_result = upper_layer_.consume(down, remainder, remainder);
if (sub_result < 0)
return sub_result;
}
......@@ -192,19 +199,17 @@ private:
// Check whether the mandatory fields exist.
std::string sec_key;
if (auto skey_field = get_if<std::string>(&fields, "Sec-WebSocket-Key")) {
auto field_hash = hash::sha1::compute(*skey_field);
auto magic = *skey_field + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
auto field_hash = hash::sha1::compute(magic);
sec_key = detail::encode_base64(field_hash);
CAF_LOG_DEBUG("received Sec-WebSocket-Key" << *skey_field
<< "respond with" << sec_key);
} else {
auto err = make_error(pec::missing_field,
"Mandatory field Sec-WebSocket-Key not found");
down->abort_reason(std::move(err));
return false;
}
// Try initializing the upper layer.
if (auto err = upper_layer_.init(owner_, down, cfg_)) {
down->abort_reason(std::move(err));
return false;
}
// Send server handshake.
down->begin_output();
auto& buf = down->output_buffer();
......@@ -219,7 +224,13 @@ private:
append(sec_key);
append("\r\n\r\n");
down->end_output();
// Try initializing the upper layer.
if (auto err = upper_layer_.init(owner_, down, cfg_)) {
down->abort_reason(std::move(err));
return false;
}
// Done.
CAF_LOG_DEBUG("handshake completed");
handshake_complete_ = true;
return true;
}
......@@ -266,7 +277,7 @@ private:
bool handshake_complete_ = false;
/// Stores the upper layer.
UpperLayer upper_layer_;
web_socket_framing<UpperLayer> upper_layer_;
/// Stores a pointer to the owning manager for the delayed initialization.
socket_manager* owner_ = nullptr;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
namespace caf::tag {
struct io_event_oriented {};
} // namespace caf::tag
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
namespace caf::tag {
class mixed_message_oriented {
public:
mixed_message_oriented();
~mixed_message_oriented();
};
} // namespace caf::tag
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/rfc6455.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
#include <cstring>
namespace caf::detail {
void rfc6455::mask_data(uint32_t key, span<char> data) {
mask_data(key, as_writable_bytes(data));
}
void rfc6455::mask_data(uint32_t key, span<byte> data) {
auto no_key = to_network_order(key);
byte arr[4];
memcpy(arr, &no_key, 4);
size_t i = 0;
for (auto& x : data) {
x = x ^ arr[i];
i = (i + 1) % 4;
}
}
void rfc6455::assemble_frame(uint32_t mask_key, span<const char> data,
binary_buffer& out) {
assemble_frame(text_frame, mask_key, as_bytes(data), out);
}
void rfc6455::assemble_frame(uint32_t mask_key, span<const byte> data,
binary_buffer& out) {
assemble_frame(binary_frame, mask_key, data, out);
}
void rfc6455::assemble_frame(uint8_t opcode, uint32_t mask_key,
span<const byte> data, binary_buffer& out) {
// First 8 bits: FIN flag + opcode (we never fragment frames).
out.push_back(byte{static_cast<uint8_t>(0x80 | opcode)});
// Mask flag + payload length (7 bits, 7+16 bits, or 7+64 bits)
auto mask_bit = byte{static_cast<uint8_t>(mask_key == 0 ? 0x00 : 0x80)};
if (data.size() < 126) {
auto len = static_cast<uint8_t>(data.size());
out.push_back(mask_bit | byte{len});
} else if (data.size() < std::numeric_limits<uint16_t>::max()) {
auto len = static_cast<uint16_t>(data.size());
auto no_len = to_network_order(len);
byte len_data[2];
memcpy(len_data, &no_len, 2);
out.push_back(mask_bit | byte{126});
for (auto x : len_data)
out.push_back(x);
} else {
auto len = static_cast<uint64_t>(data.size());
auto no_len = to_network_order(len);
byte len_data[8];
memcpy(len_data, &no_len, 8);
out.push_back(mask_bit | byte{127});
out.insert(out.end(), len_data, len_data + 8);
}
// Masking key: 0 or 4 bytes.
if (mask_key != 0) {
auto no_key = to_network_order(mask_key);
byte key_data[4];
memcpy(key_data, &no_key, 4);
out.insert(out.end(), key_data, key_data + 4);
}
// Application data.
out.insert(out.end(), data.begin(), data.end());
}
ptrdiff_t rfc6455::decode_header(span<const byte> data, header& hdr) {
if (data.size() < 2)
return 0;
auto byte1 = to_integer<uint8_t>(data[0]);
auto byte2 = to_integer<uint8_t>(data[1]);
// Fetch FIN flag and opcode.
hdr.fin = (byte1 & 0x80) != 0;
hdr.opcode = byte1 & 0x0F;
// Decode mask bit and payload length field.
bool masked = (byte2 & 0x80) != 0;
auto len_field = byte2 & 0x7F;
size_t header_length;
if (len_field < 126) {
header_length = 2 + (masked ? 4 : 0);
hdr.payload_len = len_field;
} else if (len_field == 126) {
header_length = 4 + (masked ? 4 : 0);
} else {
header_length = 10 + (masked ? 4 : 0);
}
// Make sure we can read all the data we need.
if (data.size() < header_length)
return 0;
// Start decoding remaining header bytes.
const byte* p = data.data() + 2;
// Fetch payload size
if (len_field == 126) {
uint16_t no_len;
memcpy(&no_len, p, 2);
hdr.payload_len = from_network_order(no_len);
p += 2;
} else if (len_field == 127) {
uint64_t no_len;
memcpy(&no_len, p, 8);
hdr.payload_len = from_network_order(no_len);
p += 8;
}
// Fetch mask key.
if (masked) {
uint32_t no_key;
memcpy(&no_key, p, 4);
hdr.mask_key = from_network_order(no_key);
p += 4;
} else {
hdr.mask_key = 0;
}
// No extension bits allowed.
if (byte1 & 0x70)
return -1;
// Verify opcode and return number of consumed bytes.
switch (hdr.opcode) {
case continuation_frame:
case text_frame:
case binary_frame:
case connection_close:
case ping:
case pong:
return static_cast<ptrdiff_t>(header_length);
default:
return -1;
}
}
} // namespace caf::detail
......@@ -95,7 +95,11 @@ error multiplexer::init() {
auto pipe_handles = make_pipe();
if (!pipe_handles)
return std::move(pipe_handles.error());
add(make_counted<pollset_updater>(pipe_handles->first, this));
auto updater = make_counted<pollset_updater>(pipe_handles->first, this);
settings dummy;
if (auto err = updater->init(dummy))
return err;
add(std::move(updater));
write_handle_ = pipe_handles->second;
return none;
}
......@@ -125,6 +129,7 @@ actor_system& multiplexer::system() {
// -- thread-safe signaling ----------------------------------------------------
void multiplexer::register_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
if (shutting_down_) {
// discard
......@@ -138,11 +143,12 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) {
add(mgr);
}
} else {
write_to_pipe(0, mgr);
write_to_pipe(pollset_updater::register_reading_code, mgr);
}
}
void multiplexer::register_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
if (shutting_down_) {
// discard
......@@ -156,11 +162,30 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) {
add(mgr);
}
} else {
write_to_pipe(1, mgr);
write_to_pipe(pollset_updater::register_writing_code, mgr);
}
}
void multiplexer::init(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
if (shutting_down_) {
// discard
} else {
if (auto err = mgr->init(content(system().config()))) {
CAF_LOG_ERROR("mgr->init failed: " << err);
// The socket manager should not register itself for any events if
// initialization fails. So there's probably nothing we could do
// here other than discarding the manager.
}
}
} else {
write_to_pipe(pollset_updater::init_manager_code, mgr);
}
}
void multiplexer::close_pipe() {
CAF_LOG_TRACE("");
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) {
close(write_handle_);
......@@ -171,6 +196,7 @@ void multiplexer::close_pipe() {
// -- control flow -------------------------------------------------------------
bool multiplexer::poll_once(bool blocking) {
CAF_LOG_TRACE(CAF_ARG(blocking));
if (pollset_.empty())
return false;
// We'll call poll() until poll() succeeds or fails.
......@@ -247,6 +273,7 @@ void multiplexer::run() {
}
void multiplexer::shutdown() {
CAF_LOG_TRACE("");
if (std::this_thread::get_id() == tid_) {
CAF_LOG_DEBUG("initiate shutdown");
shutting_down_ = true;
......@@ -319,10 +346,13 @@ void multiplexer::del(ptrdiff_t index) {
}
void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
CAF_ASSERT(opcode == 0 || opcode == 1 || opcode == 4);
CAF_ASSERT(mgr != nullptr || opcode == 4);
CAF_ASSERT(opcode == pollset_updater::register_reading_code
|| opcode == pollset_updater::register_writing_code
|| opcode == pollset_updater::init_manager_code
|| opcode == pollset_updater::shutdown_code);
CAF_ASSERT(mgr != nullptr || opcode == pollset_updater::shutdown_code);
pollset_updater::msg_buf buf;
if (opcode != 4)
if (opcode != pollset_updater::shutdown_code)
mgr->ref();
buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(mgr.get());
......@@ -333,7 +363,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
if (write_handle_ != invalid_socket)
res = write(write_handle_, buf);
}
if (res <= 0 && opcode != 4)
if (res <= 0 && opcode != pollset_updater::shutdown_code)
mgr->deref();
}
......
......@@ -49,11 +49,13 @@ void actor_shell::quit(error reason) {
// -- mailbox access -----------------------------------------------------------
mailbox_element_ptr actor_shell::next_message() {
mailbox_.fetch_more();
auto& q = mailbox_.queue();
if (q.total_task_size() > 0) {
q.inc_deficit(1);
return q.next();
if (!mailbox_.blocked()) {
mailbox_.fetch_more();
auto& q = mailbox_.queue();
if (q.total_task_size() > 0) {
q.inc_deficit(1);
return q.next();
}
}
return nullptr;
}
......@@ -64,8 +66,7 @@ bool actor_shell::try_block_mailbox() {
// -- message processing -------------------------------------------------------
bool actor_shell::consume_message(
behavior& bhvr, callback<result<message>(message&)>& fallback) {
bool actor_shell::consume_message() {
CAF_LOG_TRACE("");
if (auto msg = next_message()) {
current_element_ = msg.get();
......@@ -74,10 +75,10 @@ bool actor_shell::consume_message(
auto mid = msg->mid;
if (!mid.is_response()) {
detail::default_invoke_result_visitor<actor_shell> visitor{this};
if (auto result = bhvr(msg->payload)) {
if (auto result = bhvr_(msg->payload)) {
visitor(*result);
} else {
auto fallback_result = fallback(msg->payload);
auto fallback_result = (*fallback_)(msg->payload);
visit(visitor, fallback_result);
}
} else if (auto i = multiplexed_responses_.find(mid);
......
......@@ -76,7 +76,7 @@ void middleman::init(actor_system_config& cfg) {
auto this_node = make_node_id(std::move(*node_uri));
sys_.node_.swap(this_node);
} else {
CAF_RAISE_ERROR("no valid entry for caf.middleman.this-node found");
// CAF_RAISE_ERROR("no valid entry for caf.middleman.this-node found");
}
for (auto& backend : backends_)
if (auto err = backend->init()) {
......
......@@ -26,6 +26,7 @@
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp"
......@@ -81,11 +82,13 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
}
ptrdiff_t write(pipe_socket x, span<const byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size());
}
ptrdiff_t read(pipe_socket x, span<byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size());
}
......
......@@ -20,6 +20,7 @@
#include <cstring>
#include "caf/actor_system.hpp"
#include "caf/logger.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
......@@ -38,11 +39,14 @@ pollset_updater::~pollset_updater() {
}
error pollset_updater::init(const settings&) {
CAF_LOG_TRACE("");
return nonblocking(handle(), true);
}
bool pollset_updater::handle_read_event() {
CAF_LOG_TRACE("");
for (;;) {
CAF_ASSERT((buf_.size() - buf_size_) > 0);
auto num_bytes = read(handle(), make_span(buf_.data() + buf_size_,
buf_.size() - buf_size_));
if (num_bytes > 0) {
......@@ -54,13 +58,16 @@ bool pollset_updater::handle_read_event() {
memcpy(&value, buf_.data() + 1, sizeof(intptr_t));
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
switch (opcode) {
case 0:
case register_reading_code:
parent_->register_reading(mgr);
break;
case 1:
case register_writing_code:
parent_->register_writing(mgr);
break;
case 4:
case init_manager_code:
parent_->init(mgr);
break;
case shutdown_code:
parent_->shutdown();
break;
default:
......
......@@ -64,7 +64,7 @@ void socket_manager::register_writing() {
parent_->register_writing(this);
}
actor_shell_ptr socket_manager::make_actor_shell() {
actor_shell_ptr socket_manager::make_actor_shell_impl() {
CAF_ASSERT(parent_ != nullptr);
auto hdl = parent_->system().spawn<actor_shell>(this);
return actor_shell_ptr{actor_cast<strong_actor_ptr>(std::move(hdl))};
......
......@@ -169,11 +169,13 @@ error nodelay(stream_socket x, bool new_value) {
}
ptrdiff_t read(stream_socket x, span<byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag);
}
ptrdiff_t write(stream_socket x, span<const byte> buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag);
}
......
......@@ -55,6 +55,9 @@ template <int Family>
expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
const char* addr,
bool reuse_addr, bool any) {
auto bts = [](bool x) { return x ? "true" : "false"; };
printf("%s(%d, %s, %s, %s)\n", __func__, (int) port, addr, bts(reuse_addr),
bts(any));
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
int socktype = SOCK_STREAM;
......@@ -98,21 +101,23 @@ expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(node));
auto addr = to_string(node.address());
auto make_acceptor = node.address().embeds_v4()
? new_tcp_acceptor_impl<AF_INET>
: new_tcp_acceptor_impl<AF_INET6>;
auto p = make_acceptor(node.port(), addr.c_str(), reuse_addr,
node.address().zero());
if (!p) {
CAF_LOG_WARNING("could not create tcp socket for: " << to_string(node));
bool is_v4 = node.address().embeds_v4();
bool is_zero = is_v4 ? node.address().embedded_v4().bits() == 0
: node.address().zero();
auto make_acceptor = is_v4 ? new_tcp_acceptor_impl<AF_INET>
: new_tcp_acceptor_impl<AF_INET6>;
if (auto p = make_acceptor(node.port(), addr.c_str(), reuse_addr, is_zero)) {
auto sock = socket_cast<tcp_accept_socket>(*p);
auto sguard = make_socket_guard(sock);
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(sock.id, SOMAXCONN));
CAF_LOG_DEBUG(CAF_ARG(sock.id));
return sguard.release();
} else {
CAF_LOG_WARNING("could not create tcp socket:" << CAF_ARG(node)
<< CAF_ARG(p.error()));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
to_string(node));
to_string(node), std::move(p.error()));
}
auto sock = socket_cast<tcp_accept_socket>(*p);
auto sguard = make_socket_guard(sock);
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(sock.id, SOMAXCONN));
CAF_LOG_DEBUG(CAF_ARG(sock.id));
return sguard.release();
}
expected<tcp_accept_socket>
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE detail.rfc6455
#include "caf/detail/rfc6455.hpp"
#include "caf/test/dsl.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
#include "caf/string_view.hpp"
#include <cstdint>
#include <initializer_list>
#include <vector>
using namespace caf;
namespace {
struct fixture {
using impl = detail::rfc6455;
auto bytes(std::initializer_list<uint8_t> xs) {
std::vector<byte> result;
for (auto x : xs)
result.emplace_back(static_cast<byte>(x));
return result;
}
template <class T>
auto take(const T& xs, size_t num_bytes) {
auto n = std::min(xs.size(), num_bytes);
return std::vector<typename T::value_type>{xs.begin(), xs.begin() + n};
}
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(rfc6455_tests, fixture)
CAF_TEST(masking) {
auto key = uint32_t{0xDEADC0DE};
auto data = bytes({0x12, 0x34, 0x45, 0x67, 0x89, 0x9A});
auto masked_data = data;
CAF_MESSAGE("masking XORs the repeated key to data");
impl::mask_data(key, masked_data);
CAF_CHECK_EQUAL(masked_data, bytes({
0x12 ^ 0xDE,
0x34 ^ 0xAD,
0x45 ^ 0xC0,
0x67 ^ 0xDE,
0x89 ^ 0xDE,
0x9A ^ 0xAD,
}));
CAF_MESSAGE("masking masked data again gives the original data");
impl::mask_data(key, masked_data);
CAF_CHECK_EQUAL(masked_data, data);
}
CAF_TEST(no mask key plus small data) {
std::vector<uint8_t> data{0x12, 0x34, 0x45, 0x67};
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(out, bytes({
0x82, // FIN + binary frame opcode
0x04, // data size = 4
0x12, 0x34, 0x45, 0x67, // masked data
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 2);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus small data) {
std::vector<uint8_t> data{0x12, 0x34, 0x45, 0x67};
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(out, bytes({
0x82, // FIN + binary frame opcode
0x84, // MASKED + data size = 4
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0x12, 0x34, 0x45, 0x67, // masked data
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 6);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(no mask key plus medium data) {
std::vector<uint8_t> data;
data.insert(data.end(), 126, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 8),
bytes({
0x82, // FIN + binary frame opcode
0x7E, // 126 -> uint16 size
0x00, 0x7E, // data size = 126
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 4);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus medium data) {
std::vector<uint8_t> data;
data.insert(data.end(), 126, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 12),
bytes({
0x82, // FIN + binary frame opcode
0xFE, // MASKED + 126 -> uint16 size
0x00, 0x7E, // data size = 126
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 8);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(no mask key plus large data) {
std::vector<uint8_t> data;
data.insert(data.end(), 65536, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0, as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 14),
bytes({
0x82, // FIN + binary frame opcode
0x7F, // 127 -> uint64 size
0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, // 65536
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 10);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0u);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST(valid mask key plus large data) {
std::vector<uint8_t> data;
data.insert(data.end(), 65536, 0xFF);
std::vector<byte> out;
impl::assemble_frame(impl::binary_frame, 0xDEADC0DE,
as_bytes(make_span(data)), out);
CAF_CHECK_EQUAL(take(out, 18),
bytes({
0x82, // FIN + binary frame opcode
0xFF, // MASKED + 127 -> uint64 size
0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, // 65536
0xDE, 0xAD, 0xC0, 0xDE, // mask key,
0xFF, 0xFF, 0xFF, 0xFF, // first 4 masked bytes
}));
impl::header hdr;
CAF_CHECK_EQUAL(impl::decode_header(out, hdr), 14);
CAF_CHECK_EQUAL(hdr.fin, true);
CAF_CHECK_EQUAL(hdr.mask_key, 0xDEADC0DEu);
CAF_CHECK_EQUAL(hdr.opcode, impl::binary_frame);
CAF_CHECK_EQUAL(hdr.payload_len, data.size());
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -2,6 +2,7 @@
#include "caf/error.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/settings.hpp"
#include "caf/span.hpp"
......@@ -30,6 +31,10 @@ public:
// nop
}
constexpr caf::net::socket handle() noexcept {
return caf::net::invalid_socket;
}
bool can_send_more() const noexcept {
return true;
}
......
......@@ -34,7 +34,7 @@ using namespace std::string_literals;
namespace {
using byte_span = span<const byte>;
using byte_span = span<byte>;
using svec = std::vector<std::string>;
......@@ -49,14 +49,12 @@ struct app_t {
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr down, const settings&) {
self = mgr->make_actor_shell();
bhvr = behavior{
[this](std::string& line) {
CAF_MESSAGE("received an asynchronous message: " << line);
lines.emplace_back(std::move(line));
},
};
fallback = make_type_erased_callback([](message& msg) -> result<message> {
self = mgr->make_actor_shell(down);
self->set_behavior([this](std::string& line) {
CAF_MESSAGE("received an asynchronous message: " << line);
lines.emplace_back(std::move(line));
});
self->set_fallback([](message& msg) -> result<message> {
CAF_FAIL("unexpected message: " << msg);
return make_error(sec::unexpected_message);
});
......@@ -66,7 +64,7 @@ struct app_t {
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
while (self->consume_message(bhvr, *fallback)) {
while (self->consume_message()) {
// We set abort_reason in our response handlers in case of an error.
if (down->abort_reason())
return false;
......@@ -154,12 +152,6 @@ struct app_t {
// Actor shell representing this app.
net::actor_shell_ptr self;
// Handler for consuming messages from the mailbox.
behavior bhvr;
// Handler for unexpected messages.
unique_callback_ptr<result<message>(message&)> fallback;
// Counts how many bytes we've consumed in total.
size_t consumed_bytes = 0;
......
......@@ -40,7 +40,7 @@ namespace {
/// upper layer: expect messages
/// Needs to be initilized by the layer two steps down.
struct ul_expect_messages {
using byte_span = span<const byte>;
using byte_span = span<byte>;
using input_tag = tag::message_oriented;
......
......@@ -16,9 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE net.web_socket
#define CAF_SUITE net.web_socket_server
#include "caf/net/web_socket.hpp"
#include "caf/net/web_socket_server.hpp"
#include "net-test.hpp"
......@@ -28,16 +28,19 @@
#include "caf/net/stream_transport.hpp"
using namespace caf;
using namespace caf::literals;
using namespace std::literals::string_literals;
namespace {
using byte_span = span<const byte>;
using byte_span = span<byte>;
using svec = std::vector<std::string>;
struct app_t {
std::vector<std::string> lines;
std::string text_input;
std::vector<byte> binary_input;
settings cfg;
......@@ -63,20 +66,15 @@ struct app_t {
}
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buffer, byte_span) {
constexpr auto nl = byte{'\n'};
auto e = buffer.end();
if (auto i = std::find(buffer.begin(), e, nl); i != e) {
std::string str;
auto string_size = static_cast<size_t>(std::distance(buffer.begin(), i));
str.reserve(string_size);
auto num_bytes = string_size + 1; // Also consume the newline character.
std::transform(buffer.begin(), i, std::back_inserter(str),
[](byte x) { return static_cast<char>(x); });
lines.emplace_back(std::move(str));
return num_bytes + consume(down, buffer.subspan(num_bytes), {});
}
return 0;
ptrdiff_t consume_text(LowerLayerPtr, string_view text) {
text_input.insert(text_input.end(), text.begin(), text.end());
return static_cast<ptrdiff_t>(text.size());
}
template <class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr, byte_span bytes) {
binary_input.insert(binary_input.end(), bytes.begin(), bytes.end());
return static_cast<ptrdiff_t>(bytes.size());
}
};
......@@ -87,16 +85,49 @@ struct fixture : host_fixture {
app = std::addressof(ws->upper_layer());
if (auto err = transport.init())
CAF_FAIL("failed to initialize mock transport: " << err);
rng.seed(0xD3ADC0D3);
}
void rfc6455_append(uint8_t opcode, span<const byte> bytes,
std::vector<byte>& out) {
std::vector<byte> payload{bytes.begin(), bytes.end()};
auto key = static_cast<uint32_t>(rng());
detail::rfc6455::mask_data(key, payload);
detail::rfc6455::assemble_frame(opcode, key, payload, out);
}
mock_stream_transport<net::web_socket<app_t>> transport;
void rfc6455_append(span<const byte> bytes, std::vector<byte>& out) {
rfc6455_append(detail::rfc6455::binary_frame, bytes, out);
}
net::web_socket<app_t>* ws;
void rfc6455_append(string_view text, std::vector<byte>& out) {
rfc6455_append(detail::rfc6455::text_frame, as_bytes(make_span(text)), out);
}
void push(uint8_t opcode, span<const byte> bytes) {
std::vector<byte> frame;
rfc6455_append(opcode, bytes, frame);
transport.push(frame);
}
void push(span<const byte> bytes) {
push(detail::rfc6455::binary_frame, bytes);
}
void push(string_view str) {
push(detail::rfc6455::text_frame, as_bytes(make_span(str)));
}
mock_stream_transport<net::web_socket_server<app_t>> transport;
net::web_socket_server<app_t>* ws;
app_t* app;
std::minstd_rand rng;
};
constexpr string_view opening_handshake
constexpr auto opening_handshake
= "GET /chat HTTP/1.1\r\n"
"Host: server.example.com\r\n"
"Upgrade: websocket\r\n"
......@@ -105,7 +136,7 @@ constexpr string_view opening_handshake
"Origin: http://example.com\r\n"
"Sec-WebSocket-Protocol: chat, superchat\r\n"
"Sec-WebSocket-Version: 13\r\n"
"\r\n";
"\r\n"_sv;
} // namespace
......@@ -172,13 +203,14 @@ CAF_TEST(handshakes may arrive in chunks) {
}
CAF_TEST(data may follow the handshake immediately) {
std::string buf{opening_handshake.begin(), opening_handshake.end()};
buf += "Hello WebSocket!\n";
buf += "Bye WebSocket!\n";
auto hs_bytes = as_bytes(make_span(opening_handshake));
std::vector<byte> buf{hs_bytes.begin(), hs_bytes.end()};
rfc6455_append("Hello WebSocket!\n"_sv, buf);
rfc6455_append("Bye WebSocket!\n"_sv, buf);
transport.push(buf);
CAF_CHECK_EQUAL(transport.handle_input(), static_cast<ptrdiff_t>(buf.size()));
CAF_CHECK(ws->handshake_complete());
CAF_CHECK_EQUAL(app->lines, svec({"Hello WebSocket!", "Bye WebSocket!"}));
CAF_CHECK_EQUAL(app->text_input, "Hello WebSocket!\nBye WebSocket!\n");
}
CAF_TEST(data may arrive later) {
......@@ -186,10 +218,9 @@ CAF_TEST(data may arrive later) {
CAF_CHECK_EQUAL(transport.handle_input(),
static_cast<ptrdiff_t>(opening_handshake.size()));
CAF_CHECK(ws->handshake_complete());
auto buf = "Hello WebSocket!\nBye WebSocket!\n"s;
transport.push(buf);
CAF_CHECK_EQUAL(transport.handle_input(), static_cast<ptrdiff_t>(buf.size()));
CAF_CHECK_EQUAL(app->lines, svec({"Hello WebSocket!", "Bye WebSocket!"}));
push("Hello WebSocket!\nBye WebSocket!\n"_sv);
transport.handle_input();
CAF_CHECK_EQUAL(app->text_input, "Hello WebSocket!\nBye WebSocket!\n");
}
CAF_TEST_FIXTURE_SCOPE_END()
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