Commit 179f21f3 authored by Dominik Charousset's avatar Dominik Charousset

Allow converting futures to observables

parent 95aeba25
...@@ -9,6 +9,8 @@ ...@@ -9,6 +9,8 @@
#include "caf/detail/async_cell.hpp" #include "caf/detail/async_cell.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/op/cell.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
namespace caf::async { namespace caf::async {
...@@ -128,6 +130,17 @@ public: ...@@ -128,6 +130,17 @@ public:
return {&ctx, cell_}; return {&ctx, cell_};
} }
/// Binds this future to a @ref coordinator and converts it to an
/// @ref observable.
/// @pre `valid()`
flow::observable<T> observe_on(flow::coordinator* ctx) const {
using flow_cell_t = flow::op::cell<T>;
auto ptr = make_counted<flow_cell_t>(ctx);
bind_to(ctx).then([ptr](const T& val) { ptr->set_value(val); },
[ptr](const error& what) { ptr->set_error(what); });
return flow::observable<T>{ptr};
}
/// Queries whether the result of the asynchronous computation is still /// Queries whether the result of the asynchronous computation is still
/// pending, i.e., neither `set_value` nor `set_error` has been called on the /// pending, i.e., neither `set_value` nor `set_error` has been called on the
/// @ref promise. /// @ref promise.
......
...@@ -14,13 +14,28 @@ ...@@ -14,13 +14,28 @@
using namespace caf; using namespace caf;
using namespace std::literals; using namespace std::literals;
namespace {
template <class T>
auto make_shared_val_ptr() {
return std::make_shared<std::variant<none_t, T, error>>();
}
template <class T>
auto make_observer(std::shared_ptr<std::variant<none_t, T, error>> ptr) {
return flow::make_observer([ptr](const T& val) { *ptr = val; },
[ptr](const error& what) { *ptr = what; });
}
} // namespace
BEGIN_FIXTURE_SCOPE(test_coordinator_fixture<>) BEGIN_FIXTURE_SCOPE(test_coordinator_fixture<>)
SCENARIO("actors can observe futures") { SCENARIO("actors can observe futures") {
GIVEN("a promise and future pair") { GIVEN("a promise and future pair") {
WHEN("passing a non-ready future to an actor") { WHEN("passing a non-ready future to an actor") {
THEN("the actor can observe the value via .then() later") { THEN("it can observe the value via .then() later") {
auto val = std::make_shared<std::variant<none_t, std::string, error>>(); auto val = make_shared_val_ptr<std::string>();
auto uut = async::promise<std::string>{}; auto uut = async::promise<std::string>{};
auto fut = uut.get_future(); auto fut = uut.get_future();
auto testee = sys.spawn([val, fut](event_based_actor* self) { auto testee = sys.spawn([val, fut](event_based_actor* self) {
...@@ -34,10 +49,24 @@ SCENARIO("actors can observe futures") { ...@@ -34,10 +49,24 @@ SCENARIO("actors can observe futures") {
if (CHECK(std::holds_alternative<std::string>(*val))) if (CHECK(std::holds_alternative<std::string>(*val)))
CHECK_EQ(std::get<std::string>(*val), "hello world"); CHECK_EQ(std::get<std::string>(*val), "hello world");
} }
AND_THEN("it can observe the value via .observe_on() later") {
auto val = make_shared_val_ptr<std::string>();
auto uut = async::promise<std::string>{};
auto fut = uut.get_future();
auto testee = sys.spawn([val, fut](event_based_actor* self) {
fut.observe_on(self).subscribe(make_observer(val));
});
run();
CHECK(std::holds_alternative<none_t>(*val));
uut.set_value("hello world"s);
expect((action), to(testee));
if (CHECK(std::holds_alternative<std::string>(*val)))
CHECK_EQ(std::get<std::string>(*val), "hello world");
}
} }
WHEN("passing a ready future to an actor") { WHEN("passing a ready future to an actor") {
THEN("the actor can observe the value via .then() immediately") { THEN("it can observe the value via .then() immediately") {
auto val = std::make_shared<std::variant<none_t, std::string, error>>(); auto val = make_shared_val_ptr<std::string>();
auto uut = async::promise<std::string>{}; auto uut = async::promise<std::string>{};
auto fut = uut.get_future(); auto fut = uut.get_future();
uut.set_value("hello world"s); uut.set_value("hello world"s);
...@@ -49,10 +78,22 @@ SCENARIO("actors can observe futures") { ...@@ -49,10 +78,22 @@ SCENARIO("actors can observe futures") {
if (CHECK(std::holds_alternative<std::string>(*val))) if (CHECK(std::holds_alternative<std::string>(*val)))
CHECK_EQ(std::get<std::string>(*val), "hello world"); CHECK_EQ(std::get<std::string>(*val), "hello world");
} }
AND_THEN("it can observe the value via .observe_on() immediately") {
auto val = make_shared_val_ptr<std::string>();
auto uut = async::promise<std::string>{};
auto fut = uut.get_future();
uut.set_value("hello world"s);
auto testee = sys.spawn([val, fut](event_based_actor* self) {
fut.observe_on(self).subscribe(make_observer(val));
});
run();
if (CHECK(std::holds_alternative<std::string>(*val)))
CHECK_EQ(std::get<std::string>(*val), "hello world");
}
} }
WHEN("passing a non-ready future to an actor and disposing the action") { WHEN("passing a non-ready future to an actor and disposing the action") {
THEN("the actor never observes the value") { THEN("it never observes the value with .then()") {
auto val = std::make_shared<std::variant<none_t, std::string, error>>(); auto val = make_shared_val_ptr<std::string>();
auto uut = async::promise<std::string>{}; auto uut = async::promise<std::string>{};
auto fut = uut.get_future(); auto fut = uut.get_future();
auto hdl = disposable{}; auto hdl = disposable{};
...@@ -75,7 +116,7 @@ SCENARIO("actors can observe futures") { ...@@ -75,7 +116,7 @@ SCENARIO("actors can observe futures") {
SCENARIO("never setting a value or an error breaks the promises") { SCENARIO("never setting a value or an error breaks the promises") {
GIVEN("multiple promises that point to the same cell") { GIVEN("multiple promises that point to the same cell") {
WHEN("the last promise goes out of scope") { WHEN("the last promise goes out of scope") {
THEN("the future reports a broken promise") { THEN("the future reports a broken promise when using .then()") {
using promise_t = async::promise<int32_t>; using promise_t = async::promise<int32_t>;
using future_t = async::future<int32_t>; using future_t = async::future<int32_t>;
future_t fut; future_t fut;
...@@ -119,6 +160,24 @@ SCENARIO("never setting a value or an error breaks the promises") { ...@@ -119,6 +160,24 @@ SCENARIO("never setting a value or an error breaks the promises") {
ctx->run(); ctx->run();
CHECK_EQ(observed_events, 1u); CHECK_EQ(observed_events, 1u);
} }
AND_THEN("the future reports a broken promise when using .observe_on()") {
using promise_t = async::promise<int32_t>;
using future_t = async::future<int32_t>;
future_t fut;
{
auto uut = promise_t{};
fut = uut.get_future();
CHECK(fut.pending());
}
CHECK(!fut.pending());
auto val = make_shared_val_ptr<int32_t>();
sys.spawn([val, fut](event_based_actor* self) {
fut.observe_on(self).subscribe(make_observer(val));
});
run();
if (CHECK(std::holds_alternative<error>(*val)))
CHECK_EQ(std::get<error>(*val), sec::broken_promise);
}
} }
} }
} }
......
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