Unverified Commit c089c764 authored by Noir's avatar Noir Committed by GitHub

Merge pull request #1198

Fix handling of expected<T> in response promises
parents 99972912 fb854f73
...@@ -22,6 +22,8 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -22,6 +22,8 @@ is based on [Keep a Changelog](https://keepachangelog.com).
trying to parse the input of `x` if it contains a string. The function trying to parse the input of `x` if it contains a string. The function
`get_or` already existed for `settings`, but we have added new overloads for `get_or` already existed for `settings`, but we have added new overloads for
generalizing the function to `config_value` as well. generalizing the function to `config_value` as well.
- The `typed_response_promise` received additional member functions to mirror
the interface of the untyped `response_promise`.
### Deprecated ### Deprecated
...@@ -84,6 +86,8 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -84,6 +86,8 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Skipping high-priority messages resulted in CAF lowering the priority to - Skipping high-priority messages resulted in CAF lowering the priority to
normal. This unintentional demotion has been fixed (#1171). normal. This unintentional demotion has been fixed (#1171).
- Fix undefined behavior in the experimental datagram brokers (#1174). - Fix undefined behavior in the experimental datagram brokers (#1174).
- Response promises no longer send empty messages in response to asynchronous
messages.
## [0.18.0-rc.1] - 2020-09-09 ## [0.18.0-rc.1] - 2020-09-09
......
...@@ -308,6 +308,7 @@ caf_add_component( ...@@ -308,6 +308,7 @@ caf_add_component(
policy.select_all policy.select_all
policy.select_any policy.select_any
request_timeout request_timeout
response_promise
result result
save_inspector save_inspector
selective_streaming selective_streaming
......
...@@ -53,27 +53,31 @@ public: ...@@ -53,27 +53,31 @@ public:
"mixing expected<T> with regular values is not supported"); "mixing expected<T> with regular values is not supported");
if constexpr (sizeof...(Ts) == 0 if constexpr (sizeof...(Ts) == 0
&& std::is_same<message, std::decay_t<T>>::value) && std::is_same<message, std::decay_t<T>>::value)
return deliver_impl(std::forward<T>(x)); deliver_impl(std::forward<T>(x));
else else
return deliver_impl( deliver_impl(make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
make_message(std::forward<T>(x), std::forward<Ts>(xs)...));
} }
/// Satisfies the promise by sending either an error or a non-error response /// Satisfies the promise by sending either an error or a non-error response
/// message. /// message.
template <class T> template <class T>
void deliver(expected<T> x) { void deliver(expected<T> x) {
if (x) if (x) {
return deliver(std::move(*x)); if constexpr (std::is_same<T, void>::value)
return deliver(std::move(x.error())); deliver();
else
deliver(std::move(*x));
} else {
deliver(std::move(x.error()));
}
} }
/// Satisfies the promise by delegating to another actor. /// Satisfies the promise by delegating to another actor.
template <message_priority P = message_priority::normal, class Handle = actor, template <message_priority P = message_priority::normal, class Handle = actor,
class... Ts> class... Ts>
typename response_type<typename Handle::signatures, delegated_response_type_t<
detail::implicit_conversions_t< typename Handle::signatures,
typename std::decay<Ts>::type>...>::delegated_type detail::implicit_conversions_t<typename std::decay<Ts>::type>...>
delegate(const Handle& dest, Ts&&... xs) { delegate(const Handle& dest, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "nothing to delegate"); static_assert(sizeof...(Ts) > 0, "nothing to delegate");
using token = detail::type_list<typename detail::implicit_conversions< using token = detail::type_list<typename detail::implicit_conversions<
...@@ -96,7 +100,13 @@ public: ...@@ -96,7 +100,13 @@ public:
/// Satisfies the promise by sending an empty message if this promise has a /// Satisfies the promise by sending an empty message if this promise has a
/// valid message ID, i.e., `async() == false`. /// 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. /// Returns whether this response promise replies to an asynchronous message.
bool async() const; bool async() const;
......
...@@ -58,11 +58,18 @@ struct response_type<detail::type_list<Out(In...), Fs...>, In...> { ...@@ -58,11 +58,18 @@ struct response_type<detail::type_list<Out(In...), Fs...>, In...> {
using delegated_type = delegated<Out>; using delegated_type = delegated<Out>;
}; };
/// Computes the response message for input `In...` from the list of message /// Computes the response message type for input `In...` from the list of
/// passing interfaces `Fs`. /// message passing interfaces `Fs`.
template <class Fs, class... In> template <class Fs, class... In>
using response_type_t = typename response_type<Fs, In...>::type; 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`. /// Unboxes `Xs` and calls `response_type`.
template <class Ts, class Xs> template <class Ts, class Xs>
struct response_type_unbox; struct response_type_unbox;
......
...@@ -4,9 +4,11 @@ ...@@ -4,9 +4,11 @@
#pragma once #pragma once
#include "caf/response_promise.hpp" #include <type_traits>
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/make_message.hpp"
#include "caf/response_promise.hpp"
namespace caf { namespace caf {
...@@ -48,42 +50,38 @@ public: ...@@ -48,42 +50,38 @@ public:
} }
/// Satisfies the promise by sending a non-error response message. /// Satisfies the promise by sending a non-error response message.
template <class U, class... Us> template <class... Us>
typename std::enable_if<(sizeof...(Us) > 0) std::enable_if_t<(std::is_constructible<Ts, Us>::value && ...)>
|| !std::is_convertible<U, error>::value, deliver(Us... xs) {
typed_response_promise>::type promise_.deliver(make_message(Ts{std::forward<Us>(xs)}...));
deliver(U&& x, Us&&... xs) { }
static_assert(
std::is_same<detail::type_list<Ts...>, /// Satisfies the promise by sending an empty response message.
detail::type_list<typename std::decay<U>::type, template <class L = detail::type_list<Ts...>>
typename std::decay<Us>::type...>>::value, std::enable_if_t<std::is_same<L, detail::type_list<void>>::value> deliver() {
"typed_response_promise: message type mismatched"); promise_.deliver();
promise_.deliver(std::forward<U>(x), std::forward<Us>(xs)...); }
return *this;
/// 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 /// Satisfies the promise by sending either an error or a non-error response
/// message. /// message.
template <class T> template <class T>
void deliver(expected<T> x) { std::enable_if_t<
if (x) std::is_same<detail::type_list<T>, detail::type_list<Ts...>>::value>
return deliver(std::move(*x)); deliver(expected<T> x) {
return deliver(std::move(x.error())); promise_.deliver(std::move(x));
} }
/// Satisfies the promise by delegating to another actor. /// Satisfies the promise by delegating to another actor.
template <message_priority P = message_priority::normal, class Handle = actor, template <message_priority P = message_priority::normal, class Handle = actor,
class... Us> class... Us>
typed_response_promise delegate(const Handle& dest, Us&&... xs) { auto delegate(const Handle& dest, Us&&... xs) {
promise_.template delegate<P>(dest, std::forward<Us>(xs)...); return 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;
} }
/// Returns whether this response promise replies to an asynchronous message. /// Returns whether this response promise replies to an asynchronous message.
......
...@@ -46,7 +46,7 @@ void response_promise::deliver(error x) { ...@@ -46,7 +46,7 @@ void response_promise::deliver(error x) {
deliver_impl(make_message(std::move(x))); deliver_impl(make_message(std::move(x)));
} }
void response_promise::deliver(unit_t) { void response_promise::deliver() {
deliver_impl(make_message()); deliver_impl(make_message());
} }
...@@ -75,6 +75,11 @@ void response_promise::deliver_impl(message msg) { ...@@ -75,6 +75,11 @@ void response_promise::deliver_impl(message msg) {
CAF_LOG_DEBUG("drop response: invalid promise"); CAF_LOG_DEBUG("drop response: invalid promise");
return; return;
} }
if (msg.empty() && id_.is_async()) {
CAF_LOG_DEBUG("drop response: empty response to asynchronous input");
self_.reset();
return;
}
auto dptr = self_dptr(); auto dptr = self_dptr();
if (!stages_.empty()) { if (!stages_.empty()) {
auto next = std::move(stages_.back()); auto next = std::move(stages_.back());
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#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