Commit b9ee7c0a authored by Dominik Charousset's avatar Dominik Charousset

Add `response_promise::delegate`, close #512

parent 53c92ec6
...@@ -21,7 +21,7 @@ ...@@ -21,7 +21,7 @@
#define CAF_CHECK_TYPED_INPUT_HPP #define CAF_CHECK_TYPED_INPUT_HPP
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/typed_actor.hpp" #include "caf/replies_to.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/detail/typed_actor_util.hpp" #include "caf/detail/typed_actor_util.hpp"
......
...@@ -26,6 +26,8 @@ ...@@ -26,6 +26,8 @@
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/response_type.hpp"
#include "caf/check_typed_input.hpp"
namespace caf { namespace caf {
...@@ -54,7 +56,34 @@ public: ...@@ -54,7 +56,34 @@ public:
response_promise response_promise
>::type >::type
deliver(T&&x, Ts&&... xs) { deliver(T&&x, Ts&&... xs) {
return deliver_impl(make_message(std::forward<T>(x), std::forward<Ts>(xs)...)); return deliver_impl(make_message(std::forward<T>(x),
std::forward<Ts>(xs)...));
}
/// Satisfies the promise by delegating to another actor.
template <message_priority P = message_priority::normal,
class Handle = actor, class... Ts>
typename response_type<
typename Handle::signatures,
detail::implicit_conversions_t<typename std::decay<Ts>::type>...
>::delegated_type
delegate(const Handle& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "nothing to delegate");
using token =
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid,
"receiver does not accept given message");
if (dest) {
auto mid = P == message_priority::high ? id_.with_high_priority() : id_;
dest->enqueue(make_mailbox_element(std::move(source_), mid,
std::move(stages_),
std::forward<Ts>(xs)...),
ctx_);
}
return {};
} }
/// Satisfies the promise by sending an error response message. /// Satisfies the promise by sending an error response message.
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <tuple> #include <tuple>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/replies_to.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
......
...@@ -67,6 +67,13 @@ public: ...@@ -67,6 +67,13 @@ public:
return *this; return *this;
} }
template <message_priority P = message_priority::normal,
class Handle = actor, class... Us>
typed_response_promise delegate(const Handle& dest, Us&&... xs) {
promise_.template delegate<P>(dest, std::forward<Us>(xs)...);
return *this;
}
/// Satisfies the promise by sending an error response message. /// Satisfies the promise by sending an error response message.
/// For non-requests, nothing is done. /// For non-requests, nothing is done.
inline typed_response_promise deliver(error x) { inline typed_response_promise deliver(error x) {
......
...@@ -20,9 +20,11 @@ ...@@ -20,9 +20,11 @@
#include <utility> #include <utility>
#include <algorithm> #include <algorithm>
#include "caf/local_actor.hpp"
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/logger.hpp"
#include "caf/local_actor.hpp"
namespace caf { namespace caf {
response_promise::response_promise() response_promise::response_promise()
...@@ -68,9 +70,9 @@ response_promise response_promise::deliver_impl(message msg) { ...@@ -68,9 +70,9 @@ response_promise response_promise::deliver_impl(message msg) {
return *this; return *this;
} }
if (self_) if (self_)
CAF_LOG_ERROR("response promise already satisfied"); CAF_LOG_INFO("response promise already satisfied");
else else
CAF_LOG_ERROR("invalid response promise"); CAF_LOG_INFO("invalid response promise");
return *this; return *this;
} }
......
...@@ -148,54 +148,21 @@ CAF_TEST(typed_response_promise) { ...@@ -148,54 +148,21 @@ CAF_TEST(typed_response_promise) {
typed_response_promise<int> resp; typed_response_promise<int> resp;
CAF_MESSAGE("trigger 'invalid response promise' error"); CAF_MESSAGE("trigger 'invalid response promise' error");
resp.deliver(1); // delivers on an invalid promise has no effect resp.deliver(1); // delivers on an invalid promise has no effect
self->request(foo, infinite, get_atom::value, 42).receive( auto f = make_function_view(foo);
[](int x) { CAF_CHECK_EQUAL(f(get_atom::value, 42), 84);
CAF_CHECK_EQUAL(x, 84); CAF_CHECK_EQUAL(f(get_atom::value, 42, 52), std::make_tuple(84, 104));
}, CAF_CHECK_EQUAL(f(get_atom::value, 3.14, 3.14), std::make_tuple(6.28, 6.28));
ERROR_HANDLER
);
self->request(foo, infinite, get_atom::value, 42, 52).receive(
[](int x, int y) {
CAF_CHECK_EQUAL(x, 84);
CAF_CHECK_EQUAL(y, 104);
},
ERROR_HANDLER
);
self->request(foo, infinite, get_atom::value, 3.14, 3.14).receive(
[](double x, double y) {
CAF_CHECK_EQUAL(x, 3.14 * 2);
CAF_CHECK_EQUAL(y, 3.14 * 2);
},
[](const error& err) {
CAF_ERROR("unexpected error response message received: "
<< to_string(err));
}
);
} }
CAF_TEST(typed_response_promise_chained) { CAF_TEST(typed_response_promise_chained) {
auto composed = foo * foo * foo; auto f = make_function_view(foo * foo * foo);
self->request(composed, infinite, 1).receive( CAF_CHECK_EQUAL(f(1), 8);
[](int v) {
CAF_CHECK_EQUAL(v, 8);
},
[](const error& err) {
CAF_ERROR("unexpected error response message received: "
<< to_string(err));
}
);
} }
// verify that only requests get an error response message // verify that only requests get an error response message
CAF_TEST(error_response_message) { CAF_TEST(error_response_message) {
self->request(foo, infinite, get_atom::value, 3.14).receive( auto f = make_function_view(foo);
[](double) { CAF_CHECK_EQUAL(f(get_atom::value, 3.14), sec::unexpected_message);
CAF_ERROR("unexpected ordinary response message received");
},
[](error& err) {
CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::unexpected_message));
}
);
self->send(foo, get_atom::value, 42); self->send(foo, get_atom::value, 42);
self->receive( self->receive(
[](int x) { [](int x) {
...@@ -208,7 +175,7 @@ CAF_TEST(error_response_message) { ...@@ -208,7 +175,7 @@ CAF_TEST(error_response_message) {
self->send(foo, get_atom::value, 3.14); self->send(foo, get_atom::value, 3.14);
self->receive( self->receive(
[&](error& err) { [&](error& err) {
CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::unexpected_message)); CAF_CHECK_EQUAL(err, sec::unexpected_message);
self->send(self, message{}); self->send(self, message{});
} }
); );
...@@ -230,4 +197,33 @@ CAF_TEST(satisfied_promise) { ...@@ -230,4 +197,33 @@ CAF_TEST(satisfied_promise) {
); );
} }
CAF_TEST(delegating_promises) {
using task = std::pair<typed_response_promise<int>, int>;
struct state {
std::vector<task> tasks;
};
using bar_actor = typed_actor<replies_to<int>::with<int>, reacts_to<ok_atom>>;
auto bar_fun = [](bar_actor::stateful_pointer<state> self, foo_actor worker)
-> bar_actor::behavior_type {
return {
[=](int x) -> typed_response_promise<int> {
auto& tasks = self->state.tasks;
tasks.emplace_back(self->make_response_promise<int>(), x);
self->send(self, ok_atom::value);
return tasks.back().first;
},
[=](ok_atom) {
auto& tasks = self->state.tasks;
if (!tasks.empty()) {
auto& task = tasks.back();
task.first.delegate(worker, task.second);
tasks.pop_back();
}
}
};
};
auto f = make_function_view(system.spawn(bar_fun, foo));
CAF_CHECK_EQUAL(f(42), 84);
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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