14. Flowables & Backpressure

Written by Alex Sullivan

You’ve been using Observables to do some pretty powerful stuff — but there’s one problem that you still need to cover. What happens if a subscriber can’t keep up with the next events that the Observable is emitting?

Backpresssure

That thorny scenario where operators or subscribers can’t consume next events as fast as an Observable may produce them is called backpressure, and you’ll explore it thoroughly in this chapter!

To start, open up the starter project for this chapter using IntelliJ IDEA. Navigate to SupportCode.kt and take a look around. You’ll find:

• The tried and true exampleOf method, a safeSleep method that simply calls Thread.sleep and catches any InterruptedExceptions.
• A freeMemory method that calculates the total mount of free memory the system has.

Fancy, right?

Now, head over to Main.kt and add the following code in main():

exampleOf("Zipping observable") {
  val fastObservable = Observable.interval(1, TimeUnit.MILLISECONDS)
  val slowObservable = Observable.interval(1, TimeUnit.SECONDS)
}

With the above, you’re creating two new Observables using the Observable.interval static factory method. The interval method creates an Observable that counts up from the provided number at a frequency you provide, forever, so it never terminates.

The two Observables are almost exactly the same, except one will emit a next event with a new number every millisecond, and the other will only emit every second.

Now, add the following right below the slowObservable line:

// 1
val disposable =
  Observables.zip(slowObservable, fastObservable)
    .subscribeOn(Schedulers.io())
    .subscribe { (first, second) ->
      // 2
      println("Got $first and $second")
    }
// 3
safeSleep(5000)
// 4
disposable.dispose()

That’s a solid chunk of code, so breaking it down step by step:

1. Create a new Observable by using the zip function, which, as you know, combines two Observables together. You’re using the RxKotlin factory function to keep everything neat. It also makes a Pair from the two emitted items for you.
2. Subscribe to the zipped Observable and print out both items.
3. Sleep the thread for five seconds. Since you’re subscribing to the zipped Observable on the io scheduler, the “Zipping Observable” example block would finish immediately if you didn’t sleep the thread. You’d never do this in a real application since the application would never terminate naturally like this one, but it’s necessary for the examples in this chapter.
4. As the Rx guru that you are by now, you never forget to dispose the subscriptions.

Run the Main.kt file. You should see the following:

--- Example of: Zipping observable ---
Got 0 and 0
Got 1 and 1
Got 2 and 2
Got 3 and 3

The next events are being zipped together, but it leaves one question unanswered: What’s happening to all the items that the fast Observable is emitting?

It took about five seconds to print out those four numbers, but we know in that time the fast Observable should have emitted thousands of items, since it should be emitting every millisecond.

It turns out that RxJava buffers those items under the hood. That means that it keeps a list of items that keeps growing until the downstream operators and subscribers can consume them.

Buffering danger!

Most of the time buffering next events is exactly what you want, but sometimes that buffering approach eats too much memory and can lead to OutOfMemoryError crashes!

You’ll create a new example that results in an OutOfMemoryError.

Copy the following code after the previous example:

exampleOf("Overflowing observer") {
  // 1
  val disposable = Observable.range(1, 10_000_000)
      // 2
      .subscribeOn(Schedulers.io())
      .observeOn(Schedulers.computation())
      // 3
      .subscribe {
        println("Free memory: ${freeMemory()}")
        safeSleep(100)
      }
  // 4
  safeSleep(20_000)
  disposable.dispose()
}

That’s a lot of code, so again we’ll break it down by section:

1. Create a new Observable using the range static factory method. The range method returns an Observable that emits integers starting from the first argument until the second argument — so between 1 and 10_000_000 in this example.
2. Subscribe on the io scheduler and observe on the computation scheduler. It’s important that you subscribe and observe on different threads for this example. You’ll see why later.
3. Subscribe to the Observable. In the subscribe lambda, print out the total amount of remaining free memory in the system and sleep the thread for 100 seconds. Sleeping for 100 milliseconds allows you to mimic a situation where the subscribing code is slower than the emitting code.
4. Sleep the thread for 20 seconds, so that the example has enough time to finish.

Run the code. You should see something like this:

--- Example of: Overflowing observer ---
Free memory: 3793645960
Free memory: 3769377888
Free memory: 3701230888
Free memory: 3608875976
...

But you probably won’t see an OutOfMemoryError. What gives, you may ask? RxJava is buffering integers, but an Int is so tiny that it doesn’t make much of a dent in your JVMs memory. So instead, you need to buff up the memory intensity of each item. Add the following code after the subscribeOn operator and before the observeOn operator:

.map {
  LongArray(1024 * 8)
}

Now, you’re taking each integer emitted by the range Observable and turning into a LongArray with a size of 8192. Now that’s a beefy object!

Run the code again. You should see some fireworks:

--- Example of: Overflowing observer ---
Free memory: 3793645432
Free memory: 3595255472
...
Free memory: 276971776
Free memory: 277053320
io.reactivex.exceptions.UndeliverableException: java.lang.OutOfMemoryError: Java heap space

As you consume more and more memory, and RxJava buffers more and more items, you’ll see the total amount of free memory decreasing and, eventually, you should see an OutOfMemoryError. Isn’t making software blow up the best?

Natural backpressure

Now, backpressure isn’t always a problem. In the previous example, try removing the observeOn line and run the example again.

No fireworks! What gives?

Since you removed the observeOn call, it means the subscribing code and the observing code are now both running on the same thread — the scheduler thread that you told it to run on with the subscribeOn call.

That means that, when you call safeSleep(100) in the subscribing lambda, the whole Rx chain stops for 100 milliseconds. The subscribing code is consuming items as fast as the Observable is emitting them — so there’s no backpressure!

What that means is that, if you’re not mucking about with observeOn and subscribeOn calls, you really don’t need to worry about backpressure.

Introduction to Flowables

But since you usually are mucking about with threading in your RxJava chains, the RxJava library has got your back.

Flowables are backpressure-aware versions of Observables that allow you to specify how you want to handle backpressure. A Flowable is distinct from an Observable, but they both share all of the operators and fun jazzy static constructors you’ve grown to love. You don’t need to worry about learning a whole new set of operators to go along with your new Flowable type.

Copy the following code for a new example into your main method below the previous examples:

exampleOf("Zipping flowable") {
  val slowFlowable = Flowable.interval(1, TimeUnit.SECONDS)
  val fastFlowable = Flowable.interval(1, TimeUnit.MILLISECONDS)
  val disposable =
    Flowables.zip(slowFlowable, fastFlowable)
      .subscribeOn(Schedulers.io())
      .observeOn(Schedulers.newThread())
      .subscribe { (first, second) ->
        println("Got $first and $second")
      }

  safeSleep(5000)
  disposable.dispose()
}

Hopefully, this code looks pretty familiar. It’s exactly the same code as you wrote in the “Zipping observable” example, except this time it’s using Flowable instead of Observable!

Just like Observable, Flowable has a static interval factory method that creates an instance of Flowable that counts up when subscribed to.

And just like in the previous example, you’re combining two Flowables — one fast and one slow. In the subscribe lambda you’re printing out both values from the Flowables.

There’s only one problem: You haven’t told your Flowables how to react to backpressure. Unlike Observable, Flowable won’t automatically buffer items. If you run this code, it’ll crash.

Since blowing up code is super fun, run it! You should see the following at least once in the resulting stack trace:

io.reactivex.exceptions.MissingBackpressureException: can’t deliver value 128 due to lack of requests

Since your Flowable won’t automatically buffer items like Observable does, you need to tell it how to handle that backpressure. Since the fast flowable is the one that will encounter backpressure, modify the fastFlowable value to look like the following example:

val fastFlowable = Flowable.interval(1, TimeUnit.MILLISECONDS)
    .onBackpressureDrop { println("Dropping $it") }

You’re using the onBackPressureDrop operator on your fast Flowable to instruct it how on how to handle backpressure. onBackPressureDrop can optionally take a Consumer function that lets you do something with the dropped item. Here, you’re just printing out the dropped value.

Run the code. It won’t crash this time and you should see the following:

--- Example of: Zipping flowable ---
Dropping 128
Dropping 129
Dropping 130
...
Got 0 and 0
...

No more crashing!

Why did the fast Flowable only start dropping items once it got to all the way up to the 128th value? that’s because using the observeOn method actually creates a buffer of size 128 to be more performant for bursty Flowables that can emit a lot of values at once and then stop.

Backpressure strategies

You’ve seen that you can remove back-pressured items from the stream by using the onBackpressureDrop method, but there’s actually three different ways you can handle backpressure:

1. onBackPressureDrop: Remove items from the stream as they come if the downstream consumer can’t handle them.
2. onBackPressureBuffer: Buffer the backpressured item up to a limit that you specify. You can then handle the case in which the buffer is overrun.
3. onBackPressureLatest: Hold onto the latest value and emit that value when the downstream consumer can handle it.

onBackPressureBuffer

Copy the following example into your project:

exampleOf("onBackPressureBuffer") {
  val disposable = Flowable.range(1, 100)
      .subscribeOn(Schedulers.io())
      .observeOn(Schedulers.newThread(), false, 1)
      .doOnComplete { println("We're done!") }
      .subscribe {
        println("Integer: $it")
        safeSleep(50)
      }
  safeSleep(1000)
  disposable.dispose()
}

Everything here should look pretty normal, with one exception. What’s going on with that observeOn line?

observeOn(Schedulers.newThread(), false, 1)

observeOn actually has an overloaded version of the operator that can take a boolean to delay the error across thread boundaries and, more interesting for this backpressure example, an int representing the internal buffer you learned about earlier in the chapter.

You’re setting that internal buffer value to 1 so you can clearly see the backpressure operator at play!

Note: In a real project you’d never want to set the buffer that low. Chances are that you’ll never actually want to change that buffer size either. But, if you do, make sure to give it a value of at least 16, so the performance of bursty sources doesn’t go down the tubes!

Now, it’s time to hook up the backpressure! Add the following line below the subscribeOn operator:

.onBackpressureBuffer(
  // 1
  50,
  // 2
  { println("Buffer overrun; dropping latest") },
  // 3
  BackpressureOverflowStrategy.DROP_LATEST
)

That’s a chunky operator, so breaking it down section by section:

1. onBackpressureBuffer takes in a maximum buffer count, which you’ve set to 50. If you end up needing to buffer more than 50 items, you’ll want a way to handle that situation. Which is good news because…
2. You’re also passing in a lambda to take an action if your buffer overruns. In this example, you’re just printing a message.
3. Since the buffer can overflow, you need to tell RxJava what to do in that scenario. Right now, you’re telling it to drop items that come in after the buffer overruns. You can instead use BackpressureOverflowStrategy.DROP_OLDEST to drop the oldest items in the buffer. Last but not least, you can use BackpressureOverflowStrategy.ERROR if you want to run into your old friend MissingBackpressureException.

If you run the example, you should see the following:

Integer: 1
Buffer overrun; dropping latest
Buffer overrun; dropping latest
...
Integer: 2
Integer: 3
...

The first item comes in without issue. Then, since the subscribing code is sleeping for 50 milliseconds, backpressure starts creeping up and you quickly overrun the size 50 buffer.

Since you’re using the DROP_LATEST overflow strategy, the later elements are dropped and the first items to be buffered are held onto, so once the upstream starts emitting again you get those buffered items.

onBackPressureLatest

Copy the following example into your project:

exampleOf("onBackPressureLatest") {
  val disposable = Flowable.range(1, 100)
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.newThread(), false, 1)
    .doOnComplete { println("We're done!") }
    .subscribe {
      println("Integer: $it")
      safeSleep(50)
    }
  safeSleep(1000)
  disposable.dispose()
}

This code looks pretty familiar, huh? But, as you’ve seen before, it’s missing a backpressure operator! Add the following line between subscribeOn and observeOn:

.onBackpressureLatest()

As mentioned before, onBackpressureLatest() instructs the Flowable to hold onto the latest back-pressured value and emit that when the downstream can handle it.

Run the example. you’ll see the following:

--- Example of: onBackPressureLatest ---
Integer: 1
Integer: 100
We're done!

The first item is emitted just like it was before. Then the Flowable encounters backpressure, all the way up until the last item, which, again emits OK.

You can think of onBackpressureLatest as being equivalent to using onBackpressureBuffer with a buffer size of one and a BackpressureOverflowStrategy of DROP_LATEST.

Built-in backpressure support

You’ve done a fantastic job handling backpressure in several ways. But there’s one more example to work through. It’s a quick one though!

Copy the following example into your project:

exampleOf("No backpressure") {
  val disposable = Flowable.range(1, 100)
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.newThread(), false, 1)
    .doOnComplete { println("We're done!") }
    .subscribe {
      println("Integer: $it")
      safeSleep(50)
    }
  safeSleep(1000)
  disposable.dispose()
}

Run the code. You should see the following:

--- Example of: No backpressure ---
Integer: 1
Integer: 2
Integer: 3
...

Now, you might be thinking, “Wait a minute. I thought that code would throw a MissingBackpressureException since there’s no onBackPressure... operator?!” Great observation - some Flowables actually support backpressure right out of the box!

The range operator will only produce new values when the downstream code requests them. That means that, if the subscribing code takes a long time, a new value will only be produced after it finishes its task and is ready for a new value.

Not all operators honor backpressure this way, so it’s important to look at the Javadocs for operators to see how they handle backpressure. Every Flowable operator will have a section in the Javadocs explaining how they handle backpressure.

Here’s an example of the range operators JavaDocs. You can see the section on backpressure in the image above.

Now here’s an example of the zip operators backpressure documentation.

You can see from the documentation that, as opposed to the range operator, zip expects you to handle the backpressure yourself.

Remember to check the documentation of all Flowable operators before using them to make sure you don’t get caught with unexpected backpressure handling!

Flowables, Observables, Processors and Subjects — Oh, My!

You may be feeling a little overwhelmed since you’ve just been given a whole new reactive type — Flowables! But don’t worry, Flowables are really just like Observables, but with more control over backpressure. You can even switch between the two types seamlessly.

Observable has an instance method on it called toFlowable. I know it’s crazy, but that method actually converts an Observable to a Flowable.

Since you’re moving from an Observable to a Flowable, you have to handle backpressure. toFlowable takes a BackpressureStrategy, which indicates how this Flowable should handle backpressure.

Note: BackpressureStrategy is different from the BackpressureOverflowStrategy you saw earlier, so don’t confuse them!

Choosing a BackpressureStrategy value

There’s five different BackpressureStrategy values you can pick from:

1. MISSING: Use this strategy if you’re planning to use one of the onBackpressureX strategies you saw earlier. If you don’t use one of the backpressure operators, you may get a MissingBackpressureException if you encounter backpressure.
2. ERROR: Signals MissingBackpressureException if the downstream can’t keep up.
3. BUFFER: Buffers all of the next events. This is similar to how an Observable handles backpressure by default.
4. DROP: Drops the most recent next events if the downstream can’t keep up.
5. LATEST: Keeps the latest next event, overriding it if the downstream can’t keep up.

You can see that many of the constants above are similar to the backpressure operators you saw earlier. If you need more fine-grained control when converting a Flowable to an Observable, you can always use one of the onBackpressure... methods you learned about in this chapter.

Add the following example at the bottom of the main class:

exampleOf("toFlowable") {
  val disposable = Observable.range(1, 100)
    .toFlowable(BackpressureStrategy.MISSING)
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.newThread(), false, 1)
    .subscribe {
      println("Integer: $it")
      safeSleep(50)
    }
  safeSleep(1000)
  disposable.dispose()
}

Just like before you’re using the range operator to create an Observable that emits integers. This time, however, you’re using the toFlowable method to convert the Observable into a Flowable, passing in MISSING as your backpressure strategy.

Can you guess what will happen when you run this code?

Run the app. Since you’re using the MISSING backpressure strategy and not applying one of the onBackPressure... operators, your Flowable blows up with a MissingBackpressureException.

Now swap out the BackpressureStrategy you’re supplying to the toFlowable operator in the above example with the BUFFER strategy:

.toFlowable(BackpressureStrategy.BUFFER)

Run the project again. This time you’ll see items printed normally. Try out the LATEST and DROP strategies as well. They should work exactly as you’d expect.

Processors

Since a Subject is just a fancy Observable, you could always use toFlowable on it to turn it into a Flowable. Just like before, you’ll have to supply a BackpressureStrategy. Alternatively, if you want a backpressure-aware version of your favorite subject, you can use the Processor type. it’s just like a Subject, except backpressure aware!

There’s a Processor type for each Subject you know and love! For example, if you want a backpressure aware version of BehaviorSubject, you can just use BehaviorProcessor.

Add the following example to the bottom of the main class:

exampleOf("Processor") {
  // 1
  val processor = PublishProcessor.create<Int>()
  // 2
  val disposable = processor
    .observeOn(Schedulers.newThread(), false, 1)
    .subscribe {
      println("Integer: $it")
      safeSleep(50)
    }
  // 3
  Thread().run {
    for (i in 0..100) {
      processor.onNext(i)
      safeSleep(5)
    }
  }
  safeSleep(1000)
  disposable.dispose()
}

This code is a bit different, so here’s a breakdown:

1. You’re creating a PublishProcessor, which acts just like a PublishSubject except it won’t buffer items if it experiences backpressure. PublishProcessor is to PublishSubject as Flowable is to Observable.
2. Just like before, you’re calling the overloaded version of observeOn to avoid any internal scheduler buffering. You’re subscribing directly to the processor and printing out the integer in the subscribe lambda. To simulate a slow subscriber, you’re using the safeSleep method to sleep the thread for 50 milliseconds. Unlike before, you’re not using the subscribeOn operator. Since you’ll be manually calling onNext on the processor, the items will be initially created on whatever thread calls onNext, so a subscribeOn call would have no affect.
3. In order to simulate items being generated on a separate thread, you’re creating a new Thread object and using the onNext method to send a range of integers into your processor. You’re again using the safeSleep method to ensure that all values aren’t delivered at once to emulate a more real world use case.

Note: Refer to Chapter 13, “Intro to Schedulers” chapter for more information on how subjects (and by extension processors) handle schedulers.

Run the project. Since you didn’t utilize one of the onBackPressure... methods, the project will crash with a MissingBackpressureException.

Processors will not buffer items delivered via the onNext method like a Subject would. You have to manually control the backpressure just like you do with a Flowable.

Update the example above to use the onBackPressureDrop operator:

val disposable = processor
  .onBackpressureDrop { println("Dropping $it") }
  .observeOn(Schedulers.newThread(), false, 1)
  .subscribe {
    println("Integer: $it")
    safeSleep(50)
  }

Now whenever an item is dropped due to backpressure you’ll print out a quick message explaining that the item’s been dropped.

Run the project. You should see output that looks like this:

--- Example of: Processor ---
Integer: 0
Dropping 1
Dropping 2
Dropping 3
Dropping 4
Dropping 5
Dropping 6
Dropping 7
Dropping 8
Dropping 9
Integer: 10
Dropping 11
...

Most items are dropped, and the ones that do make it through are printed out in your subscribe lambda. Processors are good to know about, but in the real world you’ll rarely need to use them. One example that might warrant using a Processor is if you find yourself sending large objects, like Bitmaps, through a Subject.

In order to avoid buffering lots of heavy duty, high memory objects you could use a Processor with the onBackpressureLatest operator. That way, only the freshest data would be stored in memory.

Key points

• Flowables offer a powerful tool for handling backpressure, which is when a stream is producing values faster than they can be consumed by an Observer. Most of the time you can ignore backpressure and use Observables, but Flowable can be super-handy if you need it.
• You’d typically use a Flowable if you have really large (like over 1000 items) streams that come at variable speeds. For example, image you have a web socket that sends down tons of data at random times. You might want to only handle the latest item, so you could use the onBackpressureLatest method to achieve that.
• If you have an Observable that emits Bitmaps (or other types which can have a really huge memory footprint), you might want to be aware of the fact that all the emitted Bitmaps will buffer if you can’t consume them fast enough, which could lead to an OutOfMemoryError. It might make sense to make use of one of the backpressure operators there as well.
• Similarly, if you are buffering high memory items into a Subject, consider using a Processor instead. Just make sure to add the proper onBackPressure... operator to ensure you aren’t hit with a MissingBackpressureException!

Flowables are a powerful and sometimes intimidating part of RxJava. But with this chapter’s help, you now have all the knowledge you need to tackle them in your own applications!

Where to go from here?

Backpressure is one of things that only show up when you least expect it. Before proceeding, invest some time in playing around with the examples in this chapter and test some operators to see what impact they have on the final result. Understanding backpressure will make your life easier with RxJava, and it will improve your confidence when working with Flowables.