rxJava buffer() with time that honours backpressure

Question

The versions of buffer operator that don't operate on time honour backpressure as per JavaDoc:

http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#buffer-int-

However, any version of buffer that involves time based buffers doesn't support backpressure, like this one

http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#buffer-long-java.util.concurrent.TimeUnit-int-

I understand this comes from the fact that once the time is ticking, you can't stop it similarly to, for example interval operator, that doesn't support backpressure either for the same reason.

What I want is a buffering operator that is both size and time based and fully supports backpressure by propagating the backpressure signals to BOTH the upstream AND the time ticking producer, something like this:

someFlowable()
.buffer(
     Flowable.interval(1, SECONDS).onBackpressureDrop(),
     10
);

So now I could drop the tick on backpressure signals.

Is this something currently achievable in rxJava2? How about Project-Reactor?

Answer 1

I've encountered the problem recently and here is my implementation. It can be used like this:

    Flowable<List<T>> bufferedFlow = (some flowable of T)
                              .compose(new BufferTransformer(1, TimeUnit.MILLISECONDS, 8))

It supports backpressure by the count you've specified.

Here is the implementation: https://gist.github.com/driventokill/c49f86fb0cc182994ef423a70e793a2d

Answer 2

I had problems with solution from https://stackoverflow.com/a/55136139/6719538 when used DisposableSubscriber as subscribers, and as far as I can see this transformer don't consider calls Suscription#request from downstream subscribers (it could overflow them). I create my version that was tested in production - BufferTransformerHonorableToBackpressure.java. fang-yang - great respect for idea.