Commit a1f4a297 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/coverage'

parents d8dfff00 a334195e
...@@ -24,6 +24,12 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -24,6 +24,12 @@ is based on [Keep a Changelog](https://keepachangelog.com).
their observer. their observer.
- The `fan_out_request` request now properly deals with actor handles that - The `fan_out_request` request now properly deals with actor handles that
respond with `void` (#1369). respond with `void` (#1369).
- Fix subscription and event handling in flow buffer operator.
- Fix undefined behavior in getter functions of the flow `mcast` operator.
- Add checks to avoid potential UB when using `prefix_and_tail` or other
operators that use the `ucast` operator internally.
- The `mcast` and `ucast` operators now stop calling `on_next` immediately when
disposed.
## [0.19.0-rc.1] - 2022-10-31 ## [0.19.0-rc.1] - 2022-10-31
......
...@@ -56,8 +56,7 @@ caf_add_component( ...@@ -56,8 +56,7 @@ caf_add_component(
async.read_result async.read_result
async.write_result async.write_result
exit_reason exit_reason
flow.observable_state flow.op.state
flow.observer_state
intrusive.inbox_result intrusive.inbox_result
intrusive.task_result intrusive.task_result
invoke_message_result invoke_message_result
...@@ -279,25 +278,28 @@ caf_add_component( ...@@ -279,25 +278,28 @@ caf_add_component(
dynamic_spawn dynamic_spawn
error error
expected expected
flow.buffer
flow.concat
flow.concat_map flow.concat_map
flow.defer
flow.empty
flow.fail
flow.flat_map flow.flat_map
flow.for_each flow.for_each
flow.generation flow.generation
flow.interval
flow.item_publisher flow.item_publisher
flow.merge
flow.mixed flow.mixed
flow.never
flow.observe_on flow.observe_on
flow.prefix_and_tail flow.op.buffer
flow.op.cell
flow.op.concat
flow.op.defer
flow.op.empty
flow.op.fail
flow.op.interval
flow.op.mcast
flow.op.merge
flow.op.never
flow.op.prefix_and_tail
flow.op.ucast
flow.op.zip_with
flow.publish flow.publish
flow.single flow.single
flow.zip_with
function_view function_view
handles handles
hash.fnv hash.fnv
......
...@@ -145,6 +145,18 @@ public: ...@@ -145,6 +145,18 @@ public:
int compare(uint8_t code, type_id_t category) const noexcept; 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 -------------------------------------------------------------- // -- modifiers --------------------------------------------------------------
/// Reverts this error to "not an error" as if calling `*this = error{}`. /// Reverts this error to "not an error" as if calling `*this = error{}`.
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/flow/observable.hpp"
#include "caf/flow/observer.hpp"
namespace caf::flow {} // namespace caf::flow
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/flow/observable.hpp"
#include "caf/flow/observer.hpp"
namespace caf::flow {
/// Combines the items emitted from `pub` and `pubs...` to appear as a single
/// stream of items.
} // namespace caf::flow
...@@ -16,7 +16,6 @@ ...@@ -16,7 +16,6 @@
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/fwd.hpp" #include "caf/flow/fwd.hpp"
#include "caf/flow/observable_decl.hpp" #include "caf/flow/observable_decl.hpp"
#include "caf/flow/observable_state.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/op/base.hpp" #include "caf/flow/op/base.hpp"
#include "caf/flow/op/buffer.hpp" #include "caf/flow/op/buffer.hpp"
...@@ -28,6 +27,7 @@ ...@@ -28,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"
...@@ -331,6 +331,13 @@ public: ...@@ -331,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();
...@@ -484,12 +491,14 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) { ...@@ -484,12 +491,14 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) {
template <class T> template <class T>
disposable observable<T>::subscribe(ignore_t) { disposable observable<T>::subscribe(ignore_t) {
return subscribe(observer<T>::ignore()); return for_each([](const T&) {});
} }
template <class T> template <class T>
template <class OnNext> template <class OnNext>
disposable observable<T>::for_each(OnNext on_next) { disposable observable<T>::for_each(OnNext on_next) {
static_assert(std::is_invocable_v<OnNext, const T&>,
"for_each: the on_next function must accept a 'const T&'");
return subscribe(make_observer(std::move(on_next))); return subscribe(make_observer(std::move(on_next)));
} }
...@@ -682,6 +691,18 @@ auto observable<T>::concat_map(F f) { ...@@ -682,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
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::flow {
/// Represents the current state of an @ref observable.
enum class observable_state {
/// Indicates that the observable is still waiting on some event or input.
idle,
/// Indicates that at least one observer subscribed.
running,
/// Indicates that the observable is waiting for observers to consume all
/// produced items before shutting down.
completing,
/// Indicates that the observable properly shut down.
completed,
/// Indicates that the observable shut down due to an error.
aborted,
/// Indicates that dispose was called.
disposed,
};
/// Returns whether `x` represents a final state, i.e., `completed`, `aborted`
/// or `disposed`.
constexpr bool is_final(observable_state x) noexcept {
return static_cast<int>(x) >= static_cast<int>(observable_state::completed);
}
/// Returns whether `x` represents an active state, i.e., `idle` or `running`.
constexpr bool is_active(observable_state x) noexcept {
return static_cast<int>(x) <= static_cast<int>(observable_state::running);
}
/// @relates observable_state
CAF_CORE_EXPORT std::string to_string(observable_state);
/// @relates observable_state
CAF_CORE_EXPORT bool from_string(std::string_view, observable_state&);
/// @relates observable_state
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<observable_state>,
observable_state&);
/// @relates observable_state
template <class Inspector>
bool inspect(Inspector& f, observable_state& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::flow
...@@ -14,7 +14,6 @@ ...@@ -14,7 +14,6 @@
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/flow/coordinated.hpp" #include "caf/flow/coordinated.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/observer_state.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
...@@ -109,12 +108,6 @@ public: ...@@ -109,12 +108,6 @@ public:
return pimpl_.compare(other.pimpl_); return pimpl_.compare(other.pimpl_);
} }
/// Returns an observer that ignores any of its inputs.
static observer ignore();
/// Returns an observer that disposes its subscription immediately.
static observer cancel();
private: private:
intrusive_ptr<impl> pimpl_; intrusive_ptr<impl> pimpl_;
}; };
...@@ -144,55 +137,6 @@ public: ...@@ -144,55 +137,6 @@ public:
namespace caf::detail { namespace caf::detail {
template <class T>
class ignoring_observer : public flow::observer_impl_base<T> {
public:
void on_next(const T&) override {
if (sub_)
sub_.request(1);
}
void on_error(const error&) override {
sub_ = nullptr;
}
void on_complete() override {
sub_ = nullptr;
}
void on_subscribe(flow::subscription sub) override {
if (!sub_) {
sub_ = std::move(sub);
sub_.request(defaults::flow::buffer_size);
} else {
sub.dispose();
}
}
private:
flow::subscription sub_;
};
template <class T>
class canceling_observer : public flow::observer_impl_base<T> {
public:
void on_next(const T&) override {
// nop
}
void on_error(const error&) override {
// nop
}
void on_complete() override {
// nop
}
void on_subscribe(flow::subscription sub) override {
sub.dispose();
}
};
template <class OnNextSignature> template <class OnNextSignature>
struct on_next_trait; struct on_next_trait;
...@@ -280,16 +224,6 @@ private: ...@@ -280,16 +224,6 @@ private:
namespace caf::flow { namespace caf::flow {
template <class T>
observer<T> observer<T>::ignore() {
return observer<T>{make_counted<detail::ignoring_observer<T>>()};
}
template <class T>
observer<T> observer<T>::cancel() {
return observer<T>{make_counted<detail::canceling_observer<T>>()};
}
/// Creates an observer from given callbacks. /// Creates an observer from given callbacks.
/// @param on_next Callback for handling incoming elements. /// @param on_next Callback for handling incoming elements.
/// @param on_error Callback for handling an error. /// @param on_error Callback for handling an error.
...@@ -522,126 +456,4 @@ private: ...@@ -522,126 +456,4 @@ private:
Token token_; 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;
}
}
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 } // namespace caf::flow
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::flow {
/// Represents the current state of an @ref observer.
enum class observer_state {
/// Indicates that no callbacks were called yet.
idle,
/// Indicates that on_subscribe was called.
subscribed,
/// Indicates that on_complete was called.
completed,
/// Indicates that on_error was called.
aborted
};
/// Returns whether `x` represents a final state, i.e., `completed`, `aborted`
/// or `disposed`.
constexpr bool is_final(observer_state x) noexcept {
return static_cast<int>(x) >= static_cast<int>(observer_state::completed);
}
/// Returns whether `x` represents an active state, i.e., `idle` or
/// `subscribed`.
constexpr bool is_active(observer_state x) noexcept {
return static_cast<int>(x) <= static_cast<int>(observer_state::subscribed);
}
/// @relates sec
CAF_CORE_EXPORT std::string to_string(observer_state);
/// @relates observer_state
CAF_CORE_EXPORT bool from_string(std::string_view, observer_state&);
/// @relates observer_state
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<observer_state>,
observer_state&);
/// @relates observer_state
template <class Inspector>
bool inspect(Inspector& f, observer_state& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::flow
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#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/state.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
...@@ -68,6 +69,24 @@ public: ...@@ -68,6 +69,24 @@ public:
// nop // 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 ----------------------------------------------- // -- callbacks for the parent -----------------------------------------------
void init(observable<input_type> vals, observable<select_token_type> ctrl) { void init(observable<input_type> vals, observable<select_token_type> ctrl) {
...@@ -75,6 +94,9 @@ public: ...@@ -75,6 +94,9 @@ public:
using ctrl_fwd_t = forwarder<select_token_type, buffer_sub, buffer_emit_t>; using ctrl_fwd_t = forwarder<select_token_type, buffer_sub, buffer_emit_t>;
vals.subscribe( vals.subscribe(
make_counted<val_fwd_t>(this, buffer_input_t{})->as_observer()); make_counted<val_fwd_t>(this, buffer_input_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( ctrl.subscribe(
make_counted<ctrl_fwd_t>(this, buffer_emit_t{})->as_observer()); make_counted<ctrl_fwd_t>(this, buffer_emit_t{})->as_observer());
} }
...@@ -82,62 +104,53 @@ public: ...@@ -82,62 +104,53 @@ public:
// -- callbacks for the forwarders ------------------------------------------- // -- callbacks for the forwarders -------------------------------------------
void fwd_on_subscribe(buffer_input_t, subscription sub) { void fwd_on_subscribe(buffer_input_t, subscription sub) {
if (!value_sub_ && out_) { if (!running() || value_sub_ || !out_) {
value_sub_ = std::move(sub);
if (demand_ > 0) {
in_flight_ += max_buf_size_;
value_sub_.request(max_buf_size_);
}
} else {
sub.dispose(); sub.dispose();
return;
} }
value_sub_ = std::move(sub);
value_sub_.request(max_buf_size_);
} }
void fwd_on_complete(buffer_input_t) { void fwd_on_complete(buffer_input_t) {
if (!value_sub_.valid())
return;
CAF_ASSERT(out_.valid());
value_sub_ = nullptr; value_sub_ = nullptr;
if (!buf_.empty()) shutdown();
do_emit();
out_.on_complete();
out_ = nullptr;
if (control_sub_) {
control_sub_.dispose();
control_sub_ = nullptr;
}
} }
void fwd_on_error(buffer_input_t, const error& what) { void fwd_on_error(buffer_input_t, const error& what) {
value_sub_ = nullptr; value_sub_ = nullptr;
do_abort(what); err_ = what;
shutdown();
} }
void fwd_on_next(buffer_input_t, const input_type& item) { void fwd_on_next(buffer_input_t, const input_type& item) {
CAF_ASSERT(in_flight_ > 0); if (running()) {
--in_flight_;
buf_.push_back(item); buf_.push_back(item);
if (buf_.size() == max_buf_size_) if (buf_.size() == max_buf_size_)
do_emit(); do_emit();
} }
}
void fwd_on_subscribe(buffer_emit_t, subscription sub) { void fwd_on_subscribe(buffer_emit_t, subscription sub) {
if (!control_sub_ && out_) { if (!running() || control_sub_ || !out_) {
control_sub_ = std::move(sub);
control_sub_.request(1);
} else {
sub.dispose(); sub.dispose();
return;
} }
control_sub_ = std::move(sub);
control_sub_.request(1);
} }
void fwd_on_complete(buffer_emit_t) { void fwd_on_complete(buffer_emit_t) {
do_abort(make_error(sec::end_of_stream, control_sub_ = nullptr;
"buffer: unexpected end of the control stream")); err_ = make_error(sec::end_of_stream,
"buffer: unexpected end of the control stream");
shutdown();
} }
void fwd_on_error(buffer_emit_t, const error& what) { void fwd_on_error(buffer_emit_t, const error& what) {
control_sub_ = nullptr; control_sub_ = nullptr;
do_abort(what); err_ = what;
shutdown();
} }
void fwd_on_next(buffer_emit_t, select_token_type) { void fwd_on_next(buffer_emit_t, select_token_type) {
...@@ -167,54 +180,78 @@ public: ...@@ -167,54 +180,78 @@ public:
void request(size_t n) override { void request(size_t n) override {
CAF_ASSERT(out_.valid()); CAF_ASSERT(out_.valid());
demand_ += n; demand_ += n;
if (value_sub_ && pending() == 0) { // If we can ship a batch, schedule an event to do so.
in_flight_ = max_buf_size_; if (demand_ == n && can_emit()) {
value_sub_.request(max_buf_size_); ctx_->delay_fn([strong_this = intrusive_ptr<buffer_sub>{this}] {
strong_this->on_request();
});
} }
} }
private: private:
size_t pending() const noexcept { void shutdown() {
return buf_.size() + in_flight_; value_sub_.dispose();
control_sub_.dispose();
switch (state_) {
case state::running:
if (!buf_.empty()) {
if (demand_ == 0) {
state_ = err_ ? state::aborted : state::completed;
return;
} }
void do_emit() {
Trait f; Trait f;
out_.on_next(f(buf_)); out_.on_next(f(buf_));
auto buffered = buf_.size();
buf_.clear(); buf_.clear();
if (value_sub_ && buffered > 0) {
in_flight_ += buffered;
value_sub_.request(buffered);
} }
if (err_)
out_.on_error(err_);
else
out_.on_complete();
out_ = nullptr;
state_ = state::disposed;
break;
default:
break;
} }
void do_dispose() {
if (value_sub_) {
value_sub_.dispose();
value_sub_ = nullptr;
} }
if (control_sub_) {
control_sub_.dispose(); void on_request() {
control_sub_ = nullptr; if (demand_ == 0 || !can_emit())
return;
if (running()) {
CAF_ASSERT(buf_.size() == max_buf_size_);
do_emit();
return;
} }
if (out_) { if (!buf_.empty())
do_emit();
if (err_)
out_.on_error(err_);
else
out_.on_complete(); out_.on_complete();
out_ = nullptr;
} }
void do_emit() {
if (demand_ == 0)
return;
Trait f;
--demand_;
auto buffered = buf_.size();
out_.on_next(f(buf_));
buf_.clear();
if (value_sub_ && buffered > 0)
value_sub_.request(buffered);
} }
void do_abort(const error& reason) { void do_dispose() {
if (value_sub_) {
value_sub_.dispose(); value_sub_.dispose();
value_sub_ = nullptr;
}
if (control_sub_) {
control_sub_.dispose(); control_sub_.dispose();
control_sub_ = nullptr;
}
if (out_) { if (out_) {
out_.on_error(reason); out_.on_complete();
out_ = nullptr;
} }
state_ = state::disposed;
} }
/// Stores the context (coordinator) that runs this flow. /// Stores the context (coordinator) that runs this flow.
...@@ -223,23 +260,32 @@ private: ...@@ -223,23 +260,32 @@ private:
/// Stores the maximum buffer size before forcing a batch. /// Stores the maximum buffer size before forcing a batch.
size_t max_buf_size_; size_t max_buf_size_;
/// Keeps track of how many items we have already requested.
size_t in_flight_ = 0;
/// Stores the elements until we can emit them. /// Stores the elements until we can emit them.
std::vector<input_type> buf_; std::vector<input_type> buf_;
/// Stores a handle to the subscribed observer. /// Stores a handle to the subscribed observer.
observer<output_type> out_; observer<output_type> out_;
/// Our subscription for the values. /// Our subscription for the values. We request `max_buf_size_` items and
/// whenever we emit a batch, we request whatever amount we have shipped. That
/// way, we should always have enough demand at the source to fill up a batch.
subscription value_sub_; subscription value_sub_;
/// Our subscription for the control tokens. /// Our subscription for the control tokens. We always request 1 item.
subscription control_sub_; subscription control_sub_;
/// Demand signaled by the observer. /// Demand signaled by the observer.
size_t demand_ = 0; size_t demand_ = 0;
/// Our current state.
/// - 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::running;
/// Caches the abort reason.
error err_;
}; };
template <class Trait> template <class Trait>
...@@ -272,6 +318,12 @@ public: ...@@ -272,6 +318,12 @@ public:
disposable subscribe(observer<output_type> out) override { disposable subscribe(observer<output_type> out) override {
auto ptr = make_counted<buffer_sub<Trait>>(super::ctx_, max_items_, out); auto ptr = make_counted<buffer_sub<Trait>>(super::ctx_, max_items_, out);
ptr->init(in_, select_); 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}); out.on_subscribe(subscription{ptr});
return ptr->as_disposable(); return ptr->as_disposable();
} }
......
...@@ -101,14 +101,7 @@ public: ...@@ -101,14 +101,7 @@ public:
void fwd_on_error(input_key key, const error& what) { void fwd_on_error(input_key key, const error& what) {
if (key == active_key_ || key == factory_key_) { if (key == active_key_ || key == factory_key_) {
CAF_ASSERT(out_); CAF_ASSERT(out_);
if (delay_error_) { fin(&what);
if (!err_)
err_ = what;
subscribe_next();
} else {
err_ = what;
fin();
}
} }
} }
...@@ -137,7 +130,6 @@ public: ...@@ -137,7 +130,6 @@ public:
if (out_) { if (out_) {
ctx_->delay_fn([strong_this = intrusive_ptr<concat_sub>{this}] { ctx_->delay_fn([strong_this = intrusive_ptr<concat_sub>{this}] {
if (strong_this->out_) { if (strong_this->out_) {
strong_this->err_.reset();
strong_this->fin(); strong_this->fin();
} }
}); });
...@@ -152,7 +144,7 @@ public: ...@@ -152,7 +144,7 @@ public:
} }
private: private:
void fin() { void fin(const error* err = nullptr) {
CAF_ASSERT(out_); CAF_ASSERT(out_);
if (factory_sub_) { if (factory_sub_) {
factory_sub_.dispose(); factory_sub_.dispose();
...@@ -164,8 +156,8 @@ private: ...@@ -164,8 +156,8 @@ private:
} }
factory_key_ = 0; factory_key_ = 0;
active_key_ = 0; active_key_ = 0;
if (err_) if (err)
out_.on_error(err_); out_.on_error(*err);
else else
out_.on_complete(); out_.on_complete();
out_ = nullptr; out_ = nullptr;
...@@ -177,12 +169,6 @@ private: ...@@ -177,12 +169,6 @@ private:
/// Stores a handle to the subscribed observer. /// Stores a handle to the subscribed observer.
observer<T> out_; observer<T> out_;
/// Configures whether we carry on after an error.
bool delay_error_ = false;
/// Caches an on_error reason if delay_error_ is true.
error err_;
/// Stores our input sources. The first input is active (subscribed to) while /// Stores our input sources. The first input is active (subscribed to) while
/// the others are pending (not subscribed to). /// the others are pending (not subscribed to).
std::vector<input_type> inputs_; std::vector<input_type> inputs_;
...@@ -245,14 +231,9 @@ private: ...@@ -245,14 +231,9 @@ private:
template <class Input> template <class Input>
void add(Input&& x) { void add(Input&& x) {
using input_t = std::decay_t<Input>; using input_t = std::decay_t<Input>;
if constexpr (detail::is_iterable_v<input_t>) {
for (auto& in : x)
add(in);
} else {
static_assert(is_observable_v<input_t>); static_assert(is_observable_v<input_t>);
inputs_.emplace_back(std::move(x).as_observable()); inputs_.emplace_back(std::move(x).as_observable());
} }
}
std::vector<input_type> inputs_; std::vector<input_type> inputs_;
}; };
......
...@@ -228,6 +228,9 @@ private: ...@@ -228,6 +228,9 @@ private:
buf_.pop_front(); buf_.pop_front();
--demand_; --demand_;
out_.on_next(item); out_.on_next(item);
// Note: on_next() may call dispose() and set out_ to nullptr.
if (!out_)
return;
} }
if (in_) { if (in_) {
pull(); pull();
......
...@@ -39,12 +39,13 @@ public: ...@@ -39,12 +39,13 @@ public:
// -- implementation of subscription ----------------------------------------- // -- implementation of subscription -----------------------------------------
bool disposed() const noexcept override { bool disposed() const noexcept override {
return !state_; return !state_ || state_->disposed;
} }
void dispose() override { void dispose() override {
if (state_) { if (state_) {
ctx_->delay_fn([state = std::move(state_)]() { state->do_dispose(); }); auto state = std::move(state_);
state->dispose();
} }
} }
...@@ -107,55 +108,52 @@ public: ...@@ -107,55 +108,52 @@ public:
for (auto& state : states_) for (auto& state : states_)
state->abort(reason); state->abort(reason);
states_.clear(); states_.clear();
err_ = reason;
} }
} }
size_t max_demand() const noexcept { size_t max_demand() const noexcept {
if (states_.empty()) { if (states_.empty()) {
return 0; return 0;
} else { }
auto pred = [](const state_ptr_type& x, const state_ptr_type& y) { auto pred = [](const state_ptr_type& x, const state_ptr_type& y) {
return x->demand < y->demand; return x->demand < y->demand;
}; };
auto& ptr = *std::max_element(states_.begin(), states_.end(), pred); auto& ptr = *std::max_element(states_.begin(), states_.end(), pred);
return ptr->demand; return ptr->demand;
} }
}
size_t min_demand() const noexcept { size_t min_demand() const noexcept {
if (states_.empty()) { if (states_.empty()) {
return 0; return 0;
} else { }
auto pred = [](const state_ptr_type& x, const state_ptr_type& y) { auto pred = [](const state_ptr_type& x, const state_ptr_type& y) {
return x->demand < y->demand; return x->demand < y->demand;
}; };
auto& ptr = *std::min_element(states_.begin(), states_.end(), pred); auto& ptr = *std::min_element(states_.begin(), states_.end(), pred);
ptr->demand; return ptr->demand;
}
} }
size_t max_buffered() const noexcept { size_t max_buffered() const noexcept {
if (states_.empty()) { if (states_.empty()) {
return 0; return 0;
} else { }
auto pred = [](const state_ptr_type& x, const state_ptr_type& y) { auto pred = [](const state_ptr_type& x, const state_ptr_type& y) {
return x->buf.size() < y->buf.size(); return x->buf.size() < y->buf.size();
}; };
auto& ptr = *std::max_element(states_.begin(), states_.end(), pred); auto& ptr = *std::max_element(states_.begin(), states_.end(), pred);
return ptr->buf.size(); return ptr->buf.size();
} }
}
size_t min_buffered() const noexcept { size_t min_buffered() const noexcept {
if (states_.empty()) { if (states_.empty()) {
return 0; return 0;
} else { }
auto pred = [](const state_ptr_type& x, const state_ptr_type& y) { auto pred = [](const state_ptr_type& x, const state_ptr_type& y) {
return x->buf.size() < y->buf.size(); return x->buf.size() < y->buf.size();
}; };
auto& ptr = *std::min_element(states_.begin(), states_.end(), pred); auto& ptr = *std::min_element(states_.begin(), states_.end(), pred);
ptr->buf.size(); return ptr->buf.size();
}
} }
/// Queries whether there is at least one observer subscribed to the operator. /// Queries whether there is at least one observer subscribed to the operator.
......
...@@ -214,18 +214,7 @@ private: ...@@ -214,18 +214,7 @@ private:
break; break;
} }
} }
if (out_ && inputs_.empty()) if (out_ && inputs_.empty()) {
fin();
flags_.running = false;
}
void fin() {
if (!inputs_.empty()) {
for (auto& kvp : inputs_)
if (auto& sub = kvp.second->sub)
sub.dispose();
inputs_.clear();
}
if (!err_) { if (!err_) {
out_.on_complete(); out_.on_complete();
} else { } else {
...@@ -233,6 +222,8 @@ private: ...@@ -233,6 +222,8 @@ private:
} }
out_ = nullptr; out_ = nullptr;
} }
flags_.running = false;
}
/// Selects an input object by key or returns null. /// Selects an input object by key or returns null.
input_t* get(input_key key) { input_t* get(input_key key) {
......
...@@ -121,7 +121,7 @@ public: ...@@ -121,7 +121,7 @@ public:
} }
void request(size_t demand) override { void request(size_t demand) override {
// Only called by the out_, never by the sink_. The latter triggers // Only called by out_, never by sink_ (triggers on_sink_demand_change()).
prefix_demand_ += demand; prefix_demand_ += demand;
if (sub_ && !requested_prefix_) { if (sub_ && !requested_prefix_) {
sub_.request(prefix_size_); sub_.request(prefix_size_);
......
...@@ -25,7 +25,9 @@ public: ...@@ -25,7 +25,9 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
publish(coordinator* ctx, src_ptr src) : super(ctx), source_(std::move(src)) { publish(coordinator* ctx, src_ptr src,
size_t max_buf_size = defaults::flow::buffer_size)
: super(ctx), max_buf_size_(max_buf_size), source_(std::move(src)) {
// nop // nop
} }
...@@ -74,7 +76,6 @@ public: ...@@ -74,7 +76,6 @@ public:
void auto_disconnect(bool new_value) { void auto_disconnect(bool new_value) {
auto_disconnect_ = new_value; auto_disconnect_ = new_value;
;
} }
bool connected() const noexcept { bool connected() const noexcept {
...@@ -133,7 +134,7 @@ private: ...@@ -133,7 +134,7 @@ private:
} }
size_t in_flight_ = 0; size_t in_flight_ = 0;
size_t max_buf_size_ = defaults::flow::buffer_size; size_t max_buf_size_;
subscription in_; subscription in_;
src_ptr source_; src_ptr source_;
bool connected_ = false; bool connected_ = false;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::flow::op {
/// Represents the state of a flow operator. Some operators only use a subset of
/// the possible states.
enum class state {
idle = 0b0000'0001,
running = 0b0000'0010,
completed = 0b0000'0100,
aborted = 0b0000'1000,
disposed = 0b0001'0000,
};
/// Checks whether `x` is either `completed` or `aborted`.
constexpr bool has_shut_down(state x) {
return (static_cast<int>(x) & 0b0000'1100) != 0;
}
/// @relates state
CAF_CORE_EXPORT std::string to_string(state);
/// @relates state
CAF_CORE_EXPORT bool from_string(std::string_view, state&);
/// @relates state
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<state>, state&);
/// @relates state
template <class Inspector>
bool inspect(Inspector& f, state& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::flow::op
...@@ -89,8 +89,8 @@ public: ...@@ -89,8 +89,8 @@ public:
if (!running && buf.empty()) { if (!running && buf.empty()) {
disposed = true; disposed = true;
if (out) { if (out) {
out.on_error(reason); auto out_hdl = std::move(out);
out = nullptr; out_hdl.on_error(reason);
} }
when_disposed = nullptr; when_disposed = nullptr;
when_consumed_some = nullptr; when_consumed_some = nullptr;
...@@ -99,7 +99,7 @@ public: ...@@ -99,7 +99,7 @@ public:
} }
} }
void do_dispose() { void dispose() {
if (out) { if (out) {
out.on_complete(); out.on_complete();
out = nullptr; out = nullptr;
...@@ -123,6 +123,9 @@ public: ...@@ -123,6 +123,9 @@ public:
auto got_some = demand > 0 && !buf.empty(); auto got_some = demand > 0 && !buf.empty();
for (bool run = got_some; run; run = demand > 0 && !buf.empty()) { for (bool run = got_some; run; run = demand > 0 && !buf.empty()) {
out.on_next(buf.front()); out.on_next(buf.front());
// Note: on_next may call dispose().
if (disposed)
return;
buf.pop_front(); buf.pop_front();
--demand; --demand;
} }
...@@ -132,7 +135,7 @@ public: ...@@ -132,7 +135,7 @@ public:
else else
out.on_complete(); out.on_complete();
out = nullptr; out = nullptr;
do_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);
} }
...@@ -156,16 +159,19 @@ public: ...@@ -156,16 +159,19 @@ public:
// -- implementation of subscription ----------------------------------------- // -- implementation of subscription -----------------------------------------
bool disposed() const noexcept override { bool disposed() const noexcept override {
return !state_; return !state_ || state_->disposed;
} }
void dispose() override { void dispose() override {
if (state_) { if (state_) {
ctx_->delay_fn([state = std::move(state_)]() { state->do_dispose(); }); auto state = std::move(state_);
state->dispose();
} }
} }
void request(size_t n) override { void request(size_t n) override {
if (!state_)
return;
state_->demand += n; state_->demand += n;
if (state_->when_demand_changed) if (state_->when_demand_changed)
state_->when_demand_changed.run(); state_->when_demand_changed.run();
......
...@@ -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_) {
auto& input = at(index);
if (input.sub) {
if (!err_) if (!err_)
err_ = what; err_ = what;
auto& input = at(index);
if (input.sub)
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
...@@ -172,6 +172,10 @@ public: ...@@ -172,6 +172,10 @@ public:
return disposable{std::move(pimpl_)}; return disposable{std::move(pimpl_)};
} }
bool disposed() const noexcept {
return !pimpl_ || pimpl_->disposed();
}
// -- swapping --------------------------------------------------------------- // -- swapping ---------------------------------------------------------------
void swap(subscription& other) noexcept { void swap(subscription& other) noexcept {
......
...@@ -11,7 +11,6 @@ ...@@ -11,7 +11,6 @@
#include <memory> #include <memory>
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/scheduled_actor/flow.hpp" #include "caf/scheduled_actor/flow.hpp"
......
...@@ -4,6 +4,117 @@ ...@@ -4,6 +4,117 @@
#include "core-test.hpp" #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 {
std::string to_string(observer_state x) {
switch (x) {
default:
return "???";
case observer_state::idle:
return "caf::flow::observer_state::idle";
case observer_state::subscribed:
return "caf::flow::observer_state::subscribed";
case observer_state::completed:
return "caf::flow::observer_state::completed";
case observer_state::aborted:
return "caf::flow::observer_state::aborted";
}
}
bool from_string(std::string_view in, observer_state& out) {
if (in == "caf::flow::observer_state::idle") {
out = observer_state::idle;
return true;
}
if (in == "caf::flow::observer_state::subscribed") {
out = observer_state::subscribed;
return true;
}
if (in == "caf::flow::observer_state::completed") {
out = observer_state::completed;
return true;
}
if (in == "caf::flow::observer_state::aborted") {
out = observer_state::aborted;
return true;
}
return false;
}
bool from_integer(std::underlying_type_t<observer_state> in,
observer_state& out) {
auto result = static_cast<observer_state>(in);
switch (result) {
default:
return false;
case observer_state::idle:
case observer_state::subscribed:
case observer_state::completed:
case observer_state::aborted:
out = result;
return true;
}
}
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) { std::string to_string(level lvl) {
switch (lvl) { switch (lvl) {
case level::all: case level::all:
......
...@@ -16,6 +16,323 @@ ...@@ -16,6 +16,323 @@
#include <string> #include <string>
#include <utility> #include <utility>
// -- utility for testing flows ------------------------------------------------
namespace caf::flow {
/// Represents the current state of an @ref observer.
enum class observer_state {
/// Indicates that no callbacks were called yet.
idle,
/// Indicates that on_subscribe was called.
subscribed,
/// Indicates that on_complete was called.
completed,
/// Indicates that on_error was called.
aborted
};
/// Returns whether `x` represents a final state, i.e., `completed`, `aborted`
/// or `disposed`.
constexpr bool is_final(observer_state x) noexcept {
return static_cast<int>(x) >= static_cast<int>(observer_state::completed);
}
/// Returns whether `x` represents an active state, i.e., `idle` or
/// `subscribed`.
constexpr bool is_active(observer_state x) noexcept {
return static_cast<int>(x) <= static_cast<int>(observer_state::subscribed);
}
/// @relates observer_state
std::string to_string(observer_state);
/// @relates observer_state
bool from_string(std::string_view, observer_state&);
/// @relates observer_state
bool from_integer(std::underlying_type_t<observer_state>, observer_state&);
/// @relates observer_state
template <class Inspector>
bool inspect(Inspector& f, observer_state& x) {
return default_enum_inspect(f, x);
}
/// 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 {
if (!subscribed()) {
auto what = "on_next called but observer is in state" + to_string(state);
throw std::logic_error(what);
}
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;
};
template <class T>
class canceling_observer : public flow::observer_impl_base<T> {
public:
explicit canceling_observer(bool accept_first) : accept_next(accept_first) {
// nop
}
void on_next(const T&) override {
++on_next_calls;
if (sub) {
sub.dispose();
sub = nullptr;
}
}
void on_error(const error&) override {
++on_error_calls;
if (sub)
sub = nullptr;
}
void on_complete() override {
++on_complete_calls;
if (sub)
sub = nullptr;
}
void on_subscribe(flow::subscription sub) override {
if (accept_next) {
accept_next = false;
sub.request(128);
this->sub = std::move(sub);
return;
}
sub.dispose();
}
int on_next_calls = 0;
int on_error_calls = 0;
int on_complete_calls = 0;
bool accept_next = false;
flow::subscription sub;
};
template <class T>
auto make_canceling_observer(bool accept_first = false) {
return make_counted<canceling_observer<T>>(accept_first);
}
/// @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:
using super = passive_observer<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 {
super::on_next(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;
};
inline auto make_passive_subscription() {
return make_counted<passive_subscription_impl>();
}
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 ---------------------------- // -- utility for testing serialization round-trips ----------------------------
template <class T> template <class T>
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.broadcaster
#include "caf/flow/observable.hpp"
#include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp"
namespace caf::flow {} // namespace caf::flow
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("a broadcaster pushes items to single subscribers") {
GIVEN("a broadcaster with one source and one sink") {
auto uut = flow::make_broadcaster_impl<int>(ctx.get());
auto src = flow::make_passive_observable<int>(ctx.get());
auto snk = flow::make_passive_observer<int>();
src->subscribe(uut->as_observer());
uut->subscribe(snk->as_observer());
WHEN("the source emits 10 items") {
THEN("the broadcaster forwards them to its sink") {
snk->sub.request(13);
ctx->run();
CHECK_EQ(src->demand, 13u);
snk->sub.request(7);
ctx->run();
CHECK_EQ(src->demand, 20u);
auto inputs = std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
src->push(make_span(inputs));
CHECK_EQ(src->demand, 10u);
CHECK_EQ(uut->buffered(), 0u);
CHECK_EQ(snk->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7, 8, 9, 10}));
src->complete();
ctx->run();
}
}
}
}
SCENARIO("a broadcaster pushes items to all subscribers at the same time") {
GIVEN("a broadcaster with one source and three sinks") {
auto uut = flow::make_broadcaster_impl<int>(ctx.get());
auto src = flow::make_passive_observable<int>(ctx.get());
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto snk3 = flow::make_passive_observer<int>();
src->subscribe(uut->as_observer());
uut->subscribe(snk1->as_observer());
uut->subscribe(snk2->as_observer());
uut->subscribe(snk3->as_observer());
WHEN("the source emits 10 items") {
THEN("the broadcaster forwards them to all sinks") {
snk1->sub.request(13);
ctx->run();
CHECK_EQ(src->demand, 13u);
snk2->sub.request(7);
ctx->run();
CHECK_EQ(src->demand, 13u);
snk3->sub.request(21);
ctx->run();
CHECK_EQ(src->demand, 21u);
auto inputs = std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
src->push(make_span(inputs));
CHECK_EQ(src->demand, 11u);
CHECK_EQ(uut->buffered(), 3u);
CHECK_EQ(snk1->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk2->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk3->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
snk2->sub.request(7);
ctx->run();
CHECK_EQ(src->demand, 11u);
CHECK_EQ(uut->buffered(), 0u);
CHECK_EQ(snk1->buf, inputs);
CHECK_EQ(snk2->buf, inputs);
CHECK_EQ(snk3->buf, inputs);
snk2->sub.request(14);
ctx->run();
CHECK_EQ(src->demand, 18u);
src->complete();
ctx->run();
}
}
}
}
SCENARIO("a broadcaster emits values before propagating completion") {
GIVEN("a broadcaster with one source and three sinks") {
auto uut = flow::make_broadcaster_impl<int>(ctx.get());
auto src = flow::make_passive_observable<int>(ctx.get());
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto snk3 = flow::make_passive_observer<int>();
src->subscribe(uut->as_observer());
uut->subscribe(snk1->as_observer());
uut->subscribe(snk2->as_observer());
uut->subscribe(snk3->as_observer());
WHEN("the source emits 10 items and then signals completion") {
THEN("the broadcaster forwards all values before signaling an error") {
snk1->sub.request(13);
snk2->sub.request(7);
snk3->sub.request(21);
ctx->run();
CHECK_EQ(src->demand, 21u);
auto inputs = std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
src->push(make_span(inputs));
src->complete();
CHECK_EQ(uut->buffered(), 3u);
CHECK_EQ(snk1->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk2->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk3->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(uut->state(), flow::observable_state::completing);
CHECK_EQ(snk1->state, flow::observer_state::subscribed);
CHECK_EQ(snk2->state, flow::observer_state::subscribed);
CHECK_EQ(snk3->state, flow::observer_state::subscribed);
snk2->sub.request(7);
ctx->run();
CHECK_EQ(snk1->buf, inputs);
CHECK_EQ(snk2->buf, inputs);
CHECK_EQ(snk3->buf, inputs);
CHECK_EQ(uut->state(), flow::observable_state::completed);
CHECK_EQ(snk1->state, flow::observer_state::completed);
CHECK_EQ(snk2->state, flow::observer_state::completed);
CHECK_EQ(snk3->state, flow::observer_state::completed);
}
}
}
}
SCENARIO("a broadcaster emits values before propagating errors") {
GIVEN("a broadcaster with one source and three sinks") {
auto uut = flow::make_broadcaster_impl<int>(ctx.get());
auto src = flow::make_passive_observable<int>(ctx.get());
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto snk3 = flow::make_passive_observer<int>();
src->subscribe(uut->as_observer());
uut->subscribe(snk1->as_observer());
uut->subscribe(snk2->as_observer());
uut->subscribe(snk3->as_observer());
WHEN("the source emits 10 items and then stops with an error") {
THEN("the broadcaster forwards all values before signaling an error") {
snk1->sub.request(13);
snk2->sub.request(7);
snk3->sub.request(21);
ctx->run();
CHECK_EQ(src->demand, 21u);
auto inputs = std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
src->push(make_span(inputs));
auto runtime_error = make_error(sec::runtime_error);
src->abort(runtime_error);
CHECK_EQ(uut->buffered(), 3u);
CHECK_EQ(snk1->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk2->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk3->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(uut->state(), flow::observable_state::completing);
CHECK_EQ(uut->err(), runtime_error);
CHECK_EQ(snk1->state, flow::observer_state::subscribed);
CHECK_EQ(snk2->state, flow::observer_state::subscribed);
CHECK_EQ(snk3->state, flow::observer_state::subscribed);
snk2->sub.request(7);
ctx->run();
CHECK_EQ(snk1->buf, inputs);
CHECK_EQ(snk2->buf, inputs);
CHECK_EQ(snk3->buf, inputs);
CHECK_EQ(uut->state(), flow::observable_state::aborted);
CHECK_EQ(snk1->state, flow::observer_state::aborted);
CHECK_EQ(snk2->state, flow::observer_state::aborted);
CHECK_EQ(snk3->state, flow::observer_state::aborted);
CHECK_EQ(uut->err(), runtime_error);
CHECK_EQ(snk1->err, runtime_error);
CHECK_EQ(snk2->err, runtime_error);
CHECK_EQ(snk3->err, runtime_error);
}
}
}
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.buffer
#include "caf/flow/observable.hpp"
#include "core-test.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/item_publisher.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
using namespace std::literals;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("the buffer operator groups items together") {
GIVEN("an observable") {
WHEN("calling .buffer(3)") {
THEN("the observer receives values in groups of three") {
auto inputs = std::vector<int>{1, 2, 4, 8, 16, 32, 64, 128};
auto outputs = std::vector<cow_vector<int>>{};
auto expected = std::vector<cow_vector<int>>{
cow_vector<int>{1, 2, 4},
cow_vector<int>{8, 16, 32},
cow_vector<int>{64, 128},
};
ctx->make_observable()
.from_container(inputs) //
.buffer(3)
.for_each([&outputs](const cow_vector<int>& xs) {
outputs.emplace_back(xs);
});
ctx->run();
CHECK_EQ(outputs, expected);
}
}
}
}
SCENARIO("the buffer operator forces items at regular intervals") {
GIVEN("an observable") {
WHEN("calling .buffer(3, 1s)") {
THEN("the observer receives values in groups of three or after 1s") {
auto outputs = std::make_shared<std::vector<cow_vector<int>>>();
auto expected = std::vector<cow_vector<int>>{
cow_vector<int>{1, 2, 4}, cow_vector<int>{8, 16, 32},
cow_vector<int>{}, cow_vector<int>{64},
cow_vector<int>{}, cow_vector<int>{128, 256, 512},
};
auto pub = flow::item_publisher<int>{ctx.get()};
sys.spawn([&pub, outputs](caf::event_based_actor* self) {
pub //
.as_observable()
.observe_on(self)
.buffer(3, 1s)
.for_each([outputs](const cow_vector<int>& xs) {
outputs->emplace_back(xs);
});
});
sched.run();
MESSAGE("emit the first six items");
pub.push({1, 2, 4, 8, 16, 32});
ctx->run_some();
sched.run();
MESSAGE("force an empty buffer");
advance_time(1s);
sched.run();
MESSAGE("force a buffer with a single element");
pub.push(64);
ctx->run_some();
sched.run();
advance_time(1s);
sched.run();
MESSAGE("force an empty buffer");
advance_time(1s);
sched.run();
MESSAGE("emit the last items and close the source");
pub.push({128, 256, 512});
pub.close();
ctx->run_some();
sched.run();
advance_time(1s);
sched.run();
CHECK_EQ(*outputs, expected);
}
}
}
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.concat
#include "caf/flow/concat.hpp"
#include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
template <class... Ts>
std::vector<int> ls(Ts... xs) {
return std::vector<int>{xs...};
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("concat operators combine inputs") {
GIVEN("two observables") {
WHEN("merging them to a single publisher via concat") {
THEN("the observer receives the output of both sources in order") {
auto outputs = std::vector<int>{};
auto r1 = ctx->make_observable().repeat(11).take(113);
auto r2 = ctx->make_observable().repeat(22).take(223);
ctx->make_observable()
.concat(std::move(r1), std::move(r2))
.for_each([&outputs](int x) { outputs.emplace_back(x); });
ctx->run();
if (CHECK_EQ(outputs.size(), 336u)) {
CHECK(std::all_of(outputs.begin(), outputs.begin() + 113,
[](int x) { return x == 11; }));
CHECK(std::all_of(outputs.begin() + 113, outputs.end(),
[](int x) { return x == 22; }));
}
}
}
}
}
END_FIXTURE_SCOPE()
...@@ -9,7 +9,6 @@ ...@@ -9,7 +9,6 @@
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
......
...@@ -9,7 +9,6 @@ ...@@ -9,7 +9,6 @@
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
......
...@@ -9,7 +9,6 @@ ...@@ -9,7 +9,6 @@
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
......
...@@ -10,7 +10,6 @@ ...@@ -10,7 +10,6 @@
#include "caf/async/blocking_producer.hpp" #include "caf/async/blocking_producer.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
...@@ -311,8 +310,8 @@ SCENARIO("asynchronous buffers can generate flow items") { ...@@ -311,8 +310,8 @@ SCENARIO("asynchronous buffers can generate flow items") {
THEN("the different pointer types manipulate the same ref count") { THEN("the different pointer types manipulate the same ref count") {
using buf_t = async::spsc_buffer<int>; using buf_t = async::spsc_buffer<int>;
using impl_t = flow::op::from_resource_sub<buf_t>; using impl_t = flow::op::from_resource_sub<buf_t>;
auto ptr = make_counted<impl_t>(ctx.get(), nullptr, auto obs = flow::make_auto_observer<int>();
flow::observer<int>::ignore()); auto ptr = make_counted<impl_t>(ctx.get(), nullptr, obs->as_observer());
CHECK_EQ(ptr->get_reference_count(), 1u); CHECK_EQ(ptr->get_reference_count(), 1u);
{ {
auto sub = flow::subscription{ptr.get()}; auto sub = flow::subscription{ptr.get()};
......
...@@ -9,7 +9,6 @@ ...@@ -9,7 +9,6 @@
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable.hpp" #include "caf/flow/observable.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
......
...@@ -11,7 +11,6 @@ ...@@ -11,7 +11,6 @@
#include <memory> #include <memory>
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/scheduled_actor/flow.hpp" #include "caf/scheduled_actor/flow.hpp"
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.op.buffer
#include "caf/flow/op/buffer.hpp"
#include "core-test.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/item_publisher.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
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
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("the buffer operator groups items together") {
GIVEN("an observable") {
WHEN("calling .buffer(3)") {
THEN("the observer receives values in groups of three") {
auto inputs = std::vector<int>{1, 2, 4, 8, 16, 32, 64, 128};
auto outputs = std::vector<cow_vector<int>>{};
auto expected = std::vector<cow_vector<int>>{
cow_vector<int>{1, 2, 4},
cow_vector<int>{8, 16, 32},
cow_vector<int>{64, 128},
};
ctx->make_observable()
.from_container(inputs) //
.buffer(3)
.for_each([&outputs](const cow_vector<int>& xs) {
outputs.emplace_back(xs);
});
ctx->run();
CHECK_EQ(outputs, expected);
}
}
}
}
SCENARIO("the buffer operator forces items at regular intervals") {
GIVEN("an observable") {
WHEN("calling .buffer(3, 1s)") {
THEN("the observer receives values in groups of three or after 1s") {
auto outputs = std::make_shared<std::vector<cow_vector<int>>>();
auto expected = std::vector<cow_vector<int>>{
cow_vector<int>{1, 2, 4}, cow_vector<int>{8, 16, 32},
cow_vector<int>{}, cow_vector<int>{64},
cow_vector<int>{}, cow_vector<int>{128, 256, 512},
};
auto pub = flow::item_publisher<int>{ctx.get()};
sys.spawn([&pub, outputs](caf::event_based_actor* self) {
pub.as_observable()
.observe_on(self) //
.buffer(3, 1s)
.for_each([outputs](const cow_vector<int>& xs) {
outputs->emplace_back(xs);
});
});
sched.run();
MESSAGE("emit the first six items");
pub.push({1, 2, 4, 8, 16, 32});
ctx->run_some();
sched.run();
MESSAGE("force an empty buffer");
advance_time(1s);
sched.run();
MESSAGE("force a buffer with a single element");
pub.push(64);
ctx->run_some();
sched.run();
advance_time(1s);
sched.run();
MESSAGE("force an empty buffer");
advance_time(1s);
sched.run();
MESSAGE("emit the last items and close the source");
pub.push({128, 256, 512});
pub.close();
ctx->run_some();
sched.run();
advance_time(1s);
sched.run();
CHECK_EQ(*outputs, expected);
}
}
}
}
SCENARIO("the buffer operator forwards errors") {
GIVEN("an observable that produces some values followed by an error") {
WHEN("calling .buffer() on it") {
THEN("the observer receives the values and then the error") {
auto outputs = std::make_shared<std::vector<cow_vector<int>>>();
auto err = std::make_shared<error>();
sys.spawn([outputs, err](caf::event_based_actor* self) {
auto obs = self->make_observable();
obs.iota(1)
.take(17)
.concat(obs.fail<int>(make_error(caf::sec::runtime_error)))
.buffer(7, 1s)
.do_on_error([err](const error& what) { *err = what; })
.for_each([outputs](const cow_vector<int>& xs) {
outputs->emplace_back(xs);
});
});
sched.run();
auto expected = std::vector<cow_vector<int>>{
cow_vector<int>{1, 2, 3, 4, 5, 6, 7},
cow_vector<int>{8, 9, 10, 11, 12, 13, 14},
cow_vector<int>{15, 16, 17},
};
CHECK_EQ(*outputs, expected);
CHECK_EQ(*err, caf::sec::runtime_error);
}
}
}
GIVEN("an observable that produces only an error") {
WHEN("calling .buffer() on it") {
THEN("the observer receives the error") {
auto outputs = std::make_shared<std::vector<cow_vector<int>>>();
auto err = std::make_shared<error>();
sys.spawn([outputs, err](caf::event_based_actor* self) {
self->make_observable()
.fail<int>(make_error(caf::sec::runtime_error))
.buffer(3, 1s)
.do_on_error([err](const error& what) { *err = what; })
.for_each([outputs](const cow_vector<int>& xs) {
outputs->emplace_back(xs);
});
});
sched.run();
CHECK(outputs->empty());
CHECK_EQ(*err, caf::sec::runtime_error);
}
}
}
}
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()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.op.cell
#include "caf/flow/op/cell.hpp"
#include "core-test.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
using int_cell = flow::op::cell<int>;
using int_cell_ptr = intrusive_ptr<int_cell>;
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
int_cell_ptr make_cell() {
return make_counted<int_cell>(ctx.get());
}
flow::observable<int> lift(int_cell_ptr cell) {
return flow::observable<int>{cell};
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("a null cell emits zero items") {
GIVEN("an integer cell with an observer") {
WHEN("calling set_null on the cell") {
THEN("the observer receives the completed event") {
auto uut = make_cell();
auto snk = flow::make_passive_observer<int>();
lift(uut).subscribe(snk->as_observer());
REQUIRE(snk->subscribed());
snk->sub.request(128);
ctx->run();
REQUIRE(snk->subscribed());
uut->set_null();
ctx->run();
CHECK(snk->completed());
CHECK(snk->buf.empty());
}
}
}
GIVEN("an integer cell without on bserver") {
WHEN("calling set_null on the cell") {
THEN("observers receive completed events immediately after subscribing") {
auto uut = make_cell();
uut->set_null();
auto snk = flow::make_passive_observer<int>();
lift(uut).subscribe(snk->as_observer());
REQUIRE(snk->subscribed());
snk->sub.request(128);
ctx->run();
CHECK(snk->completed());
CHECK(snk->buf.empty());
}
}
}
}
SCENARIO("a cell with a value emits exactly one item") {
GIVEN("an integer cell with an observer") {
WHEN("calling set_value on the cell") {
THEN("the observer receives on_next and then on_complete") {
auto uut = make_cell();
auto snk = flow::make_passive_observer<int>();
lift(uut).subscribe(snk->as_observer());
REQUIRE(snk->subscribed());
snk->sub.request(128);
ctx->run();
REQUIRE(snk->subscribed());
uut->set_value(42);
ctx->run();
CHECK(snk->completed());
CHECK_EQ(snk->buf, std::vector<int>{42});
}
}
WHEN("disposing the subscription before calling set_value on the cell") {
THEN("the observer does not receive the item") {
auto uut = make_cell();
auto snk = flow::make_passive_observer<int>();
lift(uut).subscribe(snk->as_observer());
REQUIRE(snk->subscribed());
snk->request(128);
ctx->run();
// Normally, we'd call snk->unsubscribe() here. However, that nulls the
// subscription. We want the sub.disposed() call below actually call
// cell_sub::disposed() to have coverage on that member function.
snk->sub.ptr()->dispose();
snk->state = flow::observer_state::idle;
ctx->run();
CHECK(snk->sub.disposed());
CHECK(snk->idle());
uut->set_value(42);
ctx->run();
CHECK(snk->idle());
CHECK(snk->buf.empty());
}
}
}
GIVEN("an integer cell without on bserver") {
WHEN("calling set_null on the cell") {
THEN("the observer receives on_next and then on_complete immediately") {
auto uut = make_cell();
uut->set_value(42);
auto snk = flow::make_passive_observer<int>();
lift(uut).subscribe(snk->as_observer());
REQUIRE(snk->subscribed());
snk->sub.request(128);
ctx->run();
CHECK(snk->completed());
CHECK_EQ(snk->buf, std::vector<int>{42});
}
}
}
}
SCENARIO("a failed cell emits zero item") {
GIVEN("an integer cell with an observer") {
WHEN("calling set_error on the cell") {
THEN("the observer receives on_error") {
auto uut = make_cell();
auto snk = flow::make_passive_observer<int>();
lift(uut).subscribe(snk->as_observer());
REQUIRE(snk->subscribed());
snk->sub.request(128);
ctx->run();
REQUIRE(snk->subscribed());
uut->set_error(sec::runtime_error);
ctx->run();
CHECK(snk->aborted());
CHECK(snk->buf.empty());
CHECK_EQ(snk->err, sec::runtime_error);
}
}
}
GIVEN("an integer cell without on bserver") {
WHEN("calling set_error on the cell") {
THEN("the observer receives on_error immediately when subscribing") {
auto uut = make_cell();
uut->set_error(sec::runtime_error);
auto snk = flow::make_passive_observer<int>();
lift(uut).subscribe(snk->as_observer());
REQUIRE(snk->subscribed());
snk->sub.request(128);
ctx->run();
CHECK(snk->aborted());
CHECK(snk->buf.empty());
CHECK_EQ(snk->err, sec::runtime_error);
}
}
}
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.op.concat
#include "caf/flow/op/concat.hpp"
#include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
// Like op::empty, but calls on_complete immediately instead of waiting for the
// observer to request items. We use this to get more coverage on edge cases.
template <class T>
class insta_empty : public flow::op::cold<T> {
public:
// -- member types -----------------------------------------------------------
using super = flow::op::cold<T>;
using output_type = T;
// -- constructors, destructors, and assignment operators --------------------
explicit insta_empty(flow::coordinator* ctx) : super(ctx) {
// nop
}
// -- implementation of observable<T>::impl ----------------------------------
disposable subscribe(flow::observer<output_type> out) override {
auto sub = make_counted<flow::passive_subscription_impl>();
out.on_subscribe(flow::subscription{sub});
out.on_complete();
return sub->as_disposable();
}
};
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
// Creates a flow::op::concat<T>
template <class T, class... Inputs>
auto make_operator(Inputs&&... inputs) {
return make_counted<flow::op::concat<T>>(ctx.get(),
std::forward<Inputs>(inputs)...);
}
// Similar to concat::subscribe, but returns a concat_sub pointer instead of
// type-erasing it into a disposable.
template <class T, class... Ts>
auto raw_sub(flow::observer<T> out, Ts&&... xs) {
using flow::observable;
using input_type = std::variant<observable<T>, observable<observable<T>>>;
auto vec = std::vector<input_type>{std::forward<Ts>(xs).as_observable()...};
auto ptr = make_counted<flow::op::concat_sub<T>>(ctx.get(), out, vec);
out.on_subscribe(flow::subscription{ptr});
return ptr;
}
template <class T>
auto make_insta_empty() {
return flow::observable<T>{make_counted<insta_empty<T>>(ctx.get())};
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("concat operators combine inputs") {
GIVEN("two observables that emit values") {
WHEN("concatenating them to a single publisher") {
THEN("the observer receives the output of both sources in order") {
auto outputs = std::vector<int>{};
auto r1 = ctx->make_observable().repeat(11).take(113);
auto r2 = ctx->make_observable().repeat(22).take(223);
auto sub = ctx->make_observable()
.concat(std::move(r1), std::move(r2))
.for_each([&outputs](int x) { outputs.emplace_back(x); });
CHECK(!sub.disposed());
ctx->run();
if (CHECK_EQ(outputs.size(), 336u)) {
CHECK(std::all_of(outputs.begin(), outputs.begin() + 113,
[](int x) { return x == 11; }));
CHECK(std::all_of(outputs.begin() + 113, outputs.end(),
[](int x) { return x == 22; }));
}
}
}
WHEN("concatenating them but disposing the operator") {
THEN("the observer only an on_complete event") {
auto r1 = ctx->make_observable().repeat(11).take(113);
auto r2 = ctx->make_observable().repeat(22).take(223);
auto snk = flow::make_passive_observer<int>();
auto sub = ctx->make_observable()
.concat(std::move(r1), std::move(r2))
.subscribe(snk->as_observer());
ctx->run();
sub.dispose();
ctx->run();
CHECK(snk->completed());
CHECK(snk->buf.empty());
}
}
}
GIVEN("an observable of observable") {
WHEN("concatenating it") {
THEN("the observer receives all items") {
auto snk = flow::make_auto_observer<int>();
ctx->make_observable()
.from_container(
std::vector{ctx->make_observable().just(1).as_observable(),
ctx->make_observable().just(2).as_observable(),
make_insta_empty<int>()})
.concat(ctx->make_observable().just(3))
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->completed());
CHECK_EQ(snk->buf, std::vector<int>({1, 2, 3}));
}
}
WHEN("concatenating it but disposing the operator") {
THEN("the observer only an on_complete event") {
auto snk = flow::make_passive_observer<int>();
auto sub = ctx->make_observable()
.never<flow::observable<int>>()
.concat()
.subscribe(snk->as_observer());
ctx->run();
sub.dispose();
ctx->run();
CHECK(snk->completed());
CHECK(snk->buf.empty());
}
}
}
GIVEN("two observables, whereas the first produces an error") {
WHEN("concatenating them to a single publisher") {
THEN("the observer only receives an error") {
auto outputs = std::vector<int>{};
auto r1 = ctx->make_observable().fail<int>(sec::runtime_error);
auto r2 = ctx->make_observable().iota(1).take(3);
auto snk = flow::make_auto_observer<int>();
ctx->make_observable()
.concat(std::move(r1), std::move(r2))
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->aborted());
CHECK(snk->buf.empty());
CHECK_EQ(snk->err, sec::runtime_error);
}
}
}
GIVEN("two observables, whereas the second one produces an error") {
WHEN("concatenating them to a single publisher") {
THEN("the observer receives the first set of items and then an error") {
auto outputs = std::vector<int>{};
auto r1 = ctx->make_observable().iota(1).take(3);
auto r2 = ctx->make_observable().fail<int>(sec::runtime_error);
auto snk = flow::make_auto_observer<int>();
ctx->make_observable()
.concat(std::move(r1), std::move(r2))
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->aborted());
CHECK_EQ(snk->buf, std::vector<int>({1, 2, 3}));
CHECK_EQ(snk->err, sec::runtime_error);
}
}
}
}
SCENARIO("empty concat operators only call on_complete") {
GIVEN("a concat operator with no inputs") {
WHEN("subscribing to it") {
THEN("the observer only receives an on_complete event") {
auto snk = flow::make_auto_observer<int>();
auto sub = make_operator<int>()->subscribe(snk->as_observer());
ctx->run();
CHECK(sub.disposed());
CHECK(snk->completed());
CHECK(snk->buf.empty());
}
}
}
}
SCENARIO("the concat operator disposes unexpected subscriptions") {
GIVEN("a concat operator with no inputs") {
WHEN("subscribing to it") {
THEN("the observer only receives an on_complete event") {
auto snk = flow::make_passive_observer<int>();
auto r1 = ctx->make_observable().just(1).as_observable();
auto r2 = ctx->make_observable().just(2).as_observable();
auto uut = raw_sub(snk->as_observer(), r1, r2);
auto sub = make_counted<flow::passive_subscription_impl>();
ctx->run();
CHECK(!sub->disposed());
uut->fwd_on_subscribe(42, flow::subscription{sub});
CHECK(sub->disposed());
snk->request(127);
ctx->run();
CHECK(snk->completed());
CHECK_EQ(snk->buf, std::vector<int>({1, 2}));
}
}
}
}
END_FIXTURE_SCOPE()
...@@ -2,12 +2,13 @@ ...@@ -2,12 +2,13 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.defer #define CAF_SUITE flow.op.defer
#include "caf/flow/observable_builder.hpp" #include "caf/flow/op/defer.hpp"
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
using namespace caf; using namespace caf;
......
...@@ -2,12 +2,13 @@ ...@@ -2,12 +2,13 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.empty #define CAF_SUITE flow.op.empty
#include "caf/flow/observable_builder.hpp" #include "caf/flow/op/empty.hpp"
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
using namespace caf; using namespace caf;
......
...@@ -2,7 +2,7 @@ ...@@ -2,7 +2,7 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.fail #define CAF_SUITE flow.op.fail
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
......
...@@ -2,12 +2,13 @@ ...@@ -2,12 +2,13 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.interval #define CAF_SUITE flow.op.interval
#include "caf/flow/observable_builder.hpp" #include "caf/flow/op/interval.hpp"
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
using namespace std::literals; using namespace std::literals;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.op.mcast
#include "caf/flow/op/mcast.hpp"
#include "core-test.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
using int_mcast = flow::op::mcast<int>;
using int_mcast_ptr = intrusive_ptr<int_mcast>;
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
int_mcast_ptr make_mcast() {
return make_counted<int_mcast>(ctx.get());
}
auto lift(int_mcast_ptr mcast) {
return flow::observable<int>{mcast};
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("closed mcast operators appear empty") {
GIVEN("a closed mcast operator") {
WHEN("subscribing to it") {
THEN("the observer receives an on_complete event") {
auto uut = make_mcast();
uut->close();
auto snk = flow::make_auto_observer<int>();
lift(uut).subscribe(snk->as_observer());
ctx->run();
CHECK(snk->completed());
}
}
}
}
SCENARIO("aborted mcast operators fail when subscribed") {
GIVEN("an aborted mcast operator") {
WHEN("subscribing to it") {
THEN("the observer receives an on_error event") {
auto uut = make_mcast();
uut->abort(sec::runtime_error);
auto snk = flow::make_auto_observer<int>();
lift(uut).subscribe(snk->as_observer());
ctx->run();
CHECK(snk->aborted());
}
}
}
}
SCENARIO("mcast operators buffer items that they cannot ship immediately") {
GIVEN("an mcast operator with three observers") {
WHEN("pushing more data than the observers have requested") {
THEN("items are buffered individually") {
MESSAGE("subscribe three observers to a fresh mcast operator");
auto uut = make_mcast();
CHECK(!uut->has_observers());
CHECK_EQ(uut->observer_count(), 0u);
CHECK_EQ(uut->max_demand(), 0u);
CHECK_EQ(uut->min_demand(), 0u);
CHECK_EQ(uut->max_buffered(), 0u);
CHECK_EQ(uut->min_buffered(), 0u);
auto o1 = flow::make_passive_observer<int>();
auto o2 = flow::make_passive_observer<int>();
auto o3 = flow::make_passive_observer<int>();
CHECK_EQ(uut->observer_count(), 0u);
auto sub1 = uut->subscribe(o1->as_observer());
CHECK_EQ(uut->observer_count(), 1u);
auto sub2 = uut->subscribe(o2->as_observer());
CHECK_EQ(uut->observer_count(), 2u);
auto sub3 = uut->subscribe(o3->as_observer());
CHECK(uut->has_observers());
CHECK_EQ(uut->observer_count(), 3u);
CHECK_EQ(uut->max_demand(), 0u);
CHECK_EQ(uut->min_demand(), 0u);
CHECK_EQ(uut->max_buffered(), 0u);
CHECK_EQ(uut->min_buffered(), 0u);
MESSAGE("trigger request for items");
o1->request(3);
o2->request(5);
o3->request(7);
ctx->run();
CHECK_EQ(uut->max_demand(), 7u);
CHECK_EQ(uut->min_demand(), 3u);
CHECK_EQ(uut->max_buffered(), 0u);
CHECK_EQ(uut->min_buffered(), 0u);
MESSAGE("push more items than we have demand for");
for (auto i = 0; i < 8; ++i)
uut->push_all(i);
CHECK_EQ(uut->max_demand(), 0u);
CHECK_EQ(uut->min_demand(), 0u);
CHECK_EQ(uut->max_buffered(), 5u);
CHECK_EQ(uut->min_buffered(), 1u);
MESSAGE("drop the subscriber with the largest buffer");
sub1.dispose();
ctx->run();
CHECK_EQ(uut->max_demand(), 0u);
CHECK_EQ(uut->min_demand(), 0u);
CHECK_EQ(uut->max_buffered(), 3u);
CHECK_EQ(uut->min_buffered(), 1u);
}
}
}
}
END_FIXTURE_SCOPE()
...@@ -2,15 +2,14 @@ ...@@ -2,15 +2,14 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.merge #define CAF_SUITE flow.op.merge
#include "caf/flow/merge.hpp" #include "caf/flow/op/merge.hpp"
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/item_publisher.hpp" #include "caf/flow/item_publisher.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/op/merge.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
using namespace caf; using namespace caf;
...@@ -31,12 +30,49 @@ struct fixture : test_coordinator_fixture<> { ...@@ -31,12 +30,49 @@ struct fixture : test_coordinator_fixture<> {
xs.push_back(y); xs.push_back(y);
return xs; return xs;
} }
// Creates a flow::op::merge<T>
template <class T, class... Inputs>
auto make_operator(Inputs&&... inputs) {
return make_counted<flow::op::merge<T>>(ctx.get(),
std::forward<Inputs>(inputs)...);
}
// Similar to merge::subscribe, but returns a merge_sub pointer instead of
// type-erasing it into a disposable.
template <class T, class... Ts>
auto raw_sub(flow::observer<T> out, Ts&&... xs) {
using flow::observable;
auto ptr = make_counted<flow::op::merge_sub<T>>(ctx.get(), out);
(ptr->subscribe_to(xs), ...);
out.on_subscribe(flow::subscription{ptr});
return ptr;
}
}; };
} // namespace } // namespace
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("the merge operator combine inputs") {
GIVEN("two observables") {
WHEN("merging them to a single observable") {
THEN("the observer receives the output of both sources") {
using ivec = std::vector<int>;
auto snk = flow::make_auto_observer<int>();
ctx->make_observable()
.repeat(11)
.take(113)
.merge(ctx->make_observable().repeat(22).take(223))
.subscribe(snk->as_observer());
ctx->run();
CHECK_EQ(snk->state, flow::observer_state::completed);
CHECK_EQ(snk->sorted_buf(), concat(ivec(113, 11), ivec(223, 22)));
}
}
}
}
SCENARIO("mergers round-robin over their inputs") { SCENARIO("mergers round-robin over their inputs") {
GIVEN("a merger with no inputs") { GIVEN("a merger with no inputs") {
auto uut = flow::make_observable<flow::op::merge<int>>(ctx.get()); auto uut = flow::make_observable<flow::op::merge<int>>(ctx.get());
...@@ -50,7 +86,7 @@ SCENARIO("mergers round-robin over their inputs") { ...@@ -50,7 +86,7 @@ SCENARIO("mergers round-robin over their inputs") {
} }
} }
} }
GIVEN("a round-robin merger with one input that completes") { GIVEN("a merger with one input that completes") {
WHEN("subscribing to the merger and requesting before the first push") { WHEN("subscribing to the merger and requesting before the first push") {
auto src = flow::item_publisher<int>{ctx.get()}; auto src = flow::item_publisher<int>{ctx.get()};
auto uut = make_counted<flow::op::merge<int>>(ctx.get(), auto uut = make_counted<flow::op::merge<int>>(ctx.get(),
...@@ -84,7 +120,7 @@ SCENARIO("mergers round-robin over their inputs") { ...@@ -84,7 +120,7 @@ SCENARIO("mergers round-robin over their inputs") {
CHECK_EQ(snk->buf, ls(1, 2, 3, 4, 5)); CHECK_EQ(snk->buf, ls(1, 2, 3, 4, 5));
} }
} }
AND_WHEN("subscribing to the merger pushing before the first request") { WHEN("subscribing to the merger pushing before the first request") {
auto src = flow::item_publisher<int>{ctx.get()}; auto src = flow::item_publisher<int>{ctx.get()};
auto uut = make_counted<flow::op::merge<int>>(ctx.get(), auto uut = make_counted<flow::op::merge<int>>(ctx.get(),
src.as_observable()); src.as_observable());
...@@ -119,7 +155,7 @@ SCENARIO("mergers round-robin over their inputs") { ...@@ -119,7 +155,7 @@ SCENARIO("mergers round-robin over their inputs") {
} }
} }
} }
GIVEN("a round-robin merger with one input that aborts after some items") { GIVEN("a merger with one input that aborts after some items") {
WHEN("subscribing to the merger") { WHEN("subscribing to the merger") {
auto src = flow::item_publisher<int>{ctx.get()}; auto src = flow::item_publisher<int>{ctx.get()};
auto uut = make_counted<flow::op::merge<int>>(ctx.get(), auto uut = make_counted<flow::op::merge<int>>(ctx.get(),
...@@ -167,20 +203,71 @@ SCENARIO("mergers round-robin over their inputs") { ...@@ -167,20 +203,71 @@ SCENARIO("mergers round-robin over their inputs") {
} }
} }
SCENARIO("the merge operator combine inputs") { SCENARIO("empty merge operators only call on_complete") {
GIVEN("two observables") { GIVEN("a merge operator with no inputs") {
WHEN("merging them to a single observable") { WHEN("subscribing to it") {
THEN("the observer receives the output of both sources") { THEN("the observer only receives an on_complete event") {
using ivec = std::vector<int>;
auto snk = flow::make_auto_observer<int>(); auto snk = flow::make_auto_observer<int>();
ctx->make_observable() auto sub = make_operator<int>()->subscribe(snk->as_observer());
.repeat(11)
.take(113)
.merge(ctx->make_observable().repeat(22).take(223))
.subscribe(snk->as_observer());
ctx->run(); ctx->run();
CHECK_EQ(snk->state, flow::observer_state::completed); CHECK(sub.disposed());
CHECK_EQ(snk->sorted_buf(), concat(ivec(113, 11), ivec(223, 22))); CHECK(snk->completed());
CHECK(snk->buf.empty());
}
}
}
}
SCENARIO("the merge operator disposes unexpected subscriptions") {
GIVEN("a merge operator with no inputs") {
WHEN("subscribing to it") {
THEN("the observer only receives an on_complete event") {
auto snk = flow::make_passive_observer<int>();
auto r1 = ctx->make_observable().just(1).as_observable();
auto r2 = ctx->make_observable().just(2).as_observable();
auto uut = raw_sub(snk->as_observer(), r1, r2);
auto sub = make_counted<flow::passive_subscription_impl>();
ctx->run();
CHECK(!sub->disposed());
uut->fwd_on_subscribe(42, flow::subscription{sub});
CHECK(sub->disposed());
snk->request(127);
ctx->run();
CHECK(snk->completed());
CHECK_EQ(snk->buf, std::vector<int>({1, 2}));
}
}
}
}
SCENARIO("the merge operator drops inputs with no pending data on error") {
GIVEN("a merge operator with two inputs") {
WHEN("one of the inputs fails") {
THEN("the operator drops the other input right away") {
auto snk = flow::make_auto_observer<int>();
auto uut
= raw_sub(snk->as_observer(), ctx->make_observable().never<int>(),
ctx->make_observable().fail<int>(sec::runtime_error));
ctx->run();
CHECK(uut->disposed());
}
}
}
}
SCENARIO("the merge operator drops inputs when disposed") {
GIVEN("a merge operator with two inputs") {
WHEN("one of the inputs fails") {
THEN("the operator drops the other input right away") {
auto snk = flow::make_auto_observer<int>();
auto uut = raw_sub(snk->as_observer(),
ctx->make_observable().never<int>(),
ctx->make_observable().never<int>());
ctx->run();
CHECK(!uut->disposed());
uut->dispose();
ctx->run();
CHECK(uut->disposed());
} }
} }
} }
......
...@@ -2,12 +2,13 @@ ...@@ -2,12 +2,13 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.never #define CAF_SUITE flow.op.never
#include "caf/flow/observable_builder.hpp" #include "caf/flow/op/never.hpp"
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
using namespace caf; using namespace caf;
...@@ -33,8 +34,9 @@ SCENARIO("the never operator never invokes callbacks except when disposed") { ...@@ -33,8 +34,9 @@ SCENARIO("the never operator never invokes callbacks except when disposed") {
ctx->run(); ctx->run();
CHECK(snk1->buf.empty()); CHECK(snk1->buf.empty());
CHECK_EQ(snk1->state, flow::observer_state::subscribed); CHECK_EQ(snk1->state, flow::observer_state::subscribed);
sub1.dispose(); sub1.ptr()->dispose();
ctx->run(); ctx->run();
CHECK(sub1.ptr()->disposed());
CHECK_EQ(snk1->state, flow::observer_state::completed); CHECK_EQ(snk1->state, flow::observer_state::completed);
MESSAGE("dispose only affects the subscription, " MESSAGE("dispose only affects the subscription, "
"the never operator remains unchanged"); "the never operator remains unchanged");
......
...@@ -4,14 +4,14 @@ ...@@ -4,14 +4,14 @@
#define CAF_SUITE flow.prefix_and_tail #define CAF_SUITE flow.prefix_and_tail
#include "caf/flow/observable.hpp" #include "caf/flow/op/prefix_and_tail.hpp"
#include "core-test.hpp" #include "core-test.hpp"
#include <memory> #include <memory>
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp" #include "caf/flow/observable.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
...@@ -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()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.op.ucast
#include "caf/flow/op/ucast.hpp"
#include "core-test.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
using int_ucast = flow::op::ucast<int>;
using int_ucast_ptr = intrusive_ptr<int_ucast>;
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
int_ucast_ptr make_ucast() {
return make_counted<int_ucast>(ctx.get());
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("closed ucast operators appear empty") {
GIVEN("a closed ucast operator") {
WHEN("subscribing to it") {
THEN("the observer receives an on_complete event") {
auto uut = make_ucast();
uut->close();
auto snk = flow::make_auto_observer<int>();
uut->subscribe(snk->as_observer());
ctx->run();
CHECK(snk->completed());
}
}
}
}
SCENARIO("aborted ucast operators fail when subscribed") {
GIVEN("an aborted ucast operator") {
WHEN("subscribing to it") {
THEN("the observer receives an on_error event") {
auto uut = make_ucast();
uut->abort(sec::runtime_error);
auto snk = flow::make_auto_observer<int>();
uut->subscribe(snk->as_observer());
ctx->run();
CHECK(snk->aborted());
}
}
}
}
SCENARIO("ucast operators may only be subscribed to once") {
GIVEN("a ucast operator") {
WHEN("two observers subscribe to it") {
THEN("the second subscription fails") {
auto uut = make_ucast();
auto o1 = flow::make_passive_observer<int>();
auto o2 = flow::make_passive_observer<int>();
auto sub1 = uut->subscribe(o1->as_observer());
auto sub2 = uut->subscribe(o2->as_observer());
CHECK(o1->subscribed());
CHECK(!sub1.disposed());
CHECK(o2->aborted());
CHECK(sub2.disposed());
}
}
}
}
SCENARIO("observers may cancel ucast subscriptions at any time") {
GIVEN("a ucast operator") {
WHEN("the observer disposes its subscription in on_next") {
THEN("no further items arrive") {
auto uut = make_ucast();
auto snk = flow::make_canceling_observer<int>(true);
auto sub = uut->subscribe(snk->as_observer());
CHECK(!sub.disposed());
uut->push(1);
uut->push(2);
ctx->run();
CHECK(sub.disposed());
CHECK_EQ(snk->on_next_calls, 1);
}
}
}
}
SCENARIO("ucast operators deliver pending items before raising errors") {
GIVEN("a ucast operator with pending items") {
WHEN("an error event occurs") {
THEN("the operator still delivers the pending items first") {
auto uut = make_ucast();
auto snk = flow::make_auto_observer<int>();
uut->subscribe(snk->as_observer());
uut->push(1);
uut->push(2);
uut->abort(sec::runtime_error);
ctx->run();
CHECK(snk->aborted());
CHECK_EQ(snk->buf, std::vector<int>({1, 2}));
}
}
}
}
SCENARIO("requesting from disposed ucast operators is a no-op") {
GIVEN("a ucast operator with a disposed subscription") {
WHEN("calling request() on the subscription") {
THEN("the demand is ignored") {
auto uut = make_ucast();
auto snk = flow::make_canceling_observer<int>(true);
auto sub = uut->subscribe(snk->as_observer());
CHECK(!sub.disposed());
uut->push(1);
uut->push(2);
ctx->run();
CHECK(sub.disposed());
dynamic_cast<flow::subscription::impl*>(sub.ptr())->request(42);
ctx->run();
CHECK(sub.disposed());
CHECK_EQ(snk->on_next_calls, 1);
}
}
}
}
END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.op.zip_with
#include "caf/flow/op/zip_with.hpp"
#include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
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
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("zip_with combines inputs") {
GIVEN("two observables") {
WHEN("merging them with zip_with") {
THEN("the observer receives the combined output of both sources") {
auto snk = flow::make_passive_observer<int>();
ctx->make_observable()
.zip_with([](int x, int y) { return x + y; },
ctx->make_observable().repeat(11).take(113),
ctx->make_observable().repeat(22).take(223))
.subscribe(snk->as_observer());
ctx->run();
REQUIRE_EQ(snk->state, flow::observer_state::subscribed);
snk->sub.request(64);
ctx->run();
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK_EQ(snk->buf.size(), 64u);
snk->sub.request(64);
ctx->run();
CHECK_EQ(snk->state, flow::observer_state::completed);
CHECK_EQ(snk->buf.size(), 113u);
CHECK_EQ(snk->buf, std::vector<int>(113, 33));
}
}
}
}
SCENARIO("zip_with emits nothing when zipping an empty observable") {
GIVEN("two observables, one of them empty") {
WHEN("merging them with zip_with") {
THEN("the observer sees on_complete immediately") {
auto snk = flow::make_auto_observer<int>();
ctx->make_observable()
.zip_with([](int x, int y, int z) { return x + y + z; },
ctx->make_observable().repeat(11),
ctx->make_observable().repeat(22),
ctx->make_observable().empty<int>())
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->buf.empty());
CHECK_EQ(snk->state, flow::observer_state::completed);
}
}
}
}
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()
...@@ -9,6 +9,7 @@ ...@@ -9,6 +9,7 @@
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/observable_builder.hpp" #include "caf/flow/observable_builder.hpp"
#include "caf/flow/op/cell.hpp"
#include "caf/flow/scoped_coordinator.hpp" #include "caf/flow/scoped_coordinator.hpp"
using namespace caf; using namespace caf;
...@@ -117,4 +118,117 @@ SCENARIO("publish creates a connectable observable") { ...@@ -117,4 +118,117 @@ SCENARIO("publish creates a connectable observable") {
} }
} }
SCENARIO("connectable observables forward errors") {
GIVEN("a connectable with a cell and two subscribers") {
WHEN("the cell fails") {
THEN("all subscribers receive the error") {
auto cell = make_counted<flow::op::cell<int>>(ctx.get());
auto snk1 = flow::make_auto_observer<int>();
auto snk2 = flow::make_auto_observer<int>();
flow::observable<int>{cell}.share(2).compose(subscribe_all(snk1, snk2));
ctx->run();
CHECK(snk1->subscribed());
CHECK(snk2->subscribed());
cell->set_error(sec::runtime_error);
ctx->run();
CHECK(snk1->aborted());
CHECK(snk2->aborted());
}
}
}
GIVEN("an already failed connectable") {
WHEN("subscribing to it") {
THEN("the subscribers receive the error immediately") {
auto cell = make_counted<flow::op::cell<int>>(ctx.get());
auto conn = flow::observable<int>{cell}.share();
cell->set_error(sec::runtime_error);
// First subscriber to trigger subscription to the cell.
conn.subscribe(flow::make_auto_observer<int>()->as_observer());
ctx->run();
// After this point, new subscribers should be aborted right away.
auto snk = flow::make_auto_observer<int>();
auto sub = conn.subscribe(snk->as_observer());
CHECK(sub.disposed());
CHECK(snk->aborted());
ctx->run();
}
}
}
}
SCENARIO("observers that dispose their subscription do not affect others") {
GIVEN("a connectable with two subscribers") {
WHEN("one of the subscribers disposes its subscription") {
THEN("the other subscriber still receives all data") {
using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl(), 5);
auto sub1 = uut->subscribe(snk1->as_observer());
auto sub2 = uut->subscribe(snk2->as_observer());
uut->connect();
ctx->run();
snk1->request(7);
snk2->request(3);
ctx->run();
CHECK_EQ(snk1->buf, ls(1, 2, 3, 4, 5, 6, 7));
CHECK_EQ(snk2->buf, ls(1, 2, 3));
snk2->sub.dispose();
ctx->run();
snk1->request(42);
ctx->run();
CHECK_EQ(snk1->buf, ls(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12));
}
}
}
}
SCENARIO("publishers with auto_disconnect auto-dispose their subscription") {
GIVEN("a connectable with two subscribers") {
WHEN("both subscribers drop out and auto_disconnect is enabled") {
THEN("the publisher becomes disconnected") {
using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl(), 5);
auto sub1 = uut->subscribe(snk1->as_observer());
auto sub2 = uut->subscribe(snk2->as_observer());
uut->auto_disconnect(true);
uut->connect();
CHECK(uut->connected());
ctx->run();
snk1->request(7);
snk2->request(3);
ctx->run();
CHECK_EQ(snk1->buf, ls(1, 2, 3, 4, 5, 6, 7));
CHECK_EQ(snk2->buf, ls(1, 2, 3));
snk1->sub.dispose();
snk2->sub.dispose();
ctx->run();
CHECK(!uut->connected());
}
}
}
}
SCENARIO("publishers dispose unexpected subscriptions") {
GIVEN("an initialized publish operator") {
WHEN("calling on_subscribe with unexpected subscriptions") {
THEN("the operator disposes them immediately") {
using impl_t = flow::op::publish<int>;
auto snk1 = flow::make_passive_observer<int>();
auto iota = ctx->make_observable().iota(1).take(12).as_observable();
auto uut = make_counted<impl_t>(ctx.get(), iota.pimpl());
uut->subscribe(snk1->as_observer());
uut->connect();
auto sub = flow::make_passive_subscription();
uut->on_subscribe(flow::subscription{sub});
CHECK(sub->disposed());
}
}
}
}
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.zip_with
#include "caf/flow/observable_builder.hpp"
#include "core-test.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("zip_with combines inputs") {
GIVEN("two observables") {
WHEN("merging them with zip_with") {
THEN("the observer receives the combined output of both sources") {
auto snk = flow::make_passive_observer<int>();
ctx->make_observable()
.zip_with([](int x, int y) { return x + y; },
ctx->make_observable().repeat(11).take(113),
ctx->make_observable().repeat(22).take(223))
.subscribe(snk->as_observer());
ctx->run();
REQUIRE_EQ(snk->state, flow::observer_state::subscribed);
snk->sub.request(64);
ctx->run();
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK_EQ(snk->buf.size(), 64u);
snk->sub.request(64);
ctx->run();
CHECK_EQ(snk->state, flow::observer_state::completed);
CHECK_EQ(snk->buf.size(), 113u);
CHECK_EQ(snk->buf, std::vector<int>(113, 33));
}
}
}
}
SCENARIO("zip_with emits nothing when zipping an empty observable") {
GIVEN("two observables, one of them empty") {
WHEN("merging them with zip_with") {
THEN("the observer sees on_complete immediately") {
auto snk = flow::make_auto_observer<int>();
ctx->make_observable()
.zip_with([](int x, int y, int z) { return x + y + z; },
ctx->make_observable().repeat(11),
ctx->make_observable().repeat(22),
ctx->make_observable().empty<int>())
.subscribe(snk->as_observer());
ctx->run();
CHECK(snk->buf.empty());
CHECK_EQ(snk->state, flow::observer_state::completed);
}
}
}
}
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