Commit 05b34523 authored by Dominik Charousset's avatar Dominik Charousset

Improve coverage for the prefix_and_tail operator

parent bac8991f
...@@ -207,6 +207,10 @@ public: ...@@ -207,6 +207,10 @@ public:
bool disposed() const noexcept override; bool disposed() const noexcept override;
}; };
inline auto make_passive_subscription() {
return make_counted<passive_subscription_impl>();
}
namespace op { namespace op {
/// An observable that does nothing when subscribed except returning a trivial /// An observable that does nothing when subscribed except returning a trivial
......
...@@ -22,6 +22,16 @@ namespace { ...@@ -22,6 +22,16 @@ 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();
// Similar to prefix_and_tail::subscribe, but returns a merge_sub pointer
// instead of type-erasing it into a disposable.
template <class T, class Observer>
auto raw_sub(Observer out, size_t psize) {
using flow::op::prefix_and_tail_sub;
auto ptr = make_counted<prefix_and_tail_sub<T>>(ctx.get(), out, psize);
out.on_subscribe(flow::subscription{ptr});
return ptr;
}
}; };
template <class T, class... Ts> template <class T, class... Ts>
...@@ -29,11 +39,10 @@ auto ls(T x, Ts... xs) { ...@@ -29,11 +39,10 @@ auto ls(T x, Ts... xs) {
return std::vector<T>{x, xs...}; return std::vector<T>{x, xs...};
} }
// Note: last is inclusive.
template <class T> template <class T>
auto ls_range(T first, T last) { auto ls_range(T first, T last) {
auto result = std::vector<T>{}; auto result = std::vector<T>{};
for (; first <= last; ++first) for (; first < last; ++first)
result.push_back(first); result.push_back(first);
return result; return result;
} }
...@@ -140,7 +149,7 @@ SCENARIO("prefix_and_tail splits off initial elements") { ...@@ -140,7 +149,7 @@ SCENARIO("prefix_and_tail splits off initial elements") {
.subscribe(snk->as_observer()); .subscribe(snk->as_observer());
ctx->run(); ctx->run();
CHECK_EQ(flat_map_calls, 1); CHECK_EQ(flat_map_calls, 1);
CHECK_EQ(snk->buf, ls_range(8, 256)); CHECK_EQ(snk->buf, ls_range(8, 257));
CHECK_EQ(snk->state, flow::observer_state::completed); CHECK_EQ(snk->state, flow::observer_state::completed);
CHECK_EQ(snk->err, error{}); CHECK_EQ(snk->err, error{});
} }
...@@ -214,4 +223,141 @@ SCENARIO("head_and_tail splits off the first element") { ...@@ -214,4 +223,141 @@ SCENARIO("head_and_tail splits off the first element") {
} }
} }
SCENARIO("head_and_tail forwards errors") {
using tuple_t = cow_tuple<int, flow::observable<int>>;
GIVEN("an observable that emits on_error only") {
WHEN("applying a head_and_tail operator to it") {
THEN("the observer for the head receives on_error") {
auto failed = false;
auto got_tail = false;
ctx->make_observable()
.fail<int>(sec::runtime_error)
.head_and_tail()
.do_on_error([&failed](const error& what) {
failed = true;
CHECK_EQ(what, sec::runtime_error);
})
.for_each([&got_tail](const tuple_t&) { got_tail = true; });
ctx->run();
CHECK(failed);
CHECK(!got_tail);
}
}
}
GIVEN("an observable that emits one value and then on_error") {
WHEN("applying a head_and_tail operator to it") {
THEN("the observer for the tail receives on_error") {
auto head_failed = false;
auto tail_failed = false;
auto got_tail = false;
auto tail_values = 0;
ctx->make_observable()
.just(1)
.concat(ctx->make_observable().fail<int>(sec::runtime_error))
.head_and_tail()
.do_on_error([&head_failed](const error&) { head_failed = true; })
.flat_map([&](const tuple_t& x) {
auto& [head, tail] = x.data();
got_tail = true;
CHECK_EQ(head, 1);
return tail;
})
.do_on_error([&tail_failed](const error& what) {
tail_failed = true;
CHECK_EQ(what, sec::runtime_error);
})
.for_each([&tail_values](int) { ++tail_values; });
ctx->run();
CHECK(got_tail);
CHECK(!head_failed);
CHECK(tail_failed);
CHECK_EQ(tail_values, 0);
}
}
}
}
SCENARIO("head_and_tail requests the prefix as soon as possible") {
using tuple_t = cow_tuple<cow_vector<int>, flow::observable<int>>;
GIVEN("an observable that delays the call to on_subscribe") {
WHEN("the observer requests before on_subscribe from the input arrives") {
THEN("head_and_tail requests the prefix immediately") {
auto snk = flow::make_passive_observer<tuple_t>();
auto uut = raw_sub<int>(snk->as_observer(), 7);
snk->request(42);
ctx->run();
auto in_sub = flow::make_passive_subscription();
uut->on_subscribe(flow::subscription{in_sub});
CHECK_EQ(in_sub->demand, 7u);
}
}
}
}
SCENARIO("head_and_tail disposes unexpected subscriptions") {
using tuple_t = cow_tuple<cow_vector<int>, flow::observable<int>>;
GIVEN("a subscribed head_and_tail operator") {
WHEN("on_subscribe gets called again") {
THEN("the unexpected subscription gets disposed") {
auto snk = flow::make_passive_observer<tuple_t>();
auto uut = raw_sub<int>(snk->as_observer(), 7);
auto sub1 = flow::make_passive_subscription();
auto sub2 = flow::make_passive_subscription();
uut->on_subscribe(flow::subscription{sub1});
uut->on_subscribe(flow::subscription{sub2});
CHECK(!sub1->disposed());
CHECK(sub2->disposed());
}
}
}
}
SCENARIO("disposing head_and_tail disposes the input subscription") {
using tuple_t = cow_tuple<cow_vector<int>, flow::observable<int>>;
GIVEN("a subscribed head_and_tail operator") {
WHEN("calling dispose on the operator") {
THEN("the operator disposes its input") {
auto snk = flow::make_passive_observer<tuple_t>();
auto uut = raw_sub<int>(snk->as_observer(), 7);
auto sub = flow::make_passive_subscription();
uut->on_subscribe(flow::subscription{sub});
CHECK(!uut->disposed());
CHECK(!sub->disposed());
uut->dispose();
CHECK(uut->disposed());
CHECK(sub->disposed());
}
}
}
}
SCENARIO("disposing the tail of head_and_tail disposes the operator") {
using tuple_t = cow_tuple<cow_vector<int>, flow::observable<int>>;
GIVEN("a subscribed head_and_tail operator") {
WHEN("calling dispose the subscription to the tail") {
THEN("the operator gets disposed") {
auto got_tail = false;
auto tail_values = 0;
auto snk = flow::make_passive_observer<int>();
auto sub = //
ctx->make_observable()
.iota(1) //
.take(7)
.prefix_and_tail(3)
.for_each([&](const tuple_t& x) {
got_tail = true;
auto [prefix, tail] = x.data();
auto sub = tail.subscribe(snk->as_observer());
sub.dispose();
});
ctx->run();
CHECK(got_tail);
CHECK_EQ(tail_values, 0);
CHECK(sub.disposed());
CHECK(snk->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