Commit e3d69731 authored by Dominik Charousset's avatar Dominik Charousset

Fix handling of expected<T> in response promises

parent 6b2cbd4f
......@@ -308,6 +308,7 @@ caf_add_component(
policy.select_all
policy.select_any
request_timeout
response_promise
result
save_inspector
selective_streaming
......
......@@ -53,27 +53,31 @@ public:
"mixing expected<T> with regular values is not supported");
if constexpr (sizeof...(Ts) == 0
&& std::is_same<message, std::decay_t<T>>::value)
return deliver_impl(std::forward<T>(x));
deliver_impl(std::forward<T>(x));
else
return deliver_impl(
make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
deliver_impl(make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
}
/// Satisfies the promise by sending either an error or a non-error response
/// message.
template <class T>
void deliver(expected<T> x) {
if (x)
return deliver(std::move(*x));
return deliver(std::move(x.error()));
if (x) {
if constexpr (std::is_same<T, void>::value)
deliver_impl(make_message());
else
deliver(std::move(*x));
} else {
deliver(std::move(x.error()));
}
}
/// 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
delegated_response_type_t<
typename Handle::signatures,
detail::implicit_conversions_t<typename std::decay<Ts>::type>...>
delegate(const Handle& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "nothing to delegate");
using token = detail::type_list<typename detail::implicit_conversions<
......@@ -96,7 +100,13 @@ public:
/// Satisfies the promise by sending an empty message if this promise has a
/// valid message ID, i.e., `async() == false`.
void deliver(unit_t x);
void deliver();
/// Satisfies the promise by sending an empty message if this promise has a
/// valid message ID, i.e., `async() == false`.
void deliver(unit_t) {
deliver();
}
/// Returns whether this response promise replies to an asynchronous message.
bool async() const;
......
......@@ -58,11 +58,18 @@ struct response_type<detail::type_list<Out(In...), Fs...>, In...> {
using delegated_type = delegated<Out>;
};
/// Computes the response message for input `In...` from the list of message
/// passing interfaces `Fs`.
/// Computes the response message type for input `In...` from the list of
/// message passing interfaces `Fs`.
template <class Fs, class... In>
using response_type_t = typename response_type<Fs, In...>::type;
/// Computes the response message type for input `In...` from the list of
/// message passing interfaces `Fs` and returns the corresponding
/// `delegated<T>`.
template <class Fs, class... In>
using delegated_response_type_t =
typename response_type<Fs, In...>::delegated_type;
/// Unboxes `Xs` and calls `response_type`.
template <class Ts, class Xs>
struct response_type_unbox;
......
......@@ -4,9 +4,11 @@
#pragma once
#include "caf/response_promise.hpp"
#include <type_traits>
#include "caf/detail/type_list.hpp"
#include "caf/make_message.hpp"
#include "caf/response_promise.hpp"
namespace caf {
......@@ -48,42 +50,38 @@ public:
}
/// Satisfies the promise by sending a non-error response message.
template <class U, class... Us>
typename std::enable_if<(sizeof...(Us) > 0)
|| !std::is_convertible<U, error>::value,
typed_response_promise>::type
deliver(U&& x, Us&&... xs) {
static_assert(
std::is_same<detail::type_list<Ts...>,
detail::type_list<typename std::decay<U>::type,
typename std::decay<Us>::type...>>::value,
"typed_response_promise: message type mismatched");
promise_.deliver(std::forward<U>(x), std::forward<Us>(xs)...);
return *this;
template <class... Us>
std::enable_if_t<(std::is_constructible<Ts, Us>::value && ...)>
deliver(Us... xs) {
promise_.deliver(make_message(Ts{std::forward<Us>(xs)}...));
}
/// Satisfies the promise by sending an empty response message.
template <class L = detail::type_list<Ts...>>
std::enable_if_t<std::is_same<L, detail::type_list<void>>::value> deliver() {
promise_.deliver();
}
/// Satisfies the promise by sending an error response message.
/// For non-requests, nothing is done.
void deliver(error x) {
promise_.deliver(std::move(x));
}
/// Satisfies the promise by sending either an error or a non-error response
/// message.
template <class T>
void deliver(expected<T> x) {
if (x)
return deliver(std::move(*x));
return deliver(std::move(x.error()));
std::enable_if_t<
std::is_same<detail::type_list<T>, detail::type_list<Ts...>>::value>
deliver(expected<T> x) {
promise_.deliver(std::move(x));
}
/// Satisfies the promise by delegating to another actor.
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.
/// For non-requests, nothing is done.
typed_response_promise deliver(error x) {
promise_.deliver(std::move(x));
return *this;
auto delegate(const Handle& dest, Us&&... xs) {
return promise_.template delegate<P>(dest, std::forward<Us>(xs)...);
}
/// Returns whether this response promise replies to an asynchronous message.
......
......@@ -46,7 +46,7 @@ void response_promise::deliver(error x) {
deliver_impl(make_message(std::move(x)));
}
void response_promise::deliver(unit_t) {
void response_promise::deliver() {
deliver_impl(make_message());
}
......@@ -75,6 +75,11 @@ void response_promise::deliver_impl(message msg) {
CAF_LOG_DEBUG("drop response: invalid promise");
return;
}
if (msg.empty() && id_.is_async()) {
CAF_LOG_DEBUG("drop response: empty response to asynchronous input");
self_.reset();
return;
}
auto dptr = self_dptr();
if (!stages_.empty()) {
auto next = std::move(stages_.back());
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2021 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. *
******************************************************************************/
#define CAF_SUITE response_promise
#include "caf/response_promise.hpp"
#include "core-test.hpp"
using namespace caf;
namespace {
behavior adder() {
return {
[](int x, int y) { return x + y; },
[](ok_atom) {},
};
}
behavior delegator(event_based_actor* self, actor worker) {
return {
[=](int x, int y) {
auto promise = self->make_response_promise();
return promise.delegate(worker, x, y);
},
[=](ok_atom) {
auto promise = self->make_response_promise();
return promise.delegate(worker, ok_atom_v);
},
};
}
behavior requester_v1(event_based_actor* self, actor worker) {
return {
[=](int x, int y) {
auto rp = self->make_response_promise();
self->request(worker, infinite, x, y)
.then(
[rp](int result) mutable {
CHECK(rp.pending());
rp.deliver(result);
},
[rp](error err) mutable {
CHECK(rp.pending());
rp.deliver(std::move(err));
});
return rp;
},
[=](ok_atom) {
auto rp = self->make_response_promise();
self->request(worker, infinite, ok_atom_v)
.then(
[rp]() mutable {
CHECK(rp.pending());
rp.deliver();
},
[rp](error err) mutable {
CHECK(rp.pending());
rp.deliver(std::move(err));
});
return rp;
},
};
}
behavior requester_v2(event_based_actor* self, actor worker) {
return {
[=](int x, int y) {
auto rp = self->make_response_promise();
auto deliver = [rp](expected<int> x) mutable {
CHECK(rp.pending());
rp.deliver(std::move(x));
};
self->request(worker, infinite, x, y)
.then([deliver](int result) mutable { deliver(result); },
[deliver](error err) mutable { deliver(std::move(err)); });
return rp;
},
[=](ok_atom) {
auto rp = self->make_response_promise();
auto deliver = [rp](expected<void> x) mutable {
CHECK(rp.pending());
rp.deliver(std::move(x));
};
self->request(worker, infinite, ok_atom_v)
.then([deliver]() mutable { deliver({}); },
[deliver](error err) mutable { deliver(std::move(err)); });
return rp;
},
};
}
} // namespace
CAF_TEST_FIXTURE_SCOPE(response_promise_tests, test_coordinator_fixture<>)
SCENARIO("response promises allow delaying of response messages") {
auto adder_hdl = sys.spawn(adder);
std::map<std::string, actor> impls;
impls["with a value or an error"] = sys.spawn(requester_v1, adder_hdl);
impls["with an expected<T>"] = sys.spawn(requester_v2, adder_hdl);
for (auto& [desc, hdl] : impls) {
GIVEN("a dispatcher that calls deliver " << desc << " on its promise") {
WHEN("sending a request with two integers to the dispatcher") {
inject((int, int), from(self).to(hdl).with(3, 4));
THEN("clients receive the response from the dispatcher") {
expect((int, int), from(hdl).to(adder_hdl).with(3, 4));
expect((int), from(adder_hdl).to(hdl).with(7));
expect((int), from(hdl).to(self).with(7));
}
}
WHEN("sending ok_atom to the dispatcher synchronously") {
auto res = self->request(hdl, infinite, ok_atom_v);
auto fetch_result = [&] {
message result;
res.receive([] {}, // void result
[&](const error& reason) {
result = make_message(reason);
});
return result;
};
THEN("clients receive an empty response from the dispatcher") {
expect((ok_atom), from(self).to(hdl));
expect((ok_atom), from(hdl).to(adder_hdl));
expect((void), from(adder_hdl).to(hdl));
CHECK(fetch_result().empty());
}
}
WHEN("sending ok_atom to the dispatcher asynchronously") {
THEN("clients receive no response from the dispatcher") {
inject((ok_atom), from(self).to(hdl).with(ok_atom_v));
expect((ok_atom), from(hdl).to(adder_hdl));
expect((void), from(adder_hdl).to(hdl));
CHECK(self->mailbox().empty());
}
}
}
}
}
SCENARIO("response promises allow delegation") {
GIVEN("a dispatcher that calls delegate on its promise") {
auto adder_hdl = sys.spawn(adder);
auto hdl = sys.spawn(delegator, adder_hdl);
WHEN("sending a request to the dispatcher") {
inject((int, int), from(self).to(hdl).with(3, 4));
THEN("clients receive the response from the adder") {
expect((int, int), from(self).to(adder_hdl).with(3, 4));
expect((int), from(adder_hdl).to(self).with(7));
}
}
}
}
CAF_TEST_FIXTURE_SCOPE_END()
This diff is collapsed.
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