rx-java Backpressure Creating backpressured data sources

Example

Creating backpressured data sources is the relatively easier task when dealing with backpressure in general because the library already offers static methods on Observable that handle backpressure for the developer. We can distinguish two kinds of factory methods: cold "generators" that either return and generate elements based on downstream demand and hot "pushers" that usually bridge non-reactive and/or non-backpressurable data sources and layer some backpressure handling on top of them.

just

The most basic backpressure aware source is created via just:

Observable.just(1).subscribe(new Subscriber<Integer>() {
    @Override
    public void onStart() {
        request(0);
    }

    @Override
    public void onNext(Integer v) {
        System.out.println(v);
    }
   
    // the rest is omitted for brevity
}

Since we explicitly don't request in onStart, this will not print anything. just is great when there is a constant value we'd like to jump-start a sequence.

Unfortunately, just is often mistaken for a way to compute something dynamically to be consumed by Subscribers:

int counter;

int computeValue() {
   return ++counter;
}

Observable<Integer> o = Observable.just(computeValue());

o.subscribe(System.out:println);
o.subscribe(System.out:println);

Surprising to some, this prints 1 twice instead of printing 1 and 2 respectively. If the call is rewritten, it becomes obvious why it works so:

int temp = computeValue();

Observable<Integer> o = Observable.just(temp);

The computeValue is called as part of the main routine and not in response to the subscribers subscribing.

fromCallable

What people actually need is the method fromCallable:

Observable<Integer> o = Observable.fromCallable(() -> computeValue());

Here the computeValue is executed only when a subscriber subscribes and for each of them, printing the expected 1 and 2. Naturally, fromCallable also properly supports backpressure and won't emit the computed value unless requested. Note however that the computation does happen anyway. In case the computation itself should be delayed until the downstream actually requests, we can use just with map:

Observable.just("This doesn't matter").map(ignored -> computeValue())...

just won't emit its constant value until requested when it is mapped to the result of the computeValue, still called for each subscriber individually.

from

If the data is already available as an array of objects, a list of objects or any Iterable source, the respective from overloads will handle the backpressure and emission of such sources:

 Observable.from(Arrays.asList(1, 2, 3, 4, 5)).subscribe(System.out::println);

For convenience (and avoiding warnings about generic array creation) there are 2 to 10 argument overloads to just that internally delegate to from.

The from(Iterable) also gives an interesting opportunity. Many value generation can be expressed in a form of a state-machine. Each requested element triggers a state transition and computation of the returned value.

Writing such state machines as Iterables is somewhat complicated (but still easier than writing an Observable for consuming it) and unlike C#, Java doesn't have any support from the compiler to build such state machines by simply writing classically looking code (with yield return and yield break). Some libraries offer some help, such as Google Guava's AbstractIterable and IxJava's Ix.generate() and Ix.forloop(). These are by themselves worthy of a full series so let's see some very basic Iterable source that repeats some constant value indefinitely:

Iterable<Integer> iterable = () -> new Iterator<Integer>() {
    @Override
    public boolean hasNext() {
        return true;
    }

    @Override
    public Integer next() {
        return 1;
    }
};

Observable.from(iterable).take(5).subscribe(System.out::println);

If we'd consume the iterator via classic for-loop, that would result in an infinite loop. Since we build an Observable out of it, we can express our will to consume only the first 5 of it and then stop requesting anything. This is the true power of lazily evaluating and computing inside Observables.

create(SyncOnSubscribe)

Sometimes, the data source to be converted into the reactive world itself is synchronous (blocking) and pull-like, that is, we have to call some get or read method to get the next piece of data. One could, of course, turn that into an Iterable but when such sources are associated with resources, we may leak those resources if the downstream unsubscribes the sequence before it would end.

To handle such cases, RxJava has the SyncOnSubscribe class. One can extend it and implement its methods or use one of its lambda-based factory methods to build an instance.

SyncOnSubscribe<Integer, InputStream> binaryReader = SyncOnSubscribe.createStateful(
     () -> new FileInputStream("data.bin"),
     (inputstream, output) -> {
         try {
             int byte = inputstream.read();
             if (byte < 0) {
                 output.onCompleted();
             } else {
                 output.onNext(byte);
             }
         } catch (IOException ex) {
             output.onError(ex);
         }
         return inputstream;
     },
     inputstream -> {
         try {
             inputstream.close();
         } catch (IOException ex) {
             RxJavaHooks.onError(ex);
         }
     } 
 );

 Observable<Integer> o = Observable.create(binaryReader);

Generally, SyncOnSubscribe uses 3 callbacks.

The first callbacks allows one to create a per-subscriber state, such as the FileInputStream in the example; the file will be opened independently to each individual subscriber.

The second callback takes this state object and provides an output Observer whose onXXX methods can be called to emit values. This callback is executed as many times as the downstream requested. At each invocation, it has to call onNext at most once optionally followed by either onError or onCompleted. In the example we call onCompleted() if the read byte is negative, indicating and end of file, and call onError in case the read throws an IOException.

The final callback gets invoked when the downstream unsubscribes (closing the inputstream) or when the previous callback called the terminal methods; it allows freeing up resources. Since not all sources need all these features, the static methods of SyncOnSubscribe let's one create instances without them.

Unfortunately, many method calls across the JVM and other libraries throw checked exceptions and need to be wrapped into try-catches as the functional interfaces used by this class don't allow throwing checked exceptions.

Of course, we can imitate other typical sources, such as an unbounded range with it:

SyncOnSubscribe.createStateful(
     () -> 0,
     (current, output) -> {
         output.onNext(current);
         return current + 1;
     },
     e -> { }
);

In this setup, the current starts out with 0 and next time the lambda is invoked, the parameter current now holds 1.

There is a variant of SyncOnSubscribe called AsyncOnSubscribe that looks quite similar with the exception that the middle callback also takes long value that represents the request amount from downstream and the callback should generate an Observable with the exact same length. This source then concatenates all these Observable into a single sequence.

 AsyncOnSubscribe.createStateful(
     () -> 0,
     (state, requested, output) -> {
         output.onNext(Observable.range(state, (int)requested));
         return state + 1;
     },
     e -> { }
 );

There is an ongoing (heated) discussion about the usefulness of this class and generally not recommended because it routinely breaks expectations about how it will actually emit those generated values and how it will respond to, or even what kind of request values it will receive in more complex consumer scenarios.

create(emitter)

Sometimes, the source to be wrapped into an Observable is already hot (such as mouse moves) or cold but not backpressurable in its API (such as an asynchronous network callback).

To handle such cases, a recent version of RxJava introduced the create(emitter) factory method. It takes two parameters:

• a callback that will be called with an instance of the Emitter<T> interface for each incoming subscriber,
• a Emitter.BackpressureMode enumeration that mandates the developer to specify the backpressure behavior to be applied. It has the usual modes, similar to onBackpressureXXX in addition to signalling a MissingBackpressureException or simply ignoring such overflow inside it altogether.

Note that it currently doesn't support additional parameters to those backpressure modes. If one needs those customization, using NONE as the backpressure mode and applying the relevant onBackpressureXXX on the resulting Observable is the way to go.

The first typical case for its use when one wants to interact with a push-based source, such as GUI events. Those APIs feature some form of addListener/removeListener calls that one can utilize:

Observable.create(emitter -> {
    ActionListener al = e -> {
        emitter.onNext(e);
    };

    button.addActionListener(al);

    emitter.setCancellation(() -> 
        button.removeListener(al));

}, BackpressureMode.BUFFER);

The Emitter is relatively straightforward to use; one can call onNext, onError and onCompleted on it and the operator handles backpressure and unsubscription management on its own. In addition, if the wrapped API supports cancellation (such as the listener removal in the example), one can use the setCancellation (or setSubscription for Subscription-like resources) to register a cancellation callback that gets invoked when the downstream unsubscribes or the onError/onCompleted is called on the provided Emitterinstance.

These methods allow only a single resource to be associated with the emitter at a time and setting a new one unsubscribes the old one automatically. If one has to handle multiple resources, create a CompositeSubscription, associate it with the emitter and then add further resources to the CompositeSubscription itself:

Observable.create(emitter -> {
    CompositeSubscription cs = new CompositeSubscription();

    Worker worker = Schedulers.computation().createWorker();

    ActionListener al = e -> {
        emitter.onNext(e);
    };

    button.addActionListener(al);

    cs.add(worker);
    cs.add(Subscriptions.create(() -> 
        button.removeActionListener(al));

    emitter.setSubscription(cs);

}, BackpressureMode.BUFFER);

The second scenario usually involves some asynchronous, callback-based API that has to be converted into an Observable.

Observable.create(emitter -> {
    
    someAPI.remoteCall(new Callback<Data>() {
        @Override
        public void onSuccess(Data data) {
            emitter.onNext(data);
            emitter.onCompleted();
        }

        @Override
        public void onFailure(Exception error) {
            emitter.onError(error);
        }
    });

}, BackpressureMode.LATEST);

In this case, the delegation works the same way. Unfortunately, usually, these classical callback-style APIs don't support cancellation, but if they do, one can setup their cancellation just like in the previoius examples (with perhaps a more involved way though). Note the use of the LATEST backpressure mode; if we know there will be only a single value, we don't need the BUFFER strategy as it allocates a default 128 element long buffer (that grows as necessary) that is never going to be fully utilized.