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

Clean up the observable interface

The `observable` interface should only have one member functions:
`subscribe`. Hence, we move the `on_request` and `on_cancel` member
functions to the new `subscription::listener` class. Also adds a new
default implementation type for subscriptions that no longer requires a
template.
parent d9842dba
......@@ -130,4 +130,7 @@ private:
intrusive_ptr<impl> pimpl_;
};
/// @relates disposable
using disposable_impl = disposable::impl;
} // namespace caf
......@@ -39,7 +39,7 @@ public:
using output_type = T;
/// Internal interface of an `observable`.
class impl : public disposable::impl {
class impl : public disposable_impl {
public:
// -- member types ---------------------------------------------------------
......@@ -49,29 +49,23 @@ public:
virtual coordinator* ctx() const noexcept = 0;
// -- conversions ----------------------------------------------------------
observable as_observable() noexcept;
// -- flow processing ------------------------------------------------------
/// Subscribes a new observer.
virtual disposable subscribe(observer<T> what) = 0;
virtual void on_request(observer_impl<T>* sink, size_t n) = 0;
virtual void on_cancel(observer_impl<T>* sink) = 0;
observable as_observable() noexcept;
/// Creates a subscription for `sink` and returns a @ref disposable to
/// cancel the observer.
disposable do_subscribe(observer_impl<T>* sink);
/// @copydoc do_subscribe
disposable do_subscribe(observer<T>& sink) {
return do_subscribe(sink.ptr());
}
// -- convenience functions ------------------------------------------------
/// 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);
disposable reject_subscription(observer_impl<T>* sink, sec code) {
sink->on_error(make_error(code));
return disposable{};
}
/// @copydoc reject_subscription
disposable reject_subscription(observer<T>& sink, sec code) {
......@@ -79,58 +73,6 @@ public:
}
};
class sub_impl final : public ref_counted, public subscription::impl {
public:
using src_type = typename observable<T>::impl;
using snk_type = typename observer<T>::impl;
CAF_INTRUSIVE_PTR_FRIENDS(sub_impl)
sub_impl(coordinator* ctx, src_type* src, snk_type* snk)
: ctx_(ctx), src_(src), snk_(snk) {
// nop
}
bool disposed() const noexcept override {
return src_ == nullptr;
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
void request(size_t n) override {
if (src_)
ctx()->delay_fn([src = src_, snk = snk_, n] { //
src->on_request(snk.get(), n);
});
}
void cancel() override {
if (src_) {
ctx()->delay_fn([src = src_, snk = snk_] { //
src->on_cancel(snk.get());
});
src_.reset();
snk_.reset();
}
}
auto* ctx() const noexcept {
return ctx_;
}
private:
coordinator* ctx_;
intrusive_ptr<src_type> src_;
intrusive_ptr<snk_type> snk_;
};
explicit observable(intrusive_ptr<impl> pimpl) noexcept
: pimpl_(std::move(pimpl)) {
// nop
......@@ -326,28 +268,15 @@ observable<T> observable<T>::impl::as_observable() noexcept {
return observable<T>{intrusive_ptr{this}};
}
template <class T>
disposable observable<T>::impl::do_subscribe(observer_impl<T>* sink) {
sink->on_subscribe(subscription{make_counted<sub_impl>(ctx(), this, sink)});
// Note: we do NOT return the subscription here because this object is private
// to the observer. Outside code must call dispose() on the observer.
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>
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> {
class observable_impl_base : public ref_counted,
public observable_impl<T>,
public subscription::listener {
public:
// -- member types -----------------------------------------------------------
......@@ -361,7 +290,13 @@ public:
// nop
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
disposable as_disposable() noexcept {
// We need to implement this only for disambiguation since `as_disposable`
// exists in multiple base classes.
return subscription::listener::as_disposable();
}
void ref_disposable() const noexcept final {
this->ref();
......@@ -371,12 +306,29 @@ public:
this->deref();
}
// -- implementation of observable_impl<T> ---------------------------------
// -- implementation of observable_impl<T> -----------------------------------
coordinator* ctx() const noexcept final {
return ctx_;
}
// -- convenience functions --------------------------------------------------
/// Creates a subscription for `sink` and returns a @ref disposable to cancel
/// the observer.
disposable do_subscribe(observer_impl<T>* sink) {
sink->on_subscribe(subscription::default_impl::make(ctx_, this, sink));
// Note: we do NOT return the subscription here because this object is
// private to the observer. Outside code cancels flows by calling
// dispose on the observer.
return sink->as_disposable();
}
/// @copydoc do_subscribe
disposable do_subscribe(observer<T>& sink) {
return do_subscribe(sink.ptr());
}
protected:
// -- member variables -------------------------------------------------------
......@@ -538,18 +490,81 @@ private:
intrusive_ptr<impl> pimpl_;
};
/// @relates processor
template <class In, class Out = In>
using processor_impl = typename processor<In, Out>::impl;
// -- representing an error as an observable -----------------------------------
template <class T>
class observable_error_impl : public observable_impl_base<T> {
/// Default base type for processor implementation types.
/// @relates processor
template <class In, class Out>
class processor_impl_base : public ref_counted,
public processor_impl<In, Out>,
public subscription::listener {
public:
// -- member types -----------------------------------------------------------
using super = observable_impl_base<T>;
using super = processor_impl<In, Out>;
using input_type = In;
using output_type = Out;
// -- constructors, destructors, and assignment operators --------------------
explicit processor_impl_base(coordinator* ctx) : ctx_(ctx) {
// nop
}
// -- implementation of disposable_impl --------------------------------------
disposable as_disposable() noexcept {
// We need to implement this only for disambiguation since `as_disposable`
// exists in multiple base classes.
return subscription::listener::as_disposable();
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
// -- implementation of processor_impl<T> -----------------------------------
coordinator* ctx() const noexcept final {
return ctx_;
}
// -- convenience functions --------------------------------------------------
/// Creates a subscription for `sink` and returns a @ref disposable to cancel
/// the observer.
disposable do_subscribe(observer_impl<Out>* sink) {
sink->on_subscribe(subscription::default_impl::make(ctx_, this, sink));
// Note: we do NOT return the subscription here because this object is
// private to the observer. Outside code cancels flows by calling
// dispose on the observer.
return disposable{intrusive_ptr<disposable_impl>{sink}};
}
/// @copydoc do_subscribe
disposable do_subscribe(observer<Out>& sink) {
return do_subscribe(sink.ptr());
}
protected:
// -- member variables -------------------------------------------------------
coordinator* ctx_;
};
// -- representing an error as an observable -----------------------------------
template <class T>
class observable_error_impl : public ref_counted, public observable_impl<T> {
public:
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(observable_error_impl)
......@@ -557,11 +572,19 @@ public:
// -- constructors, destructors, and assignment operators --------------------
observable_error_impl(coordinator* ctx, error what)
: super(ctx), what_(std::move(what)) {
: ctx_(ctx), what_(std::move(what)) {
// nop
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
void dispose() override {
// nop
......@@ -573,12 +596,8 @@ public:
// -- implementation of observable<T>::impl ----------------------------------
void on_request(observer_impl<T>*, size_t) override {
CAF_RAISE_ERROR("observable_error_impl::on_request called");
}
void on_cancel(observer_impl<T>*) override {
CAF_RAISE_ERROR("observable_error_impl::on_cancel called");
coordinator* ctx() const noexcept final {
return ctx_;
}
disposable subscribe(observer<T> what) override {
......@@ -596,8 +615,7 @@ private:
/// Broadcasts its input to all observers without modifying it.
template <class Step, class... Steps>
class broadcaster_impl
: public ref_counted,
public processor_impl<typename Step::input_type,
: public processor_impl_base<typename Step::input_type,
steps_output_type_t<Step, Steps...>> {
public:
// -- member types -----------------------------------------------------------
......@@ -606,6 +624,8 @@ public:
using output_type = steps_output_type_t<Step, Steps...>;
using super = processor_impl_base<input_type, output_type>;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(broadcaster_impl)
......@@ -614,11 +634,11 @@ public:
template <class... Ts>
explicit broadcaster_impl(coordinator* ctx, Ts&&... step_args)
: ctx_(ctx), steps_(std::forward<Ts>(step_args)...) {
: super(ctx), steps_(std::forward<Ts>(step_args)...) {
// nop
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void dispose() override {
CAF_LOG_TRACE("");
......@@ -629,14 +649,6 @@ public:
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 {
......@@ -682,26 +694,24 @@ public:
}
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
disposable subscribe(observer<output_type> sink) override {
return term_.add(this, sink);
}
// -- implementation of subscription::listener -------------------------------
void on_request(observer_impl<output_type>* sink, size_t n) override {
void on_request(disposable_impl* sink, size_t n) override {
CAF_LOG_TRACE(CAF_ARG(n));
term_.on_request(sub_, sink, n);
}
void on_cancel(observer_impl<output_type>* sink) override {
void on_cancel(disposable_impl* sink) override {
CAF_LOG_TRACE("");
term_.on_cancel(sub_, sink);
}
// -- implementation of observable<T>::impl ----------------------------------
disposable subscribe(observer<output_type> sink) override {
return term_.add(this, sink);
}
// -- properties -------------------------------------------------------------
size_t buffered() const noexcept {
......@@ -717,9 +727,6 @@ public:
}
protected:
/// Points to the coordinator that runs this observable.
coordinator* ctx_;
/// Allows us to request more items.
subscription sub_;
......@@ -957,7 +964,7 @@ public:
// nop
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void dispose() override {
for (auto& kvp : inputs_) {
......@@ -980,14 +987,14 @@ public:
// -- implementation of observable<T>::impl ----------------------------------
void on_request(observer_impl<T>* sink, size_t demand) override {
void on_request(disposable_impl* sink, size_t demand) override {
if (auto n = term_.on_request(sink, demand); n > 0) {
pull(n);
term_.push();
}
}
void on_cancel(observer_impl<T>* sink) override {
void on_cancel(disposable_impl* sink) override {
if (auto n = term_.on_cancel(sink); n > 0) {
pull(n);
term_.push();
......@@ -1338,7 +1345,7 @@ public:
// nop
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void dispose() override {
if (sub_)
......@@ -1351,9 +1358,9 @@ public:
return term_.finalized();
}
// -- implementation of observable<T>::impl ----------------------------------
// -- implementation of subscription::listener -------------------------------
void on_request(observer_impl<T>* sink, size_t demand) override {
void on_request(disposable_impl* sink, size_t demand) override {
if (auto n = term_.on_request(sink, demand); n > 0) {
in_flight_ += n;
if (sub_)
......@@ -1361,7 +1368,7 @@ public:
}
}
void on_cancel(observer_impl<T>* sink) override {
void on_cancel(disposable_impl* sink) override {
if (auto n = term_.on_cancel(sink); n > 0) {
in_flight_ += n;
if (sub_)
......@@ -1369,6 +1376,8 @@ public:
}
}
// -- implementation of observable<T>::impl ----------------------------------
disposable subscribe(observer<T> sink) override {
// On the first subscribe, we subscribe to our inputs unless the user did
// not add any inputs before that. In that case, we close immediately except
......@@ -1602,6 +1611,7 @@ auto observable<T>::concat_map(F f) {
template <class T>
class prefix_and_tail_observable_impl final
: public ref_counted,
public subscription::listener,
public observable_impl<T>, // For the forwarding to the 'tail'.
public processor_impl<T, cow_tuple<std::vector<T>, observable<T>>> {
public:
......@@ -1626,7 +1636,7 @@ public:
CAF_INTRUSIVE_PTR_FRIENDS(prefix_and_tail_observable_impl)
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void dispose() override {
if (sub_) {
......@@ -1647,56 +1657,58 @@ public:
return !sub_ && !obs_ && !tail_;
}
void ref_disposable() const noexcept override {
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept override {
void deref_disposable() const noexcept final {
this->deref();
}
// -- implementation of observable<in_t>::impl -------------------------------
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
coordinator* ctx() const noexcept final {
return ctx_;
}
// -- implementation of observable<in_t>::impl -------------------------------
disposable subscribe(observer<in_t> sink) override {
if (sink.ptr() == tail_.ptr()) {
return in_obs_t::do_subscribe(sink.ptr());
using sub_t = subscription::default_impl;
sink.on_subscribe(sub_t::make_unsafe(ctx_, this, sink.ptr()));
return sink.as_disposable();
} else {
sink.on_error(make_error(sec::invalid_observable));
return disposable{};
}
}
void on_request(observer_impl<in_t>*, size_t n) override {
if (sub_)
sub_.request(n);
}
void on_cancel(observer_impl<in_t>*) override {
if (sub_) {
sub_.cancel();
sub_ = nullptr;
}
}
// -- implementation of observable<out_t>::impl ------------------------------
disposable subscribe(observer<out_t> sink) override {
obs_ = sink;
return out_obs_t::do_subscribe(sink.ptr());
using sub_t = subscription::default_impl;
sink.on_subscribe(sub_t::make_unsafe(ctx_, this, sink.ptr()));
return sink.as_disposable();
}
void on_request(observer_impl<out_t>*, size_t) override {
// -- implementation of subscription::listener -------------------------------
void on_request(disposable_impl* sink, size_t n) override {
if (sink == obs_.ptr()) {
if (sub_ && !requested_prefix_) {
requested_prefix_ = true;
sub_.request(prefix_size_);
}
} else if (sink == tail_.ptr()) {
if (sub_)
sub_.request(n);
}
}
void on_cancel(observer_impl<out_t>*) override {
void on_cancel(disposable_impl* sink) override {
if (sink == obs_.ptr()) {
// Only has an effect when canceling immediately. Otherwise, we forward to
// tail_ and the original observer no longer is of any interest since it
// receives at most one item anyways.
......@@ -1704,6 +1716,12 @@ public:
sub_.cancel();
sub_ = nullptr;
}
} else if (sink == tail_.ptr()) {
if (sub_) {
sub_.cancel();
sub_ = nullptr;
}
}
}
// -- implementation of observer<in_t>::impl ---------------------------------
......@@ -1845,7 +1863,7 @@ public:
this->ctx()->deref_coordinator();
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void dispose() override {
CAF_LOG_TRACE("");
......@@ -1865,14 +1883,14 @@ public:
// -- implementation of observable<T>::impl ----------------------------------
void on_request(observer_impl<value_type>*, size_t n) override {
void on_request(disposable_impl*, size_t n) override {
CAF_LOG_TRACE(CAF_ARG(n));
demand_ += n;
if (demand_ == n)
pull();
}
void on_cancel(observer_impl<value_type>*) override {
void on_cancel(disposable_impl*) override {
CAF_LOG_TRACE("");
dst_ = nullptr;
dispose();
......@@ -2033,7 +2051,7 @@ public:
// nop
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void dispose() override {
// nop
......@@ -2045,11 +2063,11 @@ public:
// -- implementation of observable<T>::impl ----------------------------------
void on_request(observer_impl<output_type>* snk, size_t) override {
snk->on_complete();
void on_request(disposable_impl* snk, size_t) override {
static_cast<observer_impl<output_type>*>(snk)->on_complete();
}
void on_cancel(observer_impl<output_type>*) override {
void on_cancel(disposable_impl*) override {
// nop
}
......@@ -2081,7 +2099,7 @@ public:
// nop
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void dispose() override {
if (!disposed_) {
......@@ -2097,11 +2115,11 @@ public:
// -- implementation of observable<T>::impl ----------------------------------
void on_request(observer_impl<output_type>*, size_t) override {
void on_request(disposable_impl*, size_t) override {
// nop
}
void on_cancel(observer_impl<output_type>*) override {
void on_cancel(disposable_impl*) override {
// nop
}
......@@ -2141,7 +2159,7 @@ public:
// nop
}
// -- implementation of disposable::impl -------------------------------------
// -- implementation of disposable_impl --------------------------------------
void dispose() override {
if (is_active(state)) {
......@@ -2160,13 +2178,13 @@ public:
// -- implementation of observable_impl<T> -----------------------------------
void on_request(observer_impl<output_type>* sink, size_t n) override {
void on_request(disposable_impl* sink, size_t n) override {
if (out.ptr() == sink) {
demand += n;
}
}
void on_cancel(observer_impl<output_type>* sink) override {
void on_cancel(disposable_impl* sink) override {
if (out.ptr() == sink) {
demand = 0;
out = nullptr;
......
......@@ -32,14 +32,13 @@ class callable_source;
class interval_action;
class CAF_CORE_EXPORT interval_impl : public ref_counted,
public observable_impl<int64_t> {
class CAF_CORE_EXPORT interval_impl : public observable_impl_base<int64_t> {
public:
// -- member types -----------------------------------------------------------
using output_type = int64_t;
using super = observable_impl<int64_t>;
using super = observable_impl_base<int64_t>;
// -- friends ----------------------------------------------------------------
......@@ -54,32 +53,23 @@ public:
interval_impl(coordinator* ctx, timespan initial_delay, timespan period,
int64_t max_val);
~interval_impl() override;
// -- implementation of disposable::impl -------------------------------------
void dispose() override;
bool disposed() const noexcept override;
void ref_disposable() const noexcept override;
void deref_disposable() const noexcept override;
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override;
void on_request(disposable_impl*, size_t) override;
void on_request(observer_impl<int64_t>*, size_t) override;
void on_cancel(observer_impl<int64_t>*) override;
void on_cancel(disposable_impl*) override;
disposable subscribe(observer<int64_t> what) override;
private:
void fire(interval_action*);
coordinator* ctx_;
observer<int64_t> obs_;
disposable pending_;
timespan initial_delay_;
......@@ -364,7 +354,7 @@ public:
return sub;
}
void on_request(observer_impl<output_type>* sink, size_t demand) override {
void on_request(disposable_impl* sink, size_t demand) override {
CAF_LOG_TRACE(CAF_ARG(sink) << CAF_ARG(demand));
if (auto n = term_.on_request(sink, demand); n > 0) {
auto fn = [this, n](auto&... steps) { gen_.pull(n, steps..., term_); };
......@@ -373,7 +363,7 @@ public:
}
}
void on_cancel(observer_impl<output_type>* sink) override {
void on_cancel(disposable_impl* sink) override {
CAF_LOG_TRACE(CAF_ARG(sink));
if (auto n = term_.on_cancel(sink); n > 0) {
auto fn = [this, n](auto&... steps) { gen_.pull(n, steps..., term_); };
......
......@@ -25,20 +25,16 @@ public:
using output_type = T;
/// Internal interface of a `single`.
class impl : public ref_counted, public observable_impl<T> {
class impl : public observable_impl_base<T> {
public:
using super = observable_impl<T>;
using super = observable_impl_base<T>;
CAF_INTRUSIVE_PTR_FRIENDS(impl)
explicit impl(coordinator* ctx) : ctx_(ctx) {
explicit impl(coordinator* ctx) : super(ctx) {
// nop
}
coordinator* ctx() const noexcept override {
return ctx_;
}
disposable subscribe(observer<T> what) override {
if (!std::holds_alternative<error>(value_)) {
auto res = super::do_subscribe(what.ptr());
......@@ -50,7 +46,7 @@ public:
}
}
void on_request(observer_impl<T>* sink, size_t n) override {
void on_request(disposable_impl* sink, size_t n) override {
auto pred = [sink](auto& entry) { return entry.first.ptr() == sink; };
if (auto i = std::find_if(observers_.begin(), observers_.end(), pred);
i != observers_.end()) {
......@@ -69,7 +65,7 @@ public:
}
}
void on_cancel(observer_impl<T>* sink) override {
void on_cancel(disposable_impl* sink) override {
auto pred = [sink](auto& entry) { return entry.first.ptr() == sink; };
if (auto i = std::find_if(observers_.begin(), observers_.end(), pred);
i != observers_.end())
......@@ -85,14 +81,6 @@ public:
return observers_.empty() && !std::holds_alternative<none_t>(value_);
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
void set_value(T val) {
if (std::holds_alternative<none_t>(value_)) {
value_ = std::move(val);
......@@ -119,7 +107,6 @@ public:
}
private:
coordinator* ctx_;
std::variant<none_t, T, error> value_;
std::vector<std::pair<observer<T>, size_t>> observers_;
};
......
......@@ -410,7 +410,7 @@ public:
/// 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) {
size_t on_request(disposable_impl* sink, size_t n) {
if (auto i = find(sink); i != outputs_.end()) {
i->demand += n;
push();
......@@ -422,7 +422,7 @@ public:
/// 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) {
void on_request(subscription& sub, disposable_impl* sink, size_t n) {
if (auto new_demand = on_request(sink, n); new_demand > 0 && sub)
sub.request(new_demand);
}
......@@ -430,7 +430,7 @@ public:
/// 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) {
size_t on_cancel(disposable_impl* sink) {
if (auto i = find(sink); i != outputs_.end()) {
outputs_.erase(i);
// TODO: shut down on last cancel?
......@@ -443,7 +443,7 @@ public:
/// 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) {
void on_cancel(subscription& sub, disposable_impl* sink) {
if (auto new_demand = on_cancel(sink); new_demand > 0 && sub)
sub.request(new_demand);
}
......@@ -574,8 +574,7 @@ public:
}
private:
typename std::vector<output_t>::iterator
find(observer_impl<output_type>* sink) {
typename std::vector<output_t>::iterator find(disposable_impl* 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);
......
......@@ -4,13 +4,15 @@
#pragma once
#include <cstddef>
#include "caf/detail/core_export.hpp"
#include "caf/disposable.hpp"
#include "caf/flow/fwd.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
#include <cstddef>
#include <type_traits>
namespace caf::flow {
/// Controls the flow of items from publishers to subscribers.
......@@ -18,7 +20,7 @@ class CAF_CORE_EXPORT subscription {
public:
// -- nested types -----------------------------------------------------------
/// Internal impl of a `disposable`.
/// Internal interface of a `subscription`.
class CAF_CORE_EXPORT impl : public disposable::impl {
public:
~impl() override;
......@@ -34,8 +36,7 @@ public:
};
/// A trivial subscription type that drops all member function calls.
class CAF_CORE_EXPORT nop_impl final : public ref_counted,
public subscription::impl {
class CAF_CORE_EXPORT nop_impl final : public ref_counted, public impl {
public:
// -- friends --------------------------------------------------------------
......@@ -55,6 +56,70 @@ public:
bool disposed_ = false;
};
/// Describes a listener to the subscription that will receive an event
/// whenever the observer calls `request` or `cancel`.
class CAF_CORE_EXPORT listener : public disposable::impl {
public:
virtual ~listener();
virtual void on_request(disposable::impl* sink, size_t n) = 0;
virtual void on_cancel(disposable::impl* sink) = 0;
};
/// Default implementation for subscriptions that forward `request` and
/// `cancel` to a @ref listener.
class default_impl final : public ref_counted, public impl {
public:
CAF_INTRUSIVE_PTR_FRIENDS(default_impl)
default_impl(coordinator* ctx, listener* src, disposable::impl* snk)
: ctx_(ctx), src_(src), snk_(snk) {
// nop
}
bool disposed() const noexcept override;
void ref_disposable() const noexcept override;
void deref_disposable() const noexcept override;
void request(size_t n) override;
void cancel() override;
auto* ctx() const noexcept {
return ctx_;
}
/// Creates a new subscription object.
/// @param ctx The owner of @p src and @p snk.
/// @param src The @ref observable that emits items.
/// @param snk the @ref observer that consumes items.
/// @returns an instance of @ref default_impl in a @ref subscription handle.
template <class Observable, class Observer>
static subscription make(coordinator* ctx, Observable* src, Observer* snk) {
static_assert(std::is_base_of_v<listener, Observable>);
static_assert(std::is_base_of_v<disposable_impl, Observer>);
static_assert(std::is_same_v<typename Observable::output_type,
typename Observer::input_type>);
intrusive_ptr<impl> ptr{new default_impl(ctx, src, snk), false};
return subscription{std::move(ptr)};
}
/// Like @ref make but without any type checking.
static subscription make_unsafe(coordinator* ctx, listener* src,
disposable_impl* snk) {
intrusive_ptr<impl> ptr{new default_impl(ctx, src, snk), false};
return subscription{std::move(ptr)};
}
private:
coordinator* ctx_;
intrusive_ptr<listener> src_;
intrusive_ptr<disposable_impl> snk_;
};
// -- constructors, destructors, and assignment operators --------------------
explicit subscription(intrusive_ptr<impl> pimpl) noexcept
......
......@@ -41,14 +41,13 @@ struct zipper_input {
/// 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...>> {
class zipper_impl : public observable_impl_base<zipper_output_t<F, Ts...>> {
public:
// -- member types -----------------------------------------------------------
using output_type = zipper_output_t<F, Ts...>;
using super = observable_impl<output_type>;
using super = observable_impl_base<output_type>;
// -- friends ----------------------------------------------------------------
......@@ -60,7 +59,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
zipper_impl(coordinator* ctx, F fn, observable<Ts>... inputs)
: ctx_(ctx), fn_(std::move(fn)), inputs_(inputs...) {
: super(ctx), fn_(std::move(fn)), inputs_(inputs...) {
// nop
}
......@@ -86,21 +85,9 @@ public:
return term_.finalized();
}
void ref_disposable() const noexcept override {
this->ref();
}
void deref_disposable() const noexcept override {
this->deref();
}
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override {
return ctx_;
}
void on_request(observer_impl<output_type>* sink, size_t demand) override {
void on_request(disposable_impl* sink, size_t demand) override {
if (auto n = term_.on_request(sink, demand); n > 0) {
demand_ += n;
for_each_input([n](auto, auto& input) {
......@@ -110,7 +97,7 @@ public:
}
}
void on_cancel(observer_impl<output_type>* sink) override {
void on_cancel(disposable_impl* sink) override {
if (auto n = term_.on_cancel(sink); n > 0) {
demand_ += n;
for_each_input([n](auto, auto& input) {
......@@ -241,7 +228,6 @@ private:
term_.fin();
}
coordinator* ctx_;
size_t demand_ = 0;
F fn_;
std::tuple<zipper_input<Ts>...> inputs_;
......
......@@ -62,14 +62,10 @@ interval_impl::interval_impl(coordinator* ctx, timespan initial_delay,
interval_impl::interval_impl(coordinator* ctx, timespan initial_delay,
timespan period, int64_t max_val)
: ctx_(ctx), initial_delay_(initial_delay), period_(period), max_(max_val) {
: super(ctx), initial_delay_(initial_delay), period_(period), max_(max_val) {
CAF_ASSERT(max_val > 0);
}
interval_impl::~interval_impl() {
// nop
}
void interval_impl::dispose() {
if (obs_) {
obs_.on_complete();
......@@ -86,19 +82,7 @@ bool interval_impl::disposed() const noexcept {
return val_ == max_;
}
void interval_impl::ref_disposable() const noexcept {
this->ref();
}
void interval_impl::deref_disposable() const noexcept {
this->deref();
}
coordinator* interval_impl::ctx() const noexcept {
return ctx_;
}
void interval_impl::on_request(observer_impl<int64_t>* ptr, size_t n) {
void interval_impl::on_request(disposable_impl* ptr, size_t n) {
if (obs_.ptr() == ptr) {
if (demand_ == 0 && !pending_) {
if (val_ == 0)
......@@ -112,7 +96,7 @@ void interval_impl::on_request(observer_impl<int64_t>* ptr, size_t n) {
}
}
void interval_impl::on_cancel(observer_impl<int64_t>* ptr) {
void interval_impl::on_cancel(disposable_impl* ptr) {
if (obs_.ptr() == ptr) {
obs_ = nullptr;
pending_.dispose();
......
......@@ -4,6 +4,8 @@
#include "caf/flow/subscription.hpp"
#include "caf/flow/coordinator.hpp"
namespace caf::flow {
subscription::impl::~impl() {
......@@ -34,4 +36,37 @@ void subscription::nop_impl::request(size_t) {
// nop
}
subscription::listener::~listener() {
// nop
}
bool subscription::default_impl::disposed() const noexcept {
return src_ == nullptr;
}
void subscription::default_impl::ref_disposable() const noexcept {
this->ref();
}
void subscription::default_impl::deref_disposable() const noexcept {
this->deref();
}
void subscription::default_impl::request(size_t n) {
if (src_)
ctx()->delay_fn([src = src_, snk = snk_, n] { //
src->on_request(snk.get(), n);
});
}
void subscription::default_impl::cancel() {
if (src_) {
ctx()->delay_fn([src = src_, snk = snk_] { //
src->on_cancel(snk.get());
});
src_.reset();
snk_.reset();
}
}
} // namespace caf::flow
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