Commit 36a7a16f authored by Dominik Charousset's avatar Dominik Charousset

Fix implementation on observe_on

parent 155f7bda
...@@ -1677,14 +1677,11 @@ observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size, ...@@ -1677,14 +1677,11 @@ observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size,
size_t min_request_size) { size_t min_request_size) {
using buffer_type = async::spsc_buffer<T>; using buffer_type = async::spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size); auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf); subscribe(async::producer_resource<T>{buf});
auto down = make_counted<observable_buffer_impl<buffer_type>>(other, buf); auto adapter = make_counted<observable_buffer_impl<buffer_type>>(other, buf);
buf->set_producer(up); buf->set_consumer(adapter);
buf->set_consumer(down); other->watch(adapter->as_disposable());
subscribe(up->as_observer()); return observable<T>{std::move(adapter)};
auto hdl = observable<T>{std::move(down)};
pimpl_->ctx()->watch(hdl.as_disposable());
return hdl;
} }
// -- observable::subscribe ---------------------------------------------------- // -- observable::subscribe ----------------------------------------------------
......
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