Commit 27874815 authored by Lingxi-Li's avatar Lingxi-Li

Improve response promise

parent 0d390a06
...@@ -77,7 +77,6 @@ ...@@ -77,7 +77,6 @@
#include "caf/abstract_channel.hpp" #include "caf/abstract_channel.hpp"
#include "caf/may_have_timeout.hpp" #include "caf/may_have_timeout.hpp"
#include "caf/message_priority.hpp" #include "caf/message_priority.hpp"
#include "caf/response_promise.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/primitive_variant.hpp" #include "caf/primitive_variant.hpp"
...@@ -86,6 +85,7 @@ ...@@ -86,6 +85,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/scoped_execution_unit.hpp" #include "caf/scoped_execution_unit.hpp"
#include "caf/typed_continue_helper.hpp" #include "caf/typed_continue_helper.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/typed_event_based_actor.hpp" #include "caf/typed_event_based_actor.hpp"
#include "caf/decorator/adapter.hpp" #include "caf/decorator/adapter.hpp"
......
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include <atomic> #include <atomic>
#include <cstdint> #include <cstdint>
#include <utility>
#include <exception> #include <exception>
#include <functional> #include <functional>
#include <type_traits> #include <type_traits>
...@@ -66,9 +67,40 @@ ...@@ -66,9 +67,40 @@
namespace caf { namespace caf {
namespace detail {
template <class... Ts>
struct make_response_promise_helper {
using type = typed_response_promise<Ts...>;
};
template <class... Ts>
struct make_response_promise_helper<typed_response_promise<Ts...>>
: make_response_promise_helper<Ts...> {};
template <>
struct make_response_promise_helper<response_promise> {
using type = response_promise;
};
} // namespace detail
/// Base class for actors running on this node, either /// Base class for actors running on this node, either
/// living in an own thread or cooperatively scheduled. /// living in an own thread or cooperatively scheduled.
class local_actor : public monitorable_actor, public resumable { class local_actor : public monitorable_actor, public resumable {
private:
template <class... Ts>
typename detail::make_response_promise_helper<Ts...>::type
make_response_promise_impl() {
auto& ptr = current_element_;
if (! ptr)
return {};
auto& mid = ptr->mid;
if (mid.is_answered())
return {};
return {this, *ptr};
}
public: public:
using mailbox_type = detail::single_reader_queue<mailbox_element, using mailbox_type = detail::single_reader_queue<mailbox_element,
detail::disposer>; detail::disposer>;
...@@ -340,15 +372,28 @@ public: ...@@ -340,15 +372,28 @@ public:
/// Returns all joined groups. /// Returns all joined groups.
std::vector<group> joined_groups() const; std::vector<group> joined_groups() const;
/// Creates a `response_promise` to response to a request later on. /// Creates a `response_promise` to respond to a request later on.
response_promise make_response_promise(); inline response_promise make_response_promise() {
return make_response_promise_impl<response_promise>();
}
/// Creates a `response_promise` and responds immediately. /// Creates a `typed_response_promise` to respond to a request later on.
/// Non-requests do not get an error response message. /// `make_response_promise<typed_response_promise<int, int>>()`
/// is equivalent to `make_response_promise<int, int>()`.
template <class... Ts> template <class... Ts>
typed_response_promise<typename std::decay<Ts>::type...> auto make_response_promise()
response(Ts&&... xs) { -> decltype(make_response_promise_impl<Ts...>()) {
auto promise = make_response_promise(); return make_response_promise_impl<Ts...>();
}
/// Creates a `typed_response_promise` and responds immediately.
/// Return type is deduced from arguments.
/// Return value is implicitly convertible to untyped response promise.
template <class... Ts, class R =
typename detail::make_response_promise_helper<
typename std::decay<Ts>::type...>::type>
R response(Ts&&... xs) {
auto promise = make_response_promise<R>();
promise.deliver(std::forward<Ts>(xs)...); promise.deliver(std::forward<Ts>(xs)...);
return promise; return promise;
} }
......
...@@ -35,8 +35,8 @@ public: ...@@ -35,8 +35,8 @@ public:
/// Constructs an invalid response promise. /// Constructs an invalid response promise.
typed_response_promise() = default; typed_response_promise() = default;
inline typed_response_promise(response_promise promise) inline typed_response_promise(local_actor* self, mailbox_element& src)
: promise_(std::move(promise)) { : promise_(self, src) {
// nop // nop
} }
...@@ -45,6 +45,11 @@ public: ...@@ -45,6 +45,11 @@ public:
typed_response_promise& operator=(typed_response_promise&&) = default; typed_response_promise& operator=(typed_response_promise&&) = default;
typed_response_promise& operator=(const typed_response_promise&) = default; typed_response_promise& operator=(const typed_response_promise&) = default;
/// Implicitly convertible to untyped response promise.
inline operator response_promise& () {
return promise_;
}
/// 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 U, class... Us>
typename std::enable_if< typename std::enable_if<
...@@ -65,11 +70,6 @@ public: ...@@ -65,11 +70,6 @@ public:
promise_.deliver(std::move(x)); promise_.deliver(std::move(x));
} }
/// Returns `*this` as an untyped response promise.
inline operator response_promise& () {
return promise_;
}
/// Queries whether this promise is a valid promise that is not satisfied yet. /// Queries whether this promise is a valid promise that is not satisfied yet.
inline bool pending() const { inline bool pending() const {
return promise_.pending(); return promise_.pending();
......
...@@ -997,16 +997,6 @@ void local_actor::delayed_send_impl(message_id mid, const channel& dest, ...@@ -997,16 +997,6 @@ void local_actor::delayed_send_impl(message_id mid, const channel& dest,
mid, std::move(msg)); mid, std::move(msg));
} }
response_promise local_actor::make_response_promise() {
auto& ptr = current_element_;
if (! ptr)
return {};
auto& mid = ptr->mid;
if (mid.is_answered())
return {};
return {this, *ptr};
}
const char* local_actor::name() const { const char* local_actor::name() const {
return "actor"; return "actor";
} }
......
...@@ -68,7 +68,7 @@ public: ...@@ -68,7 +68,7 @@ public:
}); });
send(calculator, next_id_, x); send(calculator, next_id_, x);
auto& entry = promises_[next_id_++]; auto& entry = promises_[next_id_++];
entry = make_response_promise(); entry = make_response_promise<foo_promise>();
return entry; return entry;
}, },
[=](get_atom, int x, int y) -> foo2_promise { [=](get_atom, int x, int y) -> foo2_promise {
...@@ -82,7 +82,7 @@ public: ...@@ -82,7 +82,7 @@ public:
}); });
send(calculator, next_id_, x, y); send(calculator, next_id_, x, y);
auto& entry = promises2_[next_id_++]; auto& entry = promises2_[next_id_++];
entry = make_response_promise(); entry = make_response_promise<foo2_promise>();
// verify move semantics // verify move semantics
CAF_CHECK(entry.pending()); CAF_CHECK(entry.pending());
foo2_promise tmp(std::move(entry)); foo2_promise tmp(std::move(entry));
...@@ -94,7 +94,7 @@ public: ...@@ -94,7 +94,7 @@ public:
return entry; return entry;
}, },
[=](get_atom, double) -> foo3_promise { [=](get_atom, double) -> foo3_promise {
foo3_promise resp = make_response_promise(); auto resp = make_response_promise<double>();
resp.deliver(make_error(sec::unexpected_message)); resp.deliver(make_error(sec::unexpected_message));
return resp; return resp;
}, },
...@@ -147,8 +147,6 @@ CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture) ...@@ -147,8 +147,6 @@ CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture)
CAF_TEST(typed_response_promise) { CAF_TEST(typed_response_promise) {
typed_response_promise<int> resp; typed_response_promise<int> resp;
resp.deliver(1); // delivers on an invalid promise has no effect resp.deliver(1); // delivers on an invalid promise has no effect
CAF_CHECK_EQUAL(static_cast<void*>(&static_cast<response_promise&>(resp)),
static_cast<void*>(&resp));
self->request(foo, get_atom::value, 42).receive( self->request(foo, get_atom::value, 42).receive(
[](int x) { [](int x) {
CAF_CHECK_EQUAL(x, 84); CAF_CHECK_EQUAL(x, 84);
......
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