Commit 86d0fb7e authored by Dominik Charousset's avatar Dominik Charousset

More op::buffer coverage; make test code private

parent 335d3a2d
......@@ -145,6 +145,18 @@ public:
int compare(uint8_t code, type_id_t category) const noexcept;
/// Returns a copy of `this` if `!empty()` or else returns a new error from
/// given arguments.
template <class Enum, class... Ts>
error or_else(Enum code, Ts&&... args) const {
if (!empty())
return *this;
if constexpr (sizeof...(Ts) > 0)
return error{code, make_message(std::forward<Ts>(args)...)};
else
return error{code};
}
// -- modifiers --------------------------------------------------------------
/// Reverts this error to "not an error" as if calling `*this = error{}`.
......
......@@ -522,133 +522,4 @@ private:
Token token_;
};
/// An observer with minimal internal logic. Useful for writing unit tests.
template <class T>
class passive_observer : public observer_impl_base<T> {
public:
// -- implementation of observer_impl<T> -------------------------------------
void on_complete() override {
if (sub) {
sub.dispose();
sub = nullptr;
}
state = observer_state::completed;
}
void on_error(const error& what) override {
if (sub) {
sub.dispose();
sub = nullptr;
}
err = what;
state = observer_state::aborted;
}
void on_subscribe(subscription new_sub) override {
if (state == observer_state::idle) {
CAF_ASSERT(!sub);
sub = std::move(new_sub);
state = observer_state::subscribed;
} else {
new_sub.dispose();
}
}
void on_next(const T& item) override {
buf.emplace_back(item);
}
// -- convenience functions --------------------------------------------------
bool request(size_t demand) {
if (sub) {
sub.request(demand);
return true;
} else {
return false;
}
}
void unsubscribe() {
if (sub) {
sub.dispose();
state = observer_state::idle;
}
}
bool idle() const noexcept {
return state == observer_state::idle;
}
bool subscribed() const noexcept {
return state == observer_state::subscribed;
}
bool completed() const noexcept {
return state == observer_state::completed;
}
bool aborted() const noexcept {
return state == observer_state::aborted;
}
std::vector<T> sorted_buf() const {
auto result = buf;
std::sort(result.begin(), result.end());
return result;
}
// -- member variables -------------------------------------------------------
/// The subscription for requesting additional items.
subscription sub;
/// Default-constructed unless on_error was called.
error err;
/// Represents the current state of this observer.
observer_state state = observer_state::idle;
/// Stores all items received via `on_next`.
std::vector<T> buf;
};
/// @relates passive_observer
template <class T>
intrusive_ptr<passive_observer<T>> make_passive_observer() {
return make_counted<passive_observer<T>>();
}
/// Similar to @ref passive_observer but automatically requests items until
/// completed. Useful for writing unit tests.
template <class T>
class auto_observer : public passive_observer<T> {
public:
// -- implementation of observer_impl<T> -------------------------------------
void on_subscribe(subscription new_sub) override {
if (this->state == observer_state::idle) {
CAF_ASSERT(!this->sub);
this->sub = std::move(new_sub);
this->state = observer_state::subscribed;
this->sub.request(64);
} else {
new_sub.dispose();
}
}
void on_next(const T& item) override {
this->buf.emplace_back(item);
if (this->sub)
this->sub.request(1);
}
};
/// @relates auto_observer
template <class T>
intrusive_ptr<auto_observer<T>> make_auto_observer() {
return make_counted<auto_observer<T>>();
}
} // namespace caf::flow
......@@ -69,6 +69,24 @@ public:
// nop
}
// -- properties -------------------------------------------------------------
bool running() const noexcept {
return state_ == state::running;
}
const error& err() const noexcept {
return err_;
}
size_t pending() const noexcept {
return buf_.size();
}
bool can_emit() const noexcept {
return buf_.size() == max_buf_size_ || has_shut_down(state_);
}
// -- callbacks for the parent -----------------------------------------------
void init(observable<input_type> vals, observable<select_token_type> ctrl) {
......@@ -76,21 +94,22 @@ public:
using ctrl_fwd_t = forwarder<select_token_type, buffer_sub, buffer_emit_t>;
vals.subscribe(
make_counted<val_fwd_t>(this, buffer_input_t{})->as_observer());
ctrl.subscribe(
make_counted<ctrl_fwd_t>(this, buffer_emit_t{})->as_observer());
// Note: the previous subscribe might call on_error, in which case we don't
// need to try to subscribe to the control observable.
if (running())
ctrl.subscribe(
make_counted<ctrl_fwd_t>(this, buffer_emit_t{})->as_observer());
}
// -- callbacks for the forwarders -------------------------------------------
void fwd_on_subscribe(buffer_input_t, subscription sub) {
if (state_ != state::idle || value_sub_ || !out_) {
if (!running() || value_sub_ || !out_) {
sub.dispose();
return;
}
value_sub_ = std::move(sub);
value_sub_.request(max_buf_size_);
if (control_sub_)
state_ = state::running;
}
void fwd_on_complete(buffer_input_t) {
......@@ -105,27 +124,20 @@ public:
}
void fwd_on_next(buffer_input_t, const input_type& item) {
switch (state_) {
case state::idle:
case state::running:
buf_.push_back(item);
if (buf_.size() == max_buf_size_)
do_emit();
break;
default:
break;
if (running()) {
buf_.push_back(item);
if (buf_.size() == max_buf_size_)
do_emit();
}
}
void fwd_on_subscribe(buffer_emit_t, subscription sub) {
if (state_ != state::idle || control_sub_ || !out_) {
if (!running() || control_sub_ || !out_) {
sub.dispose();
return;
}
control_sub_ = std::move(sub);
control_sub_.request(1);
if (value_sub_)
state_ = state::running;
}
void fwd_on_complete(buffer_emit_t) {
......@@ -177,15 +189,10 @@ public:
}
private:
bool can_emit() const noexcept {
return buf_.size() == max_buf_size_ || has_shut_down(state_);
}
void shutdown() {
value_sub_.dispose();
control_sub_.dispose();
switch (state_) {
case state::idle:
case state::running:
if (!buf_.empty()) {
if (demand_ == 0) {
......@@ -211,33 +218,27 @@ private:
void on_request() {
if (demand_ == 0 || !can_emit())
return;
switch (state_) {
case state::idle:
case state::running:
CAF_ASSERT(buf_.size() == max_buf_size_);
do_emit();
break;
case state::completed:
case state::aborted:
if (!buf_.empty())
do_emit();
if (err_)
out_.on_error(err_);
else
out_.on_complete();
out_ = nullptr;
break;
default:
break;
if (running()) {
CAF_ASSERT(buf_.size() == max_buf_size_);
do_emit();
return;
}
if (!buf_.empty())
do_emit();
if (err_)
out_.on_error(err_);
else
out_.on_complete();
out_ = nullptr;
}
void do_emit() {
CAF_ASSERT(demand_ > 0);
if (demand_ == 0)
return;
Trait f;
--demand_;
out_.on_next(f(buf_));
auto buffered = buf_.size();
out_.on_next(f(buf_));
buf_.clear();
if (value_sub_ && buffered > 0)
value_sub_.request(buffered);
......@@ -277,12 +278,11 @@ private:
size_t demand_ = 0;
/// Our current state.
/// - idle: until we have received both subscriptions.
/// - running: emitting batches.
/// - running: alive and ready to emit batches.
/// - completed: on_complete was called but some data is still buffered.
/// - aborted: on_error was called but some data is still buffered.
/// - disposed: inactive.
state state_ = state::idle;
state state_ = state::running;
/// Caches the abort reason.
error err_;
......@@ -318,6 +318,12 @@ public:
disposable subscribe(observer<output_type> out) override {
auto ptr = make_counted<buffer_sub<Trait>>(super::ctx_, max_items_, out);
ptr->init(in_, select_);
if (!ptr->running()) {
auto err = ptr->err().or_else(sec::runtime_error,
"failed to initialize buffer subscription");
out.on_error(err);
return {};
}
out.on_subscribe(subscription{ptr});
return ptr->as_disposable();
}
......
......@@ -4,6 +4,67 @@
#include "core-test.hpp"
#include <atomic>
namespace {
/// A trivial disposable with an atomic flag.
class trivial_impl : public caf::ref_counted, public caf::disposable::impl {
public:
trivial_impl() : flag_(false) {
// nop
}
void dispose() override {
flag_ = true;
}
bool disposed() const noexcept override {
return flag_.load();
}
void ref_disposable() const noexcept override {
ref();
}
void deref_disposable() const noexcept override {
deref();
}
friend void intrusive_ptr_add_ref(const trivial_impl* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const trivial_impl* ptr) noexcept {
ptr->deref();
}
private:
std::atomic<bool> flag_;
};
} // namespace
namespace caf::flow {
disposable make_trivial_disposable() {
return disposable{make_counted<trivial_impl>()};
}
void passive_subscription_impl::request(size_t n) {
demand += n;
}
void passive_subscription_impl::dispose() {
disposed_flag = true;
}
bool passive_subscription_impl::disposed() const noexcept {
return disposed_flag;
}
} // namespace caf::flow
std::string to_string(level lvl) {
switch (lvl) {
case level::all:
......
......@@ -16,6 +16,227 @@
#include <string>
#include <utility>
// -- utility for testing flows ------------------------------------------------
namespace caf::flow {
/// Returns a trivial disposable that wraps an atomic flag.
disposable make_trivial_disposable();
/// An observer with minimal internal logic. Useful for writing unit tests.
template <class T>
class passive_observer : public observer_impl_base<T> {
public:
// -- implementation of observer_impl<T> -------------------------------------
void on_complete() override {
if (sub) {
sub.dispose();
sub = nullptr;
}
state = observer_state::completed;
}
void on_error(const error& what) override {
if (sub) {
sub.dispose();
sub = nullptr;
}
err = what;
state = observer_state::aborted;
}
void on_subscribe(subscription new_sub) override {
if (state == observer_state::idle) {
CAF_ASSERT(!sub);
sub = std::move(new_sub);
state = observer_state::subscribed;
} else {
new_sub.dispose();
}
}
void on_next(const T& item) override {
buf.emplace_back(item);
}
// -- convenience functions --------------------------------------------------
bool request(size_t demand) {
if (sub) {
sub.request(demand);
return true;
} else {
return false;
}
}
void unsubscribe() {
if (sub) {
sub.dispose();
state = observer_state::idle;
}
}
bool idle() const noexcept {
return state == observer_state::idle;
}
bool subscribed() const noexcept {
return state == observer_state::subscribed;
}
bool completed() const noexcept {
return state == observer_state::completed;
}
bool aborted() const noexcept {
return state == observer_state::aborted;
}
std::vector<T> sorted_buf() const {
auto result = buf;
std::sort(result.begin(), result.end());
return result;
}
// -- member variables -------------------------------------------------------
/// The subscription for requesting additional items.
subscription sub;
/// Default-constructed unless on_error was called.
error err;
/// Represents the current state of this observer.
observer_state state = observer_state::idle;
/// Stores all items received via `on_next`.
std::vector<T> buf;
};
/// @relates passive_observer
template <class T>
intrusive_ptr<passive_observer<T>> make_passive_observer() {
return make_counted<passive_observer<T>>();
}
/// Similar to @ref passive_observer but automatically requests items until
/// completed. Useful for writing unit tests.
template <class T>
class auto_observer : public passive_observer<T> {
public:
// -- implementation of observer_impl<T> -------------------------------------
void on_subscribe(subscription new_sub) override {
if (this->state == observer_state::idle) {
CAF_ASSERT(!this->sub);
this->sub = std::move(new_sub);
this->state = observer_state::subscribed;
this->sub.request(64);
} else {
new_sub.dispose();
}
}
void on_next(const T& item) override {
this->buf.emplace_back(item);
if (this->sub)
this->sub.request(1);
}
};
/// @relates auto_observer
template <class T>
intrusive_ptr<auto_observer<T>> make_auto_observer() {
return make_counted<auto_observer<T>>();
}
/// A subscription implementation without internal logic.
class passive_subscription_impl final : public subscription::impl_base {
public:
/// Incremented by `request`.
size_t demand = 0;
/// Flipped by `dispose`.
bool disposed_flag = false;
void request(size_t n) override;
void dispose() override;
bool disposed() const noexcept override;
};
namespace op {
/// An observable that does nothing when subscribed except returning a trivial
/// disposable. Allows tests to call on_subscribe some time later.
template <class T>
class nil_observable : public op::cold<T> {
public:
using super = op::cold<T>;
using shared_count = std::shared_ptr<size_t>;
nil_observable(coordinator* ctx, shared_count subscribe_count)
: super(ctx), subscribe_count_(std::move(subscribe_count)) {
// nop
}
disposable subscribe(observer<T>) override {
if (subscribe_count_)
*subscribe_count_ += 1;
return make_trivial_disposable();
}
shared_count subscribe_count_;
};
/// An observable that passes a trivial disposable to any observer.
template <class T>
class trivial_observable : public op::cold<T> {
public:
using super = op::cold<T>;
using shared_count = std::shared_ptr<size_t>;
trivial_observable(coordinator* ctx, shared_count subscribe_count)
: super(ctx), subscribe_count_(std::move(subscribe_count)) {
// nop
}
disposable subscribe(observer<T> out) override {
if (subscribe_count_)
*subscribe_count_ += 1;
auto ptr = make_counted<passive_subscription_impl>();
out.on_subscribe(subscription{ptr});
return make_trivial_disposable();
}
shared_count subscribe_count_;
};
} // namespace op
template <class T>
observable<T>
make_nil_observable(coordinator* ctx,
std::shared_ptr<size_t> subscribe_count = nullptr) {
auto ptr = make_counted<op::nil_observable<T>>(ctx, subscribe_count);
return observable<T>{std::move(ptr)};
}
template <class T>
observable<T>
make_trivial_observable(coordinator* ctx,
std::shared_ptr<size_t> subscribe_count = nullptr) {
auto ptr = make_counted<op::trivial_observable<T>>(ctx, subscribe_count);
return observable<T>{std::move(ptr)};
}
} // namespace caf::flow
// -- utility for testing serialization round-trips ----------------------------
template <class T>
......
......@@ -20,8 +20,60 @@ using namespace std::literals;
namespace {
constexpr auto fwd_data = flow::op::buffer_input_t{};
constexpr auto fwd_ctrl = flow::op::buffer_emit_t{};
struct skip_trait {
static constexpr bool skip_empty = true;
using input_type = int;
using output_type = cow_vector<int>;
using select_token_type = int64_t;
output_type operator()(const std::vector<input_type>& xs) {
return output_type{xs};
}
};
struct noskip_trait {
static constexpr bool skip_empty = false;
using input_type = int;
using output_type = cow_vector<int>;
using select_token_type = int64_t;
output_type operator()(const std::vector<input_type>& xs) {
return output_type{xs};
}
};
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
// Similar to buffer::subscribe, but returns a buffer_sub pointer instead of
// type-erasing it into a disposable.
template <class Trait = noskip_trait>
auto raw_sub(size_t max_items, flow::observable<int> in,
flow::observable<int64_t> select,
flow::observer<cow_vector<int>> out) {
using sub_t = flow::op::buffer_sub<Trait>;
auto ptr = make_counted<sub_t>(ctx.get(), max_items, out);
ptr->init(in, select);
out.on_subscribe(flow::subscription{ptr});
return ptr;
}
template <class Impl>
void add_subs(intrusive_ptr<Impl> uut) {
auto data_sub = make_counted<flow::passive_subscription_impl>();
uut->fwd_on_subscribe(fwd_data, flow::subscription{std::move(data_sub)});
auto ctrl_sub = make_counted<flow::passive_subscription_impl>();
uut->fwd_on_subscribe(fwd_ctrl, flow::subscription{std::move(ctrl_sub)});
}
template <class T>
auto trivial_obs() {
return flow::make_trivial_observable<T>(ctx.get());
}
};
} // namespace
......@@ -64,9 +116,8 @@ SCENARIO("the buffer operator forces items at regular intervals") {
};
auto pub = flow::item_publisher<int>{ctx.get()};
sys.spawn([&pub, outputs](caf::event_based_actor* self) {
pub //
.as_observable()
.observe_on(self)
pub.as_observable()
.observe_on(self) //
.buffer(3, 1s)
.for_each([outputs](const cow_vector<int>& xs) {
outputs->emplace_back(xs);
......@@ -152,4 +203,337 @@ SCENARIO("the buffer operator forwards errors") {
}
}
SCENARIO("buffers start to emit items once subscribed") {
GIVEN("a buffer operator") {
WHEN("the selector never calls on_subscribe") {
THEN("the buffer still emits batches") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, flow::make_nil_observable<int>(ctx.get()),
flow::make_nil_observable<int64_t>(ctx.get()),
snk->as_observer());
auto data_sub = make_counted<flow::passive_subscription_impl>();
uut->fwd_on_subscribe(fwd_data, flow::subscription{data_sub});
ctx->run();
REQUIRE_GE(data_sub->demand, 3u);
for (int i = 0; i < 3; ++i)
uut->fwd_on_next(fwd_data, i);
ctx->run();
CHECK_EQ(snk->buf.size(), 0u);
snk->request(17);
ctx->run();
if (CHECK_EQ(snk->buf.size(), 1u))
CHECK_EQ(snk->buf[0], cow_vector<int>({0, 1, 2}));
}
}
}
}
SCENARIO("buffers never subscribe to their control observable on error") {
GIVEN("a buffer operator") {
WHEN("the data observable calls on_error on subscribing it") {
THEN("the buffer never tries to subscribe to their control observable") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto cnt = std::make_shared<size_t>(0);
auto uut = raw_sub(3,
ctx->make_observable().fail<int>(sec::runtime_error),
flow::make_nil_observable<int64_t>(ctx.get(), cnt),
snk->as_observer());
ctx->run();
CHECK(snk->aborted());
CHECK_EQ(*cnt, 0u);
}
}
}
}
SCENARIO("buffers dispose unexpected subscriptions") {
GIVEN("an initialized buffer operator") {
WHEN("calling on_subscribe with unexpected subscriptions") {
THEN("the buffer disposes them immediately") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, flow::make_nil_observable<int>(ctx.get()),
flow::make_nil_observable<int64_t>(ctx.get()),
snk->as_observer());
auto data_sub = make_counted<flow::passive_subscription_impl>();
auto ctrl_sub = make_counted<flow::passive_subscription_impl>();
uut->fwd_on_subscribe(fwd_data, flow::subscription{data_sub});
uut->fwd_on_subscribe(fwd_ctrl, flow::subscription{ctrl_sub});
ctx->run();
auto data_sub_2 = make_counted<flow::passive_subscription_impl>();
auto ctrl_sub_2 = make_counted<flow::passive_subscription_impl>();
uut->fwd_on_subscribe(fwd_data, flow::subscription{data_sub_2});
uut->fwd_on_subscribe(fwd_ctrl, flow::subscription{ctrl_sub_2});
ctx->run();
CHECK(!uut->disposed());
CHECK(!data_sub->disposed());
CHECK(!ctrl_sub->disposed());
CHECK(data_sub_2->disposed());
CHECK(ctrl_sub_2->disposed());
}
}
}
}
SCENARIO("buffers emit final items after an on_error event") {
GIVEN("an initialized buffer operator") {
WHEN("calling on_error(data) on a buffer without pending data") {
THEN("the buffer forward on_error immediately") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, trivial_obs<int>(), trivial_obs<int64_t>(),
snk->as_observer());
snk->request(42);
ctx->run();
uut->fwd_on_next(fwd_data, 1);
uut->fwd_on_next(fwd_data, 2);
uut->fwd_on_next(fwd_data, 3);
CHECK_EQ(uut->pending(), 0u);
uut->fwd_on_error(fwd_data, sec::runtime_error);
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({1, 2, 3})});
CHECK(snk->aborted());
}
}
WHEN("calling on_error(data) on a buffer with pending data") {
THEN("the buffer still emits pending data before closing") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, trivial_obs<int>(), trivial_obs<int64_t>(),
snk->as_observer());
ctx->run();
uut->fwd_on_next(fwd_data, 1);
uut->fwd_on_next(fwd_data, 2);
CHECK_EQ(uut->pending(), 2u);
uut->fwd_on_error(fwd_data, sec::runtime_error);
CHECK(snk->buf.empty());
CHECK(!snk->aborted());
snk->request(42);
ctx->run();
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({1, 2})});
CHECK(snk->aborted());
}
}
WHEN("calling on_error(control) on a buffer without pending data") {
THEN("the buffer forward on_error immediately") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, trivial_obs<int>(), trivial_obs<int64_t>(),
snk->as_observer());
snk->request(42);
ctx->run();
uut->fwd_on_next(fwd_data, 1);
uut->fwd_on_next(fwd_data, 2);
uut->fwd_on_next(fwd_data, 3);
CHECK_EQ(uut->pending(), 0u);
uut->fwd_on_error(fwd_ctrl, sec::runtime_error);
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({1, 2, 3})});
CHECK(snk->aborted());
}
}
WHEN("calling on_error(control) on a buffer with pending data") {
THEN("the buffer still emits pending data before closing") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, trivial_obs<int>(), trivial_obs<int64_t>(),
snk->as_observer());
ctx->run();
uut->fwd_on_next(fwd_data, 1);
uut->fwd_on_next(fwd_data, 2);
CHECK_EQ(uut->pending(), 2u);
uut->fwd_on_error(fwd_ctrl, sec::runtime_error);
CHECK(snk->buf.empty());
CHECK(!snk->aborted());
snk->request(42);
ctx->run();
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({1, 2})});
CHECK(snk->aborted());
}
}
}
}
SCENARIO("buffers emit final items after an on_complete event") {
GIVEN("an initialized buffer operator") {
WHEN("calling on_complete(data) on a buffer without pending data") {
THEN("the buffer forward on_complete immediately") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, trivial_obs<int>(), trivial_obs<int64_t>(),
snk->as_observer());
snk->request(42);
ctx->run();
uut->fwd_on_next(fwd_data, 1);
uut->fwd_on_next(fwd_data, 2);
uut->fwd_on_next(fwd_data, 3);
CHECK_EQ(uut->pending(), 0u);
uut->fwd_on_complete(fwd_data);
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({1, 2, 3})});
CHECK(snk->completed());
}
}
WHEN("calling on_complete(data) on a buffer with pending data") {
THEN("the buffer still emits pending data before closing") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, trivial_obs<int>(), trivial_obs<int64_t>(),
snk->as_observer());
ctx->run();
uut->fwd_on_next(fwd_data, 1);
uut->fwd_on_next(fwd_data, 2);
CHECK_EQ(uut->pending(), 2u);
uut->fwd_on_complete(fwd_data);
CHECK(snk->buf.empty());
CHECK(!snk->completed());
snk->request(42);
ctx->run();
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({1, 2})});
CHECK(snk->completed());
}
}
WHEN("calling on_complete(control) on a buffer without pending data") {
THEN("the buffer raises an error immediately") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, trivial_obs<int>(), trivial_obs<int64_t>(),
snk->as_observer());
snk->request(42);
ctx->run();
uut->fwd_on_next(fwd_data, 1);
uut->fwd_on_next(fwd_data, 2);
uut->fwd_on_next(fwd_data, 3);
CHECK_EQ(uut->pending(), 0u);
uut->fwd_on_complete(fwd_ctrl);
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({1, 2, 3})});
CHECK(snk->aborted());
}
}
WHEN("calling on_complete(control) on a buffer with pending data") {
THEN("the buffer raises an error after shipping pending items") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub(3, trivial_obs<int>(), trivial_obs<int64_t>(),
snk->as_observer());
ctx->run();
uut->fwd_on_next(fwd_data, 1);
uut->fwd_on_next(fwd_data, 2);
CHECK_EQ(uut->pending(), 2u);
uut->fwd_on_complete(fwd_ctrl);
CHECK(snk->buf.empty());
CHECK(!snk->completed());
snk->request(42);
ctx->run();
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({1, 2})});
CHECK(snk->aborted());
}
}
}
}
SCENARIO("skip policies suppress empty batches") {
GIVEN("a buffer operator") {
WHEN("the control observable fires with no pending data") {
THEN("the operator omits the batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
add_subs(uut);
snk->request(42);
ctx->run();
uut->fwd_on_next(fwd_ctrl, 1);
ctx->run();
CHECK(snk->buf.empty());
}
}
WHEN("the control observable fires with pending data") {
THEN("the operator emits a partial batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
add_subs(uut);
snk->request(42);
ctx->run();
uut->fwd_on_next(fwd_data, 17);
uut->fwd_on_next(fwd_ctrl, 1);
ctx->run();
CHECK_EQ(snk->buf, std::vector{cow_vector<int>{17}});
}
}
}
}
SCENARIO("no-skip policies emit empty batches") {
GIVEN("a buffer operator") {
WHEN("the control observable fires with no pending data") {
THEN("the operator emits an empty batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub<noskip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
add_subs(uut);
snk->request(42);
ctx->run();
uut->fwd_on_next(fwd_ctrl, 1);
ctx->run();
CHECK_EQ(snk->buf, std::vector{cow_vector<int>()});
}
}
WHEN("the control observable fires with pending data") {
THEN("the operator emits a partial batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub<noskip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
add_subs(uut);
snk->request(42);
ctx->run();
uut->fwd_on_next(fwd_data, 17);
uut->fwd_on_next(fwd_ctrl, 1);
ctx->run();
CHECK_EQ(snk->buf, std::vector{cow_vector<int>{17}});
}
}
}
}
SCENARIO("disposing a buffer operator completes the flow") {
GIVEN("a buffer operator") {
WHEN("disposing the subscription operator of the operator") {
THEN("the observer receives an on_complete event") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
add_subs(uut);
snk->request(42);
ctx->run();
uut->dispose();
ctx->run();
CHECK(snk->completed());
}
}
}
}
SCENARIO("on_request actions can turn into no-ops") {
GIVEN("a buffer operator") {
WHEN("the sink requests more data right before a timeout triggers") {
THEN("the batch gets shipped and the on_request action does nothing") {
auto snk = flow::make_passive_observer<cow_vector<int>>();
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(),
snk->as_observer());
add_subs(uut);
ctx->run();
// Add three items that we can't push yet because no downstream demand.
for (int i = 0; i < 3; ++i)
uut->fwd_on_next(fwd_data, i);
CHECK(uut->can_emit());
CHECK_EQ(uut->pending(), 3u);
// Add demand, which triggers an action - but don't run it yet.
snk->request(42);
CHECK_EQ(uut->pending(), 3u);
// Fire on_next on the control channel to force the batch out.
uut->fwd_on_next(fwd_ctrl, 1);
CHECK_EQ(uut->pending(), 0u);
// Run the scheduled action: turns into a no-op now.
ctx->run();
CHECK_EQ(snk->buf, std::vector{cow_vector<int>({0, 1, 2})});
}
}
}
}
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