Commit 3ce764a9 authored by Dominik Charousset's avatar Dominik Charousset

re-implemented sync_send for typed actors

parent 5e5ccb08
......@@ -174,10 +174,8 @@ cppa/tpartial_function.hpp
cppa/tuple_cast.hpp
cppa/type_lookup_table.hpp
cppa/typed_actor.hpp
cppa/typed_actor_base.hpp
cppa/typed_actor_ptr.hpp
cppa/typed_behavior.hpp
cppa/typed_behavior_stack_based.hpp
cppa/typed_event_based_actor.hpp
cppa/uniform_type_info.hpp
cppa/unit.hpp
......@@ -347,3 +345,5 @@ unit_testing/test_tuple.cpp
unit_testing/test_typed_spawn.cpp
unit_testing/test_uniform_type.cpp
unit_testing/test_yield_interface.cpp
cppa/typed_continue_helper.hpp
cppa/sync_sender.hpp
......@@ -46,6 +46,7 @@ namespace cppa {
class actor_addr;
class actor_proxy;
class local_actor;
class blocking_actor;
class event_based_actor;
......@@ -87,6 +88,7 @@ class actor : util::comparable<actor>
, util::comparable<actor, invalid_actor_t>
, util::comparable<actor, invalid_actor_addr_t> {
friend class local_actor;
friend class detail::raw_access;
public:
......
......@@ -33,52 +33,74 @@
#include "cppa/message_id.hpp"
#include "cppa/single_timeout.hpp"
#include "cppa/typed_behavior.hpp"
#include "cppa/behavior_policy.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/detail/behavior_stack.hpp"
namespace cppa {
/**
* @brief Mixin for actors using a stack-based message processing.
* @note This mixin implicitly includes {@link single_timeout}.
*/
template<class Base, class Subtype>
class behavior_stack_based : public single_timeout<Base, Subtype> {
template<class Base, class Subtype, class BehaviorType>
class behavior_stack_based_impl : public single_timeout<Base, Subtype> {
typedef single_timeout<Base, Subtype> super;
public:
typedef behavior_stack_based combined_type;
/**************************************************************************
* typedefs and constructor *
**************************************************************************/
typedef BehaviorType behavior_type;
typedef behavior_stack_based_impl combined_type;
typedef response_handle<behavior_stack_based_impl,
any_tuple,
nonblocking_response_handle_tag>
response_handle_type;
template <typename... Ts>
behavior_stack_based(Ts&&... args) : super(std::forward<Ts>(args)...) { }
behavior_stack_based_impl(Ts&&... vs) : super(std::forward<Ts>(vs)...) { }
inline void unbecome() {
m_bhvr_stack.pop_async_back();
/**************************************************************************
* become() member function family *
**************************************************************************/
void become(behavior_type bhvr) {
do_become(std::move(bhvr), true);
}
template<bool Discard>
void become(behavior_policy<Discard>, behavior_type bhvr) {
do_become(std::move(bhvr), Discard);
}
/**
* @brief Sets the actor's behavior and discards the previous behavior
* unless {@link keep_behavior} is given as first argument.
*/
template<typename T, typename... Ts>
inline typename std::enable_if<
!is_behavior_policy<typename util::rm_const_and_ref<T>::type>::value,
void
>::type
become(T&& arg, Ts&&... args) {
do_become(match_expr_convert(std::forward<T>(arg),
std::forward<Ts>(args)...),
do_become(behavior_type{std::forward<T>(arg),
std::forward<Ts>(args)...},
true);
}
template<bool Discard, typename... Ts>
inline void become(behavior_policy<Discard>, Ts&&... args) {
do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
void become(behavior_policy<Discard>, Ts&&... args) {
do_become(behavior_type{std::forward<Ts>(args)...}, Discard);
}
inline void unbecome() {
m_bhvr_stack.pop_async_back();
}
/**************************************************************************
* convenience member function for stack manipulation *
**************************************************************************/
inline bool has_behavior() const {
return m_bhvr_stack.empty() == false;
}
......@@ -88,10 +110,6 @@ class behavior_stack_based : public single_timeout<Base, Subtype> {
return m_bhvr_stack.back();
}
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
optional<behavior&> sync_handler(message_id msg_id) override {
return m_bhvr_stack.sync_handler(msg_id);
}
......@@ -100,38 +118,71 @@ class behavior_stack_based : public single_timeout<Base, Subtype> {
m_bhvr_stack.erase(mid);
}
void become_waiting_for(behavior bhvr, message_id mf) {
//CPPA_LOG_TRACE(CPPA_MARG(mf, integer_value));
if (bhvr.timeout().valid()) {
if (bhvr.timeout().valid()) {
this->reset_timeout();
this->request_timeout(bhvr.timeout());
}
this->bhvr_stack().push_back(std::move(bhvr), mf);
}
this->bhvr_stack().push_back(std::move(bhvr), mf);
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
void do_become(behavior bhvr, bool discard_old) {
//CPPA_LOG_TRACE(CPPA_ARG(discard_old));
//if (discard_old) m_bhvr_stack.pop_async_back();
//m_bhvr_stack.push_back(std::move(bhvr));
private:
void do_become(behavior_type bhvr, bool discard_old) {
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->reset_timeout();
if (bhvr.timeout().valid()) {
//CPPA_LOG_DEBUG("request timeout: " << bhvr.timeout().to_string());
this->request_timeout(bhvr.timeout());
}
this->m_bhvr_stack.push_back(std::move(bhvr));
this->m_bhvr_stack.push_back(std::move(unbox(bhvr)));
}
static inline behavior& unbox(behavior& arg) {
return arg;
}
template<typename... Ts>
static inline behavior& unbox(typed_behavior<Ts...>& arg) {
return arg.unbox();
}
// utility for getting a pointer-to-derived-type
Subtype* dptr() {
return static_cast<Subtype*>(this);
}
protected:
// utility for getting a const pointer-to-derived-type
const Subtype* dptr() const {
return static_cast<const Subtype*>(this);
}
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
};
/**
* @brief Mixin for actors using a stack-based message processing.
* @note This mixin implicitly includes {@link single_timeout}.
*/
template<class BehaviorType>
class behavior_stack_based {
public:
template<class Base, class Subtype>
class impl : public behavior_stack_based_impl<Base, Subtype, BehaviorType> {
typedef behavior_stack_based_impl<Base, Subtype, BehaviorType> super;
public:
typedef impl combined_type;
template<typename... Ts>
impl(Ts&&... args) : super(std::forward<Ts>(args)...) { }
};
};
} // namespace cppa
#endif // CPPA_BEHAVIOR_STACK_BASED_HPP
......@@ -34,14 +34,13 @@
#include "cppa/on.hpp"
#include "cppa/extend.hpp"
#include "cppa/behavior.hpp"
#include "cppa/typed_actor.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/sync_sender.hpp"
#include "cppa/typed_actor.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/detail/response_handle_util.hpp"
namespace cppa {
/**
......@@ -49,101 +48,13 @@ namespace cppa {
* receive rather than a behavior-stack based message processing.
* @extends local_actor
*/
class blocking_actor : public extend<local_actor>::with<mailbox_based> {
class blocking_actor
: public extend<local_actor, blocking_actor>::
with<mailbox_based,
sync_sender<blocking_response_handle_tag>::impl> {
public:
typedef blocking_response_handle_tag response_handle_tag;
typedef response_handle<blocking_actor> response_handle_type;
/**************************************************************************
* sync_send[_tuple](actor, ...) *
**************************************************************************/
/**
* @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
inline response_handle_type sync_send_tuple(message_priority prio,
const actor& dest,
any_tuple what) {
return {sync_send_tuple_impl(prio, dest, std::move(what)), this};
}
inline response_handle_type sync_send_tuple(const actor& dest, any_tuple what) {
return sync_send_tuple(message_priority::normal, dest, std::move(what));
}
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline response_handle_type sync_send(message_priority prio,
const actor& dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(prio, dest,
make_any_tuple(std::forward<Ts>(what)...));
}
template<typename... Ts>
inline response_handle_type sync_send(const actor& dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(message_priority::normal,
dest, make_any_tuple(std::forward<Ts>(what)...));
}
/**************************************************************************
* timed_sync_send[_tuple](actor, ...) *
**************************************************************************/
inline response_handle_type timed_sync_send_tuple(const util::duration& rtime,
const actor& dest,
any_tuple what) {
return {timed_sync_send_tuple_impl(message_priority::normal,
dest, rtime, std::move(what)),
this};
}
template<typename... Ts>
response_handle_type timed_sync_send(const actor& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(rtime, dest,
make_any_tuple(std::forward<Ts>(what)...));
}
/**************************************************************************
* sync_send[_tuple](typed_actor<...>, ...) *
**************************************************************************/
template<typename... Rs, typename... Ts>
inline response_handle_type sync_send_tuple(message_priority prio,
const typed_actor<Rs...>& dest,
cow_tuple<Ts...> what) {
return {sync_send_tuple_impl(prio, dest, std::move(what)), this};
}
template<typename... Rs, typename... Ts>
inline response_handle_type sync_send_tuple(const typed_actor<Rs...>& dest,
cow_tuple<Ts...> what) {
return sync_send_tuple(message_priority::normal, dest, std::move(what));
}
/**************************************************************************
* utility stuff and receive() member function family *
**************************************************************************/
......
......@@ -35,6 +35,7 @@
#include "cppa/extend.hpp"
#include "cppa/logging.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/sync_sender.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/behavior_stack_based.hpp"
......@@ -54,8 +55,10 @@ class event_based_actor;
*
* @extends local_actor
*/
class event_based_actor : public extend<local_actor>::
with<mailbox_based, behavior_stack_based> {
class event_based_actor : public extend<local_actor, event_based_actor>::
with<mailbox_based,
behavior_stack_based<behavior>::impl,
sync_sender<nonblocking_response_handle_tag>::impl> {
protected:
......@@ -69,74 +72,6 @@ class event_based_actor : public extend<local_actor>::
*/
void forward_to(const actor& whom);
public:
typedef nonblocking_response_handle_tag response_handle_tag;
typedef response_handle<event_based_actor> untyped_response_handle;
/**
* @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
inline untyped_response_handle sync_send_tuple(const actor& whom,
any_tuple what) {
return {sync_send_tuple_impl(message_priority::normal,
whom, std::move(what)),
this};
}
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
inline untyped_response_handle sync_send(const actor& whom, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(whom, make_any_tuple(std::forward<Ts>(what)...));
}
/**
* @brief Sends a synchronous message with timeout @p rtime to @p whom.
* @param whom Receiver of the message.
* @param rtime Relative time until this messages times out.
* @param what Message content as tuple.
*/
untyped_response_handle timed_sync_send_tuple(const util::duration& rtime,
const actor& whom,
any_tuple what) {
return {timed_sync_send_tuple_impl(message_priority::normal, whom,
rtime, std::move(what)),
this};
}
/**
* @brief Sends a synchronous message with timeout @p rtime to @p whom.
* @param whom Receiver of the message.
* @param rtime Relative time until this messages times out.
* @param what Message elements.
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<typename... Ts>
untyped_response_handle timed_sync_send(const actor& whom,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(rtime, whom,
make_any_tuple(std::forward<Ts>(what)...));
}
};
} // namespace cppa
......
......@@ -62,7 +62,8 @@ broker_ptr init_and_launch(broker_ptr);
* and other components in the network.
* @extends local_actor
*/
class broker : public extend<local_actor>::with<behavior_stack_based> {
class broker : public extend<local_actor>::
with<behavior_stack_based<behavior>::impl> {
friend class policy::sequential_invoke;
......
......@@ -477,7 +477,9 @@ class local_actor : public extend<abstract_actor>::with<memory_cached> {
const typed_actor<Signatures...>& whom,
cow_tuple<Ts...>&& what) {
check_typed_input(whom, what);
return sync_send_tuple_impl(mp, whom.m_ptr, any_tuple{std::move(what)});
return sync_send_tuple_impl(mp,
actor{whom.m_ptr.get()},
any_tuple{std::move(what)});
}
// returns 0 if last_dequeued() is an asynchronous or sync request message,
......
......@@ -36,9 +36,11 @@
#include "cppa/message_id.hpp"
#include "cppa/typed_behavior.hpp"
#include "cppa/continue_helper.hpp"
#include "cppa/typed_continue_helper.hpp"
#include "cppa/util/type_list.hpp"
#include "cppa/detail/typed_actor_util.hpp"
#include "cppa/detail/response_handle_util.hpp"
namespace cppa {
......@@ -46,12 +48,14 @@ namespace cppa {
/**
* @brief This tag identifies response handles featuring a
* nonblocking API by providing a @p then member function.
* @relates response_handle
*/
struct nonblocking_response_handle_tag { };
/**
* @brief This tag identifies response handles featuring a
* blocking API by providing an @p await member function.
* @relates response_handle
*/
struct blocking_response_handle_tag { };
......@@ -64,12 +68,14 @@ struct blocking_response_handle_tag { };
* @tparam Tag Either {@link nonblocking_response_handle_tag} or
* {@link blocking_response_handle_tag}.
*/
template<class Self,
class Result = any_tuple,
class Tag = typename Self::response_handle_tag>
class response_handle { // default impl for nonblocking_response_handle_tag
template<class Self, class Result, class Tag>
class response_handle;
friend Self;
/******************************************************************************
* nonblocking + untyped *
******************************************************************************/
template<class Self>
class response_handle<Self, any_tuple, nonblocking_response_handle_tag> {
public:
......@@ -79,70 +85,78 @@ class response_handle { // default impl for nonblocking_response_handle_tag
response_handle& operator=(const response_handle&) = default;
/**
* @brief Sets @p bhvr as event-handler for the response message.
*/
inline continue_helper then(behavior bhvr) {
return then_impl<Result>(std::move(bhvr));
m_self->bhvr_stack().push_back(std::move(bhvr), m_mid);
return {m_mid, m_self};
}
/**
* @brief Sets @p mexpr as event-handler for the response message.
*/
template<typename... Cs, typename... Ts>
continue_helper then(const match_expr<Cs...>& arg, const Ts&... args) {
return then_impl<Result>({arg, args...});
return then(behavior{arg, args...});
}
/**
* @brief Sets @p fun as event-handler for the response message, calls
* <tt>self->handle_sync_failure()</tt> if the response message
* is an 'EXITED' or 'VOID' message.
*/
template<typename... Fs>
typename std::enable_if<
util::all_callable<Fs...>::value,
continue_helper
>::type
then(Fs... fs) {
return then_impl<Result>(detail::fs2bhvr(m_self, fs...));
return then(detail::fs2bhvr(m_self, fs...));
}
response_handle(const message_id& mid, Self* self)
: m_mid(mid), m_self(self) { }
private:
template<typename R>
typename std::enable_if<
std::is_same<R, any_tuple>::value,
continue_helper
>::type
then_impl(behavior&& bhvr) {
m_self->bhvr_stack().push_back(std::move(bhvr), m_mid);
return {m_mid, m_self};
}
message_id m_mid;
template<typename R>
typename std::enable_if<
util::is_type_list<R>::value,
continue_helper
Self* m_self;
};
/******************************************************************************
* nonblocking + typed *
******************************************************************************/
template<class Self, typename... Ts>
class response_handle<Self, util::type_list<Ts...>, nonblocking_response_handle_tag> {
public:
response_handle() = delete;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
template<typename F>
typed_continue_helper<
typename detail::lifted_result_type<
typename util::get_callable_trait<F>::result_type
>::type
then_impl(behavior&& bhvr) {
m_self->bhvr_stack().push_back(std::move(bhvr), m_mid);
>
then(F fun) {
detail::assert_types<util::type_list<Ts...>, F>();
m_self->bhvr_stack().push_back(behavior{on_arg_match >> fun}, m_mid);
return {m_mid, m_self};
}
response_handle(const message_id& mid, Self* self)
: m_mid(mid), m_self(self) { }
private:
message_id m_mid;
Self* m_self;
};
template<class Self, typename Result>
class response_handle<Self, Result, blocking_response_handle_tag> {
friend Self;
/******************************************************************************
* blocking + untyped *
******************************************************************************/
template<class Self>
class response_handle<Self, any_tuple, blocking_response_handle_tag> {
public:
......@@ -152,47 +166,75 @@ class response_handle<Self, Result, blocking_response_handle_tag> {
response_handle& operator=(const response_handle&) = default;
/**
* @brief Blocks until the response arrives and then executes @p pfun.
*/
void await(behavior& pfun) {
m_self->dequeue_response(pfun, m_mid);
}
/**
* @copydoc await(behavior&)
*/
inline void await(behavior&& pfun) {
behavior arg{std::move(pfun)};
await(arg);
behavior tmp{std::move(pfun)};
await(tmp);
}
/**
* @brief Blocks until the response arrives and then executes
* the corresponding handler.
*/
template<typename... Cs, typename... Ts>
void await(const match_expr<Cs...>& arg, const Ts&... args) {
await(match_expr_convert(arg, args...));
}
/**
* @brief Blocks until the response arrives and then executes
* the corresponding handler.
* @note Calls <tt>self->handle_sync_failure()</tt> if the response
* message is an 'EXITED' or 'VOID' message.
*/
template<typename... Fs>
typename std::enable_if<util::all_callable<Fs...>::value>::type
await(Fs... fs) {
await(detail::fs2bhvr(m_self, fs...));
}
response_handle(const message_id& mid, Self* self)
: m_mid(mid), m_self(self) { }
private:
message_id m_mid;
Self* m_self;
};
/******************************************************************************
* blocking + typed *
******************************************************************************/
template<class Self, typename... Ts>
class response_handle<Self, util::type_list<Ts...>, blocking_response_handle_tag> {
public:
typedef util::type_list<Ts...> result_types;
response_handle() = delete;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
template<typename F>
void await(F fun) {
typedef typename util::tl_map<
typename util::get_callable_trait<F>::arg_types,
util::rm_const_and_ref
>::type
arg_types;
static constexpr size_t fun_args = util::tl_size<arg_types>::value;
static_assert(fun_args <= util::tl_size<result_types>::value,
"functor takes too much arguments");
typedef typename util::tl_right<result_types, fun_args>::type recv_types;
static_assert(std::is_same<arg_types, recv_types>::value,
"wrong functor signature");
behavior tmp = detail::fs2bhvr(m_self, fun);
m_self->dequeue_response(tmp, m_mid);
}
response_handle(const message_id& mid, Self* self)
: m_mid(mid), m_self(self) { }
private:
message_id m_mid;
Self* m_self;
......
......@@ -224,7 +224,7 @@ class functor_based_typed_actor : public typed_event_based_actor<Sigs...> {
public:
typedef functor_based_typed_actor* pointer;
typedef typed_event_based_actor<Sigs...>* pointer;
typedef typename super::behavior_type behavior_type;
typedef std::function<typed_behavior<Sigs...> ()> no_arg_fun;
......@@ -264,6 +264,14 @@ struct actor_handle_from_typed_behavior<typed_behavior<Signatures...>> {
typedef typed_actor<Signatures...> type;
};
template<typename SignatureList>
struct actor_handle_from_signature_list;
template<typename... Signatures>
struct actor_handle_from_signature_list<util::type_list<Signatures...>> {
typedef typed_actor<Signatures...> type;
};
} // namespace detail
template<spawn_options Options = no_spawn_options, typename F>
......@@ -280,6 +288,17 @@ spawn_typed(F fun) {
std::move(fun));
}
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
typename detail::actor_handle_from_signature_list<
typename Impl::signatures
>::type
spawn_typed(Ts&&... args) {
return detail::spawn_fwd_args<Impl, Options>(
[&](Impl*) { },
std::forward<Ts>(args)...);
}
/*
template<class Impl, spawn_options Opts = no_spawn_options, typename... Ts>
typename Impl::typed_pointer_type spawn_typed(Ts&&... args) {
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_SYNC_SENDER_HPP
#define CPPA_SYNC_SENDER_HPP
#include "cppa/actor.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/message_priority.hpp"
#include "cppa/util/duration.hpp"
namespace cppa {
template<class Base, class Subtype, class ResponseHandleTag>
class sync_sender_impl : public Base {
public:
typedef response_handle<Subtype,
any_tuple,
ResponseHandleTag>
response_handle_type;
/**************************************************************************
* sync_send[_tuple](actor, ...) *
**************************************************************************/
/**
* @brief Sends @p what as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message content as tuple.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
response_handle_type sync_send_tuple(message_priority prio,
const actor& dest,
any_tuple what) {
return {dptr()->sync_send_tuple_impl(prio, dest, std::move(what)),
dptr()};
}
response_handle_type sync_send_tuple(const actor& dest, any_tuple what) {
return sync_send_tuple(message_priority::normal, dest, std::move(what));
}
/**
* @brief Sends <tt>{what...}</tt> as a synchronous message to @p whom.
* @param whom Receiver of the message.
* @param what Message elements.
* @returns A handle identifying a future to the response of @p whom.
* @warning The returned handle is actor specific and the response to the
* sent message cannot be received by another actor.
* @pre <tt>sizeof...(Ts) > 0</tt>
* @throws std::invalid_argument if <tt>whom == nullptr</tt>
*/
template<typename... Ts>
response_handle_type sync_send(message_priority prio,
const actor& dest,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(prio, dest,
make_any_tuple(std::forward<Ts>(what)...));
}
template<typename... Ts>
response_handle_type sync_send(const actor& dest, Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return sync_send_tuple(message_priority::normal,
dest, make_any_tuple(std::forward<Ts>(what)...));
}
/**************************************************************************
* timed_sync_send[_tuple](actor, ...) *
**************************************************************************/
response_handle_type timed_sync_send_tuple(message_priority prio,
const actor& dest,
const util::duration& rtime,
any_tuple what) {
return {dptr()->timed_sync_send_tuple_impl(prio, dest, rtime,
std::move(what)),
dptr()};
}
response_handle_type timed_sync_send_tuple(const actor& dest,
const util::duration& rtime,
any_tuple what) {
return {dptr()->timed_sync_send_tuple_impl(message_priority::normal,
dest, rtime,
std::move(what)),
dptr()};
}
template<typename... Ts>
response_handle_type timed_sync_send(message_priority prio,
const actor& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(prio, dest, rtime,
make_any_tuple(std::forward<Ts>(what)...));
}
template<typename... Ts>
response_handle_type timed_sync_send(const actor& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(message_priority::normal, dest, rtime,
make_any_tuple(std::forward<Ts>(what)...));
}
/**************************************************************************
* sync_send[_tuple](typed_actor<...>, ...) *
**************************************************************************/
template<typename... Rs, typename... Ts>
response_handle<
Subtype,
typename detail::deduce_output_type<
util::type_list<Rs...>,
util::type_list<Ts...>
>::type,
ResponseHandleTag
>
sync_send_tuple(message_priority prio,
const typed_actor<Rs...>& dest,
cow_tuple<Ts...> what) {
return {dptr()->sync_send_tuple_impl(prio, dest, std::move(what)),
dptr()};
}
template<typename... Rs, typename... Ts>
response_handle<
Subtype,
typename detail::deduce_output_type<
util::type_list<Rs...>,
util::type_list<Ts...>
>::type,
ResponseHandleTag
>
sync_send_tuple(const typed_actor<Rs...>& dest, cow_tuple<Ts...> what) {
return sync_send_tuple(message_priority::normal, dest, std::move(what));
}
template<typename... Rs, typename... Ts>
response_handle<
Subtype,
typename detail::deduce_output_type<
util::type_list<Rs...>,
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Ts
>::type
>::type...
>::type,
ResponseHandleTag
>
sync_send(message_priority prio,
const typed_actor<Rs...>& dest,
Ts&&... what) {
return sync_send_tuple(prio, dest,
make_cow_tuple(std::forward<Ts>(what)...));
}
template<typename... Rs, typename... Ts>
response_handle<
Subtype,
typename detail::deduce_output_type<
util::type_list<Rs...>,
util::type_list<
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Ts
>::type
>::type...
>
>::type,
ResponseHandleTag
>
sync_send(const typed_actor<Rs...>& dest,
Ts&&... what) {
return sync_send_tuple(message_priority::normal, dest,
make_cow_tuple(std::forward<Ts>(what)...));
}
/**************************************************************************
* timed_sync_send[_tuple](typed_actor<...>, ...) *
**************************************************************************/
template<typename... Rs, typename... Ts>
response_handle<
Subtype,
typename detail::deduce_output_type<
util::type_list<Rs...>,
util::type_list<Ts...>
>::type,
ResponseHandleTag
>
timed_sync_send_tuple(message_priority prio,
const typed_actor<Rs...>& dest,
const util::duration& rtime,
cow_tuple<Ts...> what) {
return {dptr()->timed_sync_send_tuple_impl(prio, dest, rtime,
std::move(what)),
dptr()};
}
template<typename... Rs, typename... Ts>
response_handle<
Subtype,
typename detail::deduce_output_type<
util::type_list<Rs...>,
util::type_list<Ts...>
>::type,
ResponseHandleTag
>
timed_sync_send_tuple(const typed_actor<Rs...>& dest,
const util::duration& rtime,
cow_tuple<Ts...> what) {
return {dptr()->timed_sync_send_tuple_impl(message_priority::normal,
dest, rtime,
std::move(what)),
dptr()};
}
template<typename... Rs, typename... Ts>
response_handle<
Subtype,
typename detail::deduce_output_type<
util::type_list<Rs...>,
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Ts
>::type
>::type...
>::type,
ResponseHandleTag
>
timed_sync_send(message_priority prio,
const typed_actor<Rs...>& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(prio, dest, rtime,
make_any_tuple(std::forward<Ts>(what)...));
}
template<typename... Rs, typename... Ts>
response_handle<
Subtype,
typename detail::deduce_output_type<
util::type_list<Rs...>,
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Ts
>::type
>::type...
>::type,
ResponseHandleTag
>
timed_sync_send(const typed_actor<Rs...>& dest,
const util::duration& rtime,
Ts&&... what) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return timed_sync_send_tuple(message_priority::normal, dest, rtime,
make_any_tuple(std::forward<Ts>(what)...));
}
private:
inline Subtype* dptr() {
return static_cast<Subtype*>(this);
}
};
template<class ResponseHandleTag>
class sync_sender {
public:
template<class Base, class Subtype>
class impl : public sync_sender_impl<Base, Subtype, ResponseHandleTag> {
typedef sync_sender_impl<Base, Subtype, ResponseHandleTag> super;
public:
typedef impl combined_type;
template<typename... Ts>
impl(Ts&&... args) : super(std::forward<Ts>(args)...) { }
};
};
} // namespace cppa
#endif // CPPA_SYNC_SENDER_HPP
......@@ -35,6 +35,7 @@
#include "cppa/replies_to.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/typed_behavior.hpp"
namespace cppa {
......@@ -43,6 +44,12 @@ class local_actor;
struct invalid_actor_addr_t;
template<typename... Signatures>
class typed_event_based_actor;
/**
* @brief Identifies a strongly typed actor.
*/
template<typename... Signatures>
class typed_actor : util::comparable<typed_actor<Signatures...>>
, util::comparable<typed_actor<Signatures...>, actor_addr>
......@@ -52,6 +59,22 @@ class typed_actor : util::comparable<typed_actor<Signatures...>>
public:
/**
* @brief Identifies the behavior type actors of this kind use
* for their behavior stack.
*/
typedef typed_behavior<Signatures...> behavior_type;
/**
* @brief Identifies pointers to instances of this kind of actor.
*/
typedef typed_event_based_actor<Signatures...>* pointer;
/**
* @brief Identifies the base class for this kind of actor.
*/
typedef typed_event_based_actor<Signatures...> impl;
typedef util::type_list<Signatures...> signatures;
typed_actor() = default;
......@@ -71,8 +94,8 @@ class typed_actor : util::comparable<typed_actor<Signatures...>>
return *this;
}
template<template<typename...> class Impl, typename... OtherSignatures>
typed_actor(intrusive_ptr<Impl<OtherSignatures...>> other) {
template<class Impl>
typed_actor(intrusive_ptr<Impl> other) {
set(other);
}
......@@ -105,9 +128,9 @@ class typed_actor : util::comparable<typed_actor<Signatures...>>
m_ptr = other.m_ptr;
}
template<template<typename...> class Impl, typename... OtherSignatures>
inline void set(intrusive_ptr<Impl<OtherSignatures...>>& other) {
check_signatures<signatures, util::type_list<OtherSignatures...>>();
template<class Impl>
inline void set(intrusive_ptr<Impl>& other) {
check_signatures<signatures, typename Impl::signatures>();
m_ptr = std::move(other);
}
......
......@@ -38,20 +38,20 @@
namespace cppa {
template<class, typename...>
class typed_behavior_stack_based;
template<typename... Signatures>
class typed_actor;
template<class Base, class Subtype, class BehaviorType>
class behavior_stack_based_impl;
template<typename... Signatures>
class typed_behavior {
template<typename... OtherSignatures>
friend class typed_actor;
template<class, typename...>
friend class typed_behavior_stack_based;
template<class Base, class Subtype, class BehaviorType>
friend class behavior_stack_based_impl;
typed_behavior() = delete;
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation; either version 2.1 of the License, *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_TYPED_BEHAVIOR_STACK_BASED_HPP
#define CPPA_TYPED_BEHAVIOR_STACK_BASED_HPP
#include <utility>
#include "cppa/single_timeout.hpp"
#include "cppa/behavior_policy.hpp"
#include "cppa/typed_event_based_actor.hpp"
#include "cppa/util/type_traits.hpp"
#include "cppa/detail/behavior_stack.hpp"
namespace cppa {
/**
* @brief Mixin for actors using a typed stack-based message processing.
* @note This mixin implicitly includes {@link single_timeout}.
* @note Subtype must provide a typedef for @p behavior_type.
*/
template<class Base, typename... Signatures>
class typed_behavior_stack_based : public extend<Base>::template
with<single_timeout> {
public:
typedef typed_behavior<Signatures...> behavior_type;
/**
* @brief Sets the actor's behavior and discards the previous behavior
* unless {@link keep_behavior} is given as first argument.
*/
void become(behavior_type bhvr) {
do_become(std::move(bhvr), true);
}
template<bool Discard>
void become(behavior_policy<Discard>, behavior_type bhvr) {
do_become(std::move(bhvr), Discard);
}
template<typename T, typename... Ts>
inline typename std::enable_if<
!is_behavior_policy<typename util::rm_const_and_ref<T>::type>::value,
void
>::type
become(T&& arg, Ts&&... args) {
do_become(behavior_type{std::forward<T>(arg),
std::forward<Ts>(args)...},
true);
}
template<bool Discard, typename... Ts>
void become(behavior_policy<Discard>, Ts&&... args) {
do_become(behavior_type{std::forward<Ts>(args)...}, Discard);
}
inline void unbecome() {
m_bhvr_stack.pop_async_back();
}
inline bool has_behavior() const {
return m_bhvr_stack.empty() == false;
}
inline behavior& get_behavior() {
CPPA_REQUIRE(m_bhvr_stack.empty() == false);
return m_bhvr_stack.back();
}
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
optional<behavior&> sync_handler(message_id msg_id) override {
return m_bhvr_stack.sync_handler(msg_id);
}
inline void remove_handler(message_id mid) {
m_bhvr_stack.erase(mid);
}
private:
void do_become(behavior_type bhvr, bool discard_old) {
if (discard_old) this->m_bhvr_stack.pop_async_back();
this->reset_timeout();
if (bhvr.timeout().valid()) {
this->request_timeout(bhvr.timeout());
}
this->m_bhvr_stack.push_back(std::move(bhvr.unbox()));
}
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
};
} // namespace cppa
#endif // CPPA_TYPED_BEHAVIOR_STACK_BASED_HPP
......@@ -28,71 +28,44 @@
\******************************************************************************/
#ifndef CPPA_TYPED_ACTOR_HPP
#define CPPA_TYPED_ACTOR_HPP
#ifndef CPPA_TYPED_CONTINUE_HELPER_HPP
#define CPPA_TYPED_CONTINUE_HELPER_HPP
#include "cppa/replies_to.hpp"
#include "cppa/typed_behavior.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/continue_helper.hpp"
#include "cppa/detail/typed_actor_util.hpp"
#include "cppa/util/type_traits.hpp"
namespace cppa {
template<typename... Signatures>
class typed_actor_ptr;
template<typename... Signatures>
class typed_actor : public event_based_actor {
template<typename OutputList>
class typed_continue_helper {
public:
using signatures = util::type_list<Signatures...>;
using behavior_type = typed_behavior<Signatures...>;
using typed_pointer_type = typed_actor_ptr<Signatures...>;
protected:
typedef int message_id_wrapper_tag;
virtual behavior_type make_behavior() = 0;
typed_continue_helper(message_id mid, local_actor* self) : m_ch(mid, self) { }
void init() final {
auto bhvr = make_behavior();
m_bhvr_stack.push_back(std::move(bhvr.unbox()));
template<typename F>
typed_continue_helper<typename util::get_callable_trait<F>::result_type>
continue_with(F fun) {
detail::assert_types<OutputList, F>();
m_ch.continue_with(std::move(fun));
return {m_ch};
}
void do_become(behavior&&, bool) final {
CPPA_LOG_ERROR("typed actors are not allowed to call become()");
quit(exit_reason::unallowed_function_call);
inline message_id get_message_id() const {
return m_ch.get_message_id();
}
};
} // namespace cppa
namespace cppa { namespace detail {
template<typename... Signatures>
class default_typed_actor : public typed_actor<Signatures...> {
public:
template<typename... Cases>
default_typed_actor(match_expr<Cases...> expr) : m_bhvr(std::move(expr)) { }
protected:
typed_behavior<Signatures...> make_behavior() override {
return m_bhvr;
}
typed_continue_helper(continue_helper ch) : m_ch(std::move(ch)) { }
private:
typed_behavior<Signatures...> m_bhvr;
continue_helper m_ch;
};
} } // namespace cppa::detail
} // namespace cppa
#endif // CPPA_TYPED_ACTOR_HPP
#endif // CPPA_TYPED_CONTINUE_HELPER_HPP
......@@ -33,17 +33,23 @@
#include "cppa/replies_to.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/sync_sender.hpp"
#include "cppa/mailbox_based.hpp"
#include "cppa/typed_behavior.hpp"
#include "cppa/typed_behavior_stack_based.hpp"
#include "cppa/behavior_stack_based.hpp"
namespace cppa {
template<typename... Signatures>
class typed_event_based_actor : public typed_behavior_stack_based<
extend<local_actor>::template
with<mailbox_based>,
Signatures...> {
class typed_event_based_actor
: public extend<local_actor, typed_event_based_actor<Signatures...>>::template
with<mailbox_based,
behavior_stack_based<
typed_behavior<Signatures...>
>::template impl,
sync_sender<
nonblocking_response_handle_tag
>::template impl> {
public:
......
......@@ -37,11 +37,13 @@ using namespace cppa;
struct my_request { int a; int b; };
typedef typed_actor<replies_to<my_request>::with<bool>> server_type;
bool operator==(const my_request& lhs, const my_request& rhs) {
return lhs.a == rhs.a && lhs.b == rhs.b;
}
typed_behavior<replies_to<my_request>::with<bool>> typed_server() {
server_type::behavior_type typed_server1() {
return {
on_arg_match >> [](const my_request& req) {
return req.a == req.b;
......@@ -49,33 +51,41 @@ typed_behavior<replies_to<my_request>::with<bool>> typed_server() {
};
}
/*
typed_actor_ptr<replies_to<my_request>::with<bool>>
spawn_typed_server() {
return spawn_typed(
on_arg_match >> [](const my_request& req) {
return req.a == req.b;
}
);
server_type::behavior_type typed_server2(server_type::pointer) {
return typed_server1();
}
class typed_testee : public typed_actor<replies_to<my_request>::with<bool>> {
class typed_server3 : public server_type::impl {
protected:
public:
behavior_type make_behavior() override {
return (
on_arg_match >> [](const my_request& req) {
return req.a == req.b;
typed_server3(const string& line, actor buddy) {
send(buddy, line);
}
);
behavior_type make_behavior() override {
return typed_server2(this);
}
};
*/
void test_typed_spawn() {
auto ts = spawn_typed(typed_server);
void client(event_based_actor* self, actor parent, server_type serv) {
self->sync_send(serv, my_request{0, 0}).then(
[](bool value) {
CPPA_CHECK_EQUAL(value, true);
}
)
.continue_with([=] {
self->sync_send(serv, my_request{10, 20}).then(
[=](bool value) {
CPPA_CHECK_EQUAL(value, false);
self->send(parent, atom("passed"));
}
);
});
}
void test_typed_spawn(server_type ts) {
scoped_actor self;
self->send(ts, my_request{1, 2});
self->receive(
......@@ -89,13 +99,47 @@ void test_typed_spawn() {
CPPA_CHECK_EQUAL(value, true);
}
);
self->sync_send(ts, my_request{10, 20}).await(
[](bool value) {
CPPA_CHECK_EQUAL(value, false);
}
);
self->sync_send(ts, my_request{0, 0}).await(
[](bool value) {
CPPA_CHECK_EQUAL(value, true);
}
);
self->spawn<monitored>(client, self, ts);
self->receive(
on(atom("passed")) >> CPPA_CHECKPOINT_CB()
);
self->receive(
on_arg_match >> [](const down_msg& dmsg) {
CPPA_CHECK_EQUAL(dmsg.reason, exit_reason::normal);
}
);
self->send_exit(ts, exit_reason::user_shutdown);
}
int main() {
CPPA_TEST(test_typed_spawn);
announce<my_request>(&my_request::a, &my_request::b);
test_typed_spawn();
test_typed_spawn(spawn_typed(typed_server1));
CPPA_CHECKPOINT();
await_all_actors_done();
CPPA_CHECKPOINT();
test_typed_spawn(spawn_typed(typed_server2));
CPPA_CHECKPOINT();
await_all_actors_done();
CPPA_CHECKPOINT();
{
scoped_actor self;
test_typed_spawn(spawn_typed<typed_server3>("hi there", self));
self->receive(
on("hi there") >> CPPA_CHECKPOINT_CB()
);
}
CPPA_CHECKPOINT();
await_all_actors_done();
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