Commit 45bfe5f2 authored by Dominik Charousset's avatar Dominik Charousset

Implement the publish operator

parent 36d5a8bd
...@@ -289,6 +289,7 @@ caf_add_component( ...@@ -289,6 +289,7 @@ caf_add_component(
flow.never flow.never
flow.observe_on flow.observe_on
flow.prefix_and_tail flow.prefix_and_tail
flow.publish
flow.single flow.single
flow.zip_with flow.zip_with
function_view function_view
......
...@@ -6,12 +6,13 @@ ...@@ -6,12 +6,13 @@
#include "caf/allowed_unsafe_message_type.hpp" #include "caf/allowed_unsafe_message_type.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <atomic> #include <atomic>
#include <cstddef>
namespace caf { namespace caf {
...@@ -51,6 +52,11 @@ public: ...@@ -51,6 +52,11 @@ public:
action& operator=(const action&) noexcept = default; action& operator=(const action&) noexcept = default;
action& operator=(std::nullptr_t) noexcept {
pimpl_ = nullptr;
return *this;
}
// -- observers -------------------------------------------------------------- // -- observers --------------------------------------------------------------
[[nodiscard]] bool disposed() const { [[nodiscard]] bool disposed() const {
...@@ -100,6 +106,14 @@ public: ...@@ -100,6 +106,14 @@ public:
return pimpl_; return pimpl_;
} }
explicit operator bool() const noexcept {
return static_cast<bool>(pimpl_);
}
[[nodiscard]] bool operator!() const noexcept {
return !pimpl_;
}
private: private:
impl_ptr pimpl_; impl_ptr pimpl_;
}; };
...@@ -108,7 +122,7 @@ private: ...@@ -108,7 +122,7 @@ private:
namespace caf::detail { namespace caf::detail {
template <class F> template <class F>
struct default_action_impl : ref_counted, action::impl { struct default_action_impl : detail::atomic_ref_counted, action::impl {
std::atomic<action::state> state_; std::atomic<action::state> state_;
F f_; F f_;
......
...@@ -38,6 +38,9 @@ class observer; ...@@ -38,6 +38,9 @@ class observer;
template <class T> template <class T>
class observable; class observable;
template <class T>
class connectable;
template <class In, class Out> template <class In, class Out>
class processor; class processor;
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#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/prefix_and_tail.hpp"
#include "caf/flow/op/publish.hpp"
#include "caf/flow/step.hpp" #include "caf/flow/step.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
...@@ -83,6 +84,16 @@ public: ...@@ -83,6 +84,16 @@ public:
return alloc().concat_map(std::move(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>>> observable<cow_tuple<std::vector<T>, observable<T>>>
prefix_and_tail(size_t prefix_size) && { prefix_and_tail(size_t prefix_size) && {
return alloc().prefix_and_tail(prefix_size); return alloc().prefix_and_tail(prefix_size);
...@@ -127,6 +138,143 @@ private: ...@@ -127,6 +138,143 @@ private:
} }
}; };
// -- connectable --------------------------------------------------------------
/// Resembles a regular @ref observable, except that it does not begin emitting
/// items when it is subscribed to. Only after calling `connect` will the
/// `connectable` start to emit items.
template <class T>
class connectable {
public:
/// The type of emitted items.
using output_type = T;
/// The pointer-to-implementation type.
using pimpl_type = intrusive_ptr<op::publish<T>>;
explicit connectable(pimpl_type pimpl) noexcept : pimpl_(std::move(pimpl)) {
// nop
}
connectable& operator=(std::nullptr_t) noexcept {
pimpl_.reset();
return *this;
}
connectable() noexcept = default;
connectable(connectable&&) noexcept = default;
connectable(const connectable&) noexcept = default;
connectable& operator=(connectable&&) noexcept = default;
connectable& operator=(const connectable&) noexcept = default;
/// Returns an @ref observable that automatically connects to this
/// `connectable` when reaching `subscriber_threshold` subscriptions.
observable<T> auto_connect(size_t subscriber_threshold = 1) & {
auto ptr = make_counted<op::publish<T>>(ctx(), pimpl_);
ptr->auto_connect_threshold(subscriber_threshold);
return observable<T>{ptr};
}
/// Similar to the `lvalue` overload, but converts this `connectable` directly
/// if possible, thus saving one hop on the pipeline.
observable<T> auto_connect(size_t subscriber_threshold = 1) && {
if (pimpl_->unique() && !pimpl_->connected()) {
pimpl_->auto_connect_threshold(subscriber_threshold);
return observable<T>{std::move(pimpl_)};
} else {
auto ptr = make_counted<op::publish<T>>(ctx(), pimpl_);
ptr->auto_connect_threshold(subscriber_threshold);
return observable<T>{ptr};
}
}
/// Returns an @ref observable that automatically connects to this
/// `connectable` when reaching `subscriber_threshold` subscriptions and
/// disconnects automatically after the last subscriber canceled its
/// subscription.
/// @note The threshold only applies to the initial connect, not to any
/// re-connects.
observable<T> ref_count(size_t subscriber_threshold = 1) & {
auto ptr = make_counted<op::publish<T>>(ctx(), pimpl_);
ptr->auto_connect_threshold(subscriber_threshold);
ptr->auto_disconnect(true);
return observable<T>{ptr};
}
/// Similar to the `lvalue` overload, but converts this `connectable` directly
/// if possible, thus saving one hop on the pipeline.
observable<T> ref_count(size_t subscriber_threshold = 1) && {
if (pimpl_->unique() && !pimpl_->connected()) {
pimpl_->auto_connect_threshold(subscriber_threshold);
pimpl_->auto_disconnect(true);
return observable<T>{std::move(pimpl_)};
} else {
auto ptr = make_counted<op::publish<T>>(ctx(), pimpl_);
ptr->auto_connect_threshold(subscriber_threshold);
ptr->auto_disconnect(true);
return observable<T>{ptr};
}
}
/// Connects to the source @ref observable, thus starting to emit items.
disposable connect() {
return pimpl_->connect();
}
/// @copydoc observable::compose
template <class Fn>
auto compose(Fn&& fn) & {
return fn(*this);
}
/// @copydoc observable::compose
template <class Fn>
auto compose(Fn&& fn) && {
return fn(std::move(*this));
}
template <class... Ts>
disposable subscribe(Ts&&... xs) {
return as_observable().subscribe(std::forward<Ts>(xs)...);
}
observable<T> as_observable() const& {
return observable<T>{pimpl_};
}
observable<T> as_observable() && {
return observable<T>{std::move(pimpl_)};
}
const pimpl_type& pimpl() const noexcept {
return pimpl_;
}
bool valid() const noexcept {
return pimpl_ != nullptr;
}
explicit operator bool() const noexcept {
return valid();
}
bool operator!() const noexcept {
return !valid();
}
void swap(connectable& other) {
pimpl_.swap(other.pimpl_);
}
/// @pre `valid()`
coordinator* ctx() const {
return pimpl_->ctx();
}
private:
pimpl_type pimpl_;
};
// -- transformation ----------------------------------------------------------- // -- transformation -----------------------------------------------------------
/// A special type of observer that applies a series of transformation steps to /// A special type of observer that applies a series of transformation steps to
...@@ -160,6 +308,12 @@ public: ...@@ -160,6 +308,12 @@ public:
std::make_tuple(std::move(step)))}; std::make_tuple(std::move(step)))};
} }
/// @copydoc observable::compose
template <class Fn>
auto compose(Fn&& fn) && {
return fn(std::move(*this));
}
auto take(size_t n) && { auto take(size_t n) && {
return std::move(*this).transform(limit_step<output_type>{n}); return std::move(*this).transform(limit_step<output_type>{n});
} }
...@@ -439,6 +593,20 @@ observable<cow_tuple<T, observable<T>>> observable<T>::head_and_tail() { ...@@ -439,6 +593,20 @@ observable<cow_tuple<T, observable<T>>> observable<T>::head_and_tail() {
.as_observable(); .as_observable();
} }
// -- observable::publish ------------------------------------------------------
template <class T>
connectable<T> observable<T>::publish() {
return connectable<T>{make_counted<op::publish<T>>(ctx(), pimpl_)};
}
// -- observable::share --------------------------------------------------------
template <class T>
observable<T> observable<T>::share(size_t subscriber_threshold) {
return publish().ref_count(subscriber_threshold);
}
// -- observable::to_resource -------------------------------------------------- // -- observable::to_resource --------------------------------------------------
/// Reads from an observable buffer and emits the consumed items. /// Reads from an observable buffer and emits the consumed items.
......
...@@ -168,6 +168,24 @@ public: ...@@ -168,6 +168,24 @@ public:
/// 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();
/// Convert this observable into a @ref connectable observable.
connectable<T> publish();
/// Convenience alias for `publish().ref_count(subscriber_threshold)`.
observable<T> share(size_t subscriber_threshold = 1);
/// Transform this `observable` by applying a function object to it.
template <class Fn>
auto compose(Fn&& fn) & {
return fn(*this);
}
/// Fn this `observable` by applying a function object to it.
template <class Fn>
auto compose(Fn&& fn) && {
return fn(std::move(*this));
}
/// Creates an asynchronous resource that makes emitted items available in a /// Creates an asynchronous resource that makes emitted items available in a
/// spsc buffer. /// spsc buffer.
async::consumer_resource<T> to_resource(size_t buffer_size, async::consumer_resource<T> to_resource(size_t buffer_size,
......
...@@ -542,6 +542,22 @@ public: ...@@ -542,6 +542,22 @@ public:
} }
} }
bool idle() const noexcept {
return state == observer_state::idle;
}
bool subscribed() const noexcept {
return state == observer_state::subscribed;
}
bool completed() const noexcept {
return state == observer_state::completed;
}
bool aborted() const noexcept {
return state == observer_state::aborted;
}
std::vector<T> sorted_buf() const { std::vector<T> sorted_buf() const {
auto result = buf; auto result = buf;
std::sort(result.begin(), result.end()); std::sort(result.begin(), result.end());
...@@ -557,7 +573,7 @@ public: ...@@ -557,7 +573,7 @@ public:
error err; error err;
/// Represents the current state of this observer. /// Represents the current state of this observer.
observer_state state; observer_state state = observer_state::idle;
/// Stores all items received via `on_next`. /// Stores all items received via `on_next`.
std::vector<T> buf; std::vector<T> buf;
...@@ -572,64 +588,26 @@ intrusive_ptr<passive_observer<T>> make_passive_observer() { ...@@ -572,64 +588,26 @@ intrusive_ptr<passive_observer<T>> make_passive_observer() {
/// Similar to @ref passive_observer but automatically requests items until /// Similar to @ref passive_observer but automatically requests items until
/// completed. Useful for writing unit tests. /// completed. Useful for writing unit tests.
template <class T> template <class T>
class auto_observer : public observer_impl_base<T> { class auto_observer : public passive_observer<T> {
public: public:
// -- implementation of observer_impl<T> ------------------------------------- // -- implementation of observer_impl<T> -------------------------------------
void on_complete() override {
if (sub) {
sub.dispose();
sub = nullptr;
}
state = observer_state::completed;
}
void on_error(const error& what) override {
if (sub) {
sub.dispose();
sub = nullptr;
}
err = what;
state = observer_state::aborted;
}
void on_subscribe(subscription new_sub) override { void on_subscribe(subscription new_sub) override {
if (state == observer_state::idle) { if (this->state == observer_state::idle) {
CAF_ASSERT(!sub); CAF_ASSERT(!this->sub);
sub = std::move(new_sub); this->sub = std::move(new_sub);
state = observer_state::subscribed; this->state = observer_state::subscribed;
sub.request(64); this->sub.request(64);
} else { } else {
new_sub.dispose(); new_sub.dispose();
} }
} }
void on_next(const T& item) override { void on_next(const T& item) override {
buf.emplace_back(item); this->buf.emplace_back(item);
sub.request(1); if (this->sub)
this->sub.request(1);
} }
// -- convenience functions --------------------------------------------------
std::vector<T> sorted_buf() const {
auto result = buf;
std::sort(result.begin(), result.end());
return result;
}
// -- member variables -------------------------------------------------------
/// The subscription for requesting additional items.
subscription sub;
/// Default-constructed unless on_error was called.
error err;
/// Represents the current state of this observer.
observer_state state;
/// Stores all items received via `on_next`.
std::vector<T> buf;
}; };
/// @relates auto_observer /// @relates auto_observer
......
...@@ -18,9 +18,7 @@ enum class observer_state { ...@@ -18,9 +18,7 @@ enum class observer_state {
/// Indicates that on_complete was called. /// Indicates that on_complete was called.
completed, completed,
/// Indicates that on_error was called. /// Indicates that on_error was called.
aborted, aborted
/// Indicates that dispose was called.
disposed,
}; };
/// Returns whether `x` represents a final state, i.e., `completed`, `aborted` /// Returns whether `x` represents a final state, i.e., `completed`, `aborted`
......
...@@ -9,7 +9,8 @@ ...@@ -9,7 +9,8 @@
namespace caf::flow::op { namespace caf::flow::op {
/// Abstract base type for all flow operators. /// Abstract base type for all flow operators that implement the *observable*
/// concept.
template <class T> template <class T>
class base : public coordinated { class base : public coordinated {
public: public:
......
...@@ -27,17 +27,17 @@ public: ...@@ -27,17 +27,17 @@ public:
// -- implementation of disposable_impl -------------------------------------- // -- implementation of disposable_impl --------------------------------------
void ref_coordinated() const noexcept final { void ref_coordinated() const noexcept override {
this->ref(); this->ref();
} }
void deref_coordinated() const noexcept final { void deref_coordinated() const noexcept override {
this->deref(); this->deref();
} }
// -- implementation of observable_impl<T> ----------------------------------- // -- implementation of observable_impl<T> -----------------------------------
coordinator* ctx() const noexcept final { coordinator* ctx() const noexcept override {
return ctx_; return ctx_;
} }
......
...@@ -4,7 +4,6 @@ ...@@ -4,7 +4,6 @@
#pragma once #pragma once
#include "caf/flow/observable.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp" #include "caf/flow/op/cold.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
......
...@@ -126,13 +126,17 @@ using from_generator_output_t = // ...@@ -126,13 +126,17 @@ using from_generator_output_t = //
detail::type_list<Generator, Steps...> // detail::type_list<Generator, Steps...> //
>::output_type; >::output_type;
/// Converts a `Generator` to an @ref observable.
/// @note Depending on the `Generator`, this operator may turn *cold* if copying
/// the generator results in each copy emitting the exact same sequence of
/// values. However, we should treat it as *hot* by default.
template <class Generator, class... Steps> template <class Generator, class... Steps>
class from_generator class from_generator
: public op::cold<from_generator_output_t<Generator, Steps...>> { : public hot<from_generator_output_t<Generator, Steps...>> {
public: public:
using output_type = from_generator_output_t<Generator, Steps...>; using output_type = from_generator_output_t<Generator, Steps...>;
using super = op::cold<output_type>; using super = hot<output_type>;
from_generator(coordinator* ctx, Generator gen, std::tuple<Steps...> steps) from_generator(coordinator* ctx, Generator gen, std::tuple<Steps...> steps)
: super(ctx), gen_(std::move(gen)), steps_(std::move(steps)) { : super(ctx), gen_(std::move(gen)), steps_(std::move(steps)) {
......
...@@ -27,17 +27,17 @@ public: ...@@ -27,17 +27,17 @@ public:
// -- implementation of disposable_impl -------------------------------------- // -- implementation of disposable_impl --------------------------------------
void ref_coordinated() const noexcept final { void ref_coordinated() const noexcept override {
this->ref(); this->ref();
} }
void deref_coordinated() const noexcept final { void deref_coordinated() const noexcept override {
this->deref(); this->deref();
} }
// -- implementation of observable_impl<T> ----------------------------------- // -- implementation of observable_impl<T> -----------------------------------
coordinator* ctx() const noexcept final { coordinator* ctx() const noexcept override {
return ctx_; return ctx_;
} }
......
...@@ -9,6 +9,7 @@ ...@@ -9,6 +9,7 @@
#include "caf/flow/op/empty.hpp" #include "caf/flow/op/empty.hpp"
#include "caf/flow/op/hot.hpp" #include "caf/flow/op/hot.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include <algorithm> #include <algorithm>
...@@ -19,7 +20,22 @@ namespace caf::flow::op { ...@@ -19,7 +20,22 @@ namespace caf::flow::op {
/// State shared between one multicast operator and one subscribed observer. /// State shared between one multicast operator and one subscribed observer.
template <class T> template <class T>
struct mcast_sub_state { class mcast_sub_state : public detail::plain_ref_counted {
public:
friend void intrusive_ptr_add_ref(const mcast_sub_state* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const mcast_sub_state* ptr) noexcept {
ptr->deref();
}
mcast_sub_state(coordinator* ctx, observer<T> out)
: ctx(ctx), out(std::move(out)) {
// nop
}
coordinator* ctx;
std::deque<T> buf; std::deque<T> buf;
size_t demand = 0; size_t demand = 0;
observer<T> out; observer<T> out;
...@@ -28,6 +44,9 @@ struct mcast_sub_state { ...@@ -28,6 +44,9 @@ struct mcast_sub_state {
bool running = false; bool running = false;
error err; error err;
action when_disposed;
action when_consumed_some;
void push(const T& item) { void push(const T& item) {
if (disposed) { if (disposed) {
// nop // nop
...@@ -47,6 +66,9 @@ struct mcast_sub_state { ...@@ -47,6 +66,9 @@ struct mcast_sub_state {
if (!running && buf.empty()) { if (!running && buf.empty()) {
disposed = true; disposed = true;
out.on_complete(); out.on_complete();
out = nullptr;
when_disposed = nullptr;
when_consumed_some = nullptr;
} }
} }
} }
...@@ -58,6 +80,9 @@ struct mcast_sub_state { ...@@ -58,6 +80,9 @@ struct mcast_sub_state {
if (!running && buf.empty()) { if (!running && buf.empty()) {
disposed = true; disposed = true;
out.on_error(reason); out.on_error(reason);
out = nullptr;
when_disposed = nullptr;
when_consumed_some = nullptr;
} }
} }
} }
...@@ -67,6 +92,13 @@ struct mcast_sub_state { ...@@ -67,6 +92,13 @@ struct mcast_sub_state {
out.on_complete(); out.on_complete();
out = nullptr; out = nullptr;
} }
if (when_disposed) {
ctx->delay(std::move(when_disposed));
}
if (when_consumed_some) {
when_consumed_some.dispose();
when_consumed_some = nullptr;
}
buf.clear(); buf.clear();
demand = 0; demand = 0;
disposed = true; disposed = true;
...@@ -75,24 +107,26 @@ struct mcast_sub_state { ...@@ -75,24 +107,26 @@ struct mcast_sub_state {
void do_run() { void do_run() {
auto guard = detail::make_scope_guard([this] { running = false; }); auto guard = detail::make_scope_guard([this] { running = false; });
if (!disposed) { if (!disposed) {
while (demand > 0 && !buf.empty()) { auto got_some = demand > 0 && !buf.empty();
for (bool run = got_some; run; run = demand > 0 && !buf.empty()) {
out.on_next(buf.front()); out.on_next(buf.front());
buf.pop_front(); buf.pop_front();
--demand; --demand;
} }
if (buf.empty() && closed) { if (buf.empty() && closed) {
disposed = true;
if (err) if (err)
out.on_error(err); out.on_error(err);
else else
out.on_complete(); out.on_complete();
out = nullptr;
do_dispose();
} }
} }
} }
}; };
template <class T> template <class T>
using mcast_sub_state_ptr = std::shared_ptr<mcast_sub_state<T>>; using mcast_sub_state_ptr = intrusive_ptr<mcast_sub_state<T>>;
template <class T> template <class T>
class mcast_sub : public subscription::impl_base { class mcast_sub : public subscription::impl_base {
...@@ -185,7 +219,8 @@ public: ...@@ -185,7 +219,8 @@ public:
auto pred = [](const state_ptr_type& x, const state_ptr_type& y) { auto pred = [](const state_ptr_type& x, const state_ptr_type& y) {
return x->demand < y->demand; return x->demand < y->demand;
}; };
return std::max_element(states_.begin(), states_.end(), pred)->demand; auto& ptr = *std::max_element(states_.begin(), states_.end(), pred);
return ptr->demand;
} }
} }
...@@ -196,7 +231,8 @@ public: ...@@ -196,7 +231,8 @@ public:
auto pred = [](const state_ptr_type& x, const state_ptr_type& y) { auto pred = [](const state_ptr_type& x, const state_ptr_type& y) {
return x->demand < y->demand; return x->demand < y->demand;
}; };
return std::min_element(states_.begin(), states_.end(), pred)->demand; auto& ptr = *std::min_element(states_.begin(), states_.end(), pred);
ptr->demand;
} }
} }
...@@ -207,7 +243,8 @@ public: ...@@ -207,7 +243,8 @@ public:
auto pred = [](const state_ptr_type& x, const state_ptr_type& y) { auto pred = [](const state_ptr_type& x, const state_ptr_type& y) {
return x->buf.size() < y->buf.size(); return x->buf.size() < y->buf.size();
}; };
return std::max_element(states_.begin(), states_.end(), pred)->buf.size(); auto& ptr = *std::max_element(states_.begin(), states_.end(), pred);
return ptr->buf.size();
} }
} }
...@@ -218,7 +255,8 @@ public: ...@@ -218,7 +255,8 @@ public:
auto pred = [](const state_ptr_type& x, const state_ptr_type& y) { auto pred = [](const state_ptr_type& x, const state_ptr_type& y) {
return x->buf.size() < y->buf.size(); return x->buf.size() < y->buf.size();
}; };
return std::min_element(states_.begin(), states_.end(), pred)->buf.size(); auto& ptr = *std::min_element(states_.begin(), states_.end(), pred);
ptr->buf.size();
} }
} }
...@@ -227,9 +265,20 @@ public: ...@@ -227,9 +265,20 @@ public:
return !states_.empty(); return !states_.empty();
} }
/// Queries the current number of subscribed observers.
size_t observer_count() const noexcept {
return states_.size();
}
state_ptr_type add_state(observer_type out) { state_ptr_type add_state(observer_type out) {
auto state = std::make_shared<state_type>(); auto state = make_counted<state_type>(super::ctx_, std::move(out));
state->out = std::move(out); auto mc = strong_this();
state->when_disposed = make_action([mc, state]() mutable { //
mc->do_dispose(state);
});
state->when_consumed_some = make_action([mc, state]() mutable { //
mc->on_consumed_some(*state);
});
states_.push_back(state); states_.push_back(state);
return state; return state;
} }
...@@ -250,7 +299,28 @@ public: ...@@ -250,7 +299,28 @@ public:
protected: protected:
bool closed_ = false; bool closed_ = false;
error err_; error err_;
std::vector<mcast_sub_state_ptr<T>> states_; std::vector<state_ptr_type> states_;
private:
intrusive_ptr<mcast> strong_this() {
return {this};
}
void do_dispose(state_ptr_type& state) {
auto e = states_.end();
if (auto i = std::find(states_.begin(), e, state); i != e) {
states_.erase(i);
on_dispose(*state);
}
}
virtual void on_dispose(state_type&) {
// nop
}
virtual void on_consumed_some(state_type&) {
// nop
}
}; };
} // namespace caf::flow::op } // 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/flow/observer.hpp"
#include "caf/flow/op/mcast.hpp"
#include <limits>
namespace caf::flow::op {
/// Publishes the items from a single operator to multiple subscribers.
template <class T>
class publish : public mcast<T>, public observer_impl<T> {
public:
// -- member types -----------------------------------------------------------
using super = mcast<T>;
using state_type = typename super::state_type;
using src_ptr = intrusive_ptr<base<T>>;
// -- constructors, destructors, and assignment operators --------------------
publish(coordinator* ctx, src_ptr src) : super(ctx), source_(std::move(src)) {
// 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 publish* ptr) noexcept {
ptr->ref_coordinated();
}
friend void intrusive_ptr_release(const publish* ptr) noexcept {
ptr->deref_coordinated();
}
// -- implementation of conn<T> ----------------------------------------------
coordinator* ctx() const noexcept override {
return super::ctx_;
}
disposable subscribe(observer<T> out) override {
auto result = super::subscribe(std::move(out));
if (!connected_ && super::observer_count() == auto_connect_threshold_) {
// Note: reset to 1 since the threshold only applies to the first connect.
auto_connect_threshold_ = 1;
connect();
}
return result;
}
disposable connect() {
CAF_ASSERT(source_);
CAF_ASSERT(!connected_);
connected_ = true;
return source_->subscribe(observer<T>{this});
}
// -- properties -------------------------------------------------------------
void auto_connect_threshold(size_t new_value) {
auto_connect_threshold_ = new_value;
}
void auto_disconnect(bool new_value) {
auto_disconnect_ = new_value;
;
}
bool connected() const noexcept {
return connected_;
}
// -- implementation of observer_impl<T> -------------------------------------
void on_next(const T& item) override {
--in_flight_;
this->push_all(item);
}
void on_complete() override {
this->close();
}
void on_error(const error& what) override {
this->abort(what);
}
void on_subscribe(subscription in) override {
if (!in_) {
in_ = in;
in_flight_ = max_buf_size_;
in_.request(max_buf_size_);
} else {
in.dispose();
}
}
protected:
void try_request_more() {
if (in_ && this->has_observers()) {
if (auto buf_size = this->max_buffered() + in_flight_;
max_buf_size_ > buf_size) {
auto demand = max_buf_size_ - buf_size;
in_flight_ += demand;
in_.request(demand);
}
}
}
private:
void on_dispose(state_type&) override {
try_request_more();
if (auto_disconnect_ && connected_ && super::observer_count() == 0) {
in_.dispose();
in_ = nullptr;
connected_ = false;
}
}
void on_consumed_some(state_type&) override {
try_request_more();
}
size_t in_flight_ = 0;
size_t max_buf_size_ = defaults::flow::buffer_size;
subscription in_;
src_ptr source_;
bool connected_ = false;
size_t auto_connect_threshold_ = std::numeric_limits<size_t>::max();
bool auto_disconnect_ = false;
};
} // 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.
#define CAF_SUITE flow.publish
#include "caf/flow/observer.hpp"
#include "core-test.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
using caf::flow::make_observer;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
template <class... Ts>
std::vector<int> ls(Ts... xs) {
return std::vector<int>{xs...};
}
};
template <class... Ts>
auto subscribe_all(Ts... xs) {
auto snks = std::vector<flow::observer<int>>{flow::observer<int>{xs}...};
return [snks](auto src) {
for (auto snk : snks)
src.subscribe(snk);
return src;
};
}
auto make_hot_generator() {
auto n = std::make_shared<int>(0);
auto fn = [n] {
std::optional<int> result;
if (*n < 10)
result = ++*n;
return result;
};
return std::make_pair(n, fn);
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("publish creates a connectable observable") {
GIVEN("a connectable with a hot generator") {
WHEN("connecting without any subscriber") {
THEN("all items get dropped") {
auto [n, fn] = make_hot_generator();
ctx->make_observable().from_callable(fn).publish().connect();
ctx->run();
CHECK_EQ(*n, 10);
}
}
WHEN("connecting after two observers have subscribed") {
THEN("each observer receives all items from the generator") {
auto [n, fn] = make_hot_generator();
auto snk1 = flow::make_auto_observer<int>();
auto snk2 = flow::make_auto_observer<int>();
ctx->make_observable()
.from_callable(fn)
.publish()
.compose(subscribe_all(snk1, snk2))
.connect();
ctx->run();
CHECK_EQ(*n, 10);
CHECK(snk1->completed());
CHECK_EQ(snk1->buf, ls(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
CHECK(snk2->completed());
CHECK_EQ(snk2->buf, ls(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
}
}
WHEN("adding two observers with auto_connect(2)") {
THEN("each observer receives all items from the generator") {
auto [n, fn] = make_hot_generator();
auto snk1 = flow::make_auto_observer<int>();
auto snk2 = flow::make_auto_observer<int>();
ctx->make_observable()
.from_callable(fn)
.publish()
.auto_connect(2)
.compose(subscribe_all(snk1, snk2));
ctx->run();
CHECK_EQ(*n, 10);
CHECK(snk1->completed());
CHECK_EQ(snk1->buf, ls(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
CHECK(snk2->completed());
CHECK_EQ(snk2->buf, ls(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
}
}
WHEN("adding two observers with share(2)") {
THEN("each observer receives all items from the generator") {
auto [n, fn] = make_hot_generator();
auto snk1 = flow::make_auto_observer<int>();
auto snk2 = flow::make_auto_observer<int>();
ctx->make_observable()
.from_callable(fn) //
.share(2)
.compose(subscribe_all(snk1, snk2));
ctx->run();
CHECK_EQ(*n, 10);
CHECK(snk1->completed());
CHECK_EQ(snk1->buf, ls(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
CHECK(snk2->completed());
CHECK_EQ(snk2->buf, ls(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
}
}
}
}
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