RxJava 2 – Flowable

1. Introduction

RxJava is a Reactive Extensions Java implementation that allows us to write event-driven, and asynchronous applications. More information on how to use RxJava can be found in our intro article here.

RxJava 2 was rewritten from scratch, which brought multiple new features; some of which were created as a response for issues that existed in the previous version of the framework.

One of such features is the io.reactivex.Flowable.

2. Observable vs. Flowable

In the previous version of RxJava, there was only one base class for dealing with backpressure-aware and non-backpressure-aware sources – Observable.

RxJava 2 introduced a clear distinction between these two kinds of sources – backpressure-aware sources are now represented using a dedicated class – Flowable.

Observable sources don’t support backpressure. Because of that, we should use it for sources that we merely consume and can’t influence.

Also, if we’re dealing with a big number of elements, two possible scenarios connected with backpressure can occur depending on the type of the Observable.

In case of using a so-called “cold Observable“, events are emitted lazily, so we’re safe from overflowing an observer.

When using a “hot Observable” however, this will continue to emit events, even if the consumer can’t keep up.

3. Creating a Flowable

There are different ways to create a Flowable. Conveniently for us, those methods look similar to the methods in Observable in the first version of RxJava.

3.1. Simple Flowable

We can create a Flowable using the just() method similarly as we could with Observable :

Flowable<Integer> integerFlowable = Flowable.just(1, 2, 3, 4);

Even though using the just() is quite simple, it isn’t very common to create a Flowable from static data, and it’s used for testing purposes.

3.2. Flowable from Observable

When we have an Observable we can easily transform it to Flowable using the toFlowable() method:

Observable<Integer> integerObservable = Observable.just(1, 2, 3);
Flowable<Integer> integerFlowable = integerObservable
  .toFlowable(BackpressureStrategy.BUFFER);

Notice that to be able to perform the conversion, we need to enrich the Observable with a BackpressureStrategy. We’ll describe available strategies in the next section.

3.3. Flowable from FlowableOnSubscribe

RxJava 2 introduced a functional interface FlowableOnSubscribe, which represents a Flowable that starts emitting events after the consumer subscribes to it.

Due to that, all clients will receive the same set of events, which makes FlowableOnSubscribe backpressure-safe.

When we have the FlowableOnSubscribe we can use it to create the Flowable:

FlowableOnSubscribe<Integer> flowableOnSubscribe
 = flowable -> flowable.onNext(1);
Flowable<Integer> integerFlowable = Flowable
  .create(flowableOnSubscribe, BackpressureStrategy.BUFFER);

The documentation describes many more methods to create Flowable.

4. Flowable BackpressureStrategy

Some methods like toFlowable() or create() take a BackpressureStrategy as an argument.

The BackpressureStrategy is an enumeration, which defines the backpressure behavior that we’ll apply to our Flowable.

It can cache or drop events or not implement any behavior at all, in the last case, we will be responsible for defining it, using backpressure operators.

BackpressureStrategy is similar to BackpressureMode present in the previous version of RxJava.

There are five different strategies available in RxJava 2.

4.1. Buffer

If we use the BackpressureStrategy.BUFFER, the source will buffer all the events until the subscriber can consume them:

public void thenAllValuesAreBufferedAndReceived() {
    List testList = IntStream.range(0, 100000)
      .boxed()
      .collect(Collectors.toList());
 
    Observable observable = Observable.fromIterable(testList);
    TestSubscriber<Integer> testSubscriber = observable
      .toFlowable(BackpressureStrategy.BUFFER)
      .observeOn(Schedulers.computation()).test();

    testSubscriber.awaitTerminalEvent();

    List<Integer> receivedInts = testSubscriber.getEvents()
      .get(0)
      .stream()
      .mapToInt(object -> (int) object)
      .boxed()
      .collect(Collectors.toList());

    assertEquals(testList, receivedInts);
}

It’s similar to invoking onBackpressureBuffer() method on Flowable, but it doesn’t allow to define a buffer size or the onOverflow action explicitly.

4.2. Drop

We can use the BackpressureStrategy.DROP to discard the events that cannot be consumed instead of buffering them.

Again this is similar to using onBackpressureDrop() on Flowable:

public void whenDropStrategyUsed_thenOnBackpressureDropped() {
   
    Observable observable = Observable.fromIterable(testList);
    TestSubscriber<Integer> testSubscriber = observable
      .toFlowable(BackpressureStrategy.DROP)
      .observeOn(Schedulers.computation())
      .test();
    testSubscriber.awaitTerminalEvent();
    List<Integer> receivedInts = testSubscriber.getEvents()
      .get(0)
      .stream()
      .mapToInt(object -> (int) object)
      .boxed()
      .collect(Collectors.toList());

    assertThat(receivedInts.size() < testList.size());
    assertThat(!receivedInts.contains(100000));
 }

4.3. Latest

Using the BackpressureStrategy.LATEST will force the source to keep only the latest events, thus overwriting any previous values if the consumer can’t keep up:

public void whenLatestStrategyUsed_thenTheLastElementReceived() {
  
    Observable observable = Observable.fromIterable(testList);
    TestSubscriber<Integer> testSubscriber = observable
      .toFlowable(BackpressureStrategy.LATEST)
      .observeOn(Schedulers.computation())
      .test();

    testSubscriber.awaitTerminalEvent();
    List<Integer> receivedInts = testSubscriber.getEvents()
      .get(0)
      .stream()
      .mapToInt(object -> (int) object)
      .boxed()
      .collect(Collectors.toList());

    assertThat(receivedInts.size() < testList.size());
    assertThat(receivedInts.contains(100000));
 }

BackpressureStrategy.LATEST and BackpressureStrategy.DROP look very similar when we look at the code.

However, BackpressureStrategy.LATEST will overwrite elements that our subscriber can’t handle and keep only the latest ones, hence the name.

BackpressureStrategy.DROP, on the other hand, will discard elements that can’t be handled. This means that newest elements won’t necessarily be emitted.

4.4. Error

When we’re using the BackpressureStrategy.ERROR, we’re simply saying that we don’t expect backpressure to occur. Consequently, a MissingBackpressureException should be thrown if the consumer can’t keep up with the source:

public void whenErrorStrategyUsed_thenExceptionIsThrown() {
    Observable observable = Observable.range(1, 100000);
    TestSubscriber subscriber = observable
      .toFlowable(BackpressureStrategy.ERROR)
      .observeOn(Schedulers.computation())
      .test();

    subscriber.awaitTerminalEvent();
    subscriber.assertError(MissingBackpressureException.class);
}

4.5. Missing

If we use the BackpressureStrategy.MISSING, the source will push elements without discarding or buffering.

The downstream will have to deal with overflows in this case:

public void whenMissingStrategyUsed_thenException() {
    Observable observable = Observable.range(1, 100000);
    TestSubscriber subscriber = observable
      .toFlowable(BackpressureStrategy.MISSING)
      .observeOn(Schedulers.computation())
      .test();
    subscriber.awaitTerminalEvent();
    subscriber.assertError(MissingBackpressureException.class);
}

In our tests, we’re excepting MissingbackpressureException for both ERROR and MISSING strategies. As both of them will throw such exception when the source’s internal buffer is overflown.

However, it’s worth to note that both of them have a different purpose.

We should use the former one when we don’t expect backpressure at all, and we want the source to throw an exception in case if it occurs.

The latter one could be used if we don’t want to specify a default behavior on the creation of the Flowable. And we’re going to use backpressure operators to define it later on.

5. Summary

In this tutorial, we’ve presented the new class introduced in RxJava 2 called Flowable.

To find more information about the Flowable itself and it’s API we can refer to the documentation.