Flowable Subscriber: What request() method does actually?

Admin 1
Viewed 1586

Can anyone tell what the request() method in a DisposableSubscriber does and when it is used? We use it only when you create your own Flowable with Flowable.create? The official documentation says

request(long n): Requests the specified amount from the upstream if its Subscription is set via onSubscribe already.

But I'm not understanding what does it mean. To try out I made a sample as below

private Flowable<Long> streamOfNums() {
    return Flowable.create(e -> {
        for (int i = 0; i < 500; i++) {
            e.onNext((long) i);
            Log.d(TAG, "produced "+i);
        }
    }, BackpressureStrategy.BUFFER);
}

Consumes it like

        streamOfNums()
            .subscribeOn(Schedulers.io())
            .subscribeWith(new DisposableSubscriber<Long>() {

                @Override
                protected void onStart() {
                    super.onStart();
                    Log.d(TAG, "onStart: ");
                }

                @Override
                public void onNext(Long aLong) {
                    Log.d(TAG, "onNext: ");
                    try {
                        Log.d(TAG, "consuming data :"+aLong);
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    request(4);
                }

                @Override
                public void onError(Throwable t) {

                }

                @Override
                public void onComplete() {

                }
            });

And what I can see is each time the emitter produces numbers after the delay (2000 ms) given. I have given request(4) but even without that it works in the exact same way. can anybody explain when what request does and when it is used. Can it be used for pagination scenarios?

rx-java2 rx-java android
1 Answers

request allows the consumer to tell the producer how many elements to produce. By default, DisposableSubscriber requests Long.MAX_VALUE in its onStart() method in which case further request() calls have no effect.

There is rarely the need to actually call request in such end-consumers, but otherwise you could use it to avoid buffer overflow when your end-consumer acts as an asynchronous boundary:

ExecutorService executor = Executors.newSingleThreadedExecutor();

Flowable.range(1, 500)
    .doOnNext(v -> Log.d("produced: " + v))
    .subscribeOn(Schedulers.io())
    .subscribe(new DisposableSubscriber<Long>() {
            @Override protected void onStart() {
                Log.d(TAG, "onStart: "); // <----- no super.onStart() here!
                request(1);
            }
            @Override public void onNext(Long aLong) {
                executor.execute(() -> {
                    Log.d(TAG, "onNext: ");
                    try {
                        Log.d(TAG, "consuming data :"+aLong);
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    request(1);
                });
            }
            @Override public void onError(Throwable t) {
                 executor.execute(() -> t.printStackTrace());
            }

            @Override public void onComplete() {
                 executor.execute(() -> Log.d("onComplete"));
            }
        });

Thread.sleep(100_000);
executor.shutdown();
user_611580
Related