Commit c6fc49c0 authored by Dominik Charousset's avatar Dominik Charousset

Allow tapping into buffer subscribers

parent fbeacbcf
...@@ -166,7 +166,9 @@ public: ...@@ -166,7 +166,9 @@ public:
} }
/// Creates a new observer that pushes all observed items to the resource. /// Creates a new observer that pushes all observed items to the resource.
disposable subscribe(async::producer_resource<T> resource); disposable subscribe(async::producer_resource<T> resource,
std::shared_ptr<size_t> total_requested = nullptr,
std::shared_ptr<size_t> total_pushed = nullptr);
/// Returns a transformation that applies a step function to each input. /// Returns a transformation that applies a step function to each input.
template <class Step> template <class Step>
...@@ -439,8 +441,11 @@ public: ...@@ -439,8 +441,11 @@ public:
return lift().subscribe(std::move(what)); return lift().subscribe(std::move(what));
} }
disposable subscribe(async::producer_resource<T> resource) && { disposable subscribe(async::producer_resource<T> resource,
return lift().subscribe(std::move(resource)); std::shared_ptr<size_t> total_requested = nullptr,
std::shared_ptr<size_t> total_pushed = nullptr) && {
return lift().subscribe(std::move(resource), std::move(total_requested),
std::move(total_pushed));
} }
async::consumer_resource<T> to_resource() && { async::consumer_resource<T> to_resource() && {
...@@ -2128,12 +2133,16 @@ observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size, ...@@ -2128,12 +2133,16 @@ observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size,
// -- observable::subscribe ---------------------------------------------------- // -- observable::subscribe ----------------------------------------------------
template <class T> template <class T>
disposable observable<T>::subscribe(async::producer_resource<T> resource) { disposable observable<T>::subscribe(async::producer_resource<T> resource,
std::shared_ptr<size_t> total_requested,
std::shared_ptr<size_t> total_pushed) {
using buffer_type = typename async::consumer_resource<T>::buffer_type; using buffer_type = typename async::consumer_resource<T>::buffer_type;
using adapter_type = buffer_writer_impl<buffer_type>; using adapter_type = buffer_writer_impl<buffer_type>;
if (auto buf = resource.try_open()) { if (auto buf = resource.try_open()) {
CAF_LOG_DEBUG("subscribe producer resource to flow"); CAF_LOG_DEBUG("subscribe producer resource to flow");
auto adapter = make_counted<adapter_type>(pimpl_->ctx(), buf); auto adapter = make_counted<adapter_type>(pimpl_->ctx(), buf,
std::move(total_requested),
std::move(total_pushed));
buf->set_producer(adapter); buf->set_producer(adapter);
auto obs = adapter->as_observer(); auto obs = adapter->as_observer();
pimpl_->ctx()->watch(obs.as_disposable()); pimpl_->ctx()->watch(obs.as_disposable());
......
...@@ -352,8 +352,13 @@ public: ...@@ -352,8 +352,13 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
buffer_writer_impl(coordinator* ctx, buffer_ptr buf) buffer_writer_impl(coordinator* ctx, buffer_ptr buf,
: ctx_(ctx), buf_(std::move(buf)) { std::shared_ptr<size_t> total_requested,
std::shared_ptr<size_t> total_pushed)
: ctx_(ctx),
buf_(std::move(buf)),
total_requested_(std::move(total_requested)),
total_pushed_(std::move(total_pushed)) {
CAF_ASSERT(ctx_ != nullptr); CAF_ASSERT(ctx_ != nullptr);
CAF_ASSERT(buf_ != nullptr); CAF_ASSERT(buf_ != nullptr);
} }
...@@ -393,9 +398,12 @@ public: ...@@ -393,9 +398,12 @@ public:
void on_next(span<const value_type> items) override { void on_next(span<const value_type> items) override {
CAF_LOG_TRACE(CAF_ARG(items)); CAF_LOG_TRACE(CAF_ARG(items));
if (buf_) if (buf_) {
if (total_pushed_)
*total_pushed_ += items.size();
buf_->push(items); buf_->push(items);
} }
}
void on_complete() override { void on_complete() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -459,9 +467,12 @@ public: ...@@ -459,9 +467,12 @@ public:
private: private:
void on_demand(size_t n) { void on_demand(size_t n) {
CAF_LOG_TRACE(CAF_ARG(n)); CAF_LOG_TRACE(CAF_ARG(n));
if (sub_) if (sub_) {
if (total_requested_)
*total_requested_ += n;
sub_.request(n); sub_.request(n);
} }
}
void on_cancel() { void on_cancel() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -479,6 +490,8 @@ private: ...@@ -479,6 +490,8 @@ private:
coordinator_ptr ctx_; coordinator_ptr ctx_;
buffer_ptr buf_; buffer_ptr buf_;
subscription sub_; subscription sub_;
std::shared_ptr<size_t> total_requested_;
std::shared_ptr<size_t> total_pushed_;
}; };
// -- utility observer --------------------------------------------------------- // -- utility observer ---------------------------------------------------------
......
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