Commit 342f745a authored by Jakob Otto's avatar Jakob Otto

Merge branch master

parents 5b479631 b759768a
...@@ -19,7 +19,6 @@ set(LIBCAF_NET_SRCS ...@@ -19,7 +19,6 @@ set(LIBCAF_NET_SRCS
src/socket.cpp src/socket.cpp
src/socket_manager.cpp src/socket_manager.cpp
src/stream_socket.cpp src/stream_socket.cpp
src/scribe.cpp
src/convert_ip_endpoint.cpp src/convert_ip_endpoint.cpp
src/udp_datagram_socket.cpp src/udp_datagram_socket.cpp
src/tcp_stream_socket.cpp src/tcp_stream_socket.cpp
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <memory> #include <memory>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp" #include "caf/intrusive/drr_queue.hpp"
...@@ -137,6 +138,9 @@ public: ...@@ -137,6 +138,9 @@ public:
/// Enqueues a message to the endpoint. /// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<byte> payload); 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 ------------------------------------------ // -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop. /// Initializes the manager before adding it to the multiplexer's event loop.
...@@ -154,6 +158,9 @@ protected: ...@@ -154,6 +158,9 @@ protected:
/// Stores outbound messages. /// Stores outbound messages.
message_queue_type 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>; using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
......
...@@ -23,7 +23,7 @@ ...@@ -23,7 +23,7 @@
namespace caf { namespace caf {
namespace net { namespace net {
template <class Transport, class Application> template <class Transport>
class endpoint_manager_impl : public endpoint_manager { class endpoint_manager_impl : public endpoint_manager {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -32,15 +32,13 @@ public: ...@@ -32,15 +32,13 @@ public:
using transport_type = Transport; using transport_type = Transport;
using application_type = Application; using application_type = typename transport_type::application_type;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys, endpoint_manager_impl(const multiplexer_ptr& parent, actor_system& sys,
Transport trans, Application app) Transport trans)
: super(trans.handle(), parent, sys), : super(trans.handle(), parent, sys), transport_(std::move(trans)) {
transport_(std::move(trans)),
application_(std::move(app)) {
// nop // nop
} }
...@@ -54,8 +52,16 @@ public: ...@@ -54,8 +52,16 @@ public:
return transport_; return transport_;
} }
application_type& application() { // -- timeout management -----------------------------------------------------
return application_;
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 ---------------------------------------------------- // -- interface functions ----------------------------------------------------
...@@ -91,7 +97,7 @@ public: ...@@ -91,7 +97,7 @@ public:
} }
void handle_error(sec code) override { void handle_error(sec code) override {
transport_.handle_error(application_, code); transport_.handle_error(code);
} }
serialize_fun_type serialize_fun() const noexcept override { serialize_fun_type serialize_fun() const noexcept override {
...@@ -100,7 +106,9 @@ public: ...@@ -100,7 +106,9 @@ public:
private: private:
transport_type transport_; transport_type transport_;
application_type application_;
/// Stores the id for the next timeout.
uint64_t next_timeout_id_;
}; };
} // namespace net } // namespace net
......
...@@ -27,6 +27,8 @@ namespace net { ...@@ -27,6 +27,8 @@ namespace net {
class multiplexer; class multiplexer;
class socket_manager; class socket_manager;
template <class Application, class IdType = unit_t>
class transport_worker;
struct network_socket; struct network_socket;
struct pipe_socket; struct pipe_socket;
......
...@@ -25,12 +25,11 @@ ...@@ -25,12 +25,11 @@
namespace caf { namespace caf {
namespace net { namespace net {
template <class Transport, class Application> template <class Transport>
endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx, endpoint_manager_ptr make_endpoint_manager(const multiplexer_ptr& mpx,
actor_system& sys, Transport trans, actor_system& sys, Transport trans) {
Application app) { using impl = endpoint_manager_impl<Transport>;
using impl = endpoint_manager_impl<Transport, Application>; return make_counted<impl>(mpx, sys, std::move(trans));
return make_counted<impl>(mpx, sys, std::move(trans), std::move(app));
} }
} // namespace net } // namespace net
......
...@@ -25,26 +25,49 @@ ...@@ -25,26 +25,49 @@
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp" #include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
namespace caf { namespace caf {
namespace policy { namespace net {
/// Implements a scribe policy that manages a stream socket. /// Implements a stream_transport that manages a stream socket.
class scribe { template <class Application>
class stream_transport {
public: 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_; return handle_;
} }
// -- member functions -------------------------------------------------------
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
parent.application().init(parent); if (auto err = worker_.init(parent))
parent.mask_add(net::operation::read); return err;
parent.mask_add(operation::read);
return none; return none;
} }
...@@ -59,14 +82,15 @@ public: ...@@ -59,14 +82,15 @@ public:
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes; collected_ += *num_bytes;
if (collected_ >= read_threshold_) { 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(); prepare_next_read();
} }
return true; return true;
} else { } else {
auto err = get<sec>(ret); auto err = get<sec>(ret);
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err)); CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
parent.application().handle_error(err); worker_.handle_error(err);
return false; return false;
} }
} }
...@@ -74,25 +98,88 @@ public: ...@@ -74,25 +98,88 @@ public:
template <class Parent> template <class Parent>
bool handle_write_event(Parent& parent) { bool handle_write_event(Parent& parent) {
// Try to write leftover data. // Try to write leftover data.
write_some(parent); write_some();
// Get new data from parent. // Get new data from parent.
// TODO: dont read all messages at once - get one by one. // TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr; for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) { 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. // 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> 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()) if (write_buf_.empty())
return false; return false;
auto len = write_buf_.size() - written_; auto len = write_buf_.size() - written_;
auto buf = write_buf_.data() + written_; auto buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len)); 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)) { if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
// Update state. // Update state.
...@@ -105,47 +192,27 @@ public: ...@@ -105,47 +192,27 @@ public:
} else { } else {
auto err = get<sec>(ret); auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err)); CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
parent.application().handle_error(err); worker_.handle_error(err);
return false; return false;
} }
return true; return true;
} }
template <class Parent> transport_worker<application_type> worker_;
void resolve(Parent& parent, const std::string& path, actor listener) { stream_socket handle_;
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_;
std::vector<byte> read_buf_; std::vector<byte> read_buf_;
std::vector<byte> write_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 read_threshold_;
size_t collected_; size_t collected_;
size_t max_; size_t max_;
net::receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
size_t written_; size_t written_;
}; };
} // namespace policy } // namespace net
} // namespace caf } // namespace caf
...@@ -16,64 +16,83 @@ ...@@ -16,64 +16,83 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/policy/scribe.hpp" #pragma once
#include <system_error>
#include "caf/byte.hpp" #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/span.hpp"
#include "caf/unit.hpp"
namespace caf { namespace caf {
namespace policy { namespace net {
scribe::scribe(caf::net::stream_socket handle) /// Implements a worker for transport protocols.
: handle_(handle), template <class Application, class IdType>
max_consecutive_reads_(0), class transport_worker {
read_threshold_(1024), public:
collected_(0), // -- member types -----------------------------------------------------------
max_(1024),
rd_flag_(net::receive_policy_flag::exactly), using id_type = IdType;
written_(0) {
// nop using application_type = Application;
}
// -- constructors, destructors, and assignment operators --------------------
void scribe::prepare_next_read() {
read_buf_.clear(); transport_worker(application_type application, id_type id = id_type{})
collected_ = 0; : application_(std::move(application)), id_(std::move(id)) {
// This cast does nothing, but prevents a weird compiler error on GCC <= 4.9. // nop
// 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: // -- member functions -------------------------------------------------------
if (read_buf_.size() != max_)
read_buf_.resize(max_); template <class Parent>
read_threshold_ = max_; error init(Parent& parent) {
break; return application_.init(parent);
case net::receive_policy_flag::at_most: }
if (read_buf_.size() != max_)
read_buf_.resize(max_); template <class Parent>
read_threshold_ = 1; void handle_data(Parent& parent, span<const byte> data) {
break; application_.handle_data(parent, data);
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); template <class Parent>
if (read_buf_.size() != max_size) void write_message(Parent& parent,
read_buf_.resize(max_size); std::unique_ptr<net::endpoint_manager::message> msg) {
read_threshold_ = max_; auto decorator = make_write_packet_decorator(*this, parent);
break; 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) { private:
rd_flag_ = cfg.first; application_type application_;
max_ = cfg.second; id_type id_;
prepare_next_read(); };
}
void scribe::write_packet(span<const byte> buf) { template <class Application, class IdType = unit_t>
write_buf_.insert(write_buf_.end(), buf.begin(), buf.end()); using transport_worker_ptr = std::shared_ptr<
} transport_worker<Application, IdType>>;
} // namespace policy } // namespace net
} // namespace caf } // namespace caf
...@@ -18,6 +18,9 @@ ...@@ -18,6 +18,9 @@
#pragma once #pragma once
#include "caf/byte.hpp"
#include "caf/span.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -26,34 +29,44 @@ namespace net { ...@@ -26,34 +29,44 @@ namespace net {
template <class Object, class Parent> template <class Object, class Parent>
class write_packet_decorator { class write_packet_decorator {
public: 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) { : object_(object), parent_(parent) {
// nop // nop
} }
template <class Header> // -- properties -------------------------------------------------------------
void write_packet(const Header& header, span<const byte> payload) {
object_.write_packet(parent_, header, payload);
}
actor_system& system() { actor_system& system() {
parent_.system(); return parent_.system();
} }
void cancel_timeout(atom_value type, uint64_t id) { transport_type& transport() {
parent_.cancel_timeout(type, id); return parent_.transport();
} }
void set_timeout(timestamp tout, atom_value type, uint64_t id) { // -- member functions -------------------------------------------------------
parent_.set_timeout(tout, type, id);
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() { void cancel_timeout(atom_value type, uint64_t id) {
return parent_.transport_; parent_.cancel_timeout(type, id);
} }
typename parent::application_type& application() { template <class... Ts>
return parent_.application_; uint64_t set_timeout(timestamp tout, atom_value type, Ts&&... xs) {
return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...);
} }
private: private:
......
...@@ -72,6 +72,8 @@ public: ...@@ -72,6 +72,8 @@ public:
class dummy_transport { class dummy_transport {
public: public:
using application_type = dummy_application;
dummy_transport(stream_socket handle, std::shared_ptr<std::vector<byte>> data) dummy_transport(stream_socket handle, std::shared_ptr<std::vector<byte>> data)
: handle_(handle), data_(data), read_buf_(1024) { : handle_(handle), data_(data), read_buf_(1024) {
// nop // nop
...@@ -84,7 +86,7 @@ public: ...@@ -84,7 +86,7 @@ public:
template <class Manager> template <class Manager>
error init(Manager& manager) { error init(Manager& manager) {
auto test_bytes = as_bytes(make_span(hello_test)); 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)); CAF_CHECK(manager.mask_add(operation::read_write));
return none; return none;
} }
...@@ -104,18 +106,17 @@ public: ...@@ -104,18 +106,17 @@ public:
bool handle_write_event(Manager& mgr) { bool handle_write_event(Manager& mgr) {
for (auto x = mgr.next_message(); x != nullptr; x = mgr.next_message()) { for (auto x = mgr.next_message(); x != nullptr; x = mgr.next_message()) {
auto& payload = x->payload; 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)) { if (auto num_bytes = get_if<size_t>(&res)) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + *num_bytes); buf_.erase(buf_.begin(), buf_.begin() + *num_bytes);
return write_buf_.size() > 0; return buf_.size() > 0;
} }
return get<sec>(res) == sec::unavailable_or_would_block; return get<sec>(res) == sec::unavailable_or_would_block;
} }
template <class Manager> void handle_error(sec) {
void handle_error(Manager&, sec) {
// nop // nop
} }
...@@ -143,7 +144,7 @@ private: ...@@ -143,7 +144,7 @@ private:
std::vector<byte> read_buf_; std::vector<byte> read_buf_;
std::vector<byte> write_buf_; std::vector<byte> buf_;
}; };
} // namespace } // namespace
...@@ -160,8 +161,7 @@ CAF_TEST(send and receive) { ...@@ -160,8 +161,7 @@ CAF_TEST(send and receive) {
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf}, dummy_transport{sockets.first, buf});
dummy_application{});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
...@@ -184,8 +184,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -184,8 +184,7 @@ CAF_TEST(resolve and proxy communication) {
nonblocking(sockets.second, true); nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf}, dummy_transport{sockets.first, buf});
dummy_application{});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates(); mpx->handle_updates();
run(); run();
......
...@@ -37,6 +37,8 @@ struct dummy_socket { ...@@ -37,6 +37,8 @@ struct dummy_socket {
// nop // nop
} }
dummy_socket(const dummy_socket&) = default;
dummy_socket& operator=(const dummy_socket& other) { dummy_socket& operator=(const dummy_socket& other) {
id = other.id; id = other.id;
closed = other.closed; closed = other.closed;
......
...@@ -16,9 +16,9 @@ ...@@ -16,9 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. * * 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" #include "caf/test/dsl.hpp"
...@@ -75,7 +75,7 @@ public: ...@@ -75,7 +75,7 @@ public:
template <class Transport> template <class Transport>
void write_message(Transport& transport, void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) { std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(msg->payload); transport.write_packet(span<byte>{}, msg->payload);
} }
template <class Parent> template <class Parent>
...@@ -123,24 +123,25 @@ private: ...@@ -123,24 +123,25 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
using transport_type = stream_transport<dummy_application>;
std::vector<byte> read_buf(1024); std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>(); auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair()); 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)), CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
CAF_MESSAGE("configure scribe_policy"); transport_type transport{sockets.first, dummy_application{buf}};
policy::scribe scribe{sockets.first}; transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
scribe.configure_read(net::receive_policy::exactly(hello_manager.size())); auto mgr = make_endpoint_manager(mpx, sys, transport);
auto mgr = make_endpoint_manager(mpx, sys, scribe, dummy_application{buf});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); 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))), CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
hello_manager.size()); hello_manager.size());
CAF_MESSAGE("wrote " << hello_manager.size() << " bytes.");
run(); run();
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()), CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()),
buf->size()), buf->size()),
...@@ -148,13 +149,15 @@ CAF_TEST(receive) { ...@@ -148,13 +149,15 @@ CAF_TEST(receive) {
} }
CAF_TEST(resolve and proxy communication) { CAF_TEST(resolve and proxy communication) {
using transport_type = stream_transport<dummy_application>;
std::vector<byte> read_buf(1024); std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<byte>>(); auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true); nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first}, auto mgr = make_endpoint_manager(mpx, sys,
dummy_application{buf}); transport_type{sockets.first,
dummy_application{buf}});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates(); mpx->handle_updates();
run(); run();
......
...@@ -18,7 +18,7 @@ ...@@ -18,7 +18,7 @@
#define CAF_SUITE string_application #define CAF_SUITE string_application
#include "caf/policy/scribe.hpp" #include "caf/net/stream_transport.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
...@@ -100,12 +100,9 @@ public: ...@@ -100,12 +100,9 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) { std::unique_ptr<net::endpoint_manager::message> msg) {
std::vector<byte> buf;
header_type header{static_cast<uint32_t>(msg->payload.size())}; header_type header{static_cast<uint32_t>(msg->payload.size())};
buf.resize(sizeof(header_type)); std::vector<byte> payload(msg->payload.begin(), msg->payload.end());
memcpy(buf.data(), &header, buf.size()); parent.write_packet(as_bytes(make_span(&header, 1)), make_span(payload));
buf.insert(buf.end(), msg->payload.begin(), msg->payload.end());
parent.write_packet(buf);
} }
static expected<std::vector<byte>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
...@@ -152,7 +149,8 @@ public: ...@@ -152,7 +149,8 @@ public:
if (header_.payload == 0) if (header_.payload == 0)
Base::handle_packet(parent, header_, span<const byte>{}); Base::handle_packet(parent, header_, span<const byte>{});
else else
parent.configure_read(net::receive_policy::exactly(header_.payload)); parent.transport().configure_read(
net::receive_policy::exactly(header_.payload));
await_payload_ = true; await_payload_ = true;
} }
} }
...@@ -191,6 +189,7 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) ...@@ -191,6 +189,7 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
using application_type = extend<string_application>::with< using application_type = extend<string_application>::with<
stream_string_application>; stream_string_application>;
using transport_type = stream_transport<application_type>;
std::vector<byte> read_buf(1024); std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>(); auto buf = std::make_shared<std::vector<byte>>();
...@@ -199,13 +198,15 @@ CAF_TEST(receive) { ...@@ -199,13 +198,15 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers"); CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first}, auto mgr1 = make_endpoint_manager(mpx, sys,
application_type{sys, buf}); transport_type{sockets.first,
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr1->init(), none); CAF_CHECK_EQUAL(mgr1->init(), none);
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto mgr2 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.second}, auto mgr2 = make_endpoint_manager(mpx, sys,
application_type{sys, buf}); transport_type{sockets.second,
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr2->init(), none); CAF_CHECK_EQUAL(mgr2->init(), none);
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u); 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