Commit 1245db60 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/caf-net'

parents c3582f7b f9d5c312
set(CAF_ENABLE_ACTOR_PROFILER ON CACHE BOOL "")
set(CAF_ENABLE_EXAMPLES ON CACHE BOOL "")
set(CAF_ENABLE_ROBOT_TESTS ON CACHE BOOL "")
set(CAF_ENABLE_RUNTIME_CHECKS ON CACHE BOOL "")
set(CAF_LOG_LEVEL TRACE CACHE STRING "")
set(CAF_SANITIZERS "address,undefined" CACHE STRING "")
if(NOT MSVC AND NOT CMAKE_SYSTEM MATCHES BSD)
set(CMAKE_BUILD_TYPE Debug CACHE STRING "")
endif()
......@@ -10,4 +10,11 @@ RUN dnf update -y \
libubsan \
make \
openssl-devel \
python3 \
python3-pip \
&& dnf clean all
RUN python3 -m pip install \
robotframework \
robotframework-requests \
robotframework-websocketclient
......@@ -4,6 +4,7 @@ WorkingDir="$PWD"
UsageString="
$0 build CMAKE_INIT_FILE SOURCE_DIR BUILD_DIR
OR $0 test BUILD_DIR
OR $0 test BUILD_DIR EXCLUDES
OR $0 assert WHAT
"
......@@ -43,12 +44,21 @@ elif [ $# = 2 ]; then
if [ "$1" = 'test' ] && [ -d "$2" ]; then
Mode='test'
BuildDir=`makeAbsolute $2`
Excludes=""
elif [ "$1" = 'assert' ]; then
Mode='assert'
What="$2"
else
usage
fi
elif [ $# = 3 ]; then
if [ "$1" = 'test' ] && [ -d "$2" ]; then
Mode='test'
BuildDir=`makeAbsolute $2`
Excludes="$3"
else
usage
fi
else
usage
fi
......@@ -80,9 +90,11 @@ runBuild() {
}
runTest() {
cd "$BuildDir"
$CTestCommand --output-on-failure
cd "$WorkingDir"
if [ -z "$Excludes" ]; then
$CTestCommand --test-dir "$BuildDir" --output-on-failure
else
$CTestCommand --test-dir "$BuildDir" --output-on-failure -E "$Excludes"
fi
}
runLeakSanitizerCheck() {
......
......@@ -8,6 +8,11 @@ debug_build_template: &DEBUG_BUILD_TEMPLATE
build_script: .ci/run.sh build .ci/debug-flags.cmake . build
test_script: .ci/run.sh test build
asan_build_template: &ASAN_BUILD_TEMPLATE
build_script: .ci/run.sh build .ci/asan-flags.cmake . build
# see GH issue #1396
test_script: .ci/run.sh test build openssl
# CentOS 7 EOL: June 2024
centos7_task:
container:
......@@ -38,6 +43,14 @@ freebsd12_task:
prepare_script: .ci/freebsd-12/prepare.sh
<< : *DEBUG_BUILD_TEMPLATE
# Fedora 36 EOL: May 2023
sanitizers_task:
container:
dockerfile: .ci/fedora-36/Dockerfile
<< : *RESOURCES_TEMPLATE
prepare_script: .ci/run.sh assert UBSanitizer
<< : *ASAN_BUILD_TEMPLATE
# Windows Server 2019 EOL: January 2024
windows_task:
timeout_in: 120m
......
......@@ -10,5 +10,6 @@ manual/examples
manual/html/*
manual/libcaf_core
manual/libcaf_io
manual/libcaf_net
manual/libcaf_openssl
release.txt
......@@ -27,6 +27,13 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The class `expected` now implements the monadic member functions from C++23
`std::expected` as well as `value_or`.
### Changed
- After collecting experience and feedback on the new HTTP and WebSocket APIs
introduced with 0.19.0-rc.1, we decided to completely overhaul the
user-facing, high-level APIs. Please consult the manual for the new DSL to
start servers.
### Fixed
- When exporting metrics to Prometheus, CAF now normalizes label names to meet
......@@ -39,13 +46,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The `fan_out_request` request now properly deals with actor handles that
respond with `void` (#1369).
- Fix subscription and event handling in flow buffer operator.
- Fix undefined behavior in getter functions of the flow `mcast` operator.
- Add checks to avoid potential UB when using `prefix_and_tail` or other
operators that use the `ucast` operator internally.
- The `mcast` and `ucast` operators now stop calling `on_next` immediately when
disposed.
- Actors no longer terminate despite having open streams (#1377).
<<<<<<< HEAD
- Actors reading from external sources such as SPSC buffers via a local flow
could end up in a long-running read loop. To avoid potentially starving other
actors or activities, scheduled actors now limit the amount of actions that
......@@ -58,6 +61,7 @@ is based on [Keep a Changelog](https://keepachangelog.com).
now defaults to `INADDR6_ANY` (but allowing IPv4 clients) with `INADDR_ANY` as
fallback in case opening the socket in IPv6 mode failed.
- Add missing includes that prevented CAF from compiling on GCC 13.
- Fix AddressSanitizer and LeakSanitizer findings in some flow operators.
### Deprecated
......
......@@ -26,6 +26,7 @@ option(CAF_ENABLE_CPACK "Enable packaging via CPack" OFF)
option(CAF_ENABLE_CURL_EXAMPLES "Build examples with libcurl" OFF)
option(CAF_ENABLE_PROTOBUF_EXAMPLES "Build examples with Google Protobuf" OFF)
option(CAF_ENABLE_QT6_EXAMPLES "Build examples with the Qt6 framework" OFF)
option(CAF_ENABLE_ROBOT_TESTS "Add the Robot tests to CTest " OFF)
option(CAF_ENABLE_RUNTIME_CHECKS "Build CAF with extra runtime assertions" OFF)
# -- CAF options that are on by default ----------------------------------------
......@@ -401,6 +402,12 @@ if(CAF_ENABLE_TOOLS)
add_subdirectory(tools)
endif()
# -- optionally add the Robot tests to CTest -----------------------------------
if(CAF_ENABLE_TESTING AND CAF_ENABLE_ROBOT_TESTS)
add_subdirectory(robot)
endif()
# -- add top-level compiler and linker flags that propagate to clients ---------
# Disable warnings regarding C++ classes at ABI boundaries on MSVC.
......
......@@ -68,24 +68,6 @@ if(TARGET CAF::io)
add_io_example(remoting remote_spawn)
add_io_example(remoting distributed_calculator)
# basic I/O with brokers
add_io_example(broker simple_broker)
add_io_example(broker simple_http_broker)
if(CAF_ENABLE_PROTOBUF_EXAMPLES)
find_package(Protobuf REQUIRED)
if(NOT PROTOBUF_PROTOC_EXECUTABLE)
message(FATAL_ERROR "CMake was unable to set PROTOBUF_PROTOC_EXECUTABLE")
endif()
protobuf_generate_cpp(ProtoSources ProtoHeaders "${CMAKE_CURRENT_SOURCE_DIR}/remoting/pingpong.proto")
include_directories(${PROTOBUF_INCLUDE_DIR})
include_directories(${CMAKE_CURRENT_BINARY_DIR})
add_executable(protobuf_broker broker/protobuf_broker.cpp ${ProtoSources})
target_link_libraries(protobuf_broker
PRIVATE ${PROTOBUF_LIBRARIES} CAF::internal CAF::io)
add_dependencies(protobuf_broker all_examples)
endif()
if(CAF_ENABLE_CURL_EXAMPLES)
find_package(CURL REQUIRED)
add_executable(curl_fuse curl/curl_fuse.cpp)
......@@ -107,7 +89,7 @@ else()
endfunction()
endif()
add_net_example(http secure-time-server)
add_net_example(http rest)
add_net_example(http time-server)
add_net_example(length_prefix_framing chat-client)
......@@ -116,7 +98,6 @@ add_net_example(length_prefix_framing chat-server)
add_net_example(web_socket echo)
add_net_example(web_socket hello-client)
add_net_example(web_socket quote-server)
add_net_example(web_socket secure-echo)
add_net_example(web_socket stock-ticker)
if(TARGET CAF::net AND CAF_ENABLE_QT6_EXAMPLES)
......
#include <iostream>
#include <limits>
#include <memory>
#include <string>
#include <vector>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <arpa/inet.h>
#endif
CAF_PUSH_WARNINGS
#include "pingpong.pb.h"
CAF_POP_WARNINGS
CAF_BEGIN_TYPE_ID_BLOCK(protobuf_example, first_custom_type_id)
CAF_ADD_ATOM(protobuf_example, kickoff_atom)
CAF_END_TYPE_ID_BLOCK(protobuf_example)
namespace {
using namespace caf;
using namespace caf::io;
// utility function to print an exit message with custom name
void print_on_exit(scheduled_actor* self, const std::string& name) {
self->attach_functor([=](const error& reason) {
aout(self) << name << " exited: " << to_string(reason) << std::endl;
});
}
struct ping_state {
size_t count = 0;
};
behavior ping(stateful_actor<ping_state>* self, size_t num_pings) {
print_on_exit(self, "ping");
return {
[=](kickoff_atom, const actor& pong) {
self->send(pong, ping_atom_v, 1);
self->become([=](pong_atom, int value) -> result<ping_atom, int> {
if (++(self->state.count) >= num_pings)
self->quit();
return {ping_atom_v, value + 1};
});
},
};
}
behavior pong(event_based_actor* self) {
print_on_exit(self, "pong");
return {
[=](ping_atom, int value) { return make_message(pong_atom_v, value); },
};
}
void protobuf_io(broker* self, connection_handle hdl, const actor& buddy) {
print_on_exit(self, "protobuf_io");
aout(self) << "protobuf broker started" << std::endl;
self->monitor(buddy);
self->set_down_handler([=](const down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << std::endl;
self->quit(dm.reason);
}
});
auto write = [=](const org::caf::PingOrPong& p) {
std::string buf = p.SerializeAsString();
auto s = htonl(static_cast<uint32_t>(buf.size()));
self->write(hdl, sizeof(uint32_t), &s);
self->write(hdl, buf.size(), buf.data());
self->flush(hdl);
};
auto default_callbacks = message_handler{
[=](const connection_closed_msg&) {
aout(self) << "connection closed" << std::endl;
self->send_exit(buddy, exit_reason::remote_link_unreachable);
self->quit(exit_reason::remote_link_unreachable);
},
[=](ping_atom, int i) {
aout(self) << "'ping' " << i << std::endl;
org::caf::PingOrPong p;
p.mutable_ping()->set_id(i);
write(p);
},
[=](pong_atom, int i) {
aout(self) << "'pong' " << i << std::endl;
org::caf::PingOrPong p;
p.mutable_pong()->set_id(i);
write(p);
},
};
auto await_protobuf_data = message_handler {
[=](const new_data_msg& msg) {
org::caf::PingOrPong p;
p.ParseFromArray(msg.buf.data(), static_cast<int>(msg.buf.size()));
if (p.has_ping()) {
self->send(buddy, ping_atom_v, p.ping().id());
}
else if (p.has_pong()) {
self->send(buddy, pong_atom_v, p.pong().id());
}
else {
self->quit(exit_reason::user_shutdown);
std::cerr << "neither Ping nor Pong!" << std::endl;
}
// receive next length prefix
self->configure_read(hdl, receive_policy::exactly(sizeof(uint32_t)));
self->unbecome();
},
}.or_else(default_callbacks);
auto await_length_prefix = message_handler {
[=](const new_data_msg& msg) {
uint32_t num_bytes;
memcpy(&num_bytes, msg.buf.data(), sizeof(uint32_t));
num_bytes = htonl(num_bytes);
if (num_bytes > (1024 * 1024)) {
aout(self) << "someone is trying something nasty" << std::endl;
self->quit(exit_reason::user_shutdown);
return;
}
// receive protobuf data
auto nb = static_cast<size_t>(num_bytes);
self->configure_read(hdl, receive_policy::exactly(nb));
self->become(keep_behavior, await_protobuf_data);
},
}.or_else(default_callbacks);
// initial setup
self->configure_read(hdl, receive_policy::exactly(sizeof(uint32_t)));
self->become(await_length_prefix);
}
behavior server(broker* self, const actor& buddy) {
print_on_exit(self, "server");
aout(self) << "server is running" << std::endl;
return {
[=](const new_connection_msg& msg) {
aout(self) << "server accepted new connection" << std::endl;
auto io_actor = self->fork(protobuf_io, msg.handle, buddy);
// only accept 1 connection in our example
self->quit();
},
};
}
class config : public actor_system_config {
public:
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port")
.add(host, "host,H", "set host (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode");
}
};
void run_server(actor_system& system, const config& cfg) {
std::cout << "run in server mode" << std::endl;
auto pong_actor = system.spawn(pong);
auto server_actor = system.middleman().spawn_server(server, cfg.port,
pong_actor);
if (!server_actor)
std::cerr << "unable to spawn server: " << to_string(server_actor.error())
<< std::endl;
}
void run_client(actor_system& system, const config& cfg) {
std::cout << "run in client mode" << std::endl;
auto ping_actor = system.spawn(ping, 20u);
auto io_actor = system.middleman().spawn_client(protobuf_io, cfg.host,
cfg.port, ping_actor);
if (!io_actor) {
std::cout << "cannot connect to " << cfg.host << " at port " << cfg.port
<< ": " << to_string(io_actor.error()) << std::endl;
return;
}
send_as(*io_actor, ping_actor, kickoff_atom_v, *io_actor);
}
void caf_main(actor_system& system, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client;
f(system, cfg);
}
} // namespace
CAF_MAIN(id_block::protobuf_example, io::middleman)
/******************************************************************************\
* This example program showcases how to manually manage socket IO using *
* a broker. Server and client exchange integers in a 'ping-pong protocol'. *
* *
* Minimal setup: *
* - simple_broker -s 4242 *
* - simple_broker -c localhost 4242 *
\******************************************************************************/
#include "caf/config.hpp"
#ifdef CAF_WINDOWS
# define _WIN32_WINNT 0x0600
# include <winsock2.h>
#else
# include <arpa/inet.h> // htonl
#endif
#include <cassert>
#include <cstdint>
#include <iostream>
#include <limits>
#include <memory>
#include <string>
#include <vector>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using std::cerr;
using std::cout;
using std::endl;
using namespace caf;
using namespace caf::io;
namespace {
// --(rst-attach-begin)--
// Utility function to print an exit message with custom name.
void print_on_exit(const actor& hdl, const std::string& name) {
hdl->attach_functor([=](const error& reason) {
cout << name << " exited: " << to_string(reason) << endl;
});
}
// --(rst-attach-end)--
enum class op : uint8_t {
ping,
pong,
};
behavior ping(event_based_actor* self, size_t num_pings) {
auto count = std::make_shared<size_t>(0);
return {
[=](ok_atom, const actor& pong) {
self->send(pong, ping_atom_v, int32_t{1});
self->become([=](pong_atom, int32_t value) -> result<ping_atom, int32_t> {
if (++*count >= num_pings)
self->quit();
return {ping_atom_v, value + 1};
});
},
};
}
behavior pong() {
return {
[](ping_atom, int32_t value) -> result<pong_atom, int32_t> {
return {pong_atom_v, value};
},
};
}
// Utility function for sending an integer type.
template <class T>
void write_int(broker* self, connection_handle hdl, T value) {
using unsigned_type = typename std::make_unsigned<T>::type;
if constexpr (sizeof(T) > 1) {
auto cpy = static_cast<T>(htonl(static_cast<unsigned_type>(value)));
self->write(hdl, sizeof(T), &cpy);
} else {
self->write(hdl, sizeof(T), &value);
}
}
// Utility function for reading an integer from incoming data.
template <class T>
void read_int(const void* data, T& storage) {
using unsigned_type = typename std::make_unsigned<T>::type;
memcpy(&storage, data, sizeof(T));
if constexpr (sizeof(T) > 1)
storage = static_cast<T>(ntohl(static_cast<unsigned_type>(storage)));
}
// Implementation of our broker.
behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
// We assume io_fsm manages a broker with exactly one connection,
// i.e., the connection ponted to by `hdl`.
assert(self->num_connections() == 1);
// Monitor buddy to quit broker if buddy is done.
self->monitor(buddy);
self->set_down_handler([=](down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl;
// Quit for same reason.
self->quit(dm.reason);
}
});
// Setup: we are exchanging only messages consisting of an operation
// (as uint8_t) and an integer value (int32_t).
self->configure_read(hdl, receive_policy::exactly(sizeof(uint8_t)
+ sizeof(int32_t)));
// Our message handlers.
return {
[=](const connection_closed_msg& msg) {
// Brokers can multiplex any number of connections, however
// this example assumes io_fsm to manage a broker with
// exactly one connection.
if (msg.handle == hdl) {
aout(self) << "connection closed" << endl;
// force buddy to quit
self->send_exit(buddy, exit_reason::remote_link_unreachable);
self->quit(exit_reason::remote_link_unreachable);
}
},
[=](ping_atom, int32_t i) {
aout(self) << "send {ping, " << i << "}" << endl;
write_int(self, hdl, static_cast<uint8_t>(op::ping));
write_int(self, hdl, i);
self->flush(hdl);
},
[=](pong_atom, int32_t i) {
aout(self) << "send {pong, " << i << "}" << endl;
write_int(self, hdl, static_cast<uint8_t>(op::pong));
write_int(self, hdl, i);
self->flush(hdl);
},
[=](const new_data_msg& msg) {
// Keeps track of our position in the buffer.
auto rd_pos = msg.buf.data();
// Read the operation value as uint8_t from the buffer.
auto op_val = uint8_t{0};
read_int(rd_pos, op_val);
++rd_pos;
// Read the integer value from the buffer.
auto ival = int32_t{0};
read_int(rd_pos, ival);
// Show some output.
// Send composed message to our buddy.
switch (static_cast<op>(op_val)) {
case op::ping:
aout(self) << "received {ping, " << ival << "}" << endl;
self->send(buddy, ping_atom_v, ival);
break;
case op::pong:
aout(self) << "received {pong, " << ival << "}" << endl;
self->send(buddy, pong_atom_v, ival);
break;
default:
aout(self) << "invalid value for op_val, stop" << endl;
self->quit(sec::invalid_argument);
}
},
};
}
behavior server(broker* self, const actor& buddy) {
aout(self) << "server is running" << endl;
return {
[=](const new_connection_msg& msg) {
aout(self) << "server accepted new connection" << endl;
// By forking into a new broker, we are no longer
// responsible for the connection.
auto impl = self->fork(broker_impl, msg.handle, buddy);
print_on_exit(impl, "broker_impl");
aout(self) << "quit server (only accept 1 connection)" << endl;
self->quit();
},
};
}
class config : public actor_system_config {
public:
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port")
.add(host, "host,H", "set host (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode");
}
};
void run_server(actor_system& system, const config& cfg) {
cout << "run in server mode" << endl;
auto pong_actor = system.spawn(pong);
auto server_actor = system.middleman().spawn_server(server, cfg.port,
pong_actor);
if (!server_actor) {
std::cerr << "failed to spawn server: " << to_string(server_actor.error())
<< endl;
return;
}
print_on_exit(*server_actor, "server");
print_on_exit(pong_actor, "pong");
}
void run_client(actor_system& system, const config& cfg) {
auto ping_actor = system.spawn(ping, size_t{20});
auto io_actor = system.middleman().spawn_client(broker_impl, cfg.host,
cfg.port, ping_actor);
if (!io_actor) {
std::cerr << "failed to spawn client: " << to_string(io_actor.error())
<< endl;
return;
}
print_on_exit(ping_actor, "ping actor");
print_on_exit(*io_actor, "broker");
send_as(*io_actor, ping_actor, ok_atom_v, *io_actor);
}
void caf_main(actor_system& system, const config& cfg) {
auto f = cfg.server_mode ? run_server : run_client;
f(system, cfg);
}
} // namespace
CAF_MAIN(io::middleman)
#include <chrono>
#include <iostream>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using std::cerr;
using std::cout;
using std::endl;
using namespace caf;
using namespace caf::io;
namespace {
constexpr const char http_ok[] = R"__(HTTP/1.1 200 OK
Content-Type: text/plain
Connection: keep-alive
Transfer-Encoding: chunked
d
Hi there! :)
0
)__";
template <size_t Size>
constexpr size_t cstr_size(const char (&)[Size]) {
return Size;
}
behavior connection_worker(broker* self, connection_handle hdl) {
self->configure_read(hdl, receive_policy::at_most(1024));
return {
[=](const new_data_msg& msg) {
self->write(msg.handle, cstr_size(http_ok), http_ok);
self->quit();
},
[=](const connection_closed_msg&) { self->quit(); },
};
}
behavior server(broker* self) {
auto counter = std::make_shared<int>(0);
self->set_down_handler([=](down_msg&) { ++*counter; });
self->delayed_send(self, std::chrono::seconds(1), tick_atom_v);
return {
[=](const new_connection_msg& ncm) {
auto worker = self->fork(connection_worker, ncm.handle);
self->monitor(worker);
self->link_to(worker);
},
[=](tick_atom) {
aout(self) << "Finished " << *counter << " requests per second." << endl;
*counter = 0;
self->delayed_send(self, std::chrono::seconds(1), tick_atom_v);
},
};
}
class config : public actor_system_config {
public:
uint16_t port = 0;
config() {
opt_group{custom_options_, "global"}.add(port, "port,p", "set port");
}
};
void caf_main(actor_system& system, const config& cfg) {
auto server_actor = system.middleman().spawn_server(server, cfg.port);
if (!server_actor) {
cerr << "*** cannot spawn server: " << to_string(server_actor.error())
<< endl;
return;
}
cout << "*** listening on port " << cfg.port << endl;
cout << "*** to quit the program, simply press <enter>" << endl;
// wait for any input
std::string dummy;
std::getline(std::cin, dummy);
// kill server
anon_send_exit(*server_actor, exit_reason::user_shutdown);
}
} // namespace
CAF_MAIN(io::middleman)
// Non-interactive example to illustrate how to connect flows over an
// asynchronous SPSC (Single Producer Single Consumer) buffer.
// asynchronous SPSC (Single Producer Single Consumer) buffer manually. Usually,
// CAF generates the SPSC buffers implicitly.
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
......
// This example an HTTP server that implements a REST API by forwarding requests
// to an actor. The actor in this example is a simple key-value store. The actor
// is not aware of HTTP and the HTTP server is sending regular request messages
// to actor.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/http/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <algorithm>
#include <cctype>
#include <iostream>
#include <string>
#include <utility>
namespace http = caf::net::http;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 8080;
static constexpr size_t default_max_connections = 128;
// -- configuration ------------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<size_t>("max-connections,m", "limit for concurrent clients");
opt_group{custom_options_, "tls"} //
.add<std::string>("key-file,k", "path to the private key file")
.add<std::string>("cert-file,c", "path to the certificate file");
}
};
// -- our key-value store actor ------------------------------------------------
struct kvs_actor_state {
caf::behavior make_behavior() {
using caf::result;
return {
[this](caf::get_atom, const std::string& key) -> result<std::string> {
if (auto i = data.find(key); i != data.end())
return i->second;
else
return make_error(caf::sec::no_such_key, key + " not found");
},
[this](caf::put_atom, const std::string& key, std::string& value) {
data.insert_or_assign(key, std::move(value));
},
[this](caf::delete_atom, const std::string& key) { data.erase(key); },
};
}
std::map<std::string, std::string> data;
};
using kvs_actor_impl = caf::stateful_actor<kvs_actor_state>;
// -- utility functions --------------------------------------------------------
bool is_ascii(caf::span<const std::byte> buffer) {
auto pred = [](auto x) { return isascii(static_cast<unsigned char>(x)); };
return std::all_of(buffer.begin(), buffer.end(), pred);
}
std::string to_ascii(caf::span<const std::byte> buffer) {
return std::string{reinterpret_cast<const char*>(buffer.data()),
buffer.size()};
}
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
using namespace std::literals;
namespace ssl = caf::net::ssl;
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
auto pem = ssl::format::pem;
auto key_file = caf::get_as<std::string>(cfg, "tls.key-file");
auto cert_file = caf::get_as<std::string>(cfg, "tls.cert-file");
auto max_connections = caf::get_or(cfg, "max-connections",
default_max_connections);
if (!key_file != !cert_file) {
std::cerr << "*** inconsistent TLS config: declare neither file or both\n";
return EXIT_FAILURE;
}
// Spin up our key-value store actor.
auto kvs = sys.spawn<kvs_actor_impl>();
// Open up a TCP port for incoming connections and start the server.
auto server
= http::with(sys)
// Optionally enable TLS.
.context(ssl::context::enable(key_file && cert_file)
.and_then(ssl::emplace_server(ssl::tls::v1_2))
.and_then(ssl::use_private_key_file(key_file, pem))
.and_then(ssl::use_certificate_file(cert_file, pem)))
// Bind to the user-defined port.
.accept(port)
// Limit how many clients may be connected at any given time.
.max_connections(max_connections)
// Stop the server if our key-value store actor terminates.
.monitor(kvs)
// Forward incoming requests to the kvs actor.
.route("/api/<arg>", http::method::get,
[kvs](http::responder& res, std::string key) {
auto* self = res.self();
auto prom = std::move(res).to_promise();
self->request(kvs, 2s, caf::get_atom_v, std::move(key))
.then(
[prom](const std::string& value) mutable {
prom.respond(http::status::ok, "text/plain", value);
},
[prom](const caf::error& what) mutable {
if (what == caf::sec::no_such_key)
prom.respond(http::status::not_found, "text/plain",
"Key not found.");
else
prom.respond(http::status::internal_server_error,
what);
});
})
.route("/api/<arg>", http::method::post,
[kvs](http::responder& res, std::string key) {
auto value = res.payload();
if (!is_ascii(value)) {
res.respond(http::status::bad_request, "text/plain",
"Expected an ASCII payload.");
return;
}
auto* self = res.self();
auto prom = std::move(res).to_promise();
self
->request(kvs, 2s, caf::put_atom_v, std::move(key),
to_ascii(value))
.then(
[prom]() mutable {
prom.respond(http::status::no_content);
},
[prom](const caf::error& what) mutable {
prom.respond(http::status::internal_server_error, what);
});
})
.route("/api/<arg>", http::method::del,
[kvs](http::responder& res, std::string key) {
auto* self = res.self();
auto prom = std::move(res).to_promise();
self->request(kvs, 2s, caf::delete_atom_v, std::move(key))
.then(
[prom]() mutable {
prom.respond(http::status::no_content);
},
[prom](const caf::error& what) mutable {
prom.respond(http::status::internal_server_error, what);
});
})
.route("/status", http::method::get,
[kvs](http::responder& res) {
res.respond(http::status::no_content);
})
// Launch the server.
.start();
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Note: the actor system will keep the application running for as long as the
// kvs actor stays alive.
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
// Simple HTTPS server that tells the time.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/http/serve.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <string>
#include <string_view>
#include <utility>
using namespace std::literals;
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<std::string>("cert-file", "PEM server certificate file")
.add<std::string>("key-file", "PEM key file for the certificate");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
using namespace caf;
// Sanity checking.
auto cert_file = get_or(cfg, "cert-file", ""sv);
auto key_file = get_or(cfg, "key-file", ""sv);
if (cert_file.empty()) {
std::cerr << "*** mandatory parameter 'cert-file' missing or empty\n";
return EXIT_FAILURE;
}
if (key_file.empty()) {
std::cerr << "*** mandatory parameter 'key-file' missing or empty\n";
return EXIT_FAILURE;
}
// Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port);
auto acc = net::ssl::acceptor::make_with_cert_file(port, cert_file, key_file);
if (!acc) {
std::cerr << "*** unable to initialize TLS: " << to_string(acc.error())
<< '\n';
return EXIT_FAILURE;
}
std::cout << "*** started listening for incoming connections on port " << port
<< '\n';
// Create buffers to signal events from the WebSocket server to the worker.
auto [worker_pull, server_push] = net::http::make_request_resource();
// Spin up the HTTP server.
net::http::serve(sys, std::move(*acc), std::move(server_push));
// Spin up a worker to handle the HTTP requests.
auto worker = sys.spawn([wres = worker_pull](event_based_actor* self) {
// For each incoming request ...
wres
.observe_on(self) //
.for_each([](const net::http::request& req) {
// ... we simply return the current time as string.
// Note: we cannot respond more than once to a request.
auto str = caf::deep_to_string(make_timestamp());
req.respond(net::http::status::ok, "text/plain", str);
});
});
sys.await_all_actors_done();
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -5,59 +5,83 @@
#include "caf/caf_main.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/http/serve.hpp"
#include "caf/net/http/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <string>
#include <utility>
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 8080;
static constexpr size_t default_max_connections = 128;
// -- configuration ------------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<size_t>("max-connections,m", "limit for concurrent clients");
opt_group{custom_options_, "tls"} //
.add<std::string>("key-file,k", "path to the private key file")
.add<std::string>("cert-file,c", "path to the certificate file");
}
};
// -- main ---------------------------------------------------------------------
// --(rst-main-begin)--
int caf_main(caf::actor_system& sys, const config& cfg) {
using namespace caf;
// Open up a TCP port for incoming connections.
namespace http = caf::net::http;
namespace ssl = caf::net::ssl;
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
net::tcp_accept_socket fd;
if (auto maybe_fd = net::make_tcp_accept_socket({ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
auto pem = ssl::format::pem;
auto key_file = caf::get_as<std::string>(cfg, "tls.key-file");
auto cert_file = caf::get_as<std::string>(cfg, "tls.cert-file");
auto max_connections = caf::get_or(cfg, "max-connections",
default_max_connections);
if (!key_file != !cert_file) {
std::cerr << "*** inconsistent TLS config: declare neither file or both\n";
return EXIT_FAILURE;
}
// Open up a TCP port for incoming connections and start the server.
auto server
= http::with(sys)
// Optionally enable TLS.
.context(ssl::context::enable(key_file && cert_file)
.and_then(ssl::emplace_server(ssl::tls::v1_2))
.and_then(ssl::use_private_key_file(key_file, pem))
.and_then(ssl::use_certificate_file(cert_file, pem)))
// Bind to the user-defined port.
.accept(port)
// Limit how many clients may be connected at any given time.
.max_connections(max_connections)
// Provide the time at '/'.
.route("/", http::method::get,
[](http::responder& res) {
auto str = caf::deep_to_string(caf::make_timestamp());
res.respond(http::status::ok, "text/plain", str);
})
// Launch the server.
.start();
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Create buffers to signal events from the WebSocket server to the worker.
auto [worker_pull, server_push] = net::http::make_request_resource();
// Spin up the HTTP server.
net::http::serve(sys, fd, std::move(server_push));
// Spin up a worker to handle the HTTP requests.
auto worker = sys.spawn([wres = worker_pull](event_based_actor* self) {
// For each incoming request ...
wres
.observe_on(self) //
.for_each([](const net::http::request& req) {
// ... we simply return the current time as string.
// Note: we cannot respond more than once to a request.
auto str = caf::deep_to_string(make_timestamp());
req.respond(net::http::status::ok, "text/plain", str);
});
});
sys.await_all_actors_done();
// Note: the actor system will only wait for actors on default. Since we don't
// start actors, we need to block on something else.
std::cout << "Server is up and running. Press <enter> to shut down.\n";
getchar();
std::cout << "Terminating.\n";
return EXIT_SUCCESS;
}
// --(rst-main-end)--
CAF_MAIN(caf::net::middleman)
......@@ -4,11 +4,8 @@
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/length_prefix_framing.hpp"
#include "caf/net/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
#include "caf/uuid.hpp"
......@@ -17,21 +14,10 @@
#include <iostream>
#include <utility>
// -- convenience type aliases -------------------------------------------------
using namespace std::literals;
// The trait for translating between bytes on the wire and flow items. The
// binary default trait operates on binary::frame items.
using trait = caf::net::binary::default_trait;
// Takes care converting a byte stream into a sequence of messages on the wire.
using lpf = caf::net::length_prefix_framing::bind<trait>;
// An implicitly shared type for storing a binary frame.
using bin_frame = caf::net::binary::frame;
// Each client gets a UUID for identifying it. While processing messages, we add
// this ID to the input to tag it.
using message_t = std::pair<caf::uuid, bin_frame>;
namespace lp = caf::net::lp;
namespace ssl = caf::net::ssl;
// -- constants ----------------------------------------------------------------
......@@ -47,67 +33,86 @@ struct config : caf::actor_system_config {
.add<uint16_t>("port,p", "port of the server")
.add<std::string>("host,H", "host of the server")
.add<std::string>("name,n", "set name");
opt_group{custom_options_, "tls"} //
.add<bool>("enable", "enables encryption via TLS")
.add<std::string>("ca-file", "CA file for trusted servers");
}
};
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
// Connect to the server.
// Read the configuration.
auto use_ssl = caf::get_or(cfg, "tls.enable", false);
auto port = caf::get_or(cfg, "port", default_port);
auto host = caf::get_or(cfg, "host", default_host);
auto name = caf::get_or(cfg, "name", "");
auto ca_file = caf::get_as<std::string>(cfg, "tls.ca-file");
if (name.empty()) {
std::cerr << "*** mandatory parameter 'name' missing or empty\n";
return EXIT_FAILURE;
}
auto fd = caf::net::make_connected_tcp_stream_socket(host, port);
if (!fd) {
// Connect to the server.
auto conn
= caf::net::lp::with(sys)
// Optionally enable TLS.
.context(ssl::context::enable(use_ssl)
.and_then(ssl::emplace_client(ssl::tls::v1_2))
.and_then(ssl::load_verify_file_if(ca_file)))
// Connect to "$host:$port".
.connect(host, port)
// If we don't succeed at first, try up to 10 times with 1s delay.
.retry_delay(1s)
.max_retry_count(9)
// After connecting, spin up a worker that prints received inputs.
.start([&sys, name](auto pull, auto push) {
sys.spawn([pull](caf::event_based_actor* self) {
pull
.observe_on(self) //
.do_on_error([](const caf::error& err) {
std::cout << "*** connection error: " << to_string(err) << '\n';
})
.do_finally([self] {
std::cout << "*** lost connection to server -> quit\n"
<< "*** use CTRL+D or CTRL+C to terminate\n";
self->quit();
})
.for_each([](const lp::frame& frame) {
// Interpret the bytes as ASCII characters.
auto bytes = frame.bytes();
auto str = std::string_view{
reinterpret_cast<const char*>(bytes.data()), bytes.size()};
if (std::all_of(str.begin(), str.end(), ::isprint)) {
std::cout << str << '\n';
} else {
std::cout << "<non-ascii-data of size " << bytes.size()
<< ">\n";
}
});
});
// Spin up a second worker that reads from std::cin and sends each
// line to the server. Put that to its own thread since it's doing
// blocking I/O calls.
sys.spawn<caf::detached>([push, name] {
auto lines = caf::async::make_blocking_producer(push);
if (!lines)
throw std::logic_error("failed to create blocking producer");
auto line = std::string{};
auto prefix = name + ": ";
while (std::getline(std::cin, line)) {
line.insert(line.begin(), prefix.begin(), prefix.end());
lines->push(lp::frame{caf::as_bytes(caf::make_span(line))});
line.clear();
}
});
});
if (!conn) {
std::cerr << "*** unable to connect to " << host << ":" << port << ": "
<< to_string(fd.error()) << '\n';
<< to_string(conn.error()) << '\n';
return EXIT_FAILURE;
}
std::cout << "*** connected to " << host << ":" << port << ": " << '\n';
// Create our buffers that connect the worker to the socket.
using caf::async::make_spsc_buffer_resource;
auto [lpf_pull, app_push] = make_spsc_buffer_resource<bin_frame>();
auto [app_pull, lpf_push] = make_spsc_buffer_resource<bin_frame>();
// Spin up the network backend.
lpf::run(sys, *fd, std::move(lpf_pull), std::move(lpf_push));
// Spin up a worker that simply prints received inputs.
sys.spawn([pull = std::move(app_pull)](caf::event_based_actor* self) {
pull
.observe_on(self) //
.do_finally([self] {
std::cout << "*** lost connection to server -> quit\n"
<< "*** use CTRL+D or CTRL+C to terminate\n";
self->quit();
})
.for_each([](const bin_frame& frame) {
// Interpret the bytes as ASCII characters.
auto bytes = frame.bytes();
auto str = std::string_view{reinterpret_cast<const char*>(bytes.data()),
bytes.size()};
if (std::all_of(str.begin(), str.end(), ::isprint)) {
std::cout << str << '\n';
} else {
std::cout << "<non-ascii-data of size " << bytes.size() << ">\n";
}
});
});
// Wait for user input on std::cin and send them to the server.
auto push_buf = app_push.try_open();
assert(push_buf != nullptr);
auto inputs = caf::async::blocking_producer<bin_frame>{std::move(push_buf)};
auto line = std::string{};
auto prefix = name + ": ";
while (std::getline(std::cin, line)) {
line.insert(line.begin(), prefix.begin(), prefix.end());
inputs.push(bin_frame{caf::as_bytes(caf::make_span(line))});
line.clear();
}
// Done. However, the actor system will keep the application running for as
// long as it has open ports or connections.
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
......
......@@ -4,54 +4,46 @@
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/length_prefix_framing.hpp"
#include "caf/net/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/uuid.hpp"
#include <iostream>
#include <utility>
// -- convenience type aliases -------------------------------------------------
// The trait for translating between bytes on the wire and flow items. The
// binary default trait operates on binary::frame items.
using trait = caf::net::binary::default_trait;
// Takes care converting a byte stream into a sequence of messages on the wire.
using lpf = caf::net::length_prefix_framing::bind<trait>;
// An implicitly shared type for storing a binary frame.
using bin_frame = caf::net::binary::frame;
// Each client gets a UUID for identifying it. While processing messages, we add
// this ID to the input to tag it.
using message_t = std::pair<caf::uuid, bin_frame>;
namespace lp = caf::net::lp;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 7788;
static constexpr size_t default_max_connections = 128;
// -- configuration setup ------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<size_t>("max-connections,m", "limit for concurrent clients");
opt_group{custom_options_, "tls"} //
.add<std::string>("key-file,k", "path to the private key file")
.add<std::string>("cert-file,c", "path to the certificate file");
}
};
// -- multiplexing logic -------------------------------------------------------
void worker_impl(caf::event_based_actor* self,
caf::async::consumer_resource<lpf::accept_event_t> events) {
lp::default_trait::acceptor_resource events) {
// Each client gets a UUID for identifying it. While processing messages, we
// add this ID to the input to tag it.
using message_t = std::pair<caf::uuid, lp::frame>;
// Allows us to push new flows into the central merge point.
caf::flow::item_publisher<caf::flow::observable<message_t>> msg_pub{self};
caf::flow::item_publisher<caf::flow::observable<message_t>> pub{self};
// Our central merge point combines all inputs into a single, shared flow.
auto messages = msg_pub.as_observable().merge().share();
auto messages = pub.as_observable().merge().share();
// Have one subscription for debug output. This also makes sure that the
// shared observable stays subscribed to the merger.
messages.for_each([](const message_t& msg) {
......@@ -63,8 +55,7 @@ void worker_impl(caf::event_based_actor* self,
events
.observe_on(self) //
.for_each(
[self, messages, pub = std::move(msg_pub)] //
(const lpf::accept_event_t& event) mutable {
[self, messages, pub = std::move(pub)](const auto& event) mutable {
// Each connection gets a unique ID.
auto conn = caf::uuid::random();
std::cout << "*** accepted new connection " << to_string(conn) << '\n';
......@@ -81,16 +72,19 @@ void worker_impl(caf::event_based_actor* self,
})
.subscribe(push);
// Feed messages from the `pull` end into the central merge point.
auto inputs = pull.observe_on(self)
.on_error_complete() // Cary on if a connection breaks.
.do_on_complete([conn] {
std::cout << "*** lost connection " << to_string(conn)
<< '\n';
})
.map([conn](const bin_frame& frame) {
return message_t{conn, frame};
})
.as_observable();
auto inputs
= pull.observe_on(self)
.do_on_error([](const caf::error& err) {
std::cout << "*** connection error: " << to_string(err) << '\n';
})
.on_error_complete() // Cary on if a connection breaks.
.do_on_complete([conn] {
std::cout << "*** lost connection " << to_string(conn) << '\n';
})
.map([conn](const lp::frame& frame) {
return message_t{conn, frame};
})
.as_observable();
pub.push(inputs);
});
}
......@@ -98,28 +92,42 @@ void worker_impl(caf::event_based_actor* self,
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
// Open up a TCP port for incoming connections.
namespace ssl = caf::net::ssl;
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
caf::net::tcp_accept_socket fd;
if (auto maybe_fd = caf::net::make_tcp_accept_socket(port)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
auto pem = ssl::format::pem;
auto key_file = caf::get_as<std::string>(cfg, "tls.key-file");
auto cert_file = caf::get_as<std::string>(cfg, "tls.cert-file");
auto max_connections = caf::get_or(cfg, "max-connections",
default_max_connections);
if (!key_file != !cert_file) {
std::cerr << "*** inconsistent TLS config: declare neither file or both\n";
return EXIT_FAILURE;
}
// Open up a TCP port for incoming connections and start the server.
auto server
= caf::net::lp::with(sys)
// Optionally enable TLS.
.context(ssl::context::enable(key_file && cert_file)
.and_then(ssl::emplace_server(ssl::tls::v1_2))
.and_then(ssl::use_private_key_file(key_file, pem))
.and_then(ssl::use_certificate_file(cert_file, pem)))
// Bind to the user-defined port.
.accept(port)
// Limit how many clients may be connected at any given time.
.max_connections(max_connections)
// When started, run our worker actor to handle incoming connections.
.start([&sys](lp::default_trait::acceptor_resource accept_events) {
sys.spawn(worker_impl, std::move(accept_events));
});
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Create buffers to signal events from the caf.net backend to the worker.
auto [wres, sres] = lpf::make_accept_event_resources();
// Spin up a worker to multiplex the messages.
auto worker = sys.spawn(worker_impl, std::move(wres));
// Set everything in motion.
lpf::accept(sys, fd, std::move(sres));
// Done. However, the actor system will keep the application running for as
// long as actors are still alive and for as long as it has open ports or
// connections. Since we never close the accept socket, this means the server
// is running indefinitely until the process gets killed (e.g., via CTRL+C).
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
......
......@@ -16,6 +16,8 @@ CAF_POP_WARNINGS
using namespace std;
using namespace caf;
namespace lp = caf::net::lp;
ChatWidget::ChatWidget(QWidget* parent)
: super(parent), input_(nullptr), output_(nullptr) {
// nop
......@@ -26,8 +28,8 @@ ChatWidget::~ChatWidget() {
}
void ChatWidget::init(actor_system& system, const std::string& name,
caf::async::consumer_resource<bin_frame> pull,
caf::async::producer_resource<bin_frame> push) {
caf::async::consumer_resource<lp::frame> pull,
caf::async::producer_resource<lp::frame> push) {
name_ = QString::fromUtf8(name);
print("*** hello " + name_);
super::init(system);
......@@ -37,7 +39,7 @@ void ChatWidget::init(actor_system& system, const std::string& name,
.do_finally([this] { //
print("*** chatroom offline: lost connection to the server");
})
.for_each([this](const bin_frame& frame) {
.for_each([this](const lp::frame& frame) {
auto bytes = frame.bytes();
auto str = std::string_view{reinterpret_cast<const char*>(bytes.data()),
bytes.size()};
......@@ -56,7 +58,7 @@ void ChatWidget::init(actor_system& system, const std::string& name,
auto encoded = str.toUtf8();
auto bytes = caf::as_bytes(
caf::make_span(encoded.data(), static_cast<size_t>(encoded.size())));
return bin_frame{bytes};
return lp::frame{bytes};
})
.subscribe(push);
set_message_handler([=](actor_companion*) -> message_handler {
......
......@@ -5,7 +5,7 @@
#include "caf/all.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/mixin/actor_widget.hpp"
#include "caf/net/binary/frame.hpp"
#include "caf/net/lp/frame.hpp"
CAF_PUSH_WARNINGS
#include <QLineEdit>
......@@ -30,7 +30,7 @@ public:
using super = caf::mixin::actor_widget<QWidget>;
using bin_frame = caf::net::binary::frame;
using frame = caf::net::lp::frame;
using publisher_type = caf::flow::item_publisher<QString>;
......@@ -39,8 +39,8 @@ public:
~ChatWidget();
void init(caf::actor_system& system, const std::string& name,
caf::async::consumer_resource<bin_frame> pull,
caf::async::producer_resource<bin_frame> push);
caf::async::consumer_resource<frame> pull,
caf::async::producer_resource<frame> push);
public slots:
......
......@@ -18,10 +18,8 @@
#include <vector>
#include "caf/all.hpp"
#include "caf/net/length_prefix_framing.hpp"
#include "caf/net/lp/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
CAF_PUSH_WARNINGS
#include "ui_chatwindow.h" // auto generated from chatwindow.ui
......@@ -54,7 +52,7 @@ public:
// -- main ---------------------------------------------------------------------
int caf_main(actor_system& sys, const config& cfg) {
// Connect to the server.
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
auto host = caf::get_or(cfg, "host", default_host);
auto name = caf::get_or(cfg, "name", "");
......@@ -62,22 +60,6 @@ int caf_main(actor_system& sys, const config& cfg) {
std::cerr << "*** mandatory parameter 'name' missing or empty\n";
return EXIT_FAILURE;
}
auto fd = caf::net::make_connected_tcp_stream_socket(host, port);
if (!fd) {
std::cerr << "*** unable to connect to " << host << ":" << port << ": "
<< to_string(fd.error()) << '\n';
return EXIT_FAILURE;
}
std::cout << "*** connected to " << host << ":" << port << ": " << '\n';
// Create our buffers that connect the worker to the socket.
using bin_frame = caf::net::binary::frame;
using caf::async::make_spsc_buffer_resource;
auto [lpf_pull, app_push] = make_spsc_buffer_resource<bin_frame>();
auto [app_pull, lpf_push] = make_spsc_buffer_resource<bin_frame>();
// Spin up the network backend.
using trait = caf::net::binary::default_trait;
using lpf = caf::net::length_prefix_framing::bind<trait>;
auto conn = lpf::run(sys, *fd, std::move(lpf_pull), std::move(lpf_push));
// Spin up Qt.
auto [argc, argv] = cfg.c_args_remainder();
QApplication app{argc, argv};
......@@ -85,12 +67,24 @@ int caf_main(actor_system& sys, const config& cfg) {
QMainWindow mw;
Ui::ChatWindow helper;
helper.setupUi(&mw);
helper.chatwidget->init(sys, name, std::move(app_pull), std::move(app_push));
// Connect to the server.
auto conn
= caf::net::lp::with(sys)
.connect(host, port)
.start([&](auto pull, auto push) {
std::cout << "*** connected to " << host << ":" << port << '\n';
helper.chatwidget->init(sys, name, std::move(pull), std::move(push));
});
if (!conn) {
std::cerr << "*** unable to connect to " << host << ":" << port << ": "
<< to_string(conn.error()) << '\n';
mw.close();
return app.exec();
}
// Setup and run.
auto client = helper.chatwidget->as_actor();
mw.show();
auto result = app.exec();
conn.dispose();
conn->dispose();
return result;
}
......
......@@ -5,88 +5,119 @@
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.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/with.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <utility>
static constexpr uint16_t default_port = 8080;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 7788;
static constexpr size_t default_max_connections = 128;
// -- configuration setup ------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<size_t>("max-connections,m", "limit for concurrent clients");
opt_group{custom_options_, "tls"} //
.add<std::string>("key-file,k", "path to the private key file")
.add<std::string>("cert-file,c", "path to the certificate file");
}
};
// -- main ---------------------------------------------------------------------
// --(rst-main-begin)--
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Open up a TCP port for incoming connections.
namespace http = caf::net::http;
namespace ssl = caf::net::ssl;
namespace ws = caf::net::web_socket;
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port})) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
auto pem = ssl::format::pem;
auto key_file = caf::get_as<std::string>(cfg, "tls.key-file");
auto cert_file = caf::get_as<std::string>(cfg, "tls.cert-file");
auto max_connections = caf::get_or(cfg, "max-connections",
default_max_connections);
if (!key_file != !cert_file) {
std::cerr << "*** inconsistent TLS config: declare neither file or both\n";
return EXIT_FAILURE;
}
// Convenience type aliases.
using event_t = ws::accept_event_t<>;
// Create buffers to signal events from the WebSocket server to the worker.
auto [wres, sres] = ws::make_accept_event_resources();
// Spin up a worker to handle the events.
auto worker = sys.spawn([worker_res = wres](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ...
self->make_observable()
.from_resource(worker_res)
.for_each([self](const event_t& event) {
// ... that simply pushes data back to the sender.
auto [pull, push] = event.data();
pull.observe_on(self)
.do_on_next([](const ws::frame& x) {
if (x.is_binary()) {
std::cout << "*** received a binary WebSocket frame of size "
// Open up a TCP port for incoming connections and start the server.
using trait = ws::default_trait;
auto server
= ws::with(sys)
// Optionally enable TLS.
.context(ssl::context::enable(key_file && cert_file)
.and_then(ssl::emplace_server(ssl::tls::v1_2))
.and_then(ssl::use_private_key_file(key_file, pem))
.and_then(ssl::use_certificate_file(cert_file, pem)))
// Bind to the user-defined port.
.accept(port)
// Limit how many clients may be connected at any given time.
.max_connections(max_connections)
// Accept only requests for path "/".
.on_request([](ws::acceptor<>& acc) {
// The header parameter contains fields from the WebSocket handshake
// such as the path and HTTP header fields..
auto path = acc.header().path();
std::cout << "*** new client request for path " << path << '\n';
// Accept the WebSocket connection only if the path is "/".
if (path == "/") {
// Calling `accept` causes the server to acknowledge the client and
// creates input and output buffers that go to the worker actor.
acc.accept();
} else {
// Calling `reject` aborts the connection with HTTP status code 400
// (Bad Request). The error gets converted to a string and sent to
// the client to give some indication to the client why the request
// was rejected.
auto err = caf::make_error(caf::sec::invalid_argument,
"unrecognized path, try '/'");
acc.reject(std::move(err));
}
// Note: calling nothing on `acc` also rejects the connection.
})
// When started, run our worker actor to handle incoming connections.
.start([&sys](trait::acceptor_resource<> events) {
sys.spawn([events](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ...
self->make_observable()
.from_resource(events) //
.for_each([self](const trait::accept_event<>& ev) {
// ... that simply pushes data back to the sender.
auto [pull, push] = ev.data();
pull.observe_on(self)
.do_on_next([](const ws::frame& x) {
if (x.is_binary()) {
std::cout
<< "*** received a binary WebSocket frame of size "
<< x.size() << '\n';
} else {
std::cout << "*** received a text WebSocket frame of size "
} else {
std::cout
<< "*** received a text WebSocket frame of size "
<< x.size() << '\n';
}
})
.subscribe(push);
});
});
// Callback for incoming WebSocket requests.
auto on_request = [](const caf::settings& hdr, auto& req) {
// The hdr parameter is a dictionary with fields from the WebSocket
// handshake such as the path.
auto path = caf::get_or(hdr, "web-socket.path", "/");
std::cout << "*** new client request for path " << path << '\n';
// Accept the WebSocket connection only if the path is "/".
if (path == "/") {
// Calling `accept` causes the server to acknowledge the client and
// creates input and output buffers that go to the worker actor.
req.accept();
} else {
// Calling `reject` aborts the connection with HTTP status code 400 (Bad
// Request). The error gets converted to a string and send to the client
// to give some indication to the client why the request was rejected.
auto err = caf::make_error(caf::sec::invalid_argument,
"unrecognized path, try '/'");
req.reject(std::move(err));
}
// Note: calling neither accept nor reject also rejects the connection.
};
// Set everything in motion.
ws::accept(sys, fd, std::move(sres), on_request);
}
})
.subscribe(push);
});
});
});
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
// --(rst-main-end)--
CAF_MAIN(caf::net::middleman)
......@@ -5,14 +5,14 @@
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/connect.hpp"
#include "caf/net/web_socket/with.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <utility>
using namespace std::literals;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"}
......@@ -22,63 +22,78 @@ struct config : caf::actor_system_config {
}
};
// --(rst-main-begin)--
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
namespace ws = caf::net::web_socket;
// Sanity checking.
auto server = caf::get_or(cfg, "server", caf::uri{});
if (!server.valid()) {
auto server = caf::get_as<caf::uri>(cfg, "server");
if (!server) {
std::cerr << "*** mandatory argument server missing or invalid\n";
return EXIT_FAILURE;
}
// Ask user for the hello message.
// Ask the user for the hello message.
std::string hello;
std::cout << "Please enter a hello message for the server: " << std::flush;
std::getline(std::cin, hello);
// Spin up the WebSocket.
auto hs_setup = [&cfg](ws::handshake& hs) {
if (auto str = caf::get_as<std::string>(cfg, "protocols");
str && !str->empty())
hs.protocols(std::move(*str));
};
auto conn = ws::connect(sys, server, hs_setup);
// Try to establish a connection to the server and send the hello message.
auto conn
= ws::with(sys)
// Connect to the given URI.
.connect(server)
// If we don't succeed at first, try up to 10 times with 1s delay.
.retry_delay(1s)
.max_retry_count(9)
// On success, spin up a worker to manage the connection.
.start([&sys, hello](auto pull, auto push) {
sys.spawn([hello, pull, push](caf::event_based_actor* self) {
// Open the pull handle.
pull
.observe_on(self)
// Print errors to stderr.
.do_on_error([](const caf::error& what) {
std::cerr << "*** error while reading from the WebSocket: "
<< to_string(what) << '\n';
})
// Restrict how many messages we receive if the user configured
// a limit.
.compose([self](auto in) {
if (auto limit = caf::get_as<size_t>(self->config(), "max")) {
return std::move(in).take(*limit).as_observable();
} else {
return std::move(in).as_observable();
}
})
// Print a bye-bye message if the server closes the connection.
.do_on_complete([] { //
std::cout << "Server has closed the connection\n";
})
// Print everything from the server to stdout.
.for_each([](const ws::frame& msg) {
if (msg.is_text()) {
std::cout << "Server: " << msg.as_text() << '\n';
} else if (msg.is_binary()) {
std::cout << "Server: [binary message of size "
<< msg.as_binary().size() << "]\n";
}
});
// Send our hello message and wait until the server closes the
// socket. We keep the connection alive by never closing the write
// channel.
self->make_observable()
.just(ws::frame{hello})
.concat(self->make_observable().never<ws::frame>())
.subscribe(push);
});
});
if (!conn) {
std::cerr << "*** failed to connect: " << to_string(conn.error()) << '\n';
std::cerr << "*** unable to connect to " << server->str() << ": "
<< to_string(conn.error()) << '\n';
return EXIT_FAILURE;
}
conn->run([&](const ws::connect_event_t& conn) {
sys.spawn([conn, hello](caf::event_based_actor* self) {
auto [pull, push] = conn.data();
// Print everything from the server to stdout.
pull.observe_on(self)
.do_on_error([](const caf::error& what) {
std::cerr << "*** error while reading from the WebSocket: "
<< to_string(what) << '\n';
})
.compose([self](auto in) {
if (auto limit = caf::get_as<size_t>(self->config(), "max")) {
return std::move(in).take(*limit).as_observable();
} else {
return std::move(in).as_observable();
}
})
.do_finally([] { std::cout << "Server has closed the connection\n"; })
.for_each([](const ws::frame& msg) {
if (msg.is_text()) {
std::cout << "Server: " << msg.as_text() << '\n';
} else if (msg.is_binary()) {
std::cout << "Server: [binary message of size "
<< msg.as_binary().size() << "]\n";
}
});
// Send our hello message and wait until the server closes the socket.
self->make_observable()
.just(ws::frame{hello})
.concat(self->make_observable().never<ws::frame>())
.subscribe(push);
});
});
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
// --(rst-main-end)--
CAF_MAIN(caf::net::middleman)
This diff is collapsed.
// Simple WebSocket server with TLS that sends everything it receives back to
// the sender.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/accept.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <string>
#include <string_view>
#include <utility>
using namespace std::literals;
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<std::string>("cert-file", "PEM server certificate file")
.add<std::string>("key-file", "PEM key file for the certificate");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Sanity checking.
auto cert_file = caf::get_or(cfg, "cert-file", ""sv);
auto key_file = caf::get_or(cfg, "key-file", ""sv);
if (cert_file.empty()) {
std::cerr << "*** mandatory parameter 'cert-file' missing or empty\n";
return EXIT_FAILURE;
}
if (key_file.empty()) {
std::cerr << "*** mandatory parameter 'key-file' missing or empty\n";
return EXIT_FAILURE;
}
// Create the OpenSSL context and set key and certificate.
auto port = caf::get_or(cfg, "port", default_port);
auto acc = cn::ssl::acceptor::make_with_cert_file(port, cert_file, key_file);
if (!acc) {
std::cerr << "*** unable to initialize TLS: " << to_string(acc.error())
<< '\n';
return EXIT_FAILURE;
}
std::cout << "*** started listening for incoming connections on port " << port
<< '\n';
// Convenience type aliases.
using event_t = ws::accept_event_t<>;
// Create buffers to signal events from the WebSocket server to the worker.
auto [wres, sres] = ws::make_accept_event_resources();
// Spin up a worker to handle the events.
auto worker = sys.spawn([worker_res = wres](caf::event_based_actor* self) {
// For each buffer pair, we create a new flow ...
self->make_observable()
.from_resource(worker_res)
.for_each([self](const event_t& event) {
// ... that simply pushes data back to the sender.
auto [pull, push] = event.data();
pull.observe_on(self)
.do_on_next([](const ws::frame& x) {
if (x.is_binary()) {
std::cout << "*** received a binary WebSocket frame of size "
<< x.size() << '\n';
} else {
std::cout << "*** received a text WebSocket frame of size "
<< x.size() << '\n';
}
})
.subscribe(push);
});
});
// Callback for incoming WebSocket requests.
auto on_request = [](const caf::settings& hdr, auto& req) {
// The hdr parameter is a dictionary with fields from the WebSocket
// handshake such as the path.
auto path = caf::get_or(hdr, "web-socket.path", "/");
std::cout << "*** new client request for path " << path << '\n';
// Accept the WebSocket connection only if the path is "/".
if (path == "/") {
// Calling `accept` causes the server to acknowledge the client and
// creates input and output buffers that go to the worker actor.
req.accept();
} else {
// Calling `reject` aborts the connection with HTTP status code 400 (Bad
// Request). The error gets converted to a string and send to the client
// to give some indication to the client why the request was rejected.
auto err = caf::make_error(caf::sec::invalid_argument,
"unrecognized path, try '/'");
req.reject(std::move(err));
}
// Note: calling neither accept nor reject also rejects the connection.
};
// Set everything in motion.
ws::accept(sys, std::move(*acc), std::move(sres), on_request);
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -9,10 +9,7 @@
#include "caf/event_based_actor.hpp"
#include "caf/json_writer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.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/with.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/span.hpp"
......@@ -24,8 +21,14 @@
using namespace std::literals;
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 8080;
static constexpr size_t default_max_connections = 128;
// -- custom types -------------------------------------------------------------
namespace stock {
struct info {
......@@ -47,12 +50,15 @@ bool inspect(Inspector& f, info& x) {
} // namespace stock
// -- actor for generating a random feed ---------------------------------------
struct random_feed_state {
using trait = caf::net::web_socket::default_trait;
using frame = caf::net::web_socket::frame;
using accept_event = caf::net::web_socket::accept_event_t<>;
using accept_event = trait::accept_event<>;
random_feed_state(caf::event_based_actor* selfptr,
caf::async::consumer_resource<accept_event> events,
trait::acceptor_resource<> events,
caf::timespan update_interval)
: self(selfptr), val_dist(0, 100000), index_dist(0, 19) {
// Init random number generator.
......@@ -132,39 +138,69 @@ struct random_feed_state {
using random_feed_impl = caf::stateful_actor<random_feed_state>;
// -- configuration setup ------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<size_t>("max-connections,m", "limit for concurrent clients")
.add<caf::timespan>("interval,i", "update interval");
;
opt_group{custom_options_, "tls"} //
.add<std::string>("key-file,k", "path to the private key file")
.add<std::string>("cert-file,c", "path to the certificate file");
}
};
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Open up a TCP port for incoming connections.
namespace http = caf::net::http;
namespace ssl = caf::net::ssl;
namespace ws = caf::net::web_socket;
// Read the configuration.
auto interval = caf::get_or(cfg, "interval", caf::timespan{1s});
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_accept_socket fd;
if (auto maybe_fd = cn::make_tcp_accept_socket({caf::ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
auto pem = ssl::format::pem;
auto key_file = caf::get_as<std::string>(cfg, "tls.key-file");
auto cert_file = caf::get_as<std::string>(cfg, "tls.cert-file");
auto max_connections = caf::get_or(cfg, "max-connections",
default_max_connections);
if (!key_file != !cert_file) {
std::cerr << "*** inconsistent TLS config: declare neither file or both\n";
return EXIT_FAILURE;
}
// Create buffers to signal events from the WebSocket server to the worker.
auto [wres, sres] = ws::make_accept_event_resources();
// Spin up a worker to handle the events.
auto interval = caf::get_or(cfg, "interval", caf::timespan{1s});
auto worker = sys.spawn<random_feed_impl>(std::move(wres), interval);
// Callback for incoming WebSocket requests.
auto on_request = [](const caf::settings&, auto& req) { req.accept(); };
// Set everything in motion.
ws::accept(sys, fd, std::move(sres), on_request);
// Open up a TCP port for incoming connections and start the server.
using trait = ws::default_trait;
auto server
= ws::with(sys)
// Optionally enable TLS.
.context(ssl::context::enable(key_file && cert_file)
.and_then(ssl::emplace_server(ssl::tls::v1_2))
.and_then(ssl::use_private_key_file(key_file, pem))
.and_then(ssl::use_certificate_file(cert_file, pem)))
// Bind to the user-defined port.
.accept(port)
// Limit how many clients may be connected at any given time.
.max_connections(max_connections)
// Add handler for incoming connections.
.on_request([](ws::acceptor<>& acc) {
// Ignore all header fields and accept the connection.
acc.accept();
})
// When started, run our worker actor to handle incoming connections.
.start([&sys, interval](trait::acceptor_resource<> events) {
sys.spawn<random_feed_impl>(std::move(events), interval);
});
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
......
......@@ -154,4 +154,10 @@ namespace caf::defaults::net {
/// previous connection has been closed.
constexpr auto max_connections = make_parameter("max-connections", size_t{64});
/// The default port for HTTP servers.
constexpr uint16_t http_default_port = 80;
/// The default port for HTTPS servers.
constexpr uint16_t https_default_port = 443;
} // namespace caf::defaults::net
......@@ -1049,6 +1049,13 @@ struct unboxed_oracle<std::optional<T>> {
template <class T>
using unboxed_t = typename unboxed_oracle<T>::type;
/// Evaluates to true if `T` is a std::string or is convertible to a `const
/// char*`.
template <class T>
constexpr bool is_string_or_cstring_v
= std::is_convertible_v<T, const char*>
|| std::is_same_v<std::string, std::decay_t<T>>;
} // namespace caf::detail
#undef CAF_HAS_MEMBER_TRAIT
......
......@@ -148,7 +148,7 @@ public:
/// Returns a copy of `this` if `!empty()` or else returns a new error from
/// given arguments.
template <class Enum, class... Ts>
error or_else(Enum code, Ts&&... args) const {
error or_else(Enum code, Ts&&... args) const& {
if (!empty())
return *this;
if constexpr (sizeof...(Ts) > 0)
......@@ -157,6 +157,18 @@ public:
return error{code};
}
/// Returns a copy of `this` if `!empty()` or else returns a new error from
/// given arguments.
template <class Enum, class... Ts>
error or_else(Enum code, Ts&&... args) && {
if (!empty())
return std::move(*this);
if constexpr (sizeof...(Ts) > 0)
return error{code, make_message(std::forward<Ts>(args)...)};
else
return error{code};
}
// -- modifiers --------------------------------------------------------------
/// Reverts this error to "not an error" as if calling `*this = error{}`.
......
......@@ -499,7 +499,9 @@ template <class OnNext>
disposable observable<T>::for_each(OnNext on_next) {
static_assert(std::is_invocable_v<OnNext, const T&>,
"for_each: the on_next function must accept a 'const T&'");
return subscribe(make_observer(std::move(on_next)));
using impl_type = detail::default_observer_impl<T, OnNext>;
auto ptr = make_counted<impl_type>(std::move(on_next));
return subscribe(observer<T>{std::move(ptr)});
}
// -- observable: transforming -------------------------------------------------
......
......@@ -157,15 +157,17 @@ using on_next_trait_t
template <class F>
using on_next_value_type = typename on_next_trait_t<F>::value_type;
template <class OnNext, class OnError = unit_t, class OnComplete = unit_t>
class default_observer_impl
: public flow::observer_impl_base<on_next_value_type<OnNext>> {
template <class T, class OnNext, class OnError = unit_t,
class OnComplete = unit_t>
class default_observer_impl : public flow::observer_impl_base<T> {
public:
static_assert(std::is_invocable_v<OnNext, const T&>);
static_assert(std::is_invocable_v<OnError, const error&>);
static_assert(std::is_invocable_v<OnComplete>);
using input_type = on_next_value_type<OnNext>;
using input_type = T;
explicit default_observer_impl(OnNext&& on_next_fn)
: on_next_(std::move(on_next_fn)) {
......@@ -230,8 +232,9 @@ namespace caf::flow {
/// @param on_complete Callback for handling the end-of-stream event.
template <class OnNext, class OnError, class OnComplete>
auto make_observer(OnNext on_next, OnError on_error, OnComplete on_complete) {
using impl_type = detail::default_observer_impl<OnNext, OnError, OnComplete>;
using input_type = typename impl_type::input_type;
using input_type = detail::on_next_value_type<OnNext>;
using impl_type
= detail::default_observer_impl<input_type, OnNext, OnError, OnComplete>;
auto ptr = make_counted<impl_type>(std::move(on_next), std::move(on_error),
std::move(on_complete));
return observer<input_type>{std::move(ptr)};
......@@ -242,8 +245,8 @@ auto make_observer(OnNext on_next, OnError on_error, OnComplete on_complete) {
/// @param on_error Callback for handling an error.
template <class OnNext, class OnError>
auto make_observer(OnNext on_next, OnError on_error) {
using impl_type = detail::default_observer_impl<OnNext, OnError>;
using input_type = typename impl_type::input_type;
using input_type = detail::on_next_value_type<OnNext>;
using impl_type = detail::default_observer_impl<input_type, OnNext, OnError>;
auto ptr = make_counted<impl_type>(std::move(on_next), std::move(on_error));
return observer<input_type>{std::move(ptr)};
}
......@@ -252,8 +255,8 @@ auto make_observer(OnNext on_next, OnError on_error) {
/// @param on_next Callback for handling incoming elements.
template <class OnNext>
auto make_observer(OnNext on_next) {
using impl_type = detail::default_observer_impl<OnNext>;
using input_type = typename impl_type::input_type;
using input_type = detail::on_next_value_type<OnNext>;
using impl_type = detail::default_observer_impl<input_type, OnNext>;
auto ptr = make_counted<impl_type>(std::move(on_next));
return observer<input_type>{std::move(ptr)};
}
......
......@@ -194,7 +194,7 @@ private:
value_sub_.dispose();
control_sub_.dispose();
switch (state_) {
case state::running:
case state::running: {
if (!buf_.empty()) {
if (demand_ == 0) {
state_ = err_ ? state::aborted : state::completed;
......@@ -204,13 +204,14 @@ private:
out_.on_next(f(buf_));
buf_.clear();
}
auto tmp = std::move(out_);
if (err_)
out_.on_error(err_);
tmp.on_error(err_);
else
out_.on_complete();
out_ = nullptr;
tmp.on_complete();
state_ = state::disposed;
break;
}
default:
break;
}
......@@ -226,11 +227,11 @@ private:
}
if (!buf_.empty())
do_emit();
auto tmp = std::move(out_);
if (err_)
out_.on_error(err_);
tmp.on_error(err_);
else
out_.on_complete();
out_ = nullptr;
tmp.on_complete();
}
void do_emit() {
......@@ -249,8 +250,8 @@ private:
value_sub_.dispose();
control_sub_.dispose();
if (out_) {
out_.on_complete();
out_ = nullptr;
auto tmp = std::move(out_);
tmp.on_complete();
}
state_ = state::disposed;
}
......
......@@ -16,19 +16,46 @@
namespace caf::flow::op {
/// Interface for listening on a cell.
template <class T>
class cell_listener {
public:
friend void intrusive_ptr_add_ref(const cell_listener* ptr) noexcept {
ptr->ref_listener();
}
friend void intrusive_ptr_release(const cell_listener* ptr) noexcept {
ptr->deref_listener();
}
virtual void on_next(const T& item) = 0;
virtual void on_complete() = 0;
virtual void on_error(const error& what) = 0;
virtual void ref_listener() const noexcept = 0;
virtual void deref_listener() const noexcept = 0;
};
/// Convenience alias for a reference-counting smart pointer.
template <class T>
using cell_listener_ptr = intrusive_ptr<cell_listener<T>>;
/// State shared between one multicast operator and one subscribed observer.
template <class T>
struct cell_sub_state {
std::variant<none_t, unit_t, T, error> content;
std::vector<observer<T>> listeners;
std::vector<cell_listener_ptr<T>> listeners;
void set_null() {
CAF_ASSERT(std::holds_alternative<none_t>(content));
content = unit;
std::vector<observer<T>> xs;
std::vector<cell_listener_ptr<T>> xs;
xs.swap(listeners);
for (auto& listener : xs) {
listener.on_complete();
listener->on_complete();
}
}
......@@ -36,11 +63,11 @@ struct cell_sub_state {
CAF_ASSERT(std::holds_alternative<none_t>(content));
content = std::move(item);
auto& ref = std::get<T>(content);
std::vector<observer<T>> xs;
std::vector<cell_listener_ptr<T>> xs;
xs.swap(listeners);
for (auto& listener : xs) {
listener.on_next(ref);
listener.on_complete();
listener->on_next(ref);
listener->on_complete();
}
}
......@@ -48,41 +75,43 @@ struct cell_sub_state {
CAF_ASSERT(std::holds_alternative<none_t>(content));
content = std::move(what);
auto& ref = std::get<error>(content);
std::vector<observer<T>> xs;
std::vector<cell_listener_ptr<T>> xs;
xs.swap(listeners);
for (auto& listener : xs)
listener.on_error(ref);
listener->on_error(ref);
}
void listen(observer<T> listener) {
void listen(cell_listener_ptr<T> listener) {
switch (content.index()) {
case 1:
listener.on_complete();
listener->on_complete();
break;
case 2:
listener.on_next(std::get<T>(content));
listener.on_complete();
listener->on_next(std::get<T>(content));
listener->on_complete();
break;
case 3:
listener.on_error(std::get<error>(content));
listener->on_error(std::get<error>(content));
break;
default:
listeners.emplace_back(std::move(listener));
}
}
void drop(const observer<T>& listener) {
void drop(const cell_listener_ptr<T>& listener) {
if (auto i = std::find(listeners.begin(), listeners.end(), listener);
i != listeners.end())
listeners.erase(i);
}
};
/// Convenience alias for the state of a cell.
template <class T>
using cell_sub_state_ptr = std::shared_ptr<cell_sub_state<T>>;
/// The subscription object for interfacing an observer with the cell state.
template <class T>
class cell_sub : public subscription::impl_base {
class cell_sub : public subscription::impl_base, public cell_listener<T> {
public:
// -- constructors, destructors, and assignment operators --------------------
......@@ -91,6 +120,16 @@ public:
// nop
}
// -- disambiguation ---------------------------------------------------------
friend void intrusive_ptr_add_ref(const cell_sub* ptr) noexcept {
ptr->ref_disposable();
}
friend void intrusive_ptr_release(const cell_sub* ptr) noexcept {
ptr->deref_disposable();
}
// -- implementation of subscription -----------------------------------------
bool disposed() const noexcept override {
......@@ -99,21 +138,56 @@ public:
void dispose() override {
if (state_) {
state_->drop(out_);
state_->drop(this);
state_ = nullptr;
out_ = nullptr;
}
if (out_) {
auto tmp = std::move(out_);
tmp.on_complete();
}
}
void request(size_t) override {
if (!listening_) {
listening_ = true;
ctx_->delay_fn([state = state_, out = out_] { //
state->listen(std::move(out));
auto self = cell_listener_ptr<T>{this};
ctx_->delay_fn([state = state_, self]() mutable { //
state->listen(std::move(self));
});
}
}
// -- implementation of listener<T> ------------------------------------------
void on_next(const T& item) override {
if (out_)
out_.on_next(item);
}
void on_complete() override {
state_ = nullptr;
if (out_) {
auto tmp = std::move(out_);
tmp.on_complete();
}
}
void on_error(const error& what) override {
state_ = nullptr;
if (out_) {
auto tmp = std::move(out_);
tmp.on_error(what);
}
}
void ref_listener() const noexcept override {
ref_disposable();
}
void deref_listener() const noexcept override {
deref_disposable();
}
private:
coordinator* ctx_;
bool listening_ = false;
......
......@@ -147,20 +147,20 @@ private:
void fin(const error* err = nullptr) {
CAF_ASSERT(out_);
if (factory_sub_) {
factory_sub_.dispose();
factory_sub_ = nullptr;
auto tmp = std::move(factory_sub_);
tmp.dispose();
}
if (active_sub_) {
active_sub_.dispose();
active_sub_ = nullptr;
auto tmp = std::move(active_sub_);
tmp.dispose();
}
factory_key_ = 0;
active_key_ = 0;
auto tmp = std::move(out_);
if (err)
out_.on_error(*err);
tmp.on_error(*err);
else
out_.on_complete();
out_ = nullptr;
tmp.on_complete();
}
/// Stores the context (coordinator) that runs this flow.
......
......@@ -97,11 +97,11 @@ private:
}
void fin() {
auto tmp = std::move(out_);
if (!err_)
out_.on_complete();
tmp.on_complete();
else
out_.on_error(err_);
out_ = nullptr;
tmp.on_error(err_);
}
void pull(size_t n) {
......
......@@ -114,12 +114,12 @@ private:
void do_dispose() {
if (buf_) {
buf_->cancel();
buf_ = nullptr;
auto tmp = std::move(buf_);
tmp->cancel();
}
if (out_) {
out_.on_complete();
out_ = nullptr;
auto tmp = std::move(out_);
tmp.on_complete();
}
}
......
......@@ -41,15 +41,16 @@ public:
}
void on_complete() {
CAF_ASSERT(sub->in_.valid());
sub->in_.dispose();
sub->in_ = nullptr;
if (sub->in_) {
auto tmp = std::move(sub->in_);
tmp.dispose();
}
}
void on_error(const error& what) {
if (sub->in_) {
sub->in_.dispose();
sub->in_ = nullptr;
auto tmp = std::move(sub->in_);
tmp.dispose();
}
sub->err_ = what;
}
......@@ -187,8 +188,8 @@ public:
ctx_->delay_fn([out = std::move(out_)]() mutable { out.on_complete(); });
}
if (in_) {
in_.dispose();
in_ = nullptr;
auto tmp = std::move(in_);
tmp.dispose();
}
}
......@@ -235,12 +236,12 @@ private:
if (in_) {
pull();
} else if (buf_.empty()) {
disposed_ = true;
auto tmp = std::move(out_);
if (err_)
out_.on_error(err_);
tmp.on_error(err_);
else
out_.on_complete();
out_ = nullptr;
disposed_ = true;
tmp.on_complete();
}
}
}
......
......@@ -101,8 +101,8 @@ public:
while (i != inputs_.end()) {
auto& input = *i->second;
if (auto& sub = input.sub) {
sub.dispose();
sub = nullptr;
auto tmp = std::move(input.sub);
tmp.dispose();
}
if (input.buf.empty())
i = inputs_.erase(i);
......@@ -215,12 +215,11 @@ private:
}
}
if (out_ && inputs_.empty()) {
if (!err_) {
out_.on_complete();
} else {
out_.on_error(err_);
}
out_ = nullptr;
auto tmp = std::move(out_);
if (!err_)
tmp.on_complete();
else
tmp.on_error(err_);
}
flags_.running = false;
}
......
......@@ -70,8 +70,8 @@ public:
sink_->abort(reason);
sub_ = nullptr;
} else if (out_) {
out_.on_error(reason);
out_ = nullptr;
auto tmp = std::move(out_);
tmp.on_error(reason);
}
}
......@@ -81,8 +81,8 @@ public:
sink_->close();
sub_ = nullptr;
} else if (out_) {
out_.on_complete();
out_ = nullptr;
auto tmp = std::move(out_);
tmp.on_complete();
}
}
......@@ -102,8 +102,8 @@ public:
if (out_) {
out_ = nullptr;
if (sub_) {
sub_.dispose();
sub_ = nullptr;
auto tmp = std::move(sub_);
tmp.dispose();
}
}
}
......@@ -157,8 +157,8 @@ private:
void on_sink_dispose() {
sink_ = nullptr;
if (sub_) {
sub_.dispose();
sub_ = nullptr;
auto tmp = std::move(sub_);
tmp.dispose();
}
}
......
......@@ -123,9 +123,9 @@ private:
void on_dispose(state_type&) override {
try_request_more();
if (auto_disconnect_ && connected_ && super::observer_count() == 0) {
in_.dispose();
in_ = nullptr;
connected_ = false;
auto tmp = std::move(in_);
tmp.dispose();
}
}
......
......@@ -71,13 +71,13 @@ public:
closed = true;
if (!running && buf.empty()) {
disposed = true;
if (out) {
out.on_complete();
out = nullptr;
}
when_disposed = nullptr;
when_consumed_some = nullptr;
when_demand_changed = nullptr;
if (out) {
auto tmp = std::move(out);
tmp.on_complete();
}
}
}
}
......@@ -88,33 +88,33 @@ public:
err = reason;
if (!running && buf.empty()) {
disposed = true;
if (out) {
auto out_hdl = std::move(out);
out_hdl.on_error(reason);
}
when_disposed = nullptr;
when_consumed_some = nullptr;
when_demand_changed = nullptr;
if (out) {
auto tmp = std::move(out);
tmp.on_error(reason);
}
}
}
}
void dispose() {
if (out) {
out.on_complete();
out = nullptr;
}
if (when_disposed) {
ctx->delay(std::move(when_disposed));
}
if (when_consumed_some) {
when_consumed_some.dispose();
when_consumed_some = nullptr;
auto tmp = std::move(when_consumed_some);
tmp.dispose();
}
when_demand_changed = nullptr;
buf.clear();
demand = 0;
disposed = true;
if (out) {
auto tmp = std::move(out);
tmp.on_complete();
}
}
void do_run() {
......@@ -130,11 +130,11 @@ public:
--demand;
}
if (buf.empty() && closed) {
auto tmp = std::move(out);
if (err)
out.on_error(err);
tmp.on_error(err);
else
out.on_complete();
out = nullptr;
tmp.on_complete();
dispose();
} else if (got_some && when_consumed_some) {
ctx->delay(when_consumed_some);
......
......@@ -81,8 +81,8 @@ public:
if (out_) {
for_each_input([](auto, auto& input) {
if (input.sub) {
input.sub.dispose();
input.sub = nullptr;
auto tmp = std::move(input.sub);
tmp.dispose();
}
input.buf.clear();
});
......
......@@ -176,12 +176,20 @@ public:
template <class C>
intrusive_ptr<C> downcast() const noexcept {
return (ptr_) ? dynamic_cast<C*>(get()) : nullptr;
static_assert(std::is_base_of_v<T, C>);
return intrusive_ptr<C>{ptr_ ? dynamic_cast<C*>(get()) : nullptr};
}
template <class C>
intrusive_ptr<C> upcast() const noexcept {
return (ptr_) ? static_cast<C*>(get()) : nullptr;
intrusive_ptr<C> upcast() const& noexcept {
static_assert(std::is_base_of_v<C, T>);
return intrusive_ptr<C>{ptr_ ? ptr_ : nullptr};
}
template <class C>
intrusive_ptr<C> upcast() && noexcept {
static_assert(std::is_base_of_v<C, T>);
return intrusive_ptr<C>{ptr_ ? release() : nullptr, false};
}
private:
......
......@@ -18,6 +18,12 @@
namespace caf {
/// Tag type for selecting case-insensitive algorithms.
struct ignore_case_t {};
/// Tag for selecting case-insensitive algorithms.
constexpr ignore_case_t ignore_case = ignore_case_t{};
// provide boost::split compatible interface
constexpr std::string_view is_any_of(std::string_view arg) noexcept {
......
......@@ -461,6 +461,11 @@ public:
visit_family(f, ptr.get());
}
// -- static utility functions -----------------------------------------------
/// Returns a pointer to the metric registry from the actor system.
static metric_registry* from(actor_system& sys);
// -- modifiers --------------------------------------------------------------
/// Takes ownership of all metric families in `other`.
......
......@@ -47,6 +47,9 @@ public:
// nop
}
/// Returns the `host` as string.
std::string host_str() const;
/// Returns whether `host` is empty, i.e., the host is not an IP address
/// and the string is empty.
bool empty() const noexcept {
......
......@@ -65,14 +65,11 @@ std::string_view trim(std::string_view str) {
}
bool icase_equal(std::string_view x, std::string_view y) {
if (x.size() != y.size()) {
return false;
} else {
for (size_t index = 0; index < x.size(); ++index)
if (tolower(x[index]) != tolower(y[index]))
return false;
return true;
}
auto cmp = [](const char lhs, const char rhs) {
auto to_uchar = [](char c) { return static_cast<unsigned char>(c); };
return tolower(to_uchar(lhs)) == tolower(to_uchar(rhs));
};
return std::equal(x.begin(), x.end(), y.begin(), y.end(), cmp);
}
std::pair<std::string_view, std::string_view> split_by(std::string_view str,
......
......@@ -4,6 +4,7 @@
#include "caf/telemetry/metric_registry.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/config.hpp"
#include "caf/raise_error.hpp"
......@@ -36,6 +37,10 @@ metric_registry::~metric_registry() {
// nop
}
metric_registry* metric_registry::from(actor_system& sys) {
return &sys.metrics();
}
void metric_registry::merge(metric_registry& other) {
if (this == &other)
return;
......
......@@ -27,6 +27,12 @@ caf::uri::impl_type default_instance;
namespace caf {
std::string uri::authority_type::host_str() const {
if (auto* str = std::get_if<std::string>(&host))
return *str;
return to_string(std::get<ip_address>(host));
}
uri::impl_type::impl_type() : rc_(1) {
// nop
}
......
......@@ -70,8 +70,9 @@ public:
void on_complete() override {
if (sub) {
sub.dispose();
sub = nullptr;
subscription tmp;
tmp.swap(sub);
tmp.dispose();
}
state = observer_state::completed;
}
......@@ -158,6 +159,43 @@ public:
std::vector<T> buf;
};
template <class T>
struct unsubscribe_guard {
public:
explicit unsubscribe_guard(intrusive_ptr<T> ptr) : ptr_(std::move(ptr)) {
// nop
}
unsubscribe_guard(unsubscribe_guard&&) = default;
unsubscribe_guard& operator=(unsubscribe_guard&&) = delete;
unsubscribe_guard(const unsubscribe_guard&) = delete;
unsubscribe_guard& operator=(const unsubscribe_guard&) = delete;
~unsubscribe_guard() {
if (ptr_)
ptr_->unsubscribe();
}
private:
intrusive_ptr<T> ptr_;
};
template <class T>
auto make_unsubscribe_guard(intrusive_ptr<T> ptr) {
return unsubscribe_guard<T>{std::move(ptr)};
}
template <class T1, class T2, class... Ts>
auto make_unsubscribe_guard(intrusive_ptr<T1> ptr1, intrusive_ptr<T2> ptr2,
Ts... ptrs) {
return std::make_tuple(make_unsubscribe_guard(std::move(ptr1)),
make_unsubscribe_guard(ptr2),
make_unsubscribe_guard(std::move(ptrs))...);
}
template <class T>
class canceling_observer : public flow::observer_impl_base<T> {
public:
......
......@@ -49,6 +49,10 @@ struct noskip_trait {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
~fixture() {
ctx->run();
}
// Similar to buffer::subscribe, but returns a buffer_sub pointer instead of
// type-erasing it into a disposable.
template <class Trait = noskip_trait>
......@@ -208,6 +212,7 @@ SCENARIO("buffers start to emit items once subscribed") {
WHEN("the selector never calls on_subscribe") {
THEN("the buffer still emits batches") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub(3, flow::make_nil_observable<int>(ctx.get()),
flow::make_nil_observable<int64_t>(ctx.get()),
snk->as_observer());
......@@ -251,6 +256,7 @@ SCENARIO("buffers dispose unexpected subscriptions") {
WHEN("calling on_subscribe with unexpected subscriptions") {
THEN("the buffer disposes them immediately") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub(3, flow::make_nil_observable<int>(ctx.get()),
flow::make_nil_observable<int64_t>(ctx.get()),
snk->as_observer());
......@@ -425,6 +431,7 @@ SCENARIO("skip policies suppress empty batches") {
WHEN("the control observable fires with no pending data") {
THEN("the operator omits the batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
......@@ -439,6 +446,7 @@ SCENARIO("skip policies suppress empty batches") {
WHEN("the control observable fires with pending data") {
THEN("the operator emits a partial batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
......@@ -459,6 +467,7 @@ SCENARIO("no-skip policies emit empty batches") {
WHEN("the control observable fires with no pending data") {
THEN("the operator emits an empty batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<noskip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
......@@ -473,6 +482,7 @@ SCENARIO("no-skip policies emit empty batches") {
WHEN("the control observable fires with pending data") {
THEN("the operator emits a partial batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<noskip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
......@@ -512,6 +522,7 @@ SCENARIO("on_request actions can turn into no-ops") {
WHEN("the sink requests more data right before a timeout triggers") {
THEN("the batch gets shipped and the on_request action does nothing") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
......
......@@ -115,6 +115,8 @@ SCENARIO("mcast operators buffer items that they cannot ship immediately") {
CHECK_EQ(uut->min_demand(), 0u);
CHECK_EQ(uut->max_buffered(), 3u);
CHECK_EQ(uut->min_buffered(), 1u);
sub2.dispose();
sub3.dispose();
}
}
}
......
......@@ -68,6 +68,7 @@ SCENARIO("ucast operators may only be subscribed to once") {
auto uut = make_ucast();
auto o1 = flow::make_passive_observer<int>();
auto o2 = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(o1, o2);
auto sub1 = uut->subscribe(o1->as_observer());
auto sub2 = uut->subscribe(o2->as_observer());
CHECK(o1->subscribed());
......
......@@ -37,6 +37,7 @@ SCENARIO("zip_with combines inputs") {
WHEN("merging them with zip_with") {
THEN("the observer receives the combined output of both sources") {
auto snk = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk);
ctx->make_observable()
.zip_with([](int x, int y) { return x + y; },
ctx->make_observable().repeat(11).take(113),
......@@ -200,6 +201,7 @@ SCENARIO("observers may request from zip_with operators before on_subscribe") {
THEN("the observer receives the item") {
using flow::op::zip_index;
auto snk = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk);
auto uut = make_zip_with_sub([](int, int) { return 0; },
snk->as_observer(),
flow::make_nil_observable<int>(ctx.get()),
......@@ -222,6 +224,7 @@ SCENARIO("the zip_with operators disposes unexpected subscriptions") {
THEN("the operator disposes the subscription") {
using flow::op::zip_index;
auto snk = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk);
auto uut = make_zip_with_sub([](int, int) { return 0; },
snk->as_observer(),
flow::make_nil_observable<int>(ctx.get()),
......
......@@ -21,6 +21,10 @@ namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
~fixture() {
ctx->run();
}
template <class... Ts>
std::vector<int> ls(Ts... xs) {
return std::vector<int>{xs...};
......@@ -143,13 +147,15 @@ SCENARIO("connectable observables forward errors") {
auto conn = flow::observable<int>{cell}.share();
cell->set_error(sec::runtime_error);
// First subscriber to trigger subscription to the cell.
conn.subscribe(flow::make_auto_observer<int>()->as_observer());
auto snk1 = flow::make_auto_observer<int>();
conn.subscribe(snk1->as_observer());
ctx->run();
CHECK(snk1->aborted());
// After this point, new subscribers should be aborted right away.
auto snk = flow::make_auto_observer<int>();
auto sub = conn.subscribe(snk->as_observer());
auto snk2 = flow::make_auto_observer<int>();
auto sub = conn.subscribe(snk2->as_observer());
CHECK(sub.disposed());
CHECK(snk->aborted());
CHECK(snk2->aborted());
ctx->run();
}
}
......@@ -163,6 +169,7 @@ SCENARIO("observers that dispose their subscription do not affect others") {
using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk1, snk2);
auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl(), 5);
auto sub1 = uut->subscribe(snk1->as_observer());
......@@ -191,6 +198,7 @@ SCENARIO("publishers with auto_disconnect auto-dispose their subscription") {
using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk1, snk2);
auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl(), 5);
auto sub1 = uut->subscribe(snk1->as_observer());
......@@ -218,10 +226,11 @@ SCENARIO("publishers dispose unexpected subscriptions") {
WHEN("calling on_subscribe with unexpected subscriptions") {
THEN("the operator disposes them immediately") {
using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>();
auto snk = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk);
auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl());
uut->subscribe(snk1->as_observer());
uut->subscribe(snk->as_observer());
uut->connect();
auto sub = flow::make_passive_subscription();
uut->on_subscribe(flow::subscription{sub});
......
......@@ -9,112 +9,116 @@ configure_file(test/pem.cpp.in test/pem.cpp @ONLY)
caf_add_component(
net
DEPENDENCIES
PUBLIC
$<$<CXX_COMPILER_ID:MSVC>:ws2_32>
CAF::core
OpenSSL::Crypto
OpenSSL::SSL
PRIVATE
CAF::internal
PUBLIC
$<$<CXX_COMPILER_ID:MSVC>:ws2_32>
CAF::core
OpenSSL::Crypto
OpenSSL::SSL
PRIVATE
CAF::internal
ENUM_TYPES
net.http.method
net.http.status
net.operation
net.ssl.dtls
net.ssl.errc
net.ssl.format
net.ssl.tls
net.stream_transport_error
net.web_socket.status
net.http.method
net.http.status
net.octet_stream.errc
net.ssl.dtls
net.ssl.errc
net.ssl.format
net.ssl.tls
net.web_socket.status
HEADERS
${CAF_NET_HEADERS}
${CAF_NET_HEADERS}
SOURCES
src/detail/convert_ip_endpoint.cpp
src/detail/rfc6455.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/binary/default_trait.cpp
src/net/binary/frame.cpp
src/net/binary/lower_layer.cpp
src/net/binary/upper_layer.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/lower_layer.cpp
src/net/http/serve.cpp
src/net/http/method.cpp
src/net/http/request.cpp
src/net/http/response.cpp
src/net/http/serve.cpp
src/net/http/server.cpp
src/net/http/status.cpp
src/net/http/upper_layer.cpp
src/net/http/v1.cpp
src/net/ip.cpp
src/net/length_prefix_framing.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/prometheus/server.cpp
src/net/socket.cpp
src/net/socket_event_layer.cpp
src/net/socket_manager.cpp
src/net/ssl/acceptor.cpp
src/net/ssl/connection.cpp
src/net/ssl/context.cpp
src/net/ssl/dtls.cpp
src/net/ssl/format.cpp
src/net/ssl/password.cpp
src/net/ssl/startup.cpp
src/net/ssl/tls.cpp
src/net/ssl/transport.cpp
src/net/ssl/verify.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/connect.cpp
src/net/web_socket/default_trait.cpp
src/net/web_socket/frame.cpp
src/net/web_socket/framing.cpp
src/net/web_socket/handshake.cpp
src/net/web_socket/lower_layer.cpp
src/net/web_socket/server.cpp
src/net/web_socket/upper_layer.cpp
src/detail/convert_ip_endpoint.cpp
src/detail/pollset_updater.cpp
src/detail/rfc6455.cpp
src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp
src/net/datagram_socket.cpp
src/net/dsl/config_base.cpp
src/net/generic_lower_layer.cpp
src/net/generic_upper_layer.cpp
src/net/http/config.cpp
src/net/http/lower_layer.cpp
src/net/http/method.cpp
src/net/http/request.cpp
src/net/http/request_header.cpp
src/net/http/responder.cpp
src/net/http/response.cpp
src/net/http/route.cpp
src/net/http/router.cpp
src/net/http/server.cpp
src/net/http/server_factory.cpp
src/net/http/status.cpp
src/net/http/upper_layer.cpp
src/net/http/v1.cpp
src/net/ip.cpp
src/net/lp/default_trait.cpp
src/net/lp/frame.cpp
src/net/lp/framing.cpp
src/net/lp/lower_layer.cpp
src/net/lp/upper_layer.cpp
src/net/middleman.cpp
src/net/multiplexer.cpp
src/net/network_socket.cpp
src/net/octet_stream/lower_layer.cpp
src/net/octet_stream/policy.cpp
src/net/octet_stream/transport.cpp
src/net/octet_stream/upper_layer.cpp
src/net/pipe_socket.cpp
src/net/prometheus.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/errc.cpp
src/net/ssl/format.cpp
src/net/ssl/password.cpp
src/net/ssl/startup.cpp
src/net/ssl/tcp_acceptor.cpp
src/net/ssl/tls.cpp
src/net/ssl/transport.cpp
src/net/ssl/verify.cpp
src/net/stream_socket.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/frame.cpp
src/net/web_socket/framing.cpp
src/net/web_socket/handshake.cpp
src/net/web_socket/lower_layer.cpp
src/net/web_socket/server.cpp
src/net/web_socket/upper_layer.cpp
TEST_SOURCES
${CMAKE_CURRENT_BINARY_DIR}/test/pem.cpp
test/net-test.cpp
${CMAKE_CURRENT_BINARY_DIR}/test/pem.cpp
test/net-test.cpp
TEST_SUITES
detail.convert_ip_endpoint
detail.rfc6455
net.accept_socket
net.actor_shell
net.binary.frame
net.datagram_socket
net.http.server
net.ip
net.length_prefix_framing
net.multiplexer
net.network_socket
net.operation
net.pipe_socket
net.prometheus.server
net.socket
net.socket_guard
net.ssl.transport
net.stream_socket
net.stream_transport
net.tcp_socket
net.typed_actor_shell
net.udp_datagram_socket
net.web_socket.client
net.web_socket.frame
net.web_socket.handshake
net.web_socket.server)
detail.convert_ip_endpoint
detail.rfc6455
net.accept_socket
net.actor_shell
net.datagram_socket
net.http.router
net.http.server
net.ip
net.length_prefix_framing
net.lp.frame
net.multiplexer
net.network_socket
net.octet_stream.transport
net.pipe_socket
net.socket
net.socket_guard
net.ssl.transport
net.stream_socket
net.tcp_socket
net.typed_actor_shell
net.udp_datagram_socket
net.web_socket.client
net.web_socket.frame
net.web_socket.handshake
net.web_socket.server)
This diff is collapsed.
......@@ -4,7 +4,9 @@
#pragma once
#include "caf/async/execution_context.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/settings.hpp"
......@@ -13,24 +15,29 @@ namespace caf::detail {
/// Accepts incoming clients with an Acceptor and handles them via a connection
/// factory.
template <class Acceptor, class ConnectionHandle>
template <class Acceptor>
class accept_handler : public net::socket_event_layer {
public:
// -- member types -----------------------------------------------------------
using socket_type = net::socket;
using connection_handle = ConnectionHandle;
using transport_type = typename Acceptor::transport_type;
using connection_handle = typename transport_type::connection_handle;
using factory_type = connection_factory<connection_handle>;
using factory_ptr = detail::connection_factory_ptr<connection_handle>;
// -- constructors, destructors, and assignment operators --------------------
template <class FactoryPtr, class... Ts>
accept_handler(Acceptor acc, FactoryPtr fptr, size_t max_connections)
accept_handler(Acceptor acc, factory_ptr fptr, size_t max_connections,
std::vector<strong_actor_ptr> monitored_actors = {})
: acc_(std::move(acc)),
factory_(std::move(fptr)),
max_connections_(max_connections) {
max_connections_(max_connections),
monitored_actors_(std::move(monitored_actors)) {
CAF_ASSERT(max_connections_ > 0);
}
......@@ -38,27 +45,40 @@ public:
on_conn_close_.dispose();
if (valid(acc_))
close(acc_);
if (monitor_callback_)
monitor_callback_.dispose();
}
// -- factories --------------------------------------------------------------
template <class FactoryPtr, class... Ts>
static std::unique_ptr<accept_handler>
make(Acceptor acc, FactoryPtr fptr, size_t max_connections) {
make(Acceptor acc, factory_ptr fptr, size_t max_connections,
std::vector<strong_actor_ptr> monitored_actors = {}) {
return std::make_unique<accept_handler>(std::move(acc), std::move(fptr),
max_connections);
max_connections,
std::move(monitored_actors));
}
// -- implementation of socket_event_layer -----------------------------------
error start(net::socket_manager* owner, const settings& cfg) override {
error start(net::socket_manager* owner) override {
CAF_LOG_TRACE("");
owner_ = owner;
cfg_ = cfg;
if (auto err = factory_->start(owner, cfg)) {
if (auto err = factory_->start(owner)) {
CAF_LOG_DEBUG("factory_->start failed:" << err);
return err;
}
if (!monitored_actors_.empty()) {
monitor_callback_ = make_action([this] { owner_->shutdown(); });
auto ctx = async::execution_context_ptr{owner_->mpx_ptr()};
for (auto& hdl : monitored_actors_) {
CAF_ASSERT(hdl);
hdl->get()->attach_functor([ctx, cb = monitor_callback_] {
if (!cb.disposed())
ctx->schedule(cb);
});
}
}
on_conn_close_ = make_action([this] { connection_closed(); });
owner->register_reading();
return none;
......@@ -84,7 +104,7 @@ public:
if (++open_connections_ == max_connections_)
owner_->deregister_reading();
child->add_cleanup_listener(on_conn_close_);
std::ignore = child->start(cfg_);
std::ignore = child->start();
} else if (conn.error() == sec::unavailable_or_would_block) {
// Encountered a "soft" error: simply try again later.
CAF_LOG_DEBUG("accept failed:" << conn.error());
......@@ -120,7 +140,7 @@ private:
Acceptor acc_;
detail::connection_factory_ptr<connection_handle> factory_;
factory_ptr factory_;
size_t max_connections_;
......@@ -130,12 +150,16 @@ private:
action on_conn_close_;
settings cfg_;
/// Type-erased handle to the @ref socket_manager. This reference is important
/// to keep the acceptor alive while the manager is not registered for writing
/// or reading.
disposable self_ref_;
/// An action for stopping this handler if an observed actor terminates.
action monitor_callback_;
/// List of actors that we add monitors to in `start`.
std::vector<strong_actor_ptr> monitored_actors_;
};
} // namespace caf::detail
......@@ -4,6 +4,7 @@
#pragma once
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
......@@ -21,7 +22,7 @@ public:
// nop
}
virtual error start(net::socket_manager*, const settings&) {
virtual error start(net::socket_manager*) {
return none;
}
......
......@@ -8,7 +8,8 @@
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/logger.hpp"
#include "caf/net/http/request_header.hpp"
#include "caf/sec.hpp"
#include <utility>
......@@ -29,17 +30,43 @@ public:
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using connector_pointer = net::flow_connector_ptr<Trait>;
flow_bridge_base(async::execution_context_ptr loop, connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
virtual bool write(const output_type& item) = 0;
bool running() const noexcept {
return in_ || out_;
return in_ && out_;
}
/// Initializes consumer and producer of the bridge.
error init(net::multiplexer* mpx, async::consumer_resource<output_type> pull,
async::producer_resource<input_type> push) {
// Initialize our consumer.
auto do_wakeup = make_action([this] {
if (running())
prepare_send();
});
in_ = consumer_type::make(pull.try_open(), mpx, std::move(do_wakeup));
if (!in_) {
auto err = make_error(sec::runtime_error,
"flow bridge failed to open the input resource");
push.abort(err);
return err;
}
// Initialize our producer.
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running()) {
down_->shutdown();
}
});
out_ = producer_type::make(push.try_open(), mpx, std::move(do_resume),
std::move(do_cancel));
if (!out_) {
auto err = make_error(sec::runtime_error,
"flow bridge failed to open the output resource");
in_.cancel();
return err;
}
return {};
}
void self_ref(disposable ref) {
......@@ -48,36 +75,6 @@ public:
// -- implementation of the lower_layer --------------------------------------
error start(LowerLayer* down, const settings& cfg) override {
CAF_ASSERT(down != nullptr);
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
auto do_wakeup = make_action([this] {
if (running())
prepare_send();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running()) {
down_->shutdown();
}
});
in_ = consumer_type::make(pull.try_open(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (running())
return none;
else
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
} else {
conn_ = nullptr;
return err;
}
}
void prepare_send() override {
input_type tmp;
while (down_->can_send_more()) {
......@@ -138,12 +135,6 @@ protected:
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Our event loop.
async::execution_context_ptr loop_;
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
/// Type-erased handle to the @ref socket_manager. This reference is important
/// to keep the bridge alive while the manager is not registered for writing
/// or reading.
......
......@@ -9,46 +9,38 @@
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_bridge_base.hpp"
#include "caf/fwd.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/net/lp/lower_layer.hpp"
#include "caf/net/lp/upper_layer.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace caf::net::binary {
namespace caf::detail {
/// Convenience alias for referring to the base type of @ref flow_bridge.
template <class Trait>
using flow_bridge_base_t
= detail::flow_bridge_base<upper_layer, lower_layer, Trait>;
using lp_flow_bridge_base
= flow_bridge_base<net::lp::upper_layer, net::lp::lower_layer, Trait>;
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public flow_bridge_base_t<Trait> {
class lp_flow_bridge : public lp_flow_bridge_base<Trait> {
public:
using super = flow_bridge_base_t<Trait>;
using super = lp_flow_bridge_base<Trait>;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using connector_pointer = flow_connector_ptr<Trait>;
using super::super;
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
}
bool write(const output_type& item) override {
super::down_->begin_message();
auto& bytes = super::down_->message_buffer();
return super::trait_.convert(item, bytes) && super::down_->end_message();
}
// -- implementation of binary::lower_layer ----------------------------------
// -- implementation of lp::lower_layer --------------------------------------
ptrdiff_t consume(byte_span buf) override {
if (!super::out_)
......@@ -62,4 +54,4 @@ public:
}
};
} // namespace caf::net::binary
} // namespace caf::detail
......@@ -7,8 +7,8 @@
#include "caf/actor_system.hpp"
#include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
#include "caf/net/lp/lower_layer.hpp"
#include "caf/net/lp/upper_layer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
......@@ -32,7 +32,7 @@ namespace caf::net {
/// };
/// ~~~
template <class Trait>
class message_flow_bridge : public binary::upper_layer {
class message_flow_bridge : public lp::upper_layer {
public:
/// The input type for the application.
using input_type = typename Trait::input_type;
......@@ -64,8 +64,7 @@ public:
// nop
}
error start(net::socket_manager* mgr, binary::lower_layer* down,
const settings&) {
error start(net::socket_manager* mgr, lp::lower_layer* down) {
down_ = down;
if (auto in = make_consumer_adapter(in_res_, mgr->mpx_ptr(),
do_wakeup_cb())) {
......@@ -159,7 +158,7 @@ private:
}
/// Points to the next layer down the protocol stack.
binary::lower_layer* down_ = nullptr;
lp::lower_layer* down_ = nullptr;
/// Incoming messages, serialized to the socket.
async::consumer_adapter<input_type> in_;
......
......@@ -14,13 +14,13 @@
#include <cstdint>
#include <mutex>
namespace caf::net {
namespace caf::detail {
class CAF_NET_EXPORT pollset_updater : public socket_event_layer {
class CAF_NET_EXPORT pollset_updater : public net::socket_event_layer {
public:
// -- member types -----------------------------------------------------------
using super = socket_manager;
using super = net::socket_manager;
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
......@@ -34,17 +34,17 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit pollset_updater(pipe_socket fd);
explicit pollset_updater(net::pipe_socket fd);
// -- factories --------------------------------------------------------------
static std::unique_ptr<pollset_updater> make(pipe_socket fd);
static std::unique_ptr<pollset_updater> make(net::pipe_socket fd);
// -- implementation of socket_event_layer -----------------------------------
error start(socket_manager* owner, const settings& cfg) override;
error start(net::socket_manager* owner) override;
socket handle() const override;
net::socket handle() const override;
void handle_read_event() override;
......@@ -53,11 +53,11 @@ public:
void abort(const error& reason) override;
private:
pipe_socket fd_;
socket_manager* owner_ = nullptr;
multiplexer* mpx_ = nullptr;
net::pipe_socket fd_;
net::socket_manager* owner_ = nullptr;
net::multiplexer* mpx_ = nullptr;
msg_buf buf_;
size_t buf_size_ = 0;
};
} // namespace caf::net
} // namespace caf::detail
......@@ -9,40 +9,31 @@
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_bridge_base.hpp"
#include "caf/fwd.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
#include "caf/net/web_socket/request.hpp"
#include "caf/net/web_socket/upper_layer.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace caf::net::web_socket {
namespace caf::detail {
/// Convenience alias for referring to the base type of @ref flow_bridge.
template <class Trait>
using flow_bridge_base_t
= detail::flow_bridge_base<upper_layer, lower_layer, Trait>;
template <class Trait, class Base>
using ws_flow_bridge_base_t
= detail::flow_bridge_base<Base, net::web_socket::lower_layer, Trait>;
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public flow_bridge_base_t<Trait> {
template <class Trait, class Base>
class ws_flow_bridge : public ws_flow_bridge_base_t<Trait, Base> {
public:
using super = flow_bridge_base_t<Trait>;
using super = ws_flow_bridge_base_t<Trait, Base>;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using connector_pointer = flow_connector_ptr<Trait>;
using super::super;
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
}
bool write(const output_type& item) override {
if (super::trait_.converts_to_binary(item)) {
super::down_->begin_binary_message();
......@@ -82,4 +73,4 @@ public:
}
};
} // namespace caf::net::web_socket
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/socket.hpp"
#include "caf/sec.hpp"
namespace caf::net {
/// Lifts `Socket` to an `expected<Socket>` and sets an error if `fd` is
/// invalid.
template <class Socket>
expected<Socket> checked_socket(Socket fd) {
using res_t = expected<Socket>;
if (fd.id != invalid_socket_id)
return res_t{fd};
else
return res_t{make_error(sec::runtime_error, "invalid socket handle")};
}
/// A function object that calls `checked_socket`.
static constexpr auto check_socket = [](auto fd) { return checked_socket(fd); };
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <deque>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
namespace caf::net {
template <class Factory>
using datagram_transport_base
= transport_base<datagram_transport<Factory>,
transport_worker_dispatcher<Factory, ip_endpoint>,
udp_datagram_socket, Factory, ip_endpoint>;
/// Implements a udp_transport policy that manages a datagram socket.
template <class Factory>
class datagram_transport : public datagram_transport_base<Factory> {
public:
// Maximal UDP-packet size
static constexpr size_t max_datagram_size
= std::numeric_limits<uint16_t>::max();
// -- member types -----------------------------------------------------------
using factory_type = Factory;
using id_type = ip_endpoint;
using application_type = typename factory_type::application_type;
using super = datagram_transport_base<factory_type>;
using buffer_cache_type = typename super::buffer_cache_type;
// -- constructors, destructors, and assignment operators --------------------
datagram_transport(udp_datagram_socket handle, factory_type factory)
: super(handle, std::move(factory)) {
// nop
}
// -- public member functions ------------------------------------------------
error start(endpoint_manager& manager) override {
CAF_LOG_TRACE("");
if (auto err = super::start(manager))
return err;
prepare_next_read();
return none;
}
bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) {
auto ret = read(this->handle_, this->read_buf_);
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto& [num_bytes, ep] = *res;
CAF_LOG_DEBUG("received " << num_bytes << " bytes");
this->read_buf_.resize(num_bytes);
if (auto err = this->next_layer_.handle_data(*this, this->read_buf_,
std::move(ep))) {
CAF_LOG_ERROR("handle_data failed: " << err);
return false;
}
prepare_next_read();
} else {
auto err = get<sec>(ret);
if (err == sec::unavailable_or_would_block) {
break;
} else {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
}
}
return true;
}
bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("socket", this->handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size()));
auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
return true;
}
return false;
};
do {
if (auto err = write_some())
return err == sec::unavailable_or_would_block;
} while (fetch_next_message());
return !packet_queue_.empty();
}
// TODO: remove this function. `resolve` should add workers when needed.
error add_new_worker(node_id node, id_type id) {
auto worker = this->next_layer_.add_new_worker(*this, node, id);
if (!worker)
return worker.error();
return none;
}
void write_packet(id_type id, span<byte_buffer*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty())
this->manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
packet_queue_.emplace_back(id, buffers);
}
/// Helper struct for managing outgoing packets
struct packet {
id_type id;
buffer_cache_type bytes;
size_t size;
packet(id_type id, span<byte_buffer*> bufs) : id(id) {
size = 0;
for (auto buf : bufs) {
size += buf->size();
bytes.emplace_back(std::move(*buf));
}
}
std::vector<byte_buffer*> get_buffer_ptrs() {
std::vector<byte_buffer*> ptrs;
for (auto& buf : bytes)
ptrs.emplace_back(&buf);
return ptrs;
}
};
private:
// -- utility functions ------------------------------------------------------
void prepare_next_read() {
this->read_buf_.resize(max_datagram_size);
}
error write_some() {
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = packet_queue_.front();
auto& bufs = front.bytes;
auto it = bufs.begin();
if (this->header_bufs_.size() < this->header_bufs_.capacity()) {
it->clear();
this->header_bufs_.emplace_back(std::move(*it++));
}
for (; it != bufs.end()
&& this->payload_bufs_.size() < this->payload_bufs_.capacity();
++it) {
it->clear();
this->payload_bufs_.emplace_back(std::move(*it));
}
packet_queue_.pop_front();
};
// Write as many bytes as possible.
while (!packet_queue_.empty()) {
auto& packet = packet_queue_.front();
auto ptrs = packet.get_buffer_ptrs();
auto write_ret = write(this->handle_, ptrs, packet.id);
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
CAF_LOG_WARNING_IF(*num_bytes < packet.size,
"packet was not sent completely");
recycle();
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_ERROR("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
}
return err;
}
}
return none;
}
std::deque<packet> packet_queue_;
};
} // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/expected.hpp"
#include <optional>
namespace caf::net::dsl::arg {
/// Represents a null-terminated string or `null`.
class cstring {
public:
cstring() : data_(nullptr) {
// nop
}
cstring(const char* str) : data_(str) {
// nop
}
cstring(std::string str) : data_(std::move(str)) {
// nop
}
cstring(std::optional<const char*> str) : cstring() {
if (str)
data_ = *str;
}
cstring(std::optional<std::string> str) : cstring() {
if (str)
data_ = std::move(*str);
}
cstring(caf::expected<const char*> str) : cstring() {
if (str)
data_ = *str;
}
cstring(caf::expected<std::string> str) : cstring() {
if (str)
data_ = std::move(*str);
}
cstring(cstring&&) = default;
cstring(const cstring&) = default;
cstring& operator=(cstring&&) = default;
cstring& operator=(const cstring&) = default;
/// @returns a pointer to the null-terminated string.
const char* get() const noexcept {
return std::visit(
[](auto& arg) -> const char* {
using T = std::decay_t<decltype(arg)>;
if constexpr (std::is_same_v<T, const char*>) {
return arg;
} else {
return arg.c_str();
}
},
data_);
}
bool has_value() const noexcept {
return !operator!();
}
explicit operator bool() const noexcept {
return has_value();
}
bool operator!() const noexcept {
return data_.index() == 0 && std::get<0>(data_) == nullptr;
}
private:
std::variant<const char*, std::string> data_;
};
/// Represents a value of type T or `null`.
template <class T>
class val {
public:
val() = default;
val(T value) : data_(std::move(value)) {
// nop
}
val(std::optional<T> value) : data_(std::move(value)) {
// nop
}
val(caf::expected<T> value) {
if (value)
data_ = std::move(*value);
}
val(val&&) = default;
val(const val&) = default;
val& operator=(val&&) = default;
val& operator=(const val&) = default;
const T& get() const noexcept {
return *data_;
}
explicit operator bool() const noexcept {
return data_.has_value();
}
bool operator!() const noexcept {
return !data_;
}
private:
std::optional<T> data_;
};
} // namespace caf::net::dsl::arg
......@@ -4,10 +4,9 @@
#pragma once
namespace caf::net::binary {
namespace caf::net::dsl {
class frame;
class lower_layer;
class upper_layer;
/// Base type for our DSL classes to configure a factory object.
struct base {};
} // namespace caf::net::binary
} // namespace caf::net::dsl
// 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/callback.hpp"
#include "caf/defaults.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/net/dsl/base.hpp"
#include "caf/net/dsl/config_base.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/uri.hpp"
#include <cassert>
#include <cstdint>
#include <string>
#include <utility>
namespace caf::net::dsl {
/// Meta programming utility.
template <class T>
struct client_config_tag {
using type = T;
};
/// Simple type for storing host and port information for reaching a server.
struct server_address {
/// The host name or IP address of the host.
std::string host;
/// The port to connect to.
uint16_t port;
};
/// Wraps configuration parameters for starting clients.
class client_config {
public:
/// Configuration for a client that creates the socket on demand.
class CAF_NET_EXPORT lazy : public has_ctx {
public:
/// Type for holding a client address.
using server_t = std::variant<server_address, uri>;
static constexpr std::string_view name = "lazy";
lazy(std::string host, uint16_t port)
: server(server_address{std::move(host), port}) {
// nop
}
explicit lazy(const uri& addr) : server(addr) {
// nop
}
/// The address for reaching the server or an error.
server_t server;
/// The delay between connection attempts.
timespan retry_delay = std::chrono::seconds{1};
/// The timeout when trying to connect.
timespan connection_timeout = infinite;
/// The maximum amount of retries.
size_t max_retry_count = 0;
};
using lazy_t = client_config_tag<lazy>;
static constexpr auto lazy_v = lazy_t{};
/// Configuration for a client that uses a user-provided socket.
class CAF_NET_EXPORT socket : public has_ctx {
public:
static constexpr std::string_view name = "socket";
explicit socket(stream_socket fd) : fd(fd) {
// nop
}
socket() = delete;
socket(const socket&) = delete;
socket& operator=(const socket&) = delete;
socket(socket&& other) noexcept : fd(other.fd) {
other.fd.id = invalid_socket_id;
}
socket& operator=(socket&& other) noexcept {
using std::swap;
swap(fd, other.fd);
return *this;
}
~socket() {
if (fd != invalid_socket)
close(fd);
}
/// The socket file descriptor to use.
stream_socket fd;
/// Returns the file descriptor and setting the `fd` member variable to the
/// invalid socket.
stream_socket take_fd() noexcept {
auto result = fd;
fd.id = invalid_socket_id;
return result;
}
};
using socket_t = client_config_tag<socket>;
static constexpr auto socket_v = socket_t{};
/// Configuration for a client that uses an already established SSL
/// connection.
class CAF_NET_EXPORT conn {
public:
static constexpr std::string_view name = "conn";
explicit conn(ssl::connection st) : state(std::move(st)) {
// nop
}
conn() = delete;
conn(const conn&) = delete;
conn& operator=(const conn&) = delete;
conn(conn&&) noexcept = default;
conn& operator=(conn&&) noexcept = default;
~conn() {
if (state) {
if (auto fd = state.fd(); fd != invalid_socket)
close(fd);
}
}
/// SSL state for the connection.
ssl::connection state;
};
using conn_t = client_config_tag<conn>;
static constexpr auto conn_v = conn_t{};
using fail_t = client_config_tag<error>;
static constexpr auto fail_v = fail_t{};
using value = config_impl<lazy, socket, conn>;
};
using client_config_value = client_config::value;
} // namespace caf::net::dsl
// 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/make_counted.hpp"
#include "caf/net/dsl/base.hpp"
#include "caf/net/dsl/client_config.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/ssl/tcp_acceptor.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include <cstdint>
#include <string>
namespace caf::net::dsl {
/// Base type for client factories for use with `can_connect`.
template <class Config, class Derived>
class client_factory_base {
public:
using config_type = Config;
using trait_type = typename config_type::trait_type;
using config_pointer = intrusive_ptr<config_type>;
client_factory_base(const client_factory_base&) = default;
client_factory_base& operator=(const client_factory_base&) = default;
template <class T, class... Ts>
explicit client_factory_base(dsl::client_config_tag<T> token, Ts&&... xs) {
cfg_ = config_type::make(token, std::forward<Ts>(xs)...);
}
/// Sets the callback for errors.
template <class F>
Derived& do_on_error(F callback) {
static_assert(std::is_invocable_v<F, const error&>);
cfg_->on_error = make_shared_type_erased_callback(std::move(callback));
return dref();
}
/// Sets the retry delay for connection attempts.
///
/// @param value The new retry delay.
/// @returns a reference to this `client_factory`.
Derived& retry_delay(timespan value) {
if (auto* lazy = get_if<client_config::lazy>(&cfg_->data))
lazy->retry_delay = value;
return dref();
}
/// Sets the connection timeout for connection attempts.
///
/// @param value The new connection timeout.
/// @returns a reference to this `client_factory`.
Derived& connection_timeout(timespan value) {
if (auto* lazy = get_if<client_config::lazy>(&cfg_->data))
lazy->connection_timeout = value;
return dref();
}
/// Sets the maximum number of connection retry attempts.
///
/// @param value The new maximum retry count.
/// @returns a reference to this `client_factory`.
Derived& max_retry_count(size_t value) {
if (auto* lazy = get_if<client_config::lazy>(&cfg_->data))
lazy->max_retry_count = value;
return dref();
}
config_type& config() {
return *cfg_;
}
private:
Derived& dref() {
return static_cast<Derived&>(*this);
}
config_pointer cfg_;
};
} // namespace caf::net::dsl
// 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/callback.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/net/checked_socket.hpp"
#include "caf/net/dsl/base.hpp"
#include "caf/net/dsl/get_name.hpp"
#include "caf/net/dsl/has_ctx.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/ref_counted.hpp"
#include "caf/sec.hpp"
#include "caf/uri.hpp"
#include <cassert>
#include <cstdint>
#include <string>
#include <utility>
namespace caf::net::dsl {
/// Base class for configuration objects.
class CAF_NET_EXPORT config_base : public ref_counted {
public:
explicit config_base(multiplexer* mpx) : mpx(mpx) {
// nop
}
config_base(const config_base&) = default;
config_base& operator=(const config_base&) = default;
virtual ~config_base();
virtual std::string_view name() const noexcept = 0;
virtual void fail(error err) noexcept = 0;
virtual error fail_reason() const = 0;
/// Convenience function for setting a default error if `as_has_ctx` returns
/// `nullptr` while trying to set an SSL context.
error cannot_add_ctx() {
return make_error(sec::logic_error,
"cannot add an SSL context to a config of type "
+ std::string{name()});
}
/// Inspects the data of this configuration and returns a pointer to it as
/// `has_ctx` instance if possible, `nullptr` otherwise.
virtual has_ctx* as_has_ctx() noexcept = 0;
/// Inspects the data of this configuration and returns a pointer to it as
/// `has_ctx` instance if possible, `nullptr` otherwise.
virtual const has_ctx* as_has_ctx() const noexcept = 0;
bool failed() const noexcept {
return name() == get_name<error>::value;
}
explicit operator bool() const noexcept {
return !failed();
}
bool operator!() const noexcept {
return failed();
}
/// The pointer to the parent @ref multiplexer.
multiplexer* mpx;
/// User-defined callback for errors.
shared_callback_ptr<void(const error&)> on_error;
/// Calls `on_error` if non-null.
void call_on_error(const error& what) {
if (on_error)
(*on_error)(what);
}
};
template <class... Data>
class config_impl : public config_base {
public:
using super = config_base;
template <class... Args>
explicit config_impl(const config_base& from, Args&&... args)
: super(from), data(std::forward<Args>(args)...) {
// nop
}
template <class... Args>
explicit config_impl(multiplexer* mpx, Args&&... args)
: super(mpx), data(std::forward<Args>(args)...) {
// nop
}
std::variant<error, Data...> data;
template <class F>
auto visit(F&& f) {
return std::visit([&](auto& arg) { return f(arg); }, data);
}
/// Returns the name of the configuration type.
std::string_view name() const noexcept override {
auto f = [](const auto& val) {
using val_t = std::decay_t<decltype(val)>;
return get_name<val_t>::value;
};
return std::visit(f, data);
}
void fail(error err) noexcept override {
if (!std::holds_alternative<error>(data))
data = std::move(err);
}
error fail_reason() const override {
if (auto* val = std::get_if<error>(&data))
return *val;
else
return {};
}
has_ctx* as_has_ctx() noexcept override {
return has_ctx::from(data);
}
const has_ctx* as_has_ctx() const noexcept override {
return has_ctx::from(data);
}
protected:
template <class Derived, class From, class Token, class... Args>
static intrusive_ptr<Derived> make_impl(std::in_place_type_t<Derived>,
const From& from, Token token,
Args&&... args) {
// Always construct the data in-place first. This makes sure we account
// for ownership transfers (e.g. for sockets).
auto ptr = make_counted<Derived>(from, token, std::forward<Args>(args)...);
// Discard the data if `from` contained an error. Otherwise, transfer the
// SSL context over to the refined configuration.
if (!from) {
ptr->data.template emplace<error>(std::get<error>(from.data));
} else if (auto* dst = ptr->as_has_ctx()) {
if (const auto* src = from.as_has_ctx()) {
dst->ctx = src->ctx;
} else {
ptr->data = make_error(caf::sec::logic_error,
"failed to transfer SSL context");
}
}
return ptr;
}
};
} // namespace caf::net::dsl
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.
......@@ -6,7 +6,6 @@
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/fwd.hpp"
#include <cstdint>
#include <string>
......
This diff is collapsed.
This diff is collapsed.
......@@ -8,7 +8,7 @@
#include "caf/byte_buffer.hpp"
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/http/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/span.hpp"
#include "caf/unordered_flat_map.hpp"
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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