Unverified Commit 4aa07140 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1333

Use proper interfaces for caf-net
parents 65ecf353 787d4259
...@@ -45,6 +45,7 @@ cmake_dependent_option(CAF_ENABLE_OPENSSL_MODULE "Build OpenSSL module" ON ...@@ -45,6 +45,7 @@ cmake_dependent_option(CAF_ENABLE_OPENSSL_MODULE "Build OpenSSL module" ON
# -- CAF options with non-boolean values --------------------------------------- # -- CAF options with non-boolean values ---------------------------------------
set(CAF_LOG_LEVEL "QUIET" CACHE STRING "Set log verbosity of CAF components") set(CAF_LOG_LEVEL "QUIET" CACHE STRING "Set log verbosity of CAF components")
set(CAF_EXCLUDE_TESTS "" CACHE STRING "List of excluded test suites")
set(CAF_SANITIZERS "" CACHE STRING set(CAF_SANITIZERS "" CACHE STRING
"Comma separated sanitizers, e.g., 'address,undefined'") "Comma separated sanitizers, e.g., 'address,undefined'")
set(CAF_BUILD_INFO_FILE_PATH "" CACHE FILEPATH set(CAF_BUILD_INFO_FILE_PATH "" CACHE FILEPATH
...@@ -112,11 +113,13 @@ if(CAF_ENABLE_TESTING) ...@@ -112,11 +113,13 @@ if(CAF_ENABLE_TESTING)
enable_testing() enable_testing()
function(caf_add_test_suites target) function(caf_add_test_suites target)
foreach(suiteName ${ARGN}) foreach(suiteName ${ARGN})
string(REPLACE "." "/" suitePath ${suiteName}) if(NOT ${suiteName} IN_LIST CAF_EXCLUDE_TESTS)
target_sources(${target} PRIVATE string(REPLACE "." "/" suitePath ${suiteName})
"${CMAKE_CURRENT_SOURCE_DIR}/test/${suitePath}.cpp") target_sources(${target} PRIVATE
add_test(NAME ${suiteName} "${CMAKE_CURRENT_SOURCE_DIR}/test/${suitePath}.cpp")
COMMAND ${target} -r300 -n -v5 -s "^${suiteName}$") add_test(NAME ${suiteName}
COMMAND ${target} -r300 -n -v5 -s "^${suiteName}$")
endif()
endforeach() endforeach()
endfunction() endfunction()
endif() endif()
......
...@@ -126,8 +126,10 @@ config = [ ...@@ -126,8 +126,10 @@ config = [
['FreeBSD', [ ['FreeBSD', [
numCores: 4, numCores: 4,
builds: ['debug', 'release'], builds: ['debug', 'release'],
extraBuildFlags: [ extraDebugBuildFlags: [
'CAF_SANITIZERS:STRING=address', 'CAF_SANITIZERS:STRING=address',
// Deadlocks on FreeBSD in CRYPTO_THREAD_run_once under ASAN.
'CAF_EXCLUDE_TESTS:STRING=net.ssl.transport'
], ],
]], ]],
// Non-UNIX systems. // Non-UNIX systems.
......
...@@ -5,7 +5,9 @@ ...@@ -5,7 +5,9 @@
#include "caf/caf_main.hpp" #include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/accept.hpp" #include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp" #include "caf/scheduled_actor/flow.hpp"
......
...@@ -8,7 +8,9 @@ ...@@ -8,7 +8,9 @@
#include "caf/cow_tuple.hpp" #include "caf/cow_tuple.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/accept.hpp" #include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp" #include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
......
...@@ -270,7 +270,6 @@ caf_add_component( ...@@ -270,7 +270,6 @@ caf_add_component(
detail.serialized_size detail.serialized_size
detail.type_id_list_builder detail.type_id_list_builder
detail.unique_function detail.unique_function
detail.unordered_flat_map
dictionary dictionary
dsl dsl
dynamic_spawn dynamic_spawn
...@@ -360,6 +359,7 @@ caf_add_component( ...@@ -360,6 +359,7 @@ caf_add_component(
typed_response_promise typed_response_promise
typed_spawn typed_spawn
unit unit
unordered_flat_map
uri uri
uuid) uuid)
......
...@@ -20,6 +20,7 @@ enum class hex_format { ...@@ -20,6 +20,7 @@ enum class hex_format {
template <hex_format format = hex_format::uppercase, class Buf = std::string> template <hex_format format = hex_format::uppercase, class Buf = std::string>
void append_hex(Buf& result, const void* vptr, size_t n) { void append_hex(Buf& result, const void* vptr, size_t n) {
using value_type = typename Buf::value_type;
if (n == 0) if (n == 0)
return; return;
auto xs = reinterpret_cast<const uint8_t*>(vptr); auto xs = reinterpret_cast<const uint8_t*>(vptr);
...@@ -30,8 +31,8 @@ void append_hex(Buf& result, const void* vptr, size_t n) { ...@@ -30,8 +31,8 @@ void append_hex(Buf& result, const void* vptr, size_t n) {
tbl = "0123456789abcdef"; tbl = "0123456789abcdef";
for (size_t i = 0; i < n; ++i) { for (size_t i = 0; i < n; ++i) {
auto c = xs[i]; auto c = xs[i];
result.push_back(tbl[c >> 4]); result.push_back(static_cast<value_type>(tbl[c >> 4]));
result.push_back(tbl[c & 0x0F]); result.push_back(static_cast<value_type>(tbl[c & 0x0F]));
} }
} }
......
...@@ -11,7 +11,6 @@ ...@@ -11,7 +11,6 @@
#include "caf/cow_vector.hpp" #include "caf/cow_vector.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/flow/coordinated.hpp" #include "caf/flow/coordinated.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
...@@ -33,6 +32,7 @@ ...@@ -33,6 +32,7 @@
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/unordered_flat_map.hpp"
#include <cstddef> #include <cstddef>
#include <functional> #include <functional>
......
...@@ -112,6 +112,9 @@ public: ...@@ -112,6 +112,9 @@ public:
/// Returns an observer that ignores any of its inputs. /// Returns an observer that ignores any of its inputs.
static observer ignore(); static observer ignore();
/// Returns an observer that disposes its subscription immediately.
static observer cancel();
private: private:
intrusive_ptr<impl> pimpl_; intrusive_ptr<impl> pimpl_;
}; };
...@@ -170,6 +173,26 @@ private: ...@@ -170,6 +173,26 @@ private:
flow::subscription sub_; flow::subscription sub_;
}; };
template <class T>
class canceling_observer : public flow::observer_impl_base<T> {
public:
void on_next(const T&) override {
// nop
}
void on_error(const error&) override {
// nop
}
void on_complete() override {
// nop
}
void on_subscribe(flow::subscription sub) override {
sub.dispose();
}
};
template <class OnNextSignature> template <class OnNextSignature>
struct on_next_trait; struct on_next_trait;
...@@ -262,6 +285,11 @@ observer<T> observer<T>::ignore() { ...@@ -262,6 +285,11 @@ observer<T> observer<T>::ignore() {
return observer<T>{make_counted<detail::ignoring_observer<T>>()}; return observer<T>{make_counted<detail::ignoring_observer<T>>()};
} }
template <class T>
observer<T> observer<T>::cancel() {
return observer<T>{make_counted<detail::canceling_observer<T>>()};
}
/// Creates an observer from given callbacks. /// Creates an observer from given callbacks.
/// @param on_next Callback for handling incoming elements. /// @param on_next Callback for handling incoming elements.
/// @param on_error Callback for handling an error. /// @param on_error Callback for handling an error.
......
...@@ -40,7 +40,7 @@ public: ...@@ -40,7 +40,7 @@ public:
using input_ptr = std::unique_ptr<input_t>; using input_ptr = std::unique_ptr<input_t>;
using input_map = detail::unordered_flat_map<input_key, input_ptr>; using input_map = unordered_flat_map<input_key, input_ptr>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
......
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#include <cstdint> #include <cstdint>
#include <memory> #include <memory>
#include <string_view> #include <string_view>
#include <utility>
#include <variant> #include <variant>
#include <vector> #include <vector>
...@@ -48,6 +49,9 @@ template <class Iterator, class Sentinel = Iterator> struct parser_state; ...@@ -48,6 +49,9 @@ template <class Iterator, class Sentinel = Iterator> struct parser_state;
template <class, class, int> class actor_cast_access; template <class, class, int> class actor_cast_access;
template <class K, class V, class = std::allocator<std::pair<K, V>>>
class unordered_flat_map;
// -- variadic templates ------------------------------------------------------- // -- variadic templates -------------------------------------------------------
template <class...> class const_typed_message_view; template <class...> class const_typed_message_view;
......
...@@ -19,7 +19,6 @@ ...@@ -19,7 +19,6 @@
#include "caf/actor_traits.hpp" #include "caf/actor_traits.hpp"
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
...@@ -45,6 +44,7 @@ ...@@ -45,6 +44,7 @@
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/telemetry/timer.hpp" #include "caf/telemetry/timer.hpp"
#include "caf/unordered_flat_map.hpp"
namespace caf { namespace caf {
...@@ -546,7 +546,7 @@ protected: ...@@ -546,7 +546,7 @@ protected:
std::forward_list<pending_response> awaited_responses_; std::forward_list<pending_response> awaited_responses_;
/// Stores callbacks for multiplexed responses. /// Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_; unordered_flat_map<message_id, behavior> multiplexed_responses_;
/// Customization point for setting a default `message` callback. /// Customization point for setting a default `message` callback.
default_handler default_handler_; default_handler default_handler_;
......
...@@ -175,6 +175,10 @@ enum class sec : uint8_t { ...@@ -175,6 +175,10 @@ enum class sec : uint8_t {
disposed, disposed,
/// Failed to open a resource. /// Failed to open a resource.
cannot_open_resource, cannot_open_resource,
/// Received malformed data on a network socket.
protocol_error,
/// Encountered faulty logic in the program.
logic_error,
}; };
// --(rst-sec-end)-- // --(rst-sec-end)--
......
...@@ -10,13 +10,13 @@ ...@@ -10,13 +10,13 @@
#include "caf/detail/comparable.hpp" #include "caf/detail/comparable.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
namespace caf::detail { namespace caf {
/// A map abstraction with an unsorted `std::vector` providing `O(n)` lookup. /// A map abstraction with an unsorted `std::vector` providing `O(n)` lookup.
template <class Key, class T, template <class Key, class T, class Allocator>
class Allocator = std::allocator<std::pair<Key, T>>>
class unordered_flat_map { class unordered_flat_map {
public: public:
// -- member types ---------------------------------------------------------- // -- member types ----------------------------------------------------------
...@@ -209,7 +209,7 @@ public: ...@@ -209,7 +209,7 @@ public:
auto i = find(key); auto i = find(key);
if (i == end()) if (i == end())
CAF_RAISE_ERROR(std::out_of_range, CAF_RAISE_ERROR(std::out_of_range,
"caf::detail::unordered_flat_map::at out of range"); "caf::unordered_flat_map::at out of range");
return i->second; return i->second;
} }
...@@ -278,4 +278,4 @@ bool operator>=(const unordered_flat_map<K, T, A>& xs, ...@@ -278,4 +278,4 @@ bool operator>=(const unordered_flat_map<K, T, A>& xs,
return !(xs < ys); return !(xs < ys);
} }
} // namespace caf::detail } // namespace caf
...@@ -10,7 +10,6 @@ ...@@ -10,7 +10,6 @@
#include "caf/detail/comparable.hpp" #include "caf/detail/comparable.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/hash/fnv.hpp" #include "caf/hash/fnv.hpp"
#include "caf/inspector_access.hpp" #include "caf/inspector_access.hpp"
...@@ -18,6 +17,7 @@ ...@@ -18,6 +17,7 @@
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/ip_address.hpp" #include "caf/ip_address.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/unordered_flat_map.hpp"
namespace caf { namespace caf {
...@@ -59,7 +59,7 @@ public: ...@@ -59,7 +59,7 @@ public:
using path_list = std::vector<std::string_view>; using path_list = std::vector<std::string_view>;
/// Separates the query component into key-value pairs. /// Separates the query component into key-value pairs.
using query_map = detail::unordered_flat_map<std::string, std::string>; using query_map = unordered_flat_map<std::string, std::string>;
class CAF_CORE_EXPORT impl_type { class CAF_CORE_EXPORT impl_type {
public: public:
......
...@@ -2,9 +2,9 @@ ...@@ -2,9 +2,9 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE detail.unordered_flat_map #define CAF_SUITE unordered_flat_map
#include "caf/detail/unordered_flat_map.hpp" #include "caf/unordered_flat_map.hpp"
#include "core-test.hpp" #include "core-test.hpp"
...@@ -12,7 +12,7 @@ ...@@ -12,7 +12,7 @@
#include <utility> #include <utility>
#include <vector> #include <vector>
namespace caf::detail { namespace caf {
template <class T> template <class T>
bool operator==(const unordered_flat_map<int, T>& xs, bool operator==(const unordered_flat_map<int, T>& xs,
...@@ -26,14 +26,14 @@ bool operator==(const std::vector<std::pair<int, T>>& xs, ...@@ -26,14 +26,14 @@ bool operator==(const std::vector<std::pair<int, T>>& xs,
return ys == xs; return ys == xs;
} }
} // namespace caf::detail } // namespace caf
using std::make_pair; using std::make_pair;
using std::pair; using std::pair;
using std::string; using std::string;
using std::vector; using std::vector;
using caf::detail::unordered_flat_map; using caf::unordered_flat_map;
using namespace caf; using namespace caf;
......
...@@ -4,6 +4,8 @@ file(GLOB_RECURSE CAF_NET_HEADERS "caf/*.hpp") ...@@ -4,6 +4,8 @@ file(GLOB_RECURSE CAF_NET_HEADERS "caf/*.hpp")
# -- add targets --------------------------------------------------------------- # -- add targets ---------------------------------------------------------------
configure_file(test/pem.cpp.in test/pem.cpp @ONLY)
caf_add_component( caf_add_component(
net net
DEPENDENCIES DEPENDENCIES
...@@ -18,39 +20,65 @@ caf_add_component( ...@@ -18,39 +20,65 @@ caf_add_component(
net.http.method net.http.method
net.http.status net.http.status
net.operation net.operation
net.ssl.dtls
net.ssl.errc
net.ssl.format
net.ssl.tls
net.stream_transport_error net.stream_transport_error
net.web_socket.status net.web_socket.status
HEADERS HEADERS
${CAF_NET_HEADERS} ${CAF_NET_HEADERS}
SOURCES SOURCES
src/datagram_socket.cpp
src/detail/convert_ip_endpoint.cpp src/detail/convert_ip_endpoint.cpp
src/detail/rfc6455.cpp src/detail/rfc6455.cpp
src/host.cpp
src/ip.cpp
src/multiplexer.cpp
src/net/abstract_actor_shell.cpp src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp src/net/actor_shell.cpp
src/net/datagram_socket.cpp
src/net/generic_lower_layer.cpp
src/net/generic_upper_layer.cpp
src/net/http/header.cpp src/net/http/header.cpp
src/net/http/lower_layer.cpp
src/net/http/method.cpp src/net/http/method.cpp
src/net/http/server.cpp
src/net/http/status.cpp src/net/http/status.cpp
src/net/http/upper_layer.cpp
src/net/http/v1.cpp src/net/http/v1.cpp
src/net/ip.cpp
src/net/length_prefix_framing.cpp
src/net/message_oriented.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/multiplexer.cpp
src/net/network_socket.cpp
src/net/pipe_socket.cpp
src/net/pollset_updater.cpp
src/net/socket.cpp
src/net/socket_event_layer.cpp
src/net/socket_manager.cpp
src/net/ssl/connection.cpp
src/net/ssl/context.cpp
src/net/ssl/dtls.cpp
src/net/ssl/format.cpp
src/net/ssl/startup.cpp
src/net/ssl/tls.cpp
src/net/ssl/transport.cpp
src/net/stream_oriented.cpp
src/net/stream_socket.cpp
src/net/stream_transport.cpp
src/net/tcp_accept_socket.cpp
src/net/tcp_stream_socket.cpp
src/net/this_host.cpp
src/net/udp_datagram_socket.cpp
src/net/web_socket/client.cpp
src/net/web_socket/default_trait.cpp src/net/web_socket/default_trait.cpp
src/net/web_socket/frame.cpp src/net/web_socket/frame.cpp
src/net/web_socket/framing.cpp
src/net/web_socket/handshake.cpp src/net/web_socket/handshake.cpp
src/network_socket.cpp src/net/web_socket/lower_layer.cpp
src/pipe_socket.cpp src/net/web_socket/server.cpp
src/pollset_updater.cpp src/net/web_socket/upper_layer.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_SOURCES
${CMAKE_CURRENT_BINARY_DIR}/test/pem.cpp
test/net-test.cpp test/net-test.cpp
test/pem.cpp
TEST_SUITES TEST_SUITES
detail.convert_ip_endpoint detail.convert_ip_endpoint
detail.rfc6455 detail.rfc6455
...@@ -68,6 +96,7 @@ caf_add_component( ...@@ -68,6 +96,7 @@ caf_add_component(
net.producer_adapter net.producer_adapter
net.socket net.socket
net.socket_guard net.socket_guard
net.ssl.transport
net.stream_socket net.stream_socket
net.stream_transport net.stream_transport
net.tcp_socket net.tcp_socket
...@@ -76,21 +105,3 @@ caf_add_component( ...@@ -76,21 +105,3 @@ caf_add_component(
net.web_socket.client net.web_socket.client
net.web_socket.handshake net.web_socket.handshake
net.web_socket.server) 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()
...@@ -7,7 +7,6 @@ ...@@ -7,7 +7,6 @@
#include "caf/actor_traits.hpp" #include "caf/actor_traits.hpp"
#include "caf/callback.hpp" #include "caf/callback.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp" #include "caf/intrusive/drr_queue.hpp"
...@@ -18,6 +17,7 @@ ...@@ -18,6 +17,7 @@
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/policy/normal_messages.hpp" #include "caf/policy/normal_messages.hpp"
#include "caf/unordered_flat_map.hpp"
namespace caf::net { namespace caf::net {
...@@ -40,7 +40,9 @@ public: ...@@ -40,7 +40,9 @@ public:
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>; using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
using fallback_handler = unique_callback_ptr<result<message>(message&)>; using fallback_handler_sig = result<message>(abstract_actor_shell*, message&);
using fallback_handler = unique_callback_ptr<fallback_handler_sig>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -48,6 +50,10 @@ public: ...@@ -48,6 +50,10 @@ public:
~abstract_actor_shell() override; ~abstract_actor_shell() override;
// -- properties -------------------------------------------------------------
bool terminated() const noexcept;
// -- state modifiers -------------------------------------------------------- // -- state modifiers --------------------------------------------------------
/// Detaches the shell from its owner and closes the mailbox. /// Detaches the shell from its owner and closes the mailbox.
...@@ -56,7 +62,20 @@ public: ...@@ -56,7 +62,20 @@ public:
/// Overrides the default handler for unexpected messages. /// Overrides the default handler for unexpected messages.
template <class F> template <class F>
void set_fallback(F f) { void set_fallback(F f) {
fallback_ = make_type_erased_callback(std::move(f)); using msg_res_t = result<message>;
using self_ptr_t = abstract_actor_shell*;
if constexpr (std::is_invocable_r_v<msg_res_t, F, self_ptr_t, message&>) {
fallback_ = make_type_erased_callback(std::move(f));
} else {
static_assert(std::is_invocable_r_v<msg_res_t, F, message&>,
"illegal signature for the fallback handler: must be "
"'result<message>(message&)' or "
"'result<message>(abstract_actor_shell*, message&)'");
auto g = [f = std::move(f)](self_ptr_t, message& msg) mutable {
return f(msg);
};
fallback_ = make_type_erased_callback(std::move(g));
}
} }
// -- mailbox access --------------------------------------------------------- // -- mailbox access ---------------------------------------------------------
...@@ -114,7 +133,7 @@ protected: ...@@ -114,7 +133,7 @@ protected:
fallback_handler fallback_; fallback_handler fallback_;
// Stores callbacks for multiplexed responses. // Stores callbacks for multiplexed responses.
detail::unordered_flat_map<message_id, behavior> multiplexed_responses_; unordered_flat_map<message_id, behavior> multiplexed_responses_;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -4,68 +4,67 @@ ...@@ -4,68 +4,67 @@
#pragma once #pragma once
#include "caf/logger.hpp" #include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/settings.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 { namespace caf::net {
/// A connection_acceptor accepts connections from an accept socket and creates /// A connection_acceptor accepts connections from an accept socket and creates
/// socket managers to handle them via its factory. /// socket managers to handle them via its factory.
template <class Socket, class Factory> template <class Socket, class Factory>
class connection_acceptor { class connection_acceptor : public socket_event_layer {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using input_tag = tag::io_event_oriented;
using socket_type = Socket; using socket_type = Socket;
using factory_type = Factory; using factory_type = Factory;
using read_result = typename socket_manager::read_result; using read_result = typename socket_event_layer::read_result;
using write_result = typename socket_manager::write_result; using write_result = typename socket_event_layer::write_result;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> template <class... Ts>
explicit connection_acceptor(size_t limit, Ts&&... xs) explicit connection_acceptor(Socket fd, size_t limit, Ts&&... xs)
: factory_(std::forward<Ts>(xs)...), limit_(limit) { : fd_(fd), limit_(limit), factory_(std::forward<Ts>(xs)...) {
// nop // nop
} }
// -- interface functions ---------------------------------------------------- // -- factories --------------------------------------------------------------
template <class... Ts>
static std::unique_ptr<connection_acceptor>
make(Socket fd, size_t limit, Ts&&... xs) {
return std::make_unique<connection_acceptor>(fd, limit,
std::forward<Ts>(xs)...);
}
// -- implementation of socket_event_layer -----------------------------------
template <class LowerLayerPtr> error init(socket_manager* owner, const settings& cfg) override {
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
owner_ = owner; owner_ = owner;
cfg_ = cfg; cfg_ = cfg;
if (auto err = factory_.init(owner, cfg)) if (auto err = factory_.init(owner, cfg))
return err; return err;
down->register_reading(); owner->register_reading();
return none; return none;
} }
template <class LowerLayerPtr> read_result handle_read_event() override {
read_result handle_read_event(LowerLayerPtr down) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (auto x = accept(down->handle())) { if (auto x = accept(fd_)) {
socket_manager_ptr child = factory_.make(*x, owner_->mpx_ptr()); socket_manager_ptr child = factory_.make(owner_->mpx_ptr(), *x);
if (!child) { if (!child) {
CAF_LOG_ERROR("factory failed to create a new child"); CAF_LOG_ERROR("factory failed to create a new child");
down->abort_reason(sec::runtime_error); return read_result::abort;
return read_result::stop;
} }
if (auto err = child->init(cfg_)) { if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err); CAF_LOG_ERROR("failed to initialize new child:" << err);
down->abort_reason(std::move(err)); return read_result::abort;
return read_result::stop;
} }
if (limit_ == 0) { if (limit_ == 0) {
return read_result::again; return read_result::again;
...@@ -78,74 +77,36 @@ public: ...@@ -78,74 +77,36 @@ public:
} }
} }
template <class LowerLayerPtr> read_result handle_buffered_data() override {
static read_result handle_buffered_data(LowerLayerPtr) {
return read_result::again; return read_result::again;
} }
template <class LowerLayerPtr> read_result handle_continue_reading() override {
static read_result handle_continue_reading(LowerLayerPtr) {
return read_result::again; return read_result::again;
} }
template <class LowerLayerPtr> write_result handle_write_event() override {
write_result handle_write_event(LowerLayerPtr) {
CAF_LOG_ERROR("connection_acceptor received write event"); CAF_LOG_ERROR("connection_acceptor received write event");
return write_result::stop; return write_result::stop;
} }
template <class LowerLayerPtr> void abort(const error& reason) override {
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); CAF_LOG_ERROR("connection_acceptor aborts due to an error: " << reason);
factory_.abort(reason); factory_.abort(reason);
} }
private: private:
factory_type factory_; Socket fd_;
socket_manager* owner_;
size_t limit_; size_t limit_;
size_t accepted_ = 0; factory_type factory_;
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* owner_ = nullptr;
socket_manager_ptr make(Socket connected_socket, multiplexer* mpx) {
return f_(std::move(connected_socket), mpx);
}
void abort(const error&) { size_t accepted_ = 0;
// nop
}
private: settings cfg_;
FunctionObject f_;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -4,27 +4,21 @@ ...@@ -4,27 +4,21 @@
#pragma once #pragma once
#include <memory>
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/type_id.hpp" #include "caf/type_id.hpp"
#include <memory>
#include <vector>
namespace caf::net { namespace caf::net {
// -- templates ---------------------------------------------------------------- // -- templates ----------------------------------------------------------------
template <class UpperLayer>
class stream_transport; class stream_transport;
template <class Factory> template <class Factory>
class datagram_transport; 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> template <class... Sigs>
class typed_actor_shell; class typed_actor_shell;
...@@ -38,16 +32,18 @@ class actor_shell_ptr; ...@@ -38,16 +32,18 @@ class actor_shell_ptr;
class middleman; class middleman;
class multiplexer; class multiplexer;
class socket_manager; class socket_manager;
class this_host;
// -- structs ------------------------------------------------------------------ // -- structs ------------------------------------------------------------------
struct datagram_socket;
struct network_socket; struct network_socket;
struct pipe_socket; struct pipe_socket;
struct receive_policy;
struct socket; struct socket;
struct stream_socket; struct stream_socket;
struct tcp_accept_socket; struct tcp_accept_socket;
struct tcp_stream_socket; struct tcp_stream_socket;
struct datagram_socket;
struct udp_datagram_socket; struct udp_datagram_socket;
// -- smart pointer aliases ---------------------------------------------------- // -- smart pointer aliases ----------------------------------------------------
...@@ -56,4 +52,8 @@ using multiplexer_ptr = std::shared_ptr<multiplexer>; ...@@ -56,4 +52,8 @@ using multiplexer_ptr = std::shared_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>; using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>; using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
// -- miscellaneous aliases ----------------------------------------------------
using text_buffer = std::vector<char>;
} // namespace caf::net } // 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"
namespace caf::net {
/// Bundles protocol-independent, generic member functions for (almost all)
/// lower layers.
class CAF_NET_EXPORT generic_lower_layer {
public:
virtual ~generic_lower_layer();
/// Queries whether the output device can accept more data straight away.
[[nodiscard]] virtual bool can_send_more() const noexcept = 0;
/// Halt receiving of additional bytes or messages.
virtual void suspend_reading() = 0;
/// Queries whether the lower layer is currently configured to halt receiving
/// of additional bytes or messages.
[[nodiscard]] virtual bool stopped_reading() const noexcept = 0;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net {
/// Bundles protocol-independent, generic member functions for (almost all)
/// upper layers.
class CAF_NET_EXPORT generic_upper_layer {
public:
virtual ~generic_upper_layer();
/// Prepares any pending data for sending.
/// @returns `false` in case of an error to cause the lower layer to stop,
/// `true` otherwise.
[[nodiscard]] virtual bool prepare_send() = 0;
/// Queries whether all pending data has been sent. The lower calls this
/// function to decide whether it has to wait for write events on the socket.
[[nodiscard]] virtual bool done_sending() = 0;
/// Called by the lower layer after some event triggered re-registering the
/// socket manager for read operations after it has been stopped previously
/// by the read policy. May restart consumption of bytes or messages.
virtual void continue_reading() = 0;
/// Called by the lower layer for cleaning up any state in case of an error.
virtual void abort(const error& reason) = 0;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/net/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
...@@ -7,6 +7,6 @@ ...@@ -7,6 +7,6 @@
namespace caf::net::http { namespace caf::net::http {
/// Stores context information for a request. For HTTP/2, this is the stream ID. /// Stores context information for a request. For HTTP/2, this is the stream ID.
struct context {}; class context {};
} // namespace caf::net::http } // namespace caf::net::http
...@@ -4,13 +4,23 @@ ...@@ -4,13 +4,23 @@
#pragma once #pragma once
#include "caf/net/web_socket/framing.hpp" #include "caf/fwd.hpp"
namespace caf::net { #include <string_view>
template <class UpperLayer> namespace caf::net::http {
using web_socket_framing
[[deprecated("use caf::net::web_socket::framing instead")]]
= web_socket::framing<UpperLayer>;
} // namespace caf::net class context;
class header;
class lower_layer;
class server;
class upper_layer;
using header_fields_map
= unordered_flat_map<std::string_view, std::string_view>;
enum class method : uint8_t;
enum class status : uint16_t;
} // namespace caf::net::http
...@@ -4,7 +4,7 @@ ...@@ -4,7 +4,7 @@
#pragma once #pragma once
#include "caf/detail/unordered_flat_map.hpp" #include "caf/unordered_flat_map.hpp"
#include <string_view> #include <string_view>
...@@ -12,6 +12,6 @@ namespace caf::net::http { ...@@ -12,6 +12,6 @@ namespace caf::net::http {
/// Convenience type alias for a key-value map storing header fields. /// Convenience type alias for a key-value map storing header fields.
using header_fields_map using header_fields_map
= detail::unordered_flat_map<std::string_view, std::string_view>; = unordered_flat_map<std::string_view, std::string_view>;
} // namespace caf::net::http } // 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/net_export.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/http/fwd.hpp"
namespace caf::net::http {
/// Parses HTTP requests and passes them to the upper layer.
class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer();
/// Sends the next header to the client.
virtual bool
send_header(context ctx, status code, const header_fields_map& fields)
= 0;
/// Sends the payload after the header.
virtual bool send_payload(context, const_byte_span bytes) = 0;
/// Sends a chunk of data if the full payload is unknown when starting to
/// send.
virtual bool send_chunk(context, const_byte_span bytes) = 0;
/// Sends the last chunk, completing a chunked payload.
virtual bool send_end_of_chunks() = 0;
/// Terminates an HTTP context.
virtual void fin(context) = 0;
/// Convenience function for sending header and payload. Automatically sets
/// the header fields `Content-Type` and `Content-Length`.
bool send_response(context ctx, status codek, std::string_view content_type,
const_byte_span content);
};
} // namespace caf::net::http
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
#include "caf/default_enum_inspect.hpp" #include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/net/http/fwd.hpp"
#include <cstdint> #include <cstdint>
#include <string> #include <string>
......
This diff is collapsed.
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
#include "caf/default_enum_inspect.hpp" #include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/net/http/fwd.hpp"
#include "caf/string_view.hpp" #include "caf/string_view.hpp"
#include <cstdint> #include <cstdint>
......
// 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"
#include "caf/net/generic_upper_layer.hpp"
#include "caf/net/http/fwd.hpp"
namespace caf::net::http {
/// Operates on HTTP requests.
class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 0;
/// Consumes an HTTP message.
/// @param ctx Identifies this request. The response message must pass this
/// back to the lower layer. Allows clients to send multiple
/// requests in "parallel" (i.e., send multiple requests before
/// receiving the response on the first one).
/// @param hdr The header fields for the received message.
/// @param payload The payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an
/// error.
/// @note Discarded data is lost permanently.
virtual ptrdiff_t
consume(context ctx, const header& hdr, const_byte_span payload)
= 0;
};
} // 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/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
...@@ -7,14 +7,12 @@ ...@@ -7,14 +7,12 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp" #include "caf/async/spsc_buffer.hpp"
#include "caf/net/consumer_adapter.hpp" #include "caf/net/consumer_adapter.hpp"
#include "caf/net/message_oriented.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/producer_adapter.hpp" #include "caf/net/producer_adapter.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/settings.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> #include <utility>
...@@ -22,26 +20,39 @@ namespace caf::net { ...@@ -22,26 +20,39 @@ namespace caf::net {
/// Translates between a message-oriented transport and data flows. /// Translates between a message-oriented transport and data flows.
/// ///
/// The trait class converts between the native and the wire format: /// The `Trait` provides a customization point that converts between native and
/// wire format.
/// ///
/// ~~~ /// ~~~
/// struct my_trait { /// class my_trait {
/// bool convert(const T& value, byte_buffer& bytes); /// using input_type = ...;
/// bool convert(const_byte_span bytes, T& value); /// using output_type = ...;
/// bool convert(const_byte_span bytes, input_type& value);
/// bool convert(const output_type& value, byte_buffer& bytes);
/// }; /// };
/// ~~~ /// ~~~
template <class T, class Trait, class Tag = tag::message_oriented> template <class Trait>
class message_flow_bridge : public tag::no_auto_reading { class message_flow_bridge : public message_oriented::upper_layer {
public: public:
using input_tag = Tag; /// The input type for the application.
using input_type = typename Trait::input_type;
using buffer_type = async::spsc_buffer<T>; /// The output type for of application.
using output_type = typename Trait::output_type;
using consumer_resource_t = async::consumer_resource<T>; /// The resource type we pull from. We consume the output of the application.
using pull_resource_t = async::consumer_resource<output_type>;
using producer_resource_t = async::producer_resource<T>; /// The buffer type from the @ref pull_resource_t.
using pull_buffer_t = async::spsc_buffer<output_type>;
message_flow_bridge(consumer_resource_t in_res, producer_resource_t out_res, /// Type for the producer adapter. We produce the input of the application.
using push_resource_t = async::producer_resource<input_type>;
/// The buffer type from the @ref push_resource_t.
using push_buffer_t = async::spsc_buffer<input_type>;
message_flow_bridge(pull_resource_t in_res, push_resource_t out_res,
Trait trait) Trait trait)
: in_res_(std::move(in_res)), : in_res_(std::move(in_res)),
trait_(std::move(trait)), trait_(std::move(trait)),
...@@ -53,66 +64,41 @@ public: ...@@ -53,66 +64,41 @@ public:
// nop // nop
} }
template <class LowerLayerPtr> error init(net::socket_manager* mgr, message_oriented::lower_layer* down,
error init(net::socket_manager* mgr, LowerLayerPtr, const settings& cfg) { const settings&) {
mgr_ = mgr; down_ = down;
if constexpr (caf::detail::has_init_v<Trait>) {
if (auto err = init_res(trait_.init(cfg)))
return err;
}
if (in_res_) { if (in_res_) {
in_ = consumer_adapter<buffer_type>::try_open(mgr, in_res_); in_ = consumer_adapter<pull_buffer_t>::try_open(mgr, in_res_);
in_res_ = nullptr; in_res_ = nullptr;
} }
if (out_res_) { if (out_res_) {
out_ = producer_adapter<buffer_type>::try_open(mgr, out_res_); out_ = producer_adapter<push_buffer_t>::try_open(mgr, out_res_);
out_res_ = nullptr; out_res_ = nullptr;
} }
if (!in_ && !out_) if (!in_ && !out_)
return make_error(sec::cannot_open_resource, return make_error(sec::cannot_open_resource,
"flow bridge cannot run without at least one resource"); "a flow bridge needs at least one valid resource");
return none; return none;
} }
template <class LowerLayerPtr> bool write(const input_type& item) {
bool write(LowerLayerPtr down, const T& item) { down_->begin_message();
if constexpr (std::is_same_v<Tag, tag::message_oriented>) { auto& buf = down_->message_buffer();
down->begin_message(); return trait_.convert(item, buf) && down_->end_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 { struct write_helper {
using bridge_type = message_flow_bridge; message_flow_bridge* bridge;
bridge_type* bridge;
LowerLayerPtr down;
bool aborted = false; bool aborted = false;
size_t consumed = 0;
error err; error err;
write_helper(bridge_type* bridge, LowerLayerPtr down) write_helper(message_flow_bridge* bridge) : bridge(bridge) {
: bridge(bridge), down(down) {
// nop // nop
} }
void on_next(const T& item) { void on_next(const input_type& item) {
if (!bridge->write(down, item)) { if (!bridge->write(item))
aborted = true; aborted = true;
return;
}
} }
void on_complete() { void on_complete() {
...@@ -124,20 +110,18 @@ public: ...@@ -124,20 +110,18 @@ public:
} }
}; };
template <class LowerLayerPtr> bool prepare_send() override {
bool prepare_send(LowerLayerPtr down) { write_helper helper{this};
write_helper<LowerLayerPtr> helper{this, down}; while (down_->can_send_more() && in_) {
while (down->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper); auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) { if (!again) {
if (helper.err) { if (helper.err) {
down->send_close_message(helper.err); down_->send_close_message(helper.err);
} else { } else {
down->send_close_message(); down_->send_close_message();
} }
in_ = nullptr; in_ = nullptr;
} else if (helper.aborted) { } else if (helper.aborted) {
down->abort_reason(make_error(sec::conversion_failed));
in_->cancel(); in_->cancel();
in_ = nullptr; in_ = nullptr;
return false; return false;
...@@ -148,13 +132,11 @@ public: ...@@ -148,13 +132,11 @@ public:
return true; return true;
} }
template <class LowerLayerPtr> bool done_sending() override {
bool done_sending(LowerLayerPtr) {
return !in_ || !in_->has_data(); return !in_ || !in_->has_data();
} }
template <class LowerLayerPtr> void abort(const error& reason) override {
void abort(LowerLayerPtr, const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) { if (out_) {
if (reason == sec::socket_disconnected || reason == sec::disposed) if (reason == sec::socket_disconnected || reason == sec::disposed)
...@@ -169,89 +151,35 @@ public: ...@@ -169,89 +151,35 @@ public:
} }
} }
template <class U = Tag, class LowerLayerPtr> ptrdiff_t consume(byte_span buf) {
ptrdiff_t consume(LowerLayerPtr down, byte_span buf) { if (!out_)
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; return -1;
} input_type val;
T val; if (!trait_.convert(buf, val))
if (!trait_.convert(buf, val)) {
down->abort_reason(make_error(sec::conversion_failed));
return -1; return -1;
}
if (out_->push(std::move(val)) == 0) if (out_->push(std::move(val)) == 0)
down->suspend_reading(); down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size()); return static_cast<ptrdiff_t>(buf.size());
} }
private: private:
error init_res(error err) { /// Points to the next layer down the protocol stack.
return err; message_oriented::lower_layer* down_ = nullptr;
}
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. /// Incoming messages, serialized to the socket.
consumer_adapter_ptr<buffer_type> in_; consumer_adapter_ptr<pull_buffer_t> in_;
/// Outgoing messages, deserialized from the socket. /// Outgoing messages, deserialized from the socket.
producer_adapter_ptr<buffer_type> out_; producer_adapter_ptr<push_buffer_t> out_;
/// Converts between raw bytes and items. /// Converts between raw bytes and items.
Trait trait_; Trait trait_;
/// Discarded after initialization. /// Discarded after initialization.
consumer_resource_t in_res_; pull_resource_t in_res_;
/// Discarded after initialization. /// Discarded after initialization.
producer_resource_t out_res_; push_resource_t out_res_;
}; };
} // namespace caf::net } // 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/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net::message_oriented {
class lower_layer;
/// Consumes binary messages from the lower layer.
class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 0;
/// Consumes bytes from the lower layer.
/// @param payload Payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an
/// error.
/// @note Discarded data is lost permanently.
[[nodiscard]] virtual ptrdiff_t consume(byte_span payload) = 0;
};
/// Provides access to a resource that operates on the granularity of messages,
/// e.g., a UDP socket.
class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer();
/// Pulls messages from the transport until calling `suspend_reading`.
virtual void request_messages() = 0;
/// Prepares the layer for an outgoing message, e.g., by allocating an
/// output buffer as necessary.
virtual void begin_message() = 0;
/// Returns a reference to the buffer for assembling the current message.
/// Users may only call this function and write to the buffer between
/// calling `begin_message()` and `end_message()`.
/// @note The lower layers may pre-fill the buffer, e.g., to prefix custom
/// headers.
virtual byte_buffer& message_buffer() = 0;
/// Seals and prepares a message for transfer.
/// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code.
virtual bool end_message() = 0;
/// Inform the remote endpoint that no more messages will arrive.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message() = 0;
/// Inform the remote endpoint that no more messages will arrive because of
/// an error.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message(const error& reason) = 0;
};
} // namespace caf::net::message_oriented
// 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
...@@ -39,32 +39,6 @@ public: ...@@ -39,32 +39,6 @@ public:
~middleman() override; ~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 ---------------------------------------------------- // -- interface functions ----------------------------------------------------
void start() override; void start() override;
......
// 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
...@@ -10,12 +10,12 @@ ...@@ -10,12 +10,12 @@
#include "caf/action.hpp" #include "caf/action.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp" #include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp" #include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/unordered_flat_map.hpp"
extern "C" { extern "C" {
...@@ -37,7 +37,7 @@ public: ...@@ -37,7 +37,7 @@ public:
socket_manager_ptr mgr; socket_manager_ptr mgr;
}; };
using poll_update_map = detail::unordered_flat_map<socket, poll_update>; using poll_update_map = unordered_flat_map<socket, poll_update>;
using pollfd_list = std::vector<pollfd>; using pollfd_list = std::vector<pollfd>;
......
...@@ -4,7 +4,9 @@ ...@@ -4,7 +4,9 @@
#pragma once #pragma once
#include "caf/detail/net_export.hpp"
#include "caf/net/pipe_socket.hpp" #include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include <array> #include <array>
...@@ -14,7 +16,7 @@ ...@@ -14,7 +16,7 @@
namespace caf::net { namespace caf::net {
class pollset_updater : public socket_manager { class CAF_NET_EXPORT pollset_updater : public socket_event_layer {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -37,20 +39,15 @@ public: ...@@ -37,20 +39,15 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer* parent); explicit pollset_updater(pipe_socket fd);
~pollset_updater() override; // -- factories --------------------------------------------------------------
// -- properties ------------------------------------------------------------- static std::unique_ptr<pollset_updater> make(pipe_socket fd);
/// Returns the managed socket. // -- implementation of socket_event_layer -----------------------------------
pipe_socket handle() const noexcept {
return socket_cast<pipe_socket>(handle_);
}
// -- interface functions ---------------------------------------------------- error init(socket_manager* owner, const settings& cfg) override;
error init(const settings& config) override;
read_result handle_read_event() override; read_result handle_read_event() override;
...@@ -60,11 +57,11 @@ public: ...@@ -60,11 +57,11 @@ public:
write_result handle_write_event() override; write_result handle_write_event() override;
write_result handle_continue_writing() override; void abort(const error& reason) override;
void handle_error(sec code) override;
private: private:
pipe_socket fd_;
multiplexer* mpx_ = nullptr;
msg_buf buf_; msg_buf buf_;
size_t buf_size_ = 0; size_t buf_size_ = 0;
}; };
......
...@@ -8,8 +8,15 @@ ...@@ -8,8 +8,15 @@
namespace caf::net { namespace caf::net {
/// Configures how many bytes a @ref stream_transport receives before calling
/// `consume` on its upper layer.
struct receive_policy { struct receive_policy {
/// Configures how many bytes @ref stream_transport must read before it may
/// call `consume` on its upper layer.
uint32_t min_size; uint32_t min_size;
/// Configures how many bytes @ref stream_transport may read at most before
/// it calls `consume` on its upper layer.
uint32_t max_size; uint32_t max_size;
/// @pre `min_size > 0` /// @pre `min_size > 0`
...@@ -24,6 +31,7 @@ struct receive_policy { ...@@ -24,6 +31,7 @@ struct receive_policy {
return {size, size}; return {size, size};
} }
/// @pre `max_size >= 1`
static constexpr receive_policy up_to(uint32_t max_size) { static constexpr receive_policy up_to(uint32_t max_size) {
return {1, max_size}; return {1, max_size};
} }
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// The lowest layer in a protocol stack. Called by a @ref socket_manager
/// directly.
class CAF_NET_EXPORT socket_event_layer {
public:
virtual ~socket_event_layer();
/// Encodes how to proceed after a read operation.
enum class read_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to write to the socket instead of reading
/// from the socket.
want_write,
/// Indicates that the manager no longer reads from the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Encodes how to proceed after a write operation.
enum class write_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to read from the socket instead of
/// writing to the socket.
want_read,
/// Indicates that the manager no longer writes to the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Initializes the layer.
virtual error init(socket_manager* owner, const settings& cfg) = 0;
/// Handles a read event on the managed socket.
virtual read_result handle_read_event() = 0;
/// Handles internally buffered data.
virtual read_result handle_buffered_data() = 0;
/// Handles a request to continue reading on the socket.
virtual read_result handle_continue_reading() = 0;
/// Handles a write event on the managed socket.
virtual write_result handle_write_event() = 0;
/// Called after returning `handover` from a read or write handler.
virtual bool do_handover(std::unique_ptr<socket_event_layer>& next);
/// Called on hard errors on the managed socket.
virtual void abort(const error& reason) = 0;
};
} // namespace caf::net
...@@ -12,15 +12,15 @@ namespace caf::net { ...@@ -12,15 +12,15 @@ namespace caf::net {
template <class Socket> template <class Socket>
class socket_guard { class socket_guard {
public: public:
socket_guard() noexcept : sock_(invalid_socket_id) { socket_guard() noexcept : fd_(invalid_socket_id) {
// nop // nop
} }
explicit socket_guard(Socket sock) noexcept : sock_(sock) { explicit socket_guard(Socket fd) noexcept : fd_(fd) {
// nop // nop
} }
socket_guard(socket_guard&& other) noexcept : sock_(other.release()) { socket_guard(socket_guard&& other) noexcept : fd_(other.release()) {
// nop // nop
} }
...@@ -34,28 +34,32 @@ public: ...@@ -34,28 +34,32 @@ public:
socket_guard& operator=(const socket_guard&) = delete; socket_guard& operator=(const socket_guard&) = delete;
~socket_guard() { ~socket_guard() {
if (sock_.id != invalid_socket_id) if (fd_.id != invalid_socket_id)
close(sock_); close(fd_);
} }
void reset(Socket x) noexcept { void reset(Socket x) noexcept {
if (sock_.id != invalid_socket_id) if (fd_.id != invalid_socket_id)
close(sock_); close(fd_);
sock_ = x; fd_ = x;
} }
Socket release() noexcept { Socket release() noexcept {
auto sock = sock_; auto sock = fd_;
sock_.id = invalid_socket_id; fd_.id = invalid_socket_id;
return sock; return sock;
} }
Socket get() noexcept {
return fd_;
}
Socket socket() const noexcept { Socket socket() const noexcept {
return sock_; return fd_;
} }
private: private:
Socket sock_; Socket fd_;
}; };
template <class Socket> template <class Socket>
......
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/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/ssl/errc.hpp"
#include "caf/net/stream_socket.hpp"
#include <cstddef>
namespace caf::net::ssl {
/// SSL state for a single connections.
class CAF_NET_EXPORT connection {
public:
// -- member types -----------------------------------------------------------
/// The opaque implementation type.
struct impl;
// -- constructors, destructors, and assignment operators --------------------
connection() = delete;
connection(connection&& other);
connection& operator=(connection&& other);
~connection();
// -- factories --------------------------------------------------------------
expected<connection> make(stream_socket fd);
// -- native handles ---------------------------------------------------------
/// Reinterprets `native_handle` as the native implementation type and takes
/// ownership of the handle.
static connection from_native(void* native_handle);
/// Retrieves the native handle from the connection.
void* native_handle() const noexcept;
// -- error handling ---------------------------------------------------------
/// Returns the error code for a preceding call to `connect`, `accept`,
/// `read`, `write` or `close.
/// @param ret The (negative) result from the preceding call.
errc last_error(ptrdiff_t ret) const;
/// Returns a human-readable representation of the error for a preceding call
/// to `connect`, `accept`, `read`, `write` or `close.
/// @param ret The (negative) result from the preceding call.
std::string last_error_string(ptrdiff_t ret) const;
// -- connecting and teardown ------------------------------------------------
/// Performs the client-side TLS/SSL handshake after connection to the server.
[[nodiscard]] ptrdiff_t connect();
/// Performs the server-side TLS/SSL handshake after accepting a connection
/// from a client.
[[nodiscard]] ptrdiff_t accept();
/// Gracefully closes the SSL connection.
ptrdiff_t close();
// -- reading and writing ----------------------------------------------------
/// Tries to fill @p buf with data from the managed socket.
[[nodiscard]] ptrdiff_t read(byte_span buf);
/// Tries to write bytes from @p buf to the managed socket.
[[nodiscard]] ptrdiff_t write(const_byte_span buf);
// -- properties -------------------------------------------------------------
/// Returns the number of bytes that are currently buffered outside of the
/// managed socket.
size_t buffered() const noexcept;
/// Returns the file descriptor for this connection.
stream_socket fd() const noexcept;
private:
constexpr explicit connection(impl* ptr) : pimpl_(ptr) {
// nop
}
impl* pimpl_;
};
} // namespace caf::net::ssl
// 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/ssl/dtls.hpp"
#include "caf/net/ssl/format.hpp"
#include "caf/net/ssl/fwd.hpp"
#include "caf/net/ssl/tls.hpp"
#include "caf/net/stream_socket.hpp"
namespace caf::net::ssl {
/// SSL state, shared by multiple connections.
class CAF_NET_EXPORT context {
public:
// -- member types -----------------------------------------------------------
/// The opaque implementation type.
struct impl;
// -- constructors, destructors, and assignment operators --------------------
context() = delete;
context(context&& other);
context& operator=(context&& other);
~context();
// -- factories --------------------------------------------------------------
static expected<context> make(tls min_version, tls max_version = tls::any);
static expected<context> make_server(tls min_version,
tls max_version = tls::any);
static expected<context> make_client(tls min_version,
tls max_version = tls::any);
static expected<context> make(dtls min_version, dtls max_version = dtls::any);
static expected<context> make_server(dtls min_version,
dtls max_version = dtls::any);
static expected<context> make_client(dtls min_version,
dtls max_version = dtls::any);
// -- native handles ---------------------------------------------------------
/// Reinterprets `native_handle` as the native implementation type and takes
/// ownership of the handle.
static context from_native(void* native_handle);
/// Retrieves the native handle from the context.
void* native_handle() const noexcept;
// -- error handling ---------------------------------------------------------
/// Retrieves a human-readable error description for a preceding call to
/// another member functions and removes that error from the error queue. Call
/// repeatedly until @ref has_last_error returns `false` to retrieve all
/// errors from the queue.
static std::string last_error_string();
/// Queries whether the error stack has at least one entry.
static bool has_last_error() noexcept;
// -- connections ------------------------------------------------------------
expected<connection> new_connection(stream_socket fd);
// -- certificates and keys --------------------------------------------------
/// Configure the context to use the default locations for loading CA
/// certificates.
/// @returns `true` on success, `false` otherwise and `last_error` can be used
/// to retrieve a human-readable error representation.
[[nodiscard]] bool set_default_verify_paths();
/// Configures the context to load CA certificate from a directory.
/// @param path Null-terminated string with a path to a directory. Files in
/// the directory must use the CA subject name hash value as file
/// name with a suffix to disambiguate multiple certificates,
/// e.g., `9d66eef0.0` and `9d66eef0.1`.
/// @returns `true` on success, `false` otherwise and `last_error` can be used
/// to retrieve a human-readable error representation.
[[nodiscard]] bool load_verify_dir(const char* path);
/// Loads a CA certificate file.
/// @param path Null-terminated string with a path to a single PEM file.
/// @returns `true` on success, `false` otherwise and `last_error` can be used
/// to retrieve a human-readable error representation.
[[nodiscard]] bool load_verify_file(const char* path);
/// Loads the first certificate found in given file.
/// @param path Null-terminated string with a path to a single PEM file.
[[nodiscard]] bool use_certificate_from_file(const char* path,
format file_format);
/// Loads the first private key found in given file.
[[nodiscard]] bool use_private_key_from_file(const char* path,
format file_format);
private:
constexpr explicit context(impl* ptr) : pimpl_(ptr) {
// nop
}
impl* pimpl_;
};
} // namespace caf::net::ssl
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/is_error_code_enum.hpp"
namespace caf::net::ssl {
/// Configures the allowed DTLS versions on a @ref context.
enum class dtls {
any,
v1_0,
v1_2,
};
/// @relates dtls
CAF_NET_EXPORT std::string to_string(dtls);
/// @relates dtls
CAF_NET_EXPORT bool from_string(std::string_view, dtls&);
/// @relates dtls
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<dtls>, dtls&);
/// @relates dtls
template <class Inspector>
bool inspect(Inspector& f, dtls& x) {
return default_enum_inspect(f, x);
}
/// @relates dtls
CAF_NET_EXPORT int native(dtls);
} // namespace caf::net::ssl
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/is_error_code_enum.hpp"
namespace caf::net::ssl {
/// SSL error code for I/O operations on a @ref connection.
enum class errc : uint8_t {
/// Not-an-error.
none = 0,
/// SSL has closed the connection. The underlying transport may remain open.
closed,
/// Temporary error. SSL failed to write to a socket because it needs to read
/// first.
want_read,
/// Temporary error. SSL failed to read from a socket because it needs to
/// write first.
want_write,
/// Temporary error. The SSL client handshake did not complete yet.
want_connect,
/// Temporary error. The SSL server handshake did not complete yet.
want_accept,
/// Temporary error. An application callback has asked to be called again.
want_x509_lookup,
/// Temporary error. An asynchronous is still processing data and the user
/// must call the preceding function again from the same thread.
want_async,
/// The pool for starting asynchronous jobs is exhausted.
want_async_job,
/// Temporary error. An application callback has asked to be called again.
want_client_hello,
/// The operating system reported a non-recoverable, fatal I/O error. Users
/// may consult OS-specific means to retrieve the underlying error, e.g.,
/// `errno` on UNIX or `WSAGetLastError` on Windows.
syscall_failed,
/// SSL encountered a fatal error, usually a protocol violation.
fatal,
/// An unexpected error occurred with no further explanation available.
unspecified,
};
/// @relates errc
CAF_NET_EXPORT std::string to_string(errc);
/// @relates errc
CAF_NET_EXPORT bool from_string(std::string_view, errc&);
/// @relates errc
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<errc>, errc&);
/// @relates errc
template <class Inspector>
bool inspect(Inspector& f, errc& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::net::ssl
CAF_ERROR_CODE_ENUM(caf::net::ssl::errc)
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include <string>
namespace caf::net::ssl {
/// Format of keys and certificates.
enum class format {
/// Privacy Enhanced Mail format.
pem,
/// Binary ASN1 format.
asn1,
};
/// @relates format
CAF_NET_EXPORT std::string to_string(format);
/// @relates format
CAF_NET_EXPORT bool from_string(std::string_view, format&);
/// @relates format
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<format>, format&);
/// @relates format
template <class Inspector>
bool inspect(Inspector& f, format& x) {
return default_enum_inspect(f, x);
}
/// @relates dtls
CAF_NET_EXPORT int native(format);
} // namespace caf::net::ssl
// 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::ssl {
class connection;
class context;
class transport;
enum class dtls;
enum class errc : uint8_t;
enum class format;
enum class tls;
} // namespace caf::net::ssl
...@@ -2,22 +2,18 @@ ...@@ -2,22 +2,18 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once #include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include <cstddef> namespace caf::net::ssl {
#include <cstdint>
// -- hard-coded default values for various CAF options ------------------------ /// Initializes the SSL layer. Depending on the version, this may be mandatory
/// to call before accessing any SSL functions (OpenSSL prior to version 1.1) or
/// it may have no effect (newer versions of OpenSSL).
CAF_NET_EXPORT void startup();
namespace caf::defaults::middleman { /// Cleans up any state for the SSL layer. Like @ref startup, this step is
/// mandatory for some versions of the linked SSL library.
CAF_NET_EXPORT void cleanup();
/// Maximum number of cached buffers for sending payloads. } // namespace caf::net::ssl
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.
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/is_error_code_enum.hpp"
namespace caf::net::ssl {
/// Configures the allowed DTLS versions on a @ref context.
enum class tls {
any,
v1_0,
v1_1,
v1_2,
v1_3,
};
/// @relates tls
CAF_NET_EXPORT std::string to_string(tls);
/// @relates tls
CAF_NET_EXPORT bool from_string(std::string_view, tls&);
/// @relates tls
CAF_NET_EXPORT bool from_integer(std::underlying_type_t<tls>, tls&);
/// @relates tls
template <class Inspector>
bool inspect(Inspector& f, tls& x) {
return default_enum_inspect(f, x);
}
/// @relates tls
CAF_NET_EXPORT int native(tls);
/// @relates tls
CAF_NET_EXPORT bool has(tls, tls, tls);
} // namespace caf::net::ssl
// 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/ssl/connection.hpp"
#include "caf/net/stream_transport.hpp"
namespace caf::net::ssl {
/// Implements a stream_transport that manages a stream socket with encrypted
/// communication over OpenSSL.
class CAF_NET_EXPORT transport : public stream_transport {
public:
// -- member types -----------------------------------------------------------
using super = stream_transport;
using worker_ptr = std::unique_ptr<socket_event_layer>;
class policy_impl : public stream_transport::policy {
public:
explicit policy_impl(connection conn);
ptrdiff_t read(stream_socket, byte_span) override;
ptrdiff_t write(stream_socket, const_byte_span) override;
stream_transport_error last_error(stream_socket, ptrdiff_t) override;
ptrdiff_t connect(stream_socket) override;
ptrdiff_t accept(stream_socket) override;
size_t buffered() const noexcept override;
connection conn;
};
// -- factories --------------------------------------------------------------
/// Creates a new instance of the SSL transport for a socket that has already
/// performed the SSL handshake.
/// @param conn The connection object for managing @p fd.
/// @param up The layer operating on top of this transport.
static std::unique_ptr<transport> make(connection conn, upper_layer_ptr up);
/// Returns a worker that performs the OpenSSL server handshake on the socket.
/// On success, the worker performs a handover to an `openssl_transport` that
/// runs `up`.
/// @param conn The connection object for managing @p fd.
/// @param up The layer operating on top of this transport.
static worker_ptr make_server(connection conn, upper_layer_ptr up);
/// Returns a worker that performs the OpenSSL client handshake on the socket.
/// On success, the worker performs a handover to an `openssl_transport` that
/// runs `up`.
/// @param conn The connection object for managing @p fd.
/// @param up The layer operating on top of this transport.
static worker_ptr make_client(connection conn, upper_layer_ptr up);
private:
// -- constructors, destructors, and assignment operators --------------------
transport(stream_socket fd, connection conn, upper_layer_ptr up);
policy_impl policy_impl_;
};
} // namespace caf::net::ssl
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/generic_lower_layer.hpp"
#include "caf/net/generic_upper_layer.hpp"
namespace caf::net::stream_oriented {
class lower_layer;
/// The upper layer requests bytes from the lower layer and consumes raw chunks
/// of data.
class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
/// Initializes the upper layer.
/// @param owner A pointer to the socket manager that owns the entire
/// protocol stack. Remains valid for the lifetime of the upper
/// layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error
init(socket_manager* owner, lower_layer* down, const settings& config)
= 0;
/// Consumes bytes from the lower layer.
/// @param buffer Available bytes to read.
/// @param delta Bytes that arrived since last calling this function.
/// @returns The number of consumed bytes. May be zero if waiting for more
/// input or negative to signal an error.
[[nodiscard]] virtual ptrdiff_t consume(byte_span buffer, byte_span delta)
= 0;
};
/// Provides access to a resource that operates on a byte stream, e.g., a TCP
/// socket.
class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public:
virtual ~lower_layer();
/// Configures threshold for the next receive operations. Policies remain
/// active until calling this function again.
/// @warning Calling this function in `consume` invalidates both byte spans.
virtual void configure_read(receive_policy policy) = 0;
/// Prepares the layer for outgoing traffic, e.g., by allocating an output
/// buffer as necessary.
virtual void begin_output() = 0;
/// Returns a reference to the output buffer. Users may only call this
/// function and write to the buffer between calling `begin_output()` and
/// `end_output()`.
virtual byte_buffer& output_buffer() = 0;
/// Prepares written data for transfer, e.g., by flushing buffers or
/// registering sockets for write events.
virtual bool end_output() = 0;
};
} // namespace caf::net::stream_oriented
// 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"
#include "caf/net/receive_policy.hpp"
namespace caf::net {
/// Wraps a pointer to a stream-oriented layer with a pointer to its lower
/// layer. Both pointers are then used to implement the interface required for a
/// stream-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class stream_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_output() {
lptr_->begin_output(llptr_);
}
byte_buffer& output_buffer() {
return lptr_->output_buffer(llptr_);
}
void end_output() {
lptr_->end_output(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
void configure_read(receive_policy policy) {
lptr_->configure_read(llptr_, policy);
}
bool stopped() const noexcept {
return lptr_->stopped(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
stream_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
stream_oriented_layer_ptr(const stream_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_stream_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
return stream_oriented_layer_ptr<Layer, LowerLayerPtr>{this_layer, down};
}
} // namespace caf::net
This diff is collapsed.
...@@ -9,9 +9,11 @@ ...@@ -9,9 +9,11 @@
namespace caf::net { namespace caf::net {
struct CAF_NET_EXPORT this_host { /// Groups functions for managing the host system.
class CAF_NET_EXPORT this_host {
public:
/// Initializes the network subsystem. /// Initializes the network subsystem.
static error startup(); static void startup();
/// Release any resources of the network subsystem. /// Release any resources of the network subsystem.
static void cleanup(); static void cleanup();
......
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
namespace caf::detail { namespace caf::detail {
template <template <class> class Transport, class Trait> template <class Transport, class Trait>
class ws_acceptor_factory { class ws_acceptor_factory {
public: public:
using connector_pointer = net::web_socket::flow_connector_ptr<Trait>; using connector_pointer = net::web_socket::flow_connector_ptr<Trait>;
...@@ -32,10 +32,11 @@ public: ...@@ -32,10 +32,11 @@ public:
} }
template <class Socket> template <class Socket>
net::socket_manager_ptr make(Socket fd, net::multiplexer* mpx) { net::socket_manager_ptr make(net::multiplexer* mpx, Socket fd) {
using app_t = net::web_socket::flow_bridge<Trait>; auto app = net::web_socket::flow_bridge<Trait>::make(connector_);
using stack_t = Transport<net::web_socket::server<app_t>>; auto ws = net::web_socket::server::make(std::move(app));
return net::make_socket_manager<stack_t>(fd, mpx, connector_); auto transport = Transport::make(fd, std::move(ws));
return net::socket_manager::make(mpx, fd, std::move(transport));
} }
void abort(const error&) { void abort(const error&) {
...@@ -80,8 +81,8 @@ auto make_accept_event_resources() { ...@@ -80,8 +81,8 @@ auto make_accept_event_resources() {
/// @param on_request Function object for accepting incoming requests. /// @param on_request Function object for accepting incoming requests.
/// @param limit The maximum amount of connections before closing @p fd. Passing /// @param limit The maximum amount of connections before closing @p fd. Passing
/// 0 means "no limit". /// 0 means "no limit".
template <template <class> class Transport = stream_transport, class Socket, template <class Transport = stream_transport, class Socket, class... Ts,
class... Ts, class OnRequest> class OnRequest>
void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out, void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
OnRequest on_request, size_t limit = 0) { OnRequest on_request, size_t limit = 0) {
using trait_t = default_trait; using trait_t = default_trait;
...@@ -95,8 +96,8 @@ void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out, ...@@ -95,8 +96,8 @@ void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
auto& mpx = sys.network_manager().mpx(); auto& mpx = sys.network_manager().mpx();
auto conn = std::make_shared<conn_t>(std::move(on_request), std::move(buf)); auto conn = std::make_shared<conn_t>(std::move(on_request), std::move(buf));
auto factory = factory_t{std::move(conn)}; auto factory = factory_t{std::move(conn)};
auto ptr = make_socket_manager<impl_t>(std::move(fd), &mpx, limit, auto impl = impl_t::make(fd, limit, std::move(factory));
std::move(factory)); auto ptr = socket_manager::make(&mpx, fd, std::move(impl));
mpx.init(ptr); mpx.init(ptr);
} }
// TODO: accept() should return a disposable to stop the WebSocket server. // TODO: accept() should return a disposable to stop the WebSocket server.
......
...@@ -13,12 +13,11 @@ ...@@ -13,12 +13,11 @@
#include "caf/net/http/v1.hpp" #include "caf/net/http/v1.hpp"
#include "caf/net/message_flow_bridge.hpp" #include "caf/net/message_flow_bridge.hpp"
#include "caf/net/receive_policy.hpp" #include "caf/net/receive_policy.hpp"
#include "caf/net/stream_oriented.hpp"
#include "caf/net/web_socket/framing.hpp" #include "caf/net/web_socket/framing.hpp"
#include "caf/net/web_socket/handshake.hpp" #include "caf/net/web_socket/handshake.hpp"
#include "caf/pec.hpp" #include "caf/pec.hpp"
#include "caf/settings.hpp" #include "caf/settings.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
#include <algorithm> #include <algorithm>
...@@ -28,150 +27,71 @@ namespace caf::net::web_socket { ...@@ -28,150 +27,71 @@ namespace caf::net::web_socket {
/// 6455. Initially, the layer performs the WebSocket handshake. Once completed, /// 6455. Initially, the layer performs the WebSocket handshake. Once completed,
/// this layer decodes RFC 6455 frames and forwards binary and text messages to /// this layer decodes RFC 6455 frames and forwards binary and text messages to
/// `UpperLayer`. /// `UpperLayer`.
template <class UpperLayer> class client : public stream_oriented::upper_layer {
class client {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using input_tag = tag::stream_oriented; using handshake_ptr = std::unique_ptr<handshake>;
using output_tag = tag::mixed_message_oriented; using upper_layer_ptr = std::unique_ptr<web_socket::upper_layer>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> client(handshake_ptr hs, upper_layer_ptr up);
explicit client(handshake hs, Ts&&... xs)
: upper_layer_(std::forward<Ts>(xs)...) {
handshake_.reset(new handshake(std::move(hs)));
}
// -- initialization --------------------------------------------------------- // -- factories --------------------------------------------------------------
template <class LowerLayerPtr> static std::unique_ptr<client> make(handshake_ptr hs, upper_layer_ptr up);
error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
CAF_ASSERT(handshake_ != nullptr);
owner_ = owner;
if (!handshake_->has_mandatory_fields())
return make_error(sec::runtime_error,
"handshake data lacks mandatory fields");
if (!handshake_->has_valid_key())
handshake_->randomize_key();
cfg_ = cfg;
down->begin_output();
handshake_->write_http_1_request(down->output_buffer());
down->end_output();
down->configure_read(receive_policy::up_to(handshake::max_http_size));
return none;
}
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
auto& upper_layer() noexcept { web_socket::upper_layer& upper_layer() noexcept {
return upper_layer_.upper_layer(); return framing_.upper_layer();
} }
const auto& upper_layer() const noexcept { const web_socket::upper_layer& upper_layer() const noexcept {
return upper_layer_.upper_layer(); return framing_.upper_layer();
} }
// -- role: upper layer ------------------------------------------------------ stream_oriented::lower_layer& lower_layer() noexcept {
return framing_.lower_layer();
template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) {
return handshake_complete() ? upper_layer_.prepare_send(down) : true;
} }
template <class LowerLayerPtr> const stream_oriented::lower_layer& lower_layer() const noexcept {
bool done_sending(LowerLayerPtr down) { return framing_.lower_layer();
return handshake_complete() ? upper_layer_.done_sending(down) : true;
} }
template <class LowerLayerPtr> bool handshake_completed() const noexcept {
void abort(LowerLayerPtr down, const error& reason) { return hs_ == nullptr;
if (handshake_complete())
upper_layer_.abort(down, reason);
} }
template <class LowerLayerPtr> // -- implementation of stream_oriented::upper_layer -------------------------
void continue_reading(LowerLayerPtr down) {
if (handshake_complete())
upper_layer_.continue_reading(down);
}
template <class LowerLayerPtr> error init(socket_manager* owner, stream_oriented::lower_layer* down,
ptrdiff_t consume(LowerLayerPtr down, byte_span input, byte_span delta) { const settings& config) override;
CAF_LOG_TRACE(CAF_ARG2("socket", down->handle().id)
<< CAF_ARG2("bytes", input.size()));
// Short circuit to the framing layer after the handshake completed.
if (handshake_complete())
return upper_layer_.consume(down, input, delta);
// Check whether received a HTTP header or else wait for more data or abort
// when exceeding the maximum size.
auto [hdr, remainder] = http::v1::split_header(input);
if (hdr.empty()) {
if (input.size() >= handshake::max_http_size) {
CAF_LOG_ERROR("server response exceeded maximum header size");
auto err = make_error(pec::too_many_characters,
"exceeded maximum header size");
down->abort_reason(std::move(err));
return -1;
} else {
return 0;
}
} else if (!handle_header(down, hdr)) {
return -1;
} else if (remainder.empty()) {
CAF_ASSERT(hdr.size() == input.size());
return hdr.size();
} else {
CAF_LOG_DEBUG(CAF_ARG2("socket", down->handle().id)
<< CAF_ARG2("remainder", remainder.size()));
if (auto sub_result = upper_layer_.consume(down, remainder, remainder);
sub_result >= 0) {
return hdr.size() + sub_result;
} else {
return sub_result;
}
}
}
bool handshake_complete() const noexcept { void abort(const error& reason) override;
return handshake_ == nullptr;
} ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void continue_reading() override;
bool prepare_send() override;
bool done_sending() override;
private: private:
// -- HTTP response processing ----------------------------------------------- // -- HTTP response processing -----------------------------------------------
template <class LowerLayerPtr> bool handle_header(std::string_view http);
bool handle_header(LowerLayerPtr down, std::string_view http) {
CAF_ASSERT(handshake_ != nullptr);
auto http_ok = handshake_->is_valid_http_1_response(http);
handshake_.reset();
if (http_ok) {
if (auto err = upper_layer_.init(owner_, down, cfg_)) {
CAF_LOG_DEBUG("failed to initialize WebSocket framing layer");
down->abort_reason(std::move(err));
return false;
} else {
CAF_LOG_DEBUG("completed WebSocket handshake");
return true;
}
} else {
CAF_LOG_DEBUG("received invalid WebSocket handshake");
down->abort_reason(make_error(
sec::runtime_error,
"received invalid WebSocket handshake response from server"));
return false;
}
}
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Stores the WebSocket handshake data until the handshake completed. /// Stores the WebSocket handshake data until the handshake completed.
std::unique_ptr<handshake> handshake_; handshake_ptr hs_;
/// Stores the upper layer. /// Stores the upper layer.
framing<UpperLayer> upper_layer_; framing framing_;
/// Stores a pointer to the owning manager for the delayed initialization. /// Stores a pointer to the owning manager for the delayed initialization.
socket_manager* owner_ = nullptr; socket_manager* owner_ = nullptr;
...@@ -179,17 +99,4 @@ private: ...@@ -179,17 +99,4 @@ private:
settings cfg_; settings cfg_;
}; };
template <template <class> class Transport = stream_transport, class Socket,
class T, class Trait>
void run_client(multiplexer& mpx, Socket fd, handshake hs,
async::consumer_resource<T> in, async::producer_resource<T> out,
Trait trait) {
using app_t = message_flow_bridge<T, Trait, tag::mixed_message_oriented>;
using stack_t = Transport<client<app_t>>;
auto mgr = make_socket_manager<stack_t>(fd, &mpx, std::move(hs),
std::move(in), std::move(out),
std::move(trait));
mpx.init(mgr);
}
} // namespace caf::net::web_socket } // namespace caf::net::web_socket
This diff is collapsed.
This diff is collapsed.
...@@ -18,6 +18,12 @@ class flow_bridge; ...@@ -18,6 +18,12 @@ class flow_bridge;
template <class Trait, class... Ts> template <class Trait, class... Ts>
class request; class request;
class CAF_CORE_EXPORT frame; class frame;
class framing;
class lower_layer;
class server;
class upper_layer;
enum class status : uint16_t;
} // namespace caf::net::web_socket } // namespace caf::net::web_socket
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
namespace caf::tag {
/// Tells message-oriented transports to *not* implicitly register an
/// application for reading as part of the initialization when used a base type.
struct no_auto_reading {};
} // namespace caf::tag
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