Backpressure: SerializedSubscriber

[benjchristensen]

Can we make SerializedSubscriber correctly be implemented with a bounded buffer and support reactive pull backpressure?

This should also ensure that Observable.serialize() works correctly with backpressure, both bounding its internal queue and chaining through the Producer.request calls.

[akarnokd]

Based on my experience in ReplaySubject, this might require an incompatible change. SerializedObserver needs to know that the wrapped Observer is a Subscriber and thus can participate in the backpressure logic whereas a general Observer cannot. This adds either a instanceof check on every onXYZ method or one needs to use some static factory method (i.e. SerializedObserver.from()) to return a specific subclass.

In addition, the array-based buffering as it is now, can't be kept because the queue might be partially consumed if the requested amount is less than Long.MAX_VALUE. Either the unconsumed elements need to be joined with the potential new queue or one needs to switch to a singly-linked list and do "pointer arithmetic". Single-value delivery latency will suffer either way.

[benjchristensen]

A SerializedObserver can't by definition participate in backpressure since an Observer doesn't. The only time SerializedObserver should ever be used though is by a developer directly. All internal operators should be migrated to using SerializedSubscriber as they should all be using Subscriber.

[akarnokd]

I don't believe we should add backpressure support to SerializedSubscriber: it is a write-through from the perspective of backpressure: if downstream requests 5, SerializedSubscriber will request 5. It is not meant to run with backpressure-ignoring upstream and theonBackpressureXXX are there to handle it after.

I vote on closing this issue.