Commit 35e60d41 authored by Dominik Charousset's avatar Dominik Charousset

Clean up potential UB and leaks in flow operators

parent 3c2f991e
...@@ -194,7 +194,7 @@ private: ...@@ -194,7 +194,7 @@ private:
value_sub_.dispose(); value_sub_.dispose();
control_sub_.dispose(); control_sub_.dispose();
switch (state_) { switch (state_) {
case state::running: case state::running: {
if (!buf_.empty()) { if (!buf_.empty()) {
if (demand_ == 0) { if (demand_ == 0) {
state_ = err_ ? state::aborted : state::completed; state_ = err_ ? state::aborted : state::completed;
...@@ -204,13 +204,14 @@ private: ...@@ -204,13 +204,14 @@ private:
out_.on_next(f(buf_)); out_.on_next(f(buf_));
buf_.clear(); buf_.clear();
} }
auto tmp = std::move(out_);
if (err_) if (err_)
out_.on_error(err_); tmp.on_error(err_);
else else
out_.on_complete(); tmp.on_complete();
out_ = nullptr;
state_ = state::disposed; state_ = state::disposed;
break; break;
}
default: default:
break; break;
} }
...@@ -226,11 +227,11 @@ private: ...@@ -226,11 +227,11 @@ private:
} }
if (!buf_.empty()) if (!buf_.empty())
do_emit(); do_emit();
auto tmp = std::move(out_);
if (err_) if (err_)
out_.on_error(err_); tmp.on_error(err_);
else else
out_.on_complete(); tmp.on_complete();
out_ = nullptr;
} }
void do_emit() { void do_emit() {
...@@ -249,8 +250,8 @@ private: ...@@ -249,8 +250,8 @@ private:
value_sub_.dispose(); value_sub_.dispose();
control_sub_.dispose(); control_sub_.dispose();
if (out_) { if (out_) {
out_.on_complete(); auto tmp = std::move(out_);
out_ = nullptr; tmp.on_complete();
} }
state_ = state::disposed; state_ = state::disposed;
} }
......
...@@ -16,19 +16,46 @@ ...@@ -16,19 +16,46 @@
namespace caf::flow::op { namespace caf::flow::op {
/// Interface for listening on a cell.
template <class T>
class cell_listener {
public:
friend void intrusive_ptr_add_ref(const cell_listener* ptr) noexcept {
ptr->ref_listener();
}
friend void intrusive_ptr_release(const cell_listener* ptr) noexcept {
ptr->deref_listener();
}
virtual void on_next(const T& item) = 0;
virtual void on_complete() = 0;
virtual void on_error(const error& what) = 0;
virtual void ref_listener() const noexcept = 0;
virtual void deref_listener() const noexcept = 0;
};
/// Convenience alias for a reference-counting smart pointer.
template <class T>
using cell_listener_ptr = intrusive_ptr<cell_listener<T>>;
/// State shared between one multicast operator and one subscribed observer. /// State shared between one multicast operator and one subscribed observer.
template <class T> template <class T>
struct cell_sub_state { struct cell_sub_state {
std::variant<none_t, unit_t, T, error> content; std::variant<none_t, unit_t, T, error> content;
std::vector<observer<T>> listeners; std::vector<cell_listener_ptr<T>> listeners;
void set_null() { void set_null() {
CAF_ASSERT(std::holds_alternative<none_t>(content)); CAF_ASSERT(std::holds_alternative<none_t>(content));
content = unit; content = unit;
std::vector<observer<T>> xs; std::vector<cell_listener_ptr<T>> xs;
xs.swap(listeners); xs.swap(listeners);
for (auto& listener : xs) { for (auto& listener : xs) {
listener.on_complete(); listener->on_complete();
} }
} }
...@@ -36,11 +63,11 @@ struct cell_sub_state { ...@@ -36,11 +63,11 @@ struct cell_sub_state {
CAF_ASSERT(std::holds_alternative<none_t>(content)); CAF_ASSERT(std::holds_alternative<none_t>(content));
content = std::move(item); content = std::move(item);
auto& ref = std::get<T>(content); auto& ref = std::get<T>(content);
std::vector<observer<T>> xs; std::vector<cell_listener_ptr<T>> xs;
xs.swap(listeners); xs.swap(listeners);
for (auto& listener : xs) { for (auto& listener : xs) {
listener.on_next(ref); listener->on_next(ref);
listener.on_complete(); listener->on_complete();
} }
} }
...@@ -48,41 +75,43 @@ struct cell_sub_state { ...@@ -48,41 +75,43 @@ struct cell_sub_state {
CAF_ASSERT(std::holds_alternative<none_t>(content)); CAF_ASSERT(std::holds_alternative<none_t>(content));
content = std::move(what); content = std::move(what);
auto& ref = std::get<error>(content); auto& ref = std::get<error>(content);
std::vector<observer<T>> xs; std::vector<cell_listener_ptr<T>> xs;
xs.swap(listeners); xs.swap(listeners);
for (auto& listener : xs) for (auto& listener : xs)
listener.on_error(ref); listener->on_error(ref);
} }
void listen(observer<T> listener) { void listen(cell_listener_ptr<T> listener) {
switch (content.index()) { switch (content.index()) {
case 1: case 1:
listener.on_complete(); listener->on_complete();
break; break;
case 2: case 2:
listener.on_next(std::get<T>(content)); listener->on_next(std::get<T>(content));
listener.on_complete(); listener->on_complete();
break; break;
case 3: case 3:
listener.on_error(std::get<error>(content)); listener->on_error(std::get<error>(content));
break; break;
default: default:
listeners.emplace_back(std::move(listener)); listeners.emplace_back(std::move(listener));
} }
} }
void drop(const observer<T>& listener) { void drop(const cell_listener_ptr<T>& listener) {
if (auto i = std::find(listeners.begin(), listeners.end(), listener); if (auto i = std::find(listeners.begin(), listeners.end(), listener);
i != listeners.end()) i != listeners.end())
listeners.erase(i); listeners.erase(i);
} }
}; };
/// Convenience alias for the state of a cell.
template <class T> template <class T>
using cell_sub_state_ptr = std::shared_ptr<cell_sub_state<T>>; using cell_sub_state_ptr = std::shared_ptr<cell_sub_state<T>>;
/// The subscription object for interfacing an observer with the cell state.
template <class T> template <class T>
class cell_sub : public subscription::impl_base { class cell_sub : public subscription::impl_base, public cell_listener<T> {
public: public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -91,6 +120,16 @@ public: ...@@ -91,6 +120,16 @@ public:
// nop // nop
} }
// -- disambiguation ---------------------------------------------------------
friend void intrusive_ptr_add_ref(const cell_sub* ptr) noexcept {
ptr->ref_disposable();
}
friend void intrusive_ptr_release(const cell_sub* ptr) noexcept {
ptr->deref_disposable();
}
// -- implementation of subscription ----------------------------------------- // -- implementation of subscription -----------------------------------------
bool disposed() const noexcept override { bool disposed() const noexcept override {
...@@ -99,21 +138,56 @@ public: ...@@ -99,21 +138,56 @@ public:
void dispose() override { void dispose() override {
if (state_) { if (state_) {
state_->drop(out_); state_->drop(this);
state_ = nullptr; state_ = nullptr;
out_ = nullptr; }
if (out_) {
auto tmp = std::move(out_);
tmp.on_complete();
} }
} }
void request(size_t) override { void request(size_t) override {
if (!listening_) { if (!listening_) {
listening_ = true; listening_ = true;
ctx_->delay_fn([state = state_, out = out_] { // auto self = cell_listener_ptr<T>{this};
state->listen(std::move(out)); ctx_->delay_fn([state = state_, self]() mutable { //
state->listen(std::move(self));
}); });
} }
} }
// -- implementation of listener<T> ------------------------------------------
void on_next(const T& item) override {
if (out_)
out_.on_next(item);
}
void on_complete() override {
state_ = nullptr;
if (out_) {
auto tmp = std::move(out_);
tmp.on_complete();
}
}
void on_error(const error& what) override {
state_ = nullptr;
if (out_) {
auto tmp = std::move(out_);
tmp.on_error(what);
}
}
void ref_listener() const noexcept override {
ref_disposable();
}
void deref_listener() const noexcept override {
deref_disposable();
}
private: private:
coordinator* ctx_; coordinator* ctx_;
bool listening_ = false; bool listening_ = false;
......
...@@ -147,20 +147,20 @@ private: ...@@ -147,20 +147,20 @@ private:
void fin(const error* err = nullptr) { void fin(const error* err = nullptr) {
CAF_ASSERT(out_); CAF_ASSERT(out_);
if (factory_sub_) { if (factory_sub_) {
factory_sub_.dispose(); auto tmp = std::move(factory_sub_);
factory_sub_ = nullptr; tmp.dispose();
} }
if (active_sub_) { if (active_sub_) {
active_sub_.dispose(); auto tmp = std::move(active_sub_);
active_sub_ = nullptr; tmp.dispose();
} }
factory_key_ = 0; factory_key_ = 0;
active_key_ = 0; active_key_ = 0;
auto tmp = std::move(out_);
if (err) if (err)
out_.on_error(*err); tmp.on_error(*err);
else else
out_.on_complete(); tmp.on_complete();
out_ = nullptr;
} }
/// Stores the context (coordinator) that runs this flow. /// Stores the context (coordinator) that runs this flow.
......
...@@ -97,11 +97,11 @@ private: ...@@ -97,11 +97,11 @@ private:
} }
void fin() { void fin() {
auto tmp = std::move(out_);
if (!err_) if (!err_)
out_.on_complete(); tmp.on_complete();
else else
out_.on_error(err_); tmp.on_error(err_);
out_ = nullptr;
} }
void pull(size_t n) { void pull(size_t n) {
......
...@@ -114,12 +114,12 @@ private: ...@@ -114,12 +114,12 @@ private:
void do_dispose() { void do_dispose() {
if (buf_) { if (buf_) {
buf_->cancel(); auto tmp = std::move(buf_);
buf_ = nullptr; tmp->cancel();
} }
if (out_) { if (out_) {
out_.on_complete(); auto tmp = std::move(out_);
out_ = nullptr; tmp.on_complete();
} }
} }
......
...@@ -41,15 +41,16 @@ public: ...@@ -41,15 +41,16 @@ public:
} }
void on_complete() { void on_complete() {
CAF_ASSERT(sub->in_.valid()); if (sub->in_) {
sub->in_.dispose(); auto tmp = std::move(sub->in_);
sub->in_ = nullptr; tmp.dispose();
}
} }
void on_error(const error& what) { void on_error(const error& what) {
if (sub->in_) { if (sub->in_) {
sub->in_.dispose(); auto tmp = std::move(sub->in_);
sub->in_ = nullptr; tmp.dispose();
} }
sub->err_ = what; sub->err_ = what;
} }
...@@ -187,8 +188,8 @@ public: ...@@ -187,8 +188,8 @@ public:
ctx_->delay_fn([out = std::move(out_)]() mutable { out.on_complete(); }); ctx_->delay_fn([out = std::move(out_)]() mutable { out.on_complete(); });
} }
if (in_) { if (in_) {
in_.dispose(); auto tmp = std::move(in_);
in_ = nullptr; tmp.dispose();
} }
} }
...@@ -235,12 +236,12 @@ private: ...@@ -235,12 +236,12 @@ private:
if (in_) { if (in_) {
pull(); pull();
} else if (buf_.empty()) { } else if (buf_.empty()) {
disposed_ = true;
auto tmp = std::move(out_);
if (err_) if (err_)
out_.on_error(err_); tmp.on_error(err_);
else else
out_.on_complete(); tmp.on_complete();
out_ = nullptr;
disposed_ = true;
} }
} }
} }
......
...@@ -101,8 +101,8 @@ public: ...@@ -101,8 +101,8 @@ public:
while (i != inputs_.end()) { while (i != inputs_.end()) {
auto& input = *i->second; auto& input = *i->second;
if (auto& sub = input.sub) { if (auto& sub = input.sub) {
sub.dispose(); auto tmp = std::move(input.sub);
sub = nullptr; tmp.dispose();
} }
if (input.buf.empty()) if (input.buf.empty())
i = inputs_.erase(i); i = inputs_.erase(i);
...@@ -215,12 +215,11 @@ private: ...@@ -215,12 +215,11 @@ private:
} }
} }
if (out_ && inputs_.empty()) { if (out_ && inputs_.empty()) {
if (!err_) { auto tmp = std::move(out_);
out_.on_complete(); if (!err_)
} else { tmp.on_complete();
out_.on_error(err_); else
} tmp.on_error(err_);
out_ = nullptr;
} }
flags_.running = false; flags_.running = false;
} }
......
...@@ -70,8 +70,8 @@ public: ...@@ -70,8 +70,8 @@ public:
sink_->abort(reason); sink_->abort(reason);
sub_ = nullptr; sub_ = nullptr;
} else if (out_) { } else if (out_) {
out_.on_error(reason); auto tmp = std::move(out_);
out_ = nullptr; tmp.on_error(reason);
} }
} }
...@@ -81,8 +81,8 @@ public: ...@@ -81,8 +81,8 @@ public:
sink_->close(); sink_->close();
sub_ = nullptr; sub_ = nullptr;
} else if (out_) { } else if (out_) {
out_.on_complete(); auto tmp = std::move(out_);
out_ = nullptr; tmp.on_complete();
} }
} }
...@@ -102,8 +102,8 @@ public: ...@@ -102,8 +102,8 @@ public:
if (out_) { if (out_) {
out_ = nullptr; out_ = nullptr;
if (sub_) { if (sub_) {
sub_.dispose(); auto tmp = std::move(sub_);
sub_ = nullptr; tmp.dispose();
} }
} }
} }
...@@ -157,8 +157,8 @@ private: ...@@ -157,8 +157,8 @@ private:
void on_sink_dispose() { void on_sink_dispose() {
sink_ = nullptr; sink_ = nullptr;
if (sub_) { if (sub_) {
sub_.dispose(); auto tmp = std::move(sub_);
sub_ = nullptr; tmp.dispose();
} }
} }
......
...@@ -123,9 +123,9 @@ private: ...@@ -123,9 +123,9 @@ private:
void on_dispose(state_type&) override { void on_dispose(state_type&) override {
try_request_more(); try_request_more();
if (auto_disconnect_ && connected_ && super::observer_count() == 0) { if (auto_disconnect_ && connected_ && super::observer_count() == 0) {
in_.dispose();
in_ = nullptr;
connected_ = false; connected_ = false;
auto tmp = std::move(in_);
tmp.dispose();
} }
} }
......
...@@ -71,13 +71,13 @@ public: ...@@ -71,13 +71,13 @@ public:
closed = true; closed = true;
if (!running && buf.empty()) { if (!running && buf.empty()) {
disposed = true; disposed = true;
if (out) {
out.on_complete();
out = nullptr;
}
when_disposed = nullptr; when_disposed = nullptr;
when_consumed_some = nullptr; when_consumed_some = nullptr;
when_demand_changed = nullptr; when_demand_changed = nullptr;
if (out) {
auto tmp = std::move(out);
tmp.on_complete();
}
} }
} }
} }
...@@ -88,33 +88,33 @@ public: ...@@ -88,33 +88,33 @@ public:
err = reason; err = reason;
if (!running && buf.empty()) { if (!running && buf.empty()) {
disposed = true; disposed = true;
if (out) {
auto out_hdl = std::move(out);
out_hdl.on_error(reason);
}
when_disposed = nullptr; when_disposed = nullptr;
when_consumed_some = nullptr; when_consumed_some = nullptr;
when_demand_changed = nullptr; when_demand_changed = nullptr;
if (out) {
auto tmp = std::move(out);
tmp.on_error(reason);
}
} }
} }
} }
void dispose() { void dispose() {
if (out) {
out.on_complete();
out = nullptr;
}
if (when_disposed) { if (when_disposed) {
ctx->delay(std::move(when_disposed)); ctx->delay(std::move(when_disposed));
} }
if (when_consumed_some) { if (when_consumed_some) {
when_consumed_some.dispose(); auto tmp = std::move(when_consumed_some);
when_consumed_some = nullptr; tmp.dispose();
} }
when_demand_changed = nullptr; when_demand_changed = nullptr;
buf.clear(); buf.clear();
demand = 0; demand = 0;
disposed = true; disposed = true;
if (out) {
auto tmp = std::move(out);
tmp.on_complete();
}
} }
void do_run() { void do_run() {
...@@ -130,11 +130,11 @@ public: ...@@ -130,11 +130,11 @@ public:
--demand; --demand;
} }
if (buf.empty() && closed) { if (buf.empty() && closed) {
auto tmp = std::move(out);
if (err) if (err)
out.on_error(err); tmp.on_error(err);
else else
out.on_complete(); tmp.on_complete();
out = nullptr;
dispose(); dispose();
} else if (got_some && when_consumed_some) { } else if (got_some && when_consumed_some) {
ctx->delay(when_consumed_some); ctx->delay(when_consumed_some);
......
...@@ -81,8 +81,8 @@ public: ...@@ -81,8 +81,8 @@ public:
if (out_) { if (out_) {
for_each_input([](auto, auto& input) { for_each_input([](auto, auto& input) {
if (input.sub) { if (input.sub) {
input.sub.dispose(); auto tmp = std::move(input.sub);
input.sub = nullptr; tmp.dispose();
} }
input.buf.clear(); input.buf.clear();
}); });
......
...@@ -70,8 +70,9 @@ public: ...@@ -70,8 +70,9 @@ public:
void on_complete() override { void on_complete() override {
if (sub) { if (sub) {
sub.dispose(); subscription tmp;
sub = nullptr; tmp.swap(sub);
tmp.dispose();
} }
state = observer_state::completed; state = observer_state::completed;
} }
...@@ -158,6 +159,43 @@ public: ...@@ -158,6 +159,43 @@ public:
std::vector<T> buf; std::vector<T> buf;
}; };
template <class T>
struct unsubscribe_guard {
public:
explicit unsubscribe_guard(intrusive_ptr<T> ptr) : ptr_(std::move(ptr)) {
// nop
}
unsubscribe_guard(unsubscribe_guard&&) = default;
unsubscribe_guard& operator=(unsubscribe_guard&&) = delete;
unsubscribe_guard(const unsubscribe_guard&) = delete;
unsubscribe_guard& operator=(const unsubscribe_guard&) = delete;
~unsubscribe_guard() {
if (ptr_)
ptr_->unsubscribe();
}
private:
intrusive_ptr<T> ptr_;
};
template <class T>
auto make_unsubscribe_guard(intrusive_ptr<T> ptr) {
return unsubscribe_guard<T>{std::move(ptr)};
}
template <class T1, class T2, class... Ts>
auto make_unsubscribe_guard(intrusive_ptr<T1> ptr1, intrusive_ptr<T2> ptr2,
Ts... ptrs) {
return std::make_tuple(make_unsubscribe_guard(std::move(ptr1)),
make_unsubscribe_guard(ptr2),
make_unsubscribe_guard(std::move(ptrs))...);
}
template <class T> template <class T>
class canceling_observer : public flow::observer_impl_base<T> { class canceling_observer : public flow::observer_impl_base<T> {
public: public:
......
...@@ -49,6 +49,10 @@ struct noskip_trait { ...@@ -49,6 +49,10 @@ struct noskip_trait {
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();
~fixture() {
ctx->run();
}
// Similar to buffer::subscribe, but returns a buffer_sub pointer instead of // Similar to buffer::subscribe, but returns a buffer_sub pointer instead of
// type-erasing it into a disposable. // type-erasing it into a disposable.
template <class Trait = noskip_trait> template <class Trait = noskip_trait>
...@@ -208,6 +212,7 @@ SCENARIO("buffers start to emit items once subscribed") { ...@@ -208,6 +212,7 @@ SCENARIO("buffers start to emit items once subscribed") {
WHEN("the selector never calls on_subscribe") { WHEN("the selector never calls on_subscribe") {
THEN("the buffer still emits batches") { THEN("the buffer still emits batches") {
auto snk = flow::make_passive_observer<cow_vector<int>>(); auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub(3, flow::make_nil_observable<int>(ctx.get()), auto uut = raw_sub(3, flow::make_nil_observable<int>(ctx.get()),
flow::make_nil_observable<int64_t>(ctx.get()), flow::make_nil_observable<int64_t>(ctx.get()),
snk->as_observer()); snk->as_observer());
...@@ -251,6 +256,7 @@ SCENARIO("buffers dispose unexpected subscriptions") { ...@@ -251,6 +256,7 @@ SCENARIO("buffers dispose unexpected subscriptions") {
WHEN("calling on_subscribe with unexpected subscriptions") { WHEN("calling on_subscribe with unexpected subscriptions") {
THEN("the buffer disposes them immediately") { THEN("the buffer disposes them immediately") {
auto snk = flow::make_passive_observer<cow_vector<int>>(); auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub(3, flow::make_nil_observable<int>(ctx.get()), auto uut = raw_sub(3, flow::make_nil_observable<int>(ctx.get()),
flow::make_nil_observable<int64_t>(ctx.get()), flow::make_nil_observable<int64_t>(ctx.get()),
snk->as_observer()); snk->as_observer());
...@@ -425,6 +431,7 @@ SCENARIO("skip policies suppress empty batches") { ...@@ -425,6 +431,7 @@ SCENARIO("skip policies suppress empty batches") {
WHEN("the control observable fires with no pending data") { WHEN("the control observable fires with no pending data") {
THEN("the operator omits the batch") { THEN("the operator omits the batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>(); auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(), auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(), trivial_obs<int64_t>(),
snk->as_observer()); snk->as_observer());
...@@ -439,6 +446,7 @@ SCENARIO("skip policies suppress empty batches") { ...@@ -439,6 +446,7 @@ SCENARIO("skip policies suppress empty batches") {
WHEN("the control observable fires with pending data") { WHEN("the control observable fires with pending data") {
THEN("the operator emits a partial batch") { THEN("the operator emits a partial batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>(); auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(), auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(), trivial_obs<int64_t>(),
snk->as_observer()); snk->as_observer());
...@@ -459,6 +467,7 @@ SCENARIO("no-skip policies emit empty batches") { ...@@ -459,6 +467,7 @@ SCENARIO("no-skip policies emit empty batches") {
WHEN("the control observable fires with no pending data") { WHEN("the control observable fires with no pending data") {
THEN("the operator emits an empty batch") { THEN("the operator emits an empty batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>(); auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<noskip_trait>(3, trivial_obs<int>(), auto uut = raw_sub<noskip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(), trivial_obs<int64_t>(),
snk->as_observer()); snk->as_observer());
...@@ -473,6 +482,7 @@ SCENARIO("no-skip policies emit empty batches") { ...@@ -473,6 +482,7 @@ SCENARIO("no-skip policies emit empty batches") {
WHEN("the control observable fires with pending data") { WHEN("the control observable fires with pending data") {
THEN("the operator emits a partial batch") { THEN("the operator emits a partial batch") {
auto snk = flow::make_passive_observer<cow_vector<int>>(); auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<noskip_trait>(3, trivial_obs<int>(), auto uut = raw_sub<noskip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(), trivial_obs<int64_t>(),
snk->as_observer()); snk->as_observer());
...@@ -512,6 +522,7 @@ SCENARIO("on_request actions can turn into no-ops") { ...@@ -512,6 +522,7 @@ SCENARIO("on_request actions can turn into no-ops") {
WHEN("the sink requests more data right before a timeout triggers") { WHEN("the sink requests more data right before a timeout triggers") {
THEN("the batch gets shipped and the on_request action does nothing") { THEN("the batch gets shipped and the on_request action does nothing") {
auto snk = flow::make_passive_observer<cow_vector<int>>(); auto snk = flow::make_passive_observer<cow_vector<int>>();
auto grd = make_unsubscribe_guard(snk);
auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(), auto uut = raw_sub<skip_trait>(3, trivial_obs<int>(),
trivial_obs<int64_t>(), trivial_obs<int64_t>(),
snk->as_observer()); snk->as_observer());
......
...@@ -115,6 +115,8 @@ SCENARIO("mcast operators buffer items that they cannot ship immediately") { ...@@ -115,6 +115,8 @@ SCENARIO("mcast operators buffer items that they cannot ship immediately") {
CHECK_EQ(uut->min_demand(), 0u); CHECK_EQ(uut->min_demand(), 0u);
CHECK_EQ(uut->max_buffered(), 3u); CHECK_EQ(uut->max_buffered(), 3u);
CHECK_EQ(uut->min_buffered(), 1u); CHECK_EQ(uut->min_buffered(), 1u);
sub2.dispose();
sub3.dispose();
} }
} }
} }
......
...@@ -68,6 +68,7 @@ SCENARIO("ucast operators may only be subscribed to once") { ...@@ -68,6 +68,7 @@ SCENARIO("ucast operators may only be subscribed to once") {
auto uut = make_ucast(); auto uut = make_ucast();
auto o1 = flow::make_passive_observer<int>(); auto o1 = flow::make_passive_observer<int>();
auto o2 = flow::make_passive_observer<int>(); auto o2 = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(o1, o2);
auto sub1 = uut->subscribe(o1->as_observer()); auto sub1 = uut->subscribe(o1->as_observer());
auto sub2 = uut->subscribe(o2->as_observer()); auto sub2 = uut->subscribe(o2->as_observer());
CHECK(o1->subscribed()); CHECK(o1->subscribed());
......
...@@ -37,6 +37,7 @@ SCENARIO("zip_with combines inputs") { ...@@ -37,6 +37,7 @@ SCENARIO("zip_with combines inputs") {
WHEN("merging them with zip_with") { WHEN("merging them with zip_with") {
THEN("the observer receives the combined output of both sources") { THEN("the observer receives the combined output of both sources") {
auto snk = flow::make_passive_observer<int>(); auto snk = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk);
ctx->make_observable() ctx->make_observable()
.zip_with([](int x, int y) { return x + y; }, .zip_with([](int x, int y) { return x + y; },
ctx->make_observable().repeat(11).take(113), ctx->make_observable().repeat(11).take(113),
...@@ -200,6 +201,7 @@ SCENARIO("observers may request from zip_with operators before on_subscribe") { ...@@ -200,6 +201,7 @@ SCENARIO("observers may request from zip_with operators before on_subscribe") {
THEN("the observer receives the item") { THEN("the observer receives the item") {
using flow::op::zip_index; using flow::op::zip_index;
auto snk = flow::make_passive_observer<int>(); auto snk = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk);
auto uut = make_zip_with_sub([](int, int) { return 0; }, auto uut = make_zip_with_sub([](int, int) { return 0; },
snk->as_observer(), snk->as_observer(),
flow::make_nil_observable<int>(ctx.get()), flow::make_nil_observable<int>(ctx.get()),
...@@ -222,6 +224,7 @@ SCENARIO("the zip_with operators disposes unexpected subscriptions") { ...@@ -222,6 +224,7 @@ SCENARIO("the zip_with operators disposes unexpected subscriptions") {
THEN("the operator disposes the subscription") { THEN("the operator disposes the subscription") {
using flow::op::zip_index; using flow::op::zip_index;
auto snk = flow::make_passive_observer<int>(); auto snk = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk);
auto uut = make_zip_with_sub([](int, int) { return 0; }, auto uut = make_zip_with_sub([](int, int) { return 0; },
snk->as_observer(), snk->as_observer(),
flow::make_nil_observable<int>(ctx.get()), flow::make_nil_observable<int>(ctx.get()),
......
...@@ -21,6 +21,10 @@ namespace { ...@@ -21,6 +21,10 @@ 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();
~fixture() {
ctx->run();
}
template <class... Ts> template <class... Ts>
std::vector<int> ls(Ts... xs) { std::vector<int> ls(Ts... xs) {
return std::vector<int>{xs...}; return std::vector<int>{xs...};
...@@ -143,13 +147,15 @@ SCENARIO("connectable observables forward errors") { ...@@ -143,13 +147,15 @@ SCENARIO("connectable observables forward errors") {
auto conn = flow::observable<int>{cell}.share(); auto conn = flow::observable<int>{cell}.share();
cell->set_error(sec::runtime_error); cell->set_error(sec::runtime_error);
// First subscriber to trigger subscription to the cell. // First subscriber to trigger subscription to the cell.
conn.subscribe(flow::make_auto_observer<int>()->as_observer()); auto snk1 = flow::make_auto_observer<int>();
conn.subscribe(snk1->as_observer());
ctx->run(); ctx->run();
CHECK(snk1->aborted());
// After this point, new subscribers should be aborted right away. // After this point, new subscribers should be aborted right away.
auto snk = flow::make_auto_observer<int>(); auto snk2 = flow::make_auto_observer<int>();
auto sub = conn.subscribe(snk->as_observer()); auto sub = conn.subscribe(snk2->as_observer());
CHECK(sub.disposed()); CHECK(sub.disposed());
CHECK(snk->aborted()); CHECK(snk2->aborted());
ctx->run(); ctx->run();
} }
} }
...@@ -163,6 +169,7 @@ SCENARIO("observers that dispose their subscription do not affect others") { ...@@ -163,6 +169,7 @@ SCENARIO("observers that dispose their subscription do not affect others") {
using impl_t = flow::op::publish<int>; using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>(); auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>(); auto snk2 = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk1, snk2);
auto iota = ctx->make_observable().iota(1).take(12).as_observable(); auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl(), 5); auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl(), 5);
auto sub1 = uut->subscribe(snk1->as_observer()); auto sub1 = uut->subscribe(snk1->as_observer());
...@@ -191,6 +198,7 @@ SCENARIO("publishers with auto_disconnect auto-dispose their subscription") { ...@@ -191,6 +198,7 @@ SCENARIO("publishers with auto_disconnect auto-dispose their subscription") {
using impl_t = flow::op::publish<int>; using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>(); auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>(); auto snk2 = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk1, snk2);
auto iota = ctx->make_observable().iota(1).take(12).as_observable(); auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl(), 5); auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl(), 5);
auto sub1 = uut->subscribe(snk1->as_observer()); auto sub1 = uut->subscribe(snk1->as_observer());
...@@ -218,10 +226,11 @@ SCENARIO("publishers dispose unexpected subscriptions") { ...@@ -218,10 +226,11 @@ SCENARIO("publishers dispose unexpected subscriptions") {
WHEN("calling on_subscribe with unexpected subscriptions") { WHEN("calling on_subscribe with unexpected subscriptions") {
THEN("the operator disposes them immediately") { THEN("the operator disposes them immediately") {
using impl_t = flow::op::publish<int>; using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>(); auto snk = flow::make_passive_observer<int>();
auto grd = make_unsubscribe_guard(snk);
auto iota = ctx->make_observable().iota(1).take(12).as_observable(); auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl()); auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl());
uut->subscribe(snk1->as_observer()); uut->subscribe(snk->as_observer());
uut->connect(); uut->connect();
auto sub = flow::make_passive_subscription(); auto sub = flow::make_passive_subscription();
uut->on_subscribe(flow::subscription{sub}); uut->on_subscribe(flow::subscription{sub});
......
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