Commit 70b33e8b authored by Dominik Charousset's avatar Dominik Charousset

Port response_handle to a policy-based design

parent ff24212e
......@@ -57,15 +57,6 @@
#include "caf/mixin/sender.hpp"
#include "caf/mixin/subscriber.hpp"
namespace caf {
namespace mixin {
template <>
struct is_blocking_requester<blocking_actor> : std::true_type { };
} // namespace caf
} // namespace mixin
namespace caf {
/// A thread-mapped or context-switching actor using a blocking
......
......@@ -18,8 +18,8 @@
#pragma once
#include <tuple>
#include <chrono>
#include <tuple>
#include "caf/actor.hpp"
#include "caf/check_typed_input.hpp"
......@@ -29,15 +29,13 @@
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/message_priority.hpp"
#include "caf/policy/single_response.hpp"
#include "caf/response_handle.hpp"
#include "caf/response_type.hpp"
namespace caf {
namespace mixin {
template <class T>
struct is_blocking_requester : std::false_type { };
/// A `requester` is an actor that supports
/// `self->request(...).{then|await|receive}`.
template <class Base, class Subtype>
......@@ -60,22 +58,12 @@ public:
/// @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 Handle = actor, class... Ts>
response_handle<Subtype,
response_type_t<
typename Handle::signatures,
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>,
is_blocking_requester<Subtype>::value>
request(const Handle& dest, const duration& timeout, Ts&&... xs) {
template <message_priority P = message_priority::normal, class Handle = actor,
class... Ts>
auto request(const Handle& dest, const duration& timeout, Ts&&... xs) {
using namespace detail;
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
using token = type_list<implicit_conversions_t<decay_t<Ts>>...>;
static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid,
"receiver does not accept given message");
auto self = static_cast<Subtype*>(this);
......@@ -88,29 +76,25 @@ public:
self->eq_impl(req_id.response_id(), self->ctrl(), self->context(),
make_error(sec::invalid_argument));
}
return {req_id.response_id(), self};
}
using response_type
= response_type_t<typename Handle::signatures,
detail::implicit_conversions_t<detail::decay_t<Ts>>...>;
using handle_type
= response_handle<Subtype, policy::single_response<response_type>>;
return handle_type{self, req_id.response_id()};
}
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
/// @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, class Period = std::ratio<1>,
class Handle = actor, class... Ts>
response_handle<Subtype,
response_type_t<
typename Handle::signatures,
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>,
is_blocking_requester<Subtype>::value>
request(const Handle& dest, std::chrono::duration<Rep, Period> timeout,
Ts&&... xs) {
template <message_priority P = message_priority::normal, class Rep = int,
class Period = std::ratio<1>, class Handle = actor, class... Ts>
auto request(const Handle& dest, std::chrono::duration<Rep, Period> timeout,
Ts&&... xs) {
return request(dest, duration{timeout}, std::forward<Ts>(xs)...);
}
};
} // namespace mixin
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/behavior.hpp"
#include "caf/config.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
#include "caf/error.hpp"
#include "caf/message_id.hpp"
namespace caf {
namespace policy {
/// Trivial policy for handling a single result in a `response_handler`.
/// @relates mixin::requester
/// @relates response_handle
template <class ResponseType>
class single_response {
public:
static constexpr bool is_trivial = true;
using response_type = ResponseType;
explicit single_response(message_id mid) noexcept : mid_(mid) {
// nop
}
single_response(single_response&&) noexcept = default;
single_response& operator=(single_response&&) noexcept = default;
template <class Self, class F, class OnError>
void await(Self* self, F&& f, OnError&& g) const {
behavior bhvr{std::forward<F>(f), std::forward<OnError>(g)};
self->add_awaited_response_handler(mid_, std::move(bhvr));
}
template <class Self, class F, class OnError>
void then(Self* self, F&& f, OnError&& g) const {
behavior bhvr{std::forward<F>(f), std::forward<OnError>(g)};
self->add_multiplexed_response_handler(mid_, std::move(bhvr));
}
template <class Self, class F, class OnError>
void receive(Self* self, F&& f, OnError&& g) const {
typename Self::accept_one_cond rc;
self->varargs_receive(rc, mid_, std::forward<F>(f),
std::forward<OnError>(g));
}
message_id id() const noexcept {
return mid_;
}
private:
message_id mid_;
};
} // namespace policy
} // namespace caf
......@@ -20,8 +20,10 @@
#include <type_traits>
#include "caf/actor_traits.hpp"
#include "caf/catch_all.hpp"
#include "caf/message_id.hpp"
#include "caf/none.hpp"
#include "caf/sec.hpp"
#include "caf/system_messages.hpp"
#include "caf/typed_behavior.hpp"
......@@ -33,145 +35,140 @@ namespace caf {
/// This helper class identifies an expected response message and enables
/// `request(...).then(...)`.
template <class Self, class Output, bool IsBlocking>
class response_handle;
/******************************************************************************
* nonblocking *
******************************************************************************/
template <class Self, class Output>
class response_handle<Self, Output, false> {
template <class ActorType, class Policy>
class response_handle {
public:
response_handle() = delete;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
// -- member types -----------------------------------------------------------
response_handle(message_id mid, Self* self) : mid_(mid), self_(self) {
// nop
}
using actor_type = ActorType;
template <class F, class E = detail::is_callable_t<F>>
void await(F f) const {
await_impl(f);
}
using traits = actor_traits<actor_type>;
template <class F, class OnError, class E1 = detail::is_callable_t<F>,
class E2 = detail::is_handler_for_ef<OnError, error>>
void await(F f, OnError e) const {
await_impl(f, e);
}
using policy_type = Policy;
template <class F, class E = detail::is_callable_t<F>>
void then(F f) const {
then_impl(f);
}
using response_type = typename policy_type::response_type;
template <class F, class OnError, class E1 = detail::is_callable_t<F>,
class E2 = detail::is_handler_for_ef<OnError, error>>
void then(F f, OnError e) const {
then_impl(f, e);
}
// -- constructors, destructors, and assignment operators --------------------
message_id id() const noexcept {
return mid_;
}
response_handle() = delete;
Self* self() noexcept {
return self_;
response_handle(const response_handle&) = delete;
response_handle& operator=(const response_handle&) = delete;
response_handle(response_handle&&) noexcept = default;
response_handle& operator=(response_handle&&) noexcept = delete;
template <class... Ts>
explicit response_handle(actor_type* self, Ts&&... xs)
: self_(self), policy_(std::forward<Ts>(xs)...) {
// nop
}
private:
template <class F, class OnError>
void await_impl(F& f, OnError& g) const {
// -- non-blocking API -------------------------------------------------------
template <class T = traits, class F = none_t, class OnError = none_t,
class = detail::enable_if_t<T::is_non_blocking>>
void await(F f, OnError g) const {
static_assert(detail::is_callable<F>::value, "F must provide a single, "
"non-template operator()");
static_assert(detail::is_callable<F>::value,
"OnError must provide a single, non-template operator() "
"that takes a caf::error");
using result_type = typename detail::get_callable_trait<F>::result_type;
static_assert(std::is_same<void, result_type>::value,
"response handlers are not allowed to have a return "
"type other than void");
detail::type_checker<Output, F>::check();
self_->add_awaited_response_handler(mid_,
behavior{std::move(f), std::move(g)});
detail::type_checker<response_type, F>::check();
policy_.await(self_, std::move(f), std::move(g));
}
template <class F>
void await_impl(F& f) const {
template <class T = traits, class F = none_t,
class = detail::enable_if_t<T::is_non_blocking>>
void await(F f) const {
auto self = self_;
auto on_error = [self](error& err) { self->call_error_handler(err); };
return await_impl(f, on_error);
await(std::move(f), [self](error& err) { self->call_error_handler(err); });
}
template <class F, class OnError>
void then_impl(F& f, OnError& g) const {
template <class T = traits, class F = none_t, class OnError = none_t,
class = detail::enable_if_t<T::is_non_blocking>>
void then(F f, OnError g) const {
static_assert(detail::is_callable<F>::value, "F must provide a single, "
"non-template operator()");
static_assert(detail::is_callable<F>::value,
"OnError must provide a single, non-template operator() "
"that takes a caf::error");
using result_type = typename detail::get_callable_trait<F>::result_type;
static_assert(std::is_same<void, result_type>::value,
"response handlers are not allowed to have a return "
"type other than void");
detail::type_checker<Output, F>::check();
self_->add_multiplexed_response_handler(mid_, behavior{std::move(f),
std::move(g)});
detail::type_checker<response_type, F>::check();
policy_.then(self_, std::move(f), std::move(g));
}
template <class F>
void then_impl(F& f) const {
template <class T = traits, class F = none_t,
class = detail::enable_if_t<T::is_non_blocking>>
void then(F f) const {
auto self = self_;
auto on_error = [self](error& err) { self->call_error_handler(err); };
return then_impl(f, on_error);
}
message_id mid_;
Self* self_;
};
/******************************************************************************
* blocking *
******************************************************************************/
template <class Self, class Output>
class response_handle<Self, Output, true> {
public:
response_handle() = delete;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
response_handle(message_id mid, Self* self) : mid_(mid), self_(self) {
// nop
then(std::move(f), [self](error& err) { self->call_error_handler(err); });
}
using error_handler = std::function<void(error&)>;
// -- blocking API -----------------------------------------------------------
template <class F, class OnError,
class E = detail::is_handler_for_ef<OnError, error>>
detail::is_callable_t<F> receive(F f, OnError g) {
template <class T = traits, class F = none_t, class OnError = none_t,
class = detail::enable_if_t<T::is_blocking>>
detail::is_handler_for_ef<OnError, error> receive(F f, OnError g) {
static_assert(detail::is_callable<F>::value, "F must provide a single, "
"non-template operator()");
static_assert(detail::is_callable<F>::value,
"OnError must provide a single, non-template operator() "
"that takes a caf::error");
using result_type = typename detail::get_callable_trait<F>::result_type;
static_assert(std::is_same<void, result_type>::value,
"response handlers are not allowed to have a return "
"type other than void");
detail::type_checker<Output, F>::check();
typename Self::accept_one_cond rc;
self_->varargs_receive(rc, mid_, std::move(f), std::move(g));
detail::type_checker<response_type, F>::check();
policy_.receive(self_, std::move(f), std::move(g));
}
template <class OnError, class F, class E = detail::is_callable_t<F>>
template <class T = traits, class OnError = none_t, class F = none_t,
class = detail::enable_if_t<T::is_blocking>>
detail::is_handler_for_ef<OnError, error> receive(OnError g, F f) {
// TODO: allowing blocking actors to pass the error handler in first may be
// more flexible, but it makes the API asymmetric. Consider
// deprecating / removing this member function.
receive(std::move(f), std::move(g));
}
template <class OnError, class F,
class E = detail::is_handler_for_ef<OnError, error>>
template <class T = policy_type, class OnError = none_t, class F = none_t,
class E = detail::is_handler_for_ef<OnError, error>,
class = detail::enable_if_t<T::is_trivial>>
void receive(OnError g, catch_all<F> f) {
typename Self::accept_one_cond rc;
self_->varargs_receive(rc, mid_, std::move(g), std::move(f));
// TODO: this bypasses the policy. Either we deprecate `catch_all` or *all*
// policies must support it. Currently, we only enable this member
// function on the trivial policy for backwards compatibility.
typename actor_type::accept_one_cond rc;
self_->varargs_receive(rc, id(), std::move(g), std::move(f));
}
// -- properties -------------------------------------------------------------
template <class T = policy_type, class = detail::enable_if_t<T::is_trivial>>
message_id id() const noexcept {
return mid_;
return policy_.id();
}
Self* self() noexcept {
actor_type* self() noexcept {
return self_;
}
private:
message_id mid_;
Self* self_;
/// Points to the parent actor.
actor_type* self_;
/// Configures how the actor wants to process an incoming response.
policy_type policy_;
};
} // namespace caf
......@@ -645,9 +645,9 @@ protected:
template <class... Ts>
struct test_coordinator_fixture_fetch_helper {
template <class Self, class Interface>
template <class Self, template <class> class Policy, class Interface>
std::tuple<Ts...>
operator()(caf::response_handle<Self, Interface, true>& from) const {
operator()(caf::response_handle<Self, Policy<Interface>>& from) const {
std::tuple<Ts...> result;
from.receive([&](Ts&... xs) { result = std::make_tuple(std::move(xs)...); },
[&](caf::error& err) {
......@@ -659,8 +659,8 @@ struct test_coordinator_fixture_fetch_helper {
template <class T>
struct test_coordinator_fixture_fetch_helper<T> {
template <class Self, class Interface>
T operator()(caf::response_handle<Self, Interface, true>& from) const {
template <class Self, template <class> class Policy, class Interface>
T operator()(caf::response_handle<Self, Policy<Interface>>& from) const {
T result;
from.receive([&](T& x) { result = std::move(x); },
[&](caf::error& err) {
......
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