Commit ad965456 authored by Dominik Charousset's avatar Dominik Charousset

Merge caf_net from the incubator

parents cfeb9105 c29a4903
# -- get header files for creating "proper" XCode projects ---------------------
file(GLOB_RECURSE CAF_NET_HEADERS "caf/*.hpp")
# -- add targets ---------------------------------------------------------------
caf_incubator_add_component(
net
DEPENDENCIES
PUBLIC
CAF::core
$<$<CXX_COMPILER_ID:MSVC>:ws2_32>
PRIVATE
CAF::internal
ENUM_TYPES
net.basp.connection_state
net.basp.ec
net.basp.message_type
net.http.method
net.http.status
net.operation
net.stream_transport_error
net.web_socket.status
HEADERS
${CAF_NET_HEADERS}
SOURCES
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp
src/detail/rfc6455.cpp
src/header.cpp
src/host.cpp
src/ip.cpp
src/message_queue.cpp
src/multiplexer.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/http/header.cpp
src/net/http/method.cpp
src/net/http/status.cpp
src/net/http/v1.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
src/net/packet_writer.cpp
src/net/web_socket/handshake.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
src/worker.cpp
TEST_SOURCES
test/net-test.cpp
TEST_SUITES
accept_socket
convert_ip_endpoint
datagram_socket
detail.rfc6455
header
ip
multiplexer
net.actor_shell
net.consumer_adapter
net.http.server
net.length_prefix_framing
net.operation
net.producer_adapter
net.typed_actor_shell
net.web_socket.client
net.web_socket.handshake
net.web_socket.server
network_socket
pipe_socket
socket
socket_guard
stream_socket
stream_transport
tcp_sockets
udp_datagram_socket)
if(CAF_INC_ENABLE_TESTING AND TARGET OpenSSL::SSL AND TARGET OpenSSL::Crypto)
caf_incubator_add_test_suites(caf-net-test net.openssl_transport)
target_sources(caf-net-test PRIVATE test/net/openssl_transport_constants.cpp)
target_link_libraries(caf-net-test PRIVATE OpenSSL::SSL OpenSSL::Crypto)
endif()
This diff is collapsed.
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/fwd.hpp"
namespace caf::detail {
void convert(const ip_endpoint& src, sockaddr_storage& dst);
error convert(const sockaddr_storage& src, ip_endpoint& dst);
} // namespace caf::detail
// 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::detail {
template <class T, class LowerLayerPtr>
class has_after_reading {
private:
template <class A, class B>
static auto sfinae(A& up, B& ptr)
-> decltype(up.after_reading(ptr), std::true_type{});
template <class A>
static std::false_type sfinae(A&, ...);
using sfinae_result
= decltype(sfinae(std::declval<T&>(), std::declval<LowerLayerPtr&>()));
public:
static constexpr bool value = sfinae_result::value;
};
template <class T, class LowerLayerPtr>
constexpr bool has_after_reading_v = has_after_reading<T, LowerLayerPtr>::value;
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::detail {
template <class T>
struct actor_shell_ptr_type_oracle;
template <>
struct actor_shell_ptr_type_oracle<actor> {
using type = net::actor_shell_ptr;
};
template <class... Sigs>
struct actor_shell_ptr_type_oracle<typed_actor<Sigs...>> {
using type = net::typed_actor_shell_ptr<Sigs...>;
};
template <class T>
using infer_actor_shell_ptr_type =
typename actor_shell_ptr_type_oracle<T>::type;
} // namespace caf::detail
// 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 <cstdio>
#include <cstdlib>
#include "caf/error.hpp"
#include "caf/sec.hpp"
/// Calls a C functions and returns an error if `var op rhs` returns `true`.
#define CAF_NET_SYSCALL(funname, var, op, rhs, expr) \
auto var = expr; \
if (var op rhs) \
return make_error(sec::network_syscall_failed, funname, \
last_socket_error_as_string())
/// Calls a C functions and calls exit() if `var op rhs` returns `true`.
#define CAF_NET_CRITICAL_SYSCALL(funname, var, op, rhs, expr) \
auto var = expr; \
if (var op rhs) { \
fprintf(stderr, "[FATAL] %s:%u: syscall failed: %s returned %s\n", \
__FILE__, __LINE__, funname, \
last_socket_error_as_string().c_str()); \
abort(); \
} \
static_cast<void>(0)
// 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.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.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
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/socket_sys_includes.hpp"
namespace caf::detail {
inline auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
inline auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
inline auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
inline auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
inline auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
inline auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
} // namespace caf::detail
// 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/config.hpp"
namespace caf::net {
#ifdef CAF_WINDOWS
using setsockopt_ptr = const char*;
using getsockopt_ptr = char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
using socket_size_type = int;
#else // CAF_WINDOWS
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
using socket_size_type = unsigned;
#endif // CAF_WINDOWS
} // namespace caf::net
// 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.
// This convenience header pulls in platform-specific headers for the C socket
// API. Do *not* include this header in other headers.
#pragma once
#include "caf/config.hpp"
// clang-format off
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif // CAF_WINDOWS
# ifndef NOMINMAX
# define NOMINMAX
# endif // NOMINMAX
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif // CAF_MINGW
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else // CAF_WINDOWS
# include <sys/types.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <unistd.h>
# include <sys/uio.h>
#endif
// clang-format on
// 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_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"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/local_actor.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
#include "caf/policy/normal_messages.hpp"
namespace caf::net {
class CAF_NET_EXPORT abstract_actor_shell : public local_actor,
public non_blocking_actor_base {
public:
// -- member types -----------------------------------------------------------
using super = local_actor;
struct mailbox_policy {
using queue_type = intrusive::drr_queue<policy::normal_messages>;
using deficit_type = policy::normal_messages::deficit_type;
using mapped_type = policy::normal_messages::mapped_type;
using unique_pointer = policy::normal_messages::unique_pointer;
};
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators --------------------
abstract_actor_shell(actor_config& cfg, socket_manager* owner);
~abstract_actor_shell() override;
// -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox.
void quit(error reason);
/// 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 {
return mailbox_;
}
/// Dequeues and returns the next message from the mailbox or returns
/// `nullptr` if the mailbox is empty.
mailbox_element_ptr next_message();
/// Tries to put the mailbox into the `blocked` state, causing the next
/// enqueue to register the owning socket manager for write events.
bool try_block_mailbox();
// -- message processing -----------------------------------------------------
/// Dequeues and processes the next message from the mailbox.
/// @returns `true` if a message was dequeued and processed, `false` if the
/// mailbox was empty.
bool consume_message();
/// Adds a callback for a multiplexed response.
void add_multiplexed_response_handler(message_id response_id, behavior bhvr);
// -- overridden functions of abstract_actor ---------------------------------
using abstract_actor::enqueue;
bool enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
mailbox_element* peek_at_next_mailbox_element() override;
// -- overridden functions of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override;
bool cleanup(error&& fail_state, execution_unit* host) override;
protected:
// Stores incoming actor messages.
mailbox_type mailbox_;
// Guards access to owner_.
std::mutex owner_mtx_;
// 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_;
};
} // namespace caf::net
// 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_proxy.hpp"
#include "caf/net/endpoint_manager.hpp"
namespace caf::net {
/// Implements a simple proxy forwarding all operations to a manager.
class actor_proxy_impl : public actor_proxy {
public:
using super = actor_proxy;
actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst);
~actor_proxy_impl() override;
bool enqueue(mailbox_element_ptr what, execution_unit* context) override;
void kill_proxy(execution_unit* ctx, error rsn) override;
private:
endpoint_manager_ptr dst_;
};
} // namespace caf::net
// 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_traits.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/net/abstract_actor_shell.hpp"
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
namespace caf::net {
/// Enables socket managers to communicate with actors using dynamically typed
/// messaging.
class CAF_NET_EXPORT actor_shell
: public extend<abstract_actor_shell, actor_shell>::with<mixin::sender,
mixin::requester>,
public dynamically_typed_actor_base {
public:
// -- friends ----------------------------------------------------------------
friend class actor_shell_ptr;
// -- member types -----------------------------------------------------------
using super = extend<abstract_actor_shell,
actor_shell>::with<mixin::sender, mixin::requester>;
using signatures = none_t;
using behavior_type = behavior;
// -- constructors, destructors, and assignment operators --------------------
actor_shell(actor_config& cfg, socket_manager* owner);
~actor_shell() override;
// -- state modifiers --------------------------------------------------------
/// Overrides the callbacks for incoming messages.
template <class... Fs>
void set_behavior(Fs... fs) {
bhvr_ = behavior{std::move(fs)...};
}
// -- overridden functions of local_actor ------------------------------------
const char* name() const override;
};
/// An "owning" pointer to an actor shell in the sense that it calls `quit()` on
/// the shell when going out of scope.
class CAF_NET_EXPORT actor_shell_ptr {
public:
// -- friends ----------------------------------------------------------------
friend class socket_manager;
// -- member types -----------------------------------------------------------
using handle_type = actor;
using element_type = actor_shell;
// -- constructors, destructors, and assignment operators --------------------
constexpr actor_shell_ptr() noexcept {
// nop
}
constexpr actor_shell_ptr(std::nullptr_t) noexcept {
// nop
}
actor_shell_ptr(actor_shell_ptr&& other) noexcept = default;
actor_shell_ptr& operator=(actor_shell_ptr&& other) noexcept = default;
actor_shell_ptr(const actor_shell_ptr& other) = delete;
actor_shell_ptr& operator=(const actor_shell_ptr& other) = delete;
~actor_shell_ptr();
// -- smart pointer interface ------------------------------------------------
/// Returns an actor handle to the managed actor shell.
handle_type as_actor() const noexcept;
void detach(error reason);
element_type* get() const noexcept;
element_type* operator->() const noexcept {
return get();
}
element_type& operator*() const noexcept {
return *get();
}
bool operator!() const noexcept {
return !ptr_;
}
explicit operator bool() const noexcept {
return static_cast<bool>(ptr_);
}
private:
/// @pre `ptr != nullptr`
explicit actor_shell_ptr(strong_actor_ptr ptr) noexcept;
strong_actor_ptr ptr_;
};
} // namespace caf::net
// 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/net/actor_proxy_impl.hpp"
#include "caf/net/datagram_socket.hpp"
#include "caf/net/datagram_transport.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/host.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/pollset_updater.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/socket_id.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
// 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 <map>
#include <mutex>
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/node_id.hpp"
namespace caf::net::backend {
/// Minimal backend for tcp communication.
class CAF_NET_EXPORT tcp : public middleman_backend {
public:
using peer_map = std::map<node_id, endpoint_manager_ptr>;
using emplace_return_type = std::pair<peer_map::iterator, bool>;
// -- constructors, destructors, and assignment operators --------------------
tcp(middleman& mm);
~tcp() override;
// -- interface functions ----------------------------------------------------
error init() override;
void stop() override;
expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
endpoint_manager_ptr peer(const node_id& id) override;
void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
void set_last_hop(node_id*) override;
// -- properties -------------------------------------------------------------
uint16_t port() const noexcept override;
template <class Handle>
expected<endpoint_manager_ptr>
emplace(const node_id& peer_id, Handle socket_handle) {
using transport_type = stream_transport<basp::application>;
if (auto err = nonblocking(socket_handle, true))
return err;
auto mpx = mm_.mpx();
basp::application app{proxies_};
auto mgr = make_endpoint_manager(
mpx, mm_.system(), transport_type{socket_handle, std::move(app)});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err);
return err;
}
mpx->register_reading(mgr);
emplace_return_type res;
{
const std::lock_guard<std::mutex> lock(lock_);
res = peers_.emplace(peer_id, std::move(mgr));
}
if (res.second)
return res.first->second;
else
return make_error(sec::runtime_error, "peer_id already exists");
}
private:
endpoint_manager_ptr get_peer(const node_id& id);
middleman& mm_;
peer_map peers_;
proxy_registry proxies_;
uint16_t listening_port_;
std::mutex lock_;
};
} // namespace caf::net::backend
// 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 <map>
#include "caf/detail/net_export.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/node_id.hpp"
namespace caf::net::backend {
/// Minimal backend for unit testing.
/// @warning this backend is *not* thread safe.
class CAF_NET_EXPORT test : public middleman_backend {
public:
// -- member types -----------------------------------------------------------
using peer_entry = std::pair<stream_socket, endpoint_manager_ptr>;
// -- constructors, destructors, and assignment operators --------------------
test(middleman& mm);
~test() override;
// -- interface functions ----------------------------------------------------
error init() override;
void stop() override;
endpoint_manager_ptr peer(const node_id& id) override;
expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
void set_last_hop(node_id*) override;
// -- properties -------------------------------------------------------------
stream_socket socket(const node_id& peer_id) {
return get_peer(peer_id).first;
}
uint16_t port() const noexcept override;
peer_entry& emplace(const node_id& peer_id, stream_socket first,
stream_socket second);
private:
peer_entry& get_peer(const node_id& id);
middleman& mm_;
std::map<node_id, peer_entry> peers_;
proxy_registry proxies_;
};
} // namespace caf::net::backend
// 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 <cstdint>
#include <memory>
#include <type_traits>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include "caf/actor_addr.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/callback.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/message_type.hpp"
#include "caf/net/basp/worker.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/response_promise.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/unit.hpp"
namespace caf::net::basp {
/// An implementation of BASP as an application layer protocol.
class CAF_NET_EXPORT application {
public:
// -- member types -----------------------------------------------------------
using hub_type = detail::worker_hub<worker>;
struct test_tag {};
// -- constructors, destructors, and assignment operators --------------------
explicit application(proxy_registry& proxies);
// -- static utility functions -----------------------------------------------
static auto default_app_ids() {
return std::vector<std::string>{
to_string(defaults::middleman::app_identifier)};
}
// -- interface functions ----------------------------------------------------
template <class Parent>
error init(Parent& parent) {
// Initialize member variables.
system_ = &parent.system();
executor_.system_ptr(system_);
executor_.proxy_registry_ptr(&proxies_);
// Allow unit tests to run the application without endpoint manager.
if constexpr (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager();
size_t workers;
if (auto workers_cfg = get_if<size_t>(&system_->config(),
"caf.middleman.workers"))
workers = *workers_cfg;
else
workers = std::min(3u, std::thread::hardware_concurrency() / 4u) + 1;
for (size_t i = 0; i < workers; ++i)
hub_->add_new_worker(*queue_, proxies_);
// Write handshake.
auto hdr = parent.next_header_buffer();
auto payload = parent.next_payload_buffer();
if (auto err = generate_handshake(payload))
return err;
to_bytes(header{message_type::handshake,
static_cast<uint32_t>(payload.size()), version},
hdr);
parent.write_packet(hdr, payload);
parent.transport().configure_read(receive_policy::exactly(header_size));
return none;
}
error write_message(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> ptr);
template <class Parent>
error handle_data(Parent& parent, byte_span bytes) {
static_assert(std::is_base_of<packet_writer, Parent>::value,
"parent must implement packet_writer");
size_t next_read_size = header_size;
if (auto err = handle(next_read_size, parent, bytes))
return err;
parent.transport().configure_read(receive_policy::exactly(next_read_size));
return none;
}
void resolve(packet_writer& writer, string_view path, const actor& listener);
static void new_proxy(packet_writer& writer, actor_id id);
void local_actor_down(packet_writer& writer, actor_id id, error reason);
template <class Parent>
void timeout(Parent&, const std::string&, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
// -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path);
// -- properties -------------------------------------------------------------
connection_state state() const noexcept {
return state_;
}
actor_system& system() const noexcept {
return *system_;
}
private:
// -- handling of incoming messages ------------------------------------------
error handle(size_t& next_read_size, packet_writer& writer, byte_span bytes);
error handle(packet_writer& writer, header hdr, byte_span payload);
error handle_handshake(packet_writer& writer, header hdr, byte_span payload);
error handle_actor_message(packet_writer& writer, header hdr,
byte_span payload);
error handle_resolve_request(packet_writer& writer, header rec_hdr,
byte_span received);
error handle_resolve_response(packet_writer& writer, header received_hdr,
byte_span received);
error handle_monitor_message(packet_writer& writer, header received_hdr,
byte_span received);
error handle_down_message(packet_writer& writer, header received_hdr,
byte_span received);
/// Writes the handshake payload to `buf_`.
error generate_handshake(byte_buffer& buf);
// -- member variables -------------------------------------------------------
/// Stores a pointer to the parent actor system.
actor_system* system_ = nullptr;
/// Stores the expected type of the next incoming message.
connection_state state_ = connection_state::await_handshake_header;
/// Caches the last header while waiting for the matching payload.
header hdr_;
/// Stores the ID of our peer.
node_id peer_id_;
/// Tracks which local actors our peer monitors.
std::unordered_set<actor_addr> monitored_actors_; // TODO: this is unused
/// Caches actor handles obtained via `resolve`.
std::unordered_map<uint64_t, actor> pending_resolves_;
/// Ascending ID generator for requests to our peer.
uint64_t next_request_id_ = 1;
/// Points to the factory object for generating proxies.
proxy_registry& proxies_;
/// Points to the endpoint manager that owns this applications.
endpoint_manager* manager_ = nullptr;
/// Provides pointers to the actor system as well as the registry,
/// serializers and deserializer.
scoped_execution_unit executor_;
std::unique_ptr<message_queue> queue_;
std::unique_ptr<hub_type> hub_;
};
} // namespace caf::net::basp
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/proxy_registry.hpp"
namespace caf::net::basp {
/// Factory for basp::application.
/// @relates doorman
class CAF_NET_EXPORT application_factory {
public:
using application_type = basp::application;
application_factory(proxy_registry& proxies) : proxies_(proxies) {
// nop
}
template <class Parent>
error init(Parent&) {
return none;
}
application_type make() const {
return application_type{proxies_};
}
private:
proxy_registry& proxies_;
};
} // namespace caf::net::basp
// 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 <cstdint>
#include <string>
#include <type_traits>
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net::basp {
/// @addtogroup BASP
/// Stores the state of a connection in a `basp::application`.
enum class connection_state : uint8_t {
/// Initial state for any connection to wait for the peer's handshake.
await_handshake_header,
/// Indicates that the header for the peer's handshake arrived and BASP
/// requires the payload next.
await_handshake_payload,
/// Indicates that a connection is established and this node is waiting for
/// the next BASP header.
await_header,
/// Indicates that this node has received a header with non-zero payload and
/// is waiting for the data.
await_payload,
/// Indicates that the connection is about to shut down.
shutdown,
};
/// @relates connection_state
CAF_NET_EXPORT std::string to_string(connection_state x);
/// @relates connection_state
CAF_NET_EXPORT bool from_string(string_view, connection_state&);
/// @relates connection_state
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<connection_state>,
connection_state&);
/// @relates connection_state
template <class Inspector>
bool inspect(Inspector& f, connection_state& x) {
return default_enum_inspect(f, x);
}
/// @}
} // namespace caf::net::basp
// 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 <cstddef>
#include <cstdint>
namespace caf::net::basp {
/// @addtogroup BASP
/// The current BASP version.
/// @note BASP is not backwards compatible.
constexpr uint64_t version = 1;
/// Size of a BASP header in serialized form.
constexpr size_t header_size = 13;
/// @}
} // namespace caf::net::basp
// 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 <cstdint>
#include <string>
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/is_error_code_enum.hpp"
namespace caf::net::basp {
/// BASP-specific error codes.
enum class ec : uint8_t {
invalid_magic_number = 1,
unexpected_number_of_bytes,
unexpected_payload,
missing_payload,
illegal_state,
invalid_handshake,
missing_handshake,
unexpected_handshake,
version_mismatch,
unimplemented = 10,
app_identifiers_mismatch,
invalid_payload,
invalid_scheme,
invalid_locator,
};
/// @relates ec
CAF_NET_EXPORT std::string to_string(ec x);
/// @relates ec
CAF_NET_EXPORT bool from_string(string_view, ec&);
/// @relates ec
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<ec>, ec&);
/// @relates ec
template <class Inspector>
bool inspect(Inspector& f, ec& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net::basp
CAF_ERROR_CODE_ENUM(caf::net::basp::ec)
// 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 <array>
#include <cstdint>
#include "caf/detail/comparable.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/message_type.hpp"
#include "caf/type_id.hpp"
namespace caf::net::basp {
/// @addtogroup BASP
/// The header of a Binary Actor System Protocol (BASP) message.
struct CAF_NET_EXPORT header : detail::comparable<header> {
// -- constructors, destructors, and assignment operators --------------------
constexpr header() noexcept
: type(message_type::handshake), payload_len(0), operation_data(0) {
// nop
}
constexpr header(message_type type, uint32_t payload_len,
uint64_t operation_data) noexcept
: type(type), payload_len(payload_len), operation_data(operation_data) {
// nop
}
header(const header&) noexcept = default;
header& operator=(const header&) noexcept = default;
// -- factory functions ------------------------------------------------------
/// @pre `bytes.size() == header_size`
static header from_bytes(span<const byte> bytes);
// -- comparison -------------------------------------------------------------
int compare(header other) const noexcept;
// -- member variables -------------------------------------------------------
/// Denotes the BASP operation and how `operation_data` gets interpreted.
message_type type;
/// Stores the size in bytes for the payload that follows this header.
uint32_t payload_len;
/// Stores type-specific information such as the BASP version in handshakes.
uint64_t operation_data;
};
/// Serializes a header to a byte representation.
/// @relates header
CAF_NET_EXPORT std::array<byte, header_size> to_bytes(header x);
/// Serializes a header to a byte representation.
/// @relates header
CAF_NET_EXPORT void to_bytes(header x, byte_buffer& buf);
/// @relates header
template <class Inspector>
bool inspect(Inspector& f, header& x) {
return f.object(x).fields(f.field("type", x.type),
f.field("payload_len", x.payload_len),
f.field("operation_data", x.operation_data));
}
/// @}
} // namespace caf::net::basp
namespace caf {
template <>
struct type_name<net::basp::header> {
static constexpr string_view value = "caf::net::basp::header";
};
} // namespace caf
// 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 <cstdint>
#include <mutex>
#include <vector>
#include "caf/actor_control_block.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
namespace caf::net::basp {
/// Enforces strict order of message delivery, i.e., deliver messages in the
/// same order as if they were deserialized by a single thread.
class message_queue {
public:
// -- member types -----------------------------------------------------------
/// Request for sending a message to an actor at a later time.
struct actor_msg {
uint64_t id;
strong_actor_ptr receiver;
mailbox_element_ptr content;
};
// -- constructors, destructors, and assignment operators --------------------
message_queue();
// -- mutators ---------------------------------------------------------------
/// Adds a new message to the queue or deliver it immediately if possible.
void push(execution_unit* ctx, uint64_t id, strong_actor_ptr receiver,
mailbox_element_ptr content);
/// Marks given ID as dropped, effectively skipping it without effect.
void drop(execution_unit* ctx, uint64_t id);
/// Returns the next ascending ID.
uint64_t new_id();
// -- member variables -------------------------------------------------------
/// Protects all other properties.
std::mutex lock;
/// The next available ascending ID. The counter is large enough to overflow
/// after roughly 600 years if we dispatch a message every microsecond.
uint64_t next_id;
/// The next ID that we can ship.
uint64_t next_undelivered;
/// Keeps messages in sorted order in case a message other than
/// `next_undelivered` gets ready first.
std::vector<actor_msg> pending;
};
} // namespace caf::net::basp
// 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 <cstdint>
#include <string>
#include <type_traits>
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/is_error_code_enum.hpp"
namespace caf::net::basp {
/// @addtogroup BASP
/// Describes the first header field of a BASP message and determines the
/// interpretation of the other header fields.
enum class message_type : uint8_t {
/// Sends supported BASP version and node information to the server.
///
/// ![](client_handshake.png)
handshake = 0,
/// Transmits an actor-to-actor messages.
///
/// ![](direct_message.png)
actor_message = 1,
/// Tries to resolve a path on the receiving node.
///
/// ![](resolve_request.png)
resolve_request = 2,
/// Transmits the result of a path lookup.
///
/// ![](resolve_response.png)
resolve_response = 3,
/// Informs the receiving node that the sending node has created a proxy
/// instance for one of its actors. Causes the receiving node to attach a
/// functor to the actor that triggers a down_message on termination.
///
/// ![](monitor_message.png)
monitor_message = 4,
/// Informs the receiving node that it has a proxy for an actor that has been
/// terminated.
///
/// ![](down_message.png)
down_message = 5,
/// Used to generate periodic traffic between two nodes in order to detect
/// disconnects.
///
/// ![](heartbeat.png)
heartbeat = 6,
};
/// @relates message_type
CAF_NET_EXPORT std::string to_string(message_type x);
/// @relates message_type
CAF_NET_EXPORT bool from_string(string_view, message_type&);
/// @relates message_type
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<message_type>,
message_type&);
/// @relates message_type
template <class Inspector>
bool inspect(Inspector& f, message_type& x) {
return default_enum_inspect(f, x);
}
/// @}
} // namespace caf::net::basp
// 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 <vector>
#include "caf/actor_control_block.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/config.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/execution_unit.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/node_id.hpp"
namespace caf::net::basp {
template <class Subtype>
class remote_message_handler {
public:
void handle_remote_message(execution_unit* ctx) {
// Local variables.
auto& dref = static_cast<Subtype&>(*this);
auto& payload = dref.payload_;
auto& hdr = dref.hdr_;
auto& registry = dref.system_->registry();
auto& proxies = *dref.proxies_;
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
// Deserialize payload.
actor_id src_id = 0;
node_id src_node;
actor_id dst_id = 0;
std::vector<strong_actor_ptr> fwd_stack;
message content;
binary_deserializer source{ctx, payload};
if (!(source.apply(src_node) && source.apply(src_id) && source.apply(dst_id)
&& source.apply(fwd_stack) && source.apply(content))) {
CAF_LOG_ERROR(
"failed to deserialize payload:" << CAF_ARG(source.get_error()));
return;
}
// Sanity checks.
if (dst_id == 0)
return;
// Try to fetch the receiver.
auto dst_hdl = registry.get(dst_id);
if (dst_hdl == nullptr) {
CAF_LOG_DEBUG("no actor found for given ID, drop message");
return;
}
// Try to fetch the sender.
strong_actor_ptr src_hdl;
if (src_node != none && src_id != 0)
src_hdl = proxies.get_or_put(src_node, src_id);
// Ship the message.
auto ptr = make_mailbox_element(std::move(src_hdl),
make_message_id(hdr.operation_data),
std::move(fwd_stack), std::move(content));
dref.queue_->push(ctx, dref.msg_id_, std::move(dst_hdl), std::move(ptr));
}
};
} // namespace caf::net::basp
// 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 <atomic>
#include <cstdint>
#include "caf/byte_buffer.hpp"
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/remote_message_handler.hpp"
#include "caf/net/fwd.hpp"
#include "caf/node_id.hpp"
#include "caf/resumable.hpp"
namespace caf::net::basp {
/// Deserializes payloads for BASP messages asynchronously.
class CAF_NET_EXPORT worker : public detail::abstract_worker,
public remote_message_handler<worker> {
public:
// -- friends ----------------------------------------------------------------
friend remote_message_handler<worker>;
// -- member types -----------------------------------------------------------
using super = detail::abstract_worker;
using scheduler_type = scheduler::abstract_coordinator;
using hub_type = detail::worker_hub<worker>;
// -- constructors, destructors, and assignment operators --------------------
/// Only the ::worker_hub has access to the construtor.
worker(hub_type& hub, message_queue& queue, proxy_registry& proxies);
~worker() override;
// -- management -------------------------------------------------------------
void launch(const node_id& last_hop, const basp::header& hdr,
span<const byte> payload);
// -- implementation of resumable --------------------------------------------
resume_result resume(execution_unit* ctx, size_t) override;
private:
// -- constants and assertions -----------------------------------------------
/// Stores how many bytes the "first half" of this object requires.
static constexpr size_t pointer_members_size
= sizeof(hub_type*) + sizeof(message_queue*) + sizeof(proxy_registry*)
+ sizeof(actor_system*);
static_assert(CAF_CACHE_LINE_SIZE > pointer_members_size,
"invalid cache line size");
// -- member variables -------------------------------------------------------
/// Points to our home hub.
hub_type* hub_;
/// Points to the queue for establishing strict ordering.
message_queue* queue_;
/// Points to our proxy registry / factory.
proxy_registry* proxies_;
/// Points to the parent system.
actor_system* system_;
/// Prevents false sharing when writing to `next`.
char pad_[CAF_CACHE_LINE_SIZE - pointer_members_size];
/// ID for local ordering.
uint64_t msg_id_;
/// Identifies the node that sent us `hdr_` and `payload_`.
node_id last_hop_;
/// The header for the next message. Either a direct_message or a
/// routed_message.
header hdr_;
/// Contains whatever this worker deserializes next.
byte_buffer payload_;
};
} // namespace caf::net::basp
// 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/logger.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/send.hpp"
namespace caf::net {
/// 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 input_tag = tag::io_event_oriented;
using socket_type = Socket;
using factory_type = Factory;
using read_result = typename socket_manager::read_result;
using write_result = typename socket_manager::write_result;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
explicit connection_acceptor(size_t limit, Ts&&... xs)
: factory_(std::forward<Ts>(xs)...), limit_(limit) {
// nop
}
// -- interface functions ----------------------------------------------------
template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
CAF_LOG_TRACE("");
owner_ = owner;
cfg_ = cfg;
if (auto err = factory_.init(owner, cfg))
return err;
down->register_reading();
return none;
}
template <class LowerLayerPtr>
read_result handle_read_event(LowerLayerPtr down) {
CAF_LOG_TRACE("");
if (auto x = accept(down->handle())) {
socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr());
if (!child) {
CAF_LOG_ERROR("factory failed to create a new child");
down->abort_reason(sec::runtime_error);
return read_result::stop;
}
if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
down->abort_reason(std::move(err));
return read_result::stop;
}
if (limit_ == 0) {
return read_result::again;
} else {
return ++accepted_ < limit_ ? read_result::again : read_result::stop;
}
} else {
CAF_LOG_ERROR("accept failed:" << x.error());
return read_result::stop;
}
}
template <class LowerLayerPtr>
static read_result handle_buffered_data(LowerLayerPtr) {
return read_result::again;
}
template <class LowerLayerPtr>
static read_result handle_continue_reading(LowerLayerPtr) {
return read_result::again;
}
template <class LowerLayerPtr>
write_result handle_write_event(LowerLayerPtr) {
CAF_LOG_ERROR("connection_acceptor received write event");
return write_result::stop;
}
template <class LowerLayerPtr>
static write_result handle_continue_writing(LowerLayerPtr) {
CAF_LOG_ERROR("connection_acceptor received continue writing event");
return write_result::stop;
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
factory_.abort(reason);
}
private:
factory_type factory_;
socket_manager* owner_;
size_t limit_;
size_t accepted_ = 0;
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
}
connection_acceptor_factory_adapter(connection_acceptor_factory_adapter&&)
= default;
error init(socket_manager*, const settings&) {
return none;
}
template <class Socket>
socket_manager_ptr make(Socket connected_socket, multiplexer* mpx) {
return f_(std::move(connected_socket), mpx);
}
void abort(const error&) {
// nop
}
private:
FunctionObject f_;
};
} // namespace caf::net
// 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/consumer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net {
/// Connects a socket manager to an asynchronous consumer resource. Whenever new
/// data becomes ready, the adapter registers the socket manager for writing.
template <class Buffer>
class consumer_adapter final : public ref_counted, public async::consumer {
public:
using buf_ptr = intrusive_ptr<Buffer>;
void on_producer_ready() override {
// nop
}
void on_producer_wakeup() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { //
adapter->on_wakeup();
});
}
void ref_consumer() const noexcept override {
this->ref();
}
void deref_consumer() const noexcept override {
this->deref();
}
template <class Policy, class Observer>
std::pair<bool, size_t> pull(Policy policy, size_t demand, Observer& dst) {
return buf_->pull(policy, demand, dst);
}
void cancel() {
buf_->cancel();
buf_ = nullptr;
}
bool has_data() const noexcept {
return buf_->has_data();
}
/// Tries to open the resource for reading.
/// @returns a connected adapter that reads from the resource on success,
/// `nullptr` otherwise.
template <class Resource>
static intrusive_ptr<consumer_adapter>
try_open(socket_manager* owner, Resource src) {
CAF_ASSERT(owner != nullptr);
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<consumer_adapter>;
auto adapter = ptr_type{new consumer_adapter(owner, buf), false};
buf->set_consumer(adapter);
return adapter;
} else {
return nullptr;
}
}
friend void intrusive_ptr_add_ref(const consumer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const consumer_adapter* ptr) noexcept {
ptr->deref();
}
private:
consumer_adapter(socket_manager* owner, buf_ptr buf)
: mgr_(owner), buf_(std::move(buf)) {
// nop
}
auto strong_this() {
return intrusive_ptr{this};
}
void on_wakeup() {
if (buf_ && buf_->has_consumer_event()) {
mgr_->mpx().register_writing(mgr_);
}
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
};
template <class T>
using consumer_adapter_ptr = intrusive_ptr<consumer_adapter<T>>;
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/variant.hpp"
namespace caf::net {
/// A datagram-oriented network communication endpoint.
struct CAF_NET_EXPORT datagram_socket : network_socket {
using super = network_socket;
using super::super;
};
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates datagram_socket
error CAF_NET_EXPORT allow_connreset(datagram_socket x, bool new_value);
/// Converts the result from I/O operation on a ::datagram_socket to either an
/// error code or a integer greater or equal to zero.
/// @relates datagram_socket
variant<size_t, sec> CAF_NET_EXPORT
check_datagram_socket_io_res(std::make_signed<size_t>::type res);
} // namespace caf::net
// 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 <deque>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
namespace caf::net {
template <class Factory>
using datagram_transport_base
= transport_base<datagram_transport<Factory>,
transport_worker_dispatcher<Factory, ip_endpoint>,
udp_datagram_socket, Factory, ip_endpoint>;
/// Implements a udp_transport policy that manages a datagram socket.
template <class Factory>
class datagram_transport : public datagram_transport_base<Factory> {
public:
// Maximal UDP-packet size
static constexpr size_t max_datagram_size
= std::numeric_limits<uint16_t>::max();
// -- member types -----------------------------------------------------------
using factory_type = Factory;
using id_type = ip_endpoint;
using application_type = typename factory_type::application_type;
using super = datagram_transport_base<factory_type>;
using buffer_cache_type = typename super::buffer_cache_type;
// -- constructors, destructors, and assignment operators --------------------
datagram_transport(udp_datagram_socket handle, factory_type factory)
: super(handle, std::move(factory)) {
// nop
}
// -- public member functions ------------------------------------------------
error init(endpoint_manager& manager) override {
CAF_LOG_TRACE("");
if (auto err = super::init(manager))
return err;
prepare_next_read();
return none;
}
bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) {
auto ret = read(this->handle_, this->read_buf_);
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto& [num_bytes, ep] = *res;
CAF_LOG_DEBUG("received " << num_bytes << " bytes");
this->read_buf_.resize(num_bytes);
if (auto err = this->next_layer_.handle_data(*this, this->read_buf_,
std::move(ep))) {
CAF_LOG_ERROR("handle_data failed: " << err);
return false;
}
prepare_next_read();
} else {
auto err = get<sec>(ret);
if (err == sec::unavailable_or_would_block) {
break;
} else {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
}
}
return true;
}
bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("socket", this->handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size()));
auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
return true;
}
return false;
};
do {
if (auto err = write_some())
return err == sec::unavailable_or_would_block;
} while (fetch_next_message());
return !packet_queue_.empty();
}
// TODO: remove this function. `resolve` should add workers when needed.
error add_new_worker(node_id node, id_type id) {
auto worker = this->next_layer_.add_new_worker(*this, node, id);
if (!worker)
return worker.error();
return none;
}
void write_packet(id_type id, span<byte_buffer*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty())
this->manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
packet_queue_.emplace_back(id, buffers);
}
/// Helper struct for managing outgoing packets
struct packet {
id_type id;
buffer_cache_type bytes;
size_t size;
packet(id_type id, span<byte_buffer*> bufs) : id(id) {
size = 0;
for (auto buf : bufs) {
size += buf->size();
bytes.emplace_back(std::move(*buf));
}
}
std::vector<byte_buffer*> get_buffer_ptrs() {
std::vector<byte_buffer*> ptrs;
for (auto& buf : bytes)
ptrs.emplace_back(&buf);
return ptrs;
}
};
private:
// -- utility functions ------------------------------------------------------
void prepare_next_read() {
this->read_buf_.resize(max_datagram_size);
}
error write_some() {
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = packet_queue_.front();
auto& bufs = front.bytes;
auto it = bufs.begin();
if (this->header_bufs_.size() < this->header_bufs_.capacity()) {
it->clear();
this->header_bufs_.emplace_back(std::move(*it++));
}
for (; it != bufs.end()
&& this->payload_bufs_.size() < this->payload_bufs_.capacity();
++it) {
it->clear();
this->payload_bufs_.emplace_back(std::move(*it));
}
packet_queue_.pop_front();
};
// Write as many bytes as possible.
while (!packet_queue_.empty()) {
auto& packet = packet_queue_.front();
auto ptrs = packet.get_buffer_ptrs();
auto write_ret = write(this->handle_, ptrs, packet.id);
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
CAF_LOG_WARNING_IF(*num_bytes < packet.size,
"packet was not sent completely");
recycle();
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_ERROR("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
}
return err;
}
}
return none;
}
std::deque<packet> packet_queue_;
};
} // namespace caf::net
// 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 <cstddef>
#include <cstdint>
// -- hard-coded default values for various CAF options ------------------------
namespace caf::defaults::middleman {
/// Maximum number of cached buffers for sending payloads.
constexpr size_t max_payload_buffers = 100;
/// Maximum number of cached buffers for sending headers.
constexpr size_t max_header_buffers = 10;
/// Port to listen on for tcp.
constexpr uint16_t tcp_port = 0;
} // namespace caf::defaults::middleman
// 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 <cstddef>
#include <cstdint>
#include <memory>
#include "caf/actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/variant.hpp"
namespace caf::net {
/// Manages a communication endpoint.
class CAF_NET_EXPORT endpoint_manager : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using super = socket_manager;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys);
~endpoint_manager() override;
// -- properties -------------------------------------------------------------
actor_system& system() noexcept {
return sys_;
}
const actor_system_config& config() const noexcept;
// -- queue access -----------------------------------------------------------
bool at_end_of_message_queue();
endpoint_manager_queue::message_ptr next_message();
// -- event management -------------------------------------------------------
/// Resolves a path to a remote actor.
void resolve(uri locator, actor listener);
/// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver);
/// Enqueues an event to the endpoint.
template <class... Ts>
void enqueue_event(Ts&&... xs) {
enqueue(new endpoint_manager_queue::event(std::forward<Ts>(xs)...));
}
// -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop.
// virtual error init() = 0;
protected:
bool enqueue(endpoint_manager_queue::element* ptr);
/// Points to the hosting actor system.
actor_system& sys_;
/// Stores control events and outbound messages.
endpoint_manager_queue::type queue_;
/// Stores a proxy for interacting with the actor clock.
actor timeout_proxy_;
};
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
} // namespace caf::net
// 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/abstract_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/detail/overload.hpp"
#include "caf/net/endpoint_manager.hpp"
namespace caf::net {
template <class Transport>
class endpoint_manager_impl : public endpoint_manager {
public:
// -- member types -----------------------------------------------------------
using super = endpoint_manager;
using transport_type = Transport;
using application_type = typename transport_type::application_type;
using read_result = typename super::read_result;
using write_result = typename super::write_result;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
socket handle, Transport trans)
: super(handle, parent, sys), transport_(std::move(trans)) {
// nop
}
~endpoint_manager_impl() override {
// nop
}
// -- properties -------------------------------------------------------------
transport_type& transport() {
return transport_;
}
endpoint_manager_impl& manager() {
return *this;
}
// -- interface functions ----------------------------------------------------
error init() /*override*/ {
this->register_reading();
return transport_.init(*this);
}
read_result handle_read_event() override {
return transport_.handle_read_event(*this);
}
write_result handle_write_event() override {
if (!this->queue_.blocked()) {
this->queue_.fetch_more();
auto& q = std::get<0>(this->queue_.queue().queues());
do {
q.inc_deficit(q.total_task_size());
for (auto ptr = q.next(); ptr != nullptr; ptr = q.next()) {
auto f = detail::make_overload(
[&](endpoint_manager_queue::event::resolve_request& x) {
transport_.resolve(*this, x.locator, x.listener);
},
[&](endpoint_manager_queue::event::new_proxy& x) {
transport_.new_proxy(*this, x.peer, x.id);
},
[&](endpoint_manager_queue::event::local_actor_down& x) {
transport_.local_actor_down(*this, x.observing_peer, x.id,
std::move(x.reason));
},
[&](endpoint_manager_queue::event::timeout& x) {
transport_.timeout(*this, x.type, x.id);
});
visit(f, ptr->value);
}
} while (!q.empty());
}
if (!transport_.handle_write_event(*this)) {
if (this->queue_.blocked())
return write_result::stop;
else if (!(this->queue_.empty() && this->queue_.try_block()))
return write_result::again;
else
return write_result::stop;
}
return write_result::again;
}
void handle_error(sec code) override {
transport_.handle_error(code);
}
private:
transport_type transport_;
/// Stores the id for the next timeout.
uint64_t next_timeout_id_;
error err_;
};
} // namespace caf::net
// 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 <string>
#include "caf/actor.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/serialized_size.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/unit.hpp"
#include "caf/variant.hpp"
namespace caf::net {
class CAF_NET_EXPORT endpoint_manager_queue {
public:
enum class element_type { event, message };
class element : public intrusive::singly_linked<element> {
public:
explicit element(element_type tag) : tag_(tag) {
// nop
}
virtual ~element();
virtual size_t task_size() const noexcept = 0;
element_type tag() const noexcept {
return tag_;
}
private:
element_type tag_;
};
using element_ptr = std::unique_ptr<element>;
class event final : public element {
public:
struct resolve_request {
uri locator;
actor listener;
};
struct new_proxy {
node_id peer;
actor_id id;
};
struct local_actor_down {
node_id observing_peer;
actor_id id;
error reason;
};
struct timeout {
std::string type;
uint64_t id;
};
event(uri locator, actor listener);
event(node_id peer, actor_id proxy_id);
event(node_id observing_peer, actor_id local_actor_id, error reason);
event(std::string type, uint64_t id);
~event() override;
size_t task_size() const noexcept override;
/// Holds the event data.
variant<resolve_request, new_proxy, local_actor_down, timeout> value;
};
using event_ptr = std::unique_ptr<event>;
struct event_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = event;
using unique_pointer = event_ptr;
using queue_type = intrusive::drr_queue<event_policy>;
constexpr event_policy(unit_t) {
// nop
}
static constexpr task_size_type task_size(const event&) noexcept {
return 1;
}
};
class message : public element {
public:
/// Original message to a remote actor.
mailbox_element_ptr msg;
/// ID of the receiving actor.
strong_actor_ptr receiver;
message(mailbox_element_ptr msg, strong_actor_ptr receiver);
~message() override;
size_t task_size() const noexcept override;
};
using message_ptr = std::unique_ptr<message>;
struct message_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = message;
using unique_pointer = message_ptr;
using queue_type = intrusive::drr_queue<message_policy>;
constexpr message_policy(unit_t) {
// nop
}
static task_size_type task_size(const message& msg) noexcept {
return detail::serialized_size(msg.msg->content());
}
};
struct categorized {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = element;
using unique_pointer = element_ptr;
constexpr categorized(unit_t) {
// nop
}
template <class Queue>
deficit_type quantum(const Queue&, deficit_type x) const noexcept {
return x;
}
size_t id_of(const element& x) const noexcept {
return static_cast<size_t>(x.tag());
}
};
struct policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = element;
using unique_pointer = std::unique_ptr<element>;
using queue_type = intrusive::wdrr_fixed_multiplexed_queue<
categorized, event_policy::queue_type, message_policy::queue_type>;
task_size_type task_size(const message& x) const noexcept {
return x.task_size();
}
};
using type = intrusive::fifo_inbox<policy>;
};
} // namespace caf::net
// 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 <memory>
#include "caf/intrusive_ptr.hpp"
#include "caf/type_id.hpp"
namespace caf::net {
// -- templates ----------------------------------------------------------------
template <class UpperLayer>
class stream_transport;
template <class Factory>
class datagram_transport;
template <class Application, class IdType = unit_t>
class transport_worker;
template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher;
template <class... Sigs>
class typed_actor_shell;
template <class... Sigs>
class typed_actor_shell_ptr;
// -- enumerations -------------------------------------------------------------
enum class ec : uint8_t;
// -- classes ------------------------------------------------------------------
class actor_shell;
class actor_shell_ptr;
class endpoint_manager;
class middleman;
class middleman_backend;
class multiplexer;
class socket_manager;
// -- structs ------------------------------------------------------------------
struct network_socket;
struct pipe_socket;
struct socket;
struct stream_socket;
struct tcp_accept_socket;
struct tcp_stream_socket;
struct datagram_socket;
struct udp_datagram_socket;
// -- smart pointer aliases ----------------------------------------------------
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
using multiplexer_ptr = std::shared_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
} // namespace caf::net
namespace caf::net::basp {
enum class ec : uint8_t;
} // namespace caf::net::basp
CAF_BEGIN_TYPE_ID_BLOCK(net_module, detail::net_module_begin)
CAF_ADD_TYPE_ID(net_module, (caf::net::basp::ec))
CAF_END_TYPE_ID_BLOCK(net_module)
static_assert(caf::id_block::net_module::end == caf::detail::net_module_end);
// 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/net/socket_manager.hpp"
#include "caf/net/stream_transport_error.hpp"
namespace caf::net {
template <class OnSuccess, class OnError>
struct default_handshake_worker_factory {
OnSuccess make;
OnError abort;
};
/// An connect worker calls an asynchronous `connect` callback until it
/// succeeds. On success, the worker calls a factory object to transfer
/// ownership of socket and communication policy to the create the socket
/// manager that takes care of the established connection.
template <bool IsServer, class Socket, class Policy, class Factory>
class handshake_worker : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using super = socket_manager;
using read_result = typename super::read_result;
using write_result = typename super::write_result;
handshake_worker(Socket handle, multiplexer* parent, Policy policy,
Factory factory)
: super(handle, parent),
policy_(std::move(policy)),
factory_(std::move(factory)) {
// nop
}
// -- interface functions ----------------------------------------------------
error init(const settings& config) override {
cfg_ = config;
register_writing();
return caf::none;
}
read_result handle_read_event() override {
auto fd = socket_cast<Socket>(this->handle());
if (auto res = advance_handshake(fd); res > 0) {
return read_result::handover;
} else if (res == 0) {
factory_.abort(make_error(sec::connection_closed));
return read_result::stop;
} else {
auto err = policy_.last_error(fd, res);
switch (err) {
case stream_transport_error::want_read:
case stream_transport_error::temporary:
return read_result::again;
case stream_transport_error::want_write:
return read_result::want_write;
default:
auto err = make_error(sec::cannot_connect_to_node,
policy_.fetch_error_str());
factory_.abort(std::move(err));
return read_result::stop;
}
}
}
read_result handle_buffered_data() override {
return read_result::again;
}
read_result handle_continue_reading() override {
return read_result::again;
}
write_result handle_write_event() override {
auto fd = socket_cast<Socket>(this->handle());
if (auto res = advance_handshake(fd); res > 0) {
return write_result::handover;
} else if (res == 0) {
factory_.abort(make_error(sec::connection_closed));
return write_result::stop;
} else {
switch (policy_.last_error(fd, res)) {
case stream_transport_error::want_write:
case stream_transport_error::temporary:
return write_result::again;
case stream_transport_error::want_read:
return write_result::want_read;
default:
auto err = make_error(sec::cannot_connect_to_node,
policy_.fetch_error_str());
factory_.abort(std::move(err));
return write_result::stop;
}
}
}
write_result handle_continue_writing() override {
return write_result::again;
}
void handle_error(sec code) override {
factory_.abort(make_error(code));
}
socket_manager_ptr make_next_manager(socket hdl) override {
auto ptr = factory_.make(socket_cast<Socket>(hdl), this->mpx_ptr(),
std::move(policy_));
if (ptr) {
if (auto err = ptr->init(cfg_)) {
factory_.abort(err);
return nullptr;
} else {
return ptr;
}
} else {
factory_.abort(make_error(sec::runtime_error, "factory_.make failed"));
return nullptr;
}
}
private:
ptrdiff_t advance_handshake(Socket fd) {
if constexpr (IsServer)
return policy_.accept(fd);
else
return policy_.connect(fd);
}
settings cfg_;
Policy policy_;
Factory factory_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net {
struct CAF_NET_EXPORT this_host {
/// Initializes the network subsystem.
static error startup();
/// Release any resources of the network subsystem.
static void cleanup();
};
} // namespace caf::net
// 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::http {
/// Stores context information for a request. For HTTP/2, this is the stream ID.
struct context {};
} // namespace caf::net::http
// 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/config_value.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/method.hpp"
#include "caf/net/http/status.hpp"
#include "caf/string_view.hpp"
#include "caf/uri.hpp"
#include <vector>
namespace caf::net::http {
class CAF_NET_EXPORT header {
public:
header() = default;
header(header&&) = default;
header& operator=(header&&) = default;
header(const header&);
header& operator=(const header&);
void assign(const header&);
http::method method() const noexcept {
return method_;
}
string_view path() const noexcept {
return uri_.path();
}
const uri::query_map& query() const noexcept {
return uri_.query();
}
string_view fragment() const noexcept {
return uri_.fragment();
}
string_view version() const noexcept {
return version_;
}
const header_fields_map& fields() const noexcept {
return fields_;
}
string_view field(string_view key) const noexcept {
if (auto i = fields_.find(key); i != fields_.end())
return i->second;
else
return {};
}
template <class T>
optional<T> field_as(string_view key) const noexcept {
if (auto i = fields_.find(key); i != fields_.end()) {
caf::config_value val{to_string(i->second)};
if (auto res = caf::get_as<T>(val))
return std::move(*res);
else
return {};
} else {
return {};
}
}
bool valid() const noexcept {
return !raw_.empty();
}
std::pair<status, string_view> parse(string_view raw);
bool chunked_transfer_encoding() const;
optional<size_t> content_length() const;
private:
std::vector<char> raw_;
http::method method_;
uri uri_;
string_view version_;
header_fields_map fields_;
};
} // namespace caf::net::http
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/string_view.hpp"
namespace caf::net::http {
/// Convenience type alias for a key-value map storing header fields.
using header_fields_map = detail::unordered_flat_map<string_view, string_view>;
} // namespace caf::net::http
// 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/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/string_view.hpp"
#include <cstdint>
#include <string>
#include <type_traits>
namespace caf::net::http {
/// Methods as defined by RFC 7231.
enum class method : uint8_t {
/// Requests transfer of a current selected representation for the target
/// resource.
get,
/// Identical to GET except that the server MUST NOT send a message body in
/// the response. The server SHOULD send the same header fields in response to
/// a HEAD request as it would have sent if the request had been a GET, except
/// that the payload header fields (Section 3.3) MAY be omitted.
head,
/// Requests that the target resource process the representation enclosed in
/// the request according to the resource's own specific semantics.
post,
/// Requests that the state of the target resource be created or replaced with
/// the state defined by the representation enclosed in the request message
/// payload.
put,
/// Requests that the origin server remove the association between the target
/// resource and its current functionality.
del,
/// Requests that the recipient establish a tunnel to the destination origin
/// server identified by the request-target and, if successful, thereafter
/// restrict its behavior to blind forwarding of packets, in both directions,
/// until the tunnel is closed.
connect,
/// Requests information about the communication options available for the
/// target resource, at either the origin server or an intervening
/// intermediary.
options,
/// Requests a remote, application-level loop-back of the request message.
trace,
};
/// @relates method
CAF_NET_EXPORT std::string to_string(method);
/// Converts @p x to the RFC string representation, i.e., all-uppercase.
/// @relates method
CAF_NET_EXPORT std::string to_rfc_string(method x);
/// @relates method
CAF_NET_EXPORT bool from_string(string_view, method&);
/// @relates method
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<method>, method&);
/// @relates method
template <class Inspector>
bool inspect(Inspector& f, method& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net::http
This diff is collapsed.
// 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/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/string_view.hpp"
#include <cstdint>
#include <string>
#include <type_traits>
namespace caf::net::http {
/// Status codes as defined by RFC 7231 and RFC 6585.
enum class status : uint16_t {
/// Indicates that the initial part of a request has been received and has not
/// yet been rejected by the server. The server intends to send a final
/// response after the request has been fully received and acted upon.
continue_request = 100,
/// Indicates that the server understands and is willing to comply with the
/// client's request for a change in the application protocol being used on
/// this connection.
switching_protocols = 101,
/// Indicates that the request has succeeded.
ok = 200,
/// Indicates that the request has been fulfilled and has resulted in one or
/// more new resources being created.
created = 201,
/// Indicates that the request has been accepted for processing, but the
/// processing has not been completed.
accepted = 202,
/// Indicates that the request was successful but the enclosed payload has
/// been modified from that of the origin server's 200 (OK) response by a
/// transforming proxy
non_authoritative_information = 203,
/// Indicates that the server has successfully fulfilled the request and that
/// there is no additional content to send in the response payload body.
no_content = 204,
/// Indicates that the server has fulfilled the request and desires that the
/// user agent reset the "document view".
reset_content = 205,
/// Indicates that the server is successfully fulfilling a range request for
/// the target resource by transferring one or more parts of the selected
/// representation that correspond to the satisfiable ranges found in the
/// request's Range header field.
partial_content = 206,
/// Indicates that the target resource has more than one representation and
/// information about the alternatives is being provided so that the user (or
/// user agent) can select a preferred representation.
multiple_choices = 300,
/// Indicates that the target resource has been assigned a new permanent URI.
moved_permanently = 301,
/// Indicates that the target resource resides temporarily under a different
/// URI.
found = 302,
/// Indicates that the server is redirecting the user agent to a different
/// resource, as indicated by a URI in the Location header field, which is
/// intended to provide an indirect response to the original request.
see_other = 303,
/// Indicates that a conditional GET or HEAD request has been received and
/// would have resulted in a 200 (OK) response if it were not for the fact
/// that the condition evaluated to false.
not_modified = 304,
/// Deprecated.
use_proxy = 305,
/// No longer valid.
temporary_redirect = 307,
/// Indicates that the server cannot or will not process the request due to
/// something that is perceived to be a client error (e.g., malformed request
/// syntax, invalid request message framing, or deceptive request routing).
bad_request = 400,
/// Indicates that the request has not been applied because it lacks valid
/// authentication credentials for the target resource.
unauthorized = 401,
/// Reserved for future use.
payment_required = 402,
/// Indicates that the server understood the request but refuses to authorize
/// it.
forbidden = 403,
/// Indicates that the origin server did not find a current representation for
/// the target resource or is not willing to disclose that one exists.
not_found = 404,
/// Indicates that the method received in the request-line is known by the
/// origin server but not supported by the target resource.
method_not_allowed = 405,
/// Indicates that the target resource does not have a current representation
/// that would be acceptable to the user agent, according to the proactive
/// negotiation header fields received in the request (Section 5.3), and the
/// server is unwilling to supply a default representation.
not_acceptable = 406,
/// Similar to 401 (Unauthorized), but it indicates that the client needs to
/// authenticate itself in order to use a proxy.
proxy_authentication_required = 407,
/// Indicates that the server did not receive a complete request message
/// within the time that it was prepared to wait.
request_timeout = 408,
/// Indicates that the request could not be completed due to a conflict with
/// the current state of the target resource. This code is used in situations
/// where the user might be able to resolve the conflict and resubmit the
/// request.
conflict = 409,
/// Indicates that access to the target resource is no longer available at the
/// origin server and that this condition is likely to be permanent.
gone = 410,
/// Indicates that the server refuses to accept the request without a defined
/// Content-Length.
length_required = 411,
/// Indicates that one or more conditions given in the request header fields
/// evaluated to false when tested on the server.
precondition_failed = 412,
/// Indicates that the server is refusing to process a request because the
/// request payload is larger than the server is willing or able to process.
payload_too_large = 413,
/// Indicates that the server is refusing to service the request because the
/// request-target is longer than the server is willing to interpret.
uri_too_long = 414,
/// Indicates that the origin server is refusing to service the request
/// because the payload is in a format not supported by this method on the
/// target resource.
unsupported_media_type = 415,
/// Indicates that none of the ranges in the request's Range header field
/// overlap the current extent of the selected resource or that the set of
/// ranges requested has been rejected due to invalid ranges or an excessive
/// request of small or overlapping ranges.
range_not_satisfiable = 416,
/// Indicates that the expectation given in the request's Expect header field
/// could not be met by at least one of the inbound servers.
expectation_failed = 417,
/// Indicates that the server refuses to perform the request using the current
/// protocol but might be willing to do so after the client upgrades to a
/// different protocol.
upgrade_required = 426,
/// Indicates that the origin server requires the request to be conditional.
precondition_required = 428,
/// Indicates that the user has sent too many requests in a given amount of
/// time ("rate limiting").
too_many_requests = 429,
/// Indicates that the server is unwilling to process the request because its
/// header fields are too large.
request_header_fields_too_large = 431,
/// Indicates that the server encountered an unexpected condition that
/// prevented it from fulfilling the request.
internal_server_error = 500,
/// Indicates that the server does not support the functionality required to
/// fulfill the request.
not_implemented = 501,
/// Indicates that the server, while acting as a gateway or proxy, received an
/// invalid response from an inbound server it accessed while attempting to
/// fulfill the request.
bad_gateway = 502,
/// Indicates that the server is currently unable to handle the request due to
/// a temporary overload or scheduled maintenance, which will likely be
/// alleviated after some delay.
service_unavailable = 503,
/// Indicates that the server, while acting as a gateway or proxy, did not
/// receive a timely response from an upstream server it needed to access in
/// order to complete the request.
gateway_timeout = 504,
/// Indicates that the server does not support, or refuses to support, the
/// major version of HTTP that was used in the request message.
http_version_not_supported = 505,
/// Indicates that the client needs to authenticate to gain network access.
network_authentication_required = 511,
};
/// Returns the recommended response phrase to a status code.
/// @relates status
CAF_NET_EXPORT string_view phrase(status);
/// @relates status
CAF_NET_EXPORT std::string to_string(status);
/// @relates status
CAF_NET_EXPORT bool from_string(string_view, status&);
/// @relates status
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<status>, status&);
/// @relates status
template <class Inspector>
bool inspect(Inspector& f, status& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net::http
// 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/http/header_fields_map.hpp"
#include "caf/net/http/status.hpp"
#include "caf/string_view.hpp"
#include <utility>
namespace caf::net::http::v1 {
/// Tries splitting the given byte span into an HTTP header (`first`) and a
/// remainder (`second`). Returns an empty @ref string_view as `first` for
/// incomplete HTTP headers.
CAF_NET_EXPORT std::pair<string_view, byte_span> split_header(byte_span bytes);
/// Writes an HTTP header to the buffer.
CAF_NET_EXPORT void write_header(status code, const header_fields_map& fields,
byte_buffer& buf);
/// Writes a complete HTTP response to the buffer. Automatically sets
/// Content-Type and Content-Length header fields.
CAF_NET_EXPORT void write_response(status code, string_view content_type,
string_view content, byte_buffer& buf);
/// Writes a complete HTTP response to the buffer. Automatically sets
/// Content-Type and Content-Length header fields followed by the user-defined
/// @p fields.
CAF_NET_EXPORT void
write_response(status code, string_view content_type, string_view content,
const header_fields_map& fields, byte_buffer& buf);
} // namespace caf::net::http::v1
// 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a hypertext-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// hypertext-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class hypertext_oriented_layer_ptr {
public:
using context_type = typename Layer::context_type;
using status_code_type = typename Layer::status_code_type;
using header_fields_type = typename Layer::header_fields_type;
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
/// Queries whether the underlying transport can send additional data.
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
/// Asks the underlying transport to stop receiving additional data until
/// resumed.
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
/// Returns the socket handle.
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
/// Sends a response header for answering the request identified by the
/// context.
/// @param context Identifies which request this response belongs to.
/// @param code Indicates either success or failure to the client.
/// @param fields Various informational fields for the client. When sending
/// a payload afterwards, the fields should at least include
/// the content length.
bool send_header(context_type context, status_code_type code,
const header_fields_type& fields) {
return lptr_->send_header(llptr_, context, code, fields);
}
/// Sends a payload to the client. Must follow a header.
/// @param context Identifies which request this response belongs to.
/// @param bytes Arbitrary data for the client.
/// @pre `bytes.size() > 0`
[[nodiscard]] bool send_payload(context_type context,
const_byte_span bytes) {
return lptr_->send_payload(llptr_, context, bytes);
}
/// Sends a single chunk of arbitrary data. The chunks must follow a header.
/// @pre `bytes.size() > 0`
[[nodiscard]] bool send_chunk(context_type context, const_byte_span bytes) {
return lptr_->send_chunk(llptr_, context, bytes);
}
/// Informs the client that the transfer completed, i.e., that the server
/// will not send additional chunks.
[[nodiscard]] bool send_end_of_chunks(context_type context) {
return lptr_->send_end_of_chunks(llptr_, context);
}
/// Convenience function for completing a request 'raw' (without adding
/// additional header fields) in a single function call. Calls
/// `send_header`, `send_payload` and `fin`.
bool send_raw_response(context_type context, status_code_type code,
const header_fields_type& fields,
const_byte_span content) {
if (lptr_->send_header(llptr_, context, code, fields)
&& (content.empty()
|| lptr_->send_payload(llptr_, context, content))) {
lptr_->fin(llptr_, context);
return true;
} else {
return false;
}
}
/// Convenience function for completing a request in a single function call.
/// Automatically sets the header fields 'Content-Type' and
/// 'Content-Length'. Calls `send_header`, `send_payload` and `fin`.
bool send_response(context_type context, status_code_type code,
header_fields_type fields, string_view content_type,
const_byte_span content) {
std::string len;
if (!content.empty()) {
auto len = std::to_string(content.size());
fields.emplace("Content-Length", len);
fields.emplace("Content-Type", content_type);
}
return send_raw_response(context, code, fields, content);
}
/// Convenience function for completing a request in a single function call.
/// Automatically sets the header fields 'Content-Type' and
/// 'Content-Length'. Calls `send_header`, `send_payload` and `fin`.
bool send_response(context_type context, status_code_type code,
string_view content_type, const_byte_span content) {
std::string len;
header_fields_type fields;
if (!content.empty()) {
len = std::to_string(content.size());
fields.emplace("Content-Type", content_type);
fields.emplace("Content-Length", len);
}
return send_raw_response(context, code, fields, content);
}
bool send_response(context_type context, status_code_type code,
header_fields_type fields, string_view content_type,
string_view content) {
return send_response(context, code, content_type, std::move(fields),
as_bytes(make_span(content)));
}
bool send_response(context_type context, status_code_type code,
string_view content_type, string_view content) {
return send_response(context, code, content_type,
as_bytes(make_span(content)));
}
void fin(context_type context) {
lptr_->fin(llptr_, context);
}
/// Sets an abort reason on the transport.
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
/// Returns the current abort reason on the transport or a
/// default-constructed error is no error occurred yet.
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
hypertext_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
hypertext_oriented_layer_ptr(const hypertext_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_hypertext_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
using result_t = hypertext_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
// 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 <string>
#include <vector>
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net::ip {
/// Returns all IP addresses of `host` (if any).
std::vector<ip_address> CAF_NET_EXPORT resolve(string_view host);
/// Returns all IP addresses of `host` (if any).
std::vector<ip_address> CAF_NET_EXPORT resolve(ip_address host);
/// Returns the IP addresses for a local endpoint, which is either an address,
/// an interface name, or the string "localhost".
std::vector<ip_address> CAF_NET_EXPORT local_addresses(string_view host);
/// Returns the IP addresses for a local endpoint address.
std::vector<ip_address> CAF_NET_EXPORT local_addresses(ip_address host);
/// Returns the hostname of this device.
std::string CAF_NET_EXPORT hostname();
} // namespace caf::net::ip
// 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 <cstdint>
#include <cstring>
#include <memory>
#include <type_traits>
#include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/has_after_reading.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/error.hpp"
#include "caf/net/message_flow_bridge.hpp"
#include "caf/net/message_oriented_layer_ptr.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/tag/message_oriented.hpp"
#include "caf/tag/no_auto_reading.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace caf::net {
/// Length-prefixed message framing for discretizing a Byte stream into messages
/// of varying size. The framing uses 4 Bytes for the length prefix, but
/// messages (including the 4 Bytes for the length prefix) are limited to a
/// maximum size of INT32_MAX. This limitation comes from the POSIX API (recv)
/// on 32-bit platforms.
template <class UpperLayer>
class length_prefix_framing {
public:
using input_tag = tag::stream_oriented;
using output_tag = tag::message_oriented;
using length_prefix_type = uint32_t;
static constexpr size_t hdr_size = sizeof(uint32_t);
static constexpr size_t max_message_length = INT32_MAX - sizeof(uint32_t);
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
explicit length_prefix_framing(Ts&&... xs)
: upper_layer_(std::forward<Ts>(xs)...) {
// nop
}
// -- initialization ---------------------------------------------------------
template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
if constexpr (!std::is_base_of_v<tag::no_auto_reading, UpperLayer>)
down->configure_read(receive_policy::exactly(hdr_size));
return upper_layer_.init(owner, this_layer_ptr(down), cfg);
}
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_;
}
const auto& upper_layer() const noexcept {
return upper_layer_;
}
// -- interface for the upper layer ------------------------------------------
template <class LowerLayerPtr>
static bool can_send_more(LowerLayerPtr down) noexcept {
return down->can_send_more();
}
template <class LowerLayerPtr>
static auto handle(LowerLayerPtr down) noexcept {
return down->handle();
}
template <class LowerLayerPtr>
static void suspend_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::stop());
}
template <class LowerLayerPtr>
void begin_message(LowerLayerPtr down) {
down->begin_output();
auto& buf = down->output_buffer();
message_offset_ = buf.size();
buf.insert(buf.end(), 4, byte{0});
}
template <class LowerLayerPtr>
byte_buffer& message_buffer(LowerLayerPtr down) {
return down->output_buffer();
}
template <class LowerLayerPtr>
[[nodiscard]] bool end_message(LowerLayerPtr down) {
using detail::to_network_order;
auto& buf = down->output_buffer();
CAF_ASSERT(message_offset_ < buf.size());
auto msg_begin = buf.begin() + message_offset_;
auto msg_size = std::distance(msg_begin + 4, buf.end());
if (msg_size > 0 && static_cast<size_t>(msg_size) < max_message_length) {
auto u32_size = to_network_order(static_cast<uint32_t>(msg_size));
memcpy(std::addressof(*msg_begin), &u32_size, 4);
down->end_output();
return true;
} else {
auto err = make_error(sec::runtime_error,
msg_size == 0 ? "logic error: message of size 0"
: "maximum message size exceeded");
down->abort_reason(std::move(err));
return false;
}
}
template <class LowerLayerPtr>
bool send_close_message(LowerLayerPtr) {
// nop; this framing layer has no close handshake
return true;
}
template <class LowerLayerPtr>
bool send_close_message(LowerLayerPtr, const error&) {
// nop; this framing layer has no close handshake
return true;
}
template <class LowerLayerPtr>
static void abort_reason(LowerLayerPtr down, error reason) {
return down->abort_reason(std::move(reason));
}
template <class LowerLayerPtr>
static const error& abort_reason(LowerLayerPtr down) {
return down->abort_reason();
}
// -- interface for the lower layer ------------------------------------------
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::exactly(hdr_size));
}
template <class LowerLayerPtr>
std::enable_if_t<detail::has_after_reading_v<
UpperLayer,
message_oriented_layer_ptr<length_prefix_framing, LowerLayerPtr>>>
after_reading(LowerLayerPtr down) {
return upper_layer_.after_reading(this_layer_ptr(down));
}
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
return upper_layer_.prepare_send(this_layer_ptr(down));
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr down) {
return upper_layer_.done_sending(this_layer_ptr(down));
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const error& reason) {
upper_layer_.abort(this_layer_ptr(down), reason);
}
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span input, byte_span) {
CAF_LOG_TRACE("got" << input.size() << "bytes");
auto this_layer = this_layer_ptr(down);
if (input.size() < sizeof(uint32_t)) {
auto err = make_error(sec::runtime_error,
"received too few bytes from underlying transport");
down->abort_reason(std::move(err));
return -1;
} else if (input.size() == hdr_size) {
auto u32_size = uint32_t{0};
memcpy(&u32_size, input.data(), sizeof(uint32_t));
auto msg_size = static_cast<size_t>(detail::from_network_order(u32_size));
if (msg_size == 0) {
// Ignore empty messages.
CAF_LOG_DEBUG("received empty message");
return static_cast<ptrdiff_t>(input.size());
} else if (msg_size > max_message_length) {
CAF_LOG_DEBUG("maximum message size exceeded");
auto err = make_error(sec::runtime_error,
"maximum message size exceeded");
down->abort_reason(std::move(err));
return -1;
} else {
CAF_LOG_DEBUG("wait for payload of size" << msg_size);
down->configure_read(receive_policy::exactly(hdr_size + msg_size));
return 0;
}
} else {
auto [msg_size, msg] = split(input);
if (msg_size == msg.size() && msg_size + hdr_size == input.size()) {
CAF_LOG_DEBUG("got message of size" << msg_size);
if (upper_layer_.consume(this_layer, msg) >= 0) {
if (!down->stopped())
down->configure_read(receive_policy::exactly(hdr_size));
return static_cast<ptrdiff_t>(input.size());
} else {
return -1;
}
} else {
CAF_LOG_DEBUG("received malformed message");
auto err = make_error(sec::runtime_error, "received malformed message");
down->abort_reason(std::move(err));
return -1;
}
}
}
// -- convenience functions --------------------------------------------------
static std::pair<size_t, byte_span> split(byte_span buffer) noexcept {
CAF_ASSERT(buffer.size() >= sizeof(uint32_t));
auto u32_size = uint32_t{0};
memcpy(&u32_size, buffer.data(), sizeof(uint32_t));
auto msg_size = static_cast<size_t>(detail::from_network_order(u32_size));
return std::make_pair(msg_size, buffer.subspan(sizeof(uint32_t)));
}
private:
// -- implementation details -------------------------------------------------
template <class LowerLayerPtr>
auto this_layer_ptr(LowerLayerPtr down) {
return make_message_oriented_layer_ptr(this, down);
}
// -- member variables -------------------------------------------------------
UpperLayer upper_layer_;
size_t message_offset_ = 0;
};
// -- high-level factory functions -------------------------------------------
/// Runs a WebSocket server on the connected socket `fd`.
/// @param mpx The multiplexer that takes ownership of the socket.
/// @param fd A connected stream socket.
/// @param cfg Additional configuration parameters for the protocol stack.
/// @param in Inputs for writing to the socket.
/// @param out Outputs from the socket.
/// @param trait Converts between the native and the wire format.
/// @relates length_prefix_framing
template <template <class> class Transport = stream_transport, class Socket,
class T, class Trait, class... TransportArgs>
error run_with_length_prefix_framing(multiplexer& mpx, Socket fd,
const settings& cfg,
async::consumer_resource<T> in,
async::producer_resource<T> out,
Trait trait, TransportArgs&&... args) {
using app_t = Transport<length_prefix_framing<message_flow_bridge<T, Trait>>>;
auto mgr = make_socket_manager<app_t>(fd, &mpx,
std::forward<TransportArgs>(args)...,
std::move(in), std::move(out),
std::move(trait));
return mgr->init(cfg);
}
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/make_counted.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
namespace caf::net {
template <class Transport>
endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx,
actor_system& sys, Transport trans) {
using impl = endpoint_manager_impl<Transport>;
return make_counted<impl>(mpx, sys, std::move(trans));
}
} // namespace caf::net
// 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/async/spsc_buffer.hpp"
#include "caf/net/consumer_adapter.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/producer_adapter.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/settings.hpp"
#include "caf/tag/message_oriented.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/no_auto_reading.hpp"
#include <utility>
namespace caf::net {
/// Translates between a message-oriented transport and data flows.
///
/// The trait class converts between the native and the wire format:
///
/// ~~~
/// struct my_trait {
/// bool convert(const T& value, byte_buffer& bytes);
/// bool convert(const_byte_span bytes, T& value);
/// };
/// ~~~
template <class T, class Trait, class Tag = tag::message_oriented>
class message_flow_bridge : public tag::no_auto_reading {
public:
using input_tag = Tag;
using buffer_type = async::spsc_buffer<T>;
using consumer_resource_t = async::consumer_resource<T>;
using producer_resource_t = async::producer_resource<T>;
message_flow_bridge(consumer_resource_t in_res, producer_resource_t out_res,
Trait trait)
: in_res_(std::move(in_res)),
trait_(std::move(trait)),
out_res_(std::move(out_res)) {
// nop
}
explicit message_flow_bridge(Trait trait) : trait_(std::move(trait)) {
// nop
}
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr, const settings& cfg) {
mgr_ = mgr;
if constexpr (caf::detail::has_init_v<Trait>) {
if (auto err = init_res(trait_.init(cfg)))
return err;
}
if (in_res_) {
in_ = consumer_adapter<buffer_type>::try_open(mgr, in_res_);
in_res_ = nullptr;
}
if (out_res_) {
out_ = producer_adapter<buffer_type>::try_open(mgr, out_res_);
out_res_ = nullptr;
}
if (!in_ && !out_)
return make_error(sec::cannot_open_resource,
"flow bridge cannot run without at least one resource");
return none;
}
template <class LowerLayerPtr>
bool write(LowerLayerPtr down, const T& item) {
if constexpr (std::is_same_v<Tag, tag::message_oriented>) {
down->begin_message();
auto& buf = down->message_buffer();
return trait_.convert(item, buf) && down->end_message();
} else {
static_assert(std::is_same_v<Tag, tag::mixed_message_oriented>);
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 = message_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(span<const T> items) {
CAF_ASSERT(items.size() == 1);
for (const auto& item : items) {
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 U = Tag, class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
return -1;
}
T 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 U = Tag, class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr down, byte_span buf) {
return consume(down, buf);
}
template <class U = Tag, class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, string_view buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
return -1;
}
T 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:
error init_res(error err) {
return err;
}
error init_res(consumer_resource_t in, producer_resource_t out) {
in_res_ = std::move(in);
out_res_ = std::move(out);
return caf::none;
}
error init_res(std::tuple<consumer_resource_t, producer_resource_t> in_out) {
auto& [in, out] = in_out;
return init_res(std::move(in), std::move(out));
}
error init_res(std::pair<consumer_resource_t, producer_resource_t> in_out) {
auto& [in, out] = in_out;
return init_res(std::move(in), std::move(out));
}
template <class R>
error init_res(expected<R> res) {
if (res)
return init_res(*res);
else
return std::move(res.error());
}
/// Points to the manager that runs this protocol stack.
net::socket_manager* mgr_ = nullptr;
/// Incoming messages, serialized to the socket.
consumer_adapter_ptr<buffer_type> in_;
/// Outgoing messages, deserialized from the socket.
producer_adapter_ptr<buffer_type> out_;
/// Converts between raw bytes and items.
Trait trait_;
/// Discarded after initialization.
consumer_resource_t in_res_;
/// Discarded after initialization.
producer_resource_t out_res_;
};
} // namespace caf::net
// 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a message-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_message() {
lptr_->begin_message(llptr_);
}
[[nodiscard]] auto& message_buffer() {
return lptr_->message_buffer(llptr_);
}
[[nodiscard]] bool end_message() {
return lptr_->end_message(llptr_);
}
template <class... Ts>
bool send_close_message(Ts&&... xs) {
return lptr_->send_close_message(llptr_, std::forward<Ts>(xs)...);
}
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_;
};
message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
message_oriented_layer_ptr(const 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_message_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
using result_t = message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
// 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 <chrono>
#include <set>
#include <string>
#include <thread>
#include "caf/actor_system.hpp"
#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 {
class CAF_NET_EXPORT middleman : public actor_system::module {
public:
// -- member types -----------------------------------------------------------
using module = actor_system::module;
using module_ptr = actor_system::module_ptr;
using middleman_backend_list = std::vector<middleman_backend_ptr>;
// -- static utility functions -----------------------------------------------
static void init_global_meta_objects();
// -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
~middleman() override;
// -- socket manager functions -----------------------------------------------
/// Creates a new acceptor that accepts incoming connections from @p sock and
/// creates socket managers using @p factory.
/// @param sock An accept socket in listening mode. For a TCP socket, this
/// socket must already listen to an address plus port.
/// @param factory A function object for creating socket managers that take
/// ownership of incoming connections.
/// @param limit The maximum number of connections that this acceptor should
/// establish or 0 for 'no limit'.
template <class Socket, class Factory>
auto make_acceptor(Socket sock, Factory factory, size_t limit = 0) {
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), limit);
} else {
using impl = connection_acceptor<Socket, Factory>;
auto ptr = make_socket_manager<impl>(std::move(sock), &mpx_, limit,
std::move(factory));
mpx_.init(ptr);
return ptr;
}
}
// -- interface functions ----------------------------------------------------
void start() override;
void stop() override;
void init(actor_system_config&) override;
id_t id() const override;
void* subtype_ptr() override;
// -- factory functions ------------------------------------------------------
template <class... Ts>
static module* make(actor_system& sys, detail::type_list<Ts...> token) {
std::unique_ptr<middleman> result{new middleman(sys)};
if (sizeof...(Ts) > 0) {
result->backends_.reserve(sizeof...(Ts));
create_backends(*result, token);
}
return result.release();
}
/// Adds module-specific options to the config before loading the module.
static void add_module_options(actor_system_config& cfg);
// -- remoting ---------------------------------------------------------------
expected<endpoint_manager_ptr> connect(const uri& locator);
// Publishes an actor.
template <class Handle>
void publish(Handle whom, const std::string& path) {
// TODO: Currently, we can't get the interface from the registry. Either we
// change that, or we need to associate the handle with the interface.
system().registry().put(path, whom);
}
/// Resolves a path to a remote actor.
void resolve(const uri& locator, const actor& listener);
template <class Handle = actor, class Duration = std::chrono::seconds>
expected<Handle>
remote_actor(const uri& locator, Duration timeout = std::chrono::seconds(5)) {
scoped_actor self{sys_};
resolve(locator, self);
Handle handle;
error err;
self->receive(
[&handle](strong_actor_ptr& ptr, const std::set<std::string>&) {
// TODO: This cast is not type-safe.
handle = actor_cast<Handle>(std::move(ptr));
},
[&err](const error& e) { err = e; },
after(timeout) >>
[&err] {
err = make_error(sec::runtime_error,
"manager did not respond with a proxy.");
});
if (err)
return err;
if (handle)
return handle;
return make_error(sec::runtime_error, "cast to handle-type failed");
}
// -- properties -------------------------------------------------------------
actor_system& system() {
return sys_;
}
const actor_system_config& config() const noexcept {
return sys_.config();
}
multiplexer& mpx() noexcept {
return mpx_;
}
const multiplexer& mpx() const noexcept {
return mpx_;
}
multiplexer* mpx_ptr() noexcept {
return &mpx_;
}
const multiplexer* mpx_ptr() const noexcept {
return &mpx_;
}
middleman_backend* backend(string_view scheme) const noexcept;
expected<uint16_t> port(string_view scheme) const;
private:
// -- utility functions ------------------------------------------------------
static void create_backends(middleman&, detail::type_list<>) {
// End of recursion.
}
template <class T, class... Ts>
static void create_backends(middleman& mm, detail::type_list<T, Ts...>) {
mm.backends_.emplace_back(new T(mm));
create_backends(mm, detail::type_list<Ts...>{});
}
// -- member variables -------------------------------------------------------
/// Points to the parent system.
actor_system& sys_;
/// Stores the global socket I/O multiplexer.
multiplexer mpx_;
/// Stores all available backends for managing peers.
middleman_backend_list backends_;
/// Runs the multiplexer's event loop
std::thread mpx_thread_;
};
} // namespace caf::net
// 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 <string>
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/proxy_registry.hpp"
namespace caf::net {
/// Technology-specific backend for connecting to and managing peer
/// connections.
/// @relates middleman
class CAF_NET_EXPORT middleman_backend : public proxy_registry::backend {
public:
// -- constructors, destructors, and assignment operators --------------------
middleman_backend(std::string id);
virtual ~middleman_backend();
// -- interface functions ----------------------------------------------------
/// Initializes the backend.
virtual error init() = 0;
/// @returns The endpoint manager for `peer` on success, `nullptr` otherwise.
virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Establishes a connection to a remote node.
virtual expected<endpoint_manager_ptr> get_or_connect(const uri& locator) = 0;
/// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0;
virtual void stop() = 0;
// -- properties -------------------------------------------------------------
const std::string& id() const noexcept {
return id_;
}
virtual uint16_t port() const = 0;
private:
/// Stores the technology-specific identifier.
std::string id_;
};
/// @relates middleman_backend
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
} // namespace caf::net
// 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a mixed-message-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 suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
void begin_binary_message() {
lptr_->begin_binary_message(llptr_);
}
auto& binary_message_buffer() {
return lptr_->binary_message_buffer(llptr_);
}
bool end_binary_message() {
return lptr_->end_binary_message(llptr_);
}
void begin_text_message() {
lptr_->begin_text_message(llptr_);
}
auto& text_message_buffer() {
return lptr_->text_message_buffer(llptr_);
}
bool end_text_message() {
return lptr_->end_text_message(llptr_);
}
template <class... Ts>
bool send_close_message(Ts&&... xs) {
return lptr_->send_close_message(llptr_, std::forward<Ts>(xs)...);
}
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
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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