Commit e186ad18 authored by Dominik Charousset's avatar Dominik Charousset

Add fan_out_request function to the requester

parent 70b33e8b
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <chrono> #include <chrono>
#include <tuple> #include <tuple>
#include <vector>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/check_typed_input.hpp" #include "caf/check_typed_input.hpp"
...@@ -84,16 +85,90 @@ public: ...@@ -84,16 +85,90 @@ public:
return handle_type{self, req_id.response_id()}; return handle_type{self, req_id.response_id()};
} }
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`. /// @copydoc request
/// @returns A handle identifying a future-like handle to the response.
/// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor.
template <message_priority P = message_priority::normal, class Rep = int, template <message_priority P = message_priority::normal, class Rep = int,
class Period = std::ratio<1>, class Handle = actor, class... Ts> class Period = std::ratio<1>, class Handle = actor, class... Ts>
auto request(const Handle& dest, std::chrono::duration<Rep, Period> timeout, auto request(const Handle& dest, std::chrono::duration<Rep, Period> timeout,
Ts&&... xs) { Ts&&... xs) {
return request(dest, duration{timeout}, std::forward<Ts>(xs)...); return request(dest, duration{timeout}, std::forward<Ts>(xs)...);
} }
/// Sends `{xs...}` to each actor in the range `destinations` as a synchronous
/// message. Response messages get combined into a single result according to
/// the `MergePolicy`.
/// @tparam MergePolicy Configures how individual response messages get
/// combined by the actor. The policy makes sure that the
/// response handler gets invokes at most once. In case of
/// one or more errors, the policy calls the error handler
/// exactly once, with the first error occurred.
/// @tparam Prio Specifies the priority of the synchronous messages.
/// @tparam Container A container type for holding actor handles. Must provide
/// the type aliase `value_type` as well as the member
/// functions `begin()`, `end()`, and `size()`.
/// @param destinations A container holding handles to all destination actors.
/// @param timeout Maximum duration before dropping the request. The runtime
/// system will send an error message to the actor in case the
/// receives does not respond in time.
/// @returns A helper object that takes response handlers via `.await()`,
/// `.then()`, or `.receive()`.
/// @note The returned handle is actor-specific. Only the actor that called
/// `request` can use it for setting response handlers.
template <template <class> class MergePolicy,
message_priority Prio = message_priority::normal, class Container,
class... Ts>
// TODO: replace this monstrosity with 'auto' when switching to C++17
response_handle<Subtype,
MergePolicy<response_type_t<
typename Container::value_type::signatures,
detail::implicit_conversions_t<detail::decay_t<Ts>>...>>>
fan_out_request(const Container& destinations, const duration& timeout,
Ts&&... xs) {
using handle_type = typename Container::value_type;
using namespace detail;
static_assert(sizeof...(Ts) > 0, "no message to send");
using token = type_list<implicit_conversions_t<decay_t<Ts>>...>;
static_assert(
response_type_unbox<signatures_of_t<handle_type>, token>::valid,
"receiver does not accept given message");
auto dptr = static_cast<Subtype*>(this);
std::vector<message_id> ids;
ids.reserve(destinations.size());
for (const auto& dest : destinations) {
if (!dest)
continue;
auto req_id = dptr->new_request_id(Prio);
dest->eq_impl(req_id, dptr->ctrl(), dptr->context(),
std::forward<Ts>(xs)...);
dptr->request_response_timeout(timeout, req_id);
ids.emplace_back(req_id);
}
if (ids.empty()) {
auto req_id = dptr->new_request_id(Prio);
dptr->eq_impl(req_id.response_id(), dptr->ctrl(), dptr->context(),
make_error(sec::invalid_argument));
ids.emplace_back(req_id);
}
using response_type
= response_type_t<typename handle_type::signatures,
detail::implicit_conversions_t<detail::decay_t<Ts>>...>;
using handle_type = response_handle<Subtype, MergePolicy<response_type>>;
return handle_type{dptr, std::move(ids)};
}
/// @copydoc fan_out_request
template <template <class> class MergePolicy,
message_priority Prio = message_priority::normal, class Rep,
class Period, class Container, class... Ts>
// TODO: replace this monstrosity with 'auto' when switching to C++17
response_handle<Subtype,
MergePolicy<response_type_t<
typename Container::value_type::signatures,
detail::implicit_conversions_t<detail::decay_t<Ts>>...>>>
fan_out_request(const Container& destinations,
std::chrono::duration<Rep, Period> timeout, Ts&&... xs) {
return request<MergePolicy, Prio>(destinations, duration{timeout},
std::forward<Ts>(xs)...);
}
}; };
} // namespace mixin } // namespace mixin
......
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