Commit d9842dba authored by Dominik Charousset's avatar Dominik Charousset

Support callable sources that emit optional values

parent c9aa5463
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
#include <array> #include <array>
#include <chrono> #include <chrono>
#include <functional> #include <functional>
#include <optional>
#include <string> #include <string>
#include <tuple> #include <tuple>
#include <type_traits> #include <type_traits>
...@@ -1029,6 +1030,29 @@ struct is_builtin_inspector_type<span<const byte>, false> { ...@@ -1029,6 +1030,29 @@ struct is_builtin_inspector_type<span<const byte>, false> {
static constexpr bool value = true; static constexpr bool value = true;
}; };
/// Checks whether `T` is a `std::optional`.
template <class T>
struct is_optional : std::false_type {};
template <class T>
struct is_optional<std::optional<T>> : std::true_type {};
template <class T>
constexpr bool is_optional_v = is_optional<T>::value;
template <class T>
struct unboxed_oracle {
using type = T;
};
template <class T>
struct unboxed_oracle<std::optional<T>> {
using type = T;
};
template <class T>
using unboxed_t = typename unboxed_oracle<T>::type;
} // namespace caf::detail } // namespace caf::detail
#undef CAF_HAS_MEMBER_TRAIT #undef CAF_HAS_MEMBER_TRAIT
......
...@@ -287,7 +287,11 @@ private: ...@@ -287,7 +287,11 @@ private:
template <class F> template <class F>
class callable_source { class callable_source {
public: public:
using output_type = std::decay_t<decltype(std::declval<F&>()())>; using callable_res_t = std::decay_t<decltype(std::declval<F&>()())>;
static constexpr bool boxed_output = detail::is_optional_v<callable_res_t>;
using output_type = detail::unboxed_t<callable_res_t>;
explicit callable_source(F fn) : fn_(std::move(fn)) { explicit callable_source(F fn) : fn_(std::move(fn)) {
// nop // nop
...@@ -302,9 +306,19 @@ public: ...@@ -302,9 +306,19 @@ public:
template <class Step, class... Steps> template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) { void pull(size_t n, Step& step, Steps&... steps) {
CAF_LOG_TRACE(CAF_ARG(n)); CAF_LOG_TRACE(CAF_ARG(n));
for (size_t i = 0; i < n; ++i) for (size_t i = 0; i < n; ++i) {
if (!step.on_next(fn_(), steps...)) if constexpr (boxed_output) {
return; auto val = fn_();
if (!val) {
step.on_complete(steps...);
return;
} else if (!step.on_next(*val, steps...))
return;
} else {
if (!step.on_next(fn_(), steps...))
return;
}
}
} }
private: private:
......
...@@ -26,7 +26,7 @@ struct fixture : test_coordinator_fixture<> { ...@@ -26,7 +26,7 @@ struct fixture : test_coordinator_fixture<> {
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("the repeater source repeats one value indefinitely") { SCENARIO("repeater sources repeat one value indefinitely") {
GIVEN("a repeater source") { GIVEN("a repeater source") {
WHEN("subscribing to its output") { WHEN("subscribing to its output") {
THEN("the observer receives the same value over and over again") { THEN("the observer receives the same value over and over again") {
...@@ -51,7 +51,7 @@ SCENARIO("the repeater source repeats one value indefinitely") { ...@@ -51,7 +51,7 @@ SCENARIO("the repeater source repeats one value indefinitely") {
} }
} }
SCENARIO("the container source streams its input values") { SCENARIO("container sources stream their input values") {
GIVEN("a container source") { GIVEN("a container source") {
WHEN("subscribing to its output") { WHEN("subscribing to its output") {
THEN("the observer receives the values from the container in order") { THEN("the observer receives the values from the container in order") {
...@@ -75,4 +75,124 @@ SCENARIO("the container source streams its input values") { ...@@ -75,4 +75,124 @@ SCENARIO("the container source streams its input values") {
} }
} }
SCENARIO("value sources produce exactly one input") {
GIVEN("a value source") {
WHEN("subscribing to its output") {
THEN("the observer receives one value") {
using ivec = std::vector<int>;
auto snk = flow::make_passive_observer<int>();
ctx->make_observable().just(42).subscribe(snk->as_observer());
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty());
if (CHECK(snk->sub)) {
snk->sub.request(100);
ctx->run();
CHECK_EQ(snk->buf, ivec({42}));
CHECK_EQ(snk->state, flow::observer_state::completed);
}
}
}
}
}
SCENARIO("callable sources stream values generated from a function object") {
GIVEN("a callable source returning non-optional values") {
WHEN("subscribing to its output") {
THEN("the observer receives an indefinite amount of values") {
auto f = [n = 1]() mutable { return n++; };
using ivec = std::vector<int>;
auto snk = flow::make_passive_observer<int>();
ctx->make_observable().from_callable(f).subscribe(snk->as_observer());
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty());
if (CHECK(snk->sub)) {
snk->sub.request(3);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3}));
snk->sub.request(4);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7}));
snk->sub.cancel();
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7}));
}
}
}
}
GIVEN("a callable source returning optional values") {
WHEN("subscribing to its output") {
THEN("the observer receives value until the callable return null-opt") {
auto f = [n = 1]() mutable -> std::optional<int> {
if (n < 8)
return n++;
else
return std::nullopt;
};
using ivec = std::vector<int>;
auto snk = flow::make_passive_observer<int>();
ctx->make_observable().from_callable(f).subscribe(snk->as_observer());
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty());
if (CHECK(snk->sub)) {
snk->sub.request(3);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3}));
snk->sub.request(21);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk->state, flow::observer_state::completed);
}
}
}
}
}
namespace {
class custom_pullable {
public:
using output_type = int;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
for (size_t i = 0; i < n; ++i) {
if (value_ > 7) {
step.on_complete(steps...);
return;
} else if (!step.on_next(value_++, steps...)) {
return;
}
}
}
private:
int value_ = 1;
};
} // namespace
SCENARIO("lifting converts a Pullable into an observable") {
GIVEN("a lifted implementation of the Pullable concept") {
WHEN("subscribing to its output") {
THEN("the observer receives the generated values") {
using ivec = std::vector<int>;
auto snk = flow::make_passive_observer<int>();
auto f = custom_pullable{};
ctx->make_observable().lift(f).subscribe(snk->as_observer());
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty());
if (CHECK(snk->sub)) {
snk->sub.request(3);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3}));
snk->sub.request(21);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk->state, flow::observer_state::completed);
}
}
}
}
}
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
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