Commit 3743404c authored by Joseph Noir's avatar Joseph Noir

Cleanup file structure & initialization

parent 8113c99c
......@@ -50,6 +50,7 @@ set(LIBCAF_IO_SRCS
src/newb_tcp.cpp
src/newb_udp.cpp
src/newb.cpp
src/policies.cpp
)
add_custom_target(libcaf_io)
......
......@@ -25,7 +25,6 @@
#include <tuple>
#include "caf/actor_clock.hpp"
#include "caf/callback.hpp"
#include "caf/config.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/call_cfun.hpp"
......@@ -41,11 +40,13 @@
#include "caf/mixin/behavior_changer.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/policy/accept.hpp"
#include "caf/policy/protocol.hpp"
#include "caf/policy/transport.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf {
namespace io {
namespace network {
// -- forward declarations -----------------------------------------------------
......@@ -56,34 +57,32 @@ template <class T>
struct newb;
} // namespace io
} // namespace network
// -- required to make newb an actor ------------------------------------------
template <class T>
class behavior_type_of<io::network::newb<T>> {
class behavior_type_of<io::newb<T>> {
public:
using type = behavior;
};
namespace io {
namespace network {
// -- aliases ------------------------------------------------------------------
using byte_buffer = std::vector<char>;
using header_writer = caf::callback<byte_buffer&>;
using byte_buffer = caf::policy::byte_buffer;
using header_writer = caf::policy::header_writer;
// -- newb base ----------------------------------------------------------------
struct newb_base : public network::event_handler {
newb_base(default_multiplexer& dm, native_socket sockfd);
newb_base(network::default_multiplexer& dm, network::native_socket sockfd);
virtual void start() = 0;
virtual void stop() = 0;
virtual void io_error(operation op, error err) = 0;
virtual void io_error(network::operation op, error err) = 0;
virtual void start_reading() = 0;
......@@ -95,165 +94,6 @@ struct newb_base : public network::event_handler {
};
// -- transport policy ---------------------------------------------------------
struct transport_policy {
transport_policy();
virtual ~transport_policy();
virtual io::network::rw_state write_some(newb_base*);
virtual io::network::rw_state read_some(newb_base*);
virtual bool should_deliver();
virtual bool must_read_more(newb_base*);
virtual void prepare_next_read(newb_base*);
virtual void prepare_next_write(newb_base*);
virtual void configure_read(receive_policy::config);
virtual void flush(newb_base*);
virtual byte_buffer& wr_buf();
template <class T>
error read_some(newb_base* parent, protocol_policy<T>& policy) {
CAF_LOG_TRACE("");
size_t reads = 0;
while (reads < max_consecutive_reads || must_read_more(parent)) {
auto read_result = read_some(parent);
switch (read_result) {
case rw_state::success:
if (received_bytes == 0)
return none;
if (should_deliver()) {
auto res = policy.read(receive_buffer.data(), received_bytes);
prepare_next_read(parent);
if (res)
return res;
}
break;
case rw_state::indeterminate:
// No error, but don't continue reading.
return none;
case rw_state::failure:
// Reading failed.
return sec::runtime_error;
}
++reads;
}
return none;
}
virtual expected<native_socket>
connect(const std::string&, uint16_t,
optional<io::network::protocol::network> = none);
size_t received_bytes;
size_t max_consecutive_reads;
byte_buffer offline_buffer;
byte_buffer receive_buffer;
byte_buffer send_buffer;
};
using transport_policy_ptr = std::unique_ptr<transport_policy>;
// -- accept policy ------------------------------------------------------------
struct accept_policy {
accept_policy(bool manual_read = false)
: manual_read(manual_read) {
// nop
}
virtual ~accept_policy();
virtual expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) = 0;
virtual std::pair<native_socket, transport_policy_ptr> accept(newb_base*) {
return {0, nullptr};
}
/// If `requires_raw_data` is set to true, the acceptor will only call
/// this function for new read event and let the policy handle everything
/// else.
virtual void read_event(newb_base*) {
// nop
}
virtual error write_event(newb_base*) {
return none;
}
virtual void init(newb_base&) {
// nop
}
bool manual_read;
};
using accept_policy_ptr = std::unique_ptr<accept_policy>;
// -- protocol policy ----------------------------------------------------------
struct protocol_policy_base {
virtual ~protocol_policy_base();
virtual error read(char* bytes, size_t count) = 0;
virtual error timeout(atom_value, uint32_t) = 0;
virtual void write_header(byte_buffer&, header_writer*) = 0;
virtual void prepare_for_sending(byte_buffer&, size_t, size_t, size_t) = 0;
};
template <class T>
struct protocol_policy : protocol_policy_base {
using message_type = T;
virtual ~protocol_policy() override {
// nop
}
};
template <class T>
using protocol_policy_ptr = std::unique_ptr<protocol_policy<T>>;
template <class T>
struct generic_protocol
: public io::network::protocol_policy<typename T::message_type> {
T impl;
generic_protocol(io::network::newb<typename T::message_type>* parent)
: impl(parent) {
// nop
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) override {
impl.write_header(buf, hw);
}
void prepare_for_sending(io::network::byte_buffer& buf, size_t hstart,
size_t offset, size_t plen) override {
impl.prepare_for_sending(buf, hstart, offset, plen);
}
};
// -- new broker classes -------------------------------------------------------
/// @relates newb
......@@ -261,7 +101,7 @@ struct generic_protocol
template <class Message>
struct write_handle {
newb<Message>* parent;
protocol_policy_base* protocol;
policy::protocol_base* protocol;
byte_buffer* buf;
size_t header_start;
size_t header_len;
......@@ -289,18 +129,23 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
// -- constructors and destructors -------------------------------------------
newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd)
newb(actor_config& cfg, network::default_multiplexer& dm,
network::native_socket sockfd,
policy::transport_ptr transport,
policy::protocol_ptr<Message> protocol)
: super(cfg),
newb_base(dm, sockfd),
trans(std::move(transport)),
proto(std::move(protocol)),
value_(strong_actor_ptr{}, make_message_id(),
mailbox_element::forwarding_stack{}, Message{}),
reading_(false),
writing_(false) {
CAF_LOG_TRACE("");
scheduled_actor::set_timeout_handler([&](timeout_msg& msg) {
if (protocol)
protocol->timeout(msg.type, msg.timeout_id);
proto->timeout(msg.type, msg.timeout_id);
});
proto->init(this);
}
newb() = default;
......@@ -423,8 +268,8 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
if (!reading_ && !writing_)
intrusive_ptr_add_ref(super::ctrl());
start_reading();
if (transport)
transport->prepare_next_read(this);
if (trans)
trans->prepare_next_read(this);
}
void stop() override {
......@@ -435,7 +280,8 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
stop_writing();
}
void io_error(operation op, error err) override {
void io_error(network::operation op, error err) override {
using network::operation;
if (!this->getf(this->is_cleaned_up_flag)) {
auto mptr = make_mailbox_element(nullptr, invalid_message_id, {},
io_error_msg{op, std::move(err)});
......@@ -477,13 +323,13 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
void start_writing() override {
if (!writing_) {
event_handler::backend().add(operation::write, fd(), this);
event_handler::backend().add(network::operation::write, fd(), this);
writing_ = true;
}
}
void stop_writing() override {
event_handler::backend().del(operation::write, fd(), this);
event_handler::backend().del(network::operation::write, fd(), this);
}
// -- members ----------------------------------------------------------------
......@@ -495,30 +341,30 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
/// Get a write buffer to write data to be sent by this broker.
write_handle<Message> wr_buf(header_writer* hw) {
auto& buf = transport->wr_buf();
auto& buf = trans->wr_buf();
auto hstart = buf.size();
protocol->write_header(buf, hw);
proto->write_header(buf, hw);
auto hlen = buf.size() - hstart;
return {this, protocol.get(), &buf, hstart, hlen};
return {this, proto.get(), &buf, hstart, hlen};
}
byte_buffer& wr_buf() {
return transport->wr_buf();
return trans->wr_buf();
}
void flush() {
transport->flush(this);
trans->flush(this);
}
void read_event() {
auto err = transport->read_some(this, *protocol);
auto err = trans->read_some(this, *proto);
if (err)
io_error(operation::read, std::move(err));
io_error(network::operation::read, std::move(err));
}
void write_event() {
if (transport->write_some(this) == rw_state::failure)
io_error(operation::write, sec::runtime_error);
if (trans->write_some(this) == network::rw_state::failure)
io_error(network::operation::write, sec::runtime_error);
}
void handle_error() {
......@@ -557,7 +403,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
/// Configure the number of bytes read for the next packet. (Can be ignored by
/// the transport policy if its protocol does not support this functionality.)
void configure_read(receive_policy::config config) {
transport->configure_read(config);
trans->configure_read(config);
}
/// @cond PRIVATE
......@@ -572,8 +418,8 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
// -- policies ---------------------------------------------------------------
std::unique_ptr<transport_policy> transport;
std::unique_ptr<protocol_policy<Message>> protocol;
policy::transport_ptr trans;
policy::protocol_ptr<Message> proto;
/// @endcond
......@@ -640,27 +486,26 @@ using first_argument_type
/// Spawns a new "newb" broker.
template <class Protocol, spawn_options Os = no_spawn_options, class F, class... Ts>
typename infer_handle_from_fun<F>::type
spawn_newb(actor_system& sys, F fun, transport_policy_ptr transport,
native_socket sockfd, Ts&&... xs) {
spawn_newb(actor_system& sys, F fun, policy::transport_ptr transport,
network::native_socket sockfd, Ts&&... xs) {
using impl = typename infer_handle_from_fun<F>::impl;
using first = first_argument_type<F>;
using message = typename std::remove_pointer<first>::type::message_type;
auto& dm = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
auto& dm = dynamic_cast<network::default_multiplexer&>(sys.middleman().backend());
// Setup the config.
actor_config cfg{&dm};
actor_config cfg(&dm);
detail::init_fun_factory<impl, F> fac;
auto init_fun = fac(std::move(fun), std::forward<Ts>(xs)...);
cfg.init_fun = [init_fun](local_actor* self) mutable -> behavior {
return init_fun(self);
};
auto res = sys.spawn_class<impl, Os>(cfg, dm, sockfd);
policy::protocol_ptr<message> proto(new Protocol());
auto res = sys.spawn_class<impl, Os>(cfg, dm, sockfd, std::move(transport),
std::move(proto));
// Get a reference to the newb type.
auto ptr = caf::actor_cast<caf::abstract_actor*>(res);
CAF_ASSERT(ptr != nullptr);
auto& ref = dynamic_cast<newb<message>&>(*ptr);
// Set the policies.
ref.transport = std::move(transport);
ref.protocol.reset(new Protocol(&ref));
// Start the event handler.
ref.start();
return res;
......@@ -669,42 +514,15 @@ spawn_newb(actor_system& sys, F fun, transport_policy_ptr transport,
/// Spawn a new "newb" broker client to connect to `host`:`port`.
template <class Protocol, spawn_options Os = no_spawn_options, class F, class... Ts>
expected<typename infer_handle_from_fun<F>::type>
spawn_client(actor_system& sys, F fun, transport_policy_ptr transport,
spawn_client(actor_system& sys, F fun, policy::transport_ptr transport,
std::string host, uint16_t port, Ts&&... xs) {
expected<native_socket> esock = transport->connect(host, port);
expected<network::native_socket> esock = transport->connect(host, port);
if (!esock)
return std::move(esock.error());
return spawn_newb<Protocol>(sys, fun, std::move(transport), *esock,
std::forward<Ts>(xs)...);
}
// TODO: Remove these two.
template <class Newb>
actor make_newb(actor_system& sys, native_socket sockfd) {
auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
actor_config acfg{&mpx};
auto res = sys.spawn_impl<Newb, hidden + lazy_init>(acfg, mpx, sockfd);
return actor_cast<actor>(res);
}
template <class Newb, class Transport, class Protocol>
actor make_client_newb(actor_system& sys, std::string host, uint16_t port) {
transport_policy_ptr trans{new Transport};
expected<native_socket> esock = trans->connect(host, port);
if (!esock)
return {};
auto res = make_newb<Newb>(sys, *esock);
auto ptr = caf::actor_cast<caf::abstract_actor*>(res);
CAF_ASSERT(ptr != nullptr);
auto& ref = dynamic_cast<Newb&>(*ptr);
ref.transport = std::move(trans);
ref.protocol.reset(new Protocol(&ref));
ref.start();
return res;
}
// -- new broker acceptor ------------------------------------------------------
template <class Protocol, class Fun, class... Ts>
......@@ -714,8 +532,10 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
// -- constructors and destructors -------------------------------------------
newb_acceptor(default_multiplexer& dm, native_socket sockfd, Fun f, Ts&&... xs)
newb_acceptor(network::default_multiplexer& dm, network::native_socket sockfd,
Fun f, policy::accept_policy_ptr pol, Ts&&... xs)
: newb_base(dm, sockfd),
accept_pol(std::move(pol)),
fun_(std::move(f)),
reading_(false),
writing_(false),
......@@ -781,7 +601,7 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
stop_writing();
}
void io_error(operation op, error err) override {
void io_error(network::operation op, error err) override {
CAF_LOG_ERROR("operation " << to_string(op) << " failed: "
<< backend().system().render(err));
stop();
......@@ -800,13 +620,13 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
void start_writing() override {
if (!writing_) {
event_handler::backend().add(operation::write, fd(), this);
event_handler::backend().add(network::operation::write, fd(), this);
writing_ = true;
}
}
void stop_writing() override {
event_handler::backend().del(operation::write, fd(), this);
event_handler::backend().del(network::operation::write, fd(), this);
}
// -- members ----------------------------------------------------------------
......@@ -815,12 +635,12 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
if (accept_pol->manual_read) {
accept_pol->read_event(this);
} else {
native_socket sock;
transport_policy_ptr transport;
std::tie(sock, transport) = accept_pol->accept(this);
network::native_socket sock;
policy::transport_ptr transport;
std::tie(sock, transport) = accept_pol->accept_event(this);
auto en = create_newb(sock, std::move(transport));
if (!en) {
io_error(operation::read, std::move(en.error()));
io_error(network::operation::read, std::move(en.error()));
return;
}
auto ptr = caf::actor_cast<caf::abstract_actor*>(*en);
......@@ -834,28 +654,19 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
accept_pol->write_event(this);
}
virtual expected<actor> create_newb(native_socket sockfd,
transport_policy_ptr pol) {
virtual expected<actor> create_newb(network::native_socket sockfd,
policy::transport_ptr pol) {
CAF_LOG_TRACE(CAF_ARG(sockfd));
// auto n = io::network::spawn_newb<Protocol>(this->backend().system(), fun_,
// std::move(pol), sockfd);
auto n = detail::apply_args_prefixed(
io::network::spawn_newb<Protocol, no_spawn_options, Fun, Ts...>,
io::spawn_newb<Protocol, no_spawn_options, Fun, Ts...>,
detail::get_indices(args_),
args_, this->backend().system(),
fun_, std::move(pol), sockfd
);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr) {
// TODO: Clean this up!
std::cerr << "failed to spawn newb" << std::endl;
return sec::runtime_error;
}
// auto& ref = dynamic_cast<newb_type&>(*ptr);
return n;
}
std::unique_ptr<accept_policy> accept_pol;
policy::accept_policy_ptr accept_pol;
private:
Fun fun_;
......@@ -869,20 +680,20 @@ using acceptor_ptr = caf::intrusive_ptr<newb_acceptor<P, F, Ts...>>;
template <class Protocol, class Fun, class... Ts>
acceptor_ptr<Protocol, Fun, Ts...>
make_acceptor(actor_system& sys, Fun fun, accept_policy_ptr pol,
native_socket sockfd, Ts&&... xs) {
auto& dm = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
make_acceptor(actor_system& sys, Fun fun, policy::accept_policy_ptr pol,
network::native_socket sockfd, Ts&&... xs) {
auto& dm = dynamic_cast<network::default_multiplexer&>(sys.middleman().backend());
auto res = make_counted<newb_acceptor<Protocol, Fun, Ts...>>(dm, sockfd,
std::move(fun),
std::move(pol),
std::forward<Ts>(xs)...);
res->accept_pol = std::move(pol);
res->start();
return res;
}
template <class Protocol, class F, class... Ts>
expected<caf::intrusive_ptr<newb_acceptor<Protocol, F, Ts...>>>
make_server(actor_system& sys, F fun, accept_policy_ptr pol,
make_server(actor_system& sys, F fun, policy::accept_policy_ptr pol,
uint16_t port, const char* addr, bool reuse, Ts&&... xs) {
auto esock = pol->create_socket(port, addr, reuse);
if (!esock) {
......@@ -895,13 +706,12 @@ make_server(actor_system& sys, F fun, accept_policy_ptr pol,
template <class Protocol, class F>
expected<caf::intrusive_ptr<newb_acceptor<Protocol, F>>>
make_server(actor_system& sys, F fun, accept_policy_ptr pol,
make_server(actor_system& sys, F fun, policy::accept_policy_ptr pol,
uint16_t port, const char* addr = nullptr, bool reuse = false) {
return make_server<Protocol, F>(sys, std::move(fun), std::move(pol), port,
addr, reuse);
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#pragma once
#include "caf/config.hpp"
#include "caf/policy/transport.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace io {
struct newb_base;
} // namespace io
namespace policy {
struct accept {
accept(bool manual_read = false)
: manual_read(manual_read) {
// nop
}
virtual ~accept();
virtual expected<io::network::native_socket>
create_socket(uint16_t port, const char* host, bool reuse = false) = 0;
virtual std::pair<io::network::native_socket, transport_ptr>
accept_event(io::newb_base*) {
return {0, nullptr};
}
/// If `requires_raw_data` is set to true, the acceptor will only call
/// this function for new read event and let the policy handle everything
/// else.
virtual void read_event(io::newb_base*) {
// nop
}
virtual error write_event(io::newb_base*) {
return none;
}
virtual void init(io::newb_base&) {
// nop
}
bool manual_read;
};
using accept_policy_ptr = std::unique_ptr<accept>;
} // namespace policy
} // namespace caf
......@@ -19,7 +19,7 @@
#pragma once
#include "caf/actor.hpp"
#include "caf/io/network/newb.hpp"
#include "caf/io/newb.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
......@@ -57,11 +57,11 @@ struct datagram_basp {
static constexpr size_t header_size = basp_header_len;
using message_type = new_basp_msg;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
io::newb<message_type>* parent;
message_type msg;
datagram_basp(io::network::newb<message_type>* parent) : parent(parent) {
// nop
void init(io::newb<message_type>* n) {
parent = n;
}
error read(char* bytes, size_t count) {
......@@ -90,14 +90,14 @@ struct datagram_basp {
return none;
}
size_t write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) {
size_t write_header(io::byte_buffer& buf,
io::header_writer* hw) {
CAF_ASSERT(hw != nullptr);
(*hw)(buf);
return header_size;
}
void prepare_for_sending(io::network::byte_buffer& buf,
void prepare_for_sending(io::byte_buffer& buf,
size_t hstart, size_t offset, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(),
buf.data() + hstart + offset,
......@@ -111,13 +111,12 @@ struct stream_basp {
static constexpr size_t header_size = basp_header_len;
using message_type = new_basp_msg;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
io::newb<message_type>* parent;
message_type msg;
bool expecting_header = true;
stream_basp(io::network::newb<message_type>* parent) : parent(parent) {
// TODO: this is dangerous ...
// Maybe we need an init function that is called with `start()`?
void init(io::newb<message_type>* n) {
parent = n;
parent->configure_read(io::receive_policy::exactly(basp_header_len));
}
......@@ -157,14 +156,14 @@ struct stream_basp {
return none;
}
size_t write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) {
size_t write_header(io::byte_buffer& buf,
io::header_writer* hw) {
CAF_ASSERT(hw != nullptr);
(*hw)(buf);
return header_size;
}
void prepare_for_sending(io::network::byte_buffer& buf,
void prepare_for_sending(io::byte_buffer& buf,
size_t hstart, size_t offset, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(),
buf.data() + hstart + offset,
......
......@@ -26,7 +26,7 @@
#include "caf/binary_serializer.hpp"
#include "caf/defaults.hpp"
#include "caf/error.hpp"
#include "caf/io/network/newb.hpp"
#include "caf/io/newb.hpp"
namespace caf {
namespace policy {
......@@ -80,17 +80,22 @@ struct ordering {
size_t max_pending_messages;
bool use_timeouts;
std::chrono::milliseconds pending_to = std::chrono::milliseconds(100);
io::network::newb<message_type>* parent;
io::newb<message_type>* parent;
Next next;
std::map<sequence_type, std::vector<char>, sequence_comperator> pending;
ordering(io::network::newb<message_type>* parent, bool use_timeouts = true)
: max_pending_messages(get_or(parent->config(),
"middleman.max-pending-messages",
caf::defaults::middleman::max_pending_messages)),
use_timeouts(use_timeouts),
parent(parent),
next(parent) {
void init(io::newb<message_type>* n, bool enable_timeouts = true) {
this->use_timeouts = enable_timeouts;
parent = n;
next.init(parent);
max_pending_messages = get_or(parent->config(),
"middleman.max-pending-messages",
max_pending_messages);
}
ordering()
: max_pending_messages(caf::defaults::middleman::max_pending_messages),
use_timeouts(false) {
// nop
}
......@@ -152,8 +157,8 @@ struct ordering {
return next.timeout(atm, id);
}
void write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) {
void write_header(io::byte_buffer& buf,
io::header_writer* hw) {
binary_serializer bs(&parent->backend(), buf);
bs(ordering_header{seq_write});
seq_write += 1;
......@@ -161,7 +166,7 @@ struct ordering {
return;
}
void prepare_for_sending(io::network::byte_buffer& buf,
void prepare_for_sending(io::byte_buffer& buf,
size_t hstart, size_t offset, size_t plen) {
next.prepare_for_sending(buf, hstart, offset + header_size, plen);
}
......
......@@ -19,7 +19,7 @@
#pragma once
#include "caf/actor.hpp"
#include "caf/io/network/newb.hpp"
#include "caf/io/newb.hpp"
namespace caf {
namespace policy {
......@@ -37,11 +37,11 @@ typename Inspector::result_type inspect(Inspector& fun, new_raw_msg& data) {
struct raw {
using message_type = new_raw_msg;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
io::newb<message_type>* parent;
message_type msg;
raw(io::network::newb<message_type>* parent) : parent(parent) {
// nop
void init(io::newb<message_type>* n) {
this->parent = n;
}
error read(char* bytes, size_t count) {
......@@ -55,12 +55,12 @@ struct raw {
return none;
}
size_t write_header(io::network::byte_buffer&,
io::network::header_writer*) {
size_t write_header(io::byte_buffer&,
io::header_writer*) {
return 0;
}
void prepare_for_sending(io::network::byte_buffer&, size_t, size_t, size_t) {
void prepare_for_sending(io::byte_buffer&, size_t, size_t, size_t) {
// nop
}
};
......
......@@ -25,7 +25,7 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/error.hpp"
#include "caf/io/network/newb.hpp"
#include "caf/io/newb.hpp"
namespace caf {
namespace policy {
......@@ -57,14 +57,13 @@ struct reliability {
id_type id_write = 0;
// TODO: Make this configurable.
std::chrono::milliseconds retransmit_to = std::chrono::milliseconds(100);
io::network::newb<message_type>* parent;
io::newb<message_type>* parent;
Next next;
std::unordered_map<id_type, io::network::byte_buffer> unacked;
std::unordered_map<id_type, io::byte_buffer> unacked;
reliability(io::network::newb<message_type>* parent)
: parent(parent),
next(parent) {
// nop
void init(io::newb<message_type>* n) {
parent = n;
next.init(parent);
}
error read(char* bytes, size_t count) {
......@@ -106,22 +105,22 @@ struct reliability {
return next.timeout(atm, id);
}
void write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) {
void write_header(io::byte_buffer& buf,
io::header_writer* hw) {
binary_serializer bs(&parent->backend(), buf);
bs(reliability_header{id_write, false});
next.write_header(buf, hw);
return;
}
void prepare_for_sending(io::network::byte_buffer& buf,
void prepare_for_sending(io::byte_buffer& buf,
size_t hstart, size_t offset, size_t plen) {
next.prepare_for_sending(buf, hstart, offset + header_size, plen);
// Set timeout for retransmission.
parent->set_timeout(retransmit_to, reliability_atom::value, id_write);
// Add to unacked.
unacked.emplace(id_write,
io::network::byte_buffer(buf.begin() + hstart, buf.end()));
io::byte_buffer(buf.begin() + hstart, buf.end()));
id_write += 1;
}
};
......
......@@ -18,28 +18,29 @@
#pragma once
#include "caf/io/network/newb.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/policy/accept.hpp"
#include "caf/policy/transport.hpp"
namespace caf {
namespace policy {
struct tcp_transport : public io::network::transport_policy {
struct tcp_transport : public transport {
tcp_transport();
io::network::rw_state read_some(io::network::newb_base* parent) override;
io::network::rw_state read_some(io::newb_base* parent) override;
bool should_deliver() override;
void prepare_next_read(io::network::newb_base*) override;
void prepare_next_read(io::newb_base*) override;
void configure_read(io::receive_policy::config config) override;
io::network::rw_state write_some(io::network::newb_base* parent) override;
io::network::rw_state write_some(io::newb_base* parent) override;
void prepare_next_write(io::network::newb_base* parent) override;
void prepare_next_write(io::newb_base* parent) override;
void flush(io::network::newb_base* parent) override;
void flush(io::newb_base* parent) override;
expected<io::network::native_socket>
connect(const std::string& host, uint16_t port,
......@@ -56,18 +57,18 @@ struct tcp_transport : public io::network::transport_policy {
size_t written;
};
struct accept_tcp : public io::network::accept_policy {
struct accept_tcp : public accept {
expected<io::network::native_socket>
create_socket(uint16_t port,const char* host,bool reuse = false) override;
std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept(io::network::newb_base* parent) override;
std::pair<io::network::native_socket, transport_ptr>
accept_event(io::newb_base* parent) override;
void init(io::network::newb_base& n) override;
void init(io::newb_base& n) override;
};
template <class T>
using tcp_protocol = io::network::generic_protocol<T>;
using tcp_protocol = generic_protocol<T>;
} // namespace policy
} // namespace caf
......@@ -18,35 +18,37 @@
#pragma once
#include "caf/io/network/newb.hpp"
#include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/policy/accept.hpp"
#include "caf/policy/transport.hpp"
namespace caf {
namespace policy {
struct udp_transport : public io::network::transport_policy {
struct udp_transport : public transport {
udp_transport();
io::network::rw_state read_some(io::network::newb_base* parent) override;
io::network::rw_state read_some(io::newb_base* parent) override;
inline bool should_deliver() override {
CAF_LOG_TRACE("");
return received_bytes != 0 && sender == endpoint;
}
void prepare_next_read(io::network::newb_base*) override;
void prepare_next_read(io::newb_base*) override;
inline void configure_read(io::receive_policy::config) override {
// nop
}
io::network::rw_state write_some(io::network::newb_base* parent) override;
io::network::rw_state write_some(io::newb_base* parent) override;
void prepare_next_write(io::network::newb_base* parent) override;
void prepare_next_write(io::newb_base* parent) override;
io::network::byte_buffer& wr_buf() override;
byte_buffer& wr_buf() override;
void flush(io::network::newb_base* parent) override;
void flush(io::newb_base* parent) override;
expected<io::network::native_socket>
connect(const std::string& host, uint16_t port,
......@@ -68,18 +70,18 @@ struct udp_transport : public io::network::transport_policy {
io::network::ip_endpoint sender;
};
struct accept_udp : public io::network::accept_policy {
struct accept_udp : public accept {
expected<io::network::native_socket>
create_socket(uint16_t port, const char* host, bool reuse = false) override;
std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept(io::network::newb_base*) override;
std::pair<io::network::native_socket, transport_ptr>
accept_event(io::newb_base*) override;
void init(io::network::newb_base& n) override;
void init(io::newb_base& n) override;
};
template <class T>
using udp_protocol = io::network::generic_protocol<T>;
using udp_protocol = generic_protocol<T>;
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#pragma once
#include "caf/config.hpp"
#include "caf/callback.hpp"
#include "caf/error.hpp"
namespace caf {
namespace io {
struct newb_base;
template <class T>
struct newb;
} // namespace io
namespace policy {
using byte_buffer = std::vector<char>;
using header_writer = caf::callback<byte_buffer&>;
struct protocol_base {
virtual ~protocol_base();
virtual error read(char* bytes, size_t count) = 0;
virtual error timeout(atom_value, uint32_t) = 0;
virtual void write_header(byte_buffer&, header_writer*) = 0;
virtual void prepare_for_sending(byte_buffer&, size_t, size_t, size_t) = 0;
};
template <class T>
struct protocol : protocol_base {
using message_type = T;
virtual ~protocol() override {
// nop
}
virtual void init(io::newb<message_type>* parent) = 0;
};
template <class T>
using protocol_ptr = std::unique_ptr<protocol<T>>;
template <class T>
struct generic_protocol
: public protocol<typename T::message_type> {
void init(io::newb<typename T::message_type>* parent) override {
impl.init(parent);
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(byte_buffer& buf, header_writer* hw) override {
impl.write_header(buf, hw);
}
void prepare_for_sending(byte_buffer& buf, size_t hstart,
size_t offset, size_t plen) override {
impl.prepare_for_sending(buf, hstart, offset, plen);
}
private:
T impl;
};
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#pragma once
#include "caf/config.hpp"
#include "caf/error.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/rw_state.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/logger.hpp"
#include "caf/policy/protocol.hpp"
#include "caf/sec.hpp"
namespace caf {
namespace io {
struct newb_base;
} // namespace io
namespace policy {
struct transport {
transport();
virtual ~transport();
virtual io::network::rw_state write_some(io::newb_base*);
virtual io::network::rw_state read_some(io::newb_base*);
virtual bool should_deliver();
virtual bool must_read_more(io::newb_base*);
virtual void prepare_next_read(io::newb_base*);
virtual void prepare_next_write(io::newb_base*);
virtual void configure_read(io::receive_policy::config);
virtual void flush(io::newb_base*);
virtual byte_buffer& wr_buf();
template <class T>
error read_some(io::newb_base* parent, protocol<T>& policy) {
using io::network::rw_state;
CAF_LOG_TRACE("");
size_t reads = 0;
while (reads < max_consecutive_reads || must_read_more(parent)) {
auto read_result = read_some(parent);
switch (read_result) {
case rw_state::success:
if (received_bytes == 0)
return none;
if (should_deliver()) {
auto res = policy.read(receive_buffer.data(), received_bytes);
prepare_next_read(parent);
if (res)
return res;
}
break;
case rw_state::indeterminate:
// No error, but don't continue reading.
return none;
case rw_state::failure:
// Reading failed.
return sec::runtime_error;
}
++reads;
}
return none;
}
virtual expected<io::network::native_socket>
connect(const std::string&, uint16_t,
optional<io::network::protocol::network> = none);
size_t received_bytes;
size_t max_consecutive_reads;
byte_buffer offline_buffer;
byte_buffer receive_buffer;
byte_buffer send_buffer;
};
using transport_ptr = std::unique_ptr<transport>;
} // namespace policy
} // namespace caf
......@@ -16,85 +16,16 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/network/newb.hpp"
#include "caf/io/newb.hpp"
namespace caf {
namespace io {
namespace network {
// -- newb_base ----------------------------------------------------------------
newb_base::newb_base(default_multiplexer& dm, native_socket sockfd)
newb_base::newb_base(network::default_multiplexer& dm,
network::native_socket sockfd)
: event_handler(dm, sockfd) {
// nop
}
// -- transport_policy ---------------------------------------------------------
transport_policy::transport_policy()
: received_bytes{0},
max_consecutive_reads{50} {
// nop
}
transport_policy::~transport_policy() {
// nop
}
io::network::rw_state transport_policy::write_some(newb_base*) {
return io::network::rw_state::indeterminate;
}
io::network::rw_state transport_policy::read_some(newb_base*) {
return io::network::rw_state::indeterminate;
}
bool transport_policy::should_deliver() {
return true;
}
bool transport_policy::must_read_more(newb_base*) {
return false;
}
void transport_policy::prepare_next_read(newb_base*) {
// nop
}
void transport_policy::prepare_next_write(newb_base*) {
// nop
}
void transport_policy::configure_read(receive_policy::config) {
// nop
}
void transport_policy::flush(newb_base*) {
// nop
}
byte_buffer& transport_policy::wr_buf() {
return offline_buffer;
}
expected<native_socket>
transport_policy::connect(const std::string&, uint16_t,
optional<io::network::protocol::network>) {
return sec::bad_function_call;
}
// -- accept_policy ------------------------------------------------------------
accept_policy::~accept_policy() {
// nop
}
// -- protocol_policy_base -----------------------------------------------------
protocol_policy_base::~protocol_policy_base() {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
......@@ -17,6 +17,7 @@
******************************************************************************/
#include "caf/policy/newb_tcp.hpp"
#include "caf/io/newb.hpp"
#include "caf/config.hpp"
......@@ -71,7 +72,7 @@ tcp_transport::tcp_transport()
configure_read(io::receive_policy::at_most(1024));
}
io::network::rw_state tcp_transport::read_some(io::network::newb_base* parent) {
io::network::rw_state tcp_transport::read_some(io::newb_base* parent) {
CAF_LOG_TRACE("");
size_t len = receive_buffer.size() - collected;
void* buf = receive_buffer.data() + collected;
......@@ -100,7 +101,7 @@ bool tcp_transport::should_deliver() {
return collected >= read_threshold;
}
void tcp_transport::prepare_next_read(io::network::newb_base*) {
void tcp_transport::prepare_next_read(io::newb_base*) {
collected = 0;
received_bytes = 0;
switch (rd_flag) {
......@@ -130,7 +131,7 @@ void tcp_transport::configure_read(io::receive_policy::config config) {
maximum = config.second;
}
io::network::rw_state tcp_transport::write_some(io::network::newb_base* parent) {
io::network::rw_state tcp_transport::write_some(io::newb_base* parent) {
CAF_LOG_TRACE("");
const void* buf = send_buffer.data() + written;
auto len = send_buffer.size() - written;
......@@ -150,7 +151,7 @@ io::network::rw_state tcp_transport::write_some(io::network::newb_base* parent)
return io::network::rw_state::success;
}
void tcp_transport::prepare_next_write(io::network::newb_base* parent) {
void tcp_transport::prepare_next_write(io::newb_base* parent) {
written = 0;
send_buffer.clear();
if (offline_buffer.empty()) {
......@@ -161,7 +162,7 @@ void tcp_transport::prepare_next_write(io::network::newb_base* parent) {
}
}
void tcp_transport::flush(io::network::newb_base* parent) {
void tcp_transport::flush(io::newb_base* parent) {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
......@@ -182,8 +183,8 @@ accept_tcp::create_socket(uint16_t port, const char* host, bool reuse) {
return io::network::new_tcp_acceptor_impl(port, host, reuse);
}
std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept_tcp::accept(io::network::newb_base* parent) {
std::pair<io::network::native_socket, transport_ptr>
accept_tcp::accept_event(io::newb_base* parent) {
using namespace io::network;
sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr));
......@@ -196,11 +197,11 @@ accept_tcp::accept(io::network::newb_base* parent) {
return {invalid_native_socket, nullptr};
}
}
transport_policy_ptr ptr{new tcp_transport};
transport_ptr ptr{new tcp_transport};
return {result, std::move(ptr)};
}
void accept_tcp::init(io::network::newb_base& n) {
void accept_tcp::init(io::newb_base& n) {
n.start();
}
......
......@@ -20,6 +20,9 @@
#include "caf/config.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/newb.hpp"
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
......@@ -71,7 +74,7 @@ udp_transport::udp_transport()
// nop
}
io::network::rw_state udp_transport::read_some(io::network::newb_base* parent) {
io::network::rw_state udp_transport::read_some(io::newb_base* parent) {
CAF_LOG_TRACE(CAF_ARG(parent->fd()));
memset(sender.address(), 0, sizeof(sockaddr_storage));
io::network::socket_size_type len = sizeof(sockaddr_storage);
......@@ -97,12 +100,12 @@ io::network::rw_state udp_transport::read_some(io::network::newb_base* parent) {
return io::network::rw_state::success;
}
void udp_transport::prepare_next_read(io::network::newb_base*) {
void udp_transport::prepare_next_read(io::newb_base*) {
received_bytes = 0;
receive_buffer.resize(maximum);
}
io::network::rw_state udp_transport::write_some(io::network::newb_base* parent) {
io::network::rw_state udp_transport::write_some(io::newb_base* parent) {
using namespace caf::io::network;
CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size()));
socket_size_type len = static_cast<socket_size_type>(*endpoint.clength());
......@@ -128,7 +131,7 @@ io::network::rw_state udp_transport::write_some(io::network::newb_base* parent)
return io::network::rw_state::success;
}
void udp_transport::prepare_next_write(io::network::newb_base* parent) {
void udp_transport::prepare_next_write(io::newb_base* parent) {
written = 0;
send_buffer.clear();
send_sizes.clear();
......@@ -146,7 +149,7 @@ void udp_transport::prepare_next_write(io::network::newb_base* parent) {
}
}
io::network::byte_buffer& udp_transport::wr_buf() {
io::byte_buffer& udp_transport::wr_buf() {
if (!offline_buffer.empty()) {
auto chunk_size = offline_buffer.size() - offline_sum;
offline_sizes.push_back(chunk_size);
......@@ -155,7 +158,7 @@ io::network::byte_buffer& udp_transport::wr_buf() {
return offline_buffer;
}
void udp_transport::flush(io::network::newb_base* parent) {
void udp_transport::flush(io::newb_base* parent) {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
......@@ -183,19 +186,19 @@ accept_udp::create_socket(uint16_t port, const char* host, bool reuse) {
return (*res).first;
}
std::pair<io::network::native_socket, io::network::transport_policy_ptr>
accept_udp::accept(io::network::newb_base*) {
std::pair<io::network::native_socket, transport_ptr>
accept_udp::accept_event(io::newb_base*) {
auto res = io::network::new_local_udp_endpoint_impl(0, nullptr);
if (!res) {
CAF_LOG_DEBUG("failed to create local endpoint");
return {io::network::invalid_native_socket, nullptr};
}
auto sock = std::move(res->first);
io::network::transport_policy_ptr ptr{new udp_transport};
transport_ptr ptr{new udp_transport};
return {sock, std::move(ptr)};
}
void accept_udp::init(io::network::newb_base& n) {
void accept_udp::init(io::newb_base& n) {
n.start();
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#include "caf/policy/accept.hpp"
#include "caf/policy/protocol.hpp"
#include "caf/policy/transport.hpp"
namespace caf {
namespace policy {
// -- transport_policy ---------------------------------------------------------
transport::transport()
: received_bytes{0},
max_consecutive_reads{50} {
// nop
}
transport::~transport() {
// nop
}
io::network::rw_state transport::write_some(io::newb_base*) {
return io::network::rw_state::indeterminate;
}
io::network::rw_state transport::read_some(io::newb_base*) {
return io::network::rw_state::indeterminate;
}
bool transport::should_deliver() {
return true;
}
bool transport::must_read_more(io::newb_base*) {
return false;
}
void transport::prepare_next_read(io::newb_base*) {
// nop
}
void transport::prepare_next_write(io::newb_base*) {
// nop
}
void transport::configure_read(io::receive_policy::config) {
// nop
}
void transport::flush(io::newb_base*) {
// nop
}
byte_buffer& transport::wr_buf() {
return offline_buffer;
}
expected<io::network::native_socket>
transport::connect(const std::string&, uint16_t,
optional<io::network::protocol::network>) {
return sec::bad_function_call;
}
// -- accept_policy ------------------------------------------------------------
accept::~accept() {
// nop
}
// -- protocol_policy_base -----------------------------------------------------
protocol_base::~protocol_base() {
// nop
}
} // namespace policy
} // namespace caf
......@@ -26,7 +26,7 @@
#include "caf/test/dsl.hpp"
#include "caf/io/network/newb.hpp"
#include "caf/io/newb.hpp"
#include "caf/policy/newb_basp.hpp"
#include "caf/policy/newb_ordering.hpp"
#include "caf/policy/newb_raw.hpp"
......@@ -35,12 +35,12 @@ using namespace caf;
using namespace caf::io;
using namespace caf::policy;
using io::network::byte_buffer;
using io::network::header_writer;
using network::default_multiplexer;
using network::invalid_native_socket;
using network::last_socket_error_as_string;
using network::native_socket;
using io::byte_buffer;
using io::header_writer;
using io::network::default_multiplexer;
using io::network::invalid_native_socket;
using io::network::last_socket_error_as_string;
using io::network::native_socket;
namespace {
......@@ -55,6 +55,11 @@ using quit_atom = atom_constant<atom("quit")>;
using set_atom = atom_constant<atom("set")>;
using get_atom = atom_constant<atom("get")>;
using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>;
// -- test classes -------------------------------------------------------------
struct test_state {
int value;
std::vector<std::pair<atom_value, uint32_t>> timeout_messages;
......@@ -62,7 +67,7 @@ struct test_state {
std::deque<std::pair<basp_header, uint32_t>> expected;
};
behavior dummy_broker(network::stateful_newb<new_basp_msg, test_state>* self) {
behavior dummy_broker(stateful_newb<new_basp_msg, test_state>* self) {
return {
[=](new_basp_msg& msg) {
auto& s = self->state;
......@@ -78,7 +83,7 @@ behavior dummy_broker(network::stateful_newb<new_basp_msg, test_state>* self) {
std::vector<char> payload{msg.payload, msg.payload + msg.payload_len};
s.messages.emplace_back(msg, payload);
s.messages.back().first.payload = s.messages.back().second.data();
self->transport->receive_buffer.clear();
self->trans->receive_buffer.clear();
s.expected.pop_front();
},
[=](send_atom, actor_id sender, actor_id receiver, uint32_t payload) {
......@@ -96,14 +101,14 @@ behavior dummy_broker(network::stateful_newb<new_basp_msg, test_state>* self) {
binary_serializer bs(&self->backend(), *whdl.buf);
bs(payload);
}
std::swap(self->transport->receive_buffer, self->transport->offline_buffer);
self->transport->send_buffer.clear();
self->transport->received_bytes = self->transport->receive_buffer.size();
std::swap(self->trans->receive_buffer, self->trans->offline_buffer);
self->trans->send_buffer.clear();
self->trans->received_bytes = self->trans->receive_buffer.size();
},
[=](send_atom, ordering_header& ohdr, basp_header& bhdr, uint32_t payload) {
CAF_MESSAGE("send: ohdr = " << to_string(ohdr) << " bhdr = "
<< to_string(bhdr) << " payload = " << payload);
auto& buf = self->transport->receive_buffer;
auto& buf = self->trans->receive_buffer;
binary_serializer bs(&self->backend(), buf);
bs(ohdr);
auto bhdr_start = buf.size();
......@@ -111,7 +116,7 @@ behavior dummy_broker(network::stateful_newb<new_basp_msg, test_state>* self) {
auto payload_start = buf.size();
bs(payload);
auto packet_len = buf.size();
self->transport->received_bytes = packet_len;
self->trans->received_bytes = packet_len;
stream_serializer<charbuf> out{&self->backend(),
buf.data() + bhdr_start,
sizeof(uint32_t)};
......@@ -124,69 +129,33 @@ behavior dummy_broker(network::stateful_newb<new_basp_msg, test_state>* self) {
};
}
struct dummy_transport : public network::transport_policy {
io::network::rw_state read_some(network::newb_base*) {
return receive_buffer.size() > 0 ? network::rw_state::success : network::rw_state::indeterminate;
struct dummy_transport : public transport {
io::network::rw_state read_some(newb_base*) {
return receive_buffer.size() > 0 ? network::rw_state::success
: network::rw_state::indeterminate;
//return io::network::rw_state::success;
}
};
struct accept_policy_impl : public network::accept_policy {
struct accept_policy_impl : public accept {
expected<native_socket> create_socket(uint16_t, const char*, bool) override {
return sec::bad_function_call;
}
std::pair<native_socket, network::transport_policy_ptr>
accept(network::newb_base*) override {
std::pair<native_socket, transport_ptr>
accept_event(newb_base*) override {
auto esock = network::new_local_udp_endpoint_impl(0, nullptr);
CAF_REQUIRE(esock);
network::transport_policy_ptr ptr{new dummy_transport};
transport_ptr ptr{new dummy_transport};
return {esock->first, std::move(ptr)};
}
void init(network::newb_base& n) override {
void init(newb_base& n) override {
n.handle_event(network::operation::read);
}
};
/*
template <class Protocol>
struct dummy_basp_newb_acceptor
: public network::newb_acceptor<typename Protocol::message_type> {
using super = network::newb_acceptor<typename Protocol::message_type>;
using message_tuple_t = std::tuple<ordering_header, basp_header, int>;
dummy_basp_newb_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
expected<actor> create_newb(native_socket sockfd,
network::transport_policy_ptr pol) override {
spawned.emplace_back(
network::spawn_newb<Protocol>(this->backend().system(), dummy_broker,
std::move(pol), sockfd)
);
auto ptr = caf::actor_cast<caf::abstract_actor*>(spawned.back());
if (ptr == nullptr)
return sec::runtime_error;
// auto& ref
// = dynamic_cast<network::stateful_newb<new_basp_msg, test_state>&>(*ptr);
// // TODO: Call read_some using the buffer of the ref as a destination.
// binary_serializer bs(&this->backend(), ref.transport->receive_buffer);
// bs(get<0>(msg));
// bs(get<1>(msg));
// bs(get<2>(msg));
// ref.transport->received_bytes = ref.transport->receive_buffer.size();
// ref.state.expected.emplace_back(get<1>(msg), get<2>(msg));
return spawned.back();
}
message_tuple_t msg;
std::vector<actor> spawned;
};
*/
// -- config for controlled scheduling and multiplexing ------------------------
class config : public actor_system_config {
public:
......@@ -200,8 +169,8 @@ public:
};
struct fixture {
using newb_t = network::stateful_newb<new_basp_msg, test_state>;
using protocol_t = network::generic_protocol<ordering<datagram_basp>>;
using newb_t = stateful_newb<new_basp_msg, test_state>;
using protocol_t = generic_protocol<ordering<datagram_basp>>;
config cfg;
actor_system sys;
default_multiplexer& mpx;
......@@ -216,9 +185,9 @@ struct fixture {
auto esock = network::new_local_udp_endpoint_impl(0, nullptr);
CAF_REQUIRE(esock);
// Create newb.
network::transport_policy_ptr pol{new dummy_transport};
test_newb = network::spawn_newb<protocol_t>(sys, dummy_broker,
std::move(pol), esock->first);
transport_ptr transport{new dummy_transport};
test_newb = spawn_newb<protocol_t>(sys, dummy_broker, std::move(transport),
esock->first);
}
~fixture() {
......@@ -276,9 +245,9 @@ CAF_TEST(spawn newb) {
CAF_MESSAGE("create newb");
auto esock = network::new_local_udp_endpoint_impl(0, nullptr);
CAF_REQUIRE(esock);
network::transport_policy_ptr transport{new dummy_transport};
auto n = io::network::spawn_newb<protocol_t>(sys, my_newb, std::move(transport),
esock->first);
transport_ptr transport{new dummy_transport};
auto n = spawn_newb<protocol_t>(sys, my_newb, std::move(transport),
esock->first);
exec_all();
CAF_MESSAGE("send test message");
self->send(n, 3);
......@@ -306,9 +275,9 @@ CAF_TEST(spawn stateful newb) {
CAF_MESSAGE("create newb");
auto esock = network::new_local_udp_endpoint_impl(0, nullptr);
CAF_REQUIRE(esock);
network::transport_policy_ptr transport{new dummy_transport};
auto n = io::network::spawn_newb<protocol_t>(sys, my_newb, std::move(transport),
esock->first);
transport_ptr transport{new dummy_transport};
auto n = spawn_newb<protocol_t>(sys, my_newb, std::move(transport),
esock->first);
exec_all();
CAF_MESSAGE("set value in state");
self->send(n, set_atom::value, 3);
......@@ -342,9 +311,9 @@ CAF_TEST(read event) {
exec_all();
CAF_MESSAGE("copy them into the buffer");
auto& dummy = deref<newb_t>(test_newb);
auto& buf = dummy.transport->receive_buffer;
auto& buf = dummy.trans->receive_buffer;
write_packet(buf, ohdr, bhdr, payload);
dummy.transport->received_bytes = buf.size();
dummy.trans->received_bytes = buf.size();
CAF_MESSAGE("trigger a read event");
dummy.read_event();
CAF_MESSAGE("check the basp header and payload");
......@@ -417,15 +386,15 @@ CAF_TEST(message ordering) {
anon_send(test_newb, expect_atom::value, bhdr_second, payload_second);
exec_all();
auto& dummy = deref<newb_t>(test_newb);
auto& buf = dummy.transport->receive_buffer;
auto& buf = dummy.trans->receive_buffer;
CAF_MESSAGE("read second message first");
write_packet(buf, ohdr_second, bhdr_second, payload_second);
dummy.transport->received_bytes = buf.size();
dummy.trans->received_bytes = buf.size();
dummy.read_event();
CAF_MESSAGE("followed by first message");
buf.clear();
write_packet(buf, ohdr_first, bhdr_first, payload_first);
dummy.transport->received_bytes = buf.size();
dummy.trans->received_bytes = buf.size();
dummy.read_event();
}
......@@ -442,16 +411,4 @@ CAF_TEST(write buf) {
// Message handler will check if the expected message was received.
}
/*
CAF_TEST(newb acceptor) {
CAF_MESSAGE("trigger read event on acceptor");
na->handle_event(network::operation::read);
auto& dummy = dynamic_cast<acceptor_t&>(*na.get());
CAF_CHECK(!dummy.spawned.empty());
CAF_MESSAGE("shutdown the create newb");
for (actor& d : dummy.spawned)
anon_send_exit(d, exit_reason::user_shutdown);
}
*/
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