Commit 3a449590 authored by Dominik Charousset's avatar Dominik Charousset

Implement zip_with operator

parent 4586e948
...@@ -56,6 +56,8 @@ caf_add_component( ...@@ -56,6 +56,8 @@ caf_add_component(
async.read_result async.read_result
async.write_result async.write_result
exit_reason exit_reason
flow.observable_state
flow.observer_state
intrusive.inbox_result intrusive.inbox_result
intrusive.task_result intrusive.task_result
invoke_message_result invoke_message_result
...@@ -146,7 +148,7 @@ caf_add_component( ...@@ -146,7 +148,7 @@ caf_add_component(
src/event_based_actor.cpp src/event_based_actor.cpp
src/execution_unit.cpp src/execution_unit.cpp
src/flow/coordinator.cpp src/flow/coordinator.cpp
src/flow/observable_builder src/flow/observable_builder.cpp
src/flow/scoped_coordinator.cpp src/flow/scoped_coordinator.cpp
src/flow/subscription.cpp src/flow/subscription.cpp
src/forwarding_actor_proxy.cpp src/forwarding_actor_proxy.cpp
...@@ -283,6 +285,7 @@ caf_add_component( ...@@ -283,6 +285,7 @@ caf_add_component(
dynamic_spawn dynamic_spawn
error error
expected expected
flow.broadcaster
flow.concat flow.concat
flow.concat_map flow.concat_map
flow.flat_map flow.flat_map
...@@ -292,6 +295,7 @@ caf_add_component( ...@@ -292,6 +295,7 @@ caf_add_component(
flow.observe_on flow.observe_on
flow.prefix_and_tail flow.prefix_and_tail
flow.single flow.single
flow.zip_with
function_view function_view
fused_downstream_manager fused_downstream_manager
handles handles
......
...@@ -237,6 +237,14 @@ struct IUnknown; ...@@ -237,6 +237,14 @@ struct IUnknown;
static_cast<void>(0) static_cast<void>(0)
#endif #endif
// CAF_DEBUG_STMT(stmt): evaluates to stmt when compiling with runtime checks
// and to an empty expression otherwise.
#ifndef CAF_ENABLE_RUNTIME_CHECKS
# define CAF_DEBUG_STMT(stmt) static_cast<void>(0)
#else
# define CAF_DEBUG_STMT(stmt) stmt
#endif
// Convenience macros. // Convenience macros.
#define CAF_IGNORE_UNUSED(x) static_cast<void>(x) #define CAF_IGNORE_UNUSED(x) static_cast<void>(x)
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/fwd.hpp" #include "caf/flow/fwd.hpp"
#include "caf/flow/observable_state.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/step.hpp" #include "caf/flow/step.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
...@@ -58,8 +59,23 @@ public: ...@@ -58,8 +59,23 @@ public:
observable as_observable() noexcept; observable as_observable() noexcept;
protected: /// Creates a subscription for `sink` and returns a @ref disposable to
disposable do_subscribe(observer_impl<T>* snk); /// cancel the observer.
disposable do_subscribe(observer_impl<T>* sink);
/// @copydoc do_subscribe
disposable do_subscribe(observer<T>& sink) {
return do_subscribe(sink.ptr());
}
/// Calls `on_error` on the `sink` with given `code` and returns a
/// default-constructed @ref disposable.
disposable reject_subscription(observer_impl<T>* sink, sec code);
/// @copydoc reject_subscription
disposable reject_subscription(observer<T>& sink, sec code) {
return reject_subscription(sink.ptr(), code);
}
}; };
class sub_impl final : public ref_counted, public subscription::impl { class sub_impl final : public ref_counted, public subscription::impl {
...@@ -288,6 +304,11 @@ public: ...@@ -288,6 +304,11 @@ public:
pimpl_.swap(other.pimpl_); pimpl_.swap(other.pimpl_);
} }
/// @pre `valid()`
coordinator* ctx() const {
return pimpl_->ctx();
}
private: private:
intrusive_ptr<impl> pimpl_; intrusive_ptr<impl> pimpl_;
}; };
...@@ -298,16 +319,73 @@ observable<T> observable<T>::impl::as_observable() noexcept { ...@@ -298,16 +319,73 @@ observable<T> observable<T>::impl::as_observable() noexcept {
} }
template <class T> template <class T>
disposable observable<T>::impl::do_subscribe(observer_impl<T>* snk) { disposable observable<T>::impl::do_subscribe(observer_impl<T>* sink) {
snk->on_subscribe(subscription{make_counted<sub_impl>(ctx(), this, snk)}); sink->on_subscribe(subscription{make_counted<sub_impl>(ctx(), this, sink)});
// Note: we do NOT return the subscription here because this object is private // Note: we do NOT return the subscription here because this object is private
// to the observer. Outside code must call dispose() on the observer. // to the observer. Outside code must call dispose() on the observer.
return disposable{intrusive_ptr<typename disposable::impl>{snk}}; return disposable{intrusive_ptr<typename disposable::impl>{sink}};
} }
template <class T>
disposable observable<T>::impl::reject_subscription(observer_impl<T>* sink, //
sec code) {
sink->on_error(make_error(code));
return disposable{};
}
/// @relates observable
template <class T> template <class T>
using observable_impl = typename observable<T>::impl; using observable_impl = typename observable<T>::impl;
/// Default base type for observable implementation types.
template <class T>
class observable_impl_base : public ref_counted, public observable_impl<T> {
public:
// -- member types -----------------------------------------------------------
using super = observable_impl<T>;
using output_type = T;
// -- constructors, destructors, and assignment operators --------------------
explicit observable_impl_base(coordinator* ctx) : ctx_(ctx) {
// nop
}
// -- implementation of disposable::impl -------------------------------------
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
// -- implementation of observable_impl<T> ---------------------------------
coordinator* ctx() const noexcept final {
return ctx_;
}
protected:
// -- member variables -------------------------------------------------------
coordinator* ctx_;
};
/// Convenience function for creating a sub-type of @ref observable_impl without
/// exposing the derived type.
template <class T, class... Ts>
intrusive_ptr<observable_impl<typename T::output_type>>
make_observable_impl(coordinator* ctx, Ts&&... xs) {
using out_t = typename T::output_type;
using res_t = intrusive_ptr<observable_impl<out_t>>;
auto ptr = new T(ctx, std::forward<Ts>(xs)...);
return res_t{static_cast<observable_impl<out_t>*>(ptr), false};
}
/// Base type for classes that represent a definition of an `observable` which /// Base type for classes that represent a definition of an `observable` which
/// has not yet been converted to an actual `observable`. /// has not yet been converted to an actual `observable`.
template <class T> template <class T>
...@@ -752,13 +830,13 @@ private: ...@@ -752,13 +830,13 @@ private:
template <class T> template <class T>
using buffered_observable_impl_ptr = intrusive_ptr<buffered_observable_impl<T>>; using buffered_observable_impl_ptr = intrusive_ptr<buffered_observable_impl<T>>;
template <class T> template <class Impl>
struct term_step { struct term_step {
buffered_observable_impl<T>* pimpl; Impl* pimpl;
using output_type = T; using output_type = typename Impl::output_type;
bool on_next(const T& item) { bool on_next(const output_type& item) {
pimpl->append_to_buf(item); pimpl->append_to_buf(item);
return true; return true;
} }
...@@ -923,17 +1001,100 @@ private: ...@@ -923,17 +1001,100 @@ private:
/// Broadcasts its input to all observers without modifying it. /// Broadcasts its input to all observers without modifying it.
template <class T> template <class T>
class broadcaster_impl : public buffered_processor_impl<T, T> { class broadcaster_impl : public ref_counted, public processor_impl<T, T> {
public: public:
using super = buffered_processor_impl<T, T>; // -- constructors, destructors, and assignment operators --------------------
using super::super; explicit broadcaster_impl(coordinator* ctx) : ctx_(ctx) {
// nop
}
private: // -- friends ----------------------------------------------------------------
bool do_on_next(span<const T> items) override {
this->append_to_buf(items.begin(), items.end()); CAF_INTRUSIVE_PTR_FRIENDS(broadcaster_impl)
return true;
// -- implementation of disposable::impl -------------------------------------
void dispose() override {
CAF_LOG_TRACE("");
term_.dispose();
} }
bool disposed() const noexcept override {
return !term_.active();
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observer<T>::impl ------------------------------------
void on_subscribe(subscription sub) override {
if (term_.start(sub))
sub_ = std::move(sub);
}
void on_next(span<const T> items) override {
term_.on_next(items);
term_.push();
}
void on_complete() override {
term_.on_complete();
}
void on_error(const error& what) override {
term_.on_error(what);
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<T>* sink, size_t n) override {
CAF_LOG_TRACE(CAF_ARG(n));
term_.on_request(sub_, sink, n);
}
void on_cancel(observer_impl<T>* sink) override {
CAF_LOG_TRACE("");
term_.on_cancel(sub_, sink);
}
disposable subscribe(observer<T> sink) override {
return term_.add(this, sink);
}
// -- properties -------------------------------------------------------------
size_t buffered() const noexcept {
return term_.buffered();
}
observable_state state() const noexcept {
return term_.state();
}
const error& err() const noexcept {
return term_.err();
}
protected:
/// Points to the coordinator that runs this observable.
coordinator* ctx_;
/// Allows us to request more items.
subscription sub_;
/// Pushes data to the observers.
broadcast_step<T> term_;
}; };
template <class T> template <class T>
...@@ -980,7 +1141,7 @@ public: ...@@ -980,7 +1141,7 @@ public:
void on_complete() override { void on_complete() override {
super::sub_ = nullptr; super::sub_ = nullptr;
auto f = [this](auto& step, auto&... steps) { auto f = [this](auto& step, auto&... steps) {
term_step<output_type> term{this}; term_step<impl> term{this};
step.on_complete(steps..., term); step.on_complete(steps..., term);
}; };
std::apply(f, steps); std::apply(f, steps);
...@@ -989,7 +1150,7 @@ public: ...@@ -989,7 +1150,7 @@ public:
void on_error(const error& what) override { void on_error(const error& what) override {
super::sub_ = nullptr; super::sub_ = nullptr;
auto f = [this, &what](auto& step, auto&... steps) { auto f = [this, &what](auto& step, auto&... steps) {
term_step<output_type> term{this}; term_step<impl> term{this};
step.on_error(what, steps..., term); step.on_error(what, steps..., term);
}; };
std::apply(f, steps); std::apply(f, steps);
...@@ -1000,7 +1161,7 @@ public: ...@@ -1000,7 +1161,7 @@ public:
private: private:
bool do_on_next(span<const input_type> items) override { bool do_on_next(span<const input_type> items) override {
auto f = [this, items](auto& step, auto&... steps) { auto f = [this, items](auto& step, auto&... steps) {
term_step<output_type> term{this}; term_step<impl> term{this};
for (auto&& item : items) for (auto&& item : items)
if (!step.on_next(item, steps..., term)) if (!step.on_next(item, steps..., term))
return false; return false;
...@@ -1970,4 +2131,195 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) { ...@@ -1970,4 +2131,195 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) {
} }
} }
// -- custom operators ---------------------------------------------------------
/// An observable that represents an empty range. As soon as an observer
/// requests values from this observable, it calls `on_complete`.
template <class T>
class empty_observable_impl : public ref_counted, public observable_impl<T> {
public:
// -- member types -----------------------------------------------------------
using output_type = T;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(empty_observable_impl)
// -- constructors, destructors, and assignment operators --------------------
explicit empty_observable_impl(coordinator* ctx) : ctx_(ctx) {
// nop
}
// -- implementation of disposable::impl -------------------------------------
void dispose() override {
// nop
}
bool disposed() const noexcept override {
return true;
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<output_type>* snk, size_t) override {
snk->on_complete();
}
void on_cancel(observer_impl<output_type>*) override {
// nop
}
disposable subscribe(observer<output_type> sink) override {
return this->do_subscribe(sink.ptr());
}
private:
coordinator* ctx_;
};
/// An observable with minimal internal logic. Useful for writing unit tests.
template <class T>
class passive_observable : public ref_counted, public observable_impl<T> {
public:
// -- member types -----------------------------------------------------------
using output_type = T;
using super = observable_impl<T>;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(passive_observable)
// -- constructors, destructors, and assignment operators --------------------
passive_observable(coordinator* ctx) : ctx_(ctx) {
// nop
}
// -- implementation of disposable::impl -------------------------------------
void dispose() override {
if (is_active(state)) {
demand = 0;
if (out) {
out.on_complete();
out = nullptr;
}
state = observable_state::completed;
}
}
bool disposed() const noexcept override {
return !is_active(state);
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable_impl<T> -----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<output_type>* sink, size_t n) override {
if (out.ptr() == sink) {
demand += n;
}
}
void on_cancel(observer_impl<output_type>* sink) override {
if (out.ptr() == sink) {
demand = 0;
out = nullptr;
state = observable_state::completed;
}
}
disposable subscribe(observer<output_type> sink) override {
if (state == observable_state::idle) {
state = observable_state::running;
out = sink;
return this->do_subscribe(sink.ptr());
} else if (is_final(state)) {
return this->reject_subscription(sink, sec::disposed);
} else {
return this->reject_subscription(sink, sec::too_many_observers);
}
}
// -- pushing items ----------------------------------------------------------
void push(span<const output_type> items) {
if (items.size() > demand)
CAF_RAISE_ERROR("observables must not emit more items than requested");
demand -= items.size();
out.on_next(items);
}
void push(const output_type& item) {
push(make_span(&item, 1));
}
void complete() {
if (is_active(state)) {
demand = 0;
if (out) {
out.on_complete();
out = nullptr;
}
state = observable_state::completed;
}
}
void abort(const error& reason) {
if (is_active(state)) {
demand = 0;
if (out) {
out.on_error(reason);
out = nullptr;
}
state = observable_state::aborted;
}
}
// -- member variables -------------------------------------------------------
observable_state state = observable_state::idle;
size_t demand = 0;
observer<T> out;
private:
coordinator* ctx_;
};
/// @relates passive_observable
template <class T>
intrusive_ptr<passive_observable<T>> make_passive_observable(coordinator* ctx) {
return make_counted<passive_observable<T>>(ctx);
}
} // namespace caf::flow } // namespace caf::flow
...@@ -130,6 +130,10 @@ public: ...@@ -130,6 +130,10 @@ public:
[[nodiscard]] observable<T> [[nodiscard]] observable<T>
from_resource(async::consumer_resource<T> res) const; from_resource(async::consumer_resource<T> res) const;
/// Creates an @ref observable that emits a sequence of integers spaced by the
/// @p period.
/// @param initial_delay Delay of the first integer after subscribing.
/// @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> [[nodiscard]] observable<int64_t>
interval(std::chrono::duration<Rep, Period> initial_delay, interval(std::chrono::duration<Rep, Period> initial_delay,
...@@ -141,6 +145,31 @@ public: ...@@ -141,6 +145,31 @@ public:
return observable<int64_t>{std::move(ptr)}; return observable<int64_t>{std::move(ptr)};
} }
/// Creates an @ref observable that emits a sequence of integers spaced by the
/// @p delay.
/// @param delay Time delay between two integer values.
template <class Rep, class Period>
[[nodiscard]] observable<int64_t>
interval(std::chrono::duration<Rep, Period> delay) {
return interval(delay, delay);
}
/// Creates an @ref observable that emits a single item after the @p delay.
template <class Rep, class Period>
[[nodiscard]] observable<int64_t>
timer(std::chrono::duration<Rep, Period> delay) {
auto ptr = make_counted<interval_impl>(ctx_, delay, delay, 1);
ctx_->watch(ptr->as_disposable());
return observable<int64_t>{std::move(ptr)};
}
/// Creates an @ref observable without any values that simply calls
/// `on_complete` after subscribing to it.
template <class T>
[[nodiscard]] observable<T> empty() {
return observable<T>{make_counted<empty_observable_impl<T>>(ctx_)};
}
private: private:
explicit observable_builder(coordinator* ctx) : ctx_(ctx) { explicit observable_builder(coordinator* ctx) : ctx_(ctx) {
// nop // nop
...@@ -272,9 +301,9 @@ class generation final ...@@ -272,9 +301,9 @@ class generation final
public: public:
using output_type = transform_processor_output_type_t<Generator, Steps...>; using output_type = transform_processor_output_type_t<Generator, Steps...>;
class impl : public buffered_observable_impl<output_type> { class impl : public observable_impl_base<output_type> {
public: public:
using super = buffered_observable_impl<output_type>; using super = observable_impl_base<output_type>;
template <class... Ts> template <class... Ts>
impl(coordinator* ctx, Generator gen, Ts&&... steps) impl(coordinator* ctx, Generator gen, Ts&&... steps)
...@@ -282,17 +311,88 @@ public: ...@@ -282,17 +311,88 @@ public:
// nop // nop
} }
// -- implementation of disposable::impl -----------------------------------
void dispose() override {
disposed_ = true;
if (out_) {
out_.on_complete();
out_ = nullptr;
}
}
bool disposed() const noexcept override {
return disposed_;
}
// -- implementation of observable_impl<T> ---------------------------------
disposable subscribe(observer<output_type> what) override {
if (out_) {
return super::reject_subscription(what, sec::too_many_observers);
} else if (disposed_) {
return super::reject_subscription(what, sec::disposed);
} else {
out_ = what;
return super::do_subscribe(what);
}
}
void on_request(observer_impl<output_type>* sink, size_t n) override {
if (sink == out_.ptr()) {
auto fn = [this, n](auto&... steps) {
term_step<impl> term{this};
gen_.pull(n, steps..., term);
};
std::apply(fn, steps_);
push();
}
}
void on_cancel(observer_impl<output_type>* sink) override {
if (sink == out_.ptr()) {
buf_.clear();
out_ = nullptr;
disposed_ = true;
}
}
// -- callbacks for term_step ----------------------------------------------
void append_to_buf(const output_type& item) {
CAF_ASSERT(out_.valid());
buf_.emplace_back(item);
}
void shutdown() {
CAF_ASSERT(out_.valid());
push();
out_.on_complete();
out_ = nullptr;
disposed_ = true;
}
void abort(const error& reason) {
CAF_ASSERT(out_.valid());
push();
out_.on_error(reason);
out_ = nullptr;
disposed_ = true;
}
private: private:
virtual void pull(size_t n) { void push() {
auto fn = [this, n](auto&... steps) { if (!buf_.empty()) {
term_step<output_type> term{this}; out_.on_next(make_span(buf_));
gen_.pull(n, steps..., term); buf_.clear();
}; }
std::apply(fn, steps_);
} }
Generator gen_; Generator gen_;
std::tuple<Steps...> steps_; std::tuple<Steps...> steps_;
observer<output_type> out_;
bool disposed_ = false;
std::vector<output_type> buf_;
}; };
template <class... Ts> template <class... Ts>
...@@ -339,10 +439,19 @@ public: ...@@ -339,10 +439,19 @@ public:
} }
observable<output_type> as_observable() && override { observable<output_type> as_observable() && override {
auto pimpl = make_counted<impl>(ctx_, std::move(gen_), std::move(steps_)); auto pimpl = make_observable_impl<impl>(ctx_, std::move(gen_),
std::move(steps_));
return observable<output_type>{std::move(pimpl)}; return observable<output_type>{std::move(pimpl)};
} }
coordinator* ctx() const noexcept {
return ctx_;
}
constexpr bool valid() const noexcept {
return true;
}
private: private:
coordinator* ctx_; coordinator* ctx_;
Generator gen_; Generator 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/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::flow {
/// Represents the current state of an @ref observable.
enum class observable_state {
/// Indicates that the observable is still waiting on some event or input.
idle,
/// Indicates that at least one observer subscribed.
running,
/// Indicates that the observable is waiting for observers to consume all
/// produced items before shutting down.
completing,
/// Indicates that the observable properly shut down.
completed,
/// Indicates that the observable shut down due to an error.
aborted,
/// Indicates that dispose was called.
disposed,
};
/// Returns whether `x` represents a final state, i.e., `completed`, `aborted`
/// or `disposed`.
constexpr bool is_final(observable_state x) noexcept {
return static_cast<int>(x) >= static_cast<int>(observable_state::completed);
}
/// Returns whether `x` represents an active state, i.e., `idle` or `running`.
constexpr bool is_active(observable_state x) noexcept {
return static_cast<int>(x) <= static_cast<int>(observable_state::running);
}
/// @relates observable_state
CAF_CORE_EXPORT std::string to_string(observable_state);
/// @relates observable_state
CAF_CORE_EXPORT bool from_string(string_view, observable_state&);
/// @relates observable_state
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<observable_state>,
observable_state&);
/// @relates observable_state
template <class Inspector>
bool inspect(Inspector& f, observable_state& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::flow
...@@ -11,6 +11,7 @@ ...@@ -11,6 +11,7 @@
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/observer_state.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
...@@ -140,148 +141,6 @@ private: ...@@ -140,148 +141,6 @@ private:
template <class T> template <class T>
using observer_impl = typename observer<T>::impl; using observer_impl = typename observer<T>::impl;
// -- writing observed values to a buffer --------------------------------------
/// Writes observed values to a bounded buffer.
template <class Buffer>
class buffer_writer_impl : public ref_counted,
public observer_impl<typename Buffer::value_type>,
public async::producer {
public:
// -- member types -----------------------------------------------------------
using buffer_ptr = intrusive_ptr<Buffer>;
using value_type = typename Buffer::value_type;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(buffer_writer_impl)
// -- constructors, destructors, and assignment operators --------------------
buffer_writer_impl(coordinator* ctx, buffer_ptr buf)
: ctx_(ctx), buf_(std::move(buf)) {
CAF_ASSERT(ctx_ != nullptr);
CAF_ASSERT(buf_ != nullptr);
}
~buffer_writer_impl() {
if (buf_)
buf_->close();
}
// -- implementation of disposable::impl -------------------------------------
void dispose() override {
CAF_LOG_TRACE("");
on_complete();
}
bool disposed() const noexcept override {
return buf_ == nullptr;
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
// -- implementation of observer<T>::impl ------------------------------------
void on_next(span<const value_type> items) override {
CAF_LOG_TRACE(CAF_ARG(items));
if (buf_)
buf_->push(items);
}
void on_complete() override {
CAF_LOG_TRACE("");
if (buf_) {
buf_->close();
buf_ = nullptr;
sub_ = nullptr;
}
}
void on_error(const error& what) override {
CAF_LOG_TRACE(CAF_ARG(what));
if (buf_) {
buf_->abort(what);
buf_ = nullptr;
sub_ = nullptr;
}
}
void on_subscribe(subscription sub) override {
CAF_LOG_TRACE("");
if (buf_ && !sub_) {
CAF_LOG_DEBUG("add subscription");
sub_ = std::move(sub);
} else {
CAF_LOG_DEBUG("already have a subscription or buffer no longer valid");
sub.cancel();
}
}
// -- implementation of async::producer: must be thread-safe -----------------
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
CAF_LOG_TRACE("");
ctx_->schedule_fn([ptr{strong_ptr()}] {
CAF_LOG_TRACE("");
ptr->on_cancel();
});
}
void on_consumer_demand(size_t demand) override {
CAF_LOG_TRACE(CAF_ARG(demand));
ctx_->schedule_fn([ptr{strong_ptr()}, demand] { //
CAF_LOG_TRACE(CAF_ARG(demand));
ptr->on_demand(demand);
});
}
void ref_producer() const noexcept final {
this->ref();
}
void deref_producer() const noexcept final {
this->deref();
}
private:
void on_demand(size_t n) {
CAF_LOG_TRACE(CAF_ARG(n));
if (sub_)
sub_.request(n);
}
void on_cancel() {
CAF_LOG_TRACE("");
if (sub_) {
sub_.cancel();
sub_ = nullptr;
}
buf_ = nullptr;
}
intrusive_ptr<buffer_writer_impl> strong_ptr() {
return {this};
}
coordinator_ptr ctx_;
buffer_ptr buf_;
subscription sub_;
};
} // namespace caf::flow } // namespace caf::flow
namespace caf::detail { namespace caf::detail {
...@@ -473,4 +332,316 @@ auto make_observer_from_ptr(SmartPointer ptr) { ...@@ -473,4 +332,316 @@ auto make_observer_from_ptr(SmartPointer ptr) {
[ptr] { ptr->on_complete(); }); [ptr] { ptr->on_complete(); });
} }
// -- writing observed values to an async buffer -------------------------------
/// Writes observed values to a bounded buffer.
template <class Buffer>
class buffer_writer_impl : public ref_counted,
public observer_impl<typename Buffer::value_type>,
public async::producer {
public:
// -- member types -----------------------------------------------------------
using buffer_ptr = intrusive_ptr<Buffer>;
using value_type = typename Buffer::value_type;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(buffer_writer_impl)
// -- constructors, destructors, and assignment operators --------------------
buffer_writer_impl(coordinator* ctx, buffer_ptr buf)
: ctx_(ctx), buf_(std::move(buf)) {
CAF_ASSERT(ctx_ != nullptr);
CAF_ASSERT(buf_ != nullptr);
}
~buffer_writer_impl() {
if (buf_)
buf_->close();
}
// -- implementation of disposable::impl -------------------------------------
void dispose() override {
CAF_LOG_TRACE("");
on_complete();
}
bool disposed() const noexcept override {
return buf_ == nullptr;
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
// -- implementation of observer<T>::impl ------------------------------------
void on_next(span<const value_type> items) override {
CAF_LOG_TRACE(CAF_ARG(items));
if (buf_)
buf_->push(items);
}
void on_complete() override {
CAF_LOG_TRACE("");
if (buf_) {
buf_->close();
buf_ = nullptr;
sub_ = nullptr;
}
}
void on_error(const error& what) override {
CAF_LOG_TRACE(CAF_ARG(what));
if (buf_) {
buf_->abort(what);
buf_ = nullptr;
sub_ = nullptr;
}
}
void on_subscribe(subscription sub) override {
CAF_LOG_TRACE("");
if (buf_ && !sub_) {
CAF_LOG_DEBUG("add subscription");
sub_ = std::move(sub);
} else {
CAF_LOG_DEBUG("already have a subscription or buffer no longer valid");
sub.cancel();
}
}
// -- implementation of async::producer: must be thread-safe -----------------
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
CAF_LOG_TRACE("");
ctx_->schedule_fn([ptr{strong_ptr()}] {
CAF_LOG_TRACE("");
ptr->on_cancel();
});
}
void on_consumer_demand(size_t demand) override {
CAF_LOG_TRACE(CAF_ARG(demand));
ctx_->schedule_fn([ptr{strong_ptr()}, demand] { //
CAF_LOG_TRACE(CAF_ARG(demand));
ptr->on_demand(demand);
});
}
void ref_producer() const noexcept final {
this->ref();
}
void deref_producer() const noexcept final {
this->deref();
}
private:
void on_demand(size_t n) {
CAF_LOG_TRACE(CAF_ARG(n));
if (sub_)
sub_.request(n);
}
void on_cancel() {
CAF_LOG_TRACE("");
if (sub_) {
sub_.cancel();
sub_ = nullptr;
}
buf_ = nullptr;
}
intrusive_ptr<buffer_writer_impl> strong_ptr() {
return {this};
}
coordinator_ptr ctx_;
buffer_ptr buf_;
subscription sub_;
};
// -- utility observer ---------------------------------------------------------
/// Forwards all events to a parent.
template <class T, class Parent, class Token = unit_t>
class forwarder : public ref_counted, public observer_impl<T> {
public:
CAF_INTRUSIVE_PTR_FRIENDS(forwarder)
explicit forwarder(intrusive_ptr<Parent> parent, Token token = Token{})
: parent(std::move(parent)), token(std::move(token)) {
// nop
}
void on_complete() override {
if (parent) {
if constexpr (std::is_same_v<Token, unit_t>)
parent->fwd_on_complete(this);
else
parent->fwd_on_complete(this, token);
parent = nullptr;
}
}
void on_error(const error& what) override {
if (parent) {
if constexpr (std::is_same_v<Token, unit_t>)
parent->fwd_on_error(this, what);
else
parent->fwd_on_error(this, token, what);
parent = nullptr;
}
}
void on_subscribe(subscription new_sub) override {
if (parent) {
if constexpr (std::is_same_v<Token, unit_t>)
parent->fwd_on_subscribe(this, std::move(new_sub));
else
parent->fwd_on_subscribe(this, token, std::move(new_sub));
} else {
new_sub.cancel();
}
}
void on_next(span<const T> items) override {
if (parent) {
if constexpr (std::is_same_v<Token, unit_t>)
parent->fwd_on_next(this, items);
else
parent->fwd_on_next(this, token, items);
}
}
void dispose() override {
on_complete();
}
bool disposed() const noexcept override {
return !parent;
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
intrusive_ptr<Parent> parent;
Token token;
};
/// An observer with minimal internal logic. Useful for writing unit tests.
template <class T>
class passive_observer : public ref_counted, public observer_impl<T> {
public:
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(passive_observer)
// -- implementation of disposable::impl -------------------------------------
void dispose() override {
if (!disposed()) {
if (sub) {
sub.cancel();
sub = nullptr;
}
state = observer_state::disposed;
}
}
bool disposed() const noexcept override {
switch (state) {
case observer_state::completed:
case observer_state::aborted:
case observer_state::disposed:
return true;
default:
return false;
}
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
// -- implementation of observer_impl<T> -------------------------------------
void on_complete() override {
if (!disposed()) {
if (sub) {
sub.cancel();
sub = nullptr;
}
state = observer_state::completed;
}
}
void on_error(const error& what) override {
if (!disposed()) {
if (sub) {
sub.cancel();
sub = nullptr;
}
err = what;
state = observer_state::aborted;
}
}
void on_subscribe(subscription new_sub) override {
if (state == observer_state::idle) {
CAF_ASSERT(!sub);
sub = std::move(new_sub);
state = observer_state::subscribed;
} else {
new_sub.cancel();
}
}
void on_next(span<const T> items) override {
buf.insert(buf.end(), items.begin(), items.end());
}
// -- member variables -------------------------------------------------------
/// The subscription for requesting additional items.
subscription sub;
/// Default-constructed unless on_error was called.
error err;
/// Represents the current state of this observer.
observer_state state;
/// Stores all items received via `on_next`.
std::vector<T> buf;
};
/// @relates passive_observer
template <class T>
intrusive_ptr<passive_observer<T>> make_passive_observer() {
return make_counted<passive_observer<T>>();
}
} // namespace caf::flow } // namespace caf::flow
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::flow {
/// Represents the current state of an @ref observer.
enum class observer_state {
/// Indicates that no callbacks were called yet.
idle,
/// Indicates that on_subscribe was called.
subscribed,
/// Indicates that on_complete was called.
completed,
/// Indicates that on_error was called.
aborted,
/// Indicates that dispose was called.
disposed,
};
/// @relates sec
CAF_CORE_EXPORT std::string to_string(observer_state);
/// @relates observer_state
CAF_CORE_EXPORT bool from_string(string_view, observer_state&);
/// @relates observer_state
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<observer_state>,
observer_state&);
/// @relates observer_state
template <class Inspector>
bool inspect(Inspector& f, observer_state& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::flow
...@@ -4,10 +4,15 @@ ...@@ -4,10 +4,15 @@
#pragma once #pragma once
#include <type_traits>
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.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 <type_traits>
namespace caf::flow { namespace caf::flow {
...@@ -248,4 +253,328 @@ struct on_error_complete_step { ...@@ -248,4 +253,328 @@ struct on_error_complete_step {
} }
}; };
/// Wraps logic for pushing data to multiple observers with broadcast semantics,
/// i.e., all observers see the same items at the same time and the flow adjusts
/// to the slowest observer. This step may only be used as terminal step.
template <class T>
class broadcast_step {
public:
// -- member types -----------------------------------------------------------
using output_type = T;
using observer_impl_t = observer_impl<output_type>;
struct output_t {
size_t demand;
observer<output_type> sink;
};
// -- constructors, destructors, and assignment operators --------------------
broadcast_step() {
// Reserve some buffer space in order to avoid frequent re-allocations while
// warming up.
buf_.reserve(32);
}
// -- properties -------------------------------------------------------------
size_t min_demand() const noexcept {
if (!outputs_.empty()) {
auto i = outputs_.begin();
auto init = (*i++).demand;
return std::accumulate(i, outputs_.end(), init,
[](size_t x, const output_t& y) {
return std::min(x, y.demand);
});
} else {
return 0;
}
}
size_t max_demand() const noexcept {
if (!outputs_.empty()) {
auto i = outputs_.begin();
auto init = (*i++).demand;
return std::accumulate(i, outputs_.end(), init,
[](size_t x, const output_t& y) {
return std::max(x, y.demand);
});
} else {
return 0;
}
}
/// Returns how many items are currently buffered at this step.
size_t buffered() const noexcept {
return buf_.size();
}
/// Returns the number of current observers.
size_t num_observers() const noexcept {
return outputs_.size();
}
/// Convenience function for calling `is_active(state())`;
bool active() const noexcept {
return is_active(state_);
}
/// Queries whether the current state is `observable_state::completing`.
bool completing() const noexcept {
return state_ == observable_state::completing;
}
/// Convenience function for calling `is_final(state())`;
bool finalized() const noexcept {
return is_final(state_);
}
/// Returns the current state.
observable_state state() const noexcept {
return state_;
}
const error& err() const noexcept {
return err_;
}
void err(error x) {
err_ = std::move(x);
}
// -- demand management ------------------------------------------------------
size_t next_demand() {
auto have = buf_.size() + in_flight_;
auto want = max_demand();
if (want > have) {
auto delta = want - have;
in_flight_ += delta;
return delta;
} else {
return 0;
}
}
// -- callbacks for the parent -----------------------------------------------
/// Tries to add a new observer.
bool add(observer<output_type> sink) {
if (is_active(state_)) {
outputs_.emplace_back(output_t{0, std::move(sink)});
return true;
} else if (err_) {
sink.on_error(err_);
return false;
} else {
sink.on_error(make_error(sec::disposed));
return false;
}
}
/// Tries to add a new observer and returns `parent->do_subscribe(sink)` on
/// success or a default-constructed @ref disposable otherwise.
template <class Parent>
disposable add(Parent* parent, observer<output_type> sink) {
if (add(sink)) {
return parent->do_subscribe(sink);
} else {
return disposable{};
}
}
/// Requests `n` more items for `sink`.
/// @returns New demand to signal upstream or 0.
/// @note Calls @ref push.
size_t on_request(observer_impl_t* sink, size_t n) {
if (auto i = find(sink); i != outputs_.end()) {
i->demand += n;
push();
return next_demand();
} else {
return 0;
}
}
/// Requests `n` more items for `sink`.
/// @note Calls @ref push and may call `sub.request(n)`.
void on_request(subscription& sub, observer_impl_t* sink, size_t n) {
if (auto new_demand = on_request(sink, n); new_demand > 0 && sub)
sub.request(new_demand);
}
/// Removes `sink` from the observer set.
/// @returns New demand to signal upstream or 0.
/// @note Calls @ref push.
size_t on_cancel(observer_impl_t* sink) {
if (auto i = find(sink); i != outputs_.end()) {
outputs_.erase(i);
push();
return next_demand();
} else {
return 0;
}
}
/// Requests `n` more items for `sink`.
/// @note Calls @ref push and may call `sub.request(n)`.
void on_cancel(subscription& sub, observer_impl_t* sink) {
if (auto new_demand = on_cancel(sink); new_demand > 0 && sub)
sub.request(new_demand);
}
/// Tries to deliver items from the buffer to the observers.
void push() {
// Must not be re-entered. Any on_request call must use the event loop.
CAF_ASSERT(!pushing_);
CAF_DEBUG_STMT(pushing_ = true);
// Sanity checking.
if (outputs_.empty())
return;
// Push data downstream and adjust demand on each path.
if (auto n = std::min(min_demand(), buf_.size()); n > 0) {
auto items = span<output_type>{buf_.data(), n};
for (auto& out : outputs_) {
out.demand -= n;
out.sink.on_next(items);
}
buf_.erase(buf_.begin(), buf_.begin() + n);
}
if (state_ == observable_state::completing && buf_.empty()) {
if (!err_) {
for (auto& out : outputs_)
out.sink.on_complete();
state_ = observable_state::completed;
} else {
for (auto& out : outputs_)
out.sink.on_error(err_);
state_ = observable_state::aborted;
}
}
CAF_DEBUG_STMT(pushing_ = false);
}
/// Checks whether the broadcaster currently has no pending data.
bool idle() {
return buf_.empty();
}
/// Calls `on_complete` on all observers and drops any pending data.
void close() {
buf_.clear();
if (!err_)
for (auto& out : outputs_)
out.sink.on_complete();
else
for (auto& out : outputs_)
out.sink.on_error(err_);
outputs_.clear();
}
/// Calls `on_error` on all observers and drops any pending data.
void abort(const error& reason) {
err_ = reason;
close();
}
// -- callbacks for steps ----------------------------------------------------
bool on_next(const output_type& item) {
// Note: we may receive more data than what we have requested.
if (in_flight_ > 0)
--in_flight_;
buf_.emplace_back(item);
return true;
}
void on_next(span<const output_type> items) {
// Note: we may receive more data than what we have requested.
if (in_flight_ >= items.size())
in_flight_ -= items.size();
else
in_flight_ = 0;
buf_.insert(buf_.end(), items.begin(), items.end());
}
void fin() {
if (is_active(state_)) {
if (idle()) {
close();
state_ = err_ ? observable_state::aborted : observable_state::completed;
} else {
state_ = observable_state::completing;
}
}
}
void on_complete() {
fin();
}
void on_error(const error& what) {
err_ = what;
fin();
}
// -- callbacks for the parent -----------------------------------------------
void dispose() {
on_complete();
}
/// Tries to set the state from `idle` to `running`.
bool start() {
if (state_ == observable_state::idle) {
state_ = observable_state::running;
return true;
} else {
return false;
}
}
/// Tries to set the state from `idle` to `running`. On success, requests
/// items on `sub` if there is already demand. Calls `sub.cancel()` when
/// returning `false`.
bool start(subscription& sub) {
if (start()) {
if (auto n = next_demand(); n > 0)
sub.request(n);
return true;
} else {
sub.cancel();
return false;
}
}
private:
typename std::vector<output_t>::iterator
find(observer_impl<output_type>* sink) {
auto e = outputs_.end();
auto pred = [sink](const output_t& out) { return out.sink.ptr() == sink; };
return std::find_if(outputs_.begin(), e, pred);
}
/// Buffers outbound items until we can ship them.
std::vector<output_type> buf_;
/// Keeps track of how many items have been requested but did not arrive yet.
size_t in_flight_ = 0;
/// Stores handles to the observer plus their demand.
std::vector<output_t> outputs_;
/// Keeps track of our current state.
observable_state state_ = observable_state::idle;
/// Stores the on_error argument.
error err_;
#ifdef CAF_ENABLE_RUNTIME_CHECKS
/// Protect against re-entering `push`.
bool pushing_ = false;
#endif
};
} // namespace caf::flow } // namespace caf::flow
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/async/policy.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/observable_state.hpp"
#include "caf/flow/observer.hpp"
#include <type_traits>
#include <utility>
namespace caf::flow {
template <class F, class... Ts>
struct zipper_oracle {
using output_type
= decltype(std::declval<F&>()(std::declval<const Ts&>()...));
};
template <class F, class... Ts>
using zipper_output_t = typename zipper_oracle<F, Ts...>::output_type;
template <size_t Index>
using zipper_index = std::integral_constant<size_t, Index>;
template <class T>
struct zipper_input {
using value_type = T;
explicit zipper_input(observable<T> in) : in(std::move(in)) {
// nop
}
observable<T> in;
subscription sub;
std::vector<T> buf;
};
/// Combines items from any number of observables using a zip function.
template <class F, class... Ts>
class zipper_impl : public ref_counted,
public observable_impl<zipper_output_t<F, Ts...>> {
public:
// -- member types -----------------------------------------------------------
using output_type = zipper_output_t<F, Ts...>;
using super = observable_impl<output_type>;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(zipper_impl)
template <class, class, class>
friend class forwarder;
// -- constructors, destructors, and assignment operators --------------------
zipper_impl(coordinator* ctx, F fn, observable<Ts>... inputs)
: ctx_(ctx), fn_(std::move(fn)), inputs_(inputs...) {
// nop
}
// -- implementation of disposable::impl -------------------------------------
void dispose() override {
broken_ = true;
if (buffered() == 0) {
fin();
} else {
for_each_input([](auto, auto& input) {
input.in = nullptr;
if (input.sub) {
input.sub.cancel();
input.sub = nullptr;
}
// Do not clear the buffer to allow already arrived items to go through.
});
}
}
bool disposed() const noexcept override {
return term_.finalized();
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<output_type>* sink, size_t demand) override {
if (auto n = term_.on_request(sink, demand); n > 0) {
demand_ += n;
for_each_input([n](auto, auto& input) {
if (input.sub)
input.sub.request(n);
});
}
}
void on_cancel(observer_impl<output_type>* sink) override {
if (auto n = term_.on_cancel(sink); n > 0) {
demand_ += n;
for_each_input([n](auto, auto& input) {
if (input.sub)
input.sub.request(n);
});
}
}
disposable subscribe(observer<output_type> sink) override {
// On the first subscribe, we subscribe to our inputs.
auto res = term_.add(this, sink);
if (res && term_.start()) {
for_each_input([this](auto index, auto& input) {
using input_t = std::decay_t<decltype(input)>;
using value_t = typename input_t::value_type;
using fwd_impl = forwarder<value_t, zipper_impl, decltype(index)>;
auto fwd = make_counted<fwd_impl>(this, index);
input.in.subscribe(fwd->as_observer());
});
}
return res;
}
private:
template <size_t I>
auto& at(zipper_index<I>) {
return std::get<I>(inputs_);
}
template <class Fn, size_t... Is>
void for_each_input(Fn&& fn, std::index_sequence<Is...>) {
(fn(zipper_index<Is>{}, at(zipper_index<Is>{})), ...);
}
template <class Fn>
void for_each_input(Fn&& fn) {
for_each_input(std::forward<Fn>(fn), std::index_sequence_for<Ts...>{});
}
template <class Fn, size_t... Is>
auto fold(Fn&& fn, std::index_sequence<Is...>) {
return fn(at(zipper_index<Is>{})...);
}
template <class Fn>
auto fold(Fn&& fn) {
return fold(std::forward<Fn>(fn), std::index_sequence_for<Ts...>{});
}
size_t buffered() {
return fold([](auto&... x) { return std::min({x.buf.size()...}); });
}
template <size_t I>
void fwd_on_subscribe(ref_counted*, zipper_index<I> index, subscription sub) {
if (!term_.finalized()) {
auto& in = at(index);
if (!in.sub) {
if (demand_ > 0)
sub.request(demand_);
in.sub = std::move(sub);
} else {
sub.cancel();
}
} else {
sub.cancel();
}
}
template <size_t I>
void fwd_on_complete(ref_counted*, zipper_index<I> index) {
if (!broken_) {
broken_ = true;
if (at(index).buf.empty())
fin();
}
at(index).sub = nullptr;
}
template <size_t I>
void fwd_on_error(ref_counted*, zipper_index<I> index, const error& what) {
if (!term_.err())
term_.err(what);
if (!broken_) {
broken_ = true;
if (at(index).buf.empty())
fin();
}
at(index).sub = nullptr;
}
template <size_t I, class T>
void fwd_on_next(ref_counted*, zipper_index<I> index, span<const T> items) {
if (!term_.finalized()) {
auto& buf = at(index).buf;
buf.insert(buf.end(), items.begin(), items.end());
push();
}
}
void push() {
if (auto n = std::min(buffered(), demand_); n > 0) {
for (size_t index = 0; index < n; ++index) {
fold([this, index](auto&... x) { //
term_.on_next(fn_(x.buf[index]...));
});
}
demand_ -= n;
for_each_input([n](auto, auto& x) { //
x.buf.erase(x.buf.begin(), x.buf.begin() + n);
});
term_.push();
if (broken_ && buffered() == 0)
fin();
}
}
void fin() {
for_each_input([](auto, auto& input) {
input.in = nullptr;
if (input.sub) {
input.sub.cancel();
input.sub = nullptr;
}
input.buf.clear();
});
term_.fin();
}
coordinator* ctx_;
size_t demand_ = 0;
F fn_;
std::tuple<zipper_input<Ts>...> inputs_;
/// A zipper breaks as soon as one of its inputs signals completion or error.
bool broken_ = false;
broadcast_step<output_type> term_;
};
template <class F, class... Ts>
using zipper_impl_ptr = intrusive_ptr<zipper_impl<F, Ts...>>;
/// @param fn The zip function. Takes one element from each input at a time and
/// converts them into a single result.
/// @param input0 The input at index 0.
/// @param input1 The input at index 1.
/// @param inputs The inputs for index > 1.
template <class F, class T0, class T1, class... Ts>
auto zip_with(F fn, T0 input0, T1 input1, Ts... inputs) {
using output_type = zipper_output_t<F, typename T0::output_type, //
typename T1::output_type, //
typename Ts::output_type...>;
using impl_t = zipper_impl<F, //
typename T0::output_type, //
typename T1::output_type, //
typename Ts::output_type...>;
if (input0.valid() && input1.valid() && (inputs.valid() && ...)) {
auto ctx = input0.ctx();
auto ptr = make_counted<impl_t>(ctx, std::move(fn),
std::move(input0).as_observable(),
std::move(input1).as_observable(),
std::move(inputs).as_observable()...);
return observable<output_type>{std::move(ptr)};
} else {
return observable<output_type>{};
}
}
} // namespace caf::flow
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.broadcaster
#include "caf/flow/observable.hpp"
#include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp"
namespace caf::flow {} // namespace caf::flow
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("a broadcaster pushes items to all subscribers at the same time") {
GIVEN("a broadcaster with one source and three sinks") {
auto uut = make_counted<flow::broadcaster_impl<int>>(ctx.get());
auto src = flow::make_passive_observable<int>(ctx.get());
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto snk3 = flow::make_passive_observer<int>();
src->subscribe(uut->as_observer());
uut->subscribe(snk1->as_observer());
uut->subscribe(snk2->as_observer());
uut->subscribe(snk3->as_observer());
WHEN("the source emits 10 items") {
THEN("the broadcaster forwards them to all sinks") {
snk1->sub.request(13);
ctx->run();
CHECK_EQ(src->demand, 13u);
snk2->sub.request(7);
ctx->run();
CHECK_EQ(src->demand, 13u);
snk3->sub.request(21);
ctx->run();
CHECK_EQ(src->demand, 21u);
auto inputs = std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
src->push(make_span(inputs));
CHECK_EQ(src->demand, 11u);
CHECK_EQ(uut->buffered(), 3u);
CHECK_EQ(snk1->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk2->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk3->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
snk2->sub.request(7);
ctx->run();
CHECK_EQ(src->demand, 11u);
CHECK_EQ(uut->buffered(), 0u);
CHECK_EQ(snk1->buf, inputs);
CHECK_EQ(snk2->buf, inputs);
CHECK_EQ(snk3->buf, inputs);
snk2->sub.request(14);
ctx->run();
CHECK_EQ(src->demand, 18u);
}
}
}
}
SCENARIO("a broadcaster emits values before propagating completion") {
GIVEN("a broadcaster with one source and three sinks") {
auto uut = make_counted<flow::broadcaster_impl<int>>(ctx.get());
auto src = flow::make_passive_observable<int>(ctx.get());
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto snk3 = flow::make_passive_observer<int>();
src->subscribe(uut->as_observer());
uut->subscribe(snk1->as_observer());
uut->subscribe(snk2->as_observer());
uut->subscribe(snk3->as_observer());
WHEN("the source emits 10 items and then signals completion") {
THEN("the broadcaster forwards all values before signaling an error") {
snk1->sub.request(13);
snk2->sub.request(7);
snk3->sub.request(21);
ctx->run();
CHECK_EQ(src->demand, 21u);
auto inputs = std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
src->push(make_span(inputs));
src->complete();
CHECK_EQ(uut->buffered(), 3u);
CHECK_EQ(snk1->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk2->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk3->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(uut->state(), flow::observable_state::completing);
CHECK_EQ(snk1->state, flow::observer_state::subscribed);
CHECK_EQ(snk2->state, flow::observer_state::subscribed);
CHECK_EQ(snk3->state, flow::observer_state::subscribed);
snk2->sub.request(7);
ctx->run();
CHECK_EQ(snk1->buf, inputs);
CHECK_EQ(snk2->buf, inputs);
CHECK_EQ(snk3->buf, inputs);
CHECK_EQ(uut->state(), flow::observable_state::completed);
CHECK_EQ(snk1->state, flow::observer_state::completed);
CHECK_EQ(snk2->state, flow::observer_state::completed);
CHECK_EQ(snk3->state, flow::observer_state::completed);
}
}
}
}
SCENARIO("a broadcaster emits values before propagating errors") {
GIVEN("a broadcaster with one source and three sinks") {
auto uut = make_counted<flow::broadcaster_impl<int>>(ctx.get());
auto src = flow::make_passive_observable<int>(ctx.get());
auto snk1 = flow::make_passive_observer<int>();
auto snk2 = flow::make_passive_observer<int>();
auto snk3 = flow::make_passive_observer<int>();
src->subscribe(uut->as_observer());
uut->subscribe(snk1->as_observer());
uut->subscribe(snk2->as_observer());
uut->subscribe(snk3->as_observer());
WHEN("the source emits 10 items and then stops with an error") {
THEN("the broadcaster forwards all values before signaling an error") {
snk1->sub.request(13);
snk2->sub.request(7);
snk3->sub.request(21);
ctx->run();
CHECK_EQ(src->demand, 21u);
auto inputs = std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
src->push(make_span(inputs));
auto runtime_error = make_error(sec::runtime_error);
src->abort(runtime_error);
CHECK_EQ(uut->buffered(), 3u);
CHECK_EQ(snk1->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk2->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(snk3->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7}));
CHECK_EQ(uut->state(), flow::observable_state::completing);
CHECK_EQ(uut->err(), runtime_error);
CHECK_EQ(snk1->state, flow::observer_state::subscribed);
CHECK_EQ(snk2->state, flow::observer_state::subscribed);
CHECK_EQ(snk3->state, flow::observer_state::subscribed);
snk2->sub.request(7);
ctx->run();
CHECK_EQ(snk1->buf, inputs);
CHECK_EQ(snk2->buf, inputs);
CHECK_EQ(snk3->buf, inputs);
CHECK_EQ(uut->state(), flow::observable_state::aborted);
CHECK_EQ(snk1->state, flow::observer_state::aborted);
CHECK_EQ(snk2->state, flow::observer_state::aborted);
CHECK_EQ(snk3->state, flow::observer_state::aborted);
CHECK_EQ(uut->err(), runtime_error);
CHECK_EQ(snk1->err, runtime_error);
CHECK_EQ(snk2->err, runtime_error);
CHECK_EQ(snk3->err, runtime_error);
}
}
}
}
END_FIXTURE_SCOPE()
...@@ -14,6 +14,8 @@ using namespace std::literals; ...@@ -14,6 +14,8 @@ using namespace std::literals;
using namespace caf; using namespace caf;
using i64_list = std::vector<int64_t>;
namespace { namespace {
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<> {
...@@ -25,7 +27,6 @@ struct fixture : test_coordinator_fixture<> { ...@@ -25,7 +27,6 @@ struct fixture : test_coordinator_fixture<> {
CAF_TEST_FIXTURE_SCOPE(interval_tests, fixture) CAF_TEST_FIXTURE_SCOPE(interval_tests, fixture)
SCENARIO("scoped coordinators wait on observable intervals") { SCENARIO("scoped coordinators wait on observable intervals") {
using i64_list = std::vector<int64_t>;
GIVEN("an observable interval") { GIVEN("an observable interval") {
WHEN("an observer subscribes to it") { WHEN("an observer subscribes to it") {
THEN("the coordinator blocks the current thread for the delays") { THEN("the coordinator blocks the current thread for the delays") {
...@@ -41,8 +42,7 @@ SCENARIO("scoped coordinators wait on observable intervals") { ...@@ -41,8 +42,7 @@ SCENARIO("scoped coordinators wait on observable intervals") {
} }
} }
SCENARIO("scheduled actors schedule observable intervals delays") { SCENARIO("scheduled actors schedule observable intervals on the actor clock") {
using i64_list = std::vector<int64_t>;
GIVEN("an observable interval") { GIVEN("an observable interval") {
WHEN("an observer subscribes to it") { WHEN("an observer subscribes to it") {
THEN("the actor uses the actor clock to schedule flow processing") { THEN("the actor uses the actor clock to schedule flow processing") {
...@@ -78,4 +78,19 @@ SCENARIO("scheduled actors schedule observable intervals delays") { ...@@ -78,4 +78,19 @@ SCENARIO("scheduled actors schedule observable intervals delays") {
} }
} }
SCENARIO("a timer is an observable interval with a single value") {
GIVEN("an observable timer") {
WHEN("an observer subscribes to it") {
THEN("the coordinator observes a single value") {
auto outputs = i64_list{};
ctx->make_observable()
.timer(10ms) //
.for_each([&outputs](int64_t x) { outputs.emplace_back(x); });
ctx->run();
CHECK_EQ(outputs, i64_list({0}));
}
}
}
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE flow.zip_with
#include "caf/flow/zip_with.hpp"
#include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("zip_with combines inputs") {
GIVEN("two observables") {
WHEN("merging them with zip_with") {
THEN("the observer receives the combined output of both sources") {
auto o1 = flow::make_passive_observer<int>();
auto fn = [](int x, int y) { return x + y; };
auto r1 = ctx->make_observable().repeat(11).take(113);
auto r2 = ctx->make_observable().repeat(22).take(223);
flow::zip_with(fn, std::move(r1), std::move(r2))
.subscribe(o1->as_observer());
ctx->run();
REQUIRE_EQ(o1->state, flow::observer_state::subscribed);
o1->sub.request(64);
ctx->run();
CHECK_EQ(o1->state, flow::observer_state::subscribed);
CHECK_EQ(o1->buf.size(), 64u);
o1->sub.request(64);
ctx->run();
CHECK_EQ(o1->state, flow::observer_state::completed);
CHECK_EQ(o1->buf.size(), 113u);
CHECK(std::all_of(o1->buf.begin(), o1->buf.begin(),
[](int x) { return x == 33; }));
}
}
}
}
SCENARIO("zip_with emits nothing when zipping an empty observable") {
GIVEN("two observables, one of them empty") {
WHEN("merging them with zip_with") {
THEN("the observer sees on_complete immediately") {
auto o1 = flow::make_passive_observer<int>();
auto fn = [](int x, int y, int z) { return x + y + z; };
auto r1 = ctx->make_observable().repeat(11);
auto r2 = ctx->make_observable().repeat(22);
auto r3 = ctx->make_observable().empty<int>();
flow::zip_with(fn, std::move(r1), std::move(r2), std::move(r3))
.subscribe(o1->as_observer());
ctx->run();
REQUIRE_EQ(o1->state, flow::observer_state::subscribed);
o1->sub.request(64);
ctx->run();
CHECK_EQ(o1->state, flow::observer_state::completed);
CHECK(o1->buf.empty());
}
}
}
}
SCENARIO("zip_with may only have more than one subscriber") {
GIVEN("two observables") {
WHEN("merging them with zip_with") {
THEN("all observer receives the combined output of both sources") {
auto o1 = flow::make_passive_observer<int>();
auto o2 = flow::make_passive_observer<int>();
auto fn = [](int x, int y) { return x + y; };
auto r1 = ctx->make_observable().repeat(11).take(113);
auto r2 = ctx->make_observable().repeat(22).take(223);
auto zip = flow::zip_with(fn, std::move(r1), std::move(r2));
zip.subscribe(o1->as_observer());
zip.subscribe(o2->as_observer());
ctx->run();
REQUIRE_EQ(o1->state, flow::observer_state::subscribed);
o1->sub.request(64);
o2->sub.request(64);
ctx->run();
CHECK_EQ(o1->state, flow::observer_state::subscribed);
CHECK_EQ(o1->buf.size(), 64u);
CHECK_EQ(o2->buf.size(), 64u);
o1->sub.request(64);
o2->dispose();
ctx->run();
CHECK_EQ(o1->state, flow::observer_state::completed);
CHECK_EQ(o1->buf.size(), 113u);
CHECK(std::all_of(o1->buf.begin(), o1->buf.begin(),
[](int x) { return x == 33; }));
CHECK_EQ(o2->buf.size(), 64u);
CHECK(std::all_of(o2->buf.begin(), o2->buf.begin(),
[](int x) { return x == 33; }));
}
}
}
}
END_FIXTURE_SCOPE()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment