Commit c9aa5463 authored by Dominik Charousset's avatar Dominik Charousset

Reduce redundancies in flow operators

parent bd86505d
......@@ -292,6 +292,7 @@ caf_add_component(
flow.error
flow.flat_map
flow.for_each
flow.generation
flow.interval
flow.merge
flow.never
......
......@@ -544,11 +544,11 @@ using processor_impl = typename processor<In, Out>::impl;
// -- representing an error as an observable -----------------------------------
template <class T>
class observable_error_impl : public ref_counted, public observable_impl<T> {
class observable_error_impl : public observable_impl_base<T> {
public:
// -- member types -----------------------------------------------------------
using output_type = T;
using super = observable_impl_base<T>;
// -- friends ----------------------------------------------------------------
......@@ -557,7 +557,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
observable_error_impl(coordinator* ctx, error what)
: ctx_(ctx), what_(std::move(what)) {
: super(ctx), what_(std::move(what)) {
// nop
}
......@@ -571,20 +571,8 @@ public:
return true;
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<T>*, size_t) override {
CAF_RAISE_ERROR("observable_error_impl::on_request called");
}
......@@ -605,26 +593,6 @@ private:
// -- broadcasting -------------------------------------------------------------
template <class Impl>
struct term_step {
Impl* pimpl;
using output_type = typename Impl::output_type;
bool on_next(const output_type& item) {
pimpl->append_to_buf(item);
return true;
}
void on_complete() {
pimpl->shutdown();
}
void on_error(const error& what) {
pimpl->abort(what);
}
};
/// Broadcasts its input to all observers without modifying it.
template <class Step, class... Steps>
class broadcaster_impl
......@@ -720,6 +688,10 @@ public:
return ctx_;
}
disposable subscribe(observer<output_type> sink) override {
return term_.add(this, sink);
}
void on_request(observer_impl<output_type>* sink, size_t n) override {
CAF_LOG_TRACE(CAF_ARG(n));
term_.on_request(sub_, sink, n);
......@@ -730,10 +702,6 @@ public:
term_.on_cancel(sub_, sink);
}
disposable subscribe(observer<output_type> sink) override {
return term_.add(this, sink);
}
// -- properties -------------------------------------------------------------
size_t buffered() const noexcept {
......@@ -962,11 +930,11 @@ struct merger_input {
/// Combines items from any number of observables.
template <class T>
class merger_impl : public ref_counted, public observable_impl<T> {
class merger_impl : public observable_impl_base<T> {
public:
// -- member types -----------------------------------------------------------
using super = observable_impl<T>;
using super = observable_impl_base<T>;
using input_t = merger_input<T>;
......@@ -985,7 +953,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit merger_impl(coordinator* ctx) : ctx_(ctx) {
explicit merger_impl(coordinator* ctx) : super(ctx) {
// nop
}
......@@ -1010,20 +978,8 @@ public:
return term_.finalized();
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<T>* sink, size_t demand) override {
if (auto n = term_.on_request(sink, demand); n > 0) {
pull(n);
......@@ -1220,9 +1176,6 @@ private:
}
};
/// Points to our scheduling context.
coordinator* ctx_;
/// Fine-tunes the behavior of the merger.
flags_t flags_;
......@@ -1364,11 +1317,11 @@ observable<T>::flat_map_optional(F f) {
/// Combines items from any number of observables.
template <class T>
class concat_impl : public ref_counted, public observable_impl<T> {
class concat_impl : public observable_impl_base<T> {
public:
// -- member types -----------------------------------------------------------
using super = observable_impl<T>;
using super = observable_impl_base<T>;
using input_key = size_t;
......@@ -1381,7 +1334,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit concat_impl(coordinator* ctx) : ctx_(ctx) {
explicit concat_impl(coordinator* ctx) : super(ctx) {
// nop
}
......@@ -1398,20 +1351,8 @@ public:
return term_.finalized();
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<T>* sink, size_t demand) override {
if (auto n = term_.on_request(sink, demand); n > 0) {
in_flight_ += n;
......@@ -1537,9 +1478,6 @@ private:
}
};
/// Points to our scheduling context.
coordinator* ctx_;
/// Fine-tunes the behavior of the concat.
flags_t flags_;
......@@ -1877,8 +1815,7 @@ observable<cow_tuple<T, observable<T>>> observable<T>::head_and_tail() {
/// @note Only supports a single observer.
template <class Buffer>
class observable_buffer_impl
: public ref_counted,
public observable_impl<typename Buffer::value_type>,
: public observable_impl_base<typename Buffer::value_type>,
public async::consumer {
public:
// -- member types -----------------------------------------------------------
......@@ -1887,7 +1824,7 @@ public:
using buffer_ptr = intrusive_ptr<Buffer>;
using super = observable_impl<value_type>;
using super = observable_impl_base<value_type>;
// -- friends ----------------------------------------------------------------
......@@ -1896,7 +1833,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
observable_buffer_impl(coordinator* ctx, buffer_ptr buf)
: ctx_(ctx), buf_(buf) {
: super(ctx), buf_(buf) {
// Unlike regular observables, we need a strong reference to the context.
// Otherwise, the buffer might call schedule_fn on a destroyed object.
this->ctx()->ref_coordinator();
......@@ -1910,10 +1847,6 @@ public:
// -- implementation of disposable::impl -------------------------------------
coordinator* ctx() const noexcept final {
return ctx_;
}
void dispose() override {
CAF_LOG_TRACE("");
if (buf_) {
......@@ -1930,14 +1863,6 @@ public:
return buf_ == nullptr;
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
void on_request(observer_impl<value_type>*, size_t n) override {
......@@ -2090,10 +2015,12 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) {
/// An observable that represents an empty range. As soon as an observer
/// requests values from this observable, it calls `on_complete`.
template <class T>
class empty_observable_impl : public ref_counted, public observable_impl<T> {
class empty_observable_impl : public observable_impl_base<T> {
public:
// -- member types -----------------------------------------------------------
using super = observable_impl_base<T>;
using output_type = T;
// -- friends ----------------------------------------------------------------
......@@ -2102,7 +2029,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit empty_observable_impl(coordinator* ctx) : ctx_(ctx) {
explicit empty_observable_impl(coordinator* ctx) : super(ctx) {
// nop
}
......@@ -2116,20 +2043,8 @@ public:
return true;
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<output_type>* snk, size_t) override {
snk->on_complete();
}
......@@ -2148,10 +2063,12 @@ private:
/// An observable that never calls any callbacks on its subscribers.
template <class T>
class mute_observable_impl : public ref_counted, public observable_impl<T> {
class mute_observable_impl : public observable_impl_base<T> {
public:
// -- member types -----------------------------------------------------------
using super = observable_impl_base<T>;
using output_type = T;
// -- friends ----------------------------------------------------------------
......@@ -2160,7 +2077,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit mute_observable_impl(coordinator* ctx) : ctx_(ctx) {
explicit mute_observable_impl(coordinator* ctx) : super(ctx) {
// nop
}
......@@ -2178,20 +2095,8 @@ public:
return disposed_;
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<output_type>*, size_t) override {
// nop
}
......@@ -2212,20 +2117,19 @@ public:
}
private:
coordinator* ctx_;
bool disposed_ = false;
std::vector<observer<output_type>> observers_;
};
/// An observable with minimal internal logic. Useful for writing unit tests.
template <class T>
class passive_observable : public ref_counted, public observable_impl<T> {
class passive_observable : public observable_impl_base<T> {
public:
// -- member types -----------------------------------------------------------
using output_type = T;
using super = observable_impl<T>;
using super = observable_impl_base<T>;
// -- friends ----------------------------------------------------------------
......@@ -2233,7 +2137,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
passive_observable(coordinator* ctx) : ctx_(ctx) {
passive_observable(coordinator* ctx) : super(ctx) {
// nop
}
......@@ -2254,20 +2158,8 @@ public:
return !is_active(state);
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable_impl<T> -----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<output_type>* sink, size_t n) override {
if (out.ptr() == sink) {
demand += n;
......@@ -2342,9 +2234,6 @@ public:
size_t demand = 0;
observer<T> out;
private:
coordinator* ctx_;
};
/// @relates passive_observable
......
......@@ -163,8 +163,8 @@ public:
return observable<int64_t>{std::move(ptr)};
}
/// Creates an @ref observable without any values that simply calls
/// `on_complete` after subscribing to it.
/// Creates an @ref observable without any values that calls `on_complete`
/// after subscribing to it.
template <class T>
[[nodiscard]] observable<T> empty() {
return observable<T>{make_counted<empty_observable_impl<T>>(ctx_)};
......@@ -176,7 +176,8 @@ public:
return observable<T>{make_counted<mute_observable_impl<T>>(ctx_)};
}
/// Creates an @ref observable without any values that also never terminates.
/// Creates an @ref observable without any values that calls `on_error` after
/// subscribing to it.
template <class T>
[[nodiscard]] observable<T> error(caf::error what) {
auto ptr = make_counted<observable_error_impl<T>>(ctx_, std::move(what));
......@@ -212,6 +213,7 @@ public:
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
CAF_LOG_TRACE(CAF_ARG(n));
while (pos_ != values_.end() && n > 0) {
if (!step.on_next(*pos_++, steps...))
return;
......@@ -243,6 +245,7 @@ public:
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
CAF_LOG_TRACE(CAF_ARG(n));
for (size_t i = 0; i < n; ++i)
if (!step.on_next(value_, steps...))
return;
......@@ -268,7 +271,9 @@ public:
value_source& operator=(const value_source&) = default;
template <class Step, class... Steps>
void pull(size_t, Step& step, Steps&... steps) {
void pull([[maybe_unused]] size_t n, Step& step, Steps&... steps) {
CAF_LOG_TRACE(CAF_ARG(n));
CAF_ASSERT(n > 0);
if (step.on_next(value_, steps...))
step.on_complete(steps...);
}
......@@ -296,6 +301,7 @@ public:
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
CAF_LOG_TRACE(CAF_ARG(n));
for (size_t i = 0; i < n; ++i)
if (!step.on_next(fn_(), steps...))
return;
......@@ -326,85 +332,46 @@ public:
// -- implementation of disposable::impl -----------------------------------
void dispose() override {
disposed_ = true;
if (out_) {
out_.on_complete();
out_ = nullptr;
}
term_.dispose();
}
bool disposed() const noexcept override {
return disposed_;
return !term_.active();
}
// -- implementation of observable_impl<T> ---------------------------------
disposable subscribe(observer<output_type> what) override {
if (out_) {
return super::reject_subscription(what, sec::too_many_observers);
} else if (disposed_) {
return super::reject_subscription(what, sec::disposed);
} else {
out_ = what;
return super::do_subscribe(what);
disposable subscribe(observer<output_type> sink) override {
CAF_LOG_TRACE(CAF_ARG2("sink", sink.ptr()));
auto sub = term_.add(this, sink);
if (sub) {
term_.start();
}
return sub;
}
void on_request(observer_impl<output_type>* sink, size_t n) override {
if (sink == out_.ptr()) {
auto fn = [this, n](auto&... steps) {
term_step<impl> term{this};
gen_.pull(n, steps..., term);
};
void on_request(observer_impl<output_type>* sink, size_t demand) override {
CAF_LOG_TRACE(CAF_ARG(sink) << CAF_ARG(demand));
if (auto n = term_.on_request(sink, demand); n > 0) {
auto fn = [this, n](auto&... steps) { gen_.pull(n, steps..., term_); };
std::apply(fn, steps_);
push();
term_.push();
}
}
void on_cancel(observer_impl<output_type>* sink) override {
if (sink == out_.ptr()) {
buf_.clear();
out_ = nullptr;
disposed_ = true;
}
}
// -- callbacks for term_step ----------------------------------------------
void append_to_buf(const output_type& item) {
CAF_ASSERT(out_.valid());
buf_.emplace_back(item);
}
void shutdown() {
CAF_ASSERT(out_.valid());
push();
out_.on_complete();
out_ = nullptr;
disposed_ = true;
CAF_LOG_TRACE(CAF_ARG(sink));
if (auto n = term_.on_cancel(sink); n > 0) {
auto fn = [this, n](auto&... steps) { gen_.pull(n, steps..., term_); };
std::apply(fn, steps_);
term_.push();
}
void abort(const error& reason) {
CAF_ASSERT(out_.valid());
push();
out_.on_error(reason);
out_ = nullptr;
disposed_ = true;
}
private:
void push() {
if (!buf_.empty()) {
out_.on_next(make_span(buf_));
buf_.clear();
}
}
Generator gen_;
std::tuple<Steps...> steps_;
observer<output_type> out_;
bool disposed_ = false;
std::vector<output_type> buf_;
broadcast_step<output_type> term_;
};
template <class... Ts>
......
// 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.generation
#include "caf/flow/observable.hpp"
#include "core-test.hpp"
#include "caf/flow/coordinator.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;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("the repeater source repeats one value indefinitely") {
GIVEN("a repeater source") {
WHEN("subscribing to its output") {
THEN("the observer receives the same value over and over again") {
using ivec = std::vector<int>;
auto snk = flow::make_passive_observer<int>();
ctx->make_observable().repeat(42).subscribe(snk->as_observer());
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty());
if (CHECK(snk->sub)) {
snk->sub.request(3);
ctx->run();
CHECK_EQ(snk->buf, ivec({42, 42, 42}));
snk->sub.request(4);
ctx->run();
CHECK_EQ(snk->buf, ivec({42, 42, 42, 42, 42, 42, 42}));
snk->sub.cancel();
ctx->run();
CHECK_EQ(snk->buf, ivec({42, 42, 42, 42, 42, 42, 42}));
}
}
}
}
}
SCENARIO("the container source streams its input values") {
GIVEN("a container source") {
WHEN("subscribing to its output") {
THEN("the observer receives the values from the container in order") {
using ivec = std::vector<int>;
auto xs = ivec{1, 2, 3, 4, 5, 6, 7};
auto snk = flow::make_passive_observer<int>();
ctx->make_observable().from_container(xs).subscribe(snk->as_observer());
CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty());
if (CHECK(snk->sub)) {
snk->sub.request(3);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3}));
snk->sub.request(21);
ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7}));
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