Commit 41752506 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/typed_broker' into develop

parents 513306eb fcf3f755
...@@ -38,27 +38,26 @@ namespace caf { ...@@ -38,27 +38,26 @@ namespace caf {
* @tparam HasSyncSend Configures whether this base class extends `sync_sender`. * @tparam HasSyncSend Configures whether this base class extends `sync_sender`.
* @extends local_actor * @extends local_actor
*/ */
template <class BehaviorType, bool HasSyncSend> template <class BehaviorType, bool HasSyncSend, class Base = local_actor>
class abstract_event_based_actor : public class abstract_event_based_actor : public Base {
std::conditional<
HasSyncSend,
extend<local_actor>
::with<mixin::sync_sender<nonblocking_response_handle_tag>::template impl>,
local_actor
>::type {
public: public:
using behavior_type = BehaviorType; using behavior_type = BehaviorType;
template <class... Ts>
abstract_event_based_actor(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
/**************************************************************************** /****************************************************************************
* become() member function family * * become() member function family *
****************************************************************************/ ****************************************************************************/
void become(behavior_type bhvr) { void become(behavior_type bhvr) {
this->do_become(std::move(unbox(bhvr)), true); this->do_become(std::move(bhvr.unbox()), true);
} }
void become(const keep_behavior_t&, behavior_type bhvr) { void become(const keep_behavior_t&, behavior_type bhvr) {
this->do_become(std::move(unbox(bhvr)), false); this->do_become(std::move(bhvr.unbox()), false);
} }
template <class T, class... Ts> template <class T, class... Ts>
...@@ -68,27 +67,33 @@ public: ...@@ -68,27 +67,33 @@ public:
>::type >::type
become(T&& x, Ts&&... xs) { become(T&& x, Ts&&... xs) {
behavior_type bhvr{std::forward<T>(x), std::forward<Ts>(xs)...}; behavior_type bhvr{std::forward<T>(x), std::forward<Ts>(xs)...};
this->do_become(std::move(unbox(bhvr)), true); this->do_become(std::move(bhvr.unbox()), true);
} }
template <class... Ts> template <class... Ts>
void become(const keep_behavior_t&, Ts&&... xs) { void become(const keep_behavior_t&, Ts&&... xs) {
behavior_type bhvr{std::forward<Ts>(xs)...}; behavior_type bhvr{std::forward<Ts>(xs)...};
this->do_become(std::move(unbox(bhvr)), false); this->do_become(std::move(bhvr.unbox()), false);
} }
void unbecome() { void unbecome() {
this->bhvr_stack_.pop_back(); this->bhvr_stack_.pop_back();
} }
};
private: template <class BehaviorType, class Base>
class abstract_event_based_actor<BehaviorType, true, Base>
: public extend<abstract_event_based_actor<BehaviorType, false, Base>>
::template with<mixin::sync_sender<nonblocking_response_handle_tag>::template impl> {
public:
using super
= typename extend<abstract_event_based_actor<BehaviorType, false, Base>>::
template with<mixin::sync_sender<nonblocking_response_handle_tag>::
template impl>;
template <class... Ts> template <class... Ts>
static behavior& unbox(typed_behavior<Ts...>& x) { abstract_event_based_actor(Ts&&... xs)
return x.unbox(); : super(std::forward<Ts>(xs)...) {
} // nop
static inline behavior& unbox(behavior& x) {
return x;
} }
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
...@@ -78,6 +78,26 @@ intrusive_ptr<C> spawn_class(execution_unit* host, ...@@ -78,6 +78,26 @@ intrusive_ptr<C> spawn_class(execution_unit* host,
detail::spawn_fwd<Ts>(xs)...); detail::spawn_fwd<Ts>(xs)...);
} }
template <spawn_options Os, class C, class BeforeLaunch, class F, class... Ts>
intrusive_ptr<C> spawn_functor_impl(execution_unit* eu, BeforeLaunch cb,
F fun, Ts&&... xs) {
constexpr bool has_blocking_base =
std::is_base_of<blocking_actor, C>::value;
static_assert(has_blocking_base || ! has_blocking_api_flag(Os),
"blocking functor-based actors "
"need to be spawned using the blocking_api flag");
static_assert(! has_blocking_base || has_blocking_api_flag(Os),
"non-blocking functor-based actors "
"cannot be spawned using the blocking_api flag");
detail::init_fun_factory<C, F> fac;
auto init = fac(std::move(fun), std::forward<Ts>(xs)...);
auto bl = [&](C* ptr) {
cb(ptr);
ptr->initial_behavior_fac(std::move(init));
};
return spawn_class<C, Os>(eu, bl);
}
/// Called by `spawn` when used to create a functor-based actor (usually /// Called by `spawn` when used to create a functor-based actor (usually
/// should not be called by users of the library). This function /// should not be called by users of the library). This function
/// selects a proper implementation class and then delegates to `spawn_class`. /// selects a proper implementation class and then delegates to `spawn_class`.
...@@ -86,7 +106,7 @@ actor spawn_functor(execution_unit* eu, BeforeLaunch cb, F fun, Ts&&... xs) { ...@@ -86,7 +106,7 @@ actor spawn_functor(execution_unit* eu, BeforeLaunch cb, F fun, Ts&&... xs) {
using trait = typename detail::get_callable_trait<F>::type; using trait = typename detail::get_callable_trait<F>::type;
using arg_types = typename trait::arg_types; using arg_types = typename trait::arg_types;
using first_arg = typename detail::tl_head<arg_types>::type; using first_arg = typename detail::tl_head<arg_types>::type;
using base_class = using impl =
typename std::conditional< typename std::conditional<
std::is_pointer<first_arg>::value, std::is_pointer<first_arg>::value,
typename std::remove_pointer<first_arg>::type, typename std::remove_pointer<first_arg>::type,
...@@ -96,21 +116,8 @@ actor spawn_functor(execution_unit* eu, BeforeLaunch cb, F fun, Ts&&... xs) { ...@@ -96,21 +116,8 @@ actor spawn_functor(execution_unit* eu, BeforeLaunch cb, F fun, Ts&&... xs) {
event_based_actor event_based_actor
>::type >::type
>::type; >::type;
constexpr bool has_blocking_base = return spawn_functor_impl<Os, impl>(eu, cb, std::move(fun),
std::is_base_of<blocking_actor, base_class>::value; std::forward<Ts>(xs)...);
static_assert(has_blocking_base || ! has_blocking_api_flag(Os),
"blocking functor-based actors "
"need to be spawned using the blocking_api flag");
static_assert(! has_blocking_base || has_blocking_api_flag(Os),
"non-blocking functor-based actors "
"cannot be spawned using the blocking_api flag");
detail::init_fun_factory<base_class, F> fac;
auto init = fac(std::move(fun), std::forward<Ts>(xs)...);
auto bl = [&](base_class* ptr) {
cb(ptr);
ptr->initial_behavior_fac(std::move(init));
};
return spawn_class<base_class, Os>(eu, bl);
} }
/// @ingroup ActorCreation /// @ingroup ActorCreation
...@@ -230,13 +237,8 @@ spawn_typed_functor(execution_unit* eu, BeforeLaunch cb, F fun, Ts&&... xs) { ...@@ -230,13 +237,8 @@ spawn_typed_functor(execution_unit* eu, BeforeLaunch cb, F fun, Ts&&... xs) {
typename detail::get_callable_trait<F>::arg_types typename detail::get_callable_trait<F>::arg_types
>::type >::type
>::type; >::type;
detail::init_fun_factory<impl, F> fac; return spawn_functor_impl<Os, impl>(eu, cb, std::move(fun),
auto init = fac(std::move(fun), std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
auto bl = [&](impl* ptr) {
cb(ptr);
ptr->initial_behavior_fac(std::move(init));
};
return spawn_class<impl, Os>(eu, bl);
} }
/// Returns a new typed actor from a functor. The first element /// Returns a new typed actor from a functor. The first element
......
...@@ -39,6 +39,13 @@ struct invalid_actor_addr_t; ...@@ -39,6 +39,13 @@ struct invalid_actor_addr_t;
template <class... Sigs> template <class... Sigs>
class typed_event_based_actor; class typed_event_based_actor;
namespace io {
template <class... Sigs>
class typed_broker;
} // namespace io
/// Identifies a strongly typed actor. /// Identifies a strongly typed actor.
/// @tparam Sigs Signature of this actor as `replies_to<...>::with<...>` /// @tparam Sigs Signature of this actor as `replies_to<...>::with<...>`
/// parameter pack. /// parameter pack.
...@@ -75,6 +82,12 @@ public: ...@@ -75,6 +82,12 @@ public:
/// Identifies the base class for this kind of actor. /// Identifies the base class for this kind of actor.
using base = typed_event_based_actor<Sigs...>; using base = typed_event_based_actor<Sigs...>;
/// Identifies pointers to brokers implementing this interface.
using broker_pointer = io::typed_broker<Sigs...>*;
/// Identifies the base class of brokers implementing this interface.
using broker_base = io::typed_broker<Sigs...>;
typed_actor() = default; typed_actor() = default;
typed_actor(typed_actor&&) = default; typed_actor(typed_actor&&) = default;
typed_actor(const typed_actor&) = default; typed_actor(const typed_actor&) = default;
......
...@@ -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/local_actor.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 local_actor {
public:
using buffer_type = std::vector<char>;
class continuation;
void enqueue(const actor_addr&, message_id,
message, execution_unit*) override;
void enqueue(mailbox_element_ptr, execution_unit*) override;
void launch(execution_unit* eu, bool lazy, bool hide);
void cleanup(uint32_t reason);
/// 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 broker_;
}
servant(abstract_broker* ptr);
void disconnect(bool invoke_disconnect_message);
bool disconnected_;
abstract_broker* 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 hdl_;
}
void io_failure(network::operation op) override;
protected:
virtual buffer_type& rd_buf() = 0;
inline new_data_msg& read_msg() {
return read_msg_.get_as_mutable<new_data_msg>(0);
}
inline const new_data_msg& read_msg() const {
return 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 hdl_;
message 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 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 accept_msg_.get_as_mutable<new_connection_msg>(0);
}
inline const new_connection_msg& accept_msg() const {
return accept_msg_.get_as<new_connection_msg>(0);
}
accept_handle hdl_;
message 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 scribes_.size();
}
std::vector<connection_handle> connections() const;
/// @cond PRIVATE
inline void add_scribe(const scribe_pointer& ptr) {
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) {
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);
void invoke_message(mailbox_element_ptr& msg);
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 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, scribes_);
}
// throws on error
inline doorman& by_id(accept_handle hdl) { return by_id(hdl, doormen_); }
bool invoke_message_from_cache();
std::map<accept_handle, doorman_pointer> doormen_;
std::map<connection_handle, scribe_pointer> scribes_;
middleman& mm_;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> 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"
......
...@@ -56,7 +56,7 @@ public: ...@@ -56,7 +56,7 @@ public:
actor_proxy_ptr make_proxy(const node_id&, actor_id) override; actor_proxy_ptr make_proxy(const node_id&, actor_id) override;
class payload_writer { class payload_writer {
public: public:
payload_writer() = default; payload_writer() = default;
payload_writer(const payload_writer&) = default; payload_writer(const payload_writer&) = default;
payload_writer& operator=(const payload_writer&) = default; payload_writer& operator=(const payload_writer&) = default;
......
...@@ -27,176 +27,17 @@ ...@@ -27,176 +27,17 @@
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.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_event_based_actor<behavior, false,
abstract_broker> {
public: public:
using super = abstract_event_based_actor<behavior, false>; using super = abstract_event_based_actor<behavior, false, abstract_broker>;
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 broker_.get();
}
servant(broker* ptr);
void set_broker(broker* ptr);
void disconnect(bool invoke_disconnect_message);
bool disconnected_;
intrusive_ptr<broker> 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 hdl_;
}
void io_failure(network::operation op) override;
protected:
virtual buffer_type& rd_buf() = 0;
inline new_data_msg& read_msg() {
return read_msg_.get_as_mutable<new_data_msg>(0);
}
inline const new_data_msg& read_msg() const {
return 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 hdl_;
message 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 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 accept_msg_.get_as_mutable<new_connection_msg>(0);
}
inline const new_connection_msg& accept_msg() const {
return accept_msg_.get_as<new_connection_msg>(0);
}
accept_handle hdl_;
message accept_msg_;
};
using doorman_pointer = intrusive_ptr<doorman>;
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 scribes_.size();
}
std::vector<connection_handle> connections() const;
/// @cond PRIVATE
void initialize() override;
template <class F, class... Ts> template <class F, class... Ts>
actor fork(F fun, connection_handle hdl, Ts&&... xs) { actor fork(F fun, connection_handle hdl, Ts&&... xs) {
...@@ -221,74 +62,7 @@ public: ...@@ -221,74 +62,7 @@ public:
fun, hdl, std::forward<Ts>(xs)...); fun, hdl, std::forward<Ts>(xs)...);
} }
inline void add_scribe(const scribe_pointer& ptr) { void initialize() override;
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) {
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);
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,
message, 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);
void launch(execution_unit* eu, bool lazy, bool hide);
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();
...@@ -296,50 +70,10 @@ protected: ...@@ -296,50 +70,10 @@ protected:
broker(middleman& parent_ref); broker(middleman& parent_ref);
virtual behavior make_behavior(); virtual behavior make_behavior();
/// Can be overridden to perform cleanup code before the
/// broker closes all its connections.
virtual void on_exit();
/// @endcond
inline middleman& parent() {
return mm_;
}
network::multiplexer& backend();
private:
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, scribes_);
}
// throws on error
inline doorman& by_id(accept_handle hdl) { return by_id(hdl, doormen_); }
bool invoke_message_from_cache();
void erase_io(int id);
void erase_acceptor(int id);
std::map<accept_handle, doorman_pointer> doormen_;
std::map<connection_handle, scribe_pointer> scribes_;
middleman& mm_;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_;
}; };
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;
......
...@@ -59,27 +59,27 @@ public: ...@@ -59,27 +59,27 @@ public:
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;
......
...@@ -278,26 +278,26 @@ public: ...@@ -278,26 +278,26 @@ public:
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;
......
...@@ -60,16 +60,19 @@ public: ...@@ -60,16 +60,19 @@ public:
/// 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;
/// Tries to create an unbound TCP doorman running `port`, optionally /// Tries to create an unbound TCP doorman running `port`, optionally
...@@ -81,17 +84,18 @@ public: ...@@ -81,17 +84,18 @@ public:
/// 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
/// accepting only connections from IP address `in`. /// accepting only connections from IP address `in`.
/// @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;
/// Simple wrapper for runnables /// Simple wrapper for runnables
...@@ -106,7 +110,7 @@ public: ...@@ -106,7 +110,7 @@ public:
/// Makes sure the multipler does not exit its event loop until /// Makes sure the multipler does not exit its event loop until
/// the destructor of `supervisor` has been called. /// the destructor of `supervisor` has been called.
class supervisor { class supervisor {
public: public:
virtual ~supervisor(); virtual ~supervisor();
}; };
......
...@@ -24,10 +24,11 @@ ...@@ -24,10 +24,11 @@
#include <functional> #include <functional>
#include "caf/spawn.hpp" #include "caf/spawn.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/mailbox_element.hpp" #include "caf/io/typed_broker.hpp"
#include "caf/io/abstract_broker.hpp"
#include "caf/io/connection_handle.hpp" #include "caf/io/connection_handle.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
...@@ -35,49 +36,118 @@ ...@@ -35,49 +36,118 @@
namespace caf { namespace caf {
namespace io { namespace io {
/// Spawns a new functor-based broker. /// @cond PRIVATE
template <spawn_options Os = no_spawn_options,
typename F = std::function<void(broker*)>, class... Ts>
actor spawn_io(F fun, Ts&&... xs) {
detail::init_fun_factory<broker, F> fac;
auto init = fac(std::move(fun), std::forward<Ts>(xs)...);
auto bl = [&](broker* ptr) {
ptr->initial_behavior_fac(std::move(init));
};
return spawn_class<broker>(nullptr, bl);
}
/// Spawns a new functor-based broker connecting to `host:port. template <spawn_options Os, class Impl, class F, class... Ts>
template <spawn_options Os = no_spawn_options, intrusive_ptr<Impl> spawn_io_client_impl(F fun, const std::string& host,
typename F = std::function<void(broker*)>, class... Ts> uint16_t port, Ts&&... xs) {
actor spawn_io_client(F fun, const std::string& host,
uint16_t port, Ts&&... xs) {
// works around an issue with GCC 4.8 that could not handle // works around an issue with GCC 4.8 that could not handle
// variadic template parameter packs inside lambdas // variadic template parameter packs inside lambdas
auto args = std::forward_as_tuple(std::forward<Ts>(xs)...); auto args = std::forward_as_tuple(std::forward<Ts>(xs)...);
auto bl = [&](broker* ptr) { auto bl = [&](Impl* 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);
detail::init_fun_factory<broker, F> fac; detail::init_fun_factory<Impl, F> fac;
auto init = detail::apply_args_prefixed(fac, detail::get_indices(args), auto init = detail::apply_args_prefixed(fac, detail::get_indices(args),
args, std::move(fun), hdl); args, std::move(fun), hdl);
ptr->initial_behavior_fac(std::move(init)); ptr->initial_behavior_fac(std::move(init));
}; };
return spawn_class<broker>(nullptr, bl); return spawn_class<Impl>(nullptr, bl);
} }
/// Spawns a new broker as server running on given `port`. template <spawn_options Os, class Impl, class F, class... Ts>
template <spawn_options Os = no_spawn_options, intrusive_ptr<Impl> spawn_io_server_impl(F fun, uint16_t port, Ts&&... xs) {
class F = std::function<void(broker*)>, class... Ts> detail::init_fun_factory<Impl, F> fac;
actor spawn_io_server(F fun, uint16_t port, Ts&&... xs) {
detail::init_fun_factory<broker, F> fac;
auto init = fac(std::move(fun), std::forward<Ts>(xs)...); auto init = fac(std::move(fun), std::forward<Ts>(xs)...);
auto bl = [&](broker* ptr) { auto bl = [&](broker* ptr) {
auto mm = middleman::instance(); auto mm = middleman::instance();
mm->backend().add_tcp_doorman(ptr, port); mm->backend().add_tcp_doorman(ptr, port);
ptr->initial_behavior_fac(std::move(init)); ptr->initial_behavior_fac(std::move(init));
}; };
return spawn_class<broker>(nullptr, bl); return spawn_class<Impl>(nullptr, bl);
}
/// @endcond
/// Spawns a new functor-based broker.
template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts>
actor spawn_io(F fun, Ts&&... xs) {
return spawn_functor<Os>(nullptr, empty_before_launch_callback{},
std::move(fun), std::forward<Ts>(xs)...);
}
/// Spawns a new functor-based broker connecting to `host:port`.
template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts>
actor spawn_io_client(F fun, const std::string& host,
uint16_t port, Ts&&... xs) {
return spawn_io_client_impl<Os, broker>(std::move(fun), host, port,
std::forward<Ts>(xs)...);
}
/// Spawns a new broker as server running on given `port`.
template <spawn_options Os = no_spawn_options,
class F = std::function<void(broker*)>, class... Ts>
actor spawn_io_server(F fun, uint16_t port, Ts&&... xs) {
return spawn_io_server_impl<Os, broker>(std::move(fun), port,
std::forward<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 impl =
typename infer_typed_broker_base<
typename detail::get_callable_trait<F>::result_type,
typename detail::tl_head<
typename detail::get_callable_trait<F>::arg_types
>::type
>::type;
return spawn_functor_impl<Os, impl>(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 impl_class = typename std::remove_pointer<first_arg>::type;
return spawn_io_client_impl<Os, impl_class>(std::move(fun), host, port,
std::forward<Ts>(xs)...);
}
/// 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 impl_class = typename std::remove_pointer<first_arg>::type;
return spawn_io_server_impl<Os, impl_class>(std::move(fun), port,
std::forward<Ts>(xs)...);
} }
} // namespace io } // namespace io
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 {
template <class... Sigs>
class typed_broker;
/// Infers the appropriate base class for a functor-based typed actor
/// from the result and the first argument of the functor.
template <class Result, class FirstArg>
struct infer_typed_broker_base;
template <class... Sigs, class FirstArg>
struct infer_typed_broker_base<typed_behavior<Sigs...>, FirstArg> {
using type = typed_broker<Sigs...>;
};
template <class... Sigs>
struct infer_typed_broker_base<void, typed_event_based_actor<Sigs...>*> {
using type = typed_broker<Sigs...>;
};
/// A typed broker mediates between actor systems and other components in
/// the network.
/// @extends local_actor
template <class... Sigs>
class typed_broker : public abstract_event_based_actor<typed_behavior<Sigs...>,
false, abstract_broker> {
public:
using signatures = detail::type_list<Sigs...>;
using behavior_type = typed_behavior<Sigs...>;
using super = abstract_event_based_actor<behavior_type, false,
abstract_broker>;
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 {
this->is_initialized(true);
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(! bhvr, "make_behavior() did not return a behavior, "
<< "has_behavior() = "
<< std::boolalpha << 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->do_become(std::move(bhvr.unbox()), true);
}
}
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 = this->scribes_.find(hdl);
if (i == this->scribes_.end()) {
CAF_LOG_ERROR("invalid handle");
throw std::invalid_argument("invalid handle");
}
auto sptr = i->second;
CAF_ASSERT(sptr->hdl() == hdl);
this->scribes_.erase(i);
using impl =
typename infer_typed_broker_base<
typename detail::get_callable_trait<F>::result_type,
typename detail::tl_head<
typename detail::get_callable_trait<F>::arg_types
>::type
>::type;
return spawn_functor_impl<no_spawn_options, impl>(
nullptr, [&sptr](abstract_broker* forked) {
sptr->set_broker(forked);
forked->add_scribe(sptr);
},
std::move(fun), hdl, std::forward<Ts>(xs)...);
}
protected:
typed_broker() {
// nop
}
typed_broker(middleman& parent_ref) : abstract_broker(parent_ref) {
// nop
}
virtual behavior_type make_behavior() {
if (this->initial_behavior_fac_) {
auto bhvr = this->initial_behavior_fac_(this);
this->initial_behavior_fac_ = nullptr;
if (bhvr)
this->do_become(std::move(bhvr), true);
}
return behavior_type::make_empty_behavior();
}
};
} // namespace io
} // namespace caf
#endif // CAF_IO_TYPED_BROKER_HPP
This diff is collapsed.
...@@ -101,22 +101,35 @@ connection_handle asio_multiplexer::new_tcp_scribe(const std::string& host, ...@@ -101,22 +101,35 @@ connection_handle asio_multiplexer::new_tcp_scribe(const std::string& host,
return connection_handle::from_int(id); return connection_handle::from_int(id);
} }
<<<<<<< HEAD
void asio_multiplexer::assign_tcp_scribe(broker* self, connection_handle hdl) { void asio_multiplexer::assign_tcp_scribe(broker* self, connection_handle hdl) {
std::lock_guard<std::mutex> lock(mtx_sockets_); std::lock_guard<std::mutex> lock(mtx_sockets_);
auto itr = unassigned_sockets_.find(hdl.id()); auto itr = unassigned_sockets_.find(hdl.id());
if (itr != unassigned_sockets_.end()) { if (itr != unassigned_sockets_.end()) {
=======
void asio_multiplexer::assign_tcp_scribe(abstract_broker* self, connection_handle hdl) {
std::lock_guard<std::mutex> lock(mtx_sockets_);
auto itr = unassigned_sockets_.find(hdl.id());
if (itr != unassigned_sockets_.end()) {
>>>>>>> 84f43f6f825114da12757fe0e7018b9464b97614
add_tcp_scribe(self, std::move(itr->second)); add_tcp_scribe(self, std::move(itr->second));
unassigned_sockets_.erase(itr); unassigned_sockets_.erase(itr);
} }
} }
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("");
<<<<<<< HEAD
class impl : public broker::scribe { class impl : public broker::scribe {
public: public:
impl(broker* ptr, Socket&& s) impl(broker* ptr, Socket&& s)
=======
class impl : public abstract_broker::scribe {
public:
impl(abstract_broker* ptr, Socket&& s)
>>>>>>> 84f43f6f825114da12757fe0e7018b9464b97614
: scribe(ptr, network::conn_hdl_from_socket(s)), : scribe(ptr, network::conn_hdl_from_socket(s)),
launched_(false), launched_(false),
stream_(s.get_io_service()) { stream_(s.get_io_service()) {
...@@ -129,8 +142,13 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self, ...@@ -129,8 +142,13 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self,
launch(); launch();
} }
} }
<<<<<<< HEAD
broker::buffer_type& wr_buf() override { return stream_.wr_buf(); } broker::buffer_type& wr_buf() override { return stream_.wr_buf(); }
broker::buffer_type& rd_buf() override { return stream_.rd_buf(); } broker::buffer_type& rd_buf() override { return stream_.rd_buf(); }
=======
abstract_broker::buffer_type& wr_buf() override { return stream_.wr_buf(); }
abstract_broker::buffer_type& rd_buf() override { return stream_.rd_buf(); }
>>>>>>> 84f43f6f825114da12757fe0e7018b9464b97614
void stop_reading() override { void stop_reading() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
stream_.stop_reading(); stream_.stop_reading();
...@@ -147,16 +165,16 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self, ...@@ -147,16 +165,16 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self,
stream_.start(this); stream_.start(this);
} }
private: private:
bool launched_; bool launched_;
stream<Socket> stream_; stream<Socket> 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;
...@@ -171,7 +189,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(broker* self, ...@@ -171,7 +189,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));
...@@ -190,7 +208,7 @@ asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) { ...@@ -190,7 +208,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(mtx_acceptors_); std::lock_guard<std::mutex> lock(mtx_acceptors_);
auto itr = unassigned_acceptors_.find(hdl.id()); auto itr = unassigned_acceptors_.find(hdl.id());
...@@ -201,13 +219,19 @@ void asio_multiplexer::assign_tcp_doorman(broker* self, accept_handle hdl) { ...@@ -201,13 +219,19 @@ 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);
<<<<<<< HEAD
class impl : public broker::doorman { class impl : public broker::doorman {
public: public:
impl(broker* ptr, default_socket_acceptor&& s, impl(broker* ptr, default_socket_acceptor&& s,
=======
class impl : public abstract_broker::doorman {
public:
impl(abstract_broker* ptr, default_socket_acceptor&& s,
>>>>>>> 84f43f6f825114da12757fe0e7018b9464b97614
network::asio_multiplexer& am) network::asio_multiplexer& am)
: doorman(ptr, network::accept_hdl_from_socket(s)), : doorman(ptr, network::accept_hdl_from_socket(s)),
acceptor_(am, s.get_io_service()) { acceptor_(am, s.get_io_service()) {
...@@ -231,16 +255,16 @@ asio_multiplexer::add_tcp_doorman(broker* self, ...@@ -231,16 +255,16 @@ asio_multiplexer::add_tcp_doorman(broker* self,
acceptor_.start(this); acceptor_.start(this);
} }
private: private:
network::acceptor<default_socket_acceptor> acceptor_; network::acceptor<default_socket_acceptor> 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;
...@@ -255,7 +279,7 @@ accept_handle asio_multiplexer::add_tcp_doorman(broker* self, ...@@ -255,7 +279,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);
......
This diff is collapsed.
...@@ -417,7 +417,7 @@ namespace network { ...@@ -417,7 +417,7 @@ namespace network {
int presult; int presult;
CAF_LOG_DEBUG("poll() " << pollset_.size() << " sockets"); CAF_LOG_DEBUG("poll() " << pollset_.size() << " sockets");
# ifdef CAF_WINDOWS # ifdef CAF_WINDOWS
presult = ::WSAPoll(pollset_.data(), presult = ::WSAPoll(pollset_.data(),
static_cast<ULONG>(pollset_.size()), -1); static_cast<ULONG>(pollset_.size()), -1);
# else # else
presult = ::poll(pollset_.data(), presult = ::poll(pollset_.data(),
...@@ -622,7 +622,7 @@ default_multiplexer& get_multiplexer_singleton() { ...@@ -622,7 +622,7 @@ default_multiplexer& get_multiplexer_singleton() {
multiplexer::supervisor_ptr default_multiplexer::make_supervisor() { multiplexer::supervisor_ptr default_multiplexer::make_supervisor() {
class impl : public multiplexer::supervisor { class impl : public multiplexer::supervisor {
public: public:
explicit impl(default_multiplexer* thisptr) : this_(thisptr) { explicit impl(default_multiplexer* thisptr) : this_(thisptr) {
// nop // nop
} }
...@@ -630,7 +630,7 @@ multiplexer::supervisor_ptr default_multiplexer::make_supervisor() { ...@@ -630,7 +630,7 @@ multiplexer::supervisor_ptr default_multiplexer::make_supervisor() {
auto ptr = this_; auto ptr = this_;
ptr->dispatch([=] { ptr->close_pipe(); }); ptr->dispatch([=] { ptr->close_pipe(); });
} }
private: private:
default_multiplexer* this_; default_multiplexer* this_;
}; };
return supervisor_ptr{new impl(this)}; return supervisor_ptr{new impl(this)};
...@@ -727,12 +727,12 @@ void default_multiplexer::dispatch_runnable(runnable_ptr ptr) { ...@@ -727,12 +727,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)),
launched_(false), launched_(false),
stream_(s.backend()) { stream_(s.backend()) {
...@@ -743,10 +743,10 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self, ...@@ -743,10 +743,10 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
stream_.configure_read(config); stream_.configure_read(config);
if (! launched_) launch(); if (! launched_) launch();
} }
broker::buffer_type& wr_buf() override { abstract_broker::buffer_type& wr_buf() override {
return stream_.wr_buf(); return stream_.wr_buf();
} }
broker::buffer_type& rd_buf() override { abstract_broker::buffer_type& rd_buf() override {
return stream_.rd_buf(); return stream_.rd_buf();
} }
void stop_reading() override { void stop_reading() override {
...@@ -764,23 +764,23 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self, ...@@ -764,23 +764,23 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
launched_ = true; launched_ = true;
stream_.start(this); stream_.start(this);
} }
private: private:
bool launched_; bool launched_;
stream<default_socket> stream_; stream<default_socket> 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)),
acceptor_(s.backend()) { acceptor_(s.backend()) {
acceptor_.init(std::move(s)); acceptor_.init(std::move(s));
...@@ -802,10 +802,10 @@ default_multiplexer::add_tcp_doorman(broker* self, ...@@ -802,10 +802,10 @@ default_multiplexer::add_tcp_doorman(broker* self,
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
acceptor_.start(this); acceptor_.start(this);
} }
private: private:
network::acceptor<default_socket_acceptor> acceptor_; network::acceptor<default_socket_acceptor> 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();
} }
...@@ -816,19 +816,19 @@ connection_handle default_multiplexer::new_tcp_scribe(const std::string& host, ...@@ -816,19 +816,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)
...@@ -844,17 +844,17 @@ default_multiplexer::new_tcp_doorman(uint16_t port, const char* in, ...@@ -844,17 +844,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;
......
...@@ -34,6 +34,8 @@ using namespace std; ...@@ -34,6 +34,8 @@ using namespace std;
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
namespace {
using ping_atom = caf::atom_constant<caf::atom("ping")>; 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 publish_atom = caf::atom_constant<caf::atom("publish")>; using publish_atom = caf::atom_constant<caf::atom("publish")>;
...@@ -55,12 +57,12 @@ void ping(event_based_actor* self, size_t num_pings) { ...@@ -55,12 +57,12 @@ void ping(event_based_actor* self, size_t num_pings) {
} }
return std::make_tuple(ping_atom::value, value + 1); return std::make_tuple(ping_atom::value, value + 1);
}, },
others >> [&] { others >> [=] {
CAF_TEST_ERROR("Unexpected message: " CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message())); << to_string(self->current_message()));
}); });
}, },
others >> [&] { others >> [=] {
CAF_TEST_ERROR("Unexpected message: " CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message())); << to_string(self->current_message()));
} }
...@@ -82,7 +84,7 @@ void pong(event_based_actor* self) { ...@@ -82,7 +84,7 @@ void pong(event_based_actor* self) {
CAF_MESSAGE("received down_msg{" << dm.reason << "}"); CAF_MESSAGE("received down_msg{" << dm.reason << "}");
self->quit(dm.reason); self->quit(dm.reason);
}, },
others >> [&] { others >> [=] {
CAF_TEST_ERROR("Unexpected message: " CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message())); << to_string(self->current_message()));
} }
...@@ -90,7 +92,7 @@ void pong(event_based_actor* self) { ...@@ -90,7 +92,7 @@ void pong(event_based_actor* self) {
// reply to 'ping' // reply to 'ping'
return std::make_tuple(pong_atom::value, value); return std::make_tuple(pong_atom::value, value);
}, },
others >> [&] { others >> [=] {
CAF_TEST_ERROR("Unexpected message: " CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message())); << to_string(self->current_message()));
} }
...@@ -146,7 +148,7 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -146,7 +148,7 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
self->quit(dm.reason); self->quit(dm.reason);
} }
}, },
others >> [&] { others >> [=] {
CAF_TEST_ERROR("Unexpected message: " CAF_TEST_ERROR("Unexpected message: "
<< to_string(self->current_message())); << to_string(self->current_message()));
} }
...@@ -194,6 +196,8 @@ void run_server(bool spawn_client, const char* bin_path) { ...@@ -194,6 +196,8 @@ void run_server(bool spawn_client, const char* bin_path) {
); );
} }
} // namespace <anonymous>
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;
namespace {
using publish_atom = atom_constant<atom("publish")>;
using ping_atom = caf::atom_constant<atom("ping")>;
using pong_atom = caf::atom_constant<atom("pong")>;
using kickoff_atom = caf::atom_constant<atom("kickoff")>;
using peer = typed_actor<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>>;
using acceptor = typed_actor<replies_to<new_connection_msg>::with<void>,
replies_to<publish_atom>::with<uint16_t>>;
behavior ping(event_based_actor* self, size_t num_pings) {
CAF_MESSAGE("num_pings: " << num_pings);
auto count = std::make_shared<size_t>(0);
return {
[=](kickoff_atom, const peer& pong) {
CAF_MESSAGE("received `kickoff_atom`");
self->send(pong, ping_atom::value, 1);
self->become(
[=](pong_atom, int value) -> std::tuple<ping_atom, 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()));
}
};
}
behavior pong(event_based_actor* self) {
CAF_MESSAGE("pong actor started");
return {
[=](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()));
}
};
}
peer::behavior_type peer_fun(peer::broker_pointer 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);
}
}
};
}
acceptor::behavior_type acceptor_fun(acceptor::broker_pointer 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;
}
};
}
void run_server(bool spawn_client, const char* bin_path) {
scoped_actor self;
auto serv = io::spawn_io_typed(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",
"typed_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;
}
);
}
} // namespace <anonymous>
CAF_TEST(test_typed_broker) {
auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc();
if (argv) {
uint16_t port = 0;
auto r = message_builder(argv, argv + argc).extract_opts({
{"client-port,c", "set port for IO client", port},
{"server,s", "run in server mode"}
});
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
cout << r.error << endl << endl << r.helptext << endl;
return;
}
if (r.opts.count("client-port") > 0) {
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");
} else {
// run in server mode
run_server(false, argv[0]);
}
}
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