Commit 6494caaf authored by Joseph Noir's avatar Joseph Noir

Extract actorized event handler

parent 4ef9215a
...@@ -47,6 +47,7 @@ set(LIBCAF_IO_SRCS ...@@ -47,6 +47,7 @@ set(LIBCAF_IO_SRCS
src/udp.cpp src/udp.cpp
src/native_socket.cpp src/native_socket.cpp
src/socket_guard.cpp src/socket_guard.cpp
src/newb_base.cpp
src/newb_tcp.cpp src/newb_tcp.cpp
src/newb_udp.cpp src/newb_udp.cpp
src/newb.cpp src/newb.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/actor_marker.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
namespace caf {
namespace io {
namespace network {
// -- forward declarations -----------------------------------------------------
struct newb_base;
} // namespace network
} // namespace io
// -- required to make newb an actor ------------------------------------------
template <>
class behavior_type_of<io::network::newb_base> {
public:
using type = behavior;
};
namespace io {
namespace network {
// -- newb base ----------------------------------------------------------------
struct newb_base : public extend<scheduled_actor, newb_base>::template
with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
public dynamically_typed_actor_base,
public network::event_handler {
using signatures = none_t;
using super = typename extend<scheduled_actor, newb_base>::
template with<mixin::sender, mixin::requester, mixin::behavior_changer>;
// -- constructors and destructors -------------------------------------------
newb_base(actor_config& cfg, network::default_multiplexer& dm,
network::native_socket sockfd);
// -- overridden modifiers of abstract_actor ---------------------------------
void enqueue(mailbox_element_ptr ptr, execution_unit*) override;
void enqueue(strong_actor_ptr src, message_id mid, message msg,
execution_unit*) override;
resumable::subtype_t subtype() const override;
const char* name() const override;
// -- overridden modifiers of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override;
void initialize() override;
bool cleanup(error&& reason, execution_unit* host) override;
// -- overridden modifiers of resumable --------------------------------------
network::multiplexer::runnable::resume_result resume(execution_unit* ctx,
size_t mt) override;
// -- requirements -----------------------------------------------------------
virtual void start() = 0;
virtual void stop() = 0;
virtual void io_error(network::operation op, error err) = 0;
virtual void start_reading() = 0;
virtual void stop_reading() = 0;
virtual void start_writing() = 0;
virtual void stop_writing() = 0;
// -- members ----------------------------------------------------------------
void init_newb();
/// Override this to set the behavior of the broker.
virtual behavior make_behavior();
};
} // namespace network
} // namespace io
} // namespace caf
...@@ -25,7 +25,6 @@ ...@@ -25,7 +25,6 @@
#include <tuple> #include <tuple>
#include "caf/actor_clock.hpp" #include "caf/actor_clock.hpp"
#include "caf/config.hpp"
#include "caf/detail/apply_args.hpp" #include "caf/detail/apply_args.hpp"
#include "caf/detail/call_cfun.hpp" #include "caf/detail/call_cfun.hpp"
#include "caf/detail/enum_to_string.hpp" #include "caf/detail/enum_to_string.hpp"
...@@ -36,6 +35,7 @@ ...@@ -36,6 +35,7 @@
#include "caf/io/network/event_handler.hpp" #include "caf/io/network/event_handler.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/newb_base.hpp"
#include "caf/io/network/rw_state.hpp" #include "caf/io/network/rw_state.hpp"
#include "caf/mixin/behavior_changer.hpp" #include "caf/mixin/behavior_changer.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
...@@ -56,44 +56,11 @@ struct protocol_policy; ...@@ -56,44 +56,11 @@ struct protocol_policy;
template <class T> template <class T>
struct newb; struct newb;
} // namespace io
// -- required to make newb an actor ------------------------------------------
template <class T>
class behavior_type_of<io::newb<T>> {
public:
using type = behavior;
};
namespace io {
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
using byte_buffer = caf::policy::byte_buffer; using byte_buffer = caf::policy::byte_buffer;
using header_writer = caf::policy::header_writer; using header_writer = caf::policy::header_writer;
// -- newb base ----------------------------------------------------------------
struct newb_base : public network::event_handler {
newb_base(network::default_multiplexer& dm, network::native_socket sockfd);
virtual void start() = 0;
virtual void stop() = 0;
virtual void io_error(network::operation op, error err) = 0;
virtual void start_reading() = 0;
virtual void stop_reading() = 0;
virtual void start_writing() = 0;
virtual void stop_writing() = 0;
};
// -- new broker classes ------------------------------------------------------- // -- new broker classes -------------------------------------------------------
/// @relates newb /// @relates newb
...@@ -115,16 +82,7 @@ struct write_handle { ...@@ -115,16 +82,7 @@ struct write_handle {
}; };
template <class Message> template <class Message>
struct newb : public extend<scheduled_actor, newb<Message>>::template struct newb : public network::newb_base {
with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
public dynamically_typed_actor_base,
public newb_base {
using super = typename extend<scheduled_actor, newb<Message>>::
template with<mixin::sender, mixin::requester, mixin::behavior_changer>;
using signatures = none_t;
using message_type = Message; using message_type = Message;
// -- constructors and destructors ------------------------------------------- // -- constructors and destructors -------------------------------------------
...@@ -133,8 +91,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -133,8 +91,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
network::native_socket sockfd, network::native_socket sockfd,
policy::transport_ptr transport, policy::transport_ptr transport,
policy::protocol_ptr<Message> protocol) policy::protocol_ptr<Message> protocol)
: super(cfg), : newb_base(cfg, dm, sockfd),
newb_base(dm, sockfd),
trans(std::move(transport)), trans(std::move(transport)),
proto(std::move(protocol)), proto(std::move(protocol)),
value_(strong_actor_ptr{}, make_message_id(), value_(strong_actor_ptr{}, make_message_id(),
...@@ -142,7 +99,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -142,7 +99,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
reading_(false), reading_(false),
writing_(false) { writing_(false) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
scheduled_actor::set_timeout_handler([&](timeout_msg& msg) { this->scheduled_actor::set_timeout_handler([&](timeout_msg& msg) {
proto->timeout(msg.type, msg.timeout_id); proto->timeout(msg.type, msg.timeout_id);
}); });
proto->init(this); proto->init(this);
...@@ -156,73 +113,6 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -156,73 +113,6 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
} }
// -- overridden modifiers of abstract_actor ---------------------------------
void enqueue(mailbox_element_ptr ptr, execution_unit*) override {
CAF_PUSH_AID(this->id());
scheduled_actor::enqueue(std::move(ptr), &backend());
}
void enqueue(strong_actor_ptr src, message_id mid, message msg,
execution_unit*) override {
enqueue(make_mailbox_element(std::move(src), mid, {}, std::move(msg)),
&backend());
}
resumable::subtype_t subtype() const override {
return resumable::io_actor;
}
const char* name() const override {
return "newb";
}
// -- overridden modifiers of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_ASSERT(eu != nullptr);
CAF_ASSERT(eu == &backend());
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
// Add implicit reference count held by middleman/multiplexer.
if (!hide)
super::register_at_system();
if (lazy && super::mailbox().try_block())
return;
intrusive_ptr_add_ref(super::ctrl());
eu->exec_later(this);
}
void initialize() override {
CAF_LOG_TRACE("");
init_newb();
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(this->has_behavior()));
if (bhvr) {
// make_behavior() did return a behavior instead of using become().
CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
this->become(std::move(bhvr));
}
}
bool cleanup(error&& reason, execution_unit* host) override {
CAF_LOG_TRACE(CAF_ARG(reason));
stop();
// TODO: Should policies be notified here?
return local_actor::cleanup(std::move(reason), host);
}
// -- overridden modifiers of resumable --------------------------------------
network::multiplexer::runnable::resume_result resume(execution_unit* ctx,
size_t mt) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(ctx == &backend());
return scheduled_actor::resume(ctx, mt);
}
// -- overridden modifiers of event handler ---------------------------------- // -- overridden modifiers of event handler ----------------------------------
void handle_event(network::operation op) override { void handle_event(network::operation op) override {
...@@ -255,7 +145,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -255,7 +145,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
} }
// Event handler reference no longer necessary. // Event handler reference no longer necessary.
if (!reading_ && !writing_) if (!reading_ && !writing_)
intrusive_ptr_release(super::ctrl()); intrusive_ptr_release(this->ctrl());
} }
// -- base requirements ------------------------------------------------------ // -- base requirements ------------------------------------------------------
...@@ -266,7 +156,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -266,7 +156,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
// This is our own reference used to manage the lifetime matching // This is our own reference used to manage the lifetime matching
// as an event handler. // as an event handler.
if (!reading_ && !writing_) if (!reading_ && !writing_)
intrusive_ptr_add_ref(super::ctrl()); intrusive_ptr_add_ref(this->ctrl());
start_reading(); start_reading();
if (trans) if (trans)
trans->prepare_next_read(this); trans->prepare_next_read(this);
...@@ -285,7 +175,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -285,7 +175,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
if (!this->getf(this->is_cleaned_up_flag)) { if (!this->getf(this->is_cleaned_up_flag)) {
auto mptr = make_mailbox_element(nullptr, invalid_message_id, {}, auto mptr = make_mailbox_element(nullptr, invalid_message_id, {},
io_error_msg{op, std::move(err)}); io_error_msg{op, std::move(err)});
switch (scheduled_actor::consume(*mptr)) { switch (this->scheduled_actor::consume(*mptr)) {
case im_success: case im_success:
this->finalize(); this->finalize();
break; break;
...@@ -334,11 +224,6 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -334,11 +224,6 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
void init_newb() {
CAF_LOG_TRACE("");
super::setf(super::is_initialized_flag);
}
/// Get a write buffer to write data to be sent by this broker. /// Get a write buffer to write data to be sent by this broker.
write_handle<Message> wr_buf(header_writer* hw) { write_handle<Message> wr_buf(header_writer* hw) {
auto& buf = trans->wr_buf(); auto& buf = trans->wr_buf();
...@@ -375,9 +260,9 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -375,9 +260,9 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
template<class Rep = int, class Period = std::ratio<1>> template<class Rep = int, class Period = std::ratio<1>>
void set_timeout(std::chrono::duration<Rep, Period> timeout, void set_timeout(std::chrono::duration<Rep, Period> timeout,
atom_value atm, uint32_t id) { atom_value atm, uint32_t id) {
//auto n = actor_clock::clock_type::now(); auto n = actor_clock::clock_type::now();
//scheduled_actor::clock().set_ordinary_timeout(n + timeout, this, atm, id); scheduled_actor::clock().set_multi_timeout(n + timeout, this, atm, id);
this->delayed_send(this, timeout, atm, id); //this->delayed_send(this, timeout, atm, id);
} }
/// Returns the `multiplexer` running this broker. /// Returns the `multiplexer` running this broker.
...@@ -391,16 +276,6 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -391,16 +276,6 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
scheduled_actor::activate(scheduled_actor::context(), value_); scheduled_actor::activate(scheduled_actor::context(), value_);
} }
/// Override this to set the behavior of the broker.
virtual behavior make_behavior() {
behavior res;
if (this->initial_behavior_fac_) {
res = this->initial_behavior_fac_(this);
this->initial_behavior_fac_ = nullptr;
}
return res;
}
/// Configure the number of bytes read for the next packet. (Can be ignored by /// 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.) /// the transport policy if its protocol does not support this functionality.)
void configure_read(receive_policy::config config) { void configure_read(receive_policy::config config) {
...@@ -526,8 +401,27 @@ spawn_client(actor_system& sys, F fun, policy::transport_ptr transport, ...@@ -526,8 +401,27 @@ spawn_client(actor_system& sys, F fun, policy::transport_ptr transport,
// -- new broker acceptor ------------------------------------------------------ // -- new broker acceptor ------------------------------------------------------
struct acceptor_base : public network::event_handler {
acceptor_base(network::default_multiplexer& dm, network::native_socket sockfd);
virtual void start() = 0;
virtual void stop() = 0;
virtual void io_error(network::operation op, error err) = 0;
virtual void start_reading() = 0;
virtual void stop_reading() = 0;
virtual void start_writing() = 0;
virtual void stop_writing() = 0;
};
template <class Protocol, class Fun, class... Ts> template <class Protocol, class Fun, class... Ts>
struct newb_acceptor : public newb_base, public caf::ref_counted { struct newb_acceptor : public acceptor_base, public caf::ref_counted {
using newb_type = typename std::remove_pointer<first_argument_type<Fun>>::type; using newb_type = typename std::remove_pointer<first_argument_type<Fun>>::type;
using message_type = typename newb_type::message_type; using message_type = typename newb_type::message_type;
...@@ -535,7 +429,7 @@ struct newb_acceptor : public newb_base, public caf::ref_counted { ...@@ -535,7 +429,7 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
newb_acceptor(network::default_multiplexer& dm, network::native_socket sockfd, newb_acceptor(network::default_multiplexer& dm, network::native_socket sockfd,
Fun f, policy::accept_ptr<message_type> pol, Ts&&... xs) Fun f, policy::accept_ptr<message_type> pol, Ts&&... xs)
: newb_base(dm, sockfd), : acceptor_base(dm, sockfd),
accept_pol(std::move(pol)), accept_pol(std::move(pol)),
fun_(std::move(f)), fun_(std::move(f)),
reading_(false), reading_(false),
......
...@@ -49,22 +49,22 @@ struct accept { ...@@ -49,22 +49,22 @@ struct accept {
create_socket(uint16_t port, const char* host, bool reuse = false) = 0; create_socket(uint16_t port, const char* host, bool reuse = false) = 0;
virtual std::pair<io::network::native_socket, transport_ptr> virtual std::pair<io::network::native_socket, transport_ptr>
accept_event(io::newb_base*) { accept_event(io::network::newb_base*) {
return {0, nullptr}; return {0, nullptr};
} }
/// If `requires_raw_data` is set to true, the acceptor will only call /// 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 /// this function for new read event and let the policy handle everything
/// else. /// else.
virtual void read_event(io::newb_base*) { virtual void read_event(io::network::newb_base*) {
// nop // nop
} }
virtual error write_event(io::newb_base*) { virtual error write_event(io::network::newb_base*) {
return none; return none;
} }
virtual void init(io::newb_base*, io::newb<Message>&) { virtual void init(io::network::newb_base*, io::newb<Message>&) {
// nop // nop
} }
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/newb_base.hpp"
#include "caf/policy/accept.hpp" #include "caf/policy/accept.hpp"
#include "caf/policy/transport.hpp" #include "caf/policy/transport.hpp"
...@@ -29,19 +30,19 @@ namespace policy { ...@@ -29,19 +30,19 @@ namespace policy {
struct tcp_transport : public transport { struct tcp_transport : public transport {
tcp_transport(); tcp_transport();
io::network::rw_state read_some(io::newb_base* parent) override; io::network::rw_state read_some(io::network::newb_base* parent) override;
bool should_deliver() override; bool should_deliver() override;
void prepare_next_read(io::newb_base*) override; void prepare_next_read(io::network::newb_base*) override;
void configure_read(io::receive_policy::config config) override; void configure_read(io::receive_policy::config config) override;
io::network::rw_state write_some(io::newb_base* parent) override; io::network::rw_state write_some(io::network::newb_base* parent) override;
void prepare_next_write(io::newb_base* parent) override; void prepare_next_write(io::network::newb_base* parent) override;
void flush(io::newb_base* parent) override; void flush(io::network::newb_base* parent) override;
expected<io::network::native_socket> expected<io::network::native_socket>
connect(const std::string& host, uint16_t port, connect(const std::string& host, uint16_t port,
...@@ -58,7 +59,7 @@ struct tcp_transport : public transport { ...@@ -58,7 +59,7 @@ struct tcp_transport : public transport {
size_t written; size_t written;
}; };
io::network::native_socket get_newb_socket(io::newb_base*); io::network::native_socket get_newb_socket(io::network::newb_base*);
template <class Message> template <class Message>
struct accept_tcp : public accept<Message> { struct accept_tcp : public accept<Message> {
...@@ -68,7 +69,7 @@ struct accept_tcp : public accept<Message> { ...@@ -68,7 +69,7 @@ struct accept_tcp : public accept<Message> {
} }
std::pair<io::network::native_socket, transport_ptr> std::pair<io::network::native_socket, transport_ptr>
accept_event(io::newb_base* parent) override { accept_event(io::network::newb_base* parent) override {
auto esock = io::network::accept_tcp_connection(get_newb_socket(parent)); auto esock = io::network::accept_tcp_connection(get_newb_socket(parent));
if (!esock) { if (!esock) {
return {io::network::invalid_native_socket, nullptr}; return {io::network::invalid_native_socket, nullptr};
...@@ -77,7 +78,7 @@ struct accept_tcp : public accept<Message> { ...@@ -77,7 +78,7 @@ struct accept_tcp : public accept<Message> {
return {*esock, std::move(ptr)}; return {*esock, std::move(ptr)};
} }
void init(io::newb_base*, io::newb<Message>& spawned) override { void init(io::network::newb_base*, io::newb<Message>& spawned) override {
spawned.start(); spawned.start();
} }
}; };
......
...@@ -30,26 +30,26 @@ namespace policy { ...@@ -30,26 +30,26 @@ namespace policy {
struct udp_transport : public transport { struct udp_transport : public transport {
udp_transport(); udp_transport();
io::network::rw_state read_some(io::newb_base* parent) override; io::network::rw_state read_some(io::network::newb_base* parent) override;
inline bool should_deliver() override { inline bool should_deliver() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return received_bytes != 0 && sender == endpoint; return received_bytes != 0 && sender == endpoint;
} }
void prepare_next_read(io::newb_base*) override; void prepare_next_read(io::network::newb_base*) override;
inline void configure_read(io::receive_policy::config) override { inline void configure_read(io::receive_policy::config) override {
// nop // nop
} }
io::network::rw_state write_some(io::newb_base* parent) override; io::network::rw_state write_some(io::network::newb_base* parent) override;
void prepare_next_write(io::newb_base* parent) override; void prepare_next_write(io::network::newb_base* parent) override;
byte_buffer& wr_buf() override; byte_buffer& wr_buf() override;
void flush(io::newb_base* parent) override; void flush(io::network::newb_base* parent) override;
expected<io::network::native_socket> expected<io::network::native_socket>
connect(const std::string& host, uint16_t port, connect(const std::string& host, uint16_t port,
...@@ -82,7 +82,7 @@ struct accept_udp : public accept<Message> { ...@@ -82,7 +82,7 @@ struct accept_udp : public accept<Message> {
} }
std::pair<io::network::native_socket, transport_ptr> std::pair<io::network::native_socket, transport_ptr>
accept_event(io::newb_base*) override { accept_event(io::network::newb_base*) override {
std::cout << "Accepting msg from UDP endpoint" << std::endl; std::cout << "Accepting msg from UDP endpoint" << std::endl;
auto res = io::network::new_local_udp_endpoint_impl(0, nullptr, true); auto res = io::network::new_local_udp_endpoint_impl(0, nullptr, true);
if (!res) { if (!res) {
...@@ -94,7 +94,8 @@ struct accept_udp : public accept<Message> { ...@@ -94,7 +94,8 @@ struct accept_udp : public accept<Message> {
return {sock, std::move(ptr)}; return {sock, std::move(ptr)};
} }
void init(io::newb_base* parent, io::newb<Message>& spawned) override { void init(io::network::newb_base* parent,
io::newb<Message>& spawned) override {
spawned.trans->prepare_next_read(parent); spawned.trans->prepare_next_read(parent);
spawned.trans->read_some(parent, *spawned.proto.get()); spawned.trans->read_some(parent, *spawned.proto.get());
spawned.start(); spawned.start();
......
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/newb_base.hpp"
#include "caf/io/network/protocol.hpp" #include "caf/io/network/protocol.hpp"
#include "caf/io/network/rw_state.hpp" #include "caf/io/network/rw_state.hpp"
#include "caf/io/receive_policy.hpp" #include "caf/io/receive_policy.hpp"
...@@ -43,26 +44,26 @@ struct transport { ...@@ -43,26 +44,26 @@ struct transport {
virtual ~transport(); virtual ~transport();
virtual io::network::rw_state write_some(io::newb_base*); virtual io::network::rw_state write_some(io::network::newb_base*);
virtual io::network::rw_state read_some(io::newb_base*); virtual io::network::rw_state read_some(io::network::newb_base*);
virtual bool should_deliver(); virtual bool should_deliver();
virtual bool must_read_more(io::newb_base*); virtual bool must_read_more(io::network::newb_base*);
virtual void prepare_next_read(io::newb_base*); virtual void prepare_next_read(io::network::newb_base*);
virtual void prepare_next_write(io::newb_base*); virtual void prepare_next_write(io::network::newb_base*);
virtual void configure_read(io::receive_policy::config); virtual void configure_read(io::receive_policy::config);
virtual void flush(io::newb_base*); virtual void flush(io::network::newb_base*);
virtual byte_buffer& wr_buf(); virtual byte_buffer& wr_buf();
template <class T> template <class T>
error read_some(io::newb_base* parent, protocol<T>& policy) { error read_some(io::network::newb_base* parent, protocol<T>& policy) {
using io::network::rw_state; using io::network::rw_state;
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
size_t reads = 0; size_t reads = 0;
......
...@@ -21,11 +21,7 @@ ...@@ -21,11 +21,7 @@
namespace caf { namespace caf {
namespace io { namespace io {
newb_base::newb_base(network::default_multiplexer& dm, // TODO: Delete this file.
network::native_socket sockfd)
: event_handler(dm, sockfd) {
// nop
}
} // namespace io } // namespace io
} // namespace caf } // 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. *
******************************************************************************/
#include "caf/io/network/newb_base.hpp"
namespace caf {
namespace io {
namespace network {
newb_base::newb_base(actor_config& cfg, network::default_multiplexer& dm,
network::native_socket sockfd)
: super(cfg),
event_handler(dm, sockfd) {
// nop
}
void newb_base::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_PUSH_AID(this->id());
scheduled_actor::enqueue(std::move(ptr), &backend());
}
void newb_base::enqueue(strong_actor_ptr src, message_id mid, message msg,
execution_unit*) {
enqueue(make_mailbox_element(std::move(src), mid, {}, std::move(msg)),
&backend());
}
resumable::subtype_t newb_base::subtype() const {
return resumable::io_actor;
}
const char* newb_base::name() const {
return "newb";
}
void newb_base::launch(execution_unit* eu, bool lazy, bool hide) {
CAF_PUSH_AID_FROM_PTR(this);
CAF_ASSERT(eu != nullptr);
CAF_ASSERT(eu == &backend());
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
// Add implicit reference count held by middleman/multiplexer.
if (!hide)
super::register_at_system();
if (lazy && super::mailbox().try_block())
return;
intrusive_ptr_add_ref(super::ctrl());
eu->exec_later(this);
}
void newb_base::initialize() {
CAF_LOG_TRACE("");
init_newb();
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(this->has_behavior()));
if (bhvr) {
// make_behavior() did return a behavior instead of using become().
CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
this->become(std::move(bhvr));
}
}
bool newb_base::cleanup(error&& reason, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason));
stop();
// TODO: Should policies be notified here?
return local_actor::cleanup(std::move(reason), host);
}
network::multiplexer::runnable::resume_result
newb_base::resume(execution_unit* ctx, size_t mt) {
CAF_PUSH_AID_FROM_PTR(this);
CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(ctx == &backend());
return scheduled_actor::resume(ctx, mt);
}
void newb_base::init_newb() {
CAF_LOG_TRACE("");
this->setf(super::is_initialized_flag);
}
behavior newb_base::make_behavior() {
behavior res;
if (this->initial_behavior_fac_) {
res = this->initial_behavior_fac_(this);
this->initial_behavior_fac_ = nullptr;
}
return res;
}
} // namespace network
} // namespace io
} // namespace caf
...@@ -72,7 +72,7 @@ tcp_transport::tcp_transport() ...@@ -72,7 +72,7 @@ tcp_transport::tcp_transport()
configure_read(io::receive_policy::at_most(1024)); configure_read(io::receive_policy::at_most(1024));
} }
io::network::rw_state tcp_transport::read_some(io::newb_base* parent) { io::network::rw_state tcp_transport::read_some(io::network::newb_base* parent) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
size_t len = receive_buffer.size() - collected; size_t len = receive_buffer.size() - collected;
void* buf = receive_buffer.data() + collected; void* buf = receive_buffer.data() + collected;
...@@ -101,7 +101,7 @@ bool tcp_transport::should_deliver() { ...@@ -101,7 +101,7 @@ bool tcp_transport::should_deliver() {
return collected >= read_threshold; return collected >= read_threshold;
} }
void tcp_transport::prepare_next_read(io::newb_base*) { void tcp_transport::prepare_next_read(io::network::newb_base*) {
collected = 0; collected = 0;
received_bytes = 0; received_bytes = 0;
switch (rd_flag) { switch (rd_flag) {
...@@ -131,7 +131,8 @@ void tcp_transport::configure_read(io::receive_policy::config config) { ...@@ -131,7 +131,8 @@ void tcp_transport::configure_read(io::receive_policy::config config) {
maximum = config.second; maximum = config.second;
} }
io::network::rw_state tcp_transport::write_some(io::newb_base* parent) { io::network::rw_state
tcp_transport::write_some(io::network::newb_base* parent) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
const void* buf = send_buffer.data() + written; const void* buf = send_buffer.data() + written;
auto len = send_buffer.size() - written; auto len = send_buffer.size() - written;
...@@ -151,7 +152,7 @@ io::network::rw_state tcp_transport::write_some(io::newb_base* parent) { ...@@ -151,7 +152,7 @@ io::network::rw_state tcp_transport::write_some(io::newb_base* parent) {
return io::network::rw_state::success; return io::network::rw_state::success;
} }
void tcp_transport::prepare_next_write(io::newb_base* parent) { void tcp_transport::prepare_next_write(io::network::newb_base* parent) {
written = 0; written = 0;
send_buffer.clear(); send_buffer.clear();
if (offline_buffer.empty()) { if (offline_buffer.empty()) {
...@@ -162,7 +163,7 @@ void tcp_transport::prepare_next_write(io::newb_base* parent) { ...@@ -162,7 +163,7 @@ void tcp_transport::prepare_next_write(io::newb_base* parent) {
} }
} }
void tcp_transport::flush(io::newb_base* parent) { void tcp_transport::flush(io::network::newb_base* parent) {
CAF_ASSERT(parent != nullptr); CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size())); CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) { if (!offline_buffer.empty() && !writing) {
...@@ -178,7 +179,7 @@ tcp_transport::connect(const std::string& host, uint16_t port, ...@@ -178,7 +179,7 @@ tcp_transport::connect(const std::string& host, uint16_t port,
return io::network::new_tcp_connection(host, port, preferred); return io::network::new_tcp_connection(host, port, preferred);
} }
io::network::native_socket get_newb_socket(io::newb_base* n) { io::network::native_socket get_newb_socket(io::network::newb_base* n) {
return n->fd(); return n->fd();
} }
......
...@@ -74,11 +74,12 @@ udp_transport::udp_transport() ...@@ -74,11 +74,12 @@ udp_transport::udp_transport()
// nop // nop
} }
io::network::rw_state udp_transport::read_some(io::newb_base* parent) { io::network::rw_state udp_transport::read_some(io::network::newb_base* parent) {
CAF_LOG_TRACE(CAF_ARG(parent->fd())); CAF_LOG_TRACE(CAF_ARG(parent->fd()));
memset(sender.address(), 0, sizeof(sockaddr_storage)); memset(sender.address(), 0, sizeof(sockaddr_storage));
io::network::socket_size_type len = sizeof(sockaddr_storage); io::network::socket_size_type len = sizeof(sockaddr_storage);
auto buf_ptr = static_cast<io::network::socket_recv_ptr>(receive_buffer.data()); auto buf_ptr =
static_cast<io::network::socket_recv_ptr>(receive_buffer.data());
auto buf_len = receive_buffer.size(); auto buf_len = receive_buffer.size();
auto sres = ::recvfrom(parent->fd(), buf_ptr, buf_len, auto sres = ::recvfrom(parent->fd(), buf_ptr, buf_len,
0, sender.address(), &len); 0, sender.address(), &len);
...@@ -100,16 +101,18 @@ io::network::rw_state udp_transport::read_some(io::newb_base* parent) { ...@@ -100,16 +101,18 @@ io::network::rw_state udp_transport::read_some(io::newb_base* parent) {
return io::network::rw_state::success; return io::network::rw_state::success;
} }
void udp_transport::prepare_next_read(io::newb_base*) { void udp_transport::prepare_next_read(io::network::newb_base*) {
received_bytes = 0; received_bytes = 0;
receive_buffer.resize(maximum); receive_buffer.resize(maximum);
} }
io::network::rw_state udp_transport::write_some(io::newb_base* parent) { io::network::rw_state
udp_transport::write_some(io::network::newb_base* parent) {
using namespace caf::io::network; using namespace caf::io::network;
CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size())); CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size()));
socket_size_type len = static_cast<socket_size_type>(*endpoint.clength()); socket_size_type len = static_cast<socket_size_type>(*endpoint.clength());
auto buf_ptr = reinterpret_cast<socket_send_ptr>(send_buffer.data() + written); auto buf_ptr =
reinterpret_cast<socket_send_ptr>(send_buffer.data() + written);
auto buf_len = send_sizes.front(); auto buf_len = send_sizes.front();
auto sres = ::sendto(parent->fd(), buf_ptr, buf_len, auto sres = ::sendto(parent->fd(), buf_ptr, buf_len,
0, endpoint.caddress(), len); 0, endpoint.caddress(), len);
...@@ -131,7 +134,7 @@ io::network::rw_state udp_transport::write_some(io::newb_base* parent) { ...@@ -131,7 +134,7 @@ io::network::rw_state udp_transport::write_some(io::newb_base* parent) {
return io::network::rw_state::success; return io::network::rw_state::success;
} }
void udp_transport::prepare_next_write(io::newb_base* parent) { void udp_transport::prepare_next_write(io::network::newb_base* parent) {
written = 0; written = 0;
send_buffer.clear(); send_buffer.clear();
send_sizes.clear(); send_sizes.clear();
...@@ -158,7 +161,7 @@ io::byte_buffer& udp_transport::wr_buf() { ...@@ -158,7 +161,7 @@ io::byte_buffer& udp_transport::wr_buf() {
return offline_buffer; return offline_buffer;
} }
void udp_transport::flush(io::newb_base* parent) { void udp_transport::flush(io::network::newb_base* parent) {
CAF_ASSERT(parent != nullptr); CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size())); CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) { if (!offline_buffer.empty() && !writing) {
......
...@@ -35,11 +35,11 @@ transport::~transport() { ...@@ -35,11 +35,11 @@ transport::~transport() {
// nop // nop
} }
io::network::rw_state transport::write_some(io::newb_base*) { io::network::rw_state transport::write_some(io::network::newb_base*) {
return io::network::rw_state::indeterminate; return io::network::rw_state::indeterminate;
} }
io::network::rw_state transport::read_some(io::newb_base*) { io::network::rw_state transport::read_some(io::network::newb_base*) {
return io::network::rw_state::indeterminate; return io::network::rw_state::indeterminate;
} }
...@@ -47,15 +47,15 @@ bool transport::should_deliver() { ...@@ -47,15 +47,15 @@ bool transport::should_deliver() {
return true; return true;
} }
bool transport::must_read_more(io::newb_base*) { bool transport::must_read_more(io::network::newb_base*) {
return false; return false;
} }
void transport::prepare_next_read(io::newb_base*) { void transport::prepare_next_read(io::network::newb_base*) {
// nop // nop
} }
void transport::prepare_next_write(io::newb_base*) { void transport::prepare_next_write(io::network::newb_base*) {
// nop // nop
} }
...@@ -63,7 +63,7 @@ void transport::configure_read(io::receive_policy::config) { ...@@ -63,7 +63,7 @@ void transport::configure_read(io::receive_policy::config) {
// nop // nop
} }
void transport::flush(io::newb_base*) { void transport::flush(io::network::newb_base*) {
// nop // nop
} }
......
...@@ -130,7 +130,7 @@ behavior dummy_broker(stateful_newb<new_basp_msg, test_state>* self) { ...@@ -130,7 +130,7 @@ behavior dummy_broker(stateful_newb<new_basp_msg, test_state>* self) {
} }
struct dummy_transport : public transport { struct dummy_transport : public transport {
io::network::rw_state read_some(newb_base*) { io::network::rw_state read_some(network::newb_base*) {
return receive_buffer.size() > 0 ? network::rw_state::success return receive_buffer.size() > 0 ? network::rw_state::success
: network::rw_state::indeterminate; : network::rw_state::indeterminate;
//return io::network::rw_state::success; //return io::network::rw_state::success;
......
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