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

Merge pull request #18

 Refactor scribe to stream-transport
parents aca857ad 1fd33248
......@@ -19,7 +19,6 @@ set(LIBCAF_NET_SRCS
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
src/tcp_stream_socket.cpp
src/tcp_accept_socket.cpp
)
......
......@@ -23,6 +23,7 @@
#include <memory>
#include "caf/actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/byte.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
......@@ -137,6 +138,9 @@ public:
/// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<byte> payload);
/// Enqueues a timeout to the endpoint.
void enqueue(timeout_msg msg);
// -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop.
......@@ -154,6 +158,9 @@ protected:
/// Stores outbound messages.
message_queue_type messages_;
/// Stores a proxy for interacting with the actor clock.
actor timeout_proxy_;
};
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
......
......@@ -23,7 +23,7 @@
namespace caf {
namespace net {
template <class Transport, class Application>
template <class Transport>
class endpoint_manager_impl : public endpoint_manager {
public:
// -- member types -----------------------------------------------------------
......@@ -32,15 +32,13 @@ public:
using transport_type = Transport;
using application_type = Application;
using application_type = typename transport_type::application_type;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
Transport trans, Application app)
: super(trans.handle(), parent, sys),
transport_(std::move(trans)),
application_(std::move(app)) {
Transport trans)
: super(trans.handle(), parent, sys), transport_(std::move(trans)) {
// nop
}
......@@ -54,8 +52,16 @@ public:
return transport_;
}
application_type& application() {
return application_;
// -- timeout management -----------------------------------------------------
template <class... Ts>
uint64_t set_timeout(actor_clock::time_point tp, atom_value type,
Ts&&... xs) {
auto act = actor_cast<abstract_actor*>(timeout_proxy_);
CAF_ASSERT(act != nullptr);
sys_.clock().set_multi_timeout(tp, act, type, next_timeout_id_);
transport_.set_timeout(next_timeout_id_, std::forward<Ts>(xs)...);
return next_timeout_id_++;
}
// -- interface functions ----------------------------------------------------
......@@ -91,7 +97,7 @@ public:
}
void handle_error(sec code) override {
transport_.handle_error(application_, code);
transport_.handle_error(code);
}
serialize_fun_type serialize_fun() const noexcept override {
......@@ -100,7 +106,9 @@ public:
private:
transport_type transport_;
application_type application_;
/// Stores the id for the next timeout.
uint64_t next_timeout_id_;
};
} // namespace net
......
......@@ -27,6 +27,8 @@ namespace net {
class multiplexer;
class socket_manager;
template <class Application, class IdType = unit_t>
class transport_worker;
struct network_socket;
struct pipe_socket;
......
......@@ -25,12 +25,11 @@
namespace caf {
namespace net {
template <class Transport, class Application>
template <class Transport>
endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx,
actor_system& sys, Transport trans,
Application app) {
using impl = endpoint_manager_impl<Transport, Application>;
return make_counted<impl>(mpx, sys, std::move(trans), std::move(app));
actor_system& sys, Transport trans) {
using impl = endpoint_manager_impl<Transport>;
return make_counted<impl>(mpx, sys, std::move(trans));
}
} // namespace net
......
......@@ -25,26 +25,49 @@
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace policy {
namespace net {
/// Implements a scribe policy that manages a stream socket.
class scribe {
/// Implements a stream_transport that manages a stream socket.
template <class Application>
class stream_transport {
public:
explicit scribe(net::stream_socket handle);
// -- member types -----------------------------------------------------------
using application_type = Application;
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application)
: worker_(std::move(application)),
handle_(handle),
// max_consecutive_reads_(0),
read_threshold_(1024),
collected_(0),
max_(1024),
rd_flag_(net::receive_policy_flag::exactly),
written_(0) {
// nop
}
// -- properties -------------------------------------------------------------
net::stream_socket handle() const noexcept {
stream_socket handle() const noexcept {
return handle_;
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent& parent) {
parent.application().init(parent);
parent.mask_add(net::operation::read);
if (auto err = worker_.init(parent))
return err;
parent.mask_add(operation::read);
return none;
}
......@@ -59,14 +82,15 @@ public:
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes;
if (collected_ >= read_threshold_) {
parent.application().handle_data(*this, read_buf_);
auto decorator = make_write_packet_decorator(*this, parent);
worker_.handle_data(decorator, read_buf_);
prepare_next_read();
}
return true;
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
parent.application().handle_error(err);
worker_.handle_error(err);
return false;
}
}
......@@ -74,25 +98,88 @@ public:
template <class Parent>
bool handle_write_event(Parent& parent) {
// Try to write leftover data.
write_some(parent);
write_some();
// Get new data from parent.
// TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
parent.application().write_message(*this, std::move(msg));
auto decorator = make_write_packet_decorator(*this, parent);
worker_.write_message(decorator, std::move(msg));
}
// Write prepared data.
return write_some(parent);
return write_some();
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
worker_.resolve(parent, path, listener);
}
template <class... Ts>
void set_timeout(uint64_t, Ts&&...) {
// nop
}
template <class Parent>
bool write_some(Parent& parent) {
void timeout(Parent& parent, atom_value value, uint64_t id) {
worker_.timeout(parent, value, id);
}
void handle_error(sec code) {
worker_.handle_error(code);
}
void prepare_next_read() {
read_buf_.clear();
collected_ = 0;
// This cast does nothing, but prevents a weird compiler error on GCC
// <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch (static_cast<net::receive_policy_flag>(rd_flag_)) {
case net::receive_policy_flag::exactly:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = max_;
break;
case net::receive_policy_flag::at_most:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = 1;
break;
case net::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (read_buf_.size() != max_size)
read_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
}
void configure_read(receive_policy::config cfg) {
rd_flag_ = cfg.first;
max_ = cfg.second;
prepare_next_read();
}
template <class Parent>
void write_packet(Parent&, span<const byte> header, span<const byte> payload,
unit_t) {
write_buf_.insert(write_buf_.end(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
}
private:
// -- private member functions -----------------------------------------------
bool write_some() {
if (write_buf_.empty())
return false;
auto len = write_buf_.size() - written_;
auto buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = net::write(handle_, make_span(buf, len));
auto ret = write(handle_, make_span(buf, len));
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
// Update state.
......@@ -105,47 +192,27 @@ public:
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
parent.application().handle_error(err);
worker_.handle_error(err);
return false;
}
return true;
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
parent.application().resolve(parent, path, listener);
}
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
parent.application().timeout(*this, value, id);
}
template <class Application>
void handle_error(Application& application, sec code) {
application.handle_error(code);
}
void prepare_next_read();
void configure_read(net::receive_policy::config cfg);
void write_packet(span<const byte> buf);
private:
net::stream_socket handle_;
transport_worker<application_type> worker_;
stream_socket handle_;
std::vector<byte> read_buf_;
std::vector<byte> write_buf_;
size_t max_consecutive_reads_; // TODO use this field!
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
size_t read_threshold_;
size_t collected_;
size_t max_;
net::receive_policy_flag rd_flag_;
receive_policy_flag rd_flag_;
size_t written_;
};
} // namespace policy
} // namespace net
} // namespace caf
......@@ -16,64 +16,83 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/policy/scribe.hpp"
#include <system_error>
#pragma once
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace caf {
namespace policy {
scribe::scribe(caf::net::stream_socket handle)
: handle_(handle),
max_consecutive_reads_(0),
read_threshold_(1024),
collected_(0),
max_(1024),
rd_flag_(net::receive_policy_flag::exactly),
written_(0) {
// nop
}
void scribe::prepare_next_read() {
read_buf_.clear();
collected_ = 0;
// This cast does nothing, but prevents a weird compiler error on GCC <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch (static_cast<net::receive_policy_flag>(rd_flag_)) {
case net::receive_policy_flag::exactly:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = max_;
break;
case net::receive_policy_flag::at_most:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = 1;
break;
case net::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (read_buf_.size() != max_size)
read_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
namespace net {
/// Implements a worker for transport protocols.
template <class Application, class IdType>
class transport_worker {
public:
// -- member types -----------------------------------------------------------
using id_type = IdType;
using application_type = Application;
// -- constructors, destructors, and assignment operators --------------------
transport_worker(application_type application, id_type id = id_type{})
: application_(std::move(application)), id_(std::move(id)) {
// nop
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent& parent) {
return application_.init(parent);
}
template <class Parent>
void handle_data(Parent& parent, span<const byte> data) {
application_.handle_data(parent, data);
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.write_message(decorator, std::move(msg));
}
template <class Parent>
void write_packet(Parent& parent, span<const byte> header,
span<const byte> payload) {
parent.write_packet(header, payload, id_);
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
application_.resolve(parent, path, listener);
}
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
application_.timeout(parent, value, id);
}
void handle_error(sec error) {
application_.handle_error(error);
}
}
void scribe::configure_read(net::receive_policy::config cfg) {
rd_flag_ = cfg.first;
max_ = cfg.second;
prepare_next_read();
}
private:
application_type application_;
id_type id_;
};
void scribe::write_packet(span<const byte> buf) {
write_buf_.insert(write_buf_.end(), buf.begin(), buf.end());
}
template <class Application, class IdType = unit_t>
using transport_worker_ptr = std::shared_ptr<
transport_worker<Application, IdType>>;
} // namespace policy
} // namespace net
} // namespace caf
......@@ -18,6 +18,9 @@
#pragma once
#include "caf/byte.hpp"
#include "caf/span.hpp"
namespace caf {
namespace net {
......@@ -26,34 +29,44 @@ namespace net {
template <class Object, class Parent>
class write_packet_decorator {
public:
dispatcher(Object& object, Parent& parent)
// -- member types -----------------------------------------------------------
using transport_type = typename Parent::transport_type;
using application_type = typename Parent::application_type;
// -- constructors, destructors, and assignment operators --------------------
write_packet_decorator(Object& object, Parent& parent)
: object_(object), parent_(parent) {
// nop
}
template <class Header>
void write_packet(const Header& header, span<const byte> payload) {
object_.write_packet(parent_, header, payload);
}
// -- properties -------------------------------------------------------------
actor_system& system() {
parent_.system();
return parent_.system();
}
void cancel_timeout(atom_value type, uint64_t id) {
parent_.cancel_timeout(type, id);
transport_type& transport() {
return parent_.transport();
}
void set_timeout(timestamp tout, atom_value type, uint64_t id) {
parent_.set_timeout(tout, type, id);
// -- member functions -------------------------------------------------------
template <class... Ts>
void write_packet(span<const byte> header, span<const byte> payload,
Ts&&... xs) {
object_.write_packet(parent_, header, payload, std::forward<Ts>(xs)...);
}
typename parent::transport_type& transport() {
return parent_.transport_;
void cancel_timeout(atom_value type, uint64_t id) {
parent_.cancel_timeout(type, id);
}
typename parent::application_type& application() {
return parent_.application_;
template <class... Ts>
uint64_t set_timeout(timestamp tout, atom_value type, Ts&&... xs) {
return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...);
}
private:
......
......@@ -72,6 +72,8 @@ public:
class dummy_transport {
public:
using application_type = dummy_application;
dummy_transport(stream_socket handle, std::shared_ptr<std::vector<byte>> data)
: handle_(handle), data_(data), read_buf_(1024) {
// nop
......@@ -84,7 +86,7 @@ public:
template <class Manager>
error init(Manager& manager) {
auto test_bytes = as_bytes(make_span(hello_test));
write_buf_.insert(write_buf_.end(), test_bytes.begin(), test_bytes.end());
buf_.insert(buf_.end(), test_bytes.begin(), test_bytes.end());
CAF_CHECK(manager.mask_add(operation::read_write));
return none;
}
......@@ -104,18 +106,17 @@ public:
bool handle_write_event(Manager& mgr) {
for (auto x = mgr.next_message(); x != nullptr; x = mgr.next_message()) {
auto& payload = x->payload;
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
buf_.insert(buf_.end(), payload.begin(), payload.end());
}
auto res = write(handle_, make_span(write_buf_));
auto res = write(handle_, make_span(buf_));
if (auto num_bytes = get_if<size_t>(&res)) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + *num_bytes);
return write_buf_.size() > 0;
buf_.erase(buf_.begin(), buf_.begin() + *num_bytes);
return buf_.size() > 0;
}
return get<sec>(res) == sec::unavailable_or_would_block;
}
template <class Manager>
void handle_error(Manager&, sec) {
void handle_error(sec) {
// nop
}
......@@ -143,7 +144,7 @@ private:
std::vector<byte> read_buf_;
std::vector<byte> write_buf_;
std::vector<byte> buf_;
};
} // namespace
......@@ -160,8 +161,7 @@ CAF_TEST(send and receive) {
sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf},
dummy_application{});
dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
......@@ -184,8 +184,7 @@ CAF_TEST(resolve and proxy communication) {
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf},
dummy_application{});
dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
......
......@@ -37,6 +37,8 @@ struct dummy_socket {
// nop
}
dummy_socket(const dummy_socket&) = default;
dummy_socket& operator=(const dummy_socket& other) {
id = other.id;
closed = other.closed;
......
......@@ -16,9 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE scribe_policy
#define CAF_SUITE stream_transport
#include "caf/policy/scribe.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/test/dsl.hpp"
......@@ -75,7 +75,7 @@ public:
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(msg->payload);
transport.write_packet(span<byte>{}, msg->payload);
}
template <class Parent>
......@@ -123,24 +123,25 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
using transport_type = stream_transport<dummy_application>;
std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
if (auto err = nonblocking(sockets.second, true))
CAF_FAIL("nonblocking() returned an error: " << err);
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
CAF_MESSAGE("configure scribe_policy");
policy::scribe scribe{sockets.first};
scribe.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, scribe, dummy_application{buf});
transport_type transport{sockets.first, dummy_application{buf}};
transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, transport);
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_MESSAGE("sending data to scribe_policy");
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
hello_manager.size());
CAF_MESSAGE("wrote " << hello_manager.size() << " bytes.");
run();
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()),
buf->size()),
......@@ -148,13 +149,15 @@ CAF_TEST(receive) {
}
CAF_TEST(resolve and proxy communication) {
using transport_type = stream_transport<dummy_application>;
std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first},
dummy_application{buf});
auto mgr = make_endpoint_manager(mpx, sys,
transport_type{sockets.first,
dummy_application{buf}});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
......
......@@ -18,7 +18,7 @@
#define CAF_SUITE string_application
#include "caf/policy/scribe.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/test/dsl.hpp"
......@@ -100,12 +100,9 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
std::vector<byte> buf;
header_type header{static_cast<uint32_t>(msg->payload.size())};
buf.resize(sizeof(header_type));
memcpy(buf.data(), &header, buf.size());
buf.insert(buf.end(), msg->payload.begin(), msg->payload.end());
parent.write_packet(buf);
std::vector<byte> payload(msg->payload.begin(), msg->payload.end());
parent.write_packet(as_bytes(make_span(&header, 1)), make_span(payload));
}
static expected<std::vector<byte>> serialize(actor_system& sys,
......@@ -152,7 +149,8 @@ public:
if (header_.payload == 0)
Base::handle_packet(parent, header_, span<const byte>{});
else
parent.configure_read(net::receive_policy::exactly(header_.payload));
parent.transport().configure_read(
net::receive_policy::exactly(header_.payload));
await_payload_ = true;
}
}
......@@ -191,6 +189,7 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
using application_type = extend<string_application>::with<
stream_string_application>;
using transport_type = stream_transport<application_type>;
std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>();
......@@ -199,13 +198,15 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first},
application_type{sys, buf});
auto mgr1 = make_endpoint_manager(mpx, sys,
transport_type{sockets.first,
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr1->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto mgr2 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.second},
application_type{sys, buf});
auto mgr2 = make_endpoint_manager(mpx, sys,
transport_type{sockets.second,
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr2->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE transport_worker
#include "caf/net/transport_worker.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
namespace {
constexpr string_view hello_test = "hello test!";
struct application_result {
bool initialized;
std::vector<byte> data_buffer;
std::string resolve_path;
actor resolve_listener;
atom_value timeout_value;
uint64_t timeout_id;
sec err;
};
struct transport_result {
std::vector<byte> packet_buffer;
ip_endpoint ep;
};
class dummy_application {
public:
dummy_application(std::shared_ptr<application_result> res)
: res_(std::move(res)){
// nop
};
~dummy_application() = default;
template <class Parent>
error init(Parent&) {
res_->initialized = true;
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(span<byte>{}, msg->payload);
}
template <class Parent>
void handle_data(Parent&, span<const byte> data) {
auto& buf = res_->data_buffer;
buf.clear();
buf.insert(buf.begin(), data.begin(), data.end());
}
template <class Parent>
void resolve(Parent&, const std::string& path, actor listener) {
res_->resolve_path = path;
res_->resolve_listener = std::move(listener);
}
template <class Parent>
void timeout(Parent&, atom_value value, uint64_t id) {
res_->timeout_value = value;
res_->timeout_id = id;
}
void handle_error(sec err) {
res_->err = err;
}
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
private:
std::shared_ptr<application_result> res_;
};
class dummy_transport {
public:
using transport_type = dummy_transport;
using application_type = dummy_application;
dummy_transport(std::shared_ptr<transport_result> res) : res_(res) {
// nop
}
void write_packet(span<const byte> header, span<const byte> payload,
ip_endpoint ep) {
auto& buf = res_->packet_buffer;
buf.insert(buf.begin(), header.begin(), header.end());
buf.insert(buf.begin(), payload.begin(), payload.end());
res_->ep = ep;
}
private:
std::shared_ptr<transport_result> res_;
};
struct fixture : test_coordinator_fixture<>, host_fixture {
using worker_type = transport_worker<dummy_application, ip_endpoint>;
fixture()
: transport_results{std::make_shared<transport_result>()},
application_results{std::make_shared<application_result>()},
transport(transport_results),
worker{dummy_application{application_results}} {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
if (auto err = parse("[::1]:12345", ep))
CAF_FAIL("parse returned an error: " << err);
worker = worker_type{dummy_application{application_results}, ep};
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
multiplexer_ptr mpx;
std::shared_ptr<transport_result> transport_results;
std::shared_ptr<application_result> application_results;
dummy_transport transport;
worker_type worker;
ip_endpoint ep;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(construction and initialization) {
CAF_CHECK_EQUAL(worker.init(transport), none);
CAF_CHECK_EQUAL(application_results->initialized, true);
}
CAF_TEST(handle_data) {
auto test_span = as_bytes(make_span(hello_test));
worker.handle_data(transport, test_span);
auto& buf = application_results->data_buffer;
string_view result{reinterpret_cast<char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(result, hello_test);
}
CAF_TEST(write_message) {
actor act;
auto strong_actor = actor_cast<strong_actor_ptr>(act);
mailbox_element::forwarding_stack stack;
auto msg = make_message();
auto elem = make_mailbox_element(strong_actor, make_message_id(12345), stack,
msg);
auto test_span = make_span(reinterpret_cast<byte*>(
const_cast<char*>(hello_test.data())),
hello_test.size());
std::vector<byte> payload(test_span.begin(), test_span.end());
auto message = detail::make_unique<endpoint_manager::message>(std::move(elem),
payload);
worker.write_message(transport, std::move(message));
auto& buf = transport_results->packet_buffer;
string_view result{reinterpret_cast<char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(result, hello_test);
CAF_CHECK_EQUAL(transport_results->ep, ep);
}
CAF_TEST(resolve) {
worker.resolve(transport, "foo", self);
CAF_CHECK_EQUAL(application_results->resolve_path, "foo");
CAF_CHECK_EQUAL(application_results->resolve_listener, self);
}
CAF_TEST(timeout) {
worker.timeout(transport, atom("bar"), 42u);
CAF_CHECK_EQUAL(application_results->timeout_value, atom("bar"));
CAF_CHECK_EQUAL(application_results->timeout_id, 42u);
}
CAF_TEST(handle_error) {
worker.handle_error(sec::feature_disabled);
CAF_CHECK_EQUAL(application_results->err, sec::feature_disabled);
}
CAF_TEST_FIXTURE_SCOPE_END()
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