Commit a78f55e3 authored by Dominik Charousset's avatar Dominik Charousset

Consolidate and clean up steps and transformations

parent 45bfe5f2
// 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/detail/comparable.hpp"
#include "caf/intrusive_cow_ptr.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <vector>
namespace caf {
/// A copy-on-write vector implementation that wraps a `std::vector`.
template <class T>
class cow_vector {
public:
// -- member types -----------------------------------------------------------
using std_type = std::vector<T>;
using size_type = typename std_type::size_type;
using const_iterator = typename std_type::const_iterator;
using const_reverse_iterator = typename std_type::const_reverse_iterator;
// -- constants --------------------------------------------------------------
static inline const size_type npos = std_type::npos;
// -- constructors, destructors, and assignment operators --------------------
cow_vector() {
impl_ = make_counted<impl>();
}
explicit cow_vector(std_type std) {
impl_ = make_counted<impl>(std::move(std));
}
cow_vector(cow_vector&&) noexcept = default;
cow_vector(const cow_vector&) noexcept = default;
cow_vector& operator=(cow_vector&&) noexcept = default;
cow_vector& operator=(const cow_vector&) noexcept = default;
// -- properties -------------------------------------------------------------
/// Returns a mutable reference to the managed vector. Copies the vector if
/// more than one reference to it exists to make sure the reference count is
/// exactly 1 when returning from this function.
std_type& unshared() {
return impl_.unshared().std;
}
/// Returns the managed STD container.
const std_type& std() const noexcept {
return impl_->std;
}
/// Returns whether the reference count of the managed object is 1.
[[nodiscard]] bool unique() const noexcept {
return impl_->unique();
}
[[nodiscard]] bool empty() const noexcept {
return impl_->std.empty();
}
size_type size() const noexcept {
return impl_->std.size();
}
size_type max_size() const noexcept {
return impl_->std.max_size();
}
// -- element access ---------------------------------------------------------
T at(size_type pos) const {
return impl_->std.at(pos);
}
T operator[](size_type pos) const {
return impl_->std[pos];
}
T front() const {
return impl_->std.front();
}
T back() const {
return impl_->std.back();
}
const T* data() const noexcept {
return impl_->std.data();
}
// -- iterator access --------------------------------------------------------
const_iterator begin() const noexcept {
return impl_->std.begin();
}
const_iterator cbegin() const noexcept {
return impl_->std.begin();
}
const_reverse_iterator rbegin() const noexcept {
return impl_->std.rbegin();
}
const_reverse_iterator crbegin() const noexcept {
return impl_->std.rbegin();
}
const_iterator end() const noexcept {
return impl_->std.end();
}
const_iterator cend() const noexcept {
return impl_->std.end();
}
const_reverse_iterator rend() const noexcept {
return impl_->std.rend();
}
const_reverse_iterator crend() const noexcept {
return impl_->std.rend();
}
// -- friends ----------------------------------------------------------------
template <class Inspector>
friend bool inspect(Inspector& f, cow_vector& x) {
if constexpr (Inspector::is_loading) {
return f.apply(x.unshared());
} else {
return f.apply(x.impl_->std);
}
}
private:
struct impl : ref_counted {
std_type std;
impl() = default;
explicit impl(std_type in) : std(std::move(in)) {
// nop
}
impl* copy() const {
return new impl{std};
}
};
intrusive_cow_ptr<impl> impl_;
};
// -- comparison ---------------------------------------------------------------
template <class T>
auto operator==(const cow_vector<T>& xs, const cow_vector<T>& ys)
-> decltype(xs.std() == ys.std()) {
return xs.std() == ys.std();
}
template <class T>
auto operator==(const cow_vector<T>& xs, const std::vector<T>& ys)
-> decltype(xs.std() == ys) {
return xs.std() == ys;
}
template <class T>
auto operator==(const std::vector<T>& xs, const cow_vector<T>& ys)
-> decltype(xs == ys.std()) {
return xs.std() == ys;
}
template <class T>
auto operator!=(const cow_vector<T>& xs, const cow_vector<T>& ys)
-> decltype(xs.std() != ys.std()) {
return xs.std() != ys.std();
}
template <class T>
auto operator!=(const cow_vector<T>& xs, const std::vector<T>& ys)
-> decltype(xs.std() != ys) {
return xs.std() != ys;
}
template <class T>
auto operator!=(const std::vector<T>& xs, const cow_vector<T>& ys)
-> decltype(xs != ys.std()) {
return xs.std() != ys;
}
} // namespace caf
...@@ -38,17 +38,29 @@ class observer; ...@@ -38,17 +38,29 @@ class observer;
template <class T> template <class T>
class observable; class observable;
template <class T> template <class Materializer, class... Steps>
class connectable; class observable_def;
template <class In, class Out> template <class Generator>
class processor; class generation_materializer;
/// A blueprint for an @ref observer that generates items and applies any number
/// of processing steps immediately before emitting them.
template <class Generator, class... Steps> template <class Generator, class... Steps>
class generation; using generation = observable_def<generation_materializer<Generator>, Steps...>;
template <class Input>
class transformation_materializer;
/// A blueprint for an @ref observer that applies a series of transformation
/// steps to its inputs and emits the results.
template <class Step, class... Steps> template <class Step, class... Steps>
class transformation; using transformation
= observable_def<transformation_materializer<typename Step::input_type>, Step,
Steps...>;
template <class T>
class connectable;
template <class T> template <class T>
struct is_observable { struct is_observable {
...@@ -60,18 +72,8 @@ struct is_observable<observable<T>> { ...@@ -60,18 +72,8 @@ struct is_observable<observable<T>> {
static constexpr bool value = true; static constexpr bool value = true;
}; };
template <class Step, class... Steps> template <class Materializer, class... Steps>
struct is_observable<transformation<Step, Steps...>> { struct is_observable<observable_def<Materializer, Steps...>> {
static constexpr bool value = true;
};
template <class Generator, class... Steps>
struct is_observable<generation<Generator, Steps...>> {
static constexpr bool value = true;
};
template <class In, class Out>
struct is_observable<processor<In, Out>> {
static constexpr bool value = true; static constexpr bool value = true;
}; };
...@@ -83,29 +85,6 @@ struct is_observable<single<T>> { ...@@ -83,29 +85,6 @@ struct is_observable<single<T>> {
template <class T> template <class T>
constexpr bool is_observable_v = is_observable<T>::value; constexpr bool is_observable_v = is_observable<T>::value;
template <class T>
struct is_observer {
static constexpr bool value = false;
};
template <class T>
struct is_observer<observer<T>> {
static constexpr bool value = true;
};
template <class Step, class... Steps>
struct is_observer<transformation<Step, Steps...>> {
static constexpr bool value = true;
};
template <class In, class Out>
struct is_observer<processor<In, Out>> {
static constexpr bool value = true;
};
template <class T>
constexpr bool is_observer_v = is_observer<T>::value;
class observable_builder; class observable_builder;
template <class T> template <class T>
...@@ -116,13 +95,19 @@ struct input_type_oracle { ...@@ -116,13 +95,19 @@ struct input_type_oracle {
template <class T> template <class T>
using input_type_t = typename input_type_oracle<T>::type; using input_type_t = typename input_type_oracle<T>::type;
template <class...>
struct output_type_oracle;
template <class T> template <class T>
struct output_type_oracle { struct output_type_oracle<T> {
using type = typename T::output_type; using type = typename T::output_type;
}; };
template <class T> template <class T0, class T1, class... Ts>
using output_type_t = typename output_type_oracle<T>::type; struct output_type_oracle<T0, T1, Ts...> : output_type_oracle<T1, Ts...> {};
template <class... Ts>
using output_type_t = typename output_type_oracle<Ts...>::type;
template <class> template <class>
struct has_impl_include { struct has_impl_include {
......
// 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
namespace caf::flow::gen {
/// A generator that emits nothing and calls `on_complete` immediately.
template <class T>
class empty {
public:
using output_type = T;
template <class Step, class... Steps>
void pull(size_t, Step& step, Steps&... steps) {
step.on_complete(steps...);
}
};
} // namespace caf::flow::gen
// 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/detail/type_traits.hpp"
#include <type_traits>
namespace caf::flow::gen {
/// A generator that emits values from a function object.
template <class F>
class from_callable {
public:
using callable_res_t = std::invoke_result_t<F>;
static constexpr bool boxed_output = detail::is_optional_v<callable_res_t>;
using output_type = detail::unboxed_t<callable_res_t>;
explicit from_callable(F fn) : fn_(std::move(fn)) {
// nop
}
from_callable(from_callable&&) = default;
from_callable(const from_callable&) = default;
from_callable& operator=(from_callable&&) = default;
from_callable& operator=(const from_callable&) = default;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
for (size_t i = 0; i < n; ++i) {
if constexpr (boxed_output) {
auto val = fn_();
if (!val) {
step.on_complete(steps...);
return;
} else if (!step.on_next(*val, steps...))
return;
} else {
if (!step.on_next(fn_(), steps...))
return;
}
}
}
private:
F fn_;
};
} // namespace caf::flow::gen
// 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 <memory>
namespace caf::flow::gen {
/// A generator that emits values from a vector.
template <class Container>
class from_container {
public:
using output_type = typename Container::value_type;
explicit from_container(Container&& values) {
values_ = std::make_shared<Container>(std::move(values));
pos_ = values_->begin();
}
from_container() = default;
from_container(from_container&&) = default;
from_container(const from_container&) = default;
from_container& operator=(from_container&&) = default;
from_container& operator=(const from_container&) = default;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
auto end = values_->end();
while (pos_ != end && n > 0) {
if (!step.on_next(*pos_++, steps...))
return;
--n;
}
if (pos_ == end)
step.on_complete(steps...);
}
private:
std::shared_ptr<Container> values_;
typename Container::const_iterator pos_;
};
} // namespace caf::flow::gen
// 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
namespace caf::flow::gen {
/// A generator that emits ascending values.
template <class T>
class iota {
public:
using output_type = T;
explicit iota(T init) : value_(std::move(init)) {
// nop
}
iota(iota&&) = default;
iota(const iota&) = default;
iota& operator=(iota&&) = default;
iota& operator=(const iota&) = default;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
for (size_t i = 0; i < n; ++i)
if (!step.on_next(value_++, steps...))
return;
}
private:
T value_;
};
} // namespace caf::flow::gen
// 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
namespace caf::flow::gen {
/// A generator that emits a single value once.
template <class T>
class just {
public:
using output_type = T;
explicit just(T value) : value_(std::move(value)) {
// nop
}
just(just&&) = default;
just(const just&) = default;
just& operator=(just&&) = default;
just& operator=(const just&) = default;
template <class Step, class... Steps>
void pull([[maybe_unused]] size_t n, Step& step, Steps&... steps) {
if (step.on_next(value_, steps...))
step.on_complete(steps...);
}
private:
T value_;
};
} // namespace caf::flow::gen
// 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
namespace caf::flow::gen {
/// A generator that emits the same value repeatedly.
template <class T>
class repeat {
public:
using output_type = T;
explicit repeat(T value) : value_(std::move(value)) {
// nop
}
repeat(repeat&&) = default;
repeat(const repeat&) = default;
repeat& operator=(repeat&&) = default;
repeat& operator=(const repeat&) = default;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
for (size_t i = 0; i < n; ++i)
if (!step.on_next(value_, steps...))
return;
}
private:
T value_;
};
} // namespace caf::flow::gen
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#include "caf/async/producer.hpp" #include "caf/async/producer.hpp"
#include "caf/async/spsc_buffer.hpp" #include "caf/async/spsc_buffer.hpp"
#include "caf/cow_tuple.hpp" #include "caf/cow_tuple.hpp"
#include "caf/cow_vector.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp" #include "caf/detail/unordered_flat_map.hpp"
...@@ -23,9 +24,9 @@ ...@@ -23,9 +24,9 @@
#include "caf/flow/op/from_resource.hpp" #include "caf/flow/op/from_resource.hpp"
#include "caf/flow/op/from_steps.hpp" #include "caf/flow/op/from_steps.hpp"
#include "caf/flow/op/merge.hpp" #include "caf/flow/op/merge.hpp"
#include "caf/flow/op/prefix_and_tail.hpp" #include "caf/flow/op/prefetch.hpp"
#include "caf/flow/op/publish.hpp" #include "caf/flow/op/publish.hpp"
#include "caf/flow/step.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"
#include "caf/logger.hpp" #include "caf/logger.hpp"
...@@ -41,103 +42,6 @@ ...@@ -41,103 +42,6 @@
namespace caf::flow { namespace caf::flow {
/// Base type for classes that represent a definition of an `observable` which
/// has not yet been converted to an actual `observable`.
template <class T>
class observable_def {
public:
virtual ~observable_def() = default;
template <class OnNext>
auto for_each(OnNext on_next) && {
return alloc().for_each(std::move(on_next));
}
template <class OnNext, class OnError>
auto for_each(OnNext on_next, OnError on_error) && {
return alloc().for_each(std::move(on_next), std::move(on_error));
}
template <class OnNext, class OnError, class OnComplete>
auto for_each(OnNext on_next, OnError on_error, OnComplete on_complete) && {
return alloc().for_each(std::move(on_next), std::move(on_error),
std::move(on_complete));
}
template <class... Inputs>
auto merge(Inputs&&... xs) && {
return alloc().as_observable().merge(std::forward<Inputs>(xs)...);
}
template <class... Inputs>
auto concat(Inputs&&... xs) && {
return alloc().as_observable().concat(std::forward<Inputs>(xs)...);
}
template <class F>
auto flat_map(F f) && {
return alloc().flat_map(std::move(f));
}
template <class F>
auto concat_map(F f) && {
return alloc().concat_map(std::move(f));
}
/// @copydoc observable::publish
auto publish() && {
return alloc().publish();
}
/// @copydoc observable::share
auto share(size_t subscriber_threshold = 1) && {
return alloc().share(subscriber_threshold);
}
observable<cow_tuple<std::vector<T>, observable<T>>>
prefix_and_tail(size_t prefix_size) && {
return alloc().prefix_and_tail(prefix_size);
}
observable<cow_tuple<T, observable<T>>> head_and_tail() && {
return alloc().head_and_tail();
}
disposable subscribe(observer<T> what) && {
return alloc().subscribe(std::move(what));
}
disposable subscribe(async::producer_resource<T> resource) && {
return alloc().subscribe(std::move(resource));
}
async::consumer_resource<T> to_resource() && {
return alloc().to_resource();
}
async::consumer_resource<T> to_resource(size_t buffer_size,
size_t min_request_size) && {
return alloc().to_resource(buffer_size, min_request_size);
}
observable<T> observe_on(coordinator* other) && {
return alloc().observe_on(other);
}
observable<T> observe_on(coordinator* other, size_t buffer_size,
size_t min_request_size) && {
return alloc().observe_on(other, buffer_size, min_request_size);
}
virtual observable<T> as_observable() && = 0;
private:
/// Allocates and returns an actual @ref observable.
decltype(auto) alloc() {
return std::move(*this).as_observable();
}
};
// -- connectable -------------------------------------------------------------- // -- connectable --------------------------------------------------------------
/// Resembles a regular @ref observable, except that it does not begin emitting /// Resembles a regular @ref observable, except that it does not begin emitting
...@@ -221,12 +125,6 @@ public: ...@@ -221,12 +125,6 @@ public:
return pimpl_->connect(); return pimpl_->connect();
} }
/// @copydoc observable::compose
template <class Fn>
auto compose(Fn&& fn) & {
return fn(*this);
}
/// @copydoc observable::compose /// @copydoc observable::compose
template <class Fn> template <class Fn>
auto compose(Fn&& fn) && { auto compose(Fn&& fn) && {
...@@ -275,37 +173,34 @@ private: ...@@ -275,37 +173,34 @@ private:
pimpl_type pimpl_; pimpl_type pimpl_;
}; };
// -- transformation ----------------------------------------------------------- /// Captures the *definition* of an observable that has not materialized yet.
template <class Materializer, class... Steps>
/// A special type of observer that applies a series of transformation steps to class observable_def {
/// its input before broadcasting the result as output.
template <class Step, class... Steps>
class transformation final
: public observable_def<steps_output_type_t<Step, Steps...>> {
public: public:
using input_type = typename Step::input_type; using output_type = output_type_t<Materializer, Steps...>;
observable_def() = delete;
observable_def(const observable_def&) = delete;
observable_def& operator=(const observable_def&) = delete;
using output_type = steps_output_type_t<Step, Steps...>; observable_def(observable_def&&) = default;
observable_def& operator=(observable_def&&) = default;
template <class Tuple> template <size_t N = sizeof...(Steps), class = std::enable_if_t<N == 0>>
transformation(observable<input_type> source, Tuple&& steps) explicit observable_def(Materializer&& materializer)
: source_(std::move(source)), steps_(std::move(steps)) { : materializer_(std::move(materializer)) {
// nop // nop
} }
transformation() = delete; observable_def(Materializer&& materializer, std::tuple<Steps...>&& steps)
transformation(const transformation&) = delete; : materializer_(std::move(materializer)), steps_(std::move(steps)) {
transformation& operator=(const transformation&) = delete; // nop
}
transformation(transformation&&) = default;
transformation& operator=(transformation&&) = default;
/// @copydoc observable::transform /// @copydoc observable::transform
template <class NewStep> template <class NewStep>
transformation<Step, Steps..., NewStep> transform(NewStep step) && { observable_def<Materializer, Steps..., NewStep> transform(NewStep step) && {
return {std::move(source_), return add_step(std::move(step));
std::tuple_cat(std::move(steps_),
std::make_tuple(std::move(step)))};
} }
/// @copydoc observable::compose /// @copydoc observable::compose
...@@ -314,175 +209,356 @@ public: ...@@ -314,175 +209,356 @@ public:
return fn(std::move(*this)); return fn(std::move(*this));
} }
/// @copydoc observable::skip
auto skip(size_t n) && {
return add_step(step::skip<output_type>{n});
}
/// @copydoc observable::take
auto take(size_t n) && { auto take(size_t n) && {
return std::move(*this).transform(limit_step<output_type>{n}); return add_step(step::take<output_type>{n});
} }
template <class Predicate> template <class Predicate>
auto filter(Predicate predicate) && { auto filter(Predicate predicate) && {
return std::move(*this).transform( return add_step(step::filter<Predicate>{std::move(predicate)});
filter_step<Predicate>{std::move(predicate)});
} }
template <class Predicate> template <class Predicate>
auto take_while(Predicate predicate) && { auto take_while(Predicate predicate) && {
return std::move(*this).transform( return add_step(step::take_while<Predicate>{std::move(predicate)});
take_while_step<Predicate>{std::move(predicate)});
} }
template <class Reducer> template <class Init, class Reducer>
auto reduce(output_type init, Reducer reducer) && { auto reduce(Init init, Reducer reducer) && {
return std::move(*this).transform( using val_t = output_type;
reduce_step<output_type, Reducer>{init, reducer}); static_assert(std::is_invocable_r_v<Init, Reducer, Init&&, const val_t&>);
return add_step(step::reduce<Reducer>{std::move(init), std::move(reducer)});
} }
auto sum() && { auto sum() && {
return std::move(*this).reduce(output_type{}, std::plus<output_type>{}); return std::move(*this).reduce(output_type{}, std::plus<output_type>{});
} }
auto to_vector() && {
using vector_type = cow_vector<output_type>;
auto append = [](vector_type&& xs, const output_type& x) {
xs.unshared().push_back(x);
return xs;
};
return std::move(*this)
.reduce(vector_type{}, append)
.filter([](const vector_type& xs) { return !xs.empty(); });
}
auto distinct() && { auto distinct() && {
return std::move(*this).transform(distinct_step<output_type>{}); return add_step(step::distinct<output_type>{});
} }
template <class F> template <class F>
auto map(F f) && { auto map(F f) && {
return std::move(*this).transform(map_step<F>{std::move(f)}); return add_step(step::map<F>{std::move(f)});
} }
template <class F> template <class F>
auto do_on_next(F f) && { auto do_on_next(F f) && {
return std::move(*this) // return add_step(step::do_on_next<F>{std::move(f)});
.transform(do_on_next_step<output_type, F>{std::move(f)});
} }
template <class F> template <class F>
auto do_on_complete(F f) && { auto do_on_complete(F f) && {
return std::move(*this) // return add_step(step::do_on_complete<output_type, F>{std::move(f)});
.transform(do_on_complete_step<output_type, F>{std::move(f)});
} }
template <class F> template <class F>
auto do_on_error(F f) && { auto do_on_error(F f) && {
return std::move(*this) // return add_step(step::do_on_error<output_type, F>{std::move(f)});
.transform(do_on_error_step<output_type, F>{std::move(f)});
} }
template <class F> template <class F>
auto do_finally(F f) && { auto do_finally(F f) && {
return std::move(*this) // return add_step(step::do_finally<output_type, F>{std::move(f)});
.transform(do_finally_step<output_type, F>{std::move(f)});
} }
auto on_error_complete() { auto on_error_complete() {
return std::move(*this) // return add_step(step::on_error_complete<output_type>{});
.transform(on_error_complete_step<output_type>{});
} }
/// Materializes the @ref observable.
observable<output_type> as_observable() && { observable<output_type> as_observable() && {
using impl_t = op::from_steps<input_type, Step, Steps...>; return materialize();
return make_observable<impl_t>(source_.ctx(), source_.pimpl(), }
std::move(steps_));
/// @copydoc observable::for_each
template <class OnNext>
auto for_each(OnNext on_next) && {
return materialize().for_each(std::move(on_next));
}
/// @copydoc observable::merge
template <class... Inputs>
auto merge(Inputs&&... xs) && {
return materialize().merge(std::forward<Inputs>(xs)...);
}
/// @copydoc observable::concat
template <class... Inputs>
auto concat(Inputs&&... xs) && {
return materialize().concat(std::forward<Inputs>(xs)...);
}
/// @copydoc observable::flat_map
template <class F>
auto flat_map(F f) && {
return materialize().flat_map(std::move(f));
}
/// @copydoc observable::concat_map
template <class F>
auto concat_map(F f) && {
return materialize().concat_map(std::move(f));
}
/// @copydoc observable::publish
auto publish() && {
return materialize().publish();
}
/// @copydoc observable::share
auto share(size_t subscriber_threshold = 1) && {
return materialize().share(subscriber_threshold);
}
/// @copydoc observable::prefix_and_tail
observable<cow_tuple<cow_vector<output_type>, observable<output_type>>>
prefix_and_tail(size_t prefix_size) && {
return materialize().prefix_and_tail(prefix_size);
}
/// @copydoc observable::head_and_tail
observable<cow_tuple<output_type, observable<output_type>>>
head_and_tail() && {
return materialize().head_and_tail();
}
/// @copydoc observable::subscribe
template <class Out>
disposable subscribe(Out&& out) && {
return materialize().subscribe(std::forward<Out>(out));
}
/// @copydoc observable::to_resource
async::consumer_resource<output_type> to_resource() && {
return materialize().to_resource();
}
/// @copydoc observable::to_resource
async::consumer_resource<output_type>
to_resource(size_t buffer_size, size_t min_request_size) && {
return materialize().to_resource(buffer_size, min_request_size);
}
/// @copydoc observable::observe_on
observable<output_type> observe_on(coordinator* other) && {
return materialize().observe_on(other);
}
/// @copydoc observable::observe_on
observable<output_type> observe_on(coordinator* other, size_t buffer_size,
size_t min_request_size) && {
return materialize().observe_on(other, buffer_size, min_request_size);
}
bool valid() const noexcept {
return materializer_.valid();
} }
private: private:
observable<input_type> source_; template <class NewStep>
std::tuple<Step, Steps...> steps_; observable_def<Materializer, Steps..., NewStep> add_step(NewStep step) {
static_assert(std::is_same_v<output_type, typename NewStep::input_type>);
return {std::move(materializer_),
std::tuple_cat(std::move(steps_),
std::make_tuple(std::move(step)))};
}
observable<output_type> materialize() {
return std::move(materializer_).materialize(std::move(steps_));
}
/// Encapsulates logic for allocating a flow operator.
Materializer materializer_;
/// Stores processing steps that the materializer fuses into a single flow
/// operator.
std::tuple<Steps...> steps_;
}; };
// -- observable::transform ---------------------------------------------------- // -- transformation -----------------------------------------------------------
/// Materializes an @ref observable from a source @ref observable and one or
/// more processing steps.
template <class Input>
class transformation_materializer {
public:
using output_type = Input;
explicit transformation_materializer(observable<Input> source)
: source_(std::move(source).pimpl()) {
// nop
}
explicit transformation_materializer(intrusive_ptr<op::base<Input>> source)
: source_(std::move(source)) {
// nop
}
transformation_materializer() = delete;
transformation_materializer(const transformation_materializer&) = delete;
transformation_materializer& operator=(const transformation_materializer&)
= delete;
transformation_materializer(transformation_materializer&&) = default;
transformation_materializer& operator=(transformation_materializer&&)
= default;
bool valid() const noexcept {
return source_ != nullptr;
}
coordinator* ctx() {
return source_->ctx();
}
template <class Step, class... Steps>
auto materialize(std::tuple<Step, Steps...>&& steps) && {
using impl_t = op::from_steps<Input, Step, Steps...>;
return make_observable<impl_t>(ctx(), source_, std::move(steps));
}
private:
intrusive_ptr<op::base<Input>> source_;
};
// -- observable: subscribing --------------------------------------------------
template <class T>
disposable observable<T>::subscribe(observer<T> what) {
if (pimpl_) {
return pimpl_->subscribe(std::move(what));
} else {
what.on_error(make_error(sec::invalid_observable));
return disposable{};
}
}
template <class T>
disposable observable<T>::subscribe(async::producer_resource<T> resource) {
using buffer_type = typename async::consumer_resource<T>::buffer_type;
using adapter_type = buffer_writer_impl<buffer_type>;
if (auto buf = resource.try_open()) {
CAF_LOG_DEBUG("subscribe producer resource to flow");
auto adapter = make_counted<adapter_type>(pimpl_->ctx(), buf);
buf->set_producer(adapter);
auto obs = adapter->as_observer();
auto sub = subscribe(std::move(obs));
pimpl_->ctx()->watch(sub);
return sub;
} else {
CAF_LOG_DEBUG("failed to open producer resource");
return {};
}
}
template <class T>
template <class OnNext>
disposable observable<T>::for_each(OnNext on_next) {
return subscribe(make_observer(std::move(on_next)));
}
// -- observable: transforming -------------------------------------------------
template <class T> template <class T>
template <class Step> template <class Step>
transformation<Step> observable<T>::transform(Step step) { transformation<Step> observable<T>::transform(Step step) {
static_assert(std::is_same_v<typename Step::input_type, T>, static_assert(std::is_same_v<typename Step::input_type, T>,
"step object does not match the input type"); "step object does not match the input type");
return {*this, std::forward_as_tuple(std::move(step))}; return {transformation_materializer<T>{pimpl()}, std::move(step)};
} }
// -- observable::take --------------------------------------------------------- template <class T>
transformation<step::distinct<T>> observable<T>::distinct() {
return transform(step::distinct<T>{});
}
template <class T> template <class T>
transformation<limit_step<T>> observable<T>::take(size_t n) { template <class F>
return {*this, std::forward_as_tuple(limit_step<T>{n})}; transformation<step::do_finally<T, F>> observable<T>::do_finally(F fn) {
return transform(step::do_finally<T, F>{std::move(fn)});
} }
// -- observable::filter ------------------------------------------------------- template <class T>
template <class F>
transformation<step::do_on_complete<T, F>> observable<T>::do_on_complete(F fn) {
return transform(step::do_on_complete<T, F>{std::move(fn)});
}
template <class T> template <class T>
template <class Predicate> template <class F>
transformation<filter_step<Predicate>> transformation<step::do_on_error<T, F>> observable<T>::do_on_error(F fn) {
observable<T>::filter(Predicate predicate) { return transform(step::do_on_error<T, F>{std::move(fn)});
using step_type = filter_step<Predicate>;
static_assert(std::is_same_v<typename step_type::input_type, T>,
"predicate does not match the input type");
return {*this, std::forward_as_tuple(step_type{std::move(predicate)})};
} }
// -- observable::take_while --------------------------------------------------- template <class T>
template <class F>
transformation<step::do_on_next<F>> observable<T>::do_on_next(F fn) {
return transform(step::do_on_next<F>{std::move(fn)});
}
template <class T> template <class T>
template <class Predicate> template <class Predicate>
transformation<take_while_step<Predicate>> transformation<step::filter<Predicate>>
observable<T>::take_while(Predicate predicate) { observable<T>::filter(Predicate predicate) {
using step_type = take_while_step<Predicate>; return transform(step::filter{std::move(predicate)});
static_assert(std::is_same_v<typename step_type::input_type, T>,
"predicate does not match the input type");
return {*this, std::forward_as_tuple(step_type{std::move(predicate)})};
} }
// -- observable::sum ----------------------------------------------------------
template <class T> template <class T>
template <class Reducer> template <class F>
transformation<reduce_step<T, Reducer>> transformation<step::map<F>> observable<T>::map(F f) {
observable<T>::reduce(T init, Reducer reducer) { return transform(step::map(std::move(f)));
return {*this, reduce_step<T, Reducer>{init, reducer}};
} }
// -- observable::distinct -----------------------------------------------------
template <class T> template <class T>
transformation<distinct_step<T>> observable<T>::distinct() { transformation<step::on_error_complete<T>> observable<T>::on_error_complete() {
return {*this, distinct_step<T>{}}; return transform(step::on_error_complete<T>{});
} }
// -- observable::map ----------------------------------------------------------
template <class T> template <class T>
template <class F> template <class Init, class Reducer>
transformation<map_step<F>> observable<T>::map(F f) { transformation<step::reduce<Reducer>>
using step_type = map_step<F>; observable<T>::reduce(Init init, Reducer reducer) {
static_assert(std::is_same_v<typename step_type::input_type, T>, static_assert(std::is_invocable_r_v<Init, Reducer, Init&&, const T&>);
"map function does not match the input type"); return transform(step::reduce<Reducer>{std::move(init), std::move(reducer)});
return {*this, std::forward_as_tuple(step_type{std::move(f)})};
} }
// -- observable::for_each -----------------------------------------------------
template <class T> template <class T>
template <class OnNext> transformation<step::skip<T>> observable<T>::skip(size_t n) {
disposable observable<T>::for_each(OnNext on_next) { return transform(step::skip<T>{n});
auto obs = make_observer(std::move(on_next));
return subscribe(std::move(obs));
} }
template <class T> template <class T>
template <class OnNext, class OnError> transformation<step::take<T>> observable<T>::take(size_t n) {
disposable observable<T>::for_each(OnNext on_next, OnError on_error) { return transform(step::take<T>{n});
auto obs = make_observer(std::move(on_next), std::move(on_error));
return subscribe(std::move(obs));
} }
template <class T> template <class T>
template <class OnNext, class OnError, class OnComplete> template <class Predicate>
disposable observable<T>::for_each(OnNext on_next, OnError on_error, transformation<step::take_while<Predicate>>
OnComplete on_complete) { observable<T>::take_while(Predicate predicate) {
auto obs = make_observer(std::move(on_next), std::move(on_error), return transform(step::take_while{std::move(predicate)});
std::move(on_complete));
return subscribe(std::move(obs));
} }
// -- observable::merge -------------------------------------------------------- // -- observable: combining ----------------------------------------------------
template <class T> template <class T>
template <class Out, class... Inputs> template <class Out, class... Inputs>
...@@ -500,7 +576,21 @@ auto observable<T>::merge(Inputs&&... xs) { ...@@ -500,7 +576,21 @@ auto observable<T>::merge(Inputs&&... xs) {
} }
} }
// -- observable::flat_map ----------------------------------------------------- template <class T>
template <class Out, class... Inputs>
auto observable<T>::concat(Inputs&&... xs) {
if constexpr (is_observable_v<Out>) {
using value_t = output_type_t<Out>;
using impl_t = op::concat<value_t>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
} else {
static_assert(
sizeof...(Inputs) > 0,
"merge without arguments expects this observable to emit observables");
using impl_t = op::concat<Out>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
}
}
template <class T> template <class T>
template <class Out, class F> template <class Out, class F>
...@@ -528,26 +618,6 @@ auto observable<T>::flat_map(F f) { ...@@ -528,26 +618,6 @@ auto observable<T>::flat_map(F f) {
} }
} }
// -- observable::concat -------------------------------------------------------
template <class T>
template <class Out, class... Inputs>
auto observable<T>::concat(Inputs&&... xs) {
if constexpr (is_observable_v<Out>) {
using value_t = output_type_t<Out>;
using impl_t = op::concat<value_t>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
} else {
static_assert(
sizeof...(Inputs) > 0,
"merge without arguments expects this observable to emit observables");
using impl_t = op::concat<Out>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
}
}
// -- observable::concat_map ---------------------------------------------------
template <class T> template <class T>
template <class Out, class F> template <class Out, class F>
auto observable<T>::concat_map(F f) { auto observable<T>::concat_map(F f) {
...@@ -570,59 +640,54 @@ auto observable<T>::concat_map(F f) { ...@@ -570,59 +640,54 @@ auto observable<T>::concat_map(F f) {
} }
} }
// -- observable::prefix_and_tail ---------------------------------------------- // -- observable: splitting ----------------------------------------------------
template <class T> template <class T>
observable<cow_tuple<std::vector<T>, observable<T>>> observable<cow_tuple<cow_vector<T>, observable<T>>>
observable<T>::prefix_and_tail(size_t prefix_size) { observable<T>::prefix_and_tail(size_t n) {
using impl_t = op::prefix_and_tail<T>; using vector_t = cow_vector<T>;
return make_observable<impl_t>(ctx(), pimpl_, prefix_size); CAF_ASSERT(n > 0);
auto do_prefetch = [](auto in) {
auto ptr = op::prefetch<T>::apply(std::move(in).as_observable().pimpl());
return observable<T>{std::move(ptr)};
};
auto split = share(2);
auto tail = split.skip(n).compose(do_prefetch);
return split //
.take(n)
.to_vector()
.filter([n](const vector_t& xs) { return xs.size() == n; })
.map([tail](const vector_t& xs) { return make_cow_tuple(xs, tail); })
.as_observable();
} }
// -- observable::prefix_and_tail ----------------------------------------------
template <class T> template <class T>
observable<cow_tuple<T, observable<T>>> observable<T>::head_and_tail() { observable<cow_tuple<T, observable<T>>> observable<T>::head_and_tail() {
using tuple_t = cow_tuple<std::vector<T>, observable<T>>; auto do_prefetch = [](auto in) {
return prefix_and_tail(1) auto ptr = op::prefetch<T>::apply(std::move(in).as_observable().pimpl());
.map([](const tuple_t& tup) { return observable<T>{std::move(ptr)};
auto& [prefix, tail] = tup.data(); };
CAF_ASSERT(prefix.size() == 1); auto split = share(2);
return make_cow_tuple(prefix.front(), tail); auto tail = split.skip(1).compose(do_prefetch);
}) return split //
.take(1)
.map([tail](const T& x) { return make_cow_tuple(x, tail); })
.as_observable(); .as_observable();
} }
// -- observable::publish ------------------------------------------------------ // -- observable: multicasting -------------------------------------------------
template <class T> template <class T>
connectable<T> observable<T>::publish() { connectable<T> observable<T>::publish() {
return connectable<T>{make_counted<op::publish<T>>(ctx(), pimpl_)}; return connectable<T>{make_counted<op::publish<T>>(ctx(), pimpl_)};
} }
// -- observable::share --------------------------------------------------------
template <class T> template <class T>
observable<T> observable<T>::share(size_t subscriber_threshold) { observable<T> observable<T>::share(size_t subscriber_threshold) {
return publish().ref_count(subscriber_threshold); return publish().ref_count(subscriber_threshold);
} }
// -- observable::to_resource -------------------------------------------------- // -- observable: observing ----------------------------------------------------
/// Reads from an observable buffer and emits the consumed items.
/// @note Only supports a single observer.
template <class T>
async::consumer_resource<T>
observable<T>::to_resource(size_t buffer_size, size_t min_request_size) {
using buffer_type = async::spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf);
buf->set_producer(up);
subscribe(up->as_observer());
return async::consumer_resource<T>{std::move(buf)};
}
// -- observable::observe_on ---------------------------------------------------
template <class T> template <class T>
observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size, observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size,
...@@ -633,24 +698,17 @@ observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size, ...@@ -633,24 +698,17 @@ observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size,
return make_observable<op::from_resource<T>>(other, std::move(pull)); return make_observable<op::from_resource<T>>(other, std::move(pull));
} }
// -- observable::subscribe ---------------------------------------------------- // -- observable: converting ---------------------------------------------------
template <class T> template <class T>
disposable observable<T>::subscribe(async::producer_resource<T> resource) { async::consumer_resource<T>
using buffer_type = typename async::consumer_resource<T>::buffer_type; observable<T>::to_resource(size_t buffer_size, size_t min_request_size) {
using adapter_type = buffer_writer_impl<buffer_type>; using buffer_type = async::spsc_buffer<T>;
if (auto buf = resource.try_open()) { auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
CAF_LOG_DEBUG("subscribe producer resource to flow"); auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf);
auto adapter = make_counted<adapter_type>(pimpl_->ctx(), buf); buf->set_producer(up);
buf->set_producer(adapter); subscribe(up->as_observer());
auto obs = adapter->as_observer(); return async::consumer_resource<T>{std::move(buf)};
auto sub = subscribe(std::move(obs));
pimpl_->ctx()->watch(sub);
return sub;
} else {
CAF_LOG_DEBUG("failed to open producer resource");
return {};
}
} }
} // namespace caf::flow } // namespace caf::flow
...@@ -8,6 +8,12 @@ ...@@ -8,6 +8,12 @@
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/gen/empty.hpp"
#include "caf/flow/gen/from_callable.hpp"
#include "caf/flow/gen/from_container.hpp"
#include "caf/flow/gen/iota.hpp"
#include "caf/flow/gen/just.hpp"
#include "caf/flow/gen/repeat.hpp"
#include "caf/flow/observable.hpp" #include "caf/flow/observable.hpp"
#include "caf/flow/op/defer.hpp" #include "caf/flow/op/defer.hpp"
#include "caf/flow/op/empty.hpp" #include "caf/flow/op/empty.hpp"
...@@ -23,21 +29,42 @@ ...@@ -23,21 +29,42 @@
namespace caf::flow { namespace caf::flow {
// -- forward declarations ----------------------------------------------------- // -- generation ---------------------------------------------------------------
/// Materializes an @ref observable from a `Generator` that produces items and
/// any number of processing steps that immediately transform the produced
/// items.
template <class Generator>
class generation_materializer {
public:
using output_type = typename Generator::output_type;
generation_materializer(coordinator* ctx, Generator generator)
: ctx_(ctx), gen_(std::move(generator)) {
// nop
}
template <class T> generation_materializer() = delete;
class repeater_source; generation_materializer(const generation_materializer&) = delete;
generation_materializer& operator=(const generation_materializer&) = delete;
template <class Container> generation_materializer(generation_materializer&&) = default;
class container_source; generation_materializer& operator=(generation_materializer&&) = default;
template <class T> template <class... Steps>
class value_source; auto materialize(std::tuple<Steps...>&& steps) && {
using impl_t = op::from_generator<Generator, Steps...>;
return make_observable<impl_t>(ctx_, std::move(gen_), std::move(steps));
}
template <class F> bool valid() const noexcept {
class callable_source; return ctx_ != nullptr;
}
// -- special-purpose observable implementations ------------------------------- private:
coordinator* ctx_;
Generator gen_;
};
// -- builder interface -------------------------------------------------------- // -- builder interface --------------------------------------------------------
...@@ -50,55 +77,72 @@ public: ...@@ -50,55 +77,72 @@ public:
observable_builder& operator=(const observable_builder&) noexcept = default; observable_builder& operator=(const observable_builder&) noexcept = default;
/// Creates a @ref generation that emits `value` indefinitely. /// Creates a @ref generation that emits values by repeatedly calling
template <class T> /// `generator.pull(...)`.
[[nodiscard]] generation<repeater_source<T>> repeat(T value) const; template <class Generator>
generation<Generator> from_generator(Generator generator) const {
/// Creates a @ref generation that emits all values from `values`. using materializer_t = generation_materializer<Generator>;
template <class Container> return generation<Generator>{materializer_t{ctx_, std::move(generator)}};
[[nodiscard]] generation<container_source<Container>> }
from_container(Container values) const;
/// Creates a @ref generation that emits `value` once. /// Creates a @ref generation that emits `value` once.
template <class T> template <class T>
[[nodiscard]] generation<value_source<T>> just(T value) const; generation<gen::just<T>> just(T value) const {
return from_generator(gen::just<T>{std::move(value)});
}
/// Creates a @ref generation that emits values by repeatedly calling `fn`. /// Creates a @ref generation that emits `value` repeatedly.
template <class F> template <class T>
[[nodiscard]] generation<callable_source<F>> from_callable(F fn) const; generation<gen::repeat<T>> repeat(T value) const {
return from_generator(gen::repeat<T>{std::move(value)});
}
/// Creates a @ref generation that emits ascending values. /// Creates a @ref generation that emits ascending values.
template <class T> template <class T>
[[nodiscard]] auto iota(T init) const { generation<gen::iota<T>> iota(T value) const {
return from_callable([x = std::move(init)]() mutable { return x++; }); return from_generator(gen::iota<T>{std::move(value)});
}
/// Creates an @ref observable without any values that calls `on_complete`
/// after subscribing to it.
template <class T>
generation<gen::empty<T>> empty() {
return from_generator(gen::empty<T>{});
}
/// Creates a @ref generation that emits ascending values.
template <class Container>
generation<gen::from_container<Container>>
from_container(Container values) const {
return from_generator(gen::from_container<Container>{std::move(values)});
}
/// Creates a @ref generation that emits ascending values.
template <class F>
generation<gen::from_callable<F>> from_callable(F fn) const {
return from_generator(gen::from_callable<F>{std::move(fn)});
} }
/// Creates a @ref generation that emits `num` ascending values, starting with /// Creates a @ref generation that emits `num` ascending values, starting with
/// `init`. /// `init`.
template <class T> template <class T>
[[nodiscard]] auto range(T init, size_t num) const { auto range(T init, size_t num) const {
return iota(init).take(num); return iota(init).take(num);
} }
/// Creates a @ref generation that emits values by repeatedly calling
/// `pullable.pull(...)`. For example implementations of the `Pullable`
/// concept, see @ref container_source, @ref repeater_source and
/// @ref callable_source.
template <class Pullable>
[[nodiscard]] generation<Pullable> lift(Pullable pullable) const;
/// Creates an @ref observable that reads and emits all values from `res`. /// Creates an @ref observable that reads and emits all values from `res`.
template <class T> template <class T>
[[nodiscard]] observable<T> observable<T> from_resource(async::consumer_resource<T> res) const {
from_resource(async::consumer_resource<T> res) const; using impl_t = op::from_resource<T>;
return make_observable<impl_t>(ctx_, std::move(res));
}
/// Creates an @ref observable that emits a sequence of integers spaced by the /// Creates an @ref observable that emits a sequence of integers spaced by the
/// @p period. /// @p period.
/// @param initial_delay Delay of the first integer after subscribing. /// @param initial_delay Delay of the first integer after subscribing.
/// @param period Delay of each consecutive integer after the first value. /// @param period Delay of each consecutive integer after the first value.
template <class Rep, class Period> template <class Rep, class Period>
[[nodiscard]] observable<int64_t> observable<int64_t> interval(std::chrono::duration<Rep, Period> initial_delay,
interval(std::chrono::duration<Rep, Period> initial_delay,
std::chrono::duration<Rep, Period> period) { std::chrono::duration<Rep, Period> period) {
return make_observable<op::interval>(ctx_, initial_delay, period); return make_observable<op::interval>(ctx_, initial_delay, period);
} }
...@@ -107,41 +151,32 @@ public: ...@@ -107,41 +151,32 @@ public:
/// @p delay. /// @p delay.
/// @param delay Time delay between two integer values. /// @param delay Time delay between two integer values.
template <class Rep, class Period> template <class Rep, class Period>
[[nodiscard]] observable<int64_t> observable<int64_t> interval(std::chrono::duration<Rep, Period> delay) {
interval(std::chrono::duration<Rep, Period> delay) {
return interval(delay, delay); return interval(delay, delay);
} }
/// Creates an @ref observable that emits a single item after the @p delay. /// Creates an @ref observable that emits a single item after the @p delay.
template <class Rep, class Period> template <class Rep, class Period>
[[nodiscard]] observable<int64_t> observable<int64_t> timer(std::chrono::duration<Rep, Period> delay) {
timer(std::chrono::duration<Rep, Period> delay) {
return make_observable<op::interval>(ctx_, delay, delay, 1); return make_observable<op::interval>(ctx_, delay, delay, 1);
} }
/// Creates an @ref observable without any values that calls `on_complete`
/// after subscribing to it.
template <class T>
[[nodiscard]] observable<T> empty() {
return make_observable<op::empty<T>>(ctx_);
}
/// Creates an @ref observable without any values that also never terminates. /// Creates an @ref observable without any values that also never terminates.
template <class T> template <class T>
[[nodiscard]] observable<T> never() { observable<T> never() {
return make_observable<op::never<T>>(ctx_); return make_observable<op::never<T>>(ctx_);
} }
/// Creates an @ref observable without any values that fails immediately when /// Creates an @ref observable without any values that fails immediately when
/// subscribing to it by calling `on_error` on the subscriber. /// subscribing to it by calling `on_error` on the subscriber.
template <class T> template <class T>
[[nodiscard]] observable<T> fail(error what) { observable<T> fail(error what) {
return make_observable<op::fail<T>>(ctx_, std::move(what)); return make_observable<op::fail<T>>(ctx_, std::move(what));
} }
/// Create a fresh @ref observable for each @ref observer using the factory. /// Create a fresh @ref observable for each @ref observer using the factory.
template <class Factory> template <class Factory>
[[nodiscard]] auto defer(Factory factory) { auto defer(Factory factory) {
return make_observable<op::defer<Factory>>(ctx_, std::move(factory)); return make_observable<op::defer<Factory>>(ctx_, std::move(factory));
} }
...@@ -218,313 +253,4 @@ private: ...@@ -218,313 +253,4 @@ private:
coordinator* ctx_; coordinator* ctx_;
}; };
// -- generation ---------------------------------------------------------------
/// Implements the `Pullable` concept for emitting values from a container.
template <class Container>
class container_source {
public:
using output_type = typename Container::value_type;
explicit container_source(Container&& values) : values_(std::move(values)) {
pos_ = values_.begin();
}
container_source() = default;
container_source(container_source&&) = default;
container_source& operator=(container_source&&) = default;
container_source(const container_source& other) : values_(other.values_) {
pos_ = values_.begin();
std::advance(pos_, std::distance(other.values_.begin(), other.pos_));
}
container_source& operator=(const container_source& other) {
values_ = other.values_;
pos_ = values_.begin();
std::advance(pos_, std::distance(other.values_.begin(), other.pos_));
return *this;
}
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;
--n;
}
if (pos_ == values_.end())
step.on_complete(steps...);
}
private:
Container values_;
typename Container::const_iterator pos_;
};
/// Implements the `Pullable` concept for emitting the same value repeatedly.
template <class T>
class repeater_source {
public:
using output_type = T;
explicit repeater_source(T value) : value_(std::move(value)) {
// nop
}
repeater_source(repeater_source&&) = default;
repeater_source(const repeater_source&) = default;
repeater_source& operator=(repeater_source&&) = default;
repeater_source& operator=(const repeater_source&) = default;
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;
}
private:
T value_;
};
/// Implements the `Pullable` concept for emitting the same value once.
template <class T>
class value_source {
public:
using output_type = T;
explicit value_source(T value) : value_(std::move(value)) {
// nop
}
value_source(value_source&&) = default;
value_source(const value_source&) = default;
value_source& operator=(value_source&&) = default;
value_source& operator=(const value_source&) = default;
template <class Step, class... 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...);
}
private:
T value_;
};
/// Implements the `Pullable` concept for emitting values generated from a
/// function object.
template <class F>
class callable_source {
public:
using callable_res_t = std::decay_t<decltype(std::declval<F&>()())>;
static constexpr bool boxed_output = detail::is_optional_v<callable_res_t>;
using output_type = detail::unboxed_t<callable_res_t>;
explicit callable_source(F fn) : fn_(std::move(fn)) {
// nop
}
callable_source(callable_source&&) = default;
callable_source(const callable_source&) = default;
callable_source& operator=(callable_source&&) = default;
callable_source& operator=(const callable_source&) = default;
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 constexpr (boxed_output) {
auto val = fn_();
if (!val) {
step.on_complete(steps...);
return;
} else if (!step.on_next(*val, steps...))
return;
} else {
if (!step.on_next(fn_(), steps...))
return;
}
}
}
private:
F fn_;
};
/// Helper class for combining multiple generation and transformation steps into
/// a single @ref observable object.
template <class Generator, class... Steps>
class generation final
: public observable_def<steps_output_type_t<Generator, Steps...>> {
public:
using output_type = steps_output_type_t<Generator, Steps...>;
template <class... Ts>
generation(coordinator* ctx, Generator gen, Ts&&... steps)
: ctx_(ctx), gen_(std::move(gen)), steps_(std::forward<Ts>(steps)...) {
// nop
}
generation() = delete;
generation(const generation&) = delete;
generation& operator=(const generation&) = delete;
generation(generation&&) = default;
generation& operator=(generation&&) = default;
/// @copydoc observable::transform
template <class NewStep>
generation<Generator, Steps..., NewStep> transform(NewStep step) && {
static_assert(std::is_same_v<typename NewStep::input_type, output_type>,
"step object does not match the output type");
return {ctx_, std::move(gen_),
std::tuple_cat(std::move(steps_),
std::make_tuple(std::move(step)))};
}
auto take(size_t n) && {
return std::move(*this).transform(limit_step<output_type>{n});
}
template <class Predicate>
auto filter(Predicate predicate) && {
return std::move(*this).transform(
filter_step<Predicate>{std::move(predicate)});
}
template <class Predicate>
auto take_while(Predicate predicate) && {
return std::move(*this).transform(
take_while_step<Predicate>{std::move(predicate)});
}
template <class Reducer>
auto reduce(output_type init, Reducer reducer) && {
return std::move(*this).transform(
reduce_step<output_type, Reducer>{init, reducer});
}
auto sum() && {
return std::move(*this).reduce(output_type{}, std::plus<output_type>{});
}
auto distinct() && {
return std::move(*this).transform(distinct_step<output_type>{});
}
template <class Fn>
auto map(Fn fn) && {
return std::move(*this).transform(map_step<Fn>{std::move(fn)});
}
template <class... Inputs>
auto merge(Inputs&&... xs) && {
return std::move(*this).as_observable().merge(std::forward<Inputs>(xs)...);
}
template <class... Inputs>
auto concat(Inputs&&... xs) && {
return std::move(*this).as_observable().concat(std::forward<Inputs>(xs)...);
}
template <class F>
auto flat_map_optional(F f) && {
return std::move(*this).transform(flat_map_optional_step<F>{std::move(f)});
}
template <class F>
auto do_on_next(F f) {
return std::move(*this) //
.transform(do_on_next_step<output_type, F>{std::move(f)});
}
template <class F>
auto do_on_complete(F f) && {
return std::move(*this).transform(
do_on_complete_step<output_type, F>{std::move(f)});
}
template <class F>
auto do_on_error(F f) && {
return std::move(*this).transform(
do_on_error_step<output_type, F>{std::move(f)});
}
template <class F>
auto do_finally(F f) && {
return std::move(*this).transform(
do_finally_step<output_type, F>{std::move(f)});
}
observable<output_type> as_observable() && override {
using impl_t = op::from_generator<Generator, Steps...>;
return make_observable<impl_t>(ctx_, std::move(gen_), std::move(steps_));
}
coordinator* ctx() const noexcept {
return ctx_;
}
constexpr bool valid() const noexcept {
return true;
}
private:
coordinator* ctx_;
Generator gen_;
std::tuple<Steps...> steps_;
};
// -- observable_builder::repeat -----------------------------------------------
template <class T>
generation<repeater_source<T>> observable_builder::repeat(T value) const {
return {ctx_, repeater_source<T>{std::move(value)}};
}
// -- observable_builder::from_container ---------------------------------------
template <class Container>
generation<container_source<Container>>
observable_builder::from_container(Container values) const {
return {ctx_, container_source<Container>{std::move(values)}};
}
// -- observable_builder::just -------------------------------------------------
template <class T>
generation<value_source<T>> observable_builder::just(T value) const {
return {ctx_, value_source<T>{std::move(value)}};
}
// -- observable_builder::from_callable ----------------------------------------
template <class F>
generation<callable_source<F>> observable_builder::from_callable(F fn) const {
return {ctx_, callable_source<F>{std::move(fn)}};
}
// -- observable_builder::from_resource ----------------------------------------
template <class T>
observable<T>
observable_builder::from_resource(async::consumer_resource<T> hdl) const {
using impl_t = op::from_resource<T>;
return make_observable<impl_t>(ctx_, std::move(hdl));
}
// -- observable_builder::lift -------------------------------------------------
template <class Pullable>
generation<Pullable> observable_builder::lift(Pullable pullable) const {
return {ctx_, std::move(pullable)};
}
} // namespace caf::flow } // namespace caf::flow
...@@ -4,15 +4,16 @@ ...@@ -4,15 +4,16 @@
#pragma once #pragma once
#include "caf/cow_vector.hpp"
#include "caf/defaults.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/flow/fwd.hpp" #include "caf/flow/fwd.hpp"
#include "caf/flow/op/base.hpp" #include "caf/flow/op/base.hpp"
#include "caf/flow/step.hpp" #include "caf/flow/step/fwd.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include <cstddef> #include <cstddef>
#include <functional>
#include <utility> #include <utility>
#include <vector> #include <vector>
...@@ -22,12 +23,16 @@ namespace caf::flow { ...@@ -22,12 +23,16 @@ namespace caf::flow {
template <class T> template <class T>
class observable { class observable {
public: public:
// -- member types -----------------------------------------------------------
/// The type of emitted items. /// The type of emitted items.
using output_type = T; using output_type = T;
/// The pointer-to-implementation type. /// The pointer-to-implementation type.
using pimpl_type = intrusive_ptr<op::base<T>>; using pimpl_type = intrusive_ptr<op::base<T>>;
// -- constructors, destructors, and assignment operators --------------------
explicit observable(pimpl_type pimpl) noexcept : pimpl_(std::move(pimpl)) { explicit observable(pimpl_type pimpl) noexcept : pimpl_(std::move(pimpl)) {
// nop // nop
} }
...@@ -43,96 +48,91 @@ public: ...@@ -43,96 +48,91 @@ public:
observable& operator=(observable&&) noexcept = default; observable& operator=(observable&&) noexcept = default;
observable& operator=(const observable&) noexcept = default; observable& operator=(const observable&) noexcept = default;
/// @copydoc impl::subscribe // -- subscribing ------------------------------------------------------------
disposable subscribe(observer<T> what) {
if (pimpl_) { /// Subscribes a new observer to the items emitted by this observable.
return pimpl_->subscribe(std::move(what)); disposable subscribe(observer<T> what);
} else {
what.on_error(make_error(sec::invalid_observable));
return disposable{};
}
}
/// Creates a new observer that pushes all observed items to the resource. /// Creates a new observer that pushes all observed items to the resource.
disposable subscribe(async::producer_resource<T> resource); disposable subscribe(async::producer_resource<T> resource);
/// Calls `on_next` for each item emitted by this observable.
template <class OnNext>
disposable for_each(OnNext on_next);
// -- transforming -----------------------------------------------------------
/// Returns a transformation that applies a step function to each input. /// Returns a transformation that applies a step function to each input.
template <class Step> template <class Step>
transformation<Step> transform(Step step); transformation<Step> transform(Step step);
/// Registers a callback for `on_next` events. /// Makes all values unique by suppressing items that have been emitted in the
/// past.
transformation<step::distinct<T>> distinct();
/// Registers a callback for `on_complete` and `on_error` events.
template <class F> template <class F>
auto do_on_next(F f) { transformation<step::do_finally<T, F>> do_finally(F f);
return transform(do_on_next_step<T, F>{std::move(f)});
}
/// Registers a callback for `on_complete` events. /// Registers a callback for `on_complete` events.
template <class F> template <class F>
auto do_on_complete(F f) { transformation<step::do_on_complete<T, F>> do_on_complete(F f);
return transform(do_on_complete_step<T, F>{std::move(f)});
}
/// Registers a callback for `on_error` events. /// Registers a callback for `on_error` events.
template <class F> template <class F>
auto do_on_error(F f) { transformation<step::do_on_error<T, F>> do_on_error(F f);
return transform(do_on_error_step<T, F>{std::move(f)});
}
/// Registers a callback that runs on `on_complete` or `on_error`. /// Registers a callback for `on_next` events.
template <class F> template <class F>
auto do_finally(F f) { transformation<step::do_on_next<F>> do_on_next(F f);
return transform(do_finally_step<T, F>{std::move(f)});
}
/// Catches errors by converting them into errors instead.
auto on_error_complete() {
return transform(on_error_complete_step<T>{});
}
/// Returns a transformation that selects only the first `n` items.
transformation<limit_step<T>> take(size_t n);
/// Returns a transformation that selects only items that satisfy `predicate`. /// Returns a transformation that selects only items that satisfy `predicate`.
template <class Predicate> template <class Predicate>
transformation<filter_step<Predicate>> filter(Predicate prediate); transformation<step::filter<Predicate>> filter(Predicate prediate);
/// Returns a transformation that selects all value until the `predicate` /// Returns a transformation that applies `f` to each input and emits the
/// returns false. /// result of the function application.
template <class Predicate> template <class F>
transformation<take_while_step<Predicate>> take_while(Predicate prediate); transformation<step::map<F>> map(F f);
/// Recovers from errors by converting `on_error` to `on_complete` events.
transformation<step::on_error_complete<T>> on_error_complete();
/// Reduces the entire sequence of items to a single value. Other names for /// Reduces the entire sequence of items to a single value. Other names for
/// the algorithm are `accumulate` and `fold`. /// the algorithm are `accumulate` and `fold`.
template <class Reducer> /// @param init The initial value for the reduction.
transformation<reduce_step<T, Reducer>> reduce(T init, Reducer reducer); /// @param reducer Binary operation function that will be applied.
template <class Init, class Reducer>
transformation<step::reduce<Reducer>> reduce(Init init, Reducer reducer);
/// Accumulates all values and emits only the final result. /// Returns a transformation that selects all but the first `n` items.
auto sum() { transformation<step::skip<T>> skip(size_t n);
return reduce(T{}, std::plus<T>{});
}
/// Makes all values unique by suppressing all items that have been emitted in /// Returns a transformation that selects only the first `n` items.
/// the past. transformation<step::take<T>> take(size_t n);
transformation<distinct_step<T>> distinct();
/// Returns a transformation that applies `f` to each input and emits the /// Returns a transformation that selects all value until the `predicate`
/// result of the function application. /// returns false.
template <class F> template <class Predicate>
transformation<map_step<F>> map(F f); transformation<step::take_while<Predicate>> take_while(Predicate prediate);
/// Calls `on_next` for each item emitted by this observable. /// Accumulates all values and emits only the final result.
template <class OnNext> auto sum() {
disposable for_each(OnNext on_next); return reduce(T{}, [](T x, T y) { return x + y; });
}
/// Calls `on_next` for each item and `on_error` for each error emitted by /// Collects all values and emits all values at once in a @ref cow_vector.
/// this observable. auto to_vector() {
template <class OnNext, class OnError> using vector_type = cow_vector<output_type>;
disposable for_each(OnNext on_next, OnError on_error); auto append = [](vector_type&& xs, const output_type& x) {
xs.unshared().push_back(x);
return xs;
};
return reduce(vector_type{}, append) //
.filter([](const vector_type& xs) { return !xs.empty(); });
}
/// Calls `on_next` for each item, `on_error` for each error and `on_complete` // -- combining --------------------------------------------------------------
/// for each completion signal emitted by this observable.
template <class OnNext, class OnError, class OnComplete>
disposable for_each(OnNext on_next, OnError on_error, OnComplete on_complete);
/// Combines the output of multiple @ref observable objects into one by /// Combines the output of multiple @ref observable objects into one by
/// merging their outputs. May also be called without arguments if the `T` is /// merging their outputs. May also be called without arguments if the `T` is
...@@ -156,53 +156,68 @@ public: ...@@ -156,53 +156,68 @@ 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);
// -- 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
/// containing an observable for the remaining elements as the second value. /// containing an observable for the remaining elements as the second value.
/// The returned observable either emits a single element (the tuple) or none /// The returned observable either emits a single element (the tuple) or none
/// if this observable never produces sufficient elements for the prefix. /// if this observable never produces sufficient elements for the prefix.
/// @pre `prefix_size > 0` /// @pre `prefix_size > 0`
observable<cow_tuple<std::vector<T>, observable<T>>> observable<cow_tuple<cow_vector<T>, observable<T>>>
prefix_and_tail(size_t prefix_size); prefix_and_tail(size_t prefix_size);
/// Similar to `prefix_and_tail(1)` but passes the single element directly in /// Similar to `prefix_and_tail(1)` but passes the single element directly in
/// the tuple instead of wrapping it in a list. /// the tuple instead of wrapping it in a list.
observable<cow_tuple<T, observable<T>>> head_and_tail(); observable<cow_tuple<T, observable<T>>> head_and_tail();
// -- multicasting -----------------------------------------------------------
/// Convert this observable into a @ref connectable observable. /// Convert this observable into a @ref connectable observable.
connectable<T> publish(); connectable<T> publish();
/// Convenience alias for `publish().ref_count(subscriber_threshold)`. /// Convenience alias for `publish().ref_count(subscriber_threshold)`.
observable<T> share(size_t subscriber_threshold = 1); observable<T> share(size_t subscriber_threshold = 1);
/// Transform this `observable` by applying a function object to it. // -- composing --------------------------------------------------------------
/// Transforms this `observable` by applying a function object to it.
template <class Fn> template <class Fn>
auto compose(Fn&& fn) & { auto compose(Fn&& fn) & {
return fn(*this); return fn(*this);
} }
/// Fn this `observable` by applying a function object to it. /// Transforms this `observable` by applying a function object to it.
template <class Fn> template <class Fn>
auto compose(Fn&& fn) && { auto compose(Fn&& fn) && {
return fn(std::move(*this)); return fn(std::move(*this));
} }
/// Creates an asynchronous resource that makes emitted items available in a // -- observing --------------------------------------------------------------
/// spsc buffer.
async::consumer_resource<T> to_resource(size_t buffer_size,
size_t min_request_size);
async::consumer_resource<T> to_resource() {
return to_resource(defaults::flow::buffer_size, defaults::flow::min_demand);
}
/// Observes items from this observable on another @ref coordinator.
/// @warning The @p other @ref coordinator *must not* run at this point.
observable observe_on(coordinator* other, size_t buffer_size, observable observe_on(coordinator* other, size_t buffer_size,
size_t min_request_size); size_t min_request_size);
/// Observes items from this observable on another @ref coordinator.
/// @warning The @p other @ref coordinator *must not* run at this point.
observable observe_on(coordinator* other) { observable observe_on(coordinator* other) {
return observe_on(other, defaults::flow::buffer_size, return observe_on(other, defaults::flow::buffer_size,
defaults::flow::min_demand); defaults::flow::min_demand);
} }
// -- converting -------------------------------------------------------------
/// Creates an asynchronous resource that makes emitted items available in an
/// SPSC buffer.
async::consumer_resource<T> to_resource(size_t buffer_size,
size_t min_request_size);
/// Creates an asynchronous resource that makes emitted items available in an
/// SPSC buffer.
async::consumer_resource<T> to_resource() {
return to_resource(defaults::flow::buffer_size, defaults::flow::min_demand);
}
const observable& as_observable() const& noexcept { const observable& as_observable() const& noexcept {
return std::move(*this); return std::move(*this);
} }
...@@ -211,10 +226,16 @@ public: ...@@ -211,10 +226,16 @@ public:
return std::move(*this); return std::move(*this);
} }
const pimpl_type& pimpl() const noexcept { // -- properties -------------------------------------------------------------
const pimpl_type& pimpl() const& noexcept {
return pimpl_; return pimpl_;
} }
pimpl_type pimpl() && noexcept {
return std::move(pimpl_);
}
bool valid() const noexcept { bool valid() const noexcept {
return pimpl_ != nullptr; return pimpl_ != nullptr;
} }
...@@ -227,16 +248,20 @@ public: ...@@ -227,16 +248,20 @@ public:
return !valid(); return !valid();
} }
void swap(observable& other) {
pimpl_.swap(other.pimpl_);
}
/// @pre `valid()` /// @pre `valid()`
coordinator* ctx() const { coordinator* ctx() const {
return pimpl_->ctx(); return pimpl_->ctx();
} }
// -- swapping ---------------------------------------------------------------
void swap(observable& other) {
pimpl_.swap(other.pimpl_);
}
private: private:
// -- member variables -------------------------------------------------------
pimpl_type pimpl_; pimpl_type pimpl_;
}; };
......
...@@ -20,7 +20,7 @@ class from_generator_sub : public subscription::impl_base { ...@@ -20,7 +20,7 @@ class from_generator_sub : public subscription::impl_base {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using output_type = steps_output_type_t<Generator, Steps...>; using output_type = output_type_t<Generator, Steps...>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
......
// 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/error.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/hot.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace caf::flow::op {
/// Allows operators to subscribe to an observable immediately to force an eager
/// subscription while the observable that actually consumes the items
/// subscribes later. May only be subscribed once.
template <class T>
class prefetch : public hot<T>, public observer_impl<T> {
public:
// -- member types -----------------------------------------------------------
using super = hot<T>;
using output_type = T;
// -- constructors, destructors, and assignment operators --------------------
explicit prefetch(coordinator* ctx) : super(ctx) {
// nop
}
// -- ref counting (and disambiguation due to multiple base types) -----------
void ref_coordinated() const noexcept override {
this->ref();
}
void deref_coordinated() const noexcept override {
this->deref();
}
friend void intrusive_ptr_add_ref(const prefetch* ptr) noexcept {
ptr->ref_coordinated();
}
friend void intrusive_ptr_release(const prefetch* ptr) noexcept {
ptr->deref_coordinated();
}
// -- implementation of observable_impl<T> -----------------------------------
disposable subscribe(observer<T> out) override {
if (completed_) {
if (err_)
out.on_error(err_);
else
out.on_complete();
return {};
} else if (!out_ && sub_) {
out_ = std::move(out);
out_.on_subscribe(sub_);
return sub_.as_disposable();
} else {
auto err = make_error(sec::invalid_observable,
"prefetch cannot add more than one subscriber");
out.on_error(err);
return {};
}
}
// -- implementation of observer_impl<T> -------------------------------------
void on_next(const T& item) override {
out_.on_next(item);
}
void on_complete() override {
completed_ = true;
if (out_) {
out_.on_complete();
out_ = nullptr;
sub_ = nullptr;
}
}
void on_error(const error& what) override {
completed_ = true;
err_ = what;
if (out_) {
out_.on_error(what);
out_ = nullptr;
sub_ = nullptr;
}
}
void on_subscribe(subscription sub) override {
if (!sub_) {
sub_ = sub;
} else {
sub.dispose();
}
}
// -- convenience functions --------------------------------------------------
static intrusive_ptr<base<T>> apply(intrusive_ptr<base<T>> src) {
auto ptr = make_counted<prefetch>(src->ctx());
src->subscribe(observer<T>{ptr});
return ptr;
}
private:
bool completed_ = false;
error err_;
observer<T> out_;
subscription sub_;
};
} // namespace caf::flow::op
// 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/cow_tuple.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cell.hpp"
#include "caf/flow/subscription.hpp"
#include <cstdint>
namespace caf::flow::op {
template <class T>
class pipe_sub : public detail::plain_ref_counted,
public observer_impl<T>,
public subscription_impl {
public:
// -- constructors, destructors, and assignment operators --------------------
explicit pipe_sub(subscription in) : in_(std::move(in)) {
// nop
}
void init(observer<T> out) {
if (!completed_)
out_ = std::move(out);
else if (err_)
out.on_error(err_);
else
out.on_complete();
}
// -- ref counting -----------------------------------------------------------
void ref_coordinated() const noexcept final {
this->ref();
}
void deref_coordinated() const noexcept final {
this->deref();
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
friend void intrusive_ptr_add_ref(const pipe_sub* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const pipe_sub* ptr) noexcept {
ptr->deref();
}
// -- implementation of observer_impl<Input> ---------------------------------
void on_next(const T& item) override {
if (out_)
out_.on_next(item);
}
void on_complete() override {
if (out_) {
in_ = nullptr;
out_.on_complete();
out_ = nullptr;
completed_ = true;
} else if (!completed_) {
completed_ = true;
}
}
void on_error(const error& what) override {
if (out_) {
in_ = nullptr;
out_.on_error(what);
out_ = nullptr;
completed_ = true;
err_ = what;
} else if (!completed_) {
completed_ = true;
err_ = what;
}
}
void on_subscribe(subscription in) override {
in.dispose();
}
// -- implementation of subscription_impl ------------------------------------
bool disposed() const noexcept override {
return !out_;
}
void dispose() override {
if (in_) {
in_.dispose();
in_ = nullptr;
}
if (out_) {
out_.on_complete();
out_ = nullptr;
}
}
void request(size_t n) override {
if (in_) {
in_.request(n);
}
}
private:
subscription in_;
observer<T> out_;
/// Stores whether on_complete has been called. This may be the case even
/// before init gets called.
bool completed_ = false;
/// Stores the error passed to on_error.
error err_;
};
template <class T>
class pipe : public hot<T> {
public:
using super = hot<T>;
using sub_ptr = intrusive_ptr<pipe_sub<T>>;
explicit pipe(coordinator* ctx, sub_ptr sub)
: super(ctx), sub_(std::move(sub)) {
// nop
}
~pipe() {
if (sub_)
sub_->dispose();
}
disposable subscribe(observer<T> out) override {
if (sub_) {
sub_->init(out);
out.on_subscribe(subscription{sub_});
return disposable{std::move(sub_)};
} else {
auto err = make_error(sec::invalid_observable,
"pipes may only be subscribed once");
out.on_error(err);
return disposable{};
}
}
private:
sub_ptr sub_;
};
template <class T>
class prefix_and_tail_fwd : public detail::plain_ref_counted,
public observer_impl<T> {
public:
// -- constructors, destructors, and assignment operators --------------------
using output_type = cow_tuple<std::vector<T>, observable<T>>;
using cell_ptr = intrusive_ptr<cell<output_type>>;
prefix_and_tail_fwd(coordinator* ctx, size_t prefix_size, cell_ptr sync)
: ctx_(ctx), prefix_size_(prefix_size), sync_(std::move(sync)) {
// nop
}
// -- ref counting -----------------------------------------------------------
void ref_coordinated() const noexcept final {
this->ref();
}
void deref_coordinated() const noexcept final {
this->deref();
}
friend void intrusive_ptr_add_ref(const prefix_and_tail_fwd* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const prefix_and_tail_fwd* ptr) noexcept {
ptr->deref();
}
// -- implementation of observer_impl<T> -------------------------------------
void on_next(const T& item) override {
if (next_) {
next_.on_next(item);
} else if (sync_) {
prefix_.push_back(item);
if (prefix_.size() == prefix_size_) {
auto fwd = make_counted<pipe_sub<T>>(std::move(in_));
auto tmp = make_counted<pipe<T>>(ctx_, fwd);
output_type result{std::move(prefix_), observable<T>{tmp}};
sync_->set_value(std::move(result));
sync_ = nullptr;
next_ = observer<T>{fwd};
}
}
}
void on_complete() override {
if (next_) {
next_.on_complete();
next_ = nullptr;
} else if (sync_) {
sync_->set_null();
sync_ = nullptr;
}
}
void on_error(const error& what) override {
if (next_) {
next_.on_error(what);
next_ = nullptr;
} else if (sync_) {
sync_->set_error(what);
sync_ = nullptr;
}
}
void on_subscribe(subscription in) override {
in_ = std::move(in);
in_.request(prefix_size_);
}
private:
coordinator* ctx_;
size_t prefix_size_ = 0;
subscription in_;
std::vector<T> prefix_;
observer<T> next_;
cell_ptr sync_;
};
template <class T>
class prefix_and_tail : public cold<cow_tuple<std::vector<T>, observable<T>>> {
public:
// -- member types -----------------------------------------------------------
using output_type = cow_tuple<std::vector<T>, observable<T>>;
using super = cold<output_type>;
using observer_type = observer<T>;
// -- constructors, destructors, and assignment operators --------------------
prefix_and_tail(coordinator* ctx, intrusive_ptr<base<T>> decorated,
size_t prefix_size)
: super(ctx), decorated_(std::move(decorated)), prefix_size_(prefix_size) {
// nop
}
disposable subscribe(observer<output_type> out) override {
// The cell acts as a sort of handshake between the dispatcher and the
// observer. After producing the (prefix, tail) pair, the dispatcher goes on
// to forward items from the decorated observable to the tail.
auto sync = make_counted<cell<output_type>>(super::ctx_);
auto fwd = make_counted<prefix_and_tail_fwd<T>>(super::ctx_, prefix_size_,
sync);
std::vector<disposable> result;
result.reserve(2);
result.emplace_back(sync->subscribe(std::move(out)));
result.emplace_back(decorated_->subscribe(observer<T>{std::move(fwd)}));
return disposable::make_composite(std::move(result));
}
private:
intrusive_ptr<base<T>> decorated_;
size_t prefix_size_;
};
} // namespace caf::flow::op
...@@ -49,10 +49,6 @@ public: ...@@ -49,10 +49,6 @@ public:
// -- implementation of conn<T> ---------------------------------------------- // -- implementation of conn<T> ----------------------------------------------
coordinator* ctx() const noexcept override {
return super::ctx_;
}
disposable subscribe(observer<T> out) override { disposable subscribe(observer<T> out) override {
auto result = super::subscribe(std::move(out)); auto result = super::subscribe(std::move(out));
if (!connected_ && super::observer_count() == auto_connect_threshold_) { if (!connected_ && super::observer_count() == auto_connect_threshold_) {
......
...@@ -4,18 +4,90 @@ ...@@ -4,18 +4,90 @@
#pragma once #pragma once
#include <algorithm>
#include <utility>
#include <variant>
#include "caf/detail/overload.hpp" #include "caf/detail/overload.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/flow/observable.hpp" #include "caf/flow/observable.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include <algorithm>
#include <optional>
#include <type_traits>
#include <utility>
namespace caf::flow { namespace caf::flow {
template <class OnSuccess>
struct on_success_orcacle {
using trait = detail::get_callable_trait_t<OnSuccess>;
static_assert(trait::num_args == 1,
"OnSuccess functions must take exactly one argument");
using arg_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
};
template <class OnSuccess>
using on_success_arg_t = typename on_success_orcacle<OnSuccess>::arg_type;
template <class OnSuccess, class OnError>
class single_observer_impl
: public observer_impl_base<on_success_arg_t<OnSuccess>> {
public:
using input_type = on_success_arg_t<OnSuccess>;
single_observer_impl(OnSuccess on_success, OnError on_error)
: on_success_(std::move(on_success)), on_error_(std::move(on_error)) {
// nop
}
void on_subscribe(subscription sub) {
// Request one additional item to detect whether the observable emits more
// than one item.
sub.request(2);
sub_ = std::move(sub);
}
void on_next(const input_type& item) {
if (!result_) {
result_.emplace(item);
} else {
sub_.dispose();
sub_ = nullptr;
auto err = make_error(sec::runtime_error,
"single emitted more than one item");
on_error_(err);
}
}
void on_complete() {
if (sub_) {
sub_ = nullptr;
if (result_) {
on_success_(*result_);
result_ = std::nullopt;
} else {
auto err = make_error(sec::broken_promise,
"single failed to produce an item");
on_error_(err);
}
}
}
void on_error(const error& what) {
if (sub_) {
sub_ = nullptr;
on_error_(what);
}
}
private:
OnSuccess on_success_;
OnError on_error_;
std::optional<input_type> result_;
subscription sub_;
};
/// Similar to an `observable`, but always emits either a single value or an /// Similar to an `observable`, but always emits either a single value or an
/// error. /// error.
template <class T> template <class T>
...@@ -46,19 +118,13 @@ public: ...@@ -46,19 +118,13 @@ public:
return observable<T>{pimpl_}; return observable<T>{pimpl_};
} }
void subscribe(observer<T> what) {
if (pimpl_)
pimpl_->subscribe(std::move(what));
else
what.on_error(make_error(sec::invalid_observable));
}
template <class OnSuccess, class OnError> template <class OnSuccess, class OnError>
void subscribe(OnSuccess on_success, OnError on_error) { disposable subscribe(OnSuccess on_success, OnError on_error) {
static_assert(std::is_invocable_v<OnSuccess, const T&>); static_assert(std::is_invocable_v<OnSuccess, const T&>);
as_observable().for_each([f{std::move(on_success)}]( static_assert(std::is_invocable_v<OnError, const error&>);
const T& item) mutable { f(item); }, using impl_t = single_observer_impl<OnSuccess, OnError>;
std::move(on_error)); auto ptr = make_counted<impl_t>(std::move(on_success), std::move(on_error));
return pimpl_->subscribe(observer<T>{ptr});
} }
bool valid() const noexcept { bool valid() const noexcept {
...@@ -81,8 +147,7 @@ private: ...@@ -81,8 +147,7 @@ private:
intrusive_ptr<op::base<T>> pimpl_; intrusive_ptr<op::base<T>> pimpl_;
}; };
/// Convenience function for creating an @ref observable from a concrete /// Convenience function for creating a @ref single from a flow operator.
/// operator type.
template <class Operator, class... Ts> template <class Operator, class... Ts>
single<typename Operator::output_type> single<typename Operator::output_type>
make_single(coordinator* ctx, Ts&&... xs) { make_single(coordinator* ctx, Ts&&... xs) {
......
// 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/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
#include "caf/flow/observable_state.hpp"
#include "caf/flow/observer.hpp"
#include <algorithm>
#include <numeric>
#include <type_traits>
#include <unordered_set>
namespace caf::flow {
template <class T>
struct identity_step {
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class T>
struct limit_step {
size_t remaining;
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (remaining > 0) {
if (next.on_next(item, steps...)) {
if (--remaining > 0) {
return true;
} else {
next.on_complete(steps...);
return false;
}
}
}
return false;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class Predicate>
struct filter_step {
using trait = detail::get_callable_trait_t<Predicate>;
static_assert(std::is_convertible_v<typename trait::result_type, bool>,
"predicates must return a boolean value");
static_assert(trait::num_args == 1,
"predicates must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = input_type;
Predicate predicate;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (predicate(item))
return next.on_next(item, steps...);
else
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class Predicate>
struct take_while_step {
using trait = detail::get_callable_trait_t<Predicate>;
static_assert(std::is_convertible_v<typename trait::result_type, bool>,
"predicates must return a boolean value");
static_assert(trait::num_args == 1,
"predicates must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = input_type;
Predicate predicate;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (predicate(item)) {
return next.on_next(item, steps...);
} else {
next.on_complete(steps...);
return false;
}
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class T>
struct distinct_step {
using input_type = T;
using output_type = T;
std::unordered_set<T> prev;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (prev.insert(item).second)
return next.on_next(item, steps...);
else
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class Fn>
struct map_step {
using trait = detail::get_callable_trait_t<Fn>;
static_assert(!std::is_same_v<typename trait::result_type, void>,
"map functions may not return void");
static_assert(trait::num_args == 1,
"map functions must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = std::decay_t<typename trait::result_type>;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(fn(item), steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class T, class Reducer>
struct reduce_step {
T result;
Reducer fn;
reduce_step(T init, Reducer reducer) : result(std::move(init)), fn(reducer) {
// nop
}
reduce_step(reduce_step&&) = default;
reduce_step(const reduce_step&) = default;
reduce_step& operator=(reduce_step&&) = default;
reduce_step& operator=(const reduce_step&) = default;
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next&, Steps&...) {
result = fn(std::move(result), item);
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
if (next.on_next(result, steps...))
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
if (next.on_next(result, steps...))
next.on_error(what, steps...);
}
};
template <class Fn>
struct flat_map_optional_step {
using trait = detail::get_callable_trait_t<Fn>;
static_assert(!std::is_same_v<typename trait::result_type, void>,
"flat_map_optional functions may not return void");
static_assert(trait::num_args == 1,
"flat_map_optional functions must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using intermediate_type = std::decay_t<typename trait::result_type>;
using output_type = typename intermediate_type::value_type;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (auto val = fn(item))
return next.on_next(*val, steps...);
else
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class T, class Fn>
struct do_on_next_step {
using input_type = T;
using output_type = T;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
fn(item);
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class T, class Fn>
struct do_on_complete_step {
using input_type = T;
using output_type = T;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
fn();
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class T, class Fn>
struct do_on_error_step {
using input_type = T;
using output_type = T;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
fn(what);
next.on_error(what, steps...);
}
};
template <class T, class Fn>
struct do_finally_step {
using input_type = T;
using output_type = T;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
fn();
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
fn();
next.on_error(what, steps...);
}
};
/// Catches errors by converting them into `complete` events instead.
template <class T>
struct on_error_complete_step {
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error&, Next& next, Steps&... steps) {
next.on_complete(steps...);
}
};
/// Utility for the observables that use one or more steps.
template <class... Steps>
struct steps_output_oracle;
template <class Step>
struct steps_output_oracle<Step> {
using type = typename Step::output_type;
};
template <class Step1, class Step2, class... Steps>
struct steps_output_oracle<Step1, Step2, Steps...>
: steps_output_oracle<Step2, Steps...> {};
template <class... Steps>
using steps_output_type_t = typename steps_output_oracle<Steps...>::type;
} // namespace caf::flow
#include "caf/flow/step/distinct.hpp"
#include "caf/flow/step/do_finally.hpp"
#include "caf/flow/step/do_on_complete.hpp"
#include "caf/flow/step/do_on_error.hpp"
#include "caf/flow/step/do_on_next.hpp"
#include "caf/flow/step/filter.hpp"
#include "caf/flow/step/map.hpp"
#include "caf/flow/step/on_error_complete.hpp"
#include "caf/flow/step/reduce.hpp"
#include "caf/flow/step/skip.hpp"
#include "caf/flow/step/take.hpp"
#include "caf/flow/step/take_while.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.
#pragma once
#include "caf/fwd.hpp"
#include <unordered_set>
namespace caf::flow::step {
template <class T>
class distinct {
public:
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (prev_.insert(item).second)
return next.on_next(item, steps...);
else
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
std::unordered_set<T> prev_;
};
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class T, class F>
class do_finally {
public:
using input_type = T;
using output_type = T;
explicit do_finally(F fn) : fn_(std::move(fn)) {
// nop
}
do_finally(do_finally&&) = default;
do_finally(const do_finally&) = default;
do_finally& operator=(do_finally&&) = default;
do_finally& operator=(const do_finally&) = default;
template <class Next, class... Steps>
bool on_next(const T& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void finally(Next& next, Steps&... steps) {
fn_();
next.finally(steps...);
}
template <class Next, class... Steps>
void finally(const error& what, Next& next, Steps&... steps) {
fn_();
next.finally(what, steps...);
}
private:
F fn_;
};
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class T, class F>
class do_on_complete {
public:
using input_type = T;
using output_type = T;
explicit do_on_complete(F fn) : fn_(std::move(fn)) {
// nop
}
do_on_complete(do_on_complete&&) = default;
do_on_complete(const do_on_complete&) = default;
do_on_complete& operator=(do_on_complete&&) = default;
do_on_complete& operator=(const do_on_complete&) = default;
template <class Next, class... Steps>
bool on_next(const T& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
fn_();
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
F fn_;
};
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class T, class F>
class do_on_error {
public:
using input_type = T;
using output_type = T;
explicit do_on_error(F fn) : fn_(std::move(fn)) {
// nop
}
do_on_error(do_on_error&&) = default;
do_on_error(const do_on_error&) = default;
do_on_error& operator=(do_on_error&&) = default;
do_on_error& operator=(const do_on_error&) = default;
template <class Next, class... Steps>
bool on_next(const T& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
fn_(what);
next.on_error(what, steps...);
}
private:
F fn_;
};
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class F>
class do_on_next {
public:
using trait = detail::get_callable_trait_t<F>;
static_assert(!std::is_same_v<typename trait::result_type, void>,
"do_on_next functions may not return void");
static_assert(trait::num_args == 1,
"do_on_next functions must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = std::decay_t<typename trait::result_type>;
explicit do_on_next(F fn) : fn_(std::move(fn)) {
// nop
}
do_on_next(do_on_next&&) = default;
do_on_next(const do_on_next&) = default;
do_on_next& operator=(do_on_next&&) = default;
do_on_next& operator=(const do_on_next&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
fn_(item);
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
F fn_;
};
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class Predicate>
class filter {
public:
using trait = detail::get_callable_trait_t<Predicate>;
static_assert(std::is_convertible_v<typename trait::result_type, bool>,
"predicates must return a boolean value");
static_assert(trait::num_args == 1,
"predicates must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = input_type;
explicit filter(Predicate fn) : fn_(std::move(fn)) {
// nop
}
filter(filter&&) = default;
filter(const filter&) = default;
filter& operator=(filter&&) = default;
filter& operator=(const filter&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (fn_(item))
return next.on_next(item, steps...);
else
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
Predicate fn_;
};
} // namespace caf::flow::step
namespace caf::flow::step {
template <class>
class distinct;
template <class, class>
class do_finally;
template <class, class>
class do_on_complete;
template <class, class>
class do_on_error;
template <class>
class do_on_next;
template <class>
class filter;
template <class>
class map;
template <class>
class on_error_complete;
template <class>
class reduce;
template <class>
class skip;
template <class>
class take;
template <class>
class take_while;
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class F>
class map {
public:
using trait = detail::get_callable_trait_t<F>;
static_assert(!std::is_same_v<typename trait::result_type, void>,
"map functions may not return void");
static_assert(trait::num_args == 1,
"map functions must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = std::decay_t<typename trait::result_type>;
explicit map(F fn) : fn_(std::move(fn)) {
// nop
}
map(map&&) = default;
map(const map&) = default;
map& operator=(map&&) = default;
map& operator=(const map&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(fn_(item), steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
F fn_;
};
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class T>
class on_error_complete {
public:
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const T& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_error(Next& next, Steps&... steps) {
next.on_error(steps...);
}
template <class Next, class... Steps>
void on_error(const error&, Next& next, Steps&... steps) {
next.on_complete(steps...);
}
};
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class F>
class reduce {
public:
using trait = detail::get_callable_trait_t<F>;
static_assert(!std::is_same_v<typename trait::result_type, void>,
"reduce functions may not return void");
static_assert(trait::num_args == 2,
"reduce functions must take exactly two arguments");
using arg_type = typename trait::arg_types;
using input_type = std::decay_t<detail::tl_at_t<arg_type, 1>>;
using output_type = std::decay_t<typename trait::result_type>;
explicit reduce(output_type init, F fn)
: val_(std::move(init)), fn_(std::move(fn)) {
// nop
}
reduce(reduce&&) = default;
reduce(const reduce&) = default;
reduce& operator=(reduce&&) = default;
reduce& operator=(const reduce&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next&, Steps&...) {
val_ = fn_(std::move(val_), item);
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
if (next.on_next(val_, steps...))
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
if (next.on_next(val_, steps...))
next.on_error(what, steps...);
}
private:
output_type val_;
F fn_;
};
} // namespace caf::flow::step
// 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/fwd.hpp"
#include <cstddef>
namespace caf::flow::step {
template <class T>
class skip {
public:
using input_type = T;
using output_type = T;
explicit skip(size_t num) : remaining_(num) {
// nop
}
skip(skip&&) = default;
skip(const skip&) = default;
skip& operator=(skip&&) = default;
skip& operator=(const skip&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (remaining_ == 0) {
return next.on_next(item, steps...);
} else {
--remaining_;
return true;
}
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
size_t remaining_;
};
} // namespace caf::flow::step
// 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/fwd.hpp"
#include <cstddef>
namespace caf::flow::step {
template <class T>
class take {
public:
using input_type = T;
using output_type = T;
explicit take(size_t num) : remaining_(num) {
// nop
}
take(take&&) = default;
take(const take&) = default;
take& operator=(take&&) = default;
take& operator=(const take&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (remaining_ > 0) {
if (next.on_next(item, steps...)) {
if (--remaining_ > 0) {
return true;
} else {
next.on_complete(steps...);
return false;
}
}
}
return false;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
size_t remaining_;
};
} // namespace caf::flow::step
// 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/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include <utility>
namespace caf::flow::step {
template <class Predicate>
class take_while {
public:
using trait = detail::get_callable_trait_t<Predicate>;
static_assert(std::is_convertible_v<typename trait::result_type, bool>,
"predicates must return a boolean value");
static_assert(trait::num_args == 1,
"predicates must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = input_type;
explicit take_while(Predicate fn) : fn_(std::move(fn)) {
// nop
}
take_while(take_while&&) = default;
take_while(const take_while&) = default;
take_while& operator=(take_while&&) = default;
take_while& operator=(const take_while&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (fn_(item)) {
return next.on_next(item, steps...);
} else {
next.on_complete(steps...);
return false;
}
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
Predicate fn_;
};
} // namespace caf::flow::step
...@@ -20,6 +20,7 @@ template <class> class [[nodiscard]] error_code; ...@@ -20,6 +20,7 @@ template <class> class [[nodiscard]] error_code;
template <class> class basic_cow_string; template <class> class basic_cow_string;
template <class> class behavior_type_of; template <class> class behavior_type_of;
template <class> class callback; template <class> class callback;
template <class> class cow_vector;
template <class> class dictionary; template <class> class dictionary;
template <class> class expected; template <class> class expected;
template <class> class intrusive_cow_ptr; template <class> class intrusive_cow_ptr;
......
...@@ -26,19 +26,17 @@ SCENARIO("an empty observable terminates normally") { ...@@ -26,19 +26,17 @@ SCENARIO("an empty observable terminates normally") {
GIVEN("an empty<int32>") { GIVEN("an empty<int32>") {
WHEN("an observer subscribes") { WHEN("an observer subscribes") {
THEN("the observer receives on_complete") { THEN("the observer receives on_complete") {
auto uut = ctx->make_observable().empty<int32_t>();
auto snk = flow::make_passive_observer<int32_t>(); auto snk = flow::make_passive_observer<int32_t>();
uut.subscribe(snk->as_observer()); ctx->make_observable().empty<int32_t>().subscribe(snk->as_observer());
ctx->run(); ctx->run();
if (CHECK(snk->sub)) { CHECK(snk->subscribed());
snk->sub.request(42); snk->request(42);
ctx->run(); ctx->run();
CHECK_EQ(snk->state, flow::observer_state::completed); CHECK(snk->completed());
CHECK(snk->buf.empty()); CHECK(snk->buf.empty());
} }
} }
} }
}
} }
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
...@@ -77,11 +77,9 @@ SCENARIO("for_each iterates all values in a stream") { ...@@ -77,11 +77,9 @@ SCENARIO("for_each iterates all values in a stream") {
.as_observable() .as_observable()
.take(7) .take(7)
.map([](int x) { return x * 3; }) .map([](int x) { return x * 3; })
.for_each([&outputs](int x) { outputs.emplace_back(x); }, .do_on_error([](const error& err) { FAIL("on_error: " << err); })
[](const error& reason) { .do_on_complete([&completed] { completed = true; })
FAIL("on_error called: " << reason); .for_each([&outputs](int x) { outputs.emplace_back(x); });
},
[&completed] { completed = true; });
ctx->run(); ctx->run();
CHECK(completed); CHECK(completed);
CHECK_EQ(inputs, outputs); CHECK_EQ(inputs, outputs);
......
...@@ -149,7 +149,7 @@ SCENARIO("callable sources stream values generated from a function object") { ...@@ -149,7 +149,7 @@ SCENARIO("callable sources stream values generated from a function object") {
namespace { namespace {
class custom_pullable { class custom_generator {
public: public:
using output_type = int; using output_type = int;
...@@ -171,25 +171,24 @@ private: ...@@ -171,25 +171,24 @@ private:
} // namespace } // namespace
SCENARIO("lifting converts a Pullable into an observable") { SCENARIO("lifting converts a generator into an observable") {
GIVEN("a lifted implementation of the Pullable concept") { GIVEN("a lifted implementation of the generator concept") {
WHEN("subscribing to its output") { WHEN("subscribing to its output") {
THEN("the observer receives the generated values") { THEN("the observer receives the generated values") {
using ivec = std::vector<int>; using ivec = std::vector<int>;
auto snk = flow::make_passive_observer<int>(); auto snk = flow::make_passive_observer<int>();
auto f = custom_pullable{}; auto f = custom_generator{};
ctx->make_observable().lift(f).subscribe(snk->as_observer()); ctx->make_observable().from_generator(f).subscribe(snk->as_observer());
CHECK_EQ(snk->state, flow::observer_state::subscribed); CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty()); CHECK(snk->buf.empty());
if (CHECK(snk->sub)) { CHECK(snk->subscribed());
snk->sub.request(3); snk->request(3);
ctx->run(); ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3})); CHECK_EQ(snk->buf, ivec({1, 2, 3}));
snk->sub.request(21); snk->sub.request(21);
ctx->run(); ctx->run();
CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7})); CHECK_EQ(snk->buf, ivec({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk->state, flow::observer_state::completed); CHECK(snk->completed());
}
} }
} }
} }
......
...@@ -34,7 +34,7 @@ auto ls(T x, Ts... xs) { ...@@ -34,7 +34,7 @@ auto ls(T x, Ts... xs) {
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("prefix_and_tail splits off initial elements") { SCENARIO("prefix_and_tail splits off initial elements") {
using tuple_t = cow_tuple<std::vector<int>, flow::observable<int>>; using tuple_t = cow_tuple<cow_vector<int>, flow::observable<int>>;
GIVEN("a generation with 0 values") { GIVEN("a generation with 0 values") {
WHEN("calling prefix_and_tail(2)") { WHEN("calling prefix_and_tail(2)") {
THEN("the observer of prefix_and_tail only receives on_complete") { THEN("the observer of prefix_and_tail only receives on_complete") {
......
...@@ -72,13 +72,12 @@ SCENARIO("response handles are convertible to observables and singles") { ...@@ -72,13 +72,12 @@ SCENARIO("response handles are convertible to observables and singles") {
auto [self, launch] = sys.spawn_inactive<event_based_actor>(); auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(dummy, infinite, int32_t{42}) self->request(dummy, infinite, int32_t{42})
.as_observable<int32_t>() .as_observable<int32_t>()
.for_each( .do_on_error([&](const error& what) { result = what; })
[&](int32_t val) { .do_on_complete([&] { completed = true; })
.for_each([&](int32_t val) {
result = val; result = val;
++on_next_calls; ++on_next_calls;
}, });
[&](const error& what) { result = what; },
[&] { completed = true; });
auto aut = actor{self}; auto aut = actor{self};
launch(); launch();
expect((int32_t), from(aut).to(dummy).with(42)); expect((int32_t), from(aut).to(dummy).with(42));
...@@ -117,13 +116,12 @@ SCENARIO("response handles are convertible to observables and singles") { ...@@ -117,13 +116,12 @@ SCENARIO("response handles are convertible to observables and singles") {
auto [self, launch] = sys.spawn_inactive<event_based_actor>(); auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(dummy, infinite, int32_t{13}) self->request(dummy, infinite, int32_t{13})
.as_observable<int32_t>() .as_observable<int32_t>()
.for_each( .do_on_error([&](const error& what) { result = what; })
[&](int32_t val) { .do_on_complete([&] { completed = true; })
.for_each([&](int32_t val) {
result = val; result = val;
++on_next_calls; ++on_next_calls;
}, });
[&](const error& what) { result = what; },
[&] { completed = true; });
auto aut = actor{self}; auto aut = actor{self};
launch(); launch();
expect((int32_t), from(aut).to(dummy).with(13)); expect((int32_t), from(aut).to(dummy).with(13));
......
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