Unverified Commit f06f38c6 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1329

Bring caf_net into the main repository
parents 9e130954 0407221b
[files]
extend-exclude = ["*.pem"]
\ No newline at end of file
extend-exclude = ["*.pem", "libcaf_net/test/pem.cpp"]
......@@ -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 -----------------------------------------
......@@ -73,6 +74,14 @@ if(MSVC AND CAF_SANITIZERS)
message(FATAL_ERROR "Sanitizer builds are currently not supported on MSVC")
endif()
# -- get dependencies that are used in more than one module --------------------
if(CAF_ENABLE_OPENSSL_MODULE OR CAF_ENABLE_NET_MODULE)
if(NOT TARGET OpenSSL::SSL OR NOT TARGET OpenSSL::Crypto)
find_package(OpenSSL REQUIRED)
endif()
endif()
# -- base target setup ---------------------------------------------------------
# This target propagates compiler flags, extra dependencies, etc. All other CAF
......@@ -368,6 +377,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()
......
......@@ -127,3 +127,17 @@ if(TARGET CAF::io)
endif()
endif()
if(TARGET CAF::net)
function(add_net_example folder name)
add_example(${folder} ${name} ${ARGN})
target_link_libraries(${name} PRIVATE CAF::internal CAF::net)
endfunction()
else()
function(add_net_example)
# nop
endfunction()
endif()
add_net_example(web_socket quote-server)
add_net_example(web_socket echo)
// Simple WebSocket server that sends everything it receives back to the sender.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <utility>
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port})) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
return EXIT_FAILURE;
}
// Convenience type aliases.
using event_t = ws::accept_event_t<>;
// Create buffers to signal events from the WebSocket server to the worker.
auto [wres, sres] = ws::make_accept_event_resources();
// Spin up a worker to handle the events.
auto worker = sys.spawn([worker_res = wres](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ...
self->make_observable()
.from_resource(worker_res)
.for_each([self](const event_t& event) {
// ... that simply pushes data back to the sender.
auto [pull, push] = event.data();
self->make_observable()
.from_resource(pull)
.do_on_next([](const ws::frame& x) {
if (x.is_binary()) {
std::cout << "*** received a binary WebSocket frame of size "
<< x.size() << '\n';
} else {
std::cout << "*** received a text WebSocket frame of size "
<< x.size() << '\n';
}
})
.subscribe(push);
});
});
// Callback for incoming WebSocket requests.
auto on_request = [](const caf::settings& hdr, auto& req) {
// The hdr parameter is a dictionary with fields from the WebSocket
// handshake such as the path.
auto path = caf::get_or(hdr, "web-socket.path", "/");
std::cout << "*** new client request for path " << path << '\n';
// Accept the WebSocket connection only if the path is "/".
if (path == "/") {
// Calling `accept` causes the server to acknowledge the client and
// creates input and output buffers that go to the worker actor.
req.accept();
} else {
// Calling `reject` aborts the connection with HTTP status code 400 (Bad
// Request). The error gets converted to a string and send to the client
// to give some indication to the client why the request was rejected.
auto err = caf::make_error(caf::sec::invalid_argument,
"unrecognized path, try '/'");
req.reject(std::move(err));
}
// Note: calling neither accept nor reject also rejects the connection.
};
// Set everything in motion.
ws::accept(sys, fd, std::move(sres), on_request);
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
// Simple WebSocket server that sends local files from a working directory to
// the clients.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/cow_string.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
#include <cassert>
#include <iostream>
#include <utility>
using namespace std::literals;
static constexpr uint16_t default_port = 8080;
static constexpr std::string_view epictetus[] = {
"Wealth consists not in having great possessions, but in having few wants.",
"Don't explain your philosophy. Embody it.",
"First say to yourself what you would be; and then do what you have to do.",
"It's not what happens to you, but how you react to it that matters.",
"If you want to improve, be content to be thought foolish and stupid.",
"He who laughs at himself never runs out of things to laugh at.",
"It is impossible for a man to learn what he thinks he already knows.",
"Circumstances don't make the man, they only reveal him to himself.",
"People are not disturbed by things, but by the views they take of them.",
"Only the educated are free.",
};
static constexpr std::string_view seneca[] = {
"Luck is what happens when preparation meets opportunity.",
"All cruelty springs from weakness.",
"We suffer more often in imagination than in reality.",
"Difficulties strengthen the mind, as labor does the body.",
"If a man knows not to which port he sails, no wind is favorable.",
"It is the power of the mind to be unconquerable.",
"No man was ever wise by chance.",
"He suffers more than necessary, who suffers before it is necessary.",
"I shall never be ashamed of citing a bad author if the line is good.",
"Only time can heal what reason cannot.",
};
static constexpr std::string_view plato[] = {
"Love is a serious mental disease.",
"The measure of a man is what he does with power.",
"Ignorance, the root and stem of every evil.",
"Those who tell the stories rule society.",
"You should not honor men more than truth.",
"When men speak ill of thee, live so as nobody may believe them.",
"The beginning is the most important part of the work.",
"Necessity is the mother of invention.",
"The greatest wealth is to live content with little.",
"Beauty lies in the eyes of the beholder.",
};
caf::span<const std::string_view> quotes_by_path(std::string_view path) {
if (path == "/epictetus")
return caf::make_span(epictetus);
else if (path == "/seneca")
return caf::make_span(seneca);
else if (path == "/plato")
return caf::make_span(plato);
else
return {};
}
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
}
};
struct pick_random {
public:
pick_random() : engine_(std::random_device{}()) {
// nop
}
std::string_view operator()(caf::span<const std::string_view> quotes) {
assert(!quotes.empty());
auto dis = std::uniform_int_distribution<size_t>{0, quotes.size() - 1};
return quotes[dis(engine_)];
}
private:
std::minstd_rand engine_;
};
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port})) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
return EXIT_FAILURE;
}
// Convenience type aliases.
using event_t = ws::accept_event_t<caf::cow_string>;
// Create buffers to signal events from the WebSocket server to the worker.
auto [wres, sres] = ws::make_accept_event_resources<caf::cow_string>();
// Spin up a worker to handle the events.
auto worker = sys.spawn([worker_res = wres](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ...
self->make_observable()
.from_resource(worker_res)
.for_each([self, f = pick_random{}](const event_t& event) mutable {
// ... that pushes one random quote to the client.
auto [pull, push, path] = event.data();
auto quotes = quotes_by_path(path);
auto quote = quotes.empty() ? "Try /epictetus, /seneca or /plato."
: f(quotes);
self->make_observable().just(ws::frame{quote}).subscribe(push);
// Note: we simply drop `pull` here, which will close the buffer.
});
});
// Callback for incoming WebSocket requests.
auto on_request = [](const caf::settings& hdr, auto& req) {
// The hdr parameter is a dictionary with fields from the WebSocket
// handshake such as the path. This is only field we care about here.
auto path = caf::get_or(hdr, "web-socket.path", "/");
req.accept(caf::cow_string{std::move(path)});
};
// Set everything in motion.
ws::accept(sys, fd, std::move(sres), on_request);
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -124,6 +124,10 @@ public:
// nop
}
explicit blocking_producer(spsc_buffer_ptr<T> buf) {
impl_.emplace(std::move(buf));
}
~blocking_producer() {
if (impl_)
impl_->close();
......
......@@ -462,19 +462,23 @@ private:
intrusive_ptr<resource_ctrl<T, true>> ctrl_;
};
/// Creates spsc buffer and returns two resources connected by that buffer.
template <class T1, class T2 = T1>
using resource_pair = std::pair<consumer_resource<T1>, producer_resource<T2>>;
/// Creates an @ref spsc_buffer and returns two resources connected by that
/// buffer.
template <class T>
std::pair<consumer_resource<T>, producer_resource<T>>
resource_pair<T>
make_spsc_buffer_resource(size_t buffer_size, size_t min_request_size) {
using buffer_type = spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
return {async::consumer_resource<T>{buf}, async::producer_resource<T>{buf}};
}
/// Creates spsc buffer and returns two resources connected by that buffer.
/// Creates an @ref spsc_buffer and returns two resources connected by that
/// buffer.
template <class T>
std::pair<consumer_resource<T>, producer_resource<T>>
make_spsc_buffer_resource() {
resource_pair<T> make_spsc_buffer_resource() {
return make_spsc_buffer_resource<T>(defaults::flow::buffer_size,
defaults::flow::min_demand);
}
......
......@@ -33,15 +33,15 @@ public:
}
template <class Next, class... Steps>
void finally(Next& next, Steps&... steps) {
void on_complete(Next& next, Steps&... steps) {
fn_();
next.finally(steps...);
next.on_complete(steps...);
}
template <class Next, class... Steps>
void finally(const error& what, Next& next, Steps&... steps) {
void on_error(const error& what, Next& next, Steps&... steps) {
fn_();
next.finally(what, steps...);
next.on_error(what, steps...);
}
private:
......
......@@ -16,15 +16,12 @@ class do_on_next {
public:
using trait = detail::get_callable_trait_t<F>;
static_assert(!std::is_same_v<typename trait::result_type, void>,
"do_on_next functions may not return void");
static_assert(trait::num_args == 1,
"do_on_next functions must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = std::decay_t<typename trait::result_type>;
using output_type = input_type;
explicit do_on_next(F fn) : fn_(std::move(fn)) {
// nop
......
......@@ -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,44 @@ 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) {
// We need actual char* pointers for make_string_view. On some compilers,
// begin() and end() may return iterator types, so we use data() instead.
auto str_end = str.data() + str.size();
auto i = std::search(str.data(), str_end, sep.begin(), sep.end());
if (i != str_end) {
return {make_string_view(str.data(), 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
......
# -- get header files for creating "proper" XCode projects ---------------------
file(GLOB_RECURSE CAF_NET_HEADERS "caf/*.hpp")
# -- add targets ---------------------------------------------------------------
caf_add_component(
net
DEPENDENCIES
PUBLIC
$<$<CXX_COMPILER_ID:MSVC>:ws2_32>
CAF::core
OpenSSL::Crypto
OpenSSL::SSL
PRIVATE
CAF::internal
ENUM_TYPES
net.http.method
net.http.status
net.operation
net.stream_transport_error
net.web_socket.status
HEADERS
${CAF_NET_HEADERS}
SOURCES
src/datagram_socket.cpp
src/detail/convert_ip_endpoint.cpp
src/detail/rfc6455.cpp
src/host.cpp
src/ip.cpp
src/multiplexer.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/http/header.cpp
src/net/http/method.cpp
src/net/http/status.cpp
src/net/http/v1.cpp
src/net/middleman.cpp
src/net/web_socket/default_trait.cpp
src/net/web_socket/frame.cpp
src/net/web_socket/handshake.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
TEST_SOURCES
test/net-test.cpp
test/pem.cpp
TEST_SUITES
detail.convert_ip_endpoint
detail.rfc6455
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)
# Our OpenSSL test currently depends on <filesystem>.
check_cxx_source_compiles("
#include <cstdlib>
#include <filesystem>
int main(int, char**) {
auto cwd = std::filesystem::current_path();
auto str = cwd.string();
return str.empty() ? EXIT_FAILURE : EXIT_SUCCESS;
}
"
CAF_NET_HAS_STD_FILESYSTEM)
if(CAF_NET_HAS_STD_FILESYSTEM)
caf_add_test_suites(caf-net-test net.openssl_transport)
else()
message(STATUS "<filesystem> not working, skip OpenSSL test in CAF::net")
endif()
This diff is collapsed.
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/fwd.hpp"
namespace caf::detail {
void convert(const ip_endpoint& src, sockaddr_storage& dst);
error convert(const sockaddr_storage& src, ip_endpoint& dst);
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
namespace caf::detail {
template <class T, class LowerLayerPtr>
class has_after_reading {
private:
template <class A, class B>
static auto sfinae(A& up, B& ptr)
-> decltype(up.after_reading(ptr), std::true_type{});
template <class A>
static std::false_type sfinae(A&, ...);
using sfinae_result
= decltype(sfinae(std::declval<T&>(), std::declval<LowerLayerPtr&>()));
public:
static constexpr bool value = sfinae_result::value;
};
template <class T, class LowerLayerPtr>
constexpr bool has_after_reading_v = has_after_reading<T, LowerLayerPtr>::value;
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::detail {
template <class T>
struct actor_shell_ptr_type_oracle;
template <>
struct actor_shell_ptr_type_oracle<actor> {
using type = net::actor_shell_ptr;
};
template <class... Sigs>
struct actor_shell_ptr_type_oracle<typed_actor<Sigs...>> {
using type = net::typed_actor_shell_ptr<Sigs...>;
};
template <class T>
using infer_actor_shell_ptr_type =
typename actor_shell_ptr_type_oracle<T>::type;
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstdio>
#include <cstdlib>
#include "caf/error.hpp"
#include "caf/sec.hpp"
/// Calls a C functions and returns an error if `var op rhs` returns `true`.
#define CAF_NET_SYSCALL(funname, var, op, rhs, expr) \
auto var = expr; \
if (var op rhs) \
return make_error(sec::network_syscall_failed, funname, \
last_socket_error_as_string())
/// Calls a C functions and calls exit() if `var op rhs` returns `true`.
#define CAF_NET_CRITICAL_SYSCALL(funname, var, op, rhs, expr) \
auto var = expr; \
if (var op rhs) { \
fprintf(stderr, "[FATAL] %s:%u: syscall failed: %s returned %s\n", \
__FILE__, __LINE__, funname, \
last_socket_error_as_string().c_str()); \
abort(); \
} \
static_cast<void>(0)
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/byte_buffer.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/span.hpp"
#include <cstdint>
namespace caf::detail {
struct CAF_NET_EXPORT rfc6455 {
// -- member types -----------------------------------------------------------
struct header {
bool fin;
uint8_t opcode;
uint32_t mask_key;
uint64_t payload_len;
};
// -- constants --------------------------------------------------------------
static constexpr uint8_t continuation_frame = 0x00;
static constexpr uint8_t text_frame = 0x01;
static constexpr uint8_t binary_frame = 0x02;
static constexpr uint8_t connection_close = 0x08;
static constexpr uint8_t ping = 0x09;
static constexpr uint8_t pong = 0x0A;
// -- utility functions ------------------------------------------------------
static void mask_data(uint32_t key, span<char> data);
static void mask_data(uint32_t key, byte_span data);
static void assemble_frame(uint32_t mask_key, span<const char> data,
byte_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,
const_byte_span data, byte_buffer& out);
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/detail/socket_sys_includes.hpp"
namespace caf::detail {
inline auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
inline auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
inline auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
inline auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
inline auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
inline auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/config.hpp"
namespace caf::net {
#ifdef CAF_WINDOWS
using setsockopt_ptr = const char*;
using getsockopt_ptr = char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
using socket_size_type = int;
#else // CAF_WINDOWS
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
using socket_size_type = unsigned;
#endif // CAF_WINDOWS
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
// This convenience header pulls in platform-specific headers for the C socket
// API. Do *not* include this header in other headers.
#pragma once
#include "caf/config.hpp"
// clang-format off
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif // CAF_WINDOWS
# ifndef NOMINMAX
# define NOMINMAX
# endif // NOMINMAX
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif // CAF_MINGW
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else // CAF_WINDOWS
# include <sys/types.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <unistd.h>
# include <sys/uio.h>
#endif
// clang-format on
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_traits.hpp"
#include "caf/callback.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/local_actor.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
#include "caf/policy/normal_messages.hpp"
namespace caf::net {
class CAF_NET_EXPORT abstract_actor_shell : public local_actor,
public non_blocking_actor_base {
public:
// -- member types -----------------------------------------------------------
using super = local_actor;
struct mailbox_policy {
using queue_type = intrusive::drr_queue<policy::normal_messages>;
using deficit_type = policy::normal_messages::deficit_type;
using mapped_type = policy::normal_messages::mapped_type;
using unique_pointer = policy::normal_messages::unique_pointer;
};
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
using fallback_handler = unique_callback_ptr<result<message>(message&)>;
// -- constructors, destructors, and assignment operators --------------------
abstract_actor_shell(actor_config& cfg, socket_manager* owner);
~abstract_actor_shell() override;
// -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox.
void quit(error reason);
/// Overrides the default handler for unexpected messages.
template <class F>
void set_fallback(F f) {
fallback_ = make_type_erased_callback(std::move(f));
}
// -- mailbox access ---------------------------------------------------------
auto& mailbox() noexcept {
return mailbox_;
}
/// Dequeues and returns the next message from the mailbox or returns
/// `nullptr` if the mailbox is empty.
mailbox_element_ptr next_message();
/// Tries to put the mailbox into the `blocked` state, causing the next
/// enqueue to register the owning socket manager for write events.
bool try_block_mailbox();
// -- message processing -----------------------------------------------------
/// Dequeues and processes the next message from the mailbox.
/// @returns `true` if a message was dequeued and processed, `false` if the
/// mailbox was empty.
bool consume_message();
/// Adds a callback for a multiplexed response.
void add_multiplexed_response_handler(message_id response_id, behavior bhvr);
// -- overridden functions of abstract_actor ---------------------------------
using abstract_actor::enqueue;
bool enqueue(mailbox_element_ptr ptr, execution_unit* eu) override;
mailbox_element* peek_at_next_mailbox_element() override;
// -- overridden functions of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override;
bool cleanup(error&& fail_state, execution_unit* host) override;
protected:
// Stores incoming actor messages.
mailbox_type mailbox_;
// Guards access to owner_.
std::mutex owner_mtx_;
// Points to the owning manager (nullptr after quit was called).
socket_manager* owner_;
// Handler for consuming messages from the mailbox.
behavior bhvr_;
// Handler for unexpected messages.
fallback_handler fallback_;
// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_traits.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/net/abstract_actor_shell.hpp"
#include "caf/net/fwd.hpp"
#include "caf/none.hpp"
namespace caf::net {
/// Enables socket managers to communicate with actors using dynamically typed
/// messaging.
class CAF_NET_EXPORT actor_shell
: public extend<abstract_actor_shell, actor_shell>::with<mixin::sender,
mixin::requester>,
public dynamically_typed_actor_base {
public:
// -- friends ----------------------------------------------------------------
friend class actor_shell_ptr;
// -- member types -----------------------------------------------------------
using super = extend<abstract_actor_shell,
actor_shell>::with<mixin::sender, mixin::requester>;
using signatures = none_t;
using behavior_type = behavior;
// -- constructors, destructors, and assignment operators --------------------
actor_shell(actor_config& cfg, socket_manager* owner);
~actor_shell() override;
// -- state modifiers --------------------------------------------------------
/// Overrides the callbacks for incoming messages.
template <class... Fs>
void set_behavior(Fs... fs) {
bhvr_ = behavior{std::move(fs)...};
}
// -- overridden functions of local_actor ------------------------------------
const char* name() const override;
};
/// An "owning" pointer to an actor shell in the sense that it calls `quit()` on
/// the shell when going out of scope.
class CAF_NET_EXPORT actor_shell_ptr {
public:
// -- friends ----------------------------------------------------------------
friend class socket_manager;
// -- member types -----------------------------------------------------------
using handle_type = actor;
using element_type = actor_shell;
// -- constructors, destructors, and assignment operators --------------------
constexpr actor_shell_ptr() noexcept {
// nop
}
constexpr actor_shell_ptr(std::nullptr_t) noexcept {
// nop
}
actor_shell_ptr(actor_shell_ptr&& other) noexcept = default;
actor_shell_ptr& operator=(actor_shell_ptr&& other) noexcept = default;
actor_shell_ptr(const actor_shell_ptr& other) = delete;
actor_shell_ptr& operator=(const actor_shell_ptr& other) = delete;
~actor_shell_ptr();
// -- smart pointer interface ------------------------------------------------
/// Returns an actor handle to the managed actor shell.
handle_type as_actor() const noexcept;
void detach(error reason);
element_type* get() const noexcept;
element_type* operator->() const noexcept {
return get();
}
element_type& operator*() const noexcept {
return *get();
}
bool operator!() const noexcept {
return !ptr_;
}
explicit operator bool() const noexcept {
return static_cast<bool>(ptr_);
}
private:
/// @pre `ptr != nullptr`
explicit actor_shell_ptr(strong_actor_ptr ptr) noexcept;
strong_actor_ptr ptr_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/logger.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/send.hpp"
namespace caf::net {
/// A connection_acceptor accepts connections from an accept socket and creates
/// socket managers to handle them via its factory.
template <class Socket, class Factory>
class connection_acceptor {
public:
// -- member types -----------------------------------------------------------
using input_tag = tag::io_event_oriented;
using socket_type = Socket;
using factory_type = Factory;
using read_result = typename socket_manager::read_result;
using write_result = typename socket_manager::write_result;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
explicit connection_acceptor(size_t limit, Ts&&... xs)
: factory_(std::forward<Ts>(xs)...), limit_(limit) {
// nop
}
// -- interface functions ----------------------------------------------------
template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
CAF_LOG_TRACE("");
owner_ = owner;
cfg_ = cfg;
if (auto err = factory_.init(owner, cfg))
return err;
down->register_reading();
return none;
}
template <class LowerLayerPtr>
read_result handle_read_event(LowerLayerPtr down) {
CAF_LOG_TRACE("");
if (auto x = accept(down->handle())) {
socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr());
if (!child) {
CAF_LOG_ERROR("factory failed to create a new child");
down->abort_reason(sec::runtime_error);
return read_result::stop;
}
if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
down->abort_reason(std::move(err));
return read_result::stop;
}
if (limit_ == 0) {
return read_result::again;
} else {
return ++accepted_ < limit_ ? read_result::again : read_result::stop;
}
} else {
CAF_LOG_ERROR("accept failed:" << x.error());
return read_result::stop;
}
}
template <class LowerLayerPtr>
static read_result handle_buffered_data(LowerLayerPtr) {
return read_result::again;
}
template <class LowerLayerPtr>
static read_result handle_continue_reading(LowerLayerPtr) {
return read_result::again;
}
template <class LowerLayerPtr>
write_result handle_write_event(LowerLayerPtr) {
CAF_LOG_ERROR("connection_acceptor received write event");
return write_result::stop;
}
template <class LowerLayerPtr>
static write_result handle_continue_writing(LowerLayerPtr) {
CAF_LOG_ERROR("connection_acceptor received continue writing event");
return write_result::stop;
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
factory_.abort(reason);
}
private:
factory_type factory_;
socket_manager* owner_;
size_t limit_;
size_t accepted_ = 0;
settings cfg_;
};
/// Converts a function object into a factory object for a
/// @ref connection_acceptor.
template <class FunctionObject>
class connection_acceptor_factory_adapter {
public:
explicit connection_acceptor_factory_adapter(FunctionObject f)
: f_(std::move(f)) {
// nop
}
connection_acceptor_factory_adapter(connection_acceptor_factory_adapter&&)
= default;
error init(socket_manager*, const settings&) {
return none;
}
template <class Socket>
socket_manager_ptr make(Socket connected_socket, multiplexer* mpx) {
return f_(std::move(connected_socket), mpx);
}
void abort(const error&) {
// nop
}
private:
FunctionObject f_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/async/consumer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net {
/// Connects a socket manager to an asynchronous consumer resource. Whenever new
/// data becomes ready, the adapter registers the socket manager for writing.
template <class Buffer>
class consumer_adapter final : public ref_counted, public async::consumer {
public:
using buf_ptr = intrusive_ptr<Buffer>;
void on_producer_ready() override {
// nop
}
void on_producer_wakeup() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { //
adapter->on_wakeup();
});
}
void ref_consumer() const noexcept override {
this->ref();
}
void deref_consumer() const noexcept override {
this->deref();
}
template <class Policy, class Observer>
std::pair<bool, size_t> pull(Policy policy, size_t demand, Observer& dst) {
return buf_->pull(policy, demand, dst);
}
void cancel() {
buf_->cancel();
buf_ = nullptr;
}
bool has_data() const noexcept {
return buf_->has_data();
}
/// Tries to open the resource for reading.
/// @returns a connected adapter that reads from the resource on success,
/// `nullptr` otherwise.
template <class Resource>
static intrusive_ptr<consumer_adapter>
try_open(socket_manager* owner, Resource src) {
CAF_ASSERT(owner != nullptr);
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<consumer_adapter>;
auto adapter = ptr_type{new consumer_adapter(owner, buf), false};
buf->set_consumer(adapter);
return adapter;
} else {
return nullptr;
}
}
friend void intrusive_ptr_add_ref(const consumer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const consumer_adapter* ptr) noexcept {
ptr->deref();
}
private:
consumer_adapter(socket_manager* owner, buf_ptr buf)
: mgr_(owner), buf_(std::move(buf)) {
// nop
}
auto strong_this() {
return intrusive_ptr{this};
}
void on_wakeup() {
if (buf_ && buf_->has_consumer_event()) {
mgr_->mpx().register_writing(mgr_);
}
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
};
template <class T>
using consumer_adapter_ptr = intrusive_ptr<consumer_adapter<T>>;
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/net/network_socket.hpp"
#include <cstddef>
#include <variant>
namespace caf::net {
/// A datagram-oriented network communication endpoint.
struct CAF_NET_EXPORT datagram_socket : network_socket {
using super = network_socket;
using super::super;
};
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates datagram_socket
error CAF_NET_EXPORT allow_connreset(datagram_socket x, bool new_value);
/// Converts the result from I/O operation on a ::datagram_socket to either an
/// error code or a integer greater or equal to zero.
/// @relates datagram_socket
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 <deque>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
namespace caf::net {
template <class Factory>
using datagram_transport_base
= transport_base<datagram_transport<Factory>,
transport_worker_dispatcher<Factory, ip_endpoint>,
udp_datagram_socket, Factory, ip_endpoint>;
/// Implements a udp_transport policy that manages a datagram socket.
template <class Factory>
class datagram_transport : public datagram_transport_base<Factory> {
public:
// Maximal UDP-packet size
static constexpr size_t max_datagram_size
= std::numeric_limits<uint16_t>::max();
// -- member types -----------------------------------------------------------
using factory_type = Factory;
using id_type = ip_endpoint;
using application_type = typename factory_type::application_type;
using super = datagram_transport_base<factory_type>;
using buffer_cache_type = typename super::buffer_cache_type;
// -- constructors, destructors, and assignment operators --------------------
datagram_transport(udp_datagram_socket handle, factory_type factory)
: super(handle, std::move(factory)) {
// nop
}
// -- public member functions ------------------------------------------------
error init(endpoint_manager& manager) override {
CAF_LOG_TRACE("");
if (auto err = super::init(manager))
return err;
prepare_next_read();
return none;
}
bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) {
auto ret = read(this->handle_, this->read_buf_);
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto& [num_bytes, ep] = *res;
CAF_LOG_DEBUG("received " << num_bytes << " bytes");
this->read_buf_.resize(num_bytes);
if (auto err = this->next_layer_.handle_data(*this, this->read_buf_,
std::move(ep))) {
CAF_LOG_ERROR("handle_data failed: " << err);
return false;
}
prepare_next_read();
} else {
auto err = get<sec>(ret);
if (err == sec::unavailable_or_would_block) {
break;
} else {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
}
}
return true;
}
bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("socket", this->handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size()));
auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
return true;
}
return false;
};
do {
if (auto err = write_some())
return err == sec::unavailable_or_would_block;
} while (fetch_next_message());
return !packet_queue_.empty();
}
// TODO: remove this function. `resolve` should add workers when needed.
error add_new_worker(node_id node, id_type id) {
auto worker = this->next_layer_.add_new_worker(*this, node, id);
if (!worker)
return worker.error();
return none;
}
void write_packet(id_type id, span<byte_buffer*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty())
this->manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
packet_queue_.emplace_back(id, buffers);
}
/// Helper struct for managing outgoing packets
struct packet {
id_type id;
buffer_cache_type bytes;
size_t size;
packet(id_type id, span<byte_buffer*> bufs) : id(id) {
size = 0;
for (auto buf : bufs) {
size += buf->size();
bytes.emplace_back(std::move(*buf));
}
}
std::vector<byte_buffer*> get_buffer_ptrs() {
std::vector<byte_buffer*> ptrs;
for (auto& buf : bytes)
ptrs.emplace_back(&buf);
return ptrs;
}
};
private:
// -- utility functions ------------------------------------------------------
void prepare_next_read() {
this->read_buf_.resize(max_datagram_size);
}
error write_some() {
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = packet_queue_.front();
auto& bufs = front.bytes;
auto it = bufs.begin();
if (this->header_bufs_.size() < this->header_bufs_.capacity()) {
it->clear();
this->header_bufs_.emplace_back(std::move(*it++));
}
for (; it != bufs.end()
&& this->payload_bufs_.size() < this->payload_bufs_.capacity();
++it) {
it->clear();
this->payload_bufs_.emplace_back(std::move(*it));
}
packet_queue_.pop_front();
};
// Write as many bytes as possible.
while (!packet_queue_.empty()) {
auto& packet = packet_queue_.front();
auto ptrs = packet.get_buffer_ptrs();
auto write_ret = write(this->handle_, ptrs, packet.id);
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
CAF_LOG_WARNING_IF(*num_bytes < packet.size,
"packet was not sent completely");
recycle();
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_ERROR("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
}
return err;
}
}
return none;
}
std::deque<packet> packet_queue_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstddef>
#include <cstdint>
// -- hard-coded default values for various CAF options ------------------------
namespace caf::defaults::middleman {
/// Maximum number of cached buffers for sending payloads.
constexpr size_t max_payload_buffers = 100;
/// Maximum number of cached buffers for sending headers.
constexpr size_t max_header_buffers = 10;
/// Port to listen on for tcp.
constexpr uint16_t tcp_port = 0;
} // namespace caf::defaults::middleman
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <memory>
#include "caf/intrusive_ptr.hpp"
#include "caf/type_id.hpp"
namespace caf::net {
// -- templates ----------------------------------------------------------------
template <class UpperLayer>
class stream_transport;
template <class Factory>
class datagram_transport;
template <class Application, class IdType = unit_t>
class transport_worker;
template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher;
template <class... Sigs>
class typed_actor_shell;
template <class... Sigs>
class typed_actor_shell_ptr;
// -- classes ------------------------------------------------------------------
class actor_shell;
class actor_shell_ptr;
class middleman;
class multiplexer;
class socket_manager;
// -- structs ------------------------------------------------------------------
struct network_socket;
struct pipe_socket;
struct socket;
struct stream_socket;
struct tcp_accept_socket;
struct tcp_stream_socket;
struct datagram_socket;
struct udp_datagram_socket;
// -- smart pointer aliases ----------------------------------------------------
using 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
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_transport_error.hpp"
namespace caf::net {
template <class OnSuccess, class OnError>
struct default_handshake_worker_factory {
OnSuccess make;
OnError abort;
};
/// An connect worker calls an asynchronous `connect` callback until it
/// succeeds. On success, the worker calls a factory object to transfer
/// ownership of socket and communication policy to the create the socket
/// manager that takes care of the established connection.
template <bool IsServer, class Socket, class Policy, class Factory>
class handshake_worker : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using super = socket_manager;
using read_result = typename super::read_result;
using write_result = typename super::write_result;
handshake_worker(Socket handle, multiplexer* parent, Policy policy,
Factory factory)
: super(handle, parent),
policy_(std::move(policy)),
factory_(std::move(factory)) {
// nop
}
// -- interface functions ----------------------------------------------------
error init(const settings& config) override {
cfg_ = config;
register_writing();
return caf::none;
}
read_result handle_read_event() override {
auto fd = socket_cast<Socket>(this->handle());
if (auto res = advance_handshake(fd); res > 0) {
return read_result::handover;
} else if (res == 0) {
factory_.abort(make_error(sec::connection_closed));
return read_result::stop;
} else {
auto err = policy_.last_error(fd, res);
switch (err) {
case stream_transport_error::want_read:
case stream_transport_error::temporary:
return read_result::again;
case stream_transport_error::want_write:
return read_result::want_write;
default:
auto err = make_error(sec::cannot_connect_to_node,
policy_.fetch_error_str());
factory_.abort(std::move(err));
return read_result::stop;
}
}
}
read_result handle_buffered_data() override {
return read_result::again;
}
read_result handle_continue_reading() override {
return read_result::again;
}
write_result handle_write_event() override {
auto fd = socket_cast<Socket>(this->handle());
if (auto res = advance_handshake(fd); res > 0) {
return write_result::handover;
} else if (res == 0) {
factory_.abort(make_error(sec::connection_closed));
return write_result::stop;
} else {
switch (policy_.last_error(fd, res)) {
case stream_transport_error::want_write:
case stream_transport_error::temporary:
return write_result::again;
case stream_transport_error::want_read:
return write_result::want_read;
default:
auto err = make_error(sec::cannot_connect_to_node,
policy_.fetch_error_str());
factory_.abort(std::move(err));
return write_result::stop;
}
}
}
write_result handle_continue_writing() override {
return write_result::again;
}
void handle_error(sec code) override {
factory_.abort(make_error(code));
}
socket_manager_ptr make_next_manager(socket hdl) override {
auto ptr = factory_.make(socket_cast<Socket>(hdl), this->mpx_ptr(),
std::move(policy_));
if (ptr) {
if (auto err = ptr->init(cfg_)) {
factory_.abort(err);
return nullptr;
} else {
return ptr;
}
} else {
factory_.abort(make_error(sec::runtime_error, "factory_.make failed"));
return nullptr;
}
}
private:
ptrdiff_t advance_handshake(Socket fd) {
if constexpr (IsServer)
return policy_.accept(fd);
else
return policy_.connect(fd);
}
settings cfg_;
Policy policy_;
Factory factory_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net {
struct CAF_NET_EXPORT this_host {
/// Initializes the network subsystem.
static error startup();
/// Release any resources of the network subsystem.
static void cleanup();
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
namespace caf::net::http {
/// Stores context information for a request. For HTTP/2, this is the stream ID.
struct context {};
} // namespace caf::net::http
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/config_value.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/method.hpp"
#include "caf/net/http/status.hpp"
#include "caf/uri.hpp"
#include <string_view>
#include <vector>
namespace caf::net::http {
class CAF_NET_EXPORT header {
public:
header() = default;
header(header&&) = default;
header& operator=(header&&) = default;
header(const header&);
header& operator=(const header&);
void assign(const header&);
http::method method() const noexcept {
return method_;
}
std::string_view path() const noexcept {
return uri_.path();
}
const uri::query_map& query() const noexcept {
return uri_.query();
}
std::string_view fragment() const noexcept {
return uri_.fragment();
}
std::string_view version() const noexcept {
return version_;
}
const header_fields_map& fields() const noexcept {
return fields_;
}
std::string_view field(std::string_view key) const noexcept {
if (auto i = fields_.find(key); i != fields_.end())
return i->second;
else
return {};
}
template <class T>
std::optional<T> field_as(std::string_view key) const noexcept {
if (auto i = fields_.find(key); i != fields_.end()) {
caf::config_value val{std::string{i->second}};
if (auto res = caf::get_as<T>(val))
return std::move(*res);
else
return {};
} else {
return {};
}
}
bool valid() const noexcept {
return !raw_.empty();
}
std::pair<status, std::string_view> parse(std::string_view raw);
bool chunked_transfer_encoding() const;
std::optional<size_t> content_length() const;
private:
std::vector<char> raw_;
http::method method_;
uri uri_;
std::string_view version_;
header_fields_map fields_;
};
} // namespace caf::net::http
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/unordered_flat_map.hpp"
#include <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<std::string_view, std::string_view>;
} // namespace caf::net::http
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include <cstdint>
#include <string>
#include <string_view>
#include <type_traits>
namespace caf::net::http {
/// Methods as defined by RFC 7231.
enum class method : uint8_t {
/// Requests transfer of a current selected representation for the target
/// resource.
get,
/// Identical to GET except that the server MUST NOT send a message body in
/// the response. The server SHOULD send the same header fields in response to
/// a HEAD request as it would have sent if the request had been a GET, except
/// that the payload header fields (Section 3.3) MAY be omitted.
head,
/// Requests that the target resource process the representation enclosed in
/// the request according to the resource's own specific semantics.
post,
/// Requests that the state of the target resource be created or replaced with
/// the state defined by the representation enclosed in the request message
/// payload.
put,
/// Requests that the origin server remove the association between the target
/// resource and its current functionality.
del,
/// Requests that the recipient establish a tunnel to the destination origin
/// server identified by the request-target and, if successful, thereafter
/// restrict its behavior to blind forwarding of packets, in both directions,
/// until the tunnel is closed.
connect,
/// Requests information about the communication options available for the
/// target resource, at either the origin server or an intervening
/// intermediary.
options,
/// Requests a remote, application-level loop-back of the request message.
trace,
};
/// @relates method
CAF_NET_EXPORT std::string to_string(method);
/// Converts @p x to the RFC string representation, i.e., all-uppercase.
/// @relates method
CAF_NET_EXPORT std::string to_rfc_string(method x);
/// @relates method
CAF_NET_EXPORT bool from_string(std::string_view, method&);
/// @relates method
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<method>, method&);
/// @relates method
template <class Inspector>
bool inspect(Inspector& f, method& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net::http
This diff is collapsed.
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/string_view.hpp"
#include <cstdint>
#include <string>
#include <type_traits>
namespace caf::net::http {
/// Status codes as defined by RFC 7231 and RFC 6585.
enum class status : uint16_t {
/// Indicates that the initial part of a request has been received and has not
/// yet been rejected by the server. The server intends to send a final
/// response after the request has been fully received and acted upon.
continue_request = 100,
/// Indicates that the server understands and is willing to comply with the
/// client's request for a change in the application protocol being used on
/// this connection.
switching_protocols = 101,
/// Indicates that the request has succeeded.
ok = 200,
/// Indicates that the request has been fulfilled and has resulted in one or
/// more new resources being created.
created = 201,
/// Indicates that the request has been accepted for processing, but the
/// processing has not been completed.
accepted = 202,
/// Indicates that the request was successful but the enclosed payload has
/// been modified from that of the origin server's 200 (OK) response by a
/// transforming proxy
non_authoritative_information = 203,
/// Indicates that the server has successfully fulfilled the request and that
/// there is no additional content to send in the response payload body.
no_content = 204,
/// Indicates that the server has fulfilled the request and desires that the
/// user agent reset the "document view".
reset_content = 205,
/// Indicates that the server is successfully fulfilling a range request for
/// the target resource by transferring one or more parts of the selected
/// representation that correspond to the satisfiable ranges found in the
/// request's Range header field.
partial_content = 206,
/// Indicates that the target resource has more than one representation and
/// information about the alternatives is being provided so that the user (or
/// user agent) can select a preferred representation.
multiple_choices = 300,
/// Indicates that the target resource has been assigned a new permanent URI.
moved_permanently = 301,
/// Indicates that the target resource resides temporarily under a different
/// URI.
found = 302,
/// Indicates that the server is redirecting the user agent to a different
/// resource, as indicated by a URI in the Location header field, which is
/// intended to provide an indirect response to the original request.
see_other = 303,
/// Indicates that a conditional GET or HEAD request has been received and
/// would have resulted in a 200 (OK) response if it were not for the fact
/// that the condition evaluated to false.
not_modified = 304,
/// Deprecated.
use_proxy = 305,
/// No longer valid.
temporary_redirect = 307,
/// Indicates that the server cannot or will not process the request due to
/// something that is perceived to be a client error (e.g., malformed request
/// syntax, invalid request message framing, or deceptive request routing).
bad_request = 400,
/// Indicates that the request has not been applied because it lacks valid
/// authentication credentials for the target resource.
unauthorized = 401,
/// Reserved for future use.
payment_required = 402,
/// Indicates that the server understood the request but refuses to authorize
/// it.
forbidden = 403,
/// Indicates that the origin server did not find a current representation for
/// the target resource or is not willing to disclose that one exists.
not_found = 404,
/// Indicates that the method received in the request-line is known by the
/// origin server but not supported by the target resource.
method_not_allowed = 405,
/// Indicates that the target resource does not have a current representation
/// that would be acceptable to the user agent, according to the proactive
/// negotiation header fields received in the request (Section 5.3), and the
/// server is unwilling to supply a default representation.
not_acceptable = 406,
/// Similar to 401 (Unauthorized), but it indicates that the client needs to
/// authenticate itself in order to use a proxy.
proxy_authentication_required = 407,
/// Indicates that the server did not receive a complete request message
/// within the time that it was prepared to wait.
request_timeout = 408,
/// Indicates that the request could not be completed due to a conflict with
/// the current state of the target resource. This code is used in situations
/// where the user might be able to resolve the conflict and resubmit the
/// request.
conflict = 409,
/// Indicates that access to the target resource is no longer available at the
/// origin server and that this condition is likely to be permanent.
gone = 410,
/// Indicates that the server refuses to accept the request without a defined
/// Content-Length.
length_required = 411,
/// Indicates that one or more conditions given in the request header fields
/// evaluated to false when tested on the server.
precondition_failed = 412,
/// Indicates that the server is refusing to process a request because the
/// request payload is larger than the server is willing or able to process.
payload_too_large = 413,
/// Indicates that the server is refusing to service the request because the
/// request-target is longer than the server is willing to interpret.
uri_too_long = 414,
/// Indicates that the origin server is refusing to service the request
/// because the payload is in a format not supported by this method on the
/// target resource.
unsupported_media_type = 415,
/// Indicates that none of the ranges in the request's Range header field
/// overlap the current extent of the selected resource or that the set of
/// ranges requested has been rejected due to invalid ranges or an excessive
/// request of small or overlapping ranges.
range_not_satisfiable = 416,
/// Indicates that the expectation given in the request's Expect header field
/// could not be met by at least one of the inbound servers.
expectation_failed = 417,
/// Indicates that the server refuses to perform the request using the current
/// protocol but might be willing to do so after the client upgrades to a
/// different protocol.
upgrade_required = 426,
/// Indicates that the origin server requires the request to be conditional.
precondition_required = 428,
/// Indicates that the user has sent too many requests in a given amount of
/// time ("rate limiting").
too_many_requests = 429,
/// Indicates that the server is unwilling to process the request because its
/// header fields are too large.
request_header_fields_too_large = 431,
/// Indicates that the server encountered an unexpected condition that
/// prevented it from fulfilling the request.
internal_server_error = 500,
/// Indicates that the server does not support the functionality required to
/// fulfill the request.
not_implemented = 501,
/// Indicates that the server, while acting as a gateway or proxy, received an
/// invalid response from an inbound server it accessed while attempting to
/// fulfill the request.
bad_gateway = 502,
/// Indicates that the server is currently unable to handle the request due to
/// a temporary overload or scheduled maintenance, which will likely be
/// alleviated after some delay.
service_unavailable = 503,
/// Indicates that the server, while acting as a gateway or proxy, did not
/// receive a timely response from an upstream server it needed to access in
/// order to complete the request.
gateway_timeout = 504,
/// Indicates that the server does not support, or refuses to support, the
/// major version of HTTP that was used in the request message.
http_version_not_supported = 505,
/// Indicates that the client needs to authenticate to gain network access.
network_authentication_required = 511,
};
/// Returns the recommended response phrase to a status code.
/// @relates status
CAF_NET_EXPORT std::string_view phrase(status);
/// @relates status
CAF_NET_EXPORT std::string to_string(status);
/// @relates 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&);
/// @relates status
template <class Inspector>
bool inspect(Inspector& f, status& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net::http
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/status.hpp"
#include <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 `string_view` as `first` for
/// incomplete HTTP headers.
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,
byte_buffer& buf);
/// Writes a complete HTTP response to the buffer. Automatically sets
/// Content-Type and Content-Length header fields.
CAF_NET_EXPORT void write_response(status code, 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, std::string_view content_type,
std::string_view content,
const header_fields_map& fields,
byte_buffer& buf);
} // namespace caf::net::http::v1
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/byte_span.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a hypertext-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// hypertext-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class hypertext_oriented_layer_ptr {
public:
using context_type = typename Layer::context_type;
using status_code_type = typename Layer::status_code_type;
using header_fields_type = typename Layer::header_fields_type;
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
/// Queries whether the underlying transport can send additional data.
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
/// Asks the underlying transport to stop receiving additional data until
/// resumed.
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
/// Returns the socket handle.
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
/// Sends a response header for answering the request identified by the
/// context.
/// @param context Identifies which request this response belongs to.
/// @param code Indicates either success or failure to the client.
/// @param fields Various informational fields for the client. When sending
/// a payload afterwards, the fields should at least include
/// the content length.
bool send_header(context_type context, status_code_type code,
const header_fields_type& fields) {
return lptr_->send_header(llptr_, context, code, fields);
}
/// Sends a payload to the client. Must follow a header.
/// @param context Identifies which request this response belongs to.
/// @param bytes Arbitrary data for the client.
/// @pre `bytes.size() > 0`
[[nodiscard]] bool send_payload(context_type context,
const_byte_span bytes) {
return lptr_->send_payload(llptr_, context, bytes);
}
/// Sends a single chunk of arbitrary data. The chunks must follow a header.
/// @pre `bytes.size() > 0`
[[nodiscard]] bool send_chunk(context_type context, const_byte_span bytes) {
return lptr_->send_chunk(llptr_, context, bytes);
}
/// Informs the client that the transfer completed, i.e., that the server
/// will not send additional chunks.
[[nodiscard]] bool send_end_of_chunks(context_type context) {
return lptr_->send_end_of_chunks(llptr_, context);
}
/// Convenience function for completing a request 'raw' (without adding
/// additional header fields) in a single function call. Calls
/// `send_header`, `send_payload` and `fin`.
bool send_raw_response(context_type context, status_code_type code,
const header_fields_type& fields,
const_byte_span content) {
if (lptr_->send_header(llptr_, context, code, fields)
&& (content.empty()
|| lptr_->send_payload(llptr_, context, content))) {
lptr_->fin(llptr_, context);
return true;
} else {
return false;
}
}
/// Convenience function for completing a request in a single function call.
/// Automatically sets the header fields 'Content-Type' and
/// 'Content-Length'. Calls `send_header`, `send_payload` and `fin`.
bool send_response(context_type context, status_code_type code,
header_fields_type fields, std::string_view content_type,
const_byte_span content) {
std::string len;
if (!content.empty()) {
auto len = std::to_string(content.size());
fields.emplace("Content-Length", len);
fields.emplace("Content-Type", content_type);
}
return send_raw_response(context, code, fields, content);
}
/// Convenience function for completing a request in a single function call.
/// Automatically sets the header fields 'Content-Type' and
/// 'Content-Length'. Calls `send_header`, `send_payload` and `fin`.
bool send_response(context_type context, status_code_type code,
std::string_view content_type, const_byte_span content) {
std::string len;
header_fields_type fields;
if (!content.empty()) {
len = std::to_string(content.size());
fields.emplace("Content-Type", content_type);
fields.emplace("Content-Length", len);
}
return send_raw_response(context, code, fields, content);
}
bool send_response(context_type context, status_code_type code,
header_fields_type fields, 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,
std::string_view content_type,
std::string_view content) {
return send_response(context, code, content_type,
as_bytes(make_span(content)));
}
void fin(context_type context) {
lptr_->fin(llptr_, context);
}
/// Sets an abort reason on the transport.
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
/// Returns the current abort reason on the transport or a
/// default-constructed error is no error occurred yet.
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
hypertext_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
hypertext_oriented_layer_ptr(const hypertext_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_hypertext_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
using result_t = hypertext_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "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(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(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);
/// Returns the hostname of this device.
std::string CAF_NET_EXPORT hostname();
} // namespace caf::net::ip
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstdint>
#include <cstring>
#include <memory>
#include <type_traits>
#include "caf/byte.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/has_after_reading.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/error.hpp"
#include "caf/net/message_flow_bridge.hpp"
#include "caf/net/message_oriented_layer_ptr.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/tag/message_oriented.hpp"
#include "caf/tag/no_auto_reading.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace caf::net {
/// Length-prefixed message framing for discretizing a Byte stream into messages
/// of varying size. The framing uses 4 Bytes for the length prefix, but
/// messages (including the 4 Bytes for the length prefix) are limited to a
/// maximum size of INT32_MAX. This limitation comes from the POSIX API (recv)
/// on 32-bit platforms.
template <class UpperLayer>
class length_prefix_framing {
public:
using input_tag = tag::stream_oriented;
using output_tag = tag::message_oriented;
using length_prefix_type = uint32_t;
static constexpr size_t hdr_size = sizeof(uint32_t);
static constexpr size_t max_message_length = INT32_MAX - sizeof(uint32_t);
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
explicit length_prefix_framing(Ts&&... xs)
: upper_layer_(std::forward<Ts>(xs)...) {
// nop
}
// -- initialization ---------------------------------------------------------
template <class LowerLayerPtr>
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
if constexpr (!std::is_base_of_v<tag::no_auto_reading, UpperLayer>)
down->configure_read(receive_policy::exactly(hdr_size));
return upper_layer_.init(owner, this_layer_ptr(down), cfg);
}
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_;
}
const auto& upper_layer() const noexcept {
return upper_layer_;
}
// -- interface for the upper layer ------------------------------------------
template <class LowerLayerPtr>
static bool can_send_more(LowerLayerPtr down) noexcept {
return down->can_send_more();
}
template <class LowerLayerPtr>
static auto handle(LowerLayerPtr down) noexcept {
return down->handle();
}
template <class LowerLayerPtr>
static void suspend_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::stop());
}
template <class LowerLayerPtr>
void begin_message(LowerLayerPtr down) {
down->begin_output();
auto& buf = down->output_buffer();
message_offset_ = buf.size();
buf.insert(buf.end(), 4, std::byte{0});
}
template <class LowerLayerPtr>
byte_buffer& message_buffer(LowerLayerPtr down) {
return down->output_buffer();
}
template <class LowerLayerPtr>
[[nodiscard]] bool end_message(LowerLayerPtr down) {
using detail::to_network_order;
auto& buf = down->output_buffer();
CAF_ASSERT(message_offset_ < buf.size());
auto msg_begin = buf.begin() + message_offset_;
auto msg_size = std::distance(msg_begin + 4, buf.end());
if (msg_size > 0 && static_cast<size_t>(msg_size) < max_message_length) {
auto u32_size = to_network_order(static_cast<uint32_t>(msg_size));
memcpy(std::addressof(*msg_begin), &u32_size, 4);
down->end_output();
return true;
} else {
auto err = make_error(sec::runtime_error,
msg_size == 0 ? "logic error: message of size 0"
: "maximum message size exceeded");
down->abort_reason(std::move(err));
return false;
}
}
template <class LowerLayerPtr>
bool send_close_message(LowerLayerPtr) {
// nop; this framing layer has no close handshake
return true;
}
template <class LowerLayerPtr>
bool send_close_message(LowerLayerPtr, const error&) {
// nop; this framing layer has no close handshake
return true;
}
template <class LowerLayerPtr>
static void abort_reason(LowerLayerPtr down, error reason) {
return down->abort_reason(std::move(reason));
}
template <class LowerLayerPtr>
static const error& abort_reason(LowerLayerPtr down) {
return down->abort_reason();
}
// -- interface for the lower layer ------------------------------------------
template <class LowerLayerPtr>
void continue_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::exactly(hdr_size));
}
template <class LowerLayerPtr>
std::enable_if_t<detail::has_after_reading_v<
UpperLayer,
message_oriented_layer_ptr<length_prefix_framing, LowerLayerPtr>>>
after_reading(LowerLayerPtr down) {
return upper_layer_.after_reading(this_layer_ptr(down));
}
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
return upper_layer_.prepare_send(this_layer_ptr(down));
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr down) {
return upper_layer_.done_sending(this_layer_ptr(down));
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr down, const error& reason) {
upper_layer_.abort(this_layer_ptr(down), reason);
}
template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span input, byte_span) {
CAF_LOG_TRACE("got" << input.size() << "bytes");
auto this_layer = this_layer_ptr(down);
if (input.size() < sizeof(uint32_t)) {
auto err = make_error(sec::runtime_error,
"received too few bytes from underlying transport");
down->abort_reason(std::move(err));
return -1;
} else if (input.size() == hdr_size) {
auto u32_size = uint32_t{0};
memcpy(&u32_size, input.data(), sizeof(uint32_t));
auto msg_size = static_cast<size_t>(detail::from_network_order(u32_size));
if (msg_size == 0) {
// Ignore empty messages.
CAF_LOG_DEBUG("received empty message");
return static_cast<ptrdiff_t>(input.size());
} else if (msg_size > max_message_length) {
CAF_LOG_DEBUG("maximum message size exceeded");
auto err = make_error(sec::runtime_error,
"maximum message size exceeded");
down->abort_reason(std::move(err));
return -1;
} else {
CAF_LOG_DEBUG("wait for payload of size" << msg_size);
down->configure_read(receive_policy::exactly(hdr_size + msg_size));
return 0;
}
} else {
auto [msg_size, msg] = split(input);
if (msg_size == msg.size() && msg_size + hdr_size == input.size()) {
CAF_LOG_DEBUG("got message of size" << msg_size);
if (upper_layer_.consume(this_layer, msg) >= 0) {
if (!down->stopped())
down->configure_read(receive_policy::exactly(hdr_size));
return static_cast<ptrdiff_t>(input.size());
} else {
return -1;
}
} else {
CAF_LOG_DEBUG("received malformed message");
auto err = make_error(sec::runtime_error, "received malformed message");
down->abort_reason(std::move(err));
return -1;
}
}
}
// -- convenience functions --------------------------------------------------
static std::pair<size_t, byte_span> split(byte_span buffer) noexcept {
CAF_ASSERT(buffer.size() >= sizeof(uint32_t));
auto u32_size = uint32_t{0};
memcpy(&u32_size, buffer.data(), sizeof(uint32_t));
auto msg_size = static_cast<size_t>(detail::from_network_order(u32_size));
return std::make_pair(msg_size, buffer.subspan(sizeof(uint32_t)));
}
private:
// -- implementation details -------------------------------------------------
template <class LowerLayerPtr>
auto this_layer_ptr(LowerLayerPtr down) {
return make_message_oriented_layer_ptr(this, down);
}
// -- member variables -------------------------------------------------------
UpperLayer upper_layer_;
size_t message_offset_ = 0;
};
// -- high-level factory functions -------------------------------------------
/// Runs a WebSocket server on the connected socket `fd`.
/// @param mpx The multiplexer that takes ownership of the socket.
/// @param fd A connected stream socket.
/// @param cfg Additional configuration parameters for the protocol stack.
/// @param in Inputs for writing to the socket.
/// @param out Outputs from the socket.
/// @param trait Converts between the native and the wire format.
/// @relates length_prefix_framing
template <template <class> class Transport = stream_transport, class Socket,
class T, class Trait, class... TransportArgs>
error run_with_length_prefix_framing(multiplexer& mpx, Socket fd,
const settings& cfg,
async::consumer_resource<T> in,
async::producer_resource<T> out,
Trait trait, TransportArgs&&... args) {
using app_t = Transport<length_prefix_framing<message_flow_bridge<T, Trait>>>;
auto mgr = make_socket_manager<app_t>(fd, &mpx,
std::forward<TransportArgs>(args)...,
std::move(in), std::move(out),
std::move(trait));
return mgr->init(cfg);
}
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/net/consumer_adapter.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/producer_adapter.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/settings.hpp"
#include "caf/tag/message_oriented.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/no_auto_reading.hpp"
#include <utility>
namespace caf::net {
/// Translates between a message-oriented transport and data flows.
///
/// The trait class converts between the native and the wire format:
///
/// ~~~
/// struct my_trait {
/// bool convert(const T& value, byte_buffer& bytes);
/// bool convert(const_byte_span bytes, T& value);
/// };
/// ~~~
template <class T, class Trait, class Tag = tag::message_oriented>
class message_flow_bridge : public tag::no_auto_reading {
public:
using input_tag = Tag;
using buffer_type = async::spsc_buffer<T>;
using consumer_resource_t = async::consumer_resource<T>;
using producer_resource_t = async::producer_resource<T>;
message_flow_bridge(consumer_resource_t in_res, producer_resource_t out_res,
Trait trait)
: in_res_(std::move(in_res)),
trait_(std::move(trait)),
out_res_(std::move(out_res)) {
// nop
}
explicit message_flow_bridge(Trait trait) : trait_(std::move(trait)) {
// nop
}
template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr, const settings& cfg) {
mgr_ = mgr;
if constexpr (caf::detail::has_init_v<Trait>) {
if (auto err = init_res(trait_.init(cfg)))
return err;
}
if (in_res_) {
in_ = consumer_adapter<buffer_type>::try_open(mgr, in_res_);
in_res_ = nullptr;
}
if (out_res_) {
out_ = producer_adapter<buffer_type>::try_open(mgr, out_res_);
out_res_ = nullptr;
}
if (!in_ && !out_)
return make_error(sec::cannot_open_resource,
"flow bridge cannot run without at least one resource");
return none;
}
template <class LowerLayerPtr>
bool write(LowerLayerPtr down, const T& item) {
if constexpr (std::is_same_v<Tag, tag::message_oriented>) {
down->begin_message();
auto& buf = down->message_buffer();
return trait_.convert(item, buf) && down->end_message();
} else {
static_assert(std::is_same_v<Tag, tag::mixed_message_oriented>);
if (trait_.converts_to_binary(item)) {
down->begin_binary_message();
auto& bytes = down->binary_message_buffer();
return trait_.convert(item, bytes) && down->end_binary_message();
} else {
down->begin_text_message();
auto& text = down->text_message_buffer();
return trait_.convert(item, text) && down->end_text_message();
}
}
}
template <class LowerLayerPtr>
struct write_helper {
using bridge_type = message_flow_bridge;
bridge_type* bridge;
LowerLayerPtr down;
bool aborted = false;
size_t consumed = 0;
error err;
write_helper(bridge_type* bridge, LowerLayerPtr down)
: bridge(bridge), down(down) {
// nop
}
void on_next(const T& item) {
if (!bridge->write(down, item)) {
aborted = true;
return;
}
}
void on_complete() {
// nop
}
void on_error(const error& x) {
err = x;
}
};
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
write_helper<LowerLayerPtr> helper{this, down};
while (down->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down->send_close_message(helper.err);
} else {
down->send_close_message();
}
in_ = nullptr;
} else if (helper.aborted) {
down->abort_reason(make_error(sec::conversion_failed));
in_->cancel();
in_ = nullptr;
return false;
} else if (consumed == 0) {
return true;
}
}
return true;
}
template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) {
return !in_ || !in_->has_data();
}
template <class LowerLayerPtr>
void abort(LowerLayerPtr, const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::socket_disconnected || reason == sec::disposed)
out_->close();
else
out_->abort(reason);
out_ = nullptr;
}
if (in_) {
in_->cancel();
in_ = nullptr;
}
}
template <class U = Tag, class LowerLayerPtr>
ptrdiff_t consume(LowerLayerPtr down, byte_span buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
return -1;
}
T val;
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
return -1;
}
if (out_->push(std::move(val)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
template <class U = Tag, class LowerLayerPtr>
ptrdiff_t consume_binary(LowerLayerPtr down, byte_span buf) {
return consume(down, buf);
}
template <class U = Tag, class LowerLayerPtr>
ptrdiff_t consume_text(LowerLayerPtr down, std::string_view buf) {
if (!out_) {
down->abort_reason(make_error(sec::connection_closed));
return -1;
}
T val;
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
return -1;
}
if (out_->push(std::move(val)) == 0)
down->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
private:
error init_res(error err) {
return err;
}
error init_res(consumer_resource_t in, producer_resource_t out) {
in_res_ = std::move(in);
out_res_ = std::move(out);
return caf::none;
}
error init_res(std::tuple<consumer_resource_t, producer_resource_t> in_out) {
auto& [in, out] = in_out;
return init_res(std::move(in), std::move(out));
}
error init_res(std::pair<consumer_resource_t, producer_resource_t> in_out) {
auto& [in, out] = in_out;
return init_res(std::move(in), std::move(out));
}
template <class R>
error init_res(expected<R> res) {
if (res)
return init_res(*res);
else
return std::move(res.error());
}
/// Points to the manager that runs this protocol stack.
net::socket_manager* mgr_ = nullptr;
/// Incoming messages, serialized to the socket.
consumer_adapter_ptr<buffer_type> in_;
/// Outgoing messages, deserialized from the socket.
producer_adapter_ptr<buffer_type> out_;
/// Converts between raw bytes and items.
Trait trait_;
/// Discarded after initialization.
consumer_resource_t in_res_;
/// Discarded after initialization.
producer_resource_t out_res_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a message-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_message() {
lptr_->begin_message(llptr_);
}
[[nodiscard]] auto& message_buffer() {
return lptr_->message_buffer(llptr_);
}
[[nodiscard]] bool end_message() {
return lptr_->end_message(llptr_);
}
template <class... Ts>
bool send_close_message(Ts&&... xs) {
return lptr_->send_close_message(llptr_, std::forward<Ts>(xs)...);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
message_oriented_layer_ptr(const message_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_message_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
using result_t = message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <chrono>
#include <set>
#include <string>
#include <thread>
#include "caf/actor_system.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/scoped_actor.hpp"
namespace caf::net {
class CAF_NET_EXPORT middleman : public actor_system::module {
public:
// -- member types -----------------------------------------------------------
using module = actor_system::module;
using module_ptr = actor_system::module_ptr;
// -- static utility functions -----------------------------------------------
static void init_global_meta_objects();
// -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
~middleman() override;
// -- socket manager functions -----------------------------------------------
/// Creates a new acceptor that accepts incoming connections from @p sock and
/// creates socket managers using @p factory.
/// @param sock An accept socket in listening mode. For a TCP socket, this
/// socket must already listen to an address plus port.
/// @param factory A function object for creating socket managers that take
/// ownership of incoming connections.
/// @param limit The maximum number of connections that this acceptor should
/// establish or 0 for 'no limit'.
template <class Socket, class Factory>
auto make_acceptor(Socket sock, Factory factory, size_t limit = 0) {
using connected_socket_type = typename Socket::connected_socket_type;
if constexpr (detail::is_callable_with<Factory, connected_socket_type,
multiplexer*>::value) {
connection_acceptor_factory_adapter<Factory> adapter{std::move(factory)};
return make_acceptor(std::move(sock), std::move(adapter), limit);
} else {
using impl = connection_acceptor<Socket, Factory>;
auto ptr = make_socket_manager<impl>(std::move(sock), &mpx_, limit,
std::move(factory));
mpx_.init(ptr);
return ptr;
}
}
// -- interface functions ----------------------------------------------------
void start() override;
void stop() override;
void init(actor_system_config&) override;
id_t id() const override;
void* subtype_ptr() override;
// -- factory functions ------------------------------------------------------
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);
// -- properties -------------------------------------------------------------
actor_system& system() {
return sys_;
}
const actor_system_config& config() const noexcept {
return sys_.config();
}
multiplexer& mpx() noexcept {
return mpx_;
}
const multiplexer& mpx() const noexcept {
return mpx_;
}
multiplexer* mpx_ptr() noexcept {
return &mpx_;
}
const multiplexer* mpx_ptr() const noexcept {
return &mpx_;
}
private:
// -- member variables -------------------------------------------------------
/// Points to the parent system.
actor_system& sys_;
/// Stores the global socket I/O multiplexer.
multiplexer mpx_;
/// Runs the multiplexer's event loop
std::thread mpx_thread_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a mixed-message-oriented layer with a pointer to its
/// lower layer. Both pointers are then used to implement the interface required
/// for a mixed-message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class mixed_message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void suspend_reading() {
return lptr_->suspend_reading(llptr_);
}
void begin_binary_message() {
lptr_->begin_binary_message(llptr_);
}
auto& binary_message_buffer() {
return lptr_->binary_message_buffer(llptr_);
}
bool end_binary_message() {
return lptr_->end_binary_message(llptr_);
}
void begin_text_message() {
lptr_->begin_text_message(llptr_);
}
auto& text_message_buffer() {
return lptr_->text_message_buffer(llptr_);
}
bool end_text_message() {
return lptr_->end_text_message(llptr_);
}
template <class... Ts>
bool send_close_message(Ts&&... xs) {
return lptr_->send_close_message(llptr_, std::forward<Ts>(xs)...);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
mixed_message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
mixed_message_oriented_layer_ptr(const mixed_message_oriented_layer_ptr&)
= default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_mixed_message_oriented_layer_ptr(Layer* this_layer,
LowerLayerPtr down) {
using result_t = mixed_message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <memory>
#include <mutex>
#include <thread>
#include "caf/action.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp"
extern "C" {
struct pollfd;
} // extern "C"
namespace caf::net {
class pollset_updater;
/// Multiplexes any number of ::socket_manager objects with a ::socket.
class CAF_NET_EXPORT multiplexer {
public:
// -- member types -----------------------------------------------------------
struct poll_update {
short events = 0;
socket_manager_ptr mgr;
};
using poll_update_map = detail::unordered_flat_map<socket, poll_update>;
using pollfd_list = std::vector<pollfd>;
using manager_list = std::vector<socket_manager_ptr>;
// -- friends ----------------------------------------------------------------
friend class pollset_updater; // Needs access to the `do_*` functions.
// -- static utility functions -----------------------------------------------
/// Blocks the PIPE signal on the current thread when running on a POSIX
/// windows. Has no effect when running on Windows.
static void block_sigpipe();
// -- constructors, destructors, and assignment operators --------------------
/// @param parent Points to the owning middleman instance. May be `nullptr`
/// only for the purpose of unit testing if no @ref
/// socket_manager requires access to the @ref middleman or the
/// @ref actor_system.
explicit multiplexer(middleman* parent);
~multiplexer();
// -- initialization ---------------------------------------------------------
error init();
// -- properties -------------------------------------------------------------
/// Returns the number of currently active socket managers.
size_t num_socket_managers() const noexcept;
/// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr);
/// Returns the index of `fd` in the pollset or `-1`.
ptrdiff_t index_of(socket fd);
/// Returns the owning @ref middleman instance.
middleman& owner();
/// Returns the enclosing @ref actor_system.
actor_system& system();
/// Computes the current mask for the manager. Mostly useful for testing.
operation mask_of(const socket_manager_ptr& mgr);
// -- thread-safe signaling --------------------------------------------------
/// Registers `mgr` for read events.
/// @thread-safe
void register_reading(const socket_manager_ptr& mgr);
/// Registers `mgr` for write events.
/// @thread-safe
void register_writing(const socket_manager_ptr& mgr);
/// Triggers a continue reading event for `mgr`.
/// @thread-safe
void continue_reading(const socket_manager_ptr& mgr);
/// Triggers a continue writing event for `mgr`.
/// @thread-safe
void continue_writing(const socket_manager_ptr& mgr);
/// Schedules a call to `mgr->handle_error(sec::discarded)`.
/// @thread-safe
void discard(const socket_manager_ptr& mgr);
/// Stops further reading by `mgr`.
/// @thread-safe
void shutdown_reading(const socket_manager_ptr& mgr);
/// Stops further writing by `mgr`.
/// @thread-safe
void shutdown_writing(const socket_manager_ptr& mgr);
/// Schedules an action for execution on this multiplexer.
/// @thread-safe
void schedule(const action& what);
/// Schedules an action for execution on this multiplexer.
/// @thread-safe
template <class F>
void schedule_fn(F f) {
schedule(make_action(std::move(f)));
}
/// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe
void init(const socket_manager_ptr& mgr);
/// Signals the multiplexer to initiate shutdown.
/// @thread-safe
void shutdown();
// -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become
/// 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();
/// Runs the multiplexer until no socket event handler remains active.
void run();
protected:
// -- utility functions ------------------------------------------------------
/// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, short events, short revents);
/// Transfers socket ownership from one manager to another.
void do_handover(const socket_manager_ptr& mgr);
/// Returns a change entry for the socket at given index. Lazily creates a new
/// entry before returning if necessary.
poll_update& update_for(ptrdiff_t index);
/// Returns a change entry for the socket of the manager.
poll_update& update_for(const socket_manager_ptr& mgr);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
template <class T>
void write_to_pipe(uint8_t opcode, T* ptr);
/// @copydoc write_to_pipe
template <class Enum, class T>
std::enable_if_t<std::is_enum_v<Enum>> write_to_pipe(Enum opcode, T* ptr) {
write_to_pipe(static_cast<uint8_t>(opcode), ptr);
}
/// Queries the currently active event bitmask for `mgr`.
short active_mask_of(const socket_manager_ptr& mgr);
/// Queries whether `mgr` is currently registered for reading.
bool is_reading(const socket_manager_ptr& mgr);
/// Queries whether `mgr` is currently registered for writing.
bool is_writing(const socket_manager_ptr& mgr);
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
pollfd_list pollset_;
/// Maps sockets to their owning managers by storing the managers in the same
/// order as their sockets appear in `pollset_`.
manager_list managers_;
/// Caches changes to the events mask of managed sockets until they can safely
/// take place.
poll_update_map updates_;
/// Stores the ID of the thread this multiplexer is running in. Set when
/// calling `init()`.
std::thread::id tid_;
/// Guards `write_handle_`.
std::mutex write_lock_;
/// Used for pushing updates to the multiplexer's thread.
pipe_socket write_handle_;
/// Points to the owning middleman.
middleman* owner_;
/// Signals whether shutdown has been requested.
bool shutting_down_ = false;
private:
// -- internal callbacks the pollset updater ---------------------------------
void do_shutdown();
void do_register_reading(const socket_manager_ptr& mgr);
void do_register_writing(const socket_manager_ptr& mgr);
void do_continue_reading(const socket_manager_ptr& mgr);
void do_continue_writing(const socket_manager_ptr& mgr);
void do_discard(const socket_manager_ptr& mgr);
void do_shutdown_reading(const socket_manager_ptr& mgr);
void do_shutdown_writing(const socket_manager_ptr& mgr);
void do_init(const socket_manager_ptr& mgr);
};
} // 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 <cstdint>
#include <string>
#include <system_error>
#include <type_traits>
#include "caf/config.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp"
namespace caf::net {
/// A bidirectional network communication endpoint.
struct CAF_NET_EXPORT network_socket : socket {
using super = socket;
using super::super;
};
/// Enables or disables `SIGPIPE` events from `x`.
/// @relates network_socket
error CAF_NET_EXPORT allow_sigpipe(network_socket x, bool new_value);
/// Enables or disables `SIO_UDP_CONNRESET`error on `x`.
/// @relates network_socket
error CAF_NET_EXPORT allow_udp_connreset(network_socket x, bool new_value);
/// Get the socket buffer size for `x`.
/// @pre `x != invalid_socket`
/// @relates network_socket
expected<size_t> CAF_NET_EXPORT send_buffer_size(network_socket x);
/// Set the socket buffer size for `x`.
/// @relates network_socket
error CAF_NET_EXPORT send_buffer_size(network_socket x, size_t capacity);
/// Returns the locally assigned port of `x`.
/// @relates network_socket
expected<uint16_t> CAF_NET_EXPORT local_port(network_socket x);
/// Returns the locally assigned address of `x`.
/// @relates network_socket
expected<std::string> CAF_NET_EXPORT local_addr(network_socket x);
/// Returns the port used by the remote host of `x`.
/// @relates network_socket
expected<uint16_t> CAF_NET_EXPORT remote_port(network_socket x);
/// Returns the remote host address of `x`.
/// @relates network_socket
expected<std::string> CAF_NET_EXPORT remote_addr(network_socket x);
/// Closes the read channel for a socket.
/// @relates network_socket
void CAF_NET_EXPORT shutdown_read(network_socket x);
/// Closes the write channel for a socket.
/// @relates network_socket
void CAF_NET_EXPORT shutdown_write(network_socket x);
/// Closes the both read and write channel for a socket.
/// @relates network_socket
void CAF_NET_EXPORT shutdown(network_socket x);
} // namespace caf::net
This diff is collapsed.
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/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.
enum class operation {
none = 0b0000,
read = 0b0001,
write = 0b0010,
block_read = 0b0100,
block_write = 0b1000,
read_write = 0b0011,
read_only = 0b1001,
write_only = 0b0110,
shutdown = 0b1100,
};
/// @relates operation
[[nodiscard]] constexpr int to_integer(operation x) noexcept {
return static_cast<int>(x);
}
/// Adds the `read` flag to `x` unless the `block_read` bit is set.
/// @relates operation
[[nodiscard]] constexpr operation add_read_flag(operation x) noexcept {
if (auto bits = to_integer(x); !(bits & 0b0100))
return static_cast<operation>(bits | 0b0001);
else
return x;
}
/// Adds the `write` flag to `x` unless the `block_write` bit is set.
/// @relates operation
[[nodiscard]] constexpr operation add_write_flag(operation x) noexcept {
if (auto bits = to_integer(x); !(bits & 0b1000))
return static_cast<operation>(bits | 0b0010);
else
return x;
}
/// Removes the `read` flag from `x`.
/// @relates operation
[[nodiscard]] constexpr operation remove_read_flag(operation x) noexcept {
return static_cast<operation>(to_integer(x) & 0b1110);
}
/// Removes the `write` flag from `x`.
/// @relates operation
[[nodiscard]] constexpr operation remove_write_flag(operation x) noexcept {
return static_cast<operation>(to_integer(x) & 0b1101);
}
/// Adds the `block_read` flag to `x` and removes the `read` flag if present.
/// @relates operation
[[nodiscard]] constexpr operation block_reads(operation x) noexcept {
auto bits = to_integer(x) | 0b0100;
return static_cast<operation>(bits & 0b1110);
}
/// Adds the `block_write` flag to `x` and removes the `write` flag if present.
/// @relates operation
[[nodiscard]] constexpr operation block_writes(operation x) noexcept {
auto bits = to_integer(x) | 0b1000;
return static_cast<operation>(bits & 0b1101);
}
/// Returns whether the read flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_reading(operation x) noexcept {
return (to_integer(x) & 0b0001) == 0b0001;
}
/// Returns whether the write flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_writing(operation x) noexcept {
return (to_integer(x) & 0b0010) == 0b0010;
}
/// Returns `!is_reading(x) && !is_writing(x)`.
/// @relates operation
[[nodiscard]] constexpr bool is_idle(operation x) noexcept {
return (to_integer(x) & 0b0011) == 0b0000;
}
/// Returns whether the block read flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_read_blocked(operation x) noexcept {
return (to_integer(x) & 0b0100) == 0b0100;
}
/// Returns whether the block write flag is present in `x`.
/// @relates operation
[[nodiscard]] constexpr bool is_write_blocked(operation x) noexcept {
return (to_integer(x) & 0b1000) == 0b1000;
}
/// @relates operation
CAF_NET_EXPORT std::string to_string(operation x);
/// @relates 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&);
/// @relates operation
template <class Inspector>
bool inspect(Inspector& f, operation& x) {
return default_enum_inspect(f, x);
}
} // 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 <system_error>
#include <utility>
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp"
// Note: This API mostly wraps platform-specific functions that return ssize_t.
// We return ptrdiff_t instead, since only POSIX defines ssize_t and the two
// types are functionally equivalent.
namespace caf::net {
/// A unidirectional communication endpoint for inter-process communication.
struct CAF_NET_EXPORT pipe_socket : socket {
using super = socket;
using super::super;
};
/// Creates two connected sockets. The first socket is the read handle and the
/// second socket is the write handle.
/// @relates pipe_socket
expected<std::pair<pipe_socket, pipe_socket>> CAF_NET_EXPORT make_pipe();
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @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, 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, byte_span buf);
} // 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/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp"
#include <array>
#include <cstddef>
#include <cstdint>
#include <mutex>
namespace caf::net {
class pollset_updater : public socket_manager {
public:
// -- member types -----------------------------------------------------------
using super = socket_manager;
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
enum class code : uint8_t {
register_reading,
continue_reading,
register_writing,
continue_writing,
init_manager,
discard_manager,
shutdown_reading,
shutdown_writing,
run_action,
shutdown,
};
// -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer* parent);
~pollset_updater() override;
// -- properties -------------------------------------------------------------
/// Returns the managed socket.
pipe_socket handle() const noexcept {
return socket_cast<pipe_socket>(handle_);
}
// -- interface functions ----------------------------------------------------
error init(const settings& config) override;
read_result handle_read_event() override;
read_result handle_buffered_data() override;
read_result handle_continue_reading() override;
write_result handle_write_event() override;
write_result handle_continue_writing() override;
void handle_error(sec code) override;
private:
msg_buf buf_;
size_t buf_size_ = 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 <memory>
#include <new>
#include "caf/async/producer.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net {
/// Connects a socket manager to an asynchronous producer resource.
template <class Buffer>
class producer_adapter final : public ref_counted, public async::producer {
public:
using atomic_count = std::atomic<size_t>;
using buf_ptr = intrusive_ptr<Buffer>;
using value_type = typename Buffer::value_type;
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { //
adapter->on_cancel();
});
}
void on_consumer_demand(size_t) override {
mgr_->continue_reading();
}
void ref_producer() const noexcept override {
this->ref();
}
void deref_producer() const noexcept override {
this->deref();
}
/// Tries to open the resource for writing.
/// @returns a connected adapter that writes to the resource on success,
/// `nullptr` otherwise.
template <class Resource>
static intrusive_ptr<producer_adapter>
try_open(socket_manager* owner, Resource src) {
CAF_ASSERT(owner != nullptr);
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<producer_adapter>;
auto adapter = ptr_type{new producer_adapter(owner, buf), false};
buf->set_producer(adapter);
return adapter;
} else {
return nullptr;
}
}
/// Makes `item` available to the consumer.
/// @returns the remaining demand.
size_t push(const value_type& item) {
if (buf_) {
return buf_->push(item);
} else {
return 0;
}
}
/// Makes `items` available to the consumer.
/// @returns the remaining demand.
size_t push(span<const value_type> items) {
if (buf_) {
return buf_->push(items);
} else {
return 0;
}
}
void close() {
if (buf_) {
buf_->close();
buf_ = nullptr;
mgr_ = nullptr;
}
}
void abort(error reason) {
if (buf_) {
buf_->abort(std::move(reason));
buf_ = nullptr;
mgr_ = nullptr;
}
}
friend void intrusive_ptr_add_ref(const producer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const producer_adapter* ptr) noexcept {
ptr->deref();
}
private:
producer_adapter(socket_manager* owner, buf_ptr buf)
: mgr_(owner), buf_(std::move(buf)) {
// nop
}
void on_cancel() {
if (buf_) {
mgr_->mpx().shutdown_reading(mgr_);
buf_ = nullptr;
mgr_ = nullptr;
}
}
auto strong_this() {
return intrusive_ptr{this};
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_;
};
template <class T>
using producer_adapter_ptr = intrusive_ptr<producer_adapter<T>>;
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstdint>
namespace caf::net {
struct receive_policy {
uint32_t min_size;
uint32_t max_size;
/// @pre `min_size > 0`
/// @pre `min_size <= max_size`
static constexpr receive_policy between(uint32_t min_size,
uint32_t max_size) {
return {min_size, max_size};
}
/// @pre `size > 0`
static constexpr receive_policy exactly(uint32_t size) {
return {size, size};
}
static constexpr receive_policy up_to(uint32_t max_size) {
return {1, max_size};
}
static constexpr receive_policy stop() {
return {0, 0};
}
};
} // namespace caf::net
This diff is collapsed.
This diff is collapsed.
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstddef>
#include <limits>
#include <type_traits>
#include "caf/config.hpp"
namespace caf::net {
#ifdef CAF_WINDOWS
/// Platform-specific representation of a socket.
/// @relates socket
using socket_id = size_t;
/// Identifies the invalid socket.
constexpr socket_id invalid_socket_id = std::numeric_limits<socket_id>::max();
#else // CAF_WINDOWS
/// Platform-specific representation of a socket.
/// @relates socket
using socket_id = int;
/// Identifies the invalid socket.
constexpr socket_id invalid_socket_id = -1;
#endif // CAF_WINDOWS
/// Signed counterpart of `socket_id`.
/// @relates socket
using signed_socket_id = std::make_signed<socket_id>::type;
} // namespace caf::net
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
// This 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/web_socket/framing.hpp"
namespace caf::net {
template <class UpperLayer>
using web_socket_framing
[[deprecated("use caf::net::web_socket::framing instead")]]
= web_socket::framing<UpperLayer>;
} // namespace caf::net
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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