Commit 5d47387c authored by Dominik Charousset's avatar Dominik Charousset

Improve coverage on zip_with and improve API

parent 02ac0018
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/flow/op/never.hpp" #include "caf/flow/op/never.hpp"
#include "caf/flow/op/prefix_and_tail.hpp" #include "caf/flow/op/prefix_and_tail.hpp"
#include "caf/flow/op/publish.hpp" #include "caf/flow/op/publish.hpp"
#include "caf/flow/op/zip_with.hpp"
#include "caf/flow/step/all.hpp" #include "caf/flow/step/all.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
...@@ -330,6 +331,13 @@ public: ...@@ -330,6 +331,13 @@ public:
return materialize().concat_map(std::move(f)); return materialize().concat_map(std::move(f));
} }
/// @copydoc observable::zip_with
template <class F, class T0, class... Ts>
auto zip_with(F fn, T0 input0, Ts... inputs) {
return materialize().zip_with(std::move(fn), std::move(input0),
std::move(inputs)...);
}
/// @copydoc observable::publish /// @copydoc observable::publish
auto publish() && { auto publish() && {
return materialize().publish(); return materialize().publish();
...@@ -683,6 +691,18 @@ auto observable<T>::concat_map(F f) { ...@@ -683,6 +691,18 @@ auto observable<T>::concat_map(F f) {
} }
} }
template <class T>
template <class F, class T0, class... Ts>
auto observable<T>::zip_with(F fn, T0 input0, Ts... inputs) {
using output_type = op::zip_with_output_t<F, T, //
typename T0::output_type, //
typename Ts::output_type...>;
if (pimpl_)
return op::make_zip_with(pimpl_->ctx(), std::move(fn), *this,
std::move(input0), std::move(inputs)...);
return observable<output_type>{};
}
// -- observable: splitting ---------------------------------------------------- // -- observable: splitting ----------------------------------------------------
template <class T> template <class T>
......
...@@ -218,6 +218,8 @@ public: ...@@ -218,6 +218,8 @@ public:
} }
} }
/// Creates an @ref observable that combines the emitted from all passed
/// source observables by applying a function object.
/// @param fn The zip function. Takes one element from each input at a time /// @param fn The zip function. Takes one element from each input at a time
/// and reduces them into a single result. /// and reduces them into a single result.
/// @param input0 The input at index 0. /// @param input0 The input at index 0.
...@@ -225,24 +227,8 @@ public: ...@@ -225,24 +227,8 @@ public:
/// @param inputs The inputs for index > 1, if any. /// @param inputs The inputs for index > 1, if any.
template <class F, class T0, class T1, class... Ts> template <class F, class T0, class T1, class... Ts>
auto zip_with(F fn, T0 input0, T1 input1, Ts... inputs) { auto zip_with(F fn, T0 input0, T1 input1, Ts... inputs) {
using output_type = op::zip_with_output_t<F, // return op::make_zip_with(ctx_, std::move(fn), std::move(input0),
typename T0::output_type, std::move(input1), std::move(inputs)...);
typename T1::output_type,
typename Ts::output_type...>;
using impl_t = op::zip_with<F, //
typename T0::output_type, //
typename T1::output_type, //
typename Ts::output_type...>;
if (input0.valid() && input1.valid() && (inputs.valid() && ...)) {
auto ptr = make_counted<impl_t>(ctx_, std::move(fn),
std::move(input0).as_observable(),
std::move(input1).as_observable(),
std::move(inputs).as_observable()...);
return observable<output_type>{std::move(ptr)};
} else {
auto ptr = make_counted<op::empty<output_type>>(ctx_);
return observable<output_type>{std::move(ptr)};
}
} }
private: private:
......
...@@ -172,6 +172,15 @@ public: ...@@ -172,6 +172,15 @@ public:
template <class Out = output_type, class F> template <class Out = output_type, class F>
auto concat_map(F f); auto concat_map(F f);
/// Creates an @ref observable that combines the emitted from all passed
/// source observables by applying a function object.
/// @param fn The zip function. Takes one element from each input at a time
/// and reduces them into a single result.
/// @param input0 The first additional input.
/// @param inputs Additional inputs, if any.
template <class F, class T0, class... Ts>
auto zip_with(F fn, T0 input0, Ts... inputs);
// -- splitting -------------------------------------------------------------- // -- splitting --------------------------------------------------------------
/// Takes @p prefix_size elements from this observable and emits it in a tuple /// Takes @p prefix_size elements from this observable and emits it in a tuple
......
...@@ -7,7 +7,6 @@ ...@@ -7,7 +7,6 @@
#include "caf/flow/observable_decl.hpp" #include "caf/flow/observable_decl.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp" #include "caf/flow/op/cold.hpp"
#include "caf/flow/op/empty.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include <algorithm> #include <algorithm>
...@@ -36,7 +35,13 @@ struct zip_input { ...@@ -36,7 +35,13 @@ struct zip_input {
using value_type = T; using value_type = T;
subscription sub; subscription sub;
std::vector<T> buf; std::deque<T> buf;
T pop() {
auto result = buf.front();
buf.pop_front();
return result;
}
/// Returns whether the input can no longer produce additional items. /// Returns whether the input can no longer produce additional items.
bool at_end() const noexcept { bool at_end() const noexcept {
...@@ -136,17 +141,14 @@ public: ...@@ -136,17 +141,14 @@ public:
template <size_t I> template <size_t I>
void fwd_on_subscribe(zip_index<I> index, subscription sub) { void fwd_on_subscribe(zip_index<I> index, subscription sub) {
if (out_) { if (out_) {
auto& in = at(index); if (auto& in = at(index); !in.sub) {
if (!in.sub) {
if (demand_ > 0) if (demand_ > 0)
sub.request(demand_); sub.request(demand_);
in.sub = std::move(sub); in.sub = std::move(sub);
} else { return;
sub.dispose();
} }
} else {
sub.dispose();
} }
sub.dispose();
} }
template <size_t I> template <size_t I>
...@@ -163,12 +165,11 @@ public: ...@@ -163,12 +165,11 @@ public:
template <size_t I> template <size_t I>
void fwd_on_error(zip_index<I> index, const error& what) { void fwd_on_error(zip_index<I> index, const error& what) {
if (out_) { if (out_) {
if (!err_)
err_ = what;
auto& input = at(index); auto& input = at(index);
if (input.sub) { if (input.sub)
if (!err_)
err_ = what;
input.sub = nullptr; input.sub = nullptr;
}
if (input.buf.empty()) if (input.buf.empty())
fin(); fin();
} }
...@@ -185,15 +186,12 @@ public: ...@@ -185,15 +186,12 @@ public:
private: private:
void push() { void push() {
if (auto n = std::min(buffered(), demand_); n > 0) { if (auto n = std::min(buffered(), demand_); n > 0) {
for (size_t index = 0; index < n; ++index) {
fold([this, index](auto&... x) { //
out_.on_next(fn_(x.buf[index]...));
});
}
demand_ -= n; demand_ -= n;
for_each_input([n](auto, auto& x) { // for (size_t i = 0; i < n; ++i) {
x.buf.erase(x.buf.begin(), x.buf.begin() + n); fold([this](auto&... x) { out_.on_next(fn_(x.pop()...)); });
}); if (!out_) // on_next might call dispose()
return;
}
} }
if (at_end()) if (at_end())
fin(); fin();
...@@ -268,4 +266,25 @@ private: ...@@ -268,4 +266,25 @@ private:
std::tuple<observable<Ts>...> inputs_; std::tuple<observable<Ts>...> inputs_;
}; };
/// Creates a new zip-with operator from given inputs.
template <class F, class T0, class T1, class... Ts>
auto make_zip_with(coordinator* ctx, F fn, T0 input0, T1 input1, Ts... inputs) {
using output_type = zip_with_output_t<F, //
typename T0::output_type, //
typename T1::output_type, //
typename Ts::output_type...>;
using impl_t = zip_with<F, //
typename T0::output_type, //
typename T1::output_type, //
typename Ts::output_type...>;
if (input0.valid() && input1.valid() && (inputs.valid() && ...)) {
auto ptr = make_counted<impl_t>(ctx, std::move(fn),
std::move(input0).as_observable(),
std::move(input1).as_observable(),
std::move(inputs).as_observable()...);
return observable<output_type>{std::move(ptr)};
}
return observable<output_type>{};
}
} // namespace caf::flow::op } // namespace caf::flow::op
...@@ -17,6 +17,15 @@ namespace { ...@@ -17,6 +17,15 @@ namespace {
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator(); flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
template <class F, class Out, class... Ts>
auto make_zip_with_sub(F fn, flow::observer<Out> out, Ts... inputs) {
using impl_t = flow::op::zip_with_sub<F, typename Ts::output_type...>;
auto pack = std::make_tuple(std::move(inputs).as_observable()...);
auto sub = make_counted<impl_t>(ctx.get(), std::move(fn), out, pack);
out.on_subscribe(flow::subscription{sub});
return sub;
}
}; };
} // namespace } // namespace
...@@ -68,4 +77,164 @@ SCENARIO("zip_with emits nothing when zipping an empty observable") { ...@@ -68,4 +77,164 @@ SCENARIO("zip_with emits nothing when zipping an empty observable") {
} }
} }
SCENARIO("zip_with aborts if an input emits an error") {
GIVEN("two observables, one of them emits an error after some items") {
WHEN("merging them with zip_with") {
THEN("the observer receives all items up to the error") {
auto obs = ctx->make_observable();
auto snk = flow::make_auto_observer<int>();
obs //
.iota(1)
.take(3)
.concat(obs.fail<int>(sec::runtime_error))
.zip_with([](int x, int y) { return x + y; }, obs.iota(1).take(10))
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->aborted());
CHECK_EQ(snk->buf, std::vector<int>({2, 4, 6}));
}
}
}
GIVEN("two observables, one of them emits an error immediately") {
WHEN("merging them with zip_with") {
THEN("the observer only receives on_error") {
auto obs = ctx->make_observable();
auto snk = flow::make_auto_observer<int>();
obs //
.iota(1)
.take(3)
.zip_with([](int x, int y) { return x + y; },
obs.fail<int>(sec::runtime_error))
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->aborted());
CHECK(snk->buf.empty());
}
}
}
}
SCENARIO("zip_with on an invalid observable produces an invalid observable") {
GIVEN("a default-constructed (invalid) observable") {
WHEN("calling zip_with on it") {
THEN("the result is another invalid observable") {
auto obs = ctx->make_observable();
auto snk = flow::make_auto_observer<int>();
flow::observable<int>{}
.zip_with([](int x, int y) { return x + y; }, obs.iota(1).take(10))
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->aborted());
CHECK(snk->buf.empty());
}
}
}
GIVEN("a valid observable") {
WHEN("calling zip_with on it with an invalid observable") {
THEN("the result is another invalid observable") {
auto obs = ctx->make_observable();
auto snk = flow::make_auto_observer<int>();
obs //
.iota(1)
.take(10)
.zip_with([](int x, int y) { return x + y; }, flow::observable<int>{})
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->aborted());
CHECK(snk->buf.empty());
}
}
}
}
SCENARIO("zip_with operators can be disposed at any time") {
GIVEN("a zip_with operator that produces some items") {
WHEN("calling dispose before requesting any items") {
THEN("the observer never receives any item") {
auto obs = ctx->make_observable();
auto snk = flow::make_passive_observer<int>();
auto sub = //
obs //
.iota(1)
.take(10)
.zip_with([](int x, int y) { return x + y; }, obs.iota(1))
.subscribe(snk->as_observer());
CHECK(!sub.disposed());
sub.dispose();
ctx->run();
CHECK(snk->completed());
}
}
WHEN("calling dispose in on_subscribe") {
THEN("the observer receives no item") {
auto obs = ctx->make_observable();
auto snk = flow::make_canceling_observer<int>();
obs //
.iota(1)
.take(10)
.zip_with([](int x, int y) { return x + y; }, obs.iota(1))
.subscribe(snk->as_observer());
ctx->run();
CHECK_EQ(snk->on_next_calls, 0);
}
}
WHEN("calling dispose in on_next") {
THEN("the observer receives no additional item") {
auto obs = ctx->make_observable();
auto snk = flow::make_canceling_observer<int>(true);
obs //
.iota(1)
.take(10)
.zip_with([](int x, int y) { return x + y; }, obs.iota(1))
.subscribe(snk->as_observer());
ctx->run();
CHECK_EQ(snk->on_next_calls, 1);
}
}
}
}
SCENARIO("observers may request from zip_with operators before on_subscribe") {
GIVEN("a zip_with operator with two inputs") {
WHEN("the observer calls request before the inputs call on_subscribe") {
THEN("the observer receives the item") {
using flow::op::zip_index;
auto snk = flow::make_passive_observer<int>();
auto uut = make_zip_with_sub([](int, int) { return 0; },
snk->as_observer(),
flow::make_nil_observable<int>(ctx.get()),
flow::make_nil_observable<int>(ctx.get()));
snk->request(128);
auto sub1 = flow::make_passive_subscription();
auto sub2 = flow::make_passive_subscription();
uut->fwd_on_subscribe(zip_index<0>{}, flow::subscription{sub1});
uut->fwd_on_subscribe(zip_index<1>{}, flow::subscription{sub2});
CHECK_EQ(sub1->demand, 128u);
CHECK_EQ(sub2->demand, 128u);
}
}
}
}
SCENARIO("the zip_with operators disposes unexpected subscriptions") {
GIVEN("a zip_with operator with two inputs") {
WHEN("on_subscribe is called twice for the same input") {
THEN("the operator disposes the subscription") {
using flow::op::zip_index;
auto snk = flow::make_passive_observer<int>();
auto uut = make_zip_with_sub([](int, int) { return 0; },
snk->as_observer(),
flow::make_nil_observable<int>(ctx.get()),
flow::make_nil_observable<int>(ctx.get()));
auto sub1 = flow::make_passive_subscription();
auto sub2 = flow::make_passive_subscription();
uut->fwd_on_subscribe(zip_index<0>{}, flow::subscription{sub1});
uut->fwd_on_subscribe(zip_index<0>{}, flow::subscription{sub2});
CHECK(!sub1->disposed());
CHECK(sub2->disposed());
}
}
}
}
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