Back we where discussing the `doOn_` operators for [side effects](/Part%203%20-%20Taming%20the%20sequence/1.%20Side%20effects.md#do), we left out `doOnRequested`.

public final Observable<T> doOnRequest(Action1<java.lang.Long> onRequest)

The `doOnRequested` meta-event happens when a subscriber requests for more items. The value supplied to the action is the number of items requested.

At this moment, `doOnRequest` is in beta. It is the only beta operator that we will discuss in this book. We're making an exception, because it enables us to peek into stable backpressure functionality that is otherwise hidden. Let's see what happens in the most simple observable

Observable.range(0, 3)

.doOnRequest(i -> System.out.println("Requested " + i))

.subscribe(System.out::println);

Output

We see that `subscribe` requests the maximum number of items from the beginning. That means that `subscribe` doesn't resist values at all. Subscribe will only use backpressure if we provide a subscriber that implements backpressure. Here is a complete example for such an implementation

public class ControlledPullSubscriber<T> extends Subscriber<T> {

private final Action1<T> onNextAction;

private final Action1<Throwable> onErrorAction;

private final Action0 onCompletedAction;

public ControlledPullSubscriber(

Action1<T> onNextAction,

Action1<Throwable> onErrorAction,

Action0 onCompletedAction) {

this.onNextAction = onNextAction;

this.onErrorAction = onErrorAction;

this.onCompletedAction = onCompletedAction;

}

public ControlledPullSubscriber(

Action1<T> onNextAction,

Action1<Throwable> onErrorAction) {

this(onNextAction, onErrorAction, () -> {});

}

public ControlledPullSubscriber(Action1<T> onNextAction) {

this(onNextAction, e -> {}, () -> {});

}

@Override

public void onStart() {

request(0);

}

@Override

public void onCompleted() {

onCompletedAction.call();

}

@Override

public void onError(Throwable e) {

onErrorAction.call(e);

}

@Override

public void onNext(T t) {

onNextAction.call(t);

}

public void requestMore(int n) {

request(n);

}

}

This simple implementation will not request values unless we manually make it do so with `requestMore`.

ControlledPullSubscriber<Integer> puller =

new ControlledPullSubscriber<Integer>(System.out::println);

Observable.range(0, 3)

.doOnRequest(i -> System.out.println("Requested " + i))

.subscribe(puller);

puller.requestMore(2);

puller.requestMore(1);

Output

First we requested no emissions. Then we requested 2 and we got 2 values.

Rx operators that use queues and buffers internally should use backpressure to avoid storing an infinite amount of values. Large-scale buffering should be left to operators that explicitly serve this purpose, such as `cache`, `buffer` etc. `zip` is one operator that needs to buffer items: the first observable might emit two or more values before the second observable emits its next value. Such small asymmetries are expected and they shouldn't cause the operator to fail. For that reason, `zip` has a small buffer of 128 items.

Observable.range(0, 300)

.doOnRequest(i -> System.out.println("Requested " + i))

.zipWith(

Observable.range(10, 300),

(i1, i2) -> i1 + " - " + i2)

.take(300)

.subscribe();

Output

The `zip` operator starts by requesting enough items to fill its buffer, and requests more when it has consumed enough. The details of how many items `zip` requests isn't interesting. What the reader should take away is the realisation that some buffering and backpressure exist in Rx whether the developer requests for it or not. This gives an Rx pipeline some flexibility where you might expect none. This might trick you into thinking that your code is solid, by silently saving small tests from failing, but you're not safe until you have explicitly declared behaviour with regard to backpressure.