Commit a66f8fe2 authored by Dominik Charousset's avatar Dominik Charousset

improved message_future API

this patch allows to pass any number of functors to `then`/`await` and
moves `receive_response` and `handle_response` functions to message_future.hpp
parent 71f28b45
...@@ -577,16 +577,6 @@ message_future timed_sync_send(const actor_ptr& whom, ...@@ -577,16 +577,6 @@ message_future timed_sync_send(const actor_ptr& whom,
make_any_tuple(std::forward<Args>(what)...)); make_any_tuple(std::forward<Args>(what)...));
} }
/**
* @brief Handles a synchronous response message in an event-based way.
* @param handle A future for a synchronous response.
* @throws std::invalid_argument if given behavior does not has a valid
* timeout definition
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
*/
sync_recv_helper receive_response(const message_future& handle);
/** /**
* @brief Sends a message to the sender of the last received message. * @brief Sends a message to the sender of the last received message.
* @param what Message content as a tuple. * @param what Message content as a tuple.
......
...@@ -56,10 +56,10 @@ namespace cppa { ...@@ -56,10 +56,10 @@ namespace cppa {
class scheduler; class scheduler;
class message_future; class message_future;
class local_scheduler; class local_scheduler;
class sync_handle_helper;
namespace detail { class receive_policy; } namespace detail { class receive_policy; }
template<bool DiscardOld> template<bool DiscardOld>
struct behavior_policy { static const bool discard_old = DiscardOld; }; struct behavior_policy { static const bool discard_old = DiscardOld; };
...@@ -89,27 +89,6 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t(); ...@@ -89,27 +89,6 @@ constexpr keep_behavior_t keep_behavior = keep_behavior_t();
} // namespace <anonymous> } // namespace <anonymous>
#endif // CPPA_DOCUMENTATION #endif // CPPA_DOCUMENTATION
struct sync_recv_helper {
typedef void (*callback_type)(behavior&, message_id_t);
message_id_t m_handle;
callback_type m_fun;
inline sync_recv_helper(message_id_t handle, callback_type fun)
: m_handle(handle), m_fun(fun) { }
template<typename... Expression>
inline void operator()(Expression&&... mexpr) const {
auto bhvr = match_expr_convert(std::forward<Expression>(mexpr)...);
static_assert(std::is_same<decltype(bhvr), behavior>::value,
"no timeout specified");
if (bhvr.timeout().valid() == false || bhvr.timeout().is_zero()) {
throw std::invalid_argument("specified timeout is invalid or zero");
}
else if (!m_handle.valid() || !m_handle.is_response()) {
throw std::logic_error("handle does not point to a response");
}
m_fun(bhvr, m_handle);
}
};
class message_future; class message_future;
//inline sync_recv_helper receive_response(message_future); //inline sync_recv_helper receive_response(message_future);
...@@ -307,16 +286,6 @@ class local_actor : public memory_cached_mixin<actor> { ...@@ -307,16 +286,6 @@ class local_actor : public memory_cached_mixin<actor> {
true); true);
} }
/**
* @brief Receives a synchronous response message.
* @param handle A future for a synchronous response.
* @throws std::invalid_argument if given behavior does not has a valid
* timeout definition
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
*/
sync_recv_helper handle_response(const message_future& handle);
/** /**
* @brief Returns to a previous behavior if available. * @brief Returns to a previous behavior if available.
*/ */
...@@ -404,6 +373,10 @@ class local_actor : public memory_cached_mixin<actor> { ...@@ -404,6 +373,10 @@ class local_actor : public memory_cached_mixin<actor> {
# ifndef CPPA_DOCUMENTATION # ifndef CPPA_DOCUMENTATION
// backward compatiblity, handle_response was part of local_actor
// until version 0.6
sync_handle_helper handle_response(const message_future& mf);
local_actor(bool is_scheduled = false); local_actor(bool is_scheduled = false);
virtual bool initialized() = 0; virtual bool initialized() = 0;
......
...@@ -39,6 +39,8 @@ ...@@ -39,6 +39,8 @@
#include "cppa/message_id.hpp" #include "cppa/message_id.hpp"
#include "cppa/local_actor.hpp" #include "cppa/local_actor.hpp"
#include "cppa/util/callable_trait.hpp"
namespace cppa { namespace cppa {
/** /**
...@@ -77,9 +79,9 @@ class message_future { ...@@ -77,9 +79,9 @@ class message_future {
* @brief Sets @p mexpr as event-handler for the response message. * @brief Sets @p mexpr as event-handler for the response message.
*/ */
template<typename... Cases, typename... Args> template<typename... Cases, typename... Args>
continue_helper then(const match_expr<Cases...>& arg0, const Args&... args) { continue_helper then(const match_expr<Cases...>& a0, const Args&... as) {
consistency_check(); check_consistency();
self->become_waiting_for(match_expr_convert(arg0, args...), m_mid); self->become_waiting_for(match_expr_convert(a0, as...), m_mid);
return {m_mid}; return {m_mid};
} }
...@@ -88,7 +90,7 @@ class message_future { ...@@ -88,7 +90,7 @@ class message_future {
*/ */
template<typename... Cases, typename... Args> template<typename... Cases, typename... Args>
void await(const match_expr<Cases...>& arg0, const Args&... args) { void await(const match_expr<Cases...>& arg0, const Args&... args) {
consistency_check(); check_consistency();
self->dequeue_response(match_expr_convert(arg0, args...), m_mid); self->dequeue_response(match_expr_convert(arg0, args...), m_mid);
} }
...@@ -97,11 +99,14 @@ class message_future { ...@@ -97,11 +99,14 @@ class message_future {
* <tt>self->handle_sync_failure()</tt> if the response message * <tt>self->handle_sync_failure()</tt> if the response message
* is an 'EXITED' or 'VOID' message. * is an 'EXITED' or 'VOID' message.
*/ */
template<typename F> template<typename... Fs>
typename std::enable_if<util::is_callable<F>::value,continue_helper>::type typename std::enable_if<
then(F fun) { util::all_callable<Fs...>::value,
consistency_check(); continue_helper
self->become_waiting_for(bhvr_from_fun(fun), m_mid); >::type
then(Fs... fs) {
check_consistency();
self->become_waiting_for(fs2bhvr(std::move(fs)...), m_mid);
return {m_mid}; return {m_mid};
} }
...@@ -110,10 +115,11 @@ class message_future { ...@@ -110,10 +115,11 @@ class message_future {
* calls <tt>self->handle_sync_failure()</tt> if the response * calls <tt>self->handle_sync_failure()</tt> if the response
* message is an 'EXITED' or 'VOID' message. * message is an 'EXITED' or 'VOID' message.
*/ */
template<typename F> template<typename... Fs>
typename std::enable_if<util::is_callable<F>::value>::type await(F fun) { typename std::enable_if<util::all_callable<Fs...>::value>::type
consistency_check(); await(Fs... fs) {
self->dequeue_response(bhvr_from_fun(fun), m_mid); check_consistency();
self->dequeue_response(fs2bhvr(std::move(fs)...), m_mid);
} }
/** /**
...@@ -134,28 +140,21 @@ class message_future { ...@@ -134,28 +140,21 @@ class message_future {
message_id_t m_mid; message_id_t m_mid;
template<typename F> template<typename... Fs>
behavior bhvr_from_fun(F fun) { behavior fs2bhvr(Fs... fs) {
auto handle_sync_failure = []() -> bool {
self->handle_sync_failure();
return false; // do not treat this as a match to cause a
// continuation to be invoked only in case
// `fun` was invoked
};
auto handle_sync_timeout = []() -> bool { auto handle_sync_timeout = []() -> bool {
self->handle_sync_timeout(); self->handle_sync_timeout();
return false; return false;
}; };
return { return {
on<atom("EXITED"), std::uint32_t>() >> handle_sync_failure, on<atom("EXITED"),std::uint32_t>() >> skip_message,
on(atom("VOID")) >> skip_message,
on(atom("TIMEOUT")) >> handle_sync_timeout, on(atom("TIMEOUT")) >> handle_sync_timeout,
on(atom("VOID")) >> handle_sync_failure, (on(any_vals, arg_match) >> fs)...
on(any_vals, arg_match) >> fun,
others() >> handle_sync_failure
}; };
} }
void consistency_check() { void check_consistency() {
if (!m_mid.valid() || !m_mid.is_response()) { if (!m_mid.valid() || !m_mid.is_response()) {
throw std::logic_error("handle does not point to a response"); throw std::logic_error("handle does not point to a response");
} }
...@@ -166,6 +165,62 @@ class message_future { ...@@ -166,6 +165,62 @@ class message_future {
}; };
class sync_handle_helper {
public:
inline sync_handle_helper(const message_future& mf) : m_mf(mf) { }
template<typename... Args>
inline message_future::continue_helper operator()(Args&&... args) {
return m_mf.then(std::forward<Args>(args)...);
}
private:
message_future m_mf;
};
class sync_receive_helper {
public:
inline sync_receive_helper(const message_future& mf) : m_mf(mf) { }
template<typename... Args>
inline void operator()(Args&&... args) {
m_mf.await(std::forward<Args>(args)...);
}
private:
message_future m_mf;
};
/**
* @brief Receives a synchronous response message.
* @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
* @relates message_future
*/
inline sync_handle_helper handle_response(const message_future& f) {
return {f};
}
/**
* @brief Handles a synchronous response message in an event-based way.
* @param handle A future for a synchronous response.
* @throws std::logic_error if @p handle is not valid or if the actor
* already received the response for @p handle
* @relates message_future
*/
inline sync_receive_helper receive_response(const message_future& f) {
return {f};
}
} // namespace cppa } // namespace cppa
#endif // MESSAGE_FUTURE_HPP #endif // MESSAGE_FUTURE_HPP
...@@ -36,6 +36,8 @@ ...@@ -36,6 +36,8 @@
#include "cppa/util/rm_ref.hpp" #include "cppa/util/rm_ref.hpp"
#include "cppa/util/type_list.hpp" #include "cppa/util/type_list.hpp"
#include "cppa/util/conjunction.hpp"
namespace cppa { namespace util { namespace cppa { namespace util {
template<typename Signature> template<typename Signature>
...@@ -123,6 +125,11 @@ struct is_callable { ...@@ -123,6 +125,11 @@ struct is_callable {
}; };
template<typename... Ts>
struct all_callable {
static constexpr bool value = conjunction<is_callable<Ts>...>::value;
};
template<bool IsCallable, typename C> template<bool IsCallable, typename C>
struct get_result_type_impl { struct get_result_type_impl {
typedef typename get_callable_trait<C>::type trait_type; typedef typename get_callable_trait<C>::type trait_type;
......
...@@ -142,15 +142,6 @@ void local_actor::forward_message(const actor_ptr& new_receiver) { ...@@ -142,15 +142,6 @@ void local_actor::forward_message(const actor_ptr& new_receiver) {
} }
} }
sync_recv_helper local_actor::handle_response(const message_future& handle) {
return {handle.id(), [](behavior& bhvr, message_id_t mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
self->become_waiting_for(std::move(bhvr), mf);
}};
}
response_handle local_actor::make_response_handle() { response_handle local_actor::make_response_handle() {
auto n = m_current_node; auto n = m_current_node;
response_handle result(this, n->sender, n->mid.response_id()); response_handle result(this, n->sender, n->mid.response_id());
...@@ -171,4 +162,8 @@ void local_actor::exec_behavior_stack() { ...@@ -171,4 +162,8 @@ void local_actor::exec_behavior_stack() {
// default implementation does nothing // default implementation does nothing
} }
sync_handle_helper local_actor::handle_response(const message_future& mf) {
return ::cppa::handle_response(mf);
}
} // namespace cppa } // namespace cppa
...@@ -57,13 +57,4 @@ void receive_loop(partial_function&& rules) { ...@@ -57,13 +57,4 @@ void receive_loop(partial_function&& rules) {
receive_loop(tmp); receive_loop(tmp);
} }
sync_recv_helper receive_response(const message_future& handle) {
return {handle.id(), [](behavior& bhvr, message_id_t mf) {
if (!self->awaits(mf)) {
throw std::logic_error("response already received");
}
self->dequeue_response(bhvr, mf);
}};
}
} // namespace cppa } // namespace cppa
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