Commit 09a6675c authored by Dominik Charousset's avatar Dominik Charousset

allow functors in `message_future.{await|then}`

this patch allows functors as response message handlers and adds a
replacable callback function to `local_actor` that is invoked whenever
the synchronous response does not match the handler's signature
parent eb29f99a
......@@ -606,7 +606,7 @@ inline void delayed_send_tuple(const channel_ptr& whom,
*/
template<class Rep, class Period, typename... Args>
inline void delayed_send(const channel_ptr& whom,
const std::chrono::duration<Rep, Period>& rtime,
const std::chrono::duration<Rep,Period>& rtime,
Args&&... what) {
static_assert(sizeof...(Args) > 0, "no message to send");
if (whom) {
......
......@@ -379,7 +379,7 @@ class receive_policy {
if (id.valid() && !id.is_answered() && sender) {
sender->sync_enqueue(client,
id.response_id(),
any_tuple());
make_any_tuple(atom("VOID")));
}
hm_cleanup(client, previous_node, policy);
return hm_msg_handled;
......
......@@ -57,6 +57,13 @@ static constexpr std::uint32_t unhandled_exception = 0x00002;
*/
static constexpr std::uint32_t unallowed_function_call = 0x00003;
/**
* @brief Indicates that a synchronous message timed out
* or the receiving actor didn't handle the message properly
* and the killed actor did not handle sync failures manually.
*/
static constexpr std::uint32_t unhandled_sync_failure = 0x00004;
/**
* @brief Indicates that an actor finishied execution
* because a connection to a remote link was
......
......@@ -32,6 +32,7 @@
#define CPPA_CONTEXT_HPP
#include <cstdint>
#include <functional>
#include "cppa/group.hpp"
#include "cppa/actor.hpp"
......@@ -412,6 +413,11 @@ class local_actor : public memory_cached_mixin<actor> {
virtual void dequeue_response(behavior&, message_id_t) = 0;
inline void dequeue_response(behavior&& bhvr, message_id_t mid) {
behavior tmp{std::move(bhvr)};
dequeue_response(tmp, mid);
}
virtual void become_waiting_for(behavior&&, message_id_t) = 0;
/**
......@@ -420,6 +426,15 @@ class local_actor : public memory_cached_mixin<actor> {
*/
response_handle make_response_handle();
inline void on_sync_failure(std::function<void()> fun) {
m_sync_failure_handler = std::move(fun);
}
inline void handle_sync_failure() {
if (m_sync_failure_handler) m_sync_failure_handler();
else quit(exit_reason::unhandled_sync_failure);
}
protected:
// true if this actor uses the chained_send optimization
......@@ -442,6 +457,10 @@ class local_actor : public memory_cached_mixin<actor> {
// {group => subscription} map of all joined groups
std::map<group_ptr, group::subscription> m_subscriptions;
private:
std::function<void()> m_sync_failure_handler;
# endif // CPPA_DOCUMENTATION
protected:
......
......@@ -31,6 +31,9 @@
#ifndef MESSAGE_FUTURE_HPP
#define MESSAGE_FUTURE_HPP
#include <type_traits>
#include "cppa/on.hpp"
#include "cppa/behavior.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/message_id.hpp"
......@@ -50,23 +53,41 @@ class message_future {
/**
* @brief Sets @p mexpr as event-handler for the response message.
*/
template<typename... Expression>
void then(Expression&&... mexpr) {
auto f = [](behavior& bhvr, message_id_t mid) {
self->become_waiting_for(std::move(bhvr), mid);
};
apply(f, std::forward<Expression>(mexpr)...);
template<typename... Cases, typename... Args>
void then(const match_expr<Cases...>& arg0, const Args&... args) {
consistency_check();
self->become_waiting_for(match_expr_convert(arg0, args...), m_id);
}
/**
* @brief Blocks until the response arrives and then executes @p mexpr.
*/
template<typename... Expression>
void await(Expression&&... mexpr) {
auto f = [](behavior& bhvr, message_id_t mid) {
self->dequeue_response(bhvr, mid);
};
apply(f, std::forward<Expression>(mexpr)...);
template<typename... Cases, typename... Args>
void await(const match_expr<Cases...>& arg0, const Args&... args) {
consistency_check();
self->dequeue_response(match_expr_convert(arg0, args...), m_id);
}
/**
* @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', 'TIMEOUT', or 'VOID' message.
*/
template<typename F>
typename std::enable_if<util::is_callable<F>::value>::type then(F fun) {
consistency_check();
self->become_waiting_for(bhvr_from_fun(fun), m_id);
}
/**
* @brief Blocks until the response arrives and then executes @p @p fun,
* calls <tt>self->handle_sync_failure()</tt> if the response
* message is an 'EXITED', 'TIMEOUT', or 'VOID' message.
*/
template<typename F>
typename std::enable_if<util::is_callable<F>::value>::type await(F fun) {
consistency_check();
self->dequeue_response(bhvr_from_fun(fun), m_id);
}
/**
......@@ -87,6 +108,28 @@ class message_future {
message_id_t m_id;
template<typename F>
behavior bhvr_from_fun(F fun) {
auto handle_sync_failure = [] { self->handle_sync_failure(); };
return {
on(atom("EXITED"), any_vals) >> handle_sync_failure,
on(atom("TIMEOUT")) >> handle_sync_failure,
on(atom("VOID")) >> handle_sync_failure,
on(any_vals, arg_match) >> fun,
others() >> handle_sync_failure
};
}
void consistency_check() {
if (!m_id.valid() || !m_id.is_response()) {
throw std::logic_error("handle does not point to a response");
}
else if (!self->awaits(m_id)) {
throw std::logic_error("response already received");
}
}
/*
template<typename Fun, typename... Args>
void apply(Fun& fun, Args&&... args) {
auto bhvr = match_expr_convert(std::forward<Args>(args)...);
......@@ -103,6 +146,7 @@ class message_future {
}
fun(bhvr, m_id);
}
*/
};
......
......@@ -2,18 +2,18 @@
#include "cppa/cppa.hpp"
using namespace cppa;
using namespace cppa::placeholders;
struct popular_actor : event_based_actor { // popular actors have a buddy
actor_ptr m_buddy;
popular_actor(const actor_ptr& buddy) : m_buddy(buddy) { }
inline const actor_ptr& buddy() const { return m_buddy; }
void report_failure() {
send(buddy(), atom("failure"));
self->quit();
}
};
void report_failure() {
local_actor* s = self;
send(static_cast<popular_actor*>(s)->buddy(), atom("failure"));
self->quit();
}
/******************************************************************************\
* test case 1: *
......@@ -34,16 +34,12 @@ struct A : popular_actor {
void init() {
become (
on(atom("go"), arg_match) >> [=](const actor_ptr& next) {
sync_send(next, atom("gogo")).then (
on(atom("gogogo")) >> [=] {
send(buddy(), atom("success"));
quit();
},
others() >> report_failure,
after(std::chrono::seconds(1)) >> report_failure
);
sync_send(next, atom("gogo")).then([=] {
send(buddy(), atom("success"));
quit();
});
},
others() >> report_failure
others() >> [=] { report_failure(); }
);
}
};
......@@ -94,13 +90,10 @@ struct D : popular_actor {
become (
others() >> [=] {
m_handle = make_response_handle();
sync_send_tuple(buddy(), last_dequeued()).then(
others() >> [=] {
m_handle.apply(last_dequeued());
quit();
},
after(std::chrono::seconds(1)) >> report_failure
);
sync_send_tuple(buddy(), last_dequeued()).then([=] {
m_handle.apply(last_dequeued());
quit();
});
}
);
}
......@@ -108,21 +101,23 @@ struct D : popular_actor {
int main() {
CPPA_TEST(test__sync_send);
send(spawn<A>(self), atom("go"), spawn<B>(spawn<C>()));
receive (
on(atom("success")) >> [&] { },
on(atom("failure")) >> [&] {
CPPA_ERROR("A didn't receive a sync response");
}
);
auto await_success_message = [&] {
receive (
on(atom("success")) >> [&] { },
on(atom("failure")) >> [&] {
CPPA_ERROR("A didn't receive a sync response");
},
on(atom("DOWN"), arg_match).when(_x2 != exit_reason::normal)
>> [&](uint32_t err) {
CPPA_ERROR("A exited for reason " << err);
}
);
};
send(spawn_monitor<A>(self), atom("go"), spawn<B>(spawn<C>()));
await_success_message();
await_all_others_done();
send(spawn<A>(self), atom("go"), spawn<D>(spawn<C>()));
receive (
on(atom("success")) >> [&] { },
on(atom("failure")) >> [&] {
CPPA_ERROR("A didn't receive a sync response");
}
);
send(spawn_monitor<A>(self), atom("go"), spawn<D>(spawn<C>()));
await_success_message();
await_all_others_done();
shutdown();
return CPPA_TEST_RESULT;
......
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