Commit 84f43f6f authored by ufownl's avatar ufownl

Typed Brokers

parent e1c9577a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_MIXIN_TYPED_FUNCTOR_BASED_HPP
#define CAF_MIXIN_TYPED_FUNCTOR_BASED_HPP
namespace caf {
namespace mixin {
template <class Base, class Subtype>
class typed_functor_based : public Base {
public:
using combined_type = typed_functor_based;
using pointer = Base*;
using behavior_type = typename Base::behavior_type;
using make_behavior_fun = std::function<behavior_type(pointer)>;
using void_fun = std::function<void(pointer)>;
typed_functor_based() {
// nop
}
template <class F, class... Ts>
typed_functor_based(F f, Ts&&... xs) {
init(std::move(f), std::forward<Ts>(xs)...);
}
template <class F, class... Ts>
void init(F f, Ts&&... xs) {
using trait = typename detail::get_callable_trait<F>::type;
using arg_types = typename trait::arg_types;
using result_type = typename trait::result_type;
constexpr bool returns_behavior =
std::is_convertible<result_type, behavior_type>::value;
constexpr bool uses_first_arg = std::is_same<
typename detail::tl_head<arg_types>::type, pointer>::value;
std::integral_constant<bool, returns_behavior> token1;
std::integral_constant<bool, uses_first_arg> token2;
set(token1, token2, std::move(f), std::forward<Ts>(xs)...);
}
protected:
make_behavior_fun m_make_behavior;
private:
template <class F>
void set(std::true_type, std::true_type, F&& fun) {
// behavior_type (pointer)
m_make_behavior = std::forward<F>(fun);
}
template <class F>
void set(std::false_type, std::true_type, F fun) {
// void (pointer)
m_make_behavior = [fun](pointer ptr) {
fun(ptr);
return behavior_type{};
};
}
template <class F>
void set(std::true_type, std::false_type, F fun) {
// behavior_type (void)
m_make_behavior = [fun](pointer) { return fun(); };
}
template <class F>
void set(std::false_type, std::false_type, F fun) {
// void (void)
m_make_behavior = [fun](pointer) {
fun();
return behavior_type{};
};
}
template <class Token, typename F, typename T0, class... Ts>
void set(Token t1, std::true_type t2, F fun, T0&& x, Ts&&... xs) {
set(t1, t2, std::bind(fun, std::placeholders::_1, std::forward<T0>(x),
std::forward<Ts>(xs)...));
}
template <class Token, typename F, typename T0, class... Ts>
void set(Token t1, std::false_type t2, F fun, T0&& x, Ts&&... xs) {
set(t1, t2, std::bind(fun, std::forward<T0>(x), std::forward<Ts>(xs)...));
}
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_TYPED_FUNCTOR_BASED_HPP
...@@ -8,6 +8,7 @@ file(GLOB LIBCAF_IO_HDRS "caf/io/*.hpp" "caf/io/network/*.hpp") ...@@ -8,6 +8,7 @@ file(GLOB LIBCAF_IO_HDRS "caf/io/*.hpp" "caf/io/network/*.hpp")
# list cpp files excluding platform-dependent files # list cpp files excluding platform-dependent files
set (LIBCAF_IO_SRCS set (LIBCAF_IO_SRCS
src/basp_broker.cpp src/basp_broker.cpp
src/abstract_broker.cpp
src/broker.cpp src/broker.cpp
src/max_msg_size.cpp src/max_msg_size.cpp
src/middleman.cpp src/middleman.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_IO_ABSTRACT_BROKER_HPP
#define CAF_IO_ABSTRACT_BROKER_HPP
#include <map>
#include <vector>
#include "caf/detail/intrusive_partitioned_list.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
namespace caf {
namespace io {
class middleman;
class abstract_broker {
public:
using buffer_type = std::vector<char>;
/**
* Manages a low-level IO device for the `broker`.
*/
class servant {
public:
friend class abstract_broker;
void set_broker(abstract_broker* ptr);
virtual ~servant();
protected:
virtual void remove_from_broker() = 0;
virtual message disconnect_message() = 0;
inline abstract_broker* parent() {
return m_broker;
}
servant(abstract_broker* ptr);
void disconnect(bool invoke_disconnect_message);
bool m_disconnected;
abstract_broker* m_broker;
};
/**
* Manages a stream.
*/
class scribe : public network::stream_manager, public servant {
public:
scribe(abstract_broker* parent, connection_handle hdl);
~scribe();
/**
* Implicitly starts the read loop on first call.
*/
virtual void configure_read(receive_policy::config config) = 0;
/**
* Grants access to the output buffer.
*/
virtual buffer_type& wr_buf() = 0;
/**
* Flushes the output buffer, i.e., sends the content of
* the buffer via the network.
*/
virtual void flush() = 0;
inline connection_handle hdl() const {
return m_hdl;
}
void io_failure(network::operation op) override;
protected:
virtual buffer_type& rd_buf() = 0;
inline new_data_msg& read_msg() {
return m_read_msg.get_as_mutable<new_data_msg>(0);
}
inline const new_data_msg& read_msg() const {
return m_read_msg.get_as<new_data_msg>(0);
}
void remove_from_broker() override;
message disconnect_message() override;
void consume(const void* data, size_t num_bytes) override;
connection_handle m_hdl;
message m_read_msg;
};
using scribe_pointer = intrusive_ptr<scribe>;
/**
* Manages incoming connections.
*/
class doorman : public network::acceptor_manager, public servant {
public:
doorman(abstract_broker* parent, accept_handle hdl);
~doorman();
inline accept_handle hdl() const {
return m_hdl;
}
void io_failure(network::operation op) override;
// needs to be launched explicitly
virtual void launch() = 0;
protected:
void remove_from_broker() override;
message disconnect_message() override;
inline new_connection_msg& accept_msg() {
return m_accept_msg.get_as_mutable<new_connection_msg>(0);
}
inline const new_connection_msg& accept_msg() const {
return m_accept_msg.get_as<new_connection_msg>(0);
}
accept_handle m_hdl;
message m_accept_msg;
};
using doorman_pointer = intrusive_ptr<doorman>;
// a broker needs friends
friend class scribe;
friend class doorman;
friend class continuation;
virtual ~abstract_broker();
/**
* Modifies the receive policy for given connection.
* @param hdl Identifies the affected connection.
* @param config Contains the new receive policy.
*/
void configure_read(connection_handle hdl, receive_policy::config config);
/**
* Returns the write buffer for given connection.
*/
buffer_type& wr_buf(connection_handle hdl);
/**
* Writes `data` into the buffer for given connection.
*/
void write(connection_handle hdl, size_t data_size, const void* data);
/**
* Sends the content of the buffer for given connection.
*/
void flush(connection_handle hdl);
/**
* Returns the number of open connections.
*/
inline size_t num_connections() const {
return m_scribes.size();
}
std::vector<connection_handle> connections() const;
/** @cond PRIVATE */
virtual bool is_initialized() = 0;
inline void add_scribe(const scribe_pointer& ptr) {
m_scribes.emplace(ptr->hdl(), ptr);
}
connection_handle add_tcp_scribe(const std::string& host, uint16_t port);
void assign_tcp_scribe(connection_handle hdl);
connection_handle add_tcp_scribe(network::native_socket fd);
inline void add_doorman(const doorman_pointer& ptr) {
m_doormen.emplace(ptr->hdl(), ptr);
if (is_initialized()) {
ptr->launch();
}
}
std::pair<accept_handle, uint16_t> add_tcp_doorman(uint16_t port = 0,
const char* in = nullptr,
bool reuse_addr = false);
void assign_tcp_doorman(accept_handle hdl);
accept_handle add_tcp_doorman(network::native_socket fd);
virtual actor_addr address() const = 0;
virtual void invoke_message(mailbox_element_ptr& msg) = 0;
void invoke_message(const actor_addr& sender,
message_id mid, message& msg);
/**
* Closes all connections and acceptors.
*/
void close_all();
/**
* Closes the connection identified by `handle`.
* Unwritten data will still be send.
*/
void close(connection_handle handle);
/**
* Closes the acceptor identified by `handle`.
*/
void close(accept_handle handle);
/**
* Checks whether a connection for `handle` exists.
*/
bool valid(connection_handle handle);
/**
* Checks whether an acceptor for `handle` exists.
*/
bool valid(accept_handle handle);
// <backward_compatibility version="0.9">
static constexpr auto at_least = receive_policy_flag::at_least;
static constexpr auto at_most = receive_policy_flag::at_most;
static constexpr auto exactly = receive_policy_flag::exactly;
void receive_policy(connection_handle hdl, receive_policy_flag flag,
size_t num_bytes) CAF_DEPRECATED {
configure_read(hdl, receive_policy::config{flag, num_bytes});
}
// </backward_compatibility>
protected:
abstract_broker();
abstract_broker(middleman& parent_ref);
/** @endcond */
inline middleman& parent() {
return m_mm;
}
network::multiplexer& backend();
template <class Handle, class T>
static T& by_id(Handle hdl, std::map<Handle, intrusive_ptr<T>>& elements) {
auto i = elements.find(hdl);
if (i == elements.end()) {
throw std::invalid_argument("invalid handle");
}
return *(i->second);
}
// throws on error
inline scribe& by_id(connection_handle hdl) {
return by_id(hdl, m_scribes);
}
// throws on error
inline doorman& by_id(accept_handle hdl) { return by_id(hdl, m_doormen); }
virtual bool invoke_message_from_cache() = 0;
std::map<accept_handle, doorman_pointer> m_doormen;
std::map<connection_handle, scribe_pointer> m_scribes;
middleman& m_mm;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> m_cache;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_ABSTRACT_BROKER_HPP
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#define CAF_IO_ALL_HPP #define CAF_IO_ALL_HPP
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/typed_broker.hpp"
#include "caf/io/publish.hpp" #include "caf/io/publish.hpp"
#include "caf/io/spawn_io.hpp" #include "caf/io/spawn_io.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
......
...@@ -29,197 +29,22 @@ ...@@ -29,197 +29,22 @@
#include "caf/mixin/functor_based.hpp" #include "caf/mixin/functor_based.hpp"
#include "caf/detail/intrusive_partitioned_list.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
class broker;
class middleman;
using broker_ptr = intrusive_ptr<broker>;
/** /**
* A broker mediates between actor systems and other components in the network. * A broker mediates between actor systems and other components in the network.
* @extends local_actor * @extends local_actor
*/ */
class broker : public abstract_event_based_actor<behavior, false> { class broker : public abstract_broker,
public abstract_event_based_actor<behavior, false> {
public: public:
using super = abstract_event_based_actor<behavior, false>; using super = abstract_event_based_actor<behavior, false>;
using buffer_type = std::vector<char>;
/**
* Manages a low-level IO device for the `broker`.
*/
class servant {
public:
friend class broker;
virtual ~servant();
protected:
virtual void remove_from_broker() = 0;
virtual message disconnect_message() = 0;
inline broker* parent() {
return m_broker.get();
}
servant(broker* ptr);
void set_broker(broker* ptr);
void disconnect(bool invoke_disconnect_message);
bool m_disconnected;
intrusive_ptr<broker> m_broker;
};
/**
* Manages a stream.
*/
class scribe : public network::stream_manager, public servant {
public:
scribe(broker* parent, connection_handle hdl);
~scribe();
/**
* Implicitly starts the read loop on first call.
*/
virtual void configure_read(receive_policy::config config) = 0;
/**
* Grants access to the output buffer.
*/
virtual buffer_type& wr_buf() = 0;
/**
* Flushes the output buffer, i.e., sends the content of
* the buffer via the network.
*/
virtual void flush() = 0;
inline connection_handle hdl() const {
return m_hdl;
}
void io_failure(network::operation op) override;
protected:
virtual buffer_type& rd_buf() = 0;
inline new_data_msg& read_msg() {
return m_read_msg.get_as_mutable<new_data_msg>(0);
}
inline const new_data_msg& read_msg() const {
return m_read_msg.get_as<new_data_msg>(0);
}
void remove_from_broker() override;
message disconnect_message() override;
void consume(const void* data, size_t num_bytes) override;
connection_handle m_hdl;
message m_read_msg;
};
using scribe_pointer = intrusive_ptr<scribe>;
/**
* Manages incoming connections.
*/
class doorman : public network::acceptor_manager, public servant {
public:
doorman(broker* parent, accept_handle hdl);
~doorman();
inline accept_handle hdl() const {
return m_hdl;
}
void io_failure(network::operation op) override;
// needs to be launched explicitly
virtual void launch() = 0;
protected:
void remove_from_broker() override;
message disconnect_message() override;
inline new_connection_msg& accept_msg() {
return m_accept_msg.get_as_mutable<new_connection_msg>(0);
}
inline const new_connection_msg& accept_msg() const {
return m_accept_msg.get_as<new_connection_msg>(0);
}
accept_handle m_hdl;
message m_accept_msg;
};
using doorman_pointer = intrusive_ptr<doorman>;
class continuation; class continuation;
// a broker needs friends
friend class scribe;
friend class doorman;
friend class continuation;
~broker();
/**
* Modifies the receive policy for given connection.
* @param hdl Identifies the affected connection.
* @param config Contains the new receive policy.
*/
void configure_read(connection_handle hdl, receive_policy::config config);
/**
* Returns the write buffer for given connection.
*/
buffer_type& wr_buf(connection_handle hdl);
/**
* Writes `data` into the buffer for given connection.
*/
void write(connection_handle hdl, size_t data_size, const void* data);
/**
* Sends the content of the buffer for given connection.
*/
void flush(connection_handle hdl);
/**
* Returns the number of open connections.
*/
inline size_t num_connections() const {
return m_scribes.size();
}
std::vector<connection_handle> connections() const;
/** @cond PRIVATE */ /** @cond PRIVATE */
void initialize() override; void initialize() override;
...@@ -247,91 +72,31 @@ class broker : public abstract_event_based_actor<behavior, false> { ...@@ -247,91 +72,31 @@ class broker : public abstract_event_based_actor<behavior, false> {
fun, hdl, std::forward<Ts>(xs)...); fun, hdl, std::forward<Ts>(xs)...);
} }
inline void add_scribe(const scribe_pointer& ptr) { bool is_initialized() override {
m_scribes.emplace(ptr->hdl(), ptr); return super::is_initialized();
} }
connection_handle add_tcp_scribe(const std::string& host, uint16_t port); actor_addr address() const override {
return super::address();
void assign_tcp_scribe(connection_handle hdl);
connection_handle add_tcp_scribe(network::native_socket fd);
inline void add_doorman(const doorman_pointer& ptr) {
m_doormen.emplace(ptr->hdl(), ptr);
if (is_initialized()) {
ptr->launch();
}
} }
std::pair<accept_handle, uint16_t> add_tcp_doorman(uint16_t port = 0, void invoke_message(mailbox_element_ptr& msg) override;
const char* in = nullptr,
bool reuse_addr = false);
void assign_tcp_doorman(accept_handle hdl);
accept_handle add_tcp_doorman(network::native_socket fd);
void invoke_message(mailbox_element_ptr& msg);
void invoke_message(const actor_addr& sender, message_id mid, message& msg);
void enqueue(const actor_addr&, message_id, void enqueue(const actor_addr&, message_id,
message, execution_unit*) override; message, execution_unit*) override;
void enqueue(mailbox_element_ptr, execution_unit*) override; void enqueue(mailbox_element_ptr, execution_unit*) override;
/**
* Closes all connections and acceptors.
*/
void close_all();
/**
* Closes the connection identified by `handle`.
* Unwritten data will still be send.
*/
void close(connection_handle handle);
/**
* Closes the acceptor identified by `handle`.
*/
void close(accept_handle handle);
/**
* Checks whether a connection for `handle` exists.
*/
bool valid(connection_handle handle);
/**
* Checks whether an acceptor for `handle` exists.
*/
bool valid(accept_handle handle);
class functor_based; class functor_based;
void launch(execution_unit* eu, bool lazy, bool hide); void launch(execution_unit* eu, bool lazy, bool hide);
void cleanup(uint32_t reason); void cleanup(uint32_t reason);
// <backward_compatibility version="0.9">
static constexpr auto at_least = receive_policy_flag::at_least;
static constexpr auto at_most = receive_policy_flag::at_most;
static constexpr auto exactly = receive_policy_flag::exactly;
void receive_policy(connection_handle hdl, receive_policy_flag flag,
size_t num_bytes) CAF_DEPRECATED {
configure_read(hdl, receive_policy::config{flag, num_bytes});
}
// </backward_compatibility>
protected: protected:
broker(); broker() {}
broker(middleman& parent_ref); broker(middleman& parent_ref) : abstract_broker(parent_ref) {}
virtual behavior make_behavior() = 0; virtual behavior make_behavior() = 0;
...@@ -339,45 +104,10 @@ class broker : public abstract_event_based_actor<behavior, false> { ...@@ -339,45 +104,10 @@ class broker : public abstract_event_based_actor<behavior, false> {
* Can be overridden to perform cleanup code before the * Can be overridden to perform cleanup code before the
* broker closes all its connections. * broker closes all its connections.
*/ */
virtual void on_exit(); void on_exit() override;
/** @endcond */
inline middleman& parent() {
return m_mm;
}
network::multiplexer& backend();
private: private:
template <class Handle, class T> bool invoke_message_from_cache() override;
static T& by_id(Handle hdl, std::map<Handle, intrusive_ptr<T>>& elements) {
auto i = elements.find(hdl);
if (i == elements.end()) {
throw std::invalid_argument("invalid handle");
}
return *(i->second);
}
// throws on error
inline scribe& by_id(connection_handle hdl) {
return by_id(hdl, m_scribes);
}
// throws on error
inline doorman& by_id(accept_handle hdl) { return by_id(hdl, m_doormen); }
bool invoke_message_from_cache();
void erase_io(int id);
void erase_acceptor(int id);
std::map<accept_handle, doorman_pointer> m_doormen;
std::map<connection_handle, scribe_pointer> m_scribes;
middleman& m_mm;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> m_cache;
}; };
class broker::functor_based : public extend<broker>:: class broker::functor_based : public extend<broker>::
...@@ -395,6 +125,8 @@ class broker::functor_based : public extend<broker>:: ...@@ -395,6 +125,8 @@ class broker::functor_based : public extend<broker>::
behavior make_behavior() override; behavior make_behavior() override;
}; };
using broker_ptr = intrusive_ptr<broker>;
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -24,6 +24,7 @@ namespace caf { ...@@ -24,6 +24,7 @@ namespace caf {
namespace io { namespace io {
class basp_broker; class basp_broker;
class abstract_broker;
class broker; class broker;
class middleman; class middleman;
class receive_policy; class receive_policy;
......
...@@ -69,27 +69,27 @@ class asio_multiplexer : public multiplexer { ...@@ -69,27 +69,27 @@ class asio_multiplexer : public multiplexer {
connection_handle new_tcp_scribe(const std::string&, uint16_t) override; connection_handle new_tcp_scribe(const std::string&, uint16_t) override;
void assign_tcp_scribe(broker*, connection_handle hdl) override; void assign_tcp_scribe(abstract_broker*, connection_handle hdl) override;
template <class Socket> template <class Socket>
connection_handle add_tcp_scribe(broker*, Socket&& sock); connection_handle add_tcp_scribe(abstract_broker*, Socket&& sock);
connection_handle add_tcp_scribe(broker*, native_socket fd) override; connection_handle add_tcp_scribe(abstract_broker*, native_socket fd) override;
connection_handle add_tcp_scribe(broker*, const std::string& host, connection_handle add_tcp_scribe(abstract_broker*, const std::string& host,
uint16_t port) override; uint16_t port) override;
std::pair<accept_handle, uint16_t> new_tcp_doorman(uint16_t p, const char* in, std::pair<accept_handle, uint16_t> new_tcp_doorman(uint16_t p, const char* in,
bool rflag) override; bool rflag) override;
void assign_tcp_doorman(broker*, accept_handle hdl) override; void assign_tcp_doorman(abstract_broker*, accept_handle hdl) override;
accept_handle add_tcp_doorman(broker*, default_socket_acceptor&& sock); accept_handle add_tcp_doorman(abstract_broker*, default_socket_acceptor&& sock);
accept_handle add_tcp_doorman(broker*, native_socket fd) override; accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override;
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
add_tcp_doorman(broker*, uint16_t port, const char* in, bool rflag) override; add_tcp_doorman(abstract_broker*, uint16_t port, const char* in, bool rflag) override;
void dispatch_runnable(runnable_ptr ptr) override; void dispatch_runnable(runnable_ptr ptr) override;
......
...@@ -310,26 +310,26 @@ class default_multiplexer : public multiplexer { ...@@ -310,26 +310,26 @@ class default_multiplexer : public multiplexer {
connection_handle new_tcp_scribe(const std::string&, uint16_t) override; connection_handle new_tcp_scribe(const std::string&, uint16_t) override;
void assign_tcp_scribe(broker* ptr, connection_handle hdl) override; void assign_tcp_scribe(abstract_broker* ptr, connection_handle hdl) override;
connection_handle add_tcp_scribe(broker*, default_socket_acceptor&& sock); connection_handle add_tcp_scribe(abstract_broker*, default_socket_acceptor&& sock);
connection_handle add_tcp_scribe(broker*, native_socket fd) override; connection_handle add_tcp_scribe(abstract_broker*, native_socket fd) override;
connection_handle add_tcp_scribe(broker*, const std::string& h, connection_handle add_tcp_scribe(abstract_broker*, const std::string& h,
uint16_t port) override; uint16_t port) override;
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
new_tcp_doorman(uint16_t p, const char* in, bool rflag) override; new_tcp_doorman(uint16_t p, const char* in, bool rflag) override;
void assign_tcp_doorman(broker* ptr, accept_handle hdl) override; void assign_tcp_doorman(abstract_broker* ptr, accept_handle hdl) override;
accept_handle add_tcp_doorman(broker*, default_socket_acceptor&& sock); accept_handle add_tcp_doorman(abstract_broker*, default_socket_acceptor&& sock);
accept_handle add_tcp_doorman(broker*, native_socket fd) override; accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override;
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
add_tcp_doorman(broker*, uint16_t p, const char* in, bool rflag) override; add_tcp_doorman(abstract_broker*, uint16_t p, const char* in, bool rflag) override;
void dispatch_runnable(runnable_ptr ptr) override; void dispatch_runnable(runnable_ptr ptr) override;
......
...@@ -66,20 +66,20 @@ class multiplexer { ...@@ -66,20 +66,20 @@ class multiplexer {
* Assigns an unbound scribe identified by `hdl` to `ptr`. * Assigns an unbound scribe identified by `hdl` to `ptr`.
* @warning Do not call from outside the multiplexer's event loop. * @warning Do not call from outside the multiplexer's event loop.
*/ */
virtual void assign_tcp_scribe(broker* ptr, connection_handle hdl) = 0; virtual void assign_tcp_scribe(abstract_broker* ptr, connection_handle hdl) = 0;
/** /**
* Creates a new TCP doorman from a native socket handle. * Creates a new TCP doorman from a native socket handle.
* @warning Do not call from outside the multiplexer's event loop. * @warning Do not call from outside the multiplexer's event loop.
*/ */
virtual connection_handle add_tcp_scribe(broker* ptr, native_socket fd) = 0; virtual connection_handle add_tcp_scribe(abstract_broker* ptr, native_socket fd) = 0;
/** /**
* Tries to connect to host `h` on given `port` and returns a * Tries to connect to host `h` on given `port` and returns a
* new scribe managing the connection on success. * new scribe managing the connection on success.
* @warning Do not call from outside the multiplexer's event loop. * @warning Do not call from outside the multiplexer's event loop.
*/ */
virtual connection_handle add_tcp_scribe(broker* ptr, const std::string& host, virtual connection_handle add_tcp_scribe(abstract_broker* ptr, const std::string& host,
uint16_t port) = 0; uint16_t port) = 0;
/** /**
...@@ -95,13 +95,13 @@ class multiplexer { ...@@ -95,13 +95,13 @@ class multiplexer {
* Assigns an unbound doorman identified by `hdl` to `ptr`. * Assigns an unbound doorman identified by `hdl` to `ptr`.
* @warning Do not call from outside the multiplexer's event loop. * @warning Do not call from outside the multiplexer's event loop.
*/ */
virtual void assign_tcp_doorman(broker* ptr, accept_handle hdl) = 0; virtual void assign_tcp_doorman(abstract_broker* ptr, accept_handle hdl) = 0;
/** /**
* Creates a new TCP doorman from a native socket handle. * Creates a new TCP doorman from a native socket handle.
* @warning Do not call from outside the multiplexer's event loop. * @warning Do not call from outside the multiplexer's event loop.
*/ */
virtual accept_handle add_tcp_doorman(broker* ptr, native_socket fd) = 0; virtual accept_handle add_tcp_doorman(abstract_broker* ptr, native_socket fd) = 0;
/** /**
* Tries to create a new TCP doorman running on port `p`, optionally * Tries to create a new TCP doorman running on port `p`, optionally
...@@ -109,7 +109,7 @@ class multiplexer { ...@@ -109,7 +109,7 @@ class multiplexer {
* @warning Do not call from outside the multiplexer's event loop. * @warning Do not call from outside the multiplexer's event loop.
*/ */
virtual std::pair<accept_handle, uint16_t> virtual std::pair<accept_handle, uint16_t>
add_tcp_doorman(broker* ptr, uint16_t port, const char* in = nullptr, add_tcp_doorman(abstract_broker* ptr, uint16_t port, const char* in = nullptr,
bool reuse_addr = false) = 0; bool reuse_addr = false) = 0;
/** /**
......
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
#include "caf/spawn.hpp" #include "caf/spawn.hpp"
#include "caf/io/broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
...@@ -43,7 +43,7 @@ actor spawn_io(F fun, Ts&&... xs) { ...@@ -43,7 +43,7 @@ actor spawn_io(F fun, Ts&&... xs) {
} }
/** /**
* Spawns a new functor-based broker connecting to `host:port. * Spawns a new functor-based broker connecting to `host:port`.
*/ */
template <spawn_options Os = no_spawn_options, template <spawn_options Os = no_spawn_options,
typename F = std::function<void(broker*)>, class... Ts> typename F = std::function<void(broker*)>, class... Ts>
...@@ -69,7 +69,7 @@ actor spawn_io_client(F fun, const std::string& host, ...@@ -69,7 +69,7 @@ actor spawn_io_client(F fun, const std::string& host,
[&](broker::functor_based* ptr) { [&](broker::functor_based* ptr) {
auto mm = middleman::instance(); auto mm = middleman::instance();
auto hdl = mm->backend().add_tcp_scribe(ptr, host, port); auto hdl = mm->backend().add_tcp_scribe(ptr, host, port);
init(ptr, fun, hdl); init(ptr, std::move(fun), hdl);
}); });
} }
...@@ -93,6 +93,87 @@ actor spawn_io_server(F fun, uint16_t port, Ts&&... xs) { ...@@ -93,6 +93,87 @@ actor spawn_io_server(F fun, uint16_t port, Ts&&... xs) {
}); });
} }
/**
* Spawns a new functor-based typed-broker.
*/
template <spawn_options Os = no_spawn_options, class F, class... Ts>
typename infer_typed_actor_handle<
typename detail::get_callable_trait<F>::result_type,
typename detail::tl_head<
typename detail::get_callable_trait<F>::arg_types
>::type
>::type
spawn_io_typed(F fun, Ts&&... xs) {
using trait = typename detail::get_callable_trait<F>::type;
using arg_types = typename trait::arg_types;
using first_arg = typename detail::tl_head<arg_types>::type;
using base_class = typename std::remove_pointer<first_arg>::type;
using impl_class = typename base_class::functor_based;
return spawn_class<impl_class>(nullptr,
empty_before_launch_callback{},
std::move(fun), std::forward<Ts>(xs)...);
}
/**
* Spawns a new functor-based typed-broker connecting to `host:port`.
*/
template <spawn_options Os = no_spawn_options, class F, class... Ts>
typename infer_typed_actor_handle<
typename detail::get_callable_trait<F>::result_type,
typename detail::tl_head<
typename detail::get_callable_trait<F>::arg_types
>::type
>::type
spawn_io_client_typed(F fun, const std::string& host, uint16_t port,
Ts&&... xs) {
using trait = typename detail::get_callable_trait<F>::type;
using arg_types = typename trait::arg_types;
using first_arg = typename detail::tl_head<arg_types>::type;
using base_class = typename std::remove_pointer<first_arg>::type;
using impl_class = typename base_class::functor_based;
auto mfptr = &impl_class::template init<F, connection_handle, Ts...>;
using bi = std::function<void (impl_class*, F, connection_handle)>;
using namespace std::placeholders;
bi init = std::bind(mfptr, _1, _2, _3, std::forward<Ts>(xs)...);
return spawn_class<impl_class>(nullptr,
[&](impl_class* ptr) {
auto mm = middleman::instance();
auto hdl =
mm->backend().add_tcp_scribe(ptr,
host,
port);
init(ptr, std::move(fun), hdl);
});
}
/**
* Spawns a new typed-broker as server running on given `port`.
*/
template <spawn_options Os = no_spawn_options, class F, class... Ts>
typename infer_typed_actor_handle<
typename detail::get_callable_trait<F>::result_type,
typename detail::tl_head<
typename detail::get_callable_trait<F>::arg_types
>::type
>::type
spawn_io_server_typed(F fun, uint16_t port, Ts&&... xs) {
using trait = typename detail::get_callable_trait<F>::type;
using arg_types = typename trait::arg_types;
using first_arg = typename detail::tl_head<arg_types>::type;
using base_class = typename std::remove_pointer<first_arg>::type;
using impl_class = typename base_class::functor_based;
auto mfptr = &impl_class::template init<F, Ts...>;
using bi = std::function<void (impl_class*, F)>;
using namespace std::placeholders;
bi init = std::bind(mfptr, _1, _2, std::forward<Ts>(xs)...);
return spawn_class<impl_class>(nullptr,
[&](impl_class* ptr) {
auto mm = middleman::instance();
mm->backend().add_tcp_doorman(ptr, port);
init(ptr, std::move(fun));
});
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_IO_TYPED_BROKER_HPP
#define CAF_IO_TYPED_BROKER_HPP
#include <map>
#include <vector>
#include "caf/none.hpp"
#include "caf/config.hpp"
#include "caf/make_counted.hpp"
#include "caf/spawn.hpp"
#include "caf/extend.hpp"
#include "caf/typed_behavior.hpp"
#include "caf/local_actor.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/mixin/typed_functor_based.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/abstract_broker.hpp"
namespace caf {
namespace io {
/**
* A typed broker mediates between actor systems and other components in
* the network.
* @extends local_actor
*/
template <class... Sigs>
class typed_broker : public abstract_broker,
public abstract_event_based_actor<
typed_behavior<Sigs...>, false
> {
public:
using signatures = detail::type_list<Sigs...>;
using behavior_type = typed_behavior<Sigs...>;
using super = abstract_event_based_actor<behavior_type, false>;
using pointer = intrusive_ptr<typed_broker>;
// friend with all possible instantiations
template <class...>
friend class typed_broker;
class continuation {
public:
continuation(pointer bptr, mailbox_element_ptr mptr)
: m_self(std::move(bptr)),
m_ptr(std::move(mptr)) {
// nop
}
continuation(continuation&&) = default;
inline void operator()() {
CAF_PUSH_AID(m_self->id());
CAF_LOG_TRACE("");
m_self->invoke_message(m_ptr);
}
private:
pointer m_self;
mailbox_element_ptr m_ptr;
};
using super::send;
using super::delayed_send;
template <class... DestSigs, class... Ts>
void send(message_priority mp, const typed_actor<Sigs...>& dest, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
super::send(mp, dest, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void send(const typed_actor<DestSigs...>& dest, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
super::send(dest, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void delayed_send(message_priority mp, const typed_actor<Sigs...>& dest,
const duration& rtime, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
super::delayed_send(mp, dest, rtime, std::forward<Ts>(xs)...);
}
template <class... DestSigs, class... Ts>
void delayed_send(const typed_actor<Sigs...>& dest,
const duration& rtime, Ts&&... xs) {
detail::sender_signature_checker<
detail::type_list<Sigs...>,
detail::type_list<DestSigs...>,
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>
>::check();
super::delayed_send(dest, rtime, std::forward<Ts>(xs)...);
}
/** @cond PRIVATE */
std::set<std::string> message_types() const override {
return {Sigs::static_type_name()...};
}
void initialize() override {
// nop
}
template <class F, class... Ts>
typename infer_typed_actor_handle<
typename detail::get_callable_trait<F>::result_type,
typename detail::tl_head<
typename detail::get_callable_trait<F>::arg_types
>::type
>::type
fork(F fun, connection_handle hdl, Ts&&... xs) {
auto i = m_scribes.find(hdl);
if (i == m_scribes.end()) {
CAF_LOG_ERROR("invalid handle");
throw std::invalid_argument("invalid handle");
}
auto sptr = i->second;
CAF_ASSERT(sptr->hdl() == hdl);
m_scribes.erase(i);
using trait = typename detail::get_callable_trait<F>::type;
using arg_types = typename trait::arg_types;
using first_arg = typename detail::tl_head<arg_types>::type;
using base_class = typename std::remove_pointer<first_arg>::type;
using impl_class = typename base_class::functor_based;
return spawn_class<impl_class>(nullptr,
[sptr](impl_class* forked) {
sptr->set_broker(forked);
forked->m_scribes.emplace(sptr->hdl(),
sptr);
},
fun, hdl, std::forward<Ts>(xs)...);
}
bool is_initialized() override {
return super::is_initialized();
}
actor_addr address() const override {
return super::address();
}
void invoke_message(mailbox_element_ptr& ptr) override {
CAF_LOG_TRACE(CAF_TARG(ptr->msg, to_string));
if (super::exit_reason() != exit_reason::not_exited
|| !super::has_behavior()) {
CAF_LOG_DEBUG("actor already finished execution"
<< ", planned_exit_reason = "
<< super::planned_exit_reason()
<< ", has_behavior() = " << super::has_behavior());
if (ptr->mid.valid()) {
detail::sync_request_bouncer srb{super::exit_reason()};
srb(ptr->sender, ptr->mid);
}
return;
}
// prepare actor for invocation of message handler
try {
auto& bhvr = this->awaits_response()
? this->awaited_response_handler()
: this->bhvr_stack().back();
auto bid = this->awaited_response_id();
switch (local_actor::invoke_message(ptr, bhvr, bid)) {
case im_success: {
CAF_LOG_DEBUG("handle_message returned hm_msg_handled");
while (super::has_behavior()
&& super::planned_exit_reason() == exit_reason::not_exited
&& invoke_message_from_cache()) {
// rinse and repeat
}
break;
}
case im_dropped:
CAF_LOG_DEBUG("handle_message returned hm_drop_msg");
break;
case im_skipped: {
CAF_LOG_DEBUG("handle_message returned hm_skip_msg or hm_cache_msg");
if (ptr) {
m_cache.push_second_back(ptr.release());
}
break;
}
}
}
catch (std::exception& e) {
CAF_LOG_INFO("broker killed due to an unhandled exception: "
<< to_verbose_string(e));
// keep compiler happy in non-debug mode
static_cast<void>(e);
super::quit(exit_reason::unhandled_exception);
}
catch (...) {
CAF_LOG_ERROR("broker killed due to an unknown exception");
super::quit(exit_reason::unhandled_exception);
}
// safe to actually release behaviors now
super::bhvr_stack().cleanup();
// cleanup actor if needed
if (super::planned_exit_reason() != exit_reason::not_exited) {
cleanup(super::planned_exit_reason());
} else if (!super::has_behavior()) {
CAF_LOG_DEBUG("no behavior set, quit for normal exit reason");
super::quit(exit_reason::normal);
cleanup(super::planned_exit_reason());
}
}
void enqueue(const actor_addr& sender, message_id mid,
message msg, execution_unit* eu) override {
enqueue(mailbox_element::make(sender, mid, std::move(msg)), eu);
}
void enqueue(mailbox_element_ptr ptr, execution_unit*) override {
backend().post(continuation{this, std::move(ptr)});
}
class functor_based : public extend<typed_broker>::
template with<mixin::typed_functor_based> {
public:
using super = typename extend<typed_broker>::
template with<mixin::typed_functor_based>;
template <class... Ts>
functor_based(Ts&&... xs) : super(std::forward<Ts>(xs)...) {
// nop
}
~functor_based() {
// nop
}
behavior_type make_behavior() override {
return super::m_make_behavior(this);
}
};
void launch(execution_unit*, bool, bool is_hidden) {
// add implicit reference count held by the middleman
super::ref();
super::is_registered(!is_hidden);
CAF_PUSH_AID(super::id());
CAF_LOGF_TRACE("init and launch broker with ID " << id());
// we want to make sure initialization is executed in MM context
super::do_become(
[=](sys_atom) {
CAF_LOGF_TRACE("ID " << id());
super::unbecome();
// launch backends now, because user-defined initialization
// might call functions like add_connection
for (auto& kvp : m_doormen) {
kvp.second->launch();
}
super::is_initialized(true);
// run user-defined initialization code
auto bhvr = make_behavior();
if (bhvr) {
super::become(std::move(bhvr));
}
},
true
);
enqueue(invalid_actor_addr, invalid_message_id,
make_message(sys_atom::value), nullptr);
}
void cleanup(uint32_t reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
super::planned_exit_reason(reason);
on_exit();
close_all();
CAF_ASSERT(m_doormen.empty());
CAF_ASSERT(m_scribes.empty());
CAF_ASSERT(super::current_mailbox_element() == nullptr);
m_cache.clear();
super::cleanup(reason);
super::deref(); // release implicit reference count from middleman
}
protected:
typed_broker() {}
typed_broker(middleman& parent_ref) : abstract_broker(parent_ref) {}
virtual behavior_type make_behavior() = 0;
/**
* Can be overridden to perform cleanup code before the
* broker closes all its connections.
*/
void on_exit() override {
// nop
}
private:
bool invoke_message_from_cache() override {
CAF_LOG_TRACE("");
auto& bhvr = this->awaits_response()
? this->awaited_response_handler()
: this->bhvr_stack().back();
auto bid = this->awaited_response_id();
auto i = m_cache.second_begin();
auto e = m_cache.second_end();
CAF_LOG_DEBUG(std::distance(i, e) << " elements in cache");
return m_cache.invoke(static_cast<local_actor*>(this), i, e, bhvr, bid);
}
};
} // namespace io
} // namespace caf
#endif // CAF_IO_TYPED_BROKER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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/none.hpp"
#include "caf/config.hpp"
#include "caf/make_counted.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
namespace caf {
namespace io {
void abstract_broker::servant::set_broker(abstract_broker* new_broker) {
if (!m_disconnected) {
m_broker = new_broker;
}
}
abstract_broker::servant::~servant() {
CAF_LOG_TRACE("");
}
abstract_broker::servant::servant(abstract_broker* ptr) : m_disconnected(false), m_broker(ptr) {
// nop
}
void abstract_broker::servant::disconnect(bool invoke_disconnect_message) {
CAF_LOG_TRACE("");
if (!m_disconnected) {
CAF_LOG_DEBUG("disconnect servant from broker");
m_disconnected = true;
remove_from_broker();
if (invoke_disconnect_message) {
auto msg = disconnect_message();
m_broker->invoke_message(m_broker->address(),invalid_message_id, msg);
}
}
}
abstract_broker::scribe::scribe(abstract_broker* ptr, connection_handle conn_hdl)
: servant(ptr),
m_hdl(conn_hdl) {
std::vector<char> tmp;
m_read_msg = make_message(new_data_msg{m_hdl, std::move(tmp)});
}
void abstract_broker::scribe::remove_from_broker() {
CAF_LOG_TRACE("hdl = " << hdl().id());
m_broker->m_scribes.erase(hdl());
}
abstract_broker::scribe::~scribe() {
CAF_LOG_TRACE("");
}
message abstract_broker::scribe::disconnect_message() {
return make_message(connection_closed_msg{hdl()});
}
void abstract_broker::scribe::consume(const void*, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
if (m_disconnected) {
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an IO event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return;
}
auto& buf = rd_buf();
buf.resize(num_bytes); // make sure size is correct
read_msg().buf.swap(buf); // swap into message
m_broker->invoke_message(invalid_actor_addr, // call client
invalid_message_id, m_read_msg);
read_msg().buf.swap(buf); // swap buffer back to stream
flush(); // implicit flush of wr_buf()
}
void abstract_broker::scribe::io_failure(network::operation op) {
CAF_LOG_TRACE("id = " << hdl().id()
<< ", " << CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
disconnect(true);
}
abstract_broker::doorman::doorman(abstract_broker* ptr, accept_handle acc_hdl)
: servant(ptr), m_hdl(acc_hdl) {
auto hdl2 = connection_handle::from_int(-1);
m_accept_msg = make_message(new_connection_msg{m_hdl, hdl2});
}
abstract_broker::doorman::~doorman() {
CAF_LOG_TRACE("");
}
void abstract_broker::doorman::remove_from_broker() {
CAF_LOG_TRACE("hdl = " << hdl().id());
m_broker->m_doormen.erase(hdl());
}
message abstract_broker::doorman::disconnect_message() {
return make_message(acceptor_closed_msg{hdl()});
}
void abstract_broker::doorman::io_failure(network::operation op) {
CAF_LOG_TRACE("id = " << hdl().id() << ", "
<< CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
disconnect(true);
}
abstract_broker::~abstract_broker() {
CAF_LOG_TRACE("");
}
void abstract_broker::configure_read(connection_handle hdl, receive_policy::config cfg) {
CAF_LOG_TRACE(CAF_MARG(hdl, id) << ", cfg = {" << static_cast<int>(cfg.first)
<< ", " << cfg.second << "}");
by_id(hdl).configure_read(cfg);
}
abstract_broker::buffer_type& abstract_broker::wr_buf(connection_handle hdl) {
return by_id(hdl).wr_buf();
}
void abstract_broker::write(connection_handle hdl, size_t bs, const void* buf) {
auto& out = wr_buf(hdl);
auto first = reinterpret_cast<const char*>(buf);
auto last = first + bs;
out.insert(out.end(), first, last);
}
void abstract_broker::flush(connection_handle hdl) {
by_id(hdl).flush();
}
std::vector<connection_handle> abstract_broker::connections() const {
std::vector<connection_handle> result;
result.reserve(m_scribes.size());
for (auto& kvp : m_scribes) {
result.push_back(kvp.first);
}
return result;
}
connection_handle abstract_broker::add_tcp_scribe(const std::string& host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(host) << ", " << CAF_ARG(port));
return backend().add_tcp_scribe(this, host, port);
}
void abstract_broker::assign_tcp_scribe(connection_handle hdl) {
CAF_LOG_TRACE(CAF_MARG(hdl, id));
backend().assign_tcp_scribe(this, hdl);
}
connection_handle
abstract_broker::add_tcp_scribe(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_scribe(this, fd);
}
std::pair<accept_handle, uint16_t>
abstract_broker::add_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", in = " << (in ? in : "nullptr")
<< ", " << CAF_ARG(reuse_addr));
return backend().add_tcp_doorman(this, port, in, reuse_addr);
}
void abstract_broker::assign_tcp_doorman(accept_handle hdl) {
CAF_LOG_TRACE(CAF_MARG(hdl, id));
backend().assign_tcp_doorman(this, hdl);
}
accept_handle abstract_broker::add_tcp_doorman(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_doorman(this, fd);
}
void abstract_broker::invoke_message(const actor_addr& sender,
message_id mid, message& msg) {
auto ptr = mailbox_element::make(sender, mid, message{});
ptr->msg.swap(msg);
invoke_message(ptr);
if (ptr) {
ptr->msg.swap(msg);
}
}
void abstract_broker::close_all() {
CAF_LOG_TRACE("");
while (!m_doormen.empty()) {
// stop_reading will remove the doorman from m_doormen
m_doormen.begin()->second->stop_reading();
}
while (!m_scribes.empty()) {
// stop_reading will remove the scribe from m_scribes
m_scribes.begin()->second->stop_reading();
}
}
void abstract_broker::close(connection_handle hdl) {
by_id(hdl).stop_reading();
}
void abstract_broker::close(accept_handle hdl) {
by_id(hdl).stop_reading();
}
bool abstract_broker::valid(connection_handle hdl) {
return m_scribes.count(hdl) > 0;
}
bool abstract_broker::valid(accept_handle hdl) {
return m_doormen.count(hdl) > 0;
}
abstract_broker::abstract_broker() : m_mm(*middleman::instance()) {
// nop
}
abstract_broker::abstract_broker(middleman& ptr) : m_mm(ptr) {
// nop
}
network::multiplexer& abstract_broker::backend() {
return m_mm.backend();
}
} // namespace io
} // namespace caf
...@@ -103,7 +103,7 @@ connection_handle asio_multiplexer::new_tcp_scribe(const std::string& host, ...@@ -103,7 +103,7 @@ connection_handle asio_multiplexer::new_tcp_scribe(const std::string& host,
return connection_handle::from_int(id); return connection_handle::from_int(id);
} }
void asio_multiplexer::assign_tcp_scribe(broker* self, connection_handle hdl) { void asio_multiplexer::assign_tcp_scribe(abstract_broker* self, connection_handle hdl) {
std::lock_guard<std::mutex> lock(m_mtx_sockets); std::lock_guard<std::mutex> lock(m_mtx_sockets);
auto itr = m_unassigned_sockets.find(hdl.id()); auto itr = m_unassigned_sockets.find(hdl.id());
if (itr != m_unassigned_sockets.end()) { if (itr != m_unassigned_sockets.end()) {
...@@ -113,12 +113,12 @@ void asio_multiplexer::assign_tcp_scribe(broker* self, connection_handle hdl) { ...@@ -113,12 +113,12 @@ void asio_multiplexer::assign_tcp_scribe(broker* self, connection_handle hdl) {
} }
template <class Socket> template <class Socket>
connection_handle asio_multiplexer::add_tcp_scribe(broker* self, connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
Socket&& sock) { Socket&& sock) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
class impl : public broker::scribe { class impl : public abstract_broker::scribe {
public: public:
impl(broker* ptr, Socket&& s) impl(abstract_broker* ptr, Socket&& s)
: scribe(ptr, network::conn_hdl_from_socket(s)), : scribe(ptr, network::conn_hdl_from_socket(s)),
m_launched(false), m_launched(false),
m_stream(s.get_io_service()) { m_stream(s.get_io_service()) {
...@@ -131,8 +131,8 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self, ...@@ -131,8 +131,8 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self,
launch(); launch();
} }
} }
broker::buffer_type& wr_buf() override { return m_stream.wr_buf(); } abstract_broker::buffer_type& wr_buf() override { return m_stream.wr_buf(); }
broker::buffer_type& rd_buf() override { return m_stream.rd_buf(); } abstract_broker::buffer_type& rd_buf() override { return m_stream.rd_buf(); }
void stop_reading() override { void stop_reading() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
m_stream.stop_reading(); m_stream.stop_reading();
...@@ -153,12 +153,12 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self, ...@@ -153,12 +153,12 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self,
bool m_launched; bool m_launched;
stream<Socket> m_stream; stream<Socket> m_stream;
}; };
broker::scribe_pointer ptr = make_counted<impl>(self, std::move(sock)); abstract_broker::scribe_pointer ptr = make_counted<impl>(self, std::move(sock));
self->add_scribe(ptr); self->add_scribe(ptr);
return ptr->hdl(); return ptr->hdl();
} }
connection_handle asio_multiplexer::add_tcp_scribe(broker* self, connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
native_socket fd) { native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd)); CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd));
boost::system::error_code ec; boost::system::error_code ec;
...@@ -173,7 +173,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self, ...@@ -173,7 +173,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self,
return add_tcp_scribe(self, std::move(sock)); return add_tcp_scribe(self, std::move(sock));
} }
connection_handle asio_multiplexer::add_tcp_scribe(broker* self, connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
const std::string& host, const std::string& host,
uint16_t port) { uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(host) << ":" << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(host) << ":" << CAF_ARG(port));
...@@ -192,7 +192,7 @@ asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) { ...@@ -192,7 +192,7 @@ asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) {
return {accept_handle::from_int(id), assigned_port}; return {accept_handle::from_int(id), assigned_port};
} }
void asio_multiplexer::assign_tcp_doorman(broker* self, accept_handle hdl) { void asio_multiplexer::assign_tcp_doorman(abstract_broker* self, accept_handle hdl) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
std::lock_guard<std::mutex> lock(m_mtx_acceptors); std::lock_guard<std::mutex> lock(m_mtx_acceptors);
auto itr = m_unassigned_acceptors.find(hdl.id()); auto itr = m_unassigned_acceptors.find(hdl.id());
...@@ -203,13 +203,13 @@ void asio_multiplexer::assign_tcp_doorman(broker* self, accept_handle hdl) { ...@@ -203,13 +203,13 @@ void asio_multiplexer::assign_tcp_doorman(broker* self, accept_handle hdl) {
} }
accept_handle accept_handle
asio_multiplexer::add_tcp_doorman(broker* self, asio_multiplexer::add_tcp_doorman(abstract_broker* self,
default_socket_acceptor&& sock) { default_socket_acceptor&& sock) {
CAF_LOG_TRACE("sock.fd = " << sock.native_handle()); CAF_LOG_TRACE("sock.fd = " << sock.native_handle());
CAF_ASSERT(sock.native_handle() != network::invalid_native_socket); CAF_ASSERT(sock.native_handle() != network::invalid_native_socket);
class impl : public broker::doorman { class impl : public abstract_broker::doorman {
public: public:
impl(broker* ptr, default_socket_acceptor&& s, impl(abstract_broker* ptr, default_socket_acceptor&& s,
network::asio_multiplexer& am) network::asio_multiplexer& am)
: doorman(ptr, network::accept_hdl_from_socket(s)), : doorman(ptr, network::accept_hdl_from_socket(s)),
m_acceptor(am, s.get_io_service()) { m_acceptor(am, s.get_io_service()) {
...@@ -236,13 +236,13 @@ asio_multiplexer::add_tcp_doorman(broker* self, ...@@ -236,13 +236,13 @@ asio_multiplexer::add_tcp_doorman(broker* self,
private: private:
network::acceptor<default_socket_acceptor> m_acceptor; network::acceptor<default_socket_acceptor> m_acceptor;
}; };
broker::doorman_pointer ptr abstract_broker::doorman_pointer ptr
= make_counted<impl>(self, std::move(sock), *this); = make_counted<impl>(self, std::move(sock), *this);
self->add_doorman(ptr); self->add_doorman(ptr);
return ptr->hdl(); return ptr->hdl();
} }
accept_handle asio_multiplexer::add_tcp_doorman(broker* self, accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self,
native_socket fd) { native_socket fd) {
default_socket_acceptor sock{backend()}; default_socket_acceptor sock{backend()};
boost::system::error_code ec; boost::system::error_code ec;
...@@ -257,7 +257,7 @@ accept_handle asio_multiplexer::add_tcp_doorman(broker* self, ...@@ -257,7 +257,7 @@ accept_handle asio_multiplexer::add_tcp_doorman(broker* self,
} }
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
asio_multiplexer::add_tcp_doorman(broker* self, uint16_t port, const char* in, asio_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port, const char* in,
bool rflag) { bool rflag) {
default_socket_acceptor fd{backend()}; default_socket_acceptor fd{backend()};
ip_bind(fd, port, in, rflag); ip_bind(fd, port, in, rflag);
......
...@@ -34,106 +34,6 @@ ...@@ -34,106 +34,6 @@
namespace caf { namespace caf {
namespace io { namespace io {
broker::servant::~servant() {
CAF_LOG_TRACE("");
}
broker::servant::servant(broker* ptr) : m_disconnected(false), m_broker(ptr) {
// nop
}
void broker::servant::set_broker(broker* new_broker) {
if (!m_disconnected) {
m_broker = new_broker;
}
}
void broker::servant::disconnect(bool invoke_disconnect_message) {
CAF_LOG_TRACE("");
if (!m_disconnected) {
CAF_LOG_DEBUG("disconnect servant from broker");
m_disconnected = true;
remove_from_broker();
if (invoke_disconnect_message) {
auto msg = disconnect_message();
m_broker->invoke_message(m_broker->address(),invalid_message_id, msg);
}
}
}
broker::scribe::scribe(broker* ptr, connection_handle conn_hdl)
: servant(ptr),
m_hdl(conn_hdl) {
std::vector<char> tmp;
m_read_msg = make_message(new_data_msg{m_hdl, std::move(tmp)});
}
void broker::scribe::remove_from_broker() {
CAF_LOG_TRACE("hdl = " << hdl().id());
m_broker->m_scribes.erase(hdl());
}
broker::scribe::~scribe() {
CAF_LOG_TRACE("");
}
message broker::scribe::disconnect_message() {
return make_message(connection_closed_msg{hdl()});
}
void broker::scribe::consume(const void*, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
if (m_disconnected) {
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an IO event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return;
}
auto& buf = rd_buf();
buf.resize(num_bytes); // make sure size is correct
read_msg().buf.swap(buf); // swap into message
m_broker->invoke_message(invalid_actor_addr, // call client
invalid_message_id, m_read_msg);
read_msg().buf.swap(buf); // swap buffer back to stream
flush(); // implicit flush of wr_buf()
}
void broker::scribe::io_failure(network::operation op) {
CAF_LOG_TRACE("id = " << hdl().id()
<< ", " << CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
disconnect(true);
}
broker::doorman::doorman(broker* ptr, accept_handle acc_hdl)
: servant(ptr), m_hdl(acc_hdl) {
auto hdl2 = connection_handle::from_int(-1);
m_accept_msg = make_message(new_connection_msg{m_hdl, hdl2});
}
broker::doorman::~doorman() {
CAF_LOG_TRACE("");
}
void broker::doorman::remove_from_broker() {
CAF_LOG_TRACE("hdl = " << hdl().id());
m_broker->m_doormen.erase(hdl());
}
message broker::doorman::disconnect_message() {
return make_message(acceptor_closed_msg{hdl()});
}
void broker::doorman::io_failure(network::operation op) {
CAF_LOG_TRACE("id = " << hdl().id() << ", "
<< CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
disconnect(true);
}
class broker::continuation { class broker::continuation {
public: public:
continuation(broker_ptr bptr, mailbox_element_ptr mptr) continuation(broker_ptr bptr, mailbox_element_ptr mptr)
...@@ -218,15 +118,6 @@ void broker::invoke_message(mailbox_element_ptr& ptr) { ...@@ -218,15 +118,6 @@ void broker::invoke_message(mailbox_element_ptr& ptr) {
} }
} }
void broker::invoke_message(const actor_addr& v0, message_id v1, message& v2) {
auto ptr = mailbox_element::make(v0, v1, message{});
ptr->msg.swap(v2);
invoke_message(ptr);
if (ptr) {
ptr->msg.swap(v2);
}
}
bool broker::invoke_message_from_cache() { bool broker::invoke_message_from_cache() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto& bhvr = this->awaits_response() auto& bhvr = this->awaits_response()
...@@ -239,15 +130,8 @@ bool broker::invoke_message_from_cache() { ...@@ -239,15 +130,8 @@ bool broker::invoke_message_from_cache() {
return m_cache.invoke(static_cast<local_actor*>(this), i, e, bhvr, bid); return m_cache.invoke(static_cast<local_actor*>(this), i, e, bhvr, bid);
} }
void broker::write(connection_handle hdl, size_t bs, const void* buf) {
auto& out = wr_buf(hdl);
auto first = reinterpret_cast<const char*>(buf);
auto last = first + bs;
out.insert(out.end(), first, last);
}
void broker::enqueue(mailbox_element_ptr ptr, execution_unit*) { void broker::enqueue(mailbox_element_ptr ptr, execution_unit*) {
parent().backend().post(continuation{this, std::move(ptr)}); backend().post(continuation{this, std::move(ptr)});
} }
void broker::enqueue(const actor_addr& sender, message_id mid, message msg, void broker::enqueue(const actor_addr& sender, message_id mid, message msg,
...@@ -255,14 +139,6 @@ void broker::enqueue(const actor_addr& sender, message_id mid, message msg, ...@@ -255,14 +139,6 @@ void broker::enqueue(const actor_addr& sender, message_id mid, message msg,
enqueue(mailbox_element::make(sender, mid, std::move(msg)), eu); enqueue(mailbox_element::make(sender, mid, std::move(msg)), eu);
} }
broker::broker() : m_mm(*middleman::instance()) {
// nop
}
broker::broker(middleman& ptr) : m_mm(ptr) {
// nop
}
void broker::cleanup(uint32_t reason) { void broker::cleanup(uint32_t reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
planned_exit_reason(reason); planned_exit_reason(reason);
...@@ -296,7 +172,7 @@ void broker::launch(execution_unit*, bool, bool is_hidden) { ...@@ -296,7 +172,7 @@ void broker::launch(execution_unit*, bool, bool is_hidden) {
for (auto& kvp : m_doormen) { for (auto& kvp : m_doormen) {
kvp.second->launch(); kvp.second->launch();
} }
is_initialized(true); super::is_initialized(true);
// run user-defined initialization code // run user-defined initialization code
auto bhvr = make_behavior(); auto bhvr = make_behavior();
if (bhvr) { if (bhvr) {
...@@ -308,60 +184,6 @@ void broker::launch(execution_unit*, bool, bool is_hidden) { ...@@ -308,60 +184,6 @@ void broker::launch(execution_unit*, bool, bool is_hidden) {
make_message(sys_atom::value), nullptr); make_message(sys_atom::value), nullptr);
} }
void broker::configure_read(connection_handle hdl, receive_policy::config cfg) {
CAF_LOG_TRACE(CAF_MARG(hdl, id) << ", cfg = {" << static_cast<int>(cfg.first)
<< ", " << cfg.second << "}");
by_id(hdl).configure_read(cfg);
}
void broker::flush(connection_handle hdl) {
by_id(hdl).flush();
}
broker::buffer_type& broker::wr_buf(connection_handle hdl) {
return by_id(hdl).wr_buf();
}
broker::~broker() {
CAF_LOG_TRACE("");
}
void broker::close(connection_handle hdl) {
by_id(hdl).stop_reading();
}
void broker::close(accept_handle hdl) {
by_id(hdl).stop_reading();
}
void broker::close_all() {
CAF_LOG_TRACE("");
while (!m_doormen.empty()) {
// stop_reading will remove the doorman from m_doormen
m_doormen.begin()->second->stop_reading();
}
while (!m_scribes.empty()) {
// stop_reading will remove the scribe from m_scribes
m_scribes.begin()->second->stop_reading();
}
}
bool broker::valid(connection_handle hdl) {
return m_scribes.count(hdl) > 0;
}
bool broker::valid(accept_handle hdl) {
return m_doormen.count(hdl) > 0;
}
std::vector<connection_handle> broker::connections() const {
std::vector<connection_handle> result;
for (auto& kvp : m_scribes) {
result.push_back(kvp.first);
}
return result;
}
void broker::initialize() { void broker::initialize() {
// nop // nop
} }
...@@ -374,41 +196,5 @@ behavior broker::functor_based::make_behavior() { ...@@ -374,41 +196,5 @@ behavior broker::functor_based::make_behavior() {
return m_make_behavior(this); return m_make_behavior(this);
} }
network::multiplexer& broker::backend() {
return m_mm.backend();
}
connection_handle broker::add_tcp_scribe(const std::string& hst, uint16_t prt) {
CAF_LOG_TRACE(CAF_ARG(hst) << ", " << CAF_ARG(prt));
return backend().add_tcp_scribe(this, hst, prt);
}
void broker::assign_tcp_scribe(connection_handle hdl) {
CAF_LOG_TRACE(CAF_MARG(hdl, id));
backend().assign_tcp_scribe(this, hdl);
}
connection_handle broker::add_tcp_scribe(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_scribe(this, fd);
}
void broker::assign_tcp_doorman(accept_handle hdl) {
CAF_LOG_TRACE(CAF_MARG(hdl, id));
backend().assign_tcp_doorman(this, hdl);
}
std::pair<accept_handle, uint16_t>
broker::add_tcp_doorman(uint16_t port, const char* in, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", in = " << (in ? in : "nullptr")
<< ", " << CAF_ARG(reuse_addr));
return backend().add_tcp_doorman(this, port, in, reuse_addr);
}
accept_handle broker::add_tcp_doorman(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_doorman(this, fd);
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -726,12 +726,12 @@ void default_multiplexer::dispatch_runnable(runnable_ptr ptr) { ...@@ -726,12 +726,12 @@ void default_multiplexer::dispatch_runnable(runnable_ptr ptr) {
wr_dispatch_request(ptr.release()); wr_dispatch_request(ptr.release());
} }
connection_handle default_multiplexer::add_tcp_scribe(broker* self, connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
default_socket&& sock) { default_socket&& sock) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
class impl : public broker::scribe { class impl : public abstract_broker::scribe {
public: public:
impl(broker* ptr, default_socket&& s) impl(abstract_broker* ptr, default_socket&& s)
: scribe(ptr, network::conn_hdl_from_socket(s)), : scribe(ptr, network::conn_hdl_from_socket(s)),
m_launched(false), m_launched(false),
m_stream(s.backend()) { m_stream(s.backend()) {
...@@ -742,10 +742,10 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self, ...@@ -742,10 +742,10 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
m_stream.configure_read(config); m_stream.configure_read(config);
if (!m_launched) launch(); if (!m_launched) launch();
} }
broker::buffer_type& wr_buf() override { abstract_broker::buffer_type& wr_buf() override {
return m_stream.wr_buf(); return m_stream.wr_buf();
} }
broker::buffer_type& rd_buf() override { abstract_broker::buffer_type& rd_buf() override {
return m_stream.rd_buf(); return m_stream.rd_buf();
} }
void stop_reading() override { void stop_reading() override {
...@@ -767,19 +767,19 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self, ...@@ -767,19 +767,19 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
bool m_launched; bool m_launched;
stream<default_socket> m_stream; stream<default_socket> m_stream;
}; };
broker::scribe_pointer ptr = make_counted<impl>(self, std::move(sock)); abstract_broker::scribe_pointer ptr = make_counted<impl>(self, std::move(sock));
self->add_scribe(ptr); self->add_scribe(ptr);
return ptr->hdl(); return ptr->hdl();
} }
accept_handle accept_handle
default_multiplexer::add_tcp_doorman(broker* self, default_multiplexer::add_tcp_doorman(abstract_broker* self,
default_socket_acceptor&& sock) { default_socket_acceptor&& sock) {
CAF_LOG_TRACE("sock.fd = " << sock.fd()); CAF_LOG_TRACE("sock.fd = " << sock.fd());
CAF_ASSERT(sock.fd() != network::invalid_native_socket); CAF_ASSERT(sock.fd() != network::invalid_native_socket);
class impl : public broker::doorman { class impl : public abstract_broker::doorman {
public: public:
impl(broker* ptr, default_socket_acceptor&& s) impl(abstract_broker* ptr, default_socket_acceptor&& s)
: doorman(ptr, network::accept_hdl_from_socket(s)), : doorman(ptr, network::accept_hdl_from_socket(s)),
m_acceptor(s.backend()) { m_acceptor(s.backend()) {
m_acceptor.init(std::move(s)); m_acceptor.init(std::move(s));
...@@ -804,7 +804,7 @@ default_multiplexer::add_tcp_doorman(broker* self, ...@@ -804,7 +804,7 @@ default_multiplexer::add_tcp_doorman(broker* self,
private: private:
network::acceptor<default_socket_acceptor> m_acceptor; network::acceptor<default_socket_acceptor> m_acceptor;
}; };
broker::doorman_pointer ptr = make_counted<impl>(self, std::move(sock)); abstract_broker::doorman_pointer ptr = make_counted<impl>(self, std::move(sock));
self->add_doorman(ptr); self->add_doorman(ptr);
return ptr->hdl(); return ptr->hdl();
} }
...@@ -815,19 +815,19 @@ connection_handle default_multiplexer::new_tcp_scribe(const std::string& host, ...@@ -815,19 +815,19 @@ connection_handle default_multiplexer::new_tcp_scribe(const std::string& host,
return connection_handle::from_int(int64_from_native_socket(fd)); return connection_handle::from_int(int64_from_native_socket(fd));
} }
void default_multiplexer::assign_tcp_scribe(broker* self, void default_multiplexer::assign_tcp_scribe(abstract_broker* self,
connection_handle hdl) { connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_MARG(hdl, id)); CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_MARG(hdl, id));
add_tcp_scribe(self, static_cast<native_socket>(hdl.id())); add_tcp_scribe(self, static_cast<native_socket>(hdl.id()));
} }
connection_handle default_multiplexer::add_tcp_scribe(broker* self, connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
native_socket fd) { native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd)); CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd));
return add_tcp_scribe(self, default_socket{*this, fd}); return add_tcp_scribe(self, default_socket{*this, fd});
} }
connection_handle default_multiplexer::add_tcp_scribe(broker* self, connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
const std::string& host, const std::string& host,
uint16_t port) { uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(host) CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(host)
...@@ -843,17 +843,17 @@ default_multiplexer::new_tcp_doorman(uint16_t port, const char* in, ...@@ -843,17 +843,17 @@ default_multiplexer::new_tcp_doorman(uint16_t port, const char* in,
res.second}; res.second};
} }
void default_multiplexer::assign_tcp_doorman(broker* ptr, accept_handle hdl) { void default_multiplexer::assign_tcp_doorman(abstract_broker* ptr, accept_handle hdl) {
add_tcp_doorman(ptr, static_cast<native_socket>(hdl.id())); add_tcp_doorman(ptr, static_cast<native_socket>(hdl.id()));
} }
accept_handle default_multiplexer::add_tcp_doorman(broker* self, accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
native_socket fd) { native_socket fd) {
return add_tcp_doorman(self, default_socket_acceptor{*this, fd}); return add_tcp_doorman(self, default_socket_acceptor{*this, fd});
} }
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
default_multiplexer::add_tcp_doorman(broker* self, uint16_t port, default_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port,
const char* host, bool reuse_addr) { const char* host, bool reuse_addr) {
auto acceptor = new_tcp_acceptor(port, host, reuse_addr); auto acceptor = new_tcp_acceptor(port, host, reuse_addr);
auto bound_port = acceptor.second; auto bound_port = acceptor.second;
......
...@@ -38,6 +38,8 @@ using ping_atom = caf::atom_constant<caf::atom("ping")>; ...@@ -38,6 +38,8 @@ using ping_atom = caf::atom_constant<caf::atom("ping")>;
using pong_atom = caf::atom_constant<caf::atom("pong")>; using pong_atom = caf::atom_constant<caf::atom("pong")>;
using kickoff_atom = caf::atom_constant<caf::atom("kickoff")>; using kickoff_atom = caf::atom_constant<caf::atom("kickoff")>;
namespace {
void ping(event_based_actor* self, size_t num_pings) { void ping(event_based_actor* self, size_t num_pings) {
CAF_MESSAGE("num_pings: " << num_pings); CAF_MESSAGE("num_pings: " << num_pings);
auto count = std::make_shared<size_t>(0); auto count = std::make_shared<size_t>(0);
...@@ -96,6 +98,8 @@ void pong(event_based_actor* self) { ...@@ -96,6 +98,8 @@ void pong(event_based_actor* self) {
); );
} }
}
void peer_fun(broker* self, connection_handle hdl, const actor& buddy) { void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
CAF_MESSAGE("peer_fun called"); CAF_MESSAGE("peer_fun called");
CAF_CHECK(self != nullptr); CAF_CHECK(self != nullptr);
...@@ -170,6 +174,8 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) { ...@@ -170,6 +174,8 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
}; };
} }
namespace {
void run_server(bool spawn_client, const char* bin_path) { void run_server(bool spawn_client, const char* bin_path) {
scoped_actor self; scoped_actor self;
auto serv = io::spawn_io(peer_acceptor_fun, spawn(pong)); auto serv = io::spawn_io(peer_acceptor_fun, spawn(pong));
...@@ -193,6 +199,8 @@ void run_server(bool spawn_client, const char* bin_path) { ...@@ -193,6 +199,8 @@ void run_server(bool spawn_client, const char* bin_path) {
); );
} }
}
CAF_TEST(test_broker) { CAF_TEST(test_broker) {
auto argv = caf::test::engine::argv(); auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc(); auto argc = caf::test::engine::argc();
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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 typed_broker
#include "caf/test/unit_test.hpp"
#include <memory>
#include <iostream>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/detail/run_program.hpp"
using namespace std;
using namespace caf;
using namespace caf::io;
using ping_atom = caf::atom_constant<caf::atom("ping")>;
using pong_atom = caf::atom_constant<caf::atom("pong")>;
using kickoff_atom = caf::atom_constant<caf::atom("kickoff")>;
namespace {
void ping(event_based_actor* self, size_t num_pings) {
CAF_MESSAGE("num_pings: " << num_pings);
auto count = std::make_shared<size_t>(0);
self->become(
[=](kickoff_atom, const actor& pong) {
CAF_MESSAGE("received `kickoff_atom`");
self->send(pong, ping_atom::value, 1);
self->become(
[=](pong_atom, int value)->std::tuple<atom_value, int> {
if (++*count >= num_pings) {
CAF_MESSAGE("received " << num_pings
<< " pings, call self->quit");
self->quit();
}
return std::make_tuple(ping_atom::value, value + 1);
},
others >> [&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message()));
});
},
others >> [&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message()));
}
);
}
void pong(event_based_actor* self) {
CAF_MESSAGE("pong actor started");
self->become(
[=](ping_atom, int value) -> std::tuple<atom_value, int> {
CAF_MESSAGE("received `ping_atom`");
self->monitor(self->current_sender());
// set next behavior
self->become(
[](ping_atom, int val) {
return std::make_tuple(pong_atom::value, val);
},
[=](const down_msg& dm) {
CAF_MESSAGE("received down_msg{" << dm.reason << "}");
self->quit(dm.reason);
},
others >> [&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message()));
}
);
// reply to 'ping'
return std::make_tuple(pong_atom::value, value);
},
others >> [&] {
CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message()));
}
);
}
}
using peer_broker = typed_broker<
replies_to<connection_closed_msg>::with<void>,
replies_to<new_data_msg>::with<void>,
replies_to<ping_atom, int>::with<void>,
replies_to<pong_atom, int>::with<void>
>;
peer_broker::behavior_type
peer_fun(peer_broker* self, connection_handle hdl, const actor& buddy) {
CAF_MESSAGE("peer_fun called");
CAF_CHECK(self != nullptr);
CAF_CHECK(buddy != invalid_actor);
self->monitor(buddy);
// assume exactly one connection
auto cons = self->connections();
if (cons.size() != 1) {
cerr << "expected 1 connection, found " << cons.size() << endl;
throw std::logic_error("num_connections() != 1");
}
self->configure_read(
hdl, receive_policy::exactly(sizeof(atom_value) + sizeof(int)));
auto write = [=](atom_value type, int value) {
auto& buf = self->wr_buf(hdl);
auto first = reinterpret_cast<char*>(&type);
buf.insert(buf.end(), first, first + sizeof(atom_value));
first = reinterpret_cast<char*>(&value);
buf.insert(buf.end(), first, first + sizeof(int));
self->flush(hdl);
};
return {
[=](const connection_closed_msg&) {
CAF_MESSAGE("received connection_closed_msg");
self->quit();
},
[=](const new_data_msg& msg) {
CAF_MESSAGE("received new_data_msg");
atom_value type;
int value;
memcpy(&type, msg.buf.data(), sizeof(atom_value));
memcpy(&value, msg.buf.data() + sizeof(atom_value), sizeof(int));
self->send(buddy, type, value);
},
[=](ping_atom, int value) {
CAF_MESSAGE("received ping{" << value << "}");
write(ping_atom::value, value);
},
[=](pong_atom, int value) {
CAF_MESSAGE("received pong{" << value << "}");
write(pong_atom::value, value);
},
[=](const down_msg& dm) {
CAF_MESSAGE("received down_msg");
if (dm.source == buddy) {
self->quit(dm.reason);
}
}
};
}
using publish_atom = atom_constant<atom("publish")>;
using acceptor_broker = typed_broker<
replies_to<new_connection_msg>::with<void>,
replies_to<publish_atom>::with<uint16_t>
>;
acceptor_broker::behavior_type
peer_acceptor_fun(acceptor_broker* self, const actor& buddy) {
CAF_MESSAGE("peer_acceptor_fun");
return {
[=](const new_connection_msg& msg) {
CAF_MESSAGE("received `new_connection_msg`");
self->fork(peer_fun, msg.handle, buddy);
self->quit();
},
[=](publish_atom) {
return self->add_tcp_doorman(0, "127.0.0.1").second;
}
};
}
namespace {
void run_server(bool spawn_client, const char* bin_path) {
scoped_actor self;
auto serv = io::spawn_io_typed(peer_acceptor_fun, spawn(pong));
self->sync_send(serv, publish_atom::value).await(
[&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port);
if (spawn_client) {
auto child = detail::run_program(self, bin_path, "-n", "-s", "broker",
"--", "-c", port);
CAF_MESSAGE("block till child process has finished");
child.join();
}
}
);
self->await_all_other_actors_done();
self->receive(
[](const std::string& output) {
cout << endl << endl << "*** output of client program ***"
<< endl << output << endl;
}
);
}
}
CAF_TEST(test_typed_broker) {
auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc();
if (argv) {
message_builder{argv, argv + argc}.apply({
on("-c", arg_match) >> [&](const std::string& portstr) {
auto port = static_cast<uint16_t>(std::stoi(portstr));
auto p = spawn(ping, 10);
CAF_MESSAGE("spawn_io_client_typed...");
auto cl = spawn_io_client_typed(peer_fun, "localhost", port, p);
CAF_MESSAGE("spawn_io_client_typed finished");
anon_send(p, kickoff_atom::value, cl);
CAF_MESSAGE("`kickoff_atom` has been send");
},
on("-s") >> [&] {
run_server(false, argv[0]);
},
others >> [&] {
cerr << "usage: " << argv[0] << " [-c PORT]" << endl;
}
});
}
else {
run_server(true, caf::test::engine::path());
}
CAF_MESSAGE("block on `await_all_actors_done`");
await_all_actors_done();
CAF_MESSAGE("`await_all_actors_done` has finished");
shutdown();
}
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