Callstack blocking as a flow-control alternative to backpressure

Another way of handling an overproductive Observable is to block the callstack (parking the thread that governs the overproductive Observable). This has the disadvantage of going against the “reactive” and non-blocking model of Rx. However this can be a viable option if the problematic Observable is on a thread that can be blocked safely. Currently RxJava does not expose any operators to facilitate this.

If the Observable, all of the operators that operate on it, and the observer that is subscribed to it, are all operating in the same thread, this effectively establishes a form of backpressure by means of callstack blocking. But be aware that many Observable operators do operate in distinct threads by default (the javadocs for those operators will indicate this).

How a subscriber establishes “reactive pull” backpressure

When you subscribe to an Observable with a Subscriber, you can request reactive pull backpressure by calling Subscriber.request(n) in the Subscriber’s onStart() method (where n is the maximum number of items you want the Observable to emit before the next request() call).

Then, after handling this item (or these items) in onNext(), you can call request() again to instruct the Observable to emit another item (or items). Here is an example of a Subscriber that requests one item at a time from someObservable:

someObservable.subscribe(new Subscriber<t>() {
    @Override
    public void onStart() {
      request(1);
    }

    @Override
    public void onCompleted() {
      // gracefully handle sequence-complete
    }

    @Override
    public void onError(Throwable e) {
      // gracefully handle error
    }

    @Override
    public void onNext(t n) {
      // do something with the emitted item "n"
      // request another item:
      request(1);
    }
});

You can pass a magic number to request, request(Long.MAX_VALUE), to disable reactive pull backpressure and to ask the Observable to emit items at its own pace. request(0) is a legal call, but has no effect. Passing values less than zero to request will cause an exception to be thrown.

Reactive pull backpressure isn’t magic

Backpressure doesn’t make the problem of an overproducing Observable or an underconsuming Subscriber go away. It just moves the problem up the chain of operators to a point where it can be handled better.

Let’s take a closer look at the problem of the uneven zip.

You have two Observables, A and B, where B is inclined to emit items more frequently than A. When you try to zip these two Observables together, the zip operator combines item n from A and item n from B, but meanwhile B has also emitted items n+1 to n+m. The zip operator has to hold on to these items so it can combine them with items n+1 to n+m from A as they are emitted, but meanwhile m keeps growing and so the size of the buffer needed to hold on to these items keeps increasing.

You could attach a throttling operator to B, but this would mean ignoring some of the items B emits, which might not be appropriate. What you’d really like to do is to signal to B that it needs to slow down and then let B decide how to do this in a way that maintains the integrity of its emissions.

The reactive pull backpressure model lets you do this. It creates a sort of active pull from the Subscriber in contrast to the normal passive push Observable behavior.

The zip operator as implemented in RxJava uses this technique. It maintains a small buffer of items for each source Observable, and it requests no more items from each source Observable than would fill its buffer. Each time zip emits an item, it removes the corresponding items from its buffers and requests exactly one more item from each of its source Observables.

(Many RxJava operators exercise reactive pull backpressure. Some operators do not need to use this variety of backpressure, as they operate in the same thread as the Observable they operate on, and so they exert a form of blocking backpressure simply by not giving the Observable the opportunity to emit another item until they have finished processing the previous one. For other operators, backpressure is inappropriate as they have been explicitly designed to deal with flow control in other ways. The RxJava javadocs for those operators that are methods of the Observable class indicate which ones do not use reactive pull backpressure and why.)

For this to work, though, Observables A and B must respond correctly to the request(). If an Observable has not been written to support reactive pull backpressure (such support is not a requirement for Observables), you can apply one of the following operators to it, each of which forces a simple form of backpressure behavior:

onBackpressureBuffer

maintains a buffer of all emissions from the source Observable and emits them to downstream Subscribers according to the requests they generate

an experimental version of this operator (not available in RxJava 1.0) allows you to set the capacity of the buffer; applying this operator will cause the resulting Observable to terminate with an error if this buffer is overrun​

onBackpressureDrop

drops emissions from the source Observable unless there is a pending request from a downstream Subscriber, in which case it will emit enough items to fulfill the request
​

onBackpressureBlock (experimental, not in RxJava 1.0)

blocks the thread on which the source Observable is operating until such time as a Subscriber issues a request for items, and then unblocks the thread only so long as there are pending requests

If you do not apply any of these operators to an Observable that does not support backpressure, and if either you as the Subscriber or some operator between you and the Observable attempts to apply reactive pull backpressure, you will encounter a MissingBackpressureException which you will be notified of via your onError() callback.

Work in progress...

Things that may need explaining:

• the Producer interface (and its request method)
• other new method in Subscriber:
  • setProducer(p)
• how and when to support producers in custom observables & operators
  • point here from the "how to make a custom operator" page; maybe also from create operator doc
• Meanwhile, for info on Producers, see:
  • Pull #2963
  • Advanced RxJava a blog from Dávid Karnok

Work in progress...

See also

• RxJava 0.20.0-RC1 release notes