This week 1/2021 - RxJava 3.x - back pressure

I got deep dive into RxJava tutorial and train a lot of possible cases. However most interesting for me is back pressure feature. 

Going throw the mechanic of RxJava for this case, at the moment of subscription generator is initialized (line 3), then subscriber onStart method is called (line 13) and first elements are generated (depending of buffer size).

In onStart method should be "request" called defining how many elements should be generated a the begging. If "request" method is missing this result no further action.

In the following example there is no difference if we ask for more elements than one, generator always waits to buffer be empty in 75% (the level when buffer is refilled is defined in BaseObserveOnSubscriber) and then start to generate.

When calling onComplete and onError flow is intermittent.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37

public void backpressure() { final Flowable<Integer> generate = Flowable.generate( () -> "INITIALIZED", //initial state (state, emitter) -> { //current state // do some producing task emitter.onNext(product.get()); return "RUNNING"; // next state } ); final DefaultSubscriber<Integer> subscriber = new DefaultSubscriber<Integer>() { @Override protected void onStart() { request(1); } @Override public void onNext(final Integer o) { request(1); // do some consumer work } @Override public void onError(final Throwable throwable) { } @Override public void onComplete() { } }; int bufferSize = 6; generate.observeOn(Schedulers.newThread(), false, bufferSize) .subscribe(subscriber); }

 

I was training on version 3.0.11 of RxJava.

Sources:

[1] https://github.com/Froussios/Intro-To-RxJava

[2] https://blog.frankel.ch/backpressure-reactive-systems/