Commit a2abe9a3 authored by Dominik Charousset's avatar Dominik Charousset

Clean up caf_net and integrate it to CMake

parent ad965456
......@@ -30,10 +30,11 @@ option(CAF_ENABLE_ACTOR_PROFILER "Enable experimental profiler API" OFF)
# -- CAF options that are on by default ----------------------------------------
option(CAF_ENABLE_EXAMPLES "Build small programs showcasing CAF features" ON)
option(CAF_ENABLE_IO_MODULE "Build networking I/O module" ON)
option(CAF_ENABLE_EXCEPTIONS "Build CAF with support for exceptions" ON)
option(CAF_ENABLE_IO_MODULE "Build legacy networking I/O module" ON)
option(CAF_ENABLE_NET_MODULE "Build networking I/O module" ON)
option(CAF_ENABLE_TESTING "Build unit test suites" ON)
option(CAF_ENABLE_TOOLS "Build utility programs such as caf-run" ON)
option(CAF_ENABLE_EXCEPTIONS "Build CAF with support for exceptions" ON)
# -- CAF options that depend on others -----------------------------------------
......@@ -368,6 +369,10 @@ endfunction()
add_subdirectory(libcaf_core)
if(CAF_ENABLE_NET_MODULE)
add_subdirectory(libcaf_net)
endif()
if(CAF_ENABLE_IO_MODULE)
add_subdirectory(libcaf_io)
endif()
......
......@@ -42,6 +42,24 @@ CAF_CORE_EXPORT void split(std::vector<std::string_view>& result,
std::string_view str, char delim,
bool keep_all = true);
// TODO: use the equivalent C++20 constructor instead, once available.
constexpr std::string_view make_string_view(const char* first,
const char* last) {
auto n = static_cast<size_t>(std::distance(first, last));
return std::string_view{first, n};
}
/// Drops any leading and trailing whitespaces from `str`.
CAF_CORE_EXPORT std::string_view trim(std::string_view str);
/// Checks whether two strings are equal when ignoring upper/lower case.
CAF_CORE_EXPORT bool icase_equal(std::string_view x, std::string_view y);
/// Splits a string by a separator into a head and a tail. If `sep` was not
/// found, the tail is empty.
CAF_CORE_EXPORT std::pair<std::string_view, std::string_view>
split_by(std::string_view str, std::string_view sep);
template <class InputIterator>
std::string
join(InputIterator first, InputIterator last, std::string_view glue) {
......
......@@ -51,6 +51,41 @@ void split(std::vector<std::string_view>& result, std::string_view str,
split(result, str, std::string_view{&delim, 1}, keep_all);
}
std::string_view trim(std::string_view str) {
auto non_whitespace = [](char c) { return !isspace(c); };
if (std::any_of(str.begin(), str.end(), non_whitespace)) {
while (str.front() == ' ')
str.remove_prefix(1);
while (str.back() == ' ')
str.remove_suffix(1);
} else {
str = std::string_view{};
}
return str;
}
bool icase_equal(std::string_view x, std::string_view y) {
if (x.size() != y.size()) {
return false;
} else {
for (size_t index = 0; index < x.size(); ++index)
if (tolower(x[index]) != tolower(y[index]))
return false;
return true;
}
}
std::pair<std::string_view, std::string_view> split_by(std::string_view str,
std::string_view sep) {
auto i = std::search(str.begin(), str.end(), sep.begin(), sep.end());
if (i != str.end()) {
return {make_string_view(str.begin(), i),
make_string_view(i + sep.size(), str.end())};
} else {
return {str, {}};
}
}
void replace_all(std::string& str, std::string_view what,
std::string_view with) {
// end(what) - 1 points to the null-terminator
......
......@@ -4,7 +4,7 @@ file(GLOB_RECURSE CAF_NET_HEADERS "caf/*.hpp")
# -- add targets ---------------------------------------------------------------
caf_incubator_add_component(
caf_add_component(
net
DEPENDENCIES
PUBLIC
......@@ -13,9 +13,6 @@ caf_incubator_add_component(
PRIVATE
CAF::internal
ENUM_TYPES
net.basp.connection_state
net.basp.ec
net.basp.message_type
net.http.method
net.http.status
net.operation
......@@ -24,13 +21,11 @@ caf_incubator_add_component(
HEADERS
${CAF_NET_HEADERS}
SOURCES
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp
src/detail/convert_ip_endpoint.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
......@@ -39,8 +34,6 @@ caf_incubator_add_component(
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
......@@ -51,35 +44,33 @@ caf_incubator_add_component(
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.convert_ip_endpoint
detail.rfc6455
header
ip
multiplexer
net.accept_socket
net.actor_shell
net.consumer_adapter
net.datagram_socket
net.http.server
net.ip
net.length_prefix_framing
net.multiplexer
net.network_socket
net.operation
net.pipe_socket
net.producer_adapter
net.socket
net.socket_guard
net.stream_socket
net.stream_transport
net.tcp_socket
net.typed_actor_shell
net.udp_datagram_socket
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)
net.web_socket.server)
if(CAF_INC_ENABLE_TESTING AND TARGET OpenSSL::SSL AND TARGET OpenSSL::Crypto)
caf_incubator_add_test_suites(caf-net-test net.openssl_transport)
......
......@@ -4,20 +4,19 @@
#pragma once
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/span.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;
......@@ -43,18 +42,18 @@ struct CAF_NET_EXPORT rfc6455 {
static void mask_data(uint32_t key, span<char> data);
static void mask_data(uint32_t key, span<byte> data);
static void mask_data(uint32_t key, byte_span data);
static void assemble_frame(uint32_t mask_key, span<const char> data,
binary_buffer& out);
byte_buffer& out);
static void assemble_frame(uint32_t mask_key, span<const byte> data,
binary_buffer& out);
static void assemble_frame(uint32_t mask_key, const_byte_span data,
byte_buffer& out);
static void assemble_frame(uint8_t opcode, uint32_t mask_key,
span<const byte> data, binary_buffer& out);
const_byte_span data, byte_buffer& out);
static ptrdiff_t decode_header(span<const byte> data, header& hdr);
static ptrdiff_t decode_header(const_byte_span 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/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/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
......@@ -5,7 +5,6 @@
#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"
......
......@@ -6,7 +6,9 @@
#include "caf/detail/net_export.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/variant.hpp"
#include <cstddef>
#include <variant>
namespace caf::net {
......@@ -24,7 +26,7 @@ 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
std::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 <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
......@@ -31,17 +31,11 @@ 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;
......@@ -58,24 +52,8 @@ 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);
......@@ -9,9 +9,9 @@
#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 <string_view>
#include <vector>
namespace caf::net::http {
......@@ -34,7 +34,7 @@ public:
return method_;
}
string_view path() const noexcept {
std::string_view path() const noexcept {
return uri_.path();
}
......@@ -42,11 +42,11 @@ public:
return uri_.query();
}
string_view fragment() const noexcept {
std::string_view fragment() const noexcept {
return uri_.fragment();
}
string_view version() const noexcept {
std::string_view version() const noexcept {
return version_;
}
......@@ -54,7 +54,7 @@ public:
return fields_;
}
string_view field(string_view key) const noexcept {
std::string_view field(std::string_view key) const noexcept {
if (auto i = fields_.find(key); i != fields_.end())
return i->second;
else
......@@ -62,9 +62,9 @@ public:
}
template <class T>
optional<T> field_as(string_view key) const noexcept {
std::optional<T> field_as(std::string_view key) const noexcept {
if (auto i = fields_.find(key); i != fields_.end()) {
caf::config_value val{to_string(i->second)};
caf::config_value val{std::string{i->second}};
if (auto res = caf::get_as<T>(val))
return std::move(*res);
else
......@@ -78,17 +78,17 @@ public:
return !raw_.empty();
}
std::pair<status, string_view> parse(string_view raw);
std::pair<status, std::string_view> parse(std::string_view raw);
bool chunked_transfer_encoding() const;
optional<size_t> content_length() const;
std::optional<size_t> content_length() const;
private:
std::vector<char> raw_;
http::method method_;
uri uri_;
string_view version_;
std::string_view version_;
header_fields_map fields_;
};
......
......@@ -5,11 +5,13 @@
#pragma once
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/string_view.hpp"
#include <string_view>
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>;
using header_fields_map
= detail::unordered_flat_map<std::string_view, std::string_view>;
} // namespace caf::net::http
......@@ -6,10 +6,10 @@
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/string_view.hpp"
#include <cstdint>
#include <string>
#include <string_view>
#include <type_traits>
namespace caf::net::http {
......@@ -55,7 +55,7 @@ CAF_NET_EXPORT std::string to_string(method);
CAF_NET_EXPORT std::string to_rfc_string(method x);
/// @relates method
CAF_NET_EXPORT bool from_string(string_view, method&);
CAF_NET_EXPORT bool from_string(std::string_view, method&);
/// @relates method
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<method>, method&);
......
......@@ -130,17 +130,17 @@ public:
auto& buf = down->output_buffer();
auto size = bytes.size();
detail::append_hex(buf, &size, sizeof(size));
buf.emplace_back(byte{'\r'});
buf.emplace_back(byte{'\n'});
buf.emplace_back(std::byte{'\r'});
buf.emplace_back(std::byte{'\n'});
buf.insert(buf.end(), bytes.begin(), bytes.end());
buf.emplace_back(byte{'\r'});
buf.emplace_back(byte{'\n'});
buf.emplace_back(std::byte{'\r'});
buf.emplace_back(std::byte{'\n'});
return down->end_output();
}
template <class LowerLayerPtr>
bool send_end_of_chunks(LowerLayerPtr down, context) {
string_view str = "0\r\n\r\n";
std::string_view str = "0\r\n\r\n";
auto bytes = as_bytes(make_span(str));
down->begin_output();
auto& buf = down->output_buffer();
......@@ -272,7 +272,8 @@ private:
}
template <class LowerLayerPtr>
void write_response(LowerLayerPtr down, status code, string_view content) {
void
write_response(LowerLayerPtr down, status code, std::string_view content) {
down->begin_output();
v1::write_response(code, "text/plain", content, down->output_buffer());
down->end_output();
......@@ -293,7 +294,7 @@ private:
// -- HTTP request processing ------------------------------------------------
template <class LowerLayerPtr>
bool handle_header(LowerLayerPtr down, string_view http) {
bool handle_header(LowerLayerPtr down, std::string_view http) {
// Parse the header and reject invalid inputs.
auto [code, msg] = hdr_.parse(http);
if (code != status::ok) {
......
......@@ -168,13 +168,13 @@ enum class status : uint16_t {
/// Returns the recommended response phrase to a status code.
/// @relates status
CAF_NET_EXPORT string_view phrase(status);
CAF_NET_EXPORT std::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&);
CAF_NET_EXPORT bool from_string(std::string_view, status&);
/// @relates status
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<status>, status&);
......
......@@ -8,16 +8,17 @@
#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 <string_view>
#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
/// remainder (`second`). Returns an empty `string_view` as `first` for
/// incomplete HTTP headers.
CAF_NET_EXPORT std::pair<string_view, byte_span> split_header(byte_span bytes);
CAF_NET_EXPORT std::pair<std::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,
......@@ -25,14 +26,15 @@ CAF_NET_EXPORT void write_header(status code, const header_fields_map& fields,
/// 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);
CAF_NET_EXPORT void write_response(status code, std::string_view content_type,
std::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);
CAF_NET_EXPORT void write_response(status code, std::string_view content_type,
std::string_view content,
const header_fields_map& fields,
byte_buffer& buf);
} // namespace caf::net::http::v1
......@@ -98,7 +98,7 @@ public:
/// 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,
header_fields_type fields, std::string_view content_type,
const_byte_span content) {
std::string len;
if (!content.empty()) {
......@@ -113,7 +113,7 @@ public:
/// 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_view content_type, const_byte_span content) {
std::string len;
header_fields_type fields;
if (!content.empty()) {
......@@ -125,14 +125,15 @@ public:
}
bool send_response(context_type context, status_code_type code,
header_fields_type fields, string_view content_type,
string_view content) {
header_fields_type fields, std::string_view content_type,
std::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) {
std::string_view content_type,
std::string_view content) {
return send_response(context, code, content_type,
as_bytes(make_span(content)));
}
......
......@@ -4,23 +4,24 @@
#pragma once
#include <string>
#include <vector>
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include <string>
#include <string_view>
#include <vector>
namespace caf::net::ip {
/// Returns all IP addresses of `host` (if any).
std::vector<ip_address> CAF_NET_EXPORT resolve(string_view host);
std::vector<ip_address> CAF_NET_EXPORT resolve(std::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);
std::vector<ip_address> CAF_NET_EXPORT local_addresses(std::string_view host);
/// Returns the IP addresses for a local endpoint address.
std::vector<ip_address> CAF_NET_EXPORT local_addresses(ip_address host);
......
......@@ -93,7 +93,7 @@ public:
down->begin_output();
auto& buf = down->output_buffer();
message_offset_ = buf.size();
buf.insert(buf.end(), 4, byte{0});
buf.insert(buf.end(), 4, std::byte{0});
}
template <class LowerLayerPtr>
......
// 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
......@@ -108,13 +108,10 @@ public:
// 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_next(const T& item) {
if (!bridge->write(down, item)) {
aborted = true;
return;
}
}
......@@ -194,7 +191,7 @@ public:
}
template <class U = Tag, class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, string_view buf) {
ptrdiff_t consume_text(LowerLayerPtr down, std::string_view buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
return -1;
......
......@@ -15,7 +15,6 @@
#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"
......@@ -30,8 +29,6 @@ public:
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();
......@@ -82,59 +79,11 @@ public:
// -- 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();
}
static module* make(actor_system& sys, detail::type_list<>);
/// 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() {
......@@ -161,23 +110,7 @@ public:
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.
......@@ -186,9 +119,6 @@ private:
/// 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_;
};
......
// 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
......@@ -142,13 +142,16 @@ public:
/// ready as a result.
bool poll_once(bool blocking);
/// Runs `poll_once(false)` until it returns `true`.`
void poll();
/// Applies all pending updates.
void apply_updates();
/// Sets the thread ID to `std::this_thread::id()`.
void set_thread_id();
/// Polls until no socket event handler remains.
/// Runs the multiplexer until no socket event handler remains active.
void run();
protected:
......
......@@ -4,15 +4,16 @@
#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"
#include <cstdint>
#include <string>
#include <string_view>
#include <type_traits>
namespace caf::net {
/// Values for representing bitmask of I/O operations.
......@@ -111,7 +112,7 @@ enum class operation {
CAF_NET_EXPORT std::string to_string(operation x);
/// @relates operation
CAF_NET_EXPORT bool from_string(string_view, operation&);
CAF_NET_EXPORT bool from_string(std::string_view, operation&);
/// @relates operation
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<operation>, operation&);
......
// 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"
#include "caf/net/fwd.hpp"
#include "caf/span.hpp"
namespace caf::net {
/// Implements an interface for packet writing in application-layers.
class CAF_NET_EXPORT packet_writer {
public:
virtual ~packet_writer();
/// Returns a buffer for writing header information.
virtual byte_buffer next_header_buffer() = 0;
/// Returns a buffer for writing payload content.
virtual byte_buffer next_payload_buffer() = 0;
/// Convenience function to write a packet consisting of multiple buffers.
/// @param buffers all buffers for the packet. The first buffer is a header
/// buffer, the other buffers are payload buffer.
/// @warning this function takes ownership of `buffers`.
template <class... Ts>
void write_packet(Ts&... buffers) {
byte_buffer* bufs[] = {&buffers...};
write_impl(make_span(bufs, sizeof...(Ts)));
}
protected:
/// Implementing function for `write_packet`.
/// @param buffers a `span` containing all buffers of a packet.
virtual void write_impl(span<byte_buffer*> buffers) = 0;
};
} // 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/packet_writer.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/span.hpp"
namespace caf::net {
/// Implements the interface for transport and application policies and
/// dispatches member functions either to `object` or `parent`.
template <class Object, class Parent>
class packet_writer_decorator final : public packet_writer {
public:
// -- member types -----------------------------------------------------------
using transport_type = typename Parent::transport_type;
using application_type = typename Parent::application_type;
// -- constructors, destructors, and assignment operators --------------------
packet_writer_decorator(Object& object, Parent& parent)
: object_(object), parent_(parent) {
// nop
}
// -- properties -------------------------------------------------------------
actor_system& system() {
return parent_.system();
}
transport_type& transport() {
return parent_.transport();
}
endpoint_manager& manager() {
return parent_.manager();
}
byte_buffer next_header_buffer() override {
return transport().next_header_buffer();
}
byte_buffer next_payload_buffer() override {
return transport().next_payload_buffer();
}
protected:
void write_impl(span<byte_buffer*> buffers) override {
parent_.write_packet(object_.id(), buffers);
}
private:
Object& object_;
Parent& parent_;
};
template <class Object, class Parent>
packet_writer_decorator<Object, Parent>
make_packet_writer_decorator(Object& object, Parent& parent) {
return {object, parent};
}
} // namespace caf::net
......@@ -36,13 +36,13 @@ expected<std::pair<pipe_socket, pipe_socket>> CAF_NET_EXPORT make_pipe();
/// @param buf Memory region for reading the message to send.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
ptrdiff_t CAF_NET_EXPORT write(pipe_socket x, span<const byte> buf);
ptrdiff_t CAF_NET_EXPORT write(pipe_socket x, const_byte_span buf);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param buf Memory region for storing the received bytes.
/// @returns The number of received bytes on success, otherwise an error code.
/// @relates pipe_socket
ptrdiff_t CAF_NET_EXPORT read(pipe_socket x, span<byte> buf);
ptrdiff_t CAF_NET_EXPORT read(pipe_socket x, byte_span buf);
} // namespace caf::net
......@@ -4,14 +4,14 @@
#pragma once
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp"
#include <array>
#include <cstddef>
#include <cstdint>
#include <mutex>
#include "caf/byte.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp"
namespace caf::net {
class pollset_updater : public socket_manager {
......@@ -20,7 +20,7 @@ public:
using super = socket_manager;
using msg_buf = std::array<byte, sizeof(intptr_t) + 1>;
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
enum class code : uint8_t {
register_reading,
......
......@@ -6,6 +6,7 @@
#include <cstddef>
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/network_socket.hpp"
......@@ -46,7 +47,7 @@ error CAF_NET_EXPORT nodelay(stream_socket x, bool new_value);
/// @relates stream_socket
/// @post Either the functions returned a non-negative integer or the caller can
/// retrieve the error code by calling `last_socket_error()`.
ptrdiff_t CAF_NET_EXPORT read(stream_socket x, span<byte> buf);
ptrdiff_t CAF_NET_EXPORT read(stream_socket x, byte_span buf);
/// Sends data to `x`.
/// @param x A connected endpoint.
......@@ -55,7 +56,7 @@ ptrdiff_t CAF_NET_EXPORT read(stream_socket x, span<byte> buf);
/// or -1 in case of an error.
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
ptrdiff_t CAF_NET_EXPORT write(stream_socket x, span<const byte> buf);
ptrdiff_t CAF_NET_EXPORT write(stream_socket x, const_byte_span buf);
/// Transmits data from `x` to its peer.
/// @param x A connected endpoint.
......@@ -66,6 +67,6 @@ ptrdiff_t CAF_NET_EXPORT write(stream_socket x, span<const byte> buf);
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10`
ptrdiff_t CAF_NET_EXPORT write(stream_socket x,
std::initializer_list<span<const byte>> bufs);
std::initializer_list<const_byte_span> bufs);
} // namespace caf::net
......@@ -29,12 +29,12 @@ namespace caf::net {
struct default_stream_transport_policy {
public:
/// Reads data from the socket into the buffer.
static ptrdiff_t read(stream_socket x, span<byte> buf) {
static ptrdiff_t read(stream_socket x, byte_span buf) {
return net::read(x, buf);
}
/// Writes data from the buffer to the socket.
static ptrdiff_t write(stream_socket x, span<const byte> buf) {
static ptrdiff_t write(stream_socket x, const_byte_span buf) {
return net::write(x, buf);
}
......
......@@ -30,7 +30,7 @@ enum class stream_transport_error {
CAF_NET_EXPORT std::string to_string(stream_transport_error);
/// @relates stream_transport_error
CAF_NET_EXPORT bool from_string(string_view, stream_transport_error&);
CAF_NET_EXPORT bool from_string(std::string_view, stream_transport_error&);
/// @relates stream_transport_error
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<stream_transport_error>,
......
// 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 <stdexcept>
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/net/host.hpp"
#include "caf/raise_error.hpp"
namespace {
struct CAF_NET_EXPORT host_fixture {
host_fixture() {
if (auto err = caf::net::this_host::startup())
CAF_RAISE_ERROR(std::logic_error, "this_host::startup failed");
}
~host_fixture() {
caf::net::this_host::cleanup();
}
};
} // namespace
// 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_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/overload.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Implements a base class for transports.
/// @tparam Transport The derived type of the transport implementation.
/// @tparam NextLayer The Following Layer. Either `transport_worker` or
/// `transport_worker_dispatcher`
/// @tparam Handle The type of the related socket_handle.
/// @tparam Application The type of the application used in this stack.
/// @tparam IdType The id type of the derived transport, must match the IdType
/// of `NextLayer`.
template <class Transport, class NextLayer, class Handle, class Application,
class IdType>
class transport_base {
public:
// -- member types -----------------------------------------------------------
using next_layer_type = NextLayer;
using handle_type = Handle;
using transport_type = Transport;
using application_type = Application;
using id_type = IdType;
using buffer_cache_type = std::vector<byte_buffer>;
// -- constructors, destructors, and assignment operators --------------------
transport_base(handle_type handle, application_type application)
: next_layer_(std::move(application)), handle_(handle), manager_(nullptr) {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns the socket_handle of this transport.
handle_type handle() const noexcept {
return handle_;
}
/// Returns a reference to the `actor_system` of this transport.
/// @pre `init` must be called before calling this getter.
actor_system& system() {
return manager().system();
}
/// Returns a reference to the application of this transport.
application_type& application() {
return next_layer_.application();
}
/// Returns a reference to this transport.
transport_type& transport() {
return *reinterpret_cast<transport_type*>(this);
}
/// Returns a reference to the `endpoint_manager` of this transport.
/// @pre `init` must be called before calling this getter.
endpoint_manager& manager() {
CAF_ASSERT(manager_);
return *manager_;
}
// -- transport member functions ---------------------------------------------
/// Initializes this transport.
/// @param parent The `endpoint_manager` that manages this transport.
/// @returns `error` on failure, none on success.
virtual error init(endpoint_manager& parent) {
CAF_LOG_TRACE("");
manager_ = &parent;
auto& cfg = system().config();
max_consecutive_reads_ = get_or(this->system().config(),
"caf.middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads);
auto max_header_bufs = get_or(cfg, "caf.middleman.max-header-buffers",
defaults::middleman::max_header_buffers);
header_bufs_.reserve(max_header_bufs);
auto max_payload_bufs = get_or(cfg, "caf.middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs);
if (auto err = next_layer_.init(*this))
return err;
return none;
}
/// Resolves a remote actor using `locator` and sends the resolved actor to
/// listener on success - an `error` otherwise.
/// @param locator The `uri` of the remote actor.
/// @param listener The `actor_handle` which the result should be sent to.
auto resolve(endpoint_manager&, const uri& locator, const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(locator) << CAF_ARG(listener));
auto f = detail::make_overload(
[&](auto& layer) -> decltype(layer.resolve(*this, locator, listener)) {
return layer.resolve(*this, locator, listener);
},
[&](auto& layer) -> decltype(layer.resolve(*this, locator.path(),
listener)) {
return layer.resolve(*this, locator.path(), listener);
});
f(next_layer_);
}
/// Gets called by an actor proxy after creation.
/// @param peer The `node_id` of the remote node.
/// @param id The id of the remote actor.
void new_proxy(endpoint_manager&, const node_id& peer, actor_id id) {
next_layer_.new_proxy(*this, peer, id);
}
/// Notifies the remote endpoint that the local actor is down.
/// @param peer The `node_id` of the remote endpoint.
/// @param id The `actor_id` of the remote actor.
/// @param reason The reason why the local actor has shut down.
void local_actor_down(endpoint_manager&, const node_id& peer, actor_id id,
error reason) {
next_layer_.local_actor_down(*this, peer, id, std::move(reason));
}
/// Callback for when an error occurs.
/// @param code The error code to handle.
void handle_error(sec code) {
next_layer_.handle_error(code);
}
// -- (pure) virtual functions -----------------------------------------------
/// Called by the endpoint manager when the transport can read data from its
/// socket.
virtual bool handle_read_event(endpoint_manager&) = 0;
/// Called by the endpoint manager when the transport can write data to its
/// socket.
virtual bool handle_write_event(endpoint_manager& parent) = 0;
/// Queues a packet scattered across multiple buffers to be sent via this
/// transport.
/// @param id The id of the destination endpoint.
/// @param buffers Pointers to the buffers that make up the packet content.
virtual void write_packet(id_type id, span<byte_buffer*> buffers) = 0;
// -- buffer management ------------------------------------------------------
/// Returns the next cached header buffer or creates a new one if no buffers
/// are cached.
byte_buffer next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
/// Returns the next cached payload buffer or creates a new one if no buffers
/// are cached.
byte_buffer next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
private:
// -- utility functions ------------------------------------------------------
static byte_buffer next_buffer_impl(buffer_cache_type& cache) {
if (cache.empty())
return {};
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
}
protected:
next_layer_type next_layer_;
handle_type handle_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
byte_buffer read_buf_;
endpoint_manager* manager_;
size_t max_consecutive_reads_;
};
} // 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/logger.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/unit.hpp"
namespace caf::net {
/// Implements a worker for transport protocols.
template <class Application, class IdType>
class transport_worker {
public:
// -- member types -----------------------------------------------------------
using id_type = IdType;
using application_type = Application;
// -- constructors, destructors, and assignment operators --------------------
explicit transport_worker(application_type application,
id_type id = id_type{})
: application_(std::move(application)), id_(std::move(id)) {
// nop
}
// -- properties -------------------------------------------------------------
application_type& application() noexcept {
return application_;
}
const application_type& application() const noexcept {
return application_;
}
const id_type& id() const noexcept {
return id_;
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent& parent) {
auto writer = make_packet_writer_decorator(*this, parent);
return application_.init(writer);
}
template <class Parent>
error handle_data(Parent& parent, span<const byte> data) {
auto writer = make_packet_writer_decorator(*this, parent);
return application_.handle_data(writer, data);
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto writer = make_packet_writer_decorator(*this, parent);
if (auto err = application_.write_message(writer, std::move(msg)))
CAF_LOG_ERROR("write_message failed: " << err);
}
template <class Parent>
void resolve(Parent& parent, string_view path, const actor& listener) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.resolve(writer, path, listener);
}
template <class Parent>
void new_proxy(Parent& parent, const node_id&, actor_id id) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.new_proxy(writer, id);
}
template <class Parent>
void
local_actor_down(Parent& parent, const node_id&, actor_id id, error reason) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.local_actor_down(writer, id, std::move(reason));
}
template <class Parent>
void timeout(Parent& parent, std::string tag, uint64_t id) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.timeout(writer, std::move(tag), id);
}
void handle_error(sec error) {
application_.handle_error(error);
}
private:
application_type application_;
id_type id_;
};
template <class Application, class IdType = unit_t>
using transport_worker_ptr
= std::shared_ptr<transport_worker<Application, IdType>>;
} // 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 <unordered_map>
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf::net {
/// Implements a dispatcher that dispatches between transport and workers.
template <class Factory, class IdType>
class transport_worker_dispatcher {
public:
// -- member types -----------------------------------------------------------
using id_type = IdType;
using factory_type = Factory;
using application_type = typename factory_type::application_type;
using worker_type = transport_worker<application_type, id_type>;
using worker_ptr = transport_worker_ptr<application_type, id_type>;
// -- constructors, destructors, and assignment operators --------------------
explicit transport_worker_dispatcher(factory_type factory)
: factory_(std::move(factory)) {
// nop
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent&) {
CAF_ASSERT(workers_by_id_.empty());
return none;
}
template <class Parent>
error handle_data(Parent& parent, span<const byte> data, id_type id) {
if (auto worker = find_worker(id))
return worker->handle_data(parent, data);
// TODO: Where to get node_id from here?
auto worker = add_new_worker(parent, node_id{}, id);
if (worker)
return (*worker)->handle_data(parent, data);
else
return std::move(worker.error());
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto receiver = msg->receiver;
if (!receiver)
return;
auto nid = receiver->node();
if (auto worker = find_worker(nid)) {
worker->write_message(parent, std::move(msg));
return;
}
// TODO: where to get id_type from here?
if (auto worker = add_new_worker(parent, nid, id_type{}))
(*worker)->write_message(parent, std::move(msg));
}
template <class Parent>
void resolve(Parent& parent, const uri& locator, const actor& listener) {
if (auto worker = find_worker(make_node_id(locator)))
worker->resolve(parent, locator.path(), listener);
else
anon_send(listener,
make_error(sec::runtime_error, "could not resolve node"));
}
template <class Parent>
void new_proxy(Parent& parent, const node_id& nid, actor_id id) {
if (auto worker = find_worker(nid))
worker->new_proxy(parent, nid, id);
}
template <class Parent>
void local_actor_down(Parent& parent, const node_id& nid, actor_id id,
error reason) {
if (auto worker = find_worker(nid))
worker->local_actor_down(parent, nid, id, std::move(reason));
}
void handle_error(sec error) {
for (const auto& p : workers_by_id_) {
auto worker = p.second;
worker->handle_error(error);
}
}
template <class Parent>
expected<worker_ptr>
add_new_worker(Parent& parent, node_id node, id_type id) {
CAF_LOG_TRACE(CAF_ARG(node) << CAF_ARG(id));
auto application = factory_.make();
auto worker = std::make_shared<worker_type>(std::move(application), id);
if (auto err = worker->init(parent))
return err;
workers_by_id_.emplace(std::move(id), worker);
workers_by_node_.emplace(std::move(node), worker);
return worker;
}
private:
worker_ptr find_worker(const node_id& nid) {
return find_worker_impl(workers_by_node_, nid);
}
worker_ptr find_worker(const id_type& id) {
return find_worker_impl(workers_by_id_, id);
}
template <class Key>
worker_ptr find_worker_impl(const std::unordered_map<Key, worker_ptr>& map,
const Key& key) {
if (map.count(key) == 0) {
CAF_LOG_DEBUG("could not find worker: " << CAF_ARG(key));
return nullptr;
}
return map.at(key);
}
// -- worker lookups ---------------------------------------------------------
std::unordered_map<id_type, worker_ptr> workers_by_id_;
std::unordered_map<node_id, worker_ptr> workers_by_node_;
std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_;
factory_type factory_;
};
} // namespace caf::net
......@@ -4,10 +4,13 @@
#pragma once
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/network_socket.hpp"
#include <vector>
namespace caf::net {
/// A datagram-oriented network communication endpoint for bidirectional
......@@ -40,8 +43,8 @@ error CAF_NET_EXPORT allow_connreset(udp_datagram_socket x, bool new_value);
/// @relates udp_datagram_socket
/// @post buf was modified and the resulting integer represents the length of
/// the received datagram, even if it did not fit into the given buffer.
variant<std::pair<size_t, ip_endpoint>, sec>
CAF_NET_EXPORT read(udp_datagram_socket x, span<byte> buf);
std::variant<std::pair<size_t, ip_endpoint>, sec>
CAF_NET_EXPORT read(udp_datagram_socket x, byte_span buf);
/// Sends the content of `bufs` as a datagram to the endpoint `ep` on socket
/// `x`.
......@@ -52,9 +55,9 @@ variant<std::pair<size_t, ip_endpoint>, sec>
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
/// @pre `bufs.size() < 10`
variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
span<byte_buffer*> bufs,
ip_endpoint ep);
std::variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
span<byte_buffer*> bufs,
ip_endpoint ep);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// @param x The UDP socket for sending datagrams.
......@@ -62,13 +65,14 @@ variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
/// @param ep The enpoint to send the datagram to.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
span<const byte> buf, ip_endpoint ep);
std::variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
const_byte_span buf,
ip_endpoint ep);
/// Converts the result from I/O operation on a ::udp_datagram_socket to either
/// an error code or a non-zero positive integer.
/// @relates udp_datagram_socket
variant<size_t, sec> CAF_NET_EXPORT
std::variant<size_t, sec> CAF_NET_EXPORT
check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res);
} // namespace caf::net
......@@ -143,7 +143,7 @@ private:
// -- HTTP response processing -----------------------------------------------
template <class LowerLayerPtr>
bool handle_header(LowerLayerPtr down, string_view http) {
bool handle_header(LowerLayerPtr down, std::string_view http) {
CAF_ASSERT(handshake_ != nullptr);
auto http_ok = handshake_->is_valid_http_1_response(http);
handshake_.reset();
......
......@@ -91,7 +91,7 @@ public:
}
template <class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, caf::string_view text) {
ptrdiff_t consume_text(LowerLayerPtr down, std::string_view text) {
reader_msgput_type msg;
if (reader_.deserialize_text(text, msg)) {
if (reader_output_->push(std::move(msg)) == 0)
......
......@@ -27,7 +27,7 @@ class framing {
public:
// -- member types -----------------------------------------------------------
using binary_buffer = std::vector<byte>;
using binary_buffer = std::vector<std::byte>;
using text_buffer = std::vector<char>;
......@@ -131,7 +131,8 @@ public:
}
template <class LowerLayerPtr>
bool send_close_message(LowerLayerPtr down, status code, string_view desc) {
bool
send_close_message(LowerLayerPtr down, status code, std::string_view desc) {
ship_close(down, static_cast<uint16_t>(code), desc);
return true;
}
......@@ -282,8 +283,8 @@ private:
bool handle(LowerLayerPtr down, uint8_t opcode, byte_span payload) {
switch (opcode) {
case detail::rfc6455::text_frame: {
string_view text{reinterpret_cast<const char*>(payload.data()),
payload.size()};
std::string_view text{reinterpret_cast<const char*>(payload.data()),
payload.size()};
return upper_layer_.consume_text(this_layer_ptr(down), text) >= 0;
}
case detail::rfc6455::binary_frame:
......@@ -317,14 +318,14 @@ private:
}
template <class LowerLayerPtr>
void ship_close(LowerLayerPtr down, uint16_t code, string_view msg) {
void ship_close(LowerLayerPtr down, uint16_t code, std::string_view msg) {
uint32_t mask_key = 0;
std::vector<byte> payload;
byte_buffer payload;
payload.reserve(msg.size() + 2);
payload.push_back(static_cast<byte>((code & 0xFF00) >> 8));
payload.push_back(static_cast<byte>(code & 0x00FF));
payload.push_back(static_cast<std::byte>((code & 0xFF00) >> 8));
payload.push_back(static_cast<std::byte>(code & 0x00FF));
for (auto c : msg)
payload.push_back(static_cast<byte>(c));
payload.push_back(static_cast<std::byte>(c));
if (mask_outgoing_frames) {
mask_key = static_cast<uint32_t>(rng_());
detail::rfc6455::mask_data(mask_key, payload);
......@@ -338,10 +339,10 @@ private:
template <class LowerLayerPtr>
void ship_close(LowerLayerPtr down) {
uint32_t mask_key = 0;
byte payload[] = {
byte{0x03}, byte{0xE8}, // Error code 1000: normal close.
byte{'E'}, byte{'O'}, byte{'F'}, // "EOF" string as goodbye message.
};
std::byte payload[] = {// Error code 1000: normal close.
std::byte{0x03}, std::byte{0xE8},
// "EOF" string as goodbye message.
std::byte{'E'}, std::byte{'O'}, std::byte{'F'}};
if (mask_outgoing_frames) {
mask_key = static_cast<uint32_t>(rng_());
detail::rfc6455::mask_data(mask_key, payload);
......
......@@ -4,13 +4,13 @@
#pragma once
#include <string>
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/dictionary.hpp"
#include "caf/string_view.hpp"
#include <cstddef>
#include <string>
#include <string_view>
namespace caf::net::web_socket {
......@@ -20,7 +20,7 @@ class CAF_NET_EXPORT handshake {
public:
// -- member types -----------------------------------------------------------
using key_type = std::array<byte, 16>;
using key_type = std::array<std::byte, 16>;
// -- constants --------------------------------------------------------------
......@@ -53,7 +53,7 @@ public:
}
/// Tries to assign a key from base64 input.
bool assign_key(string_view base64_key);
bool assign_key(std::string_view base64_key);
/// Returns the key for the response message, i.e., what the server puts into
/// the HTTP field `Sec-WebSocket-Accept`.
......@@ -131,12 +131,12 @@ public:
/// - An `Upgrade` field with the value `websocket`.
/// - A `Connection` field with the value `Upgrade`.
/// - A `Sec-WebSocket-Accept` field with the value `response_key()`.
bool is_valid_http_1_response(string_view http_response) const;
bool is_valid_http_1_response(std::string_view http_response) const;
private:
// -- utility ----------------------------------------------------------------
string_view lookup(string_view field_name) const noexcept;
std::string_view lookup(std::string_view field_name) const noexcept;
// -- member variables -------------------------------------------------------
......
......@@ -22,6 +22,7 @@
#include "caf/net/web_socket/status.hpp"
#include "caf/pec.hpp"
#include "caf/settings.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
......@@ -96,7 +97,7 @@ public:
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span input, byte_span delta) {
using namespace caf::literals;
using namespace std::literals;
CAF_LOG_TRACE(CAF_ARG2("socket", down->handle().id)
<< CAF_ARG2("bytes", input.size()));
// Short circuit to the framing layer after the handshake completed.
......@@ -109,8 +110,8 @@ public:
if (input.size() >= handshake::max_http_size) {
down->begin_output();
http::v1::write_response(http::status::request_header_fields_too_large,
"text/plain"_sv,
"Header exceeds maximum size."_sv,
"text/plain"sv,
"Header exceeds maximum size."sv,
down->output_buffer());
down->end_output();
auto err = make_error(pec::too_many_characters,
......@@ -145,14 +146,15 @@ private:
// -- HTTP request processing ------------------------------------------------
template <class LowerLayerPtr>
void write_response(LowerLayerPtr down, http::status code, string_view msg) {
void
write_response(LowerLayerPtr down, http::status code, std::string_view msg) {
down->begin_output();
http::v1::write_response(code, "text/plain", msg, down->output_buffer());
down->end_output();
}
template <class LowerLayerPtr>
bool handle_header(LowerLayerPtr down, string_view http) {
bool handle_header(LowerLayerPtr down, std::string_view http) {
using namespace std::literals;
// Parse the header and reject invalid inputs.
http::header hdr;
......@@ -183,14 +185,14 @@ private:
// Store the request information in the settings for the upper layer.
auto& ws = cfg_["web-socket"].as_dictionary();
put(ws, "method", to_rfc_string(hdr.method()));
put(ws, "path", to_string(hdr.path()));
put(ws, "path", std::string{hdr.path()});
put(ws, "query", hdr.query());
put(ws, "fragment", hdr.fragment());
put(ws, "http-version", hdr.version());
if (!hdr.fields().empty()) {
auto& fields = ws["fields"].as_dictionary();
for (auto& [key, val] : hdr.fields())
put(fields, to_string(key), to_string(val));
put(fields, std::string{key}, std::string{val});
}
// Try to initialize the upper layer.
if (auto err = upper_layer_.init(owner_, down, cfg_)) {
......@@ -212,18 +214,18 @@ private:
}
template <class F>
void for_each_line(string_view input, F&& f) {
constexpr string_view eol{"\r\n"};
void for_each_line(std::string_view input, F&& f) {
constexpr std::string_view eol{"\r\n"};
for (auto pos = input.begin();;) {
auto line_end = std::search(pos, input.end(), eol.begin(), eol.end());
if (line_end == input.end())
return;
f(string_view{pos, line_end});
f(make_string_view(pos, line_end));
pos = line_end + eol.size();
}
}
static string_view trim(string_view str) {
static std::string_view trim(std::string_view str) {
str.remove_prefix(std::min(str.find_first_not_of(' '), str.size()));
auto trim_pos = str.find_last_not_of(' ');
if (trim_pos != str.npos)
......@@ -231,24 +233,6 @@ private:
return str;
}
/// Splits `str` at the first occurrence of `sep` into the head and the
/// remainder (excluding the separator).
static std::pair<string_view, string_view> split(string_view str,
string_view sep) {
auto i = std::search(str.begin(), str.end(), sep.begin(), sep.end());
if (i != str.end())
return {{str.begin(), i}, {i + sep.size(), str.end()}};
return {{str}, {}};
}
/// Convenience function for splitting twice.
static std::tuple<string_view, string_view, string_view>
split2(string_view str, string_view sep) {
auto [first, r1] = split(str, sep);
auto [second, third] = split(r1, sep);
return {first, second, third};
}
/// Stores whether the WebSocket handshake completed successfully.
bool handshake_complete_ = false;
......@@ -373,7 +357,7 @@ public:
return decorated().convert(bytes, x);
}
bool convert(string_view text, value_type& x) {
bool convert(std::string_view text, value_type& x) {
return decorated().convert(text, x);
}
......
......@@ -8,6 +8,8 @@
#include "caf/detail/net_export.hpp"
#include <cstdint>
#include <string>
#include <string_view>
#include <type_traits>
namespace caf::net::web_socket {
......@@ -83,7 +85,7 @@ enum class status : uint16_t {
CAF_NET_EXPORT std::string to_string(status);
/// @relates status
CAF_NET_EXPORT bool from_string(string_view, status&);
CAF_NET_EXPORT bool from_string(std::string_view, status&);
/// @relates status
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<status>, status&);
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/actor_system.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
namespace caf::net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), dst_(std::move(dst)) {
CAF_ASSERT(dst_ != nullptr);
dst_->enqueue_event(node(), id());
}
actor_proxy_impl::~actor_proxy_impl() {
// nop
}
void actor_proxy_impl::enqueue(mailbox_element_ptr msg, execution_unit*) {
CAF_PUSH_AID(0);
CAF_ASSERT(msg != nullptr);
CAF_LOG_SEND_EVENT(msg);
dst_->enqueue(std::move(msg), ctrl());
}
void actor_proxy_impl::kill_proxy(execution_unit* ctx, error rsn) {
cleanup(std::move(rsn), ctx);
}
} // namespace caf::net
......@@ -8,6 +8,8 @@
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp"
#include <variant>
namespace caf::net {
#ifdef CAF_WINDOWS
......@@ -33,7 +35,7 @@ error allow_connreset(datagram_socket x, bool) {
#endif // CAF_WINDOWS
variant<size_t, sec>
std::variant<size_t, sec>
check_datagram_socket_io_res(std::make_signed<size_t>::type res) {
if (res < 0) {
auto code = last_socket_error();
......
......@@ -5,7 +5,6 @@
#include "caf/detail/rfc6455.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
#include <cstring>
......@@ -15,9 +14,9 @@ void rfc6455::mask_data(uint32_t key, span<char> data) {
mask_data(key, as_writable_bytes(data));
}
void rfc6455::mask_data(uint32_t key, span<byte> data) {
void rfc6455::mask_data(uint32_t key, byte_span data) {
auto no_key = to_network_order(key);
byte arr[4];
std::byte arr[4];
memcpy(arr, &no_key, 4);
size_t i = 0;
for (auto& x : data) {
......@@ -27,44 +26,44 @@ void rfc6455::mask_data(uint32_t key, span<byte> data) {
}
void rfc6455::assemble_frame(uint32_t mask_key, span<const char> data,
binary_buffer& out) {
byte_buffer& out) {
assemble_frame(text_frame, mask_key, as_bytes(data), out);
}
void rfc6455::assemble_frame(uint32_t mask_key, span<const byte> data,
binary_buffer& out) {
void rfc6455::assemble_frame(uint32_t mask_key, const_byte_span data,
byte_buffer& out) {
assemble_frame(binary_frame, mask_key, data, out);
}
void rfc6455::assemble_frame(uint8_t opcode, uint32_t mask_key,
span<const byte> data, binary_buffer& out) {
const_byte_span data, byte_buffer& out) {
// First 8 bits: FIN flag + opcode (we never fragment frames).
out.push_back(byte{static_cast<uint8_t>(0x80 | opcode)});
out.push_back(std::byte{static_cast<uint8_t>(0x80 | opcode)});
// Mask flag + payload length (7 bits, 7+16 bits, or 7+64 bits)
auto mask_bit = byte{static_cast<uint8_t>(mask_key == 0 ? 0x00 : 0x80)};
auto mask_bit = std::byte{static_cast<uint8_t>(mask_key == 0 ? 0x00 : 0x80)};
if (data.size() < 126) {
auto len = static_cast<uint8_t>(data.size());
out.push_back(mask_bit | byte{len});
out.push_back(mask_bit | std::byte{len});
} else if (data.size() < std::numeric_limits<uint16_t>::max()) {
auto len = static_cast<uint16_t>(data.size());
auto no_len = to_network_order(len);
byte len_data[2];
std::byte len_data[2];
memcpy(len_data, &no_len, 2);
out.push_back(mask_bit | byte{126});
out.push_back(mask_bit | std::byte{126});
for (auto x : len_data)
out.push_back(x);
} else {
auto len = static_cast<uint64_t>(data.size());
auto no_len = to_network_order(len);
byte len_data[8];
std::byte len_data[8];
memcpy(len_data, &no_len, 8);
out.push_back(mask_bit | byte{127});
out.push_back(mask_bit | std::byte{127});
out.insert(out.end(), len_data, len_data + 8);
}
// Masking key: 0 or 4 bytes.
if (mask_key != 0) {
auto no_key = to_network_order(mask_key);
byte key_data[4];
std::byte key_data[4];
memcpy(key_data, &no_key, 4);
out.insert(out.end(), key_data, key_data + 4);
}
......@@ -72,11 +71,11 @@ void rfc6455::assemble_frame(uint8_t opcode, uint32_t mask_key,
out.insert(out.end(), data.begin(), data.end());
}
ptrdiff_t rfc6455::decode_header(span<const byte> data, header& hdr) {
ptrdiff_t rfc6455::decode_header(const_byte_span data, header& hdr) {
if (data.size() < 2)
return 0;
auto byte1 = to_integer<uint8_t>(data[0]);
auto byte2 = to_integer<uint8_t>(data[1]);
auto byte1 = std::to_integer<uint8_t>(data[0]);
auto byte2 = std::to_integer<uint8_t>(data[1]);
// Fetch FIN flag and opcode.
hdr.fin = (byte1 & 0x80) != 0;
hdr.opcode = byte1 & 0x0F;
......@@ -96,7 +95,7 @@ ptrdiff_t rfc6455::decode_header(span<const byte> data, header& hdr) {
if (data.size() < header_length)
return 0;
// Start decoding remaining header bytes.
const byte* p = data.data() + 2;
const std::byte* p = data.data() + 2;
// Fetch payload size
if (len_field == 126) {
uint16_t no_len;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/endpoint_manager.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys)
: super(handle, parent), sys_(sys), queue_(unit, unit, unit) {
queue_.try_block();
}
endpoint_manager::~endpoint_manager() {
// nop
}
// -- properties ---------------------------------------------------------------
const actor_system_config& endpoint_manager::config() const noexcept {
return sys_.config();
}
// -- queue access -------------------------------------------------------------
bool endpoint_manager::at_end_of_message_queue() {
return queue_.empty() && queue_.try_block();
}
endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
if (queue_.blocked())
return nullptr;
queue_.fetch_more();
auto& q = std::get<1>(queue_.queue().queues());
auto ts = q.next_task_size();
if (ts == 0)
return nullptr;
q.inc_deficit(ts);
auto result = q.next();
if (queue_.empty())
queue_.try_block();
return result;
}
// -- event management ---------------------------------------------------------
void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result;
using event_type = endpoint_manager_queue::event;
auto ptr = new event_type(std::move(locator), listener);
if (!enqueue(ptr))
anon_send(listener, resolve_atom_v, make_error(sec::request_receiver_down));
}
void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver) {
using message_type = endpoint_manager_queue::message;
auto ptr = new message_type(std::move(msg), std::move(receiver));
enqueue(ptr);
}
bool endpoint_manager::enqueue(endpoint_manager_queue::element* ptr) {
switch (queue_.push_back(ptr)) {
case intrusive::inbox_result::success:
return true;
case intrusive::inbox_result::unblocked_reader: {
auto mpx = parent_.lock();
if (mpx) {
mpx->register_writing(this);
return true;
}
return false;
}
default:
return false;
}
}
} // 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.
#include "caf/net/basp/header.hpp"
#include <cstring>
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
namespace caf::net::basp {
namespace {
void to_bytes_impl(const header& x, byte* ptr) {
*ptr = static_cast<byte>(x.type);
auto payload_len = detail::to_network_order(x.payload_len);
memcpy(ptr + 1, &payload_len, sizeof(payload_len));
auto operation_data = detail::to_network_order(x.operation_data);
memcpy(ptr + 5, &operation_data, sizeof(operation_data));
}
} // namespace
int header::compare(header other) const noexcept {
auto x = to_bytes(*this);
auto y = to_bytes(other);
return memcmp(x.data(), y.data(), header_size);
}
header header::from_bytes(span<const byte> bytes) {
CAF_ASSERT(bytes.size() >= header_size);
header result;
auto ptr = bytes.data();
result.type = *reinterpret_cast<const message_type*>(ptr);
uint32_t payload_len = 0;
memcpy(&payload_len, ptr + 1, 4);
result.payload_len = detail::from_network_order(payload_len);
uint64_t operation_data;
memcpy(&operation_data, ptr + 5, 8);
result.operation_data = detail::from_network_order(operation_data);
return result;
}
std::array<byte, header_size> to_bytes(header x) {
std::array<byte, header_size> result{};
to_bytes_impl(x, result.data());
return result;
}
void to_bytes(header x, byte_buffer& buf) {
buf.resize(header_size);
to_bytes_impl(x, buf.data());
}
} // namespace caf::net::basp
......@@ -4,11 +4,6 @@
#include "caf/net/ip.hpp"
#include <cstddef>
#include <string>
#include <utility>
#include <vector>
#include "caf/config.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
......@@ -17,7 +12,12 @@
#include "caf/ipv4_address.hpp"
#include "caf/logger.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/string_view.hpp"
#include <cstddef>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
#ifdef CAF_WINDOWS
# ifndef _WIN32_WINNT
......@@ -41,8 +41,8 @@ namespace caf::net::ip {
namespace {
// Dummy port to resolve empty string with getaddrinfo.
constexpr string_view dummy_port = "42";
constexpr string_view localhost = "localhost";
constexpr std::string_view dummy_port = "42";
constexpr std::string_view localhost = "localhost";
void* fetch_in_addr(int family, sockaddr* addr) {
if (family == AF_INET)
......@@ -142,7 +142,7 @@ void for_each_adapter(F f, bool is_link_local = false) {
} // namespace
std::vector<ip_address> resolve(string_view host) {
std::vector<ip_address> resolve(std::string_view host) {
addrinfo hint;
memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_STREAM;
......@@ -180,21 +180,21 @@ std::vector<ip_address> resolve(ip_address host) {
return resolve(to_string(host));
}
std::vector<ip_address> local_addresses(string_view host) {
std::vector<ip_address> local_addresses(std::string_view host) {
ip_address host_ip;
std::vector<ip_address> results;
if (host.empty()) {
for_each_adapter(
[&](string_view, ip_address ip) { results.push_back(ip); });
[&](std::string_view, ip_address ip) { results.push_back(ip); });
} else if (host == localhost) {
auto v6_local = ip_address{{0}, {0x1}};
auto v4_local = ip_address{make_ipv4_address(127, 0, 0, 1)};
for_each_adapter([&](string_view, ip_address ip) {
for_each_adapter([&](std::string_view, ip_address ip) {
if (ip == v4_local || ip == v6_local)
results.push_back(ip);
});
} else if (auto err = parse(host, host_ip)) {
for_each_adapter([&](string_view iface, ip_address ip) {
for_each_adapter([&](std::string_view iface, ip_address ip) {
if (iface == host)
results.push_back(ip);
});
......@@ -215,7 +215,7 @@ std::vector<ip_address> local_addresses(ip_address host) {
auto is_link_local = ll_prefix.contains(host);
std::vector<ip_address> results;
for_each_adapter(
[&](string_view, ip_address ip) {
[&](std::string_view, ip_address ip) {
if (host == ip)
results.push_back(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.
#include "caf/net/basp/message_queue.hpp"
#include <iterator>
namespace caf::net::basp {
message_queue::message_queue() : next_id(0), next_undelivered(0) {
// nop
}
void message_queue::push(execution_unit* ctx, uint64_t id,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
std::unique_lock<std::mutex> guard{lock};
CAF_ASSERT(id >= next_undelivered);
CAF_ASSERT(id < next_id);
auto first = pending.begin();
auto last = pending.end();
if (id == next_undelivered) {
// Dispatch current head.
if (receiver != nullptr)
receiver->enqueue(std::move(content), ctx);
auto next = id + 1;
// Check whether we can deliver more.
if (first == last || first->id != next) {
next_undelivered = next;
CAF_ASSERT(next_undelivered <= next_id);
return;
}
// Deliver everything until reaching a non-consecutive ID or the end.
auto i = first;
for (; i != last && i->id == next; ++i, ++next)
if (i->receiver != nullptr)
i->receiver->enqueue(std::move(i->content), ctx);
next_undelivered = next;
pending.erase(first, i);
CAF_ASSERT(next_undelivered <= next_id);
return;
}
// Get the insertion point.
auto pred = [&](const actor_msg& x) { return x.id >= id; };
pending.emplace(std::find_if(first, last, pred),
actor_msg{id, std::move(receiver), std::move(content)});
}
void message_queue::drop(execution_unit* ctx, uint64_t id) {
push(ctx, id, nullptr, nullptr);
}
uint64_t message_queue::new_id() {
std::unique_lock<std::mutex> guard{lock};
return next_id++;
}
} // namespace caf::net::basp
......@@ -17,7 +17,6 @@
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
#include <algorithm>
#include <optional>
......@@ -308,7 +307,7 @@ bool multiplexer::poll_once(bool blocking) {
// Must not happen.
auto int_code = static_cast<int>(code);
auto msg = std::generic_category().message(int_code);
string_view prefix = "poll() failed: ";
std::string_view prefix = "poll() failed: ";
msg.insert(msg.begin(), prefix.begin(), prefix.end());
CAF_CRITICAL(msg.c_str());
}
......@@ -317,6 +316,11 @@ bool multiplexer::poll_once(bool blocking) {
}
}
void multiplexer::poll() {
while (poll_once(false))
; // repeat
}
void multiplexer::apply_updates() {
CAF_LOG_DEBUG("apply" << updates_.size() << "updates");
if (!updates_.empty()) {
......@@ -478,9 +482,10 @@ multiplexer::update_for(const socket_manager_ptr& mgr) {
template <class T>
void multiplexer::write_to_pipe(uint8_t opcode, T* ptr) {
pollset_updater::msg_buf buf;
if (ptr)
if (ptr) {
intrusive_ptr_add_ref(ptr);
buf[0] = static_cast<byte>(opcode);
}
buf[0] = static_cast<std::byte>(opcode);
auto value = reinterpret_cast<intptr_t>(ptr);
memcpy(buf.data() + 1, &value, sizeof(intptr_t));
ptrdiff_t res = -1;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/backend/tcp.hpp"
#include <mutex>
#include <string>
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/application_factory.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.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::backend {
tcp::tcp(middleman& mm)
: middleman_backend("tcp"), mm_(mm), proxies_(mm.system(), *this) {
// nop
}
tcp::~tcp() {
// nop
}
error tcp::init() {
uint16_t conf_port = get_or<uint16_t>(mm_.system().config(),
"caf.middleman.tcp-port",
defaults::middleman::tcp_port);
ip_endpoint ep;
auto local_address = std::string("[::]:") + std::to_string(conf_port);
if (auto err = detail::parse(local_address, ep))
return err;
auto acceptor = make_tcp_accept_socket(ep, true);
if (!acceptor)
return acceptor.error();
auto acc_guard = make_socket_guard(*acceptor);
if (auto err = nonblocking(acc_guard.socket(), true))
return err;
auto port = local_port(*acceptor);
if (!port)
return port.error();
listening_port_ = *port;
CAF_LOG_INFO("doorman spawned on " << CAF_ARG(*port));
auto doorman_uri = make_uri("tcp://doorman");
if (!doorman_uri)
return doorman_uri.error();
auto& mpx = mm_.mpx();
auto mgr = make_endpoint_manager(
mpx, mm_.system(),
doorman{acc_guard.release(), basp::application_factory{proxies_}});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err);
return err;
}
return none;
}
void tcp::stop() {
for (const auto& p : peers_)
proxies_.erase(p.first);
peers_.clear();
}
expected<endpoint_manager_ptr> tcp::get_or_connect(const uri& locator) {
if (auto auth = locator.authority_only()) {
auto id = make_node_id(*auth);
if (auto ptr = peer(id))
return ptr;
auto host = locator.authority().host;
if (auto hostname = get_if<std::string>(&host)) {
for (const auto& addr : ip::resolve(*hostname)) {
ip_endpoint ep{addr, locator.authority().port};
auto sock = make_connected_tcp_stream_socket(ep);
if (!sock)
continue;
else
return emplace(id, *sock);
}
}
}
return sec::cannot_connect_to_node;
}
endpoint_manager_ptr tcp::peer(const node_id& id) {
return get_peer(id);
}
void tcp::resolve(const uri& locator, const actor& listener) {
if (auto p = get_or_connect(locator))
(*p)->resolve(locator, listener);
else
anon_send(listener, p.error());
}
strong_actor_ptr tcp::make_proxy(node_id nid, actor_id aid) {
using impl_type = actor_proxy_impl;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &mm_.system(), cfg,
peer(nid));
}
void tcp::set_last_hop(node_id*) {
// nop
}
uint16_t tcp::port() const noexcept {
return listening_port_;
}
endpoint_manager_ptr tcp::get_peer(const node_id& id) {
const std::lock_guard<std::mutex> lock(lock_);
auto i = peers_.find(id);
if (i != peers_.end())
return i->second;
return nullptr;
}
} // 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.
#include "caf/net/backend/test.hpp"
#include "caf/expected.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/raise_error.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf::net::backend {
test::test(middleman& mm)
: middleman_backend("test"), mm_(mm), proxies_(mm.system(), *this) {
// nop
}
test::~test() {
// nop
}
error test::init() {
return none;
}
void test::stop() {
for (const auto& p : peers_)
proxies_.erase(p.first);
peers_.clear();
}
endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second;
}
expected<endpoint_manager_ptr> test::get_or_connect(const uri& locator) {
if (auto ptr = peer(make_node_id(*locator.authority_only())))
return ptr;
return make_error(sec::runtime_error,
"connecting not implemented in test backend");
}
void test::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only();
if (id)
peer(make_node_id(*id))->resolve(locator, listener);
else
anon_send(listener, error(basp::ec::invalid_locator));
}
strong_actor_ptr test::make_proxy(node_id nid, actor_id aid) {
using impl_type = actor_proxy_impl;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &mm_.system(), cfg,
peer(nid));
}
void test::set_last_hop(node_id*) {
// nop
}
uint16_t test::port() const noexcept {
return 0;
}
test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
stream_socket second) {
using transport_type = stream_transport<basp::application>;
if (auto err = nonblocking(second, true))
CAF_LOG_ERROR("nonblocking failed: " << err);
auto mpx = mm_.mpx();
basp::application app{proxies_};
auto mgr = make_endpoint_manager(mpx, mm_.system(),
transport_type{second, std::move(app)});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err);
CAF_RAISE_ERROR("mgr->init() failed");
}
mpx->register_reading(mgr);
auto& result = peers_[peer_id];
result = std::make_pair(first, std::move(mgr));
return result;
}
test::peer_entry& test::get_peer(const node_id& id) {
auto i = peers_.find(id);
if (i != peers_.end())
return i->second;
auto sockets = make_stream_socket_pair();
if (!sockets) {
CAF_LOG_ERROR("make_stream_socket_pair failed: " << sockets.error());
CAF_RAISE_ERROR("make_stream_socket_pair failed");
}
return emplace(id, sockets->first, sockets->second);
}
} // namespace caf::net::backend
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.
#include "caf/net/endpoint_manager_queue.hpp"
namespace caf::net {
endpoint_manager_queue::element::~element() {
// nop
}
endpoint_manager_queue::event::event(uri locator, actor listener)
: element(element_type::event),
value(resolve_request{std::move(locator), std::move(listener)}) {
// nop
}
endpoint_manager_queue::event::event(node_id peer, actor_id proxy_id)
: element(element_type::event), value(new_proxy{peer, proxy_id}) {
// nop
}
endpoint_manager_queue::event::event(node_id observing_peer,
actor_id local_actor_id, error reason)
: element(element_type::event),
value(local_actor_down{observing_peer, local_actor_id, std::move(reason)}) {
// nop
}
endpoint_manager_queue::event::event(std::string tag, uint64_t id)
: element(element_type::event), value(timeout{std::move(tag), id}) {
// nop
}
endpoint_manager_queue::event::~event() {
// nop
}
size_t endpoint_manager_queue::event::task_size() const noexcept {
return 1;
}
endpoint_manager_queue::message::message(mailbox_element_ptr msg,
strong_actor_ptr receiver)
: element(element_type::message),
msg(std::move(msg)),
receiver(std::move(receiver)) {
// nop
}
size_t endpoint_manager_queue::message::task_size() const noexcept {
return message_policy::task_size(*this);
}
endpoint_manager_queue::message::~message() {
// nop
}
} // namespace caf::net
......@@ -7,24 +7,26 @@ namespace caf::net::http {
namespace {
constexpr string_view eol = "\r\n";
constexpr std::string_view eol = "\r\n";
template <class F>
bool for_each_line(string_view input, F&& f) {
bool for_each_line(std::string_view input, F&& f) {
for (auto pos = input.begin();;) {
auto line_end = std::search(pos, input.end(), eol.begin(), eol.end());
if (line_end == input.end() || std::distance(pos, input.end()) == 2) {
// Reaching the end or hitting the last empty line tells us we're done.
return true;
}
if (!f(string_view{pos, line_end}))
auto to_line_end = std::distance(pos, line_end);
CAF_ASSERT(to_line_end >= 0);
if (!f(std::string_view{pos, static_cast<size_t>(to_line_end)}))
return false;
pos = line_end + eol.size();
}
return true;
}
string_view trim(string_view str) {
std::string_view trim(std::string_view str) {
str.remove_prefix(std::min(str.find_first_not_of(' '), str.size()));
auto trim_pos = str.find_last_not_of(' ');
if (trim_pos != str.npos)
......@@ -34,24 +36,29 @@ string_view trim(string_view str) {
/// Splits `str` at the first occurrence of `sep` into the head and the
/// remainder (excluding the separator).
static std::pair<string_view, string_view> split(string_view str,
string_view sep) {
static std::pair<std::string_view, std::string_view>
split(std::string_view str, std::string_view sep) {
auto i = std::search(str.begin(), str.end(), sep.begin(), sep.end());
if (i != str.end())
return {{str.begin(), i}, {i + sep.size(), str.end()}};
return {{str}, {}};
if (i != str.end()) {
auto n = static_cast<size_t>(std::distance(str.begin(), i));
auto j = i + sep.size();
auto m = static_cast<size_t>(std::distance(j, str.end()));
return {{str.begin(), n}, {j, m}};
} else {
return {{str}, {}};
}
}
/// Convenience function for splitting twice.
static std::tuple<string_view, string_view, string_view>
split2(string_view str, string_view sep) {
static std::tuple<std::string_view, std::string_view, std::string_view>
split2(std::string_view str, std::string_view sep) {
auto [first, r1] = split(str, sep);
auto [second, third] = split(r1, sep);
return {first, second, third};
}
/// @pre `y` is all lowercase
bool case_insensitive_eq(string_view x, string_view y) {
bool case_insensitive_eq(std::string_view x, std::string_view y) {
return std::equal(x.begin(), x.end(), y.begin(), y.end(),
[](char a, char b) { return tolower(a) == b; });
}
......@@ -68,9 +75,10 @@ header& header::operator=(const header& other) {
}
void header::assign(const header& other) {
auto remap = [](const char* base, string_view src, const char* new_base) {
auto remap = [](const char* base, std::string_view src,
const char* new_base) {
auto offset = std::distance(base, src.data());
return string_view{new_base + offset, src.size()};
return std::string_view{new_base + offset, src.size()};
};
method_ = other.method_;
uri_ = other.uri_;
......@@ -88,12 +96,12 @@ void header::assign(const header& other) {
}
} else {
raw_.clear();
version_ = string_view{};
version_ = std::string_view{};
fields_.clear();
}
}
std::pair<status, string_view> header::parse(string_view raw) {
std::pair<status, std::string_view> header::parse(std::string_view raw) {
CAF_LOG_TRACE(CAF_ARG(raw));
// Sanity checking and copying of the raw input.
using namespace literals;
......@@ -103,8 +111,8 @@ std::pair<status, string_view> header::parse(string_view raw) {
};
raw_.assign(raw.begin(), raw.end());
// Parse the first line, i.e., "METHOD REQUEST-URI VERSION".
auto [first_line, remainder] = split(string_view{raw_.data(), raw_.size()},
eol);
auto [first_line, remainder]
= split(std::string_view{raw_.data(), raw_.size()}, eol);
auto [method_str, request_uri_str, version] = split2(first_line, " ");
// The path must be absolute.
if (request_uri_str.empty() || request_uri_str.front() != '/') {
......@@ -115,7 +123,7 @@ std::pair<status, string_view> header::parse(string_view raw) {
}
// The path must form a valid URI when prefixing a scheme. We don't actually
// care about the scheme, so just use "foo" here for the validation step.
if (auto res = make_uri("nil://host" + to_string(request_uri_str))) {
if (auto res = make_uri("nil://host" + std::string{request_uri_str})) {
uri_ = std::move(*res);
} else {
CAF_LOG_DEBUG("Failed to parse URI" << request_uri_str << "->"
......@@ -148,11 +156,13 @@ std::pair<status, string_view> header::parse(string_view raw) {
// Store the remaining header fields.
version_ = version;
fields_.clear();
bool ok = for_each_line(remainder, [this](string_view line) {
bool ok = for_each_line(remainder, [this](std::string_view line) {
if (auto sep = std::find(line.begin(), line.end(), ':');
sep != line.end()) {
auto key = trim({line.begin(), sep});
auto val = trim({sep + 1, line.end()});
auto n = static_cast<size_t>(std::distance(line.begin(), sep));
auto key = trim(std::string_view{line.begin(), n});
auto m = static_cast<size_t>(std::distance(sep + 1, line.end()));
auto val = trim(std::string_view{sep + 1, m});
if (!key.empty()) {
fields_.emplace(key, val);
return true;
......@@ -164,17 +174,17 @@ std::pair<status, string_view> header::parse(string_view raw) {
return {status::ok, "OK"};
} else {
raw_.clear();
version_ = string_view{};
version_ = std::string_view{};
fields_.clear();
return {status::bad_request, "Malformed header fields."};
}
}
bool header::chunked_transfer_encoding() const {
return field("Transfer-Encoding").find("chunked") != string_view::npos;
return field("Transfer-Encoding").find("chunked") != std::string_view::npos;
}
optional<size_t> header::content_length() const {
std::optional<size_t> header::content_length() const {
return field_as<size_t>("Content-Length");
}
......
......@@ -3,26 +3,26 @@
namespace caf::net::http {
std::string to_rfc_string(method x) {
using namespace caf::literals;
using namespace std::literals;
switch (x) {
case method::get:
return "GET";
return "GET"s;
case method::head:
return "HEAD";
return "HEAD"s;
case method::post:
return "POST";
return "POST"s;
case method::put:
return "PUT";
return "PUT"s;
case method::del:
return "DELETE";
return "DELETE"s;
case method::connect:
return "CONNECT";
return "CONNECT"s;
case method::options:
return "OPTIONS";
return "OPTIONS"s;
case method::trace:
return "TRACE";
return "TRACE"s;
default:
return "INVALID";
return "INVALID"s;
}
}
......
#include "caf/net/http/status.hpp"
#include <string_view>
namespace caf::net::http {
string_view phrase(status code) {
using namespace caf::literals;
std::string_view phrase(status code) {
using namespace std::literals;
switch (code) {
case status::continue_request:
return "Continue"_sv;
return "Continue"sv;
case status::switching_protocols:
return "Switching Protocols"_sv;
return "Switching Protocols"sv;
case status::ok:
return "OK"_sv;
return "OK"sv;
case status::created:
return "Created"_sv;
return "Created"sv;
case status::accepted:
return "Accepted"_sv;
return "Accepted"sv;
case status::non_authoritative_information:
return "Non-Authoritative Information"_sv;
return "Non-Authoritative Information"sv;
case status::no_content:
return "No Content"_sv;
return "No Content"sv;
case status::reset_content:
return "Reset Content"_sv;
return "Reset Content"sv;
case status::partial_content:
return "Partial Content"_sv;
return "Partial Content"sv;
case status::multiple_choices:
return "Multiple Choices"_sv;
return "Multiple Choices"sv;
case status::moved_permanently:
return "Moved Permanently"_sv;
return "Moved Permanently"sv;
case status::found:
return "Found"_sv;
return "Found"sv;
case status::see_other:
return "See Other"_sv;
return "See Other"sv;
case status::not_modified:
return "Not Modified"_sv;
return "Not Modified"sv;
case status::use_proxy:
return "Use Proxy"_sv;
return "Use Proxy"sv;
case status::temporary_redirect:
return "Temporary Redirect"_sv;
return "Temporary Redirect"sv;
case status::bad_request:
return "Bad Request"_sv;
return "Bad Request"sv;
case status::unauthorized:
return "Unauthorized"_sv;
return "Unauthorized"sv;
case status::payment_required:
return "Payment Required"_sv;
return "Payment Required"sv;
case status::forbidden:
return "Forbidden"_sv;
return "Forbidden"sv;
case status::not_found:
return "Not Found"_sv;
return "Not Found"sv;
case status::method_not_allowed:
return "Method Not Allowed"_sv;
return "Method Not Allowed"sv;
case status::not_acceptable:
return "Not Acceptable"_sv;
return "Not Acceptable"sv;
case status::proxy_authentication_required:
return "Proxy Authentication Required"_sv;
return "Proxy Authentication Required"sv;
case status::request_timeout:
return "Request Timeout"_sv;
return "Request Timeout"sv;
case status::conflict:
return "Conflict"_sv;
return "Conflict"sv;
case status::gone:
return "Gone"_sv;
return "Gone"sv;
case status::length_required:
return "Length Required"_sv;
return "Length Required"sv;
case status::precondition_failed:
return "Precondition Failed"_sv;
return "Precondition Failed"sv;
case status::payload_too_large:
return "Payload Too Large"_sv;
return "Payload Too Large"sv;
case status::uri_too_long:
return "URI Too Long"_sv;
return "URI Too Long"sv;
case status::unsupported_media_type:
return "Unsupported Media Type"_sv;
return "Unsupported Media Type"sv;
case status::range_not_satisfiable:
return "Range Not Satisfiable"_sv;
return "Range Not Satisfiable"sv;
case status::expectation_failed:
return "Expectation Failed"_sv;
return "Expectation Failed"sv;
case status::upgrade_required:
return "Upgrade Required"_sv;
return "Upgrade Required"sv;
case status::precondition_required:
return "Precondition Required"_sv;
return "Precondition Required"sv;
case status::too_many_requests:
return "Too Many Requests"_sv;
return "Too Many Requests"sv;
case status::request_header_fields_too_large:
return "Request Header Fields Too Large"_sv;
return "Request Header Fields Too Large"sv;
case status::internal_server_error:
return "Internal Server Error"_sv;
return "Internal Server Error"sv;
case status::not_implemented:
return "Not Implemented"_sv;
return "Not Implemented"sv;
case status::bad_gateway:
return "Bad Gateway"_sv;
return "Bad Gateway"sv;
case status::service_unavailable:
return "Service Unavailable"_sv;
return "Service Unavailable"sv;
case status::gateway_timeout:
return "Gateway Timeout"_sv;
return "Gateway Timeout"sv;
case status::http_version_not_supported:
return "HTTP Version Not Supported"_sv;
return "HTTP Version Not Supported"sv;
case status::network_authentication_required:
return "Network Authentication Required"_sv;
return "Network Authentication Required"sv;
default:
return "Unrecognized";
return "Unrecognized"sv;
}
}
......
#include "caf/net/http/v1.hpp"
#include <array>
#include <cstddef>
#include <string_view>
using namespace std::literals;
......@@ -14,13 +15,7 @@ struct writer {
};
writer& operator<<(writer& out, char x) {
out.buf->push_back(static_cast<byte>(x));
return out;
}
writer& operator<<(writer& out, string_view str) {
auto bytes = as_bytes(make_span(str));
out.buf->insert(out.buf->end(), bytes.begin(), bytes.end());
out.buf->push_back(static_cast<std::byte>(x));
return out;
}
......@@ -31,26 +26,27 @@ writer& operator<<(writer& out, std::string_view str) {
}
writer& operator<<(writer& out, const std::string& str) {
return out << string_view{str};
return out << std::string_view{str};
}
} // namespace
std::pair<string_view, byte_span> split_header(byte_span bytes) {
std::array<byte, 4> end_of_header{{
byte{'\r'},
byte{'\n'},
byte{'\r'},
byte{'\n'},
std::pair<std::string_view, byte_span> split_header(byte_span bytes) {
std::array<std::byte, 4> end_of_header{{
std::byte{'\r'},
std::byte{'\n'},
std::byte{'\r'},
std::byte{'\n'},
}};
if (auto i = std::search(bytes.begin(), bytes.end(), end_of_header.begin(),
end_of_header.end());
i == bytes.end()) {
return {string_view{}, bytes};
return {std::string_view{}, bytes};
} else {
auto offset = static_cast<size_t>(std::distance(bytes.begin(), i));
offset += end_of_header.size();
return {string_view{reinterpret_cast<const char*>(bytes.begin()), offset},
return {std::string_view{reinterpret_cast<const char*>(bytes.begin()),
offset},
bytes.subspan(offset)};
}
}
......@@ -65,16 +61,17 @@ void write_header(status code, const header_fields_map& fields,
out << "\r\n"sv;
}
void write_response(status code, string_view content_type, string_view content,
byte_buffer& buf) {
void write_response(status code, std::string_view content_type,
std::string_view content, byte_buffer& buf) {
header_fields_map fields;
write_response(code, content_type, content, fields, buf);
writer out{&buf};
out << content;
}
void write_response(status code, string_view content_type, string_view content,
const header_fields_map& fields, byte_buffer& buf) {
void write_response(status code, std::string_view content_type,
std::string_view content, const header_fields_map& fields,
byte_buffer& buf) {
writer out{&buf};
out << "HTTP/1.1 "sv << std::to_string(static_cast<int>(code)) << ' '
<< phrase(code) << "\r\n"sv;
......
......@@ -7,10 +7,6 @@
#include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/expected.hpp"
#include "caf/init_global_meta_objects.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/raise_error.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
......@@ -19,7 +15,7 @@
namespace caf::net {
void middleman::init_global_meta_objects() {
caf::init_global_meta_objects<id_block::net_module>();
// nop
}
middleman::middleman(actor_system& sys) : sys_(sys), mpx_(this) {
......@@ -46,8 +42,6 @@ void middleman::start() {
}
void middleman::stop() {
for (const auto& backend : backends_)
backend->stop();
mpx_.shutdown();
if (mpx_thread_.joinable())
mpx_thread_.join();
......@@ -66,11 +60,6 @@ void middleman::init(actor_system_config& cfg) {
} else {
// CAF_RAISE_ERROR("no valid entry for caf.middleman.this-node found");
}
for (auto& backend : backends_)
if (auto err = backend->init()) {
CAF_LOG_ERROR("failed to initialize backend: " << err);
CAF_RAISE_ERROR("failed to initialize backend");
}
}
middleman::module::id_t middleman::id() const {
......@@ -81,6 +70,10 @@ void* middleman::subtype_ptr() {
return this;
}
actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
return new middleman(sys);
}
void middleman::add_module_options(actor_system_config& cfg) {
config_option_adder{cfg.custom_options(), "caf.middleman"}
.add<std::vector<std::string>>("app-identifiers",
......@@ -98,36 +91,4 @@ void middleman::add_module_options(actor_system_config& cfg) {
.add<std::string>("network-backend", "legacy option");
}
expected<endpoint_manager_ptr> middleman::connect(const uri& locator) {
if (auto ptr = backend(locator.scheme()))
return ptr->get_or_connect(locator);
else
return basp::ec::invalid_scheme;
}
void middleman::resolve(const uri& locator, const actor& listener) {
auto ptr = backend(locator.scheme());
if (ptr != nullptr)
ptr->resolve(locator, listener);
else
anon_send(listener, error{basp::ec::invalid_scheme});
}
middleman_backend* middleman::backend(string_view scheme) const noexcept {
auto predicate = [&](const middleman_backend_ptr& ptr) {
return ptr->id() == scheme;
};
auto i = std::find_if(backends_.begin(), backends_.end(), predicate);
if (i != backends_.end())
return i->get();
return nullptr;
}
expected<uint16_t> middleman::port(string_view scheme) const {
if (auto ptr = backend(scheme))
return ptr->port();
else
return basp::ec::invalid_scheme;
}
} // namespace caf::net
This diff is collapsed.
......@@ -14,7 +14,6 @@
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace {
......
......@@ -17,7 +17,6 @@
#include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf::net {
......@@ -36,12 +35,12 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
}
}
ptrdiff_t write(pipe_socket x, span<const byte> buf) {
ptrdiff_t write(pipe_socket x, const_byte_span buf) {
// On Windows, a pipe consists of two stream sockets.
return write(socket_cast<stream_socket>(x), buf);
}
ptrdiff_t read(pipe_socket x, span<byte> buf) {
ptrdiff_t read(pipe_socket x, byte_span buf) {
// On Windows, a pipe consists of two stream sockets.
return read(socket_cast<stream_socket>(x), buf);
}
......@@ -67,13 +66,13 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
return std::make_pair(pipe_socket{pipefds[0]}, pipe_socket{pipefds[1]});
}
ptrdiff_t write(pipe_socket x, span<const byte> buf) {
ptrdiff_t write(pipe_socket x, const_byte_span buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size());
}
ptrdiff_t read(pipe_socket x, span<byte> buf) {
ptrdiff_t read(pipe_socket x, byte_span buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size());
......
......@@ -12,7 +12,6 @@
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf::net {
......
......@@ -11,7 +11,6 @@
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace caf::net {
......
......@@ -4,7 +4,6 @@
#include "caf/net/stream_socket.hpp"
#include "caf/byte.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
......@@ -13,7 +12,8 @@
#include "caf/net/socket.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
#include <cstddef>
#ifdef CAF_POSIX
# include <sys/uio.h>
......@@ -154,13 +154,13 @@ error nodelay(stream_socket x, bool new_value) {
return none;
}
ptrdiff_t read(stream_socket x, span<byte> buf) {
ptrdiff_t read(stream_socket x, byte_span buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag);
}
ptrdiff_t write(stream_socket x, span<const byte> buf) {
ptrdiff_t write(stream_socket x, const_byte_span buf) {
CAF_LOG_TRACE(CAF_ARG2("socket", x.id) << CAF_ARG2("bytes", buf.size()));
return ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), buf.size(),
no_sigpipe_io_flag);
......@@ -168,10 +168,10 @@ ptrdiff_t write(stream_socket x, span<const byte> buf) {
#ifdef CAF_WINDOWS
ptrdiff_t write(stream_socket x, std::initializer_list<span<const byte>> bufs) {
ptrdiff_t write(stream_socket x, std::initializer_list<const_byte_span> bufs) {
CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10];
auto convert = [](span<const byte> buf) {
auto convert = [](const_byte_span buf) {
auto data = const_cast<byte*>(buf.data());
return WSABUF{static_cast<ULONG>(buf.size()),
reinterpret_cast<CHAR*>(data)};
......@@ -185,11 +185,11 @@ ptrdiff_t write(stream_socket x, std::initializer_list<span<const byte>> bufs) {
#else // CAF_WINDOWS
ptrdiff_t write(stream_socket x, std::initializer_list<span<const byte>> bufs) {
ptrdiff_t write(stream_socket x, std::initializer_list<const_byte_span> bufs) {
CAF_ASSERT(bufs.size() < 10);
iovec buf_array[10];
auto convert = [](span<const byte> buf) {
return iovec{const_cast<byte*>(buf.data()), buf.size()};
auto convert = [](const_byte_span buf) {
return iovec{const_cast<std::byte*>(buf.data()), buf.size()};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
return writev(x.id, buf_array, static_cast<int>(bufs.size()));
......
......@@ -105,9 +105,9 @@ expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
expected<tcp_accept_socket>
make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) {
if (auto ip = get_if<ip_address>(&node.host))
if (auto ip = std::get_if<ip_address>(&node.host))
return make_tcp_accept_socket(ip_endpoint{*ip, node.port}, reuse_addr);
auto host = get<std::string>(node.host);
auto host = std::get<std::string>(node.host);
auto addrs = ip::local_addresses(host);
if (addrs.empty())
return make_error(sec::cannot_open_port, "no local interface available",
......
......@@ -13,7 +13,6 @@
#include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
#include <algorithm>
......@@ -158,9 +157,9 @@ make_connected_tcp_stream_socket(const uri::authority_type& node,
if (port == 0)
return make_error(sec::cannot_connect_to_node, "port is zero");
std::vector<ip_address> addrs;
if (auto str = get_if<std::string>(&node.host))
if (auto str = std::get_if<std::string>(&node.host))
addrs = ip::resolve(*str);
else if (auto addr = get_if<ip_address>(&node.host))
else if (auto addr = std::get_if<ip_address>(&node.host))
addrs.push_back(*addr);
if (addrs.empty())
return make_error(sec::cannot_connect_to_node, "empty authority");
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -2,11 +2,10 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE convert_ip_endpoint
#define CAF_SUITE detail.convert_ip_endpoint
#include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include <cstring>
......@@ -20,8 +19,8 @@ using namespace caf::detail;
namespace {
struct fixture : host_fixture {
fixture() : host_fixture() {
struct fixture {
fixture() {
memset(&sockaddr4_src, 0, sizeof(sockaddr_storage));
memset(&sockaddr6_src, 0, sizeof(sockaddr_storage));
auto sockaddr6_ptr = reinterpret_cast<sockaddr_in6*>(&sockaddr6_src);
......
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