How RxJava chain actually works

RxJava was created quite a while ago, but it is still widely used in large Android projects as the main tool for managing streams and multi-threading.

Sadly, barriers to entry are high. It took me so long to understand the fundamentals — days turned into months, which turned into years. (Even now, I dare not say that I know it all.)

RxJava has many traps and pitfalls. For example, the subscribeOn operator affects the thread of both the upper and the lower streams. You are not obliged to declare the thread where the result will be executed. But this may lead to crash in runtime. (For example, when you interact with the view not on the main thread).
To feel all the pain, just read this — Kotlin and Rx2. How I wasted 5 hours because of wrong brackets

Yet another post about RxJava. Why?

This not just a post, this is a heartfelt cry. RxJava framework became mainstream in 2015-2016. As I said, it is quite complicated and requires considerable effort from the developer to understand its many aspects.

Recently, Kotlin Flow has appeared and became a Google-recommended alternative to RxJava. It has skeletons in the closet, too, and hasn’t been tested by time and scale yet. However, it certainly has a great chance to oust RxJava from the market in two or three years.

Meanwhile developers still need to be ready to work with RxJava, as framework is still used in many applications. We may also encounter a product that combines RxJava and Flow with multiple mappers to each other.

I’ll explain the RxJava chain to you in simple terms, no fluff. This article is addressed both to those who make their first attempts to understand RX, and those who have years of practice.

The chain

Birth. Life. Death

The life cycle consists of four important steps:

1. Creation of Observables ⬇️
   Operators are created from top to bottom on the current execution thread. (Similar to the Builder pattern.)
2. Operators subscription ⬆️
   Performed from the bottom up. Each operator subscribes to the upper one
3. Data emission ⬇️
   Starts when all operators have successfully subscribed to each other. Goes from the top down.
4. Unsubscribing. The entire chain dies 😵
   Execution of all operators stops.

Let’s look at each of the stages in more detail.

Creation of observables ⬇️

First, Observable.just is created with the value “Hey”
If we look at the source code, we will see that our value is simply cached in ObservableJust:

After that, subscribeOn() is created similarly to Observable.just()

Note that only the creation of operators is happening here at this stage, without changing the thread.
The main difference from the previous operator is that this stream stores a reference to the upper Observable:

All other operators up to subscribe() are created in the same way and do not perform any actions.

The chain’s behaviour at the creation stage:

1. Observable.just("Hey")
   → caches the passed value
2. subscribeOn(Schedulers.io())
   → caches the subscription scheduler
3. map(String::length)
   → caches a lambda with a mapper
4. subscribeOn(Schedulers.computation())
   → caches the subscription scheduler
5. observeOn(AndroidSchedulers.mainThread())
   → caches the emitting scheduler
6. doOnSubscribe { doAction() }
   → caches the lambda that will be executed at subscribing
7. flatMap{ ... }
→ caches the lambda that will be executed at subscribing

As a result, each operator, except for the first one, contains a reference to the previous one, thus forming a LinkedList.
Creation will be executed at the chain call thread.

Let’s visualize again

Many questions may arise here. For example, which of the three subscribeOn’s will be applied to the chain and how many times will the thread change? I’ve seen different answers to these questions. Including that all subscribeOn’s will be ignored except for the top one.

Is this true? Let’s figure it out.

Subscription of operators ⬆️

The subscription process goes from the bottom up and starts immediately after the subscribe call.

At this stage, some operators can already be executed.

When we call subscribe, the subscription does not apply to the entire chain, but only to the flatMap, which is higher.
Then flatMap subscribes to the operator whose reference it saved while creating doOnSubscribe. This is why operators keep references to the upper streams.

Rx is like a cabbage that opens up gradually, leaf by leaf. Operator by operator (с) Alexey Tvorogov

The chain’s behaviour at the subscription stage:

1. flatMap{ ... }
→ Subscribes to the upper stream
2. doOnSubscribe{ ... }
   → Executes the code in the lambda
   → Subscribes to the upper stream
   → Execution thread: the one where subscribe is called
3. observeOn(AndroidSchedulers.mainThread())
→ Subscribes to the upper stream
4. subscribeOn(Schedulers.computation())
→ Changes thread to computation (previous thread = subscription thread)
   → Subscribes to the upper stream
5. map(String::length)
→ Subscribes to the upper stream
   → Subscription at computation
6. subscribeOn(Schedulers.io())
   → Changes thread to IO (previous thread = computation)
   → Subscribes to the upper stream

Of all the operators, only one doOnSubscribe and two subscribeOn were executed. The thread context changed twice: first to computation, and then to IO. That is, subscribeOn will be executed as many times as it is written.

Even if you write subscribeOn(Schedulers.io()) 3 times in a row, the previous 2 will not be ignored.

Let’s have a look at the following example:

This is how threads will change

Note that with Schedulers.io() under the hood, we use Executor with a dynamic thread pool. Chances are that sometimes thread instances will coincide, because a busy thread may become free and available again.
For example, if only this RX code is running now, the threads output will be as follows:

Thread[RxCachedThreadScheduler-1,5,main] //blue onsubscribe
Thread[RxCachedThreadScheduler-2,5,main] //red onsubscribe
Thread[RxCachedThreadScheduler-1,5,main] //yellow onsubscribe

However, no one can guarantee that exactly these instances will coincide.

On computation, the number of threads is limited and directly depends on the processor and the number of its cores.
We should only execute resource-consuming tasks on it, such as image compression or encryption. But definitely not input-output operations or any tasks with the file system.

Emission ⬇️

As soon as we reach Observable that does not have a reference to the parent stream, the emission process starts. It goes from the root Observable to the lowest one.

The chain’s behaviour:

1. Observsble.just("Hey")
→ Emits “Hey” on IO
2. subscribeOn(Schedulers.io())
→ Proxies “Hey” on the existing IO, and doesn’t change the thread
3. map(String::length)
   → Runs the code in lambda on IO
4. subscribeOn(Schedulers.computation())
→ Proxies the value, still on IO
5. observeOn(AndroidSchedulers.mainThread())
→ Changes thread to main, emits the value.
   It’s important to remember that if we’re already on the main thread, the element can be not emitted immediately, because in any case it will pass through Handler/Looper/MessageQueue.
6. doOnSubscribe { doAction() }
→ Proxies the value to the stream below, on main
7. flatMap{...}
   This one is tricky. First, let’s recall its contents:

The chain inside flatMap will go through all the stages of creation, subscription, emission. In this case, there will be no unsubscription, because timer will continue to emit values.

Operators timer/delay/interval subscribe to computation under the hood. This step isn’t obvious, but it affects the lower stream.

Therefore remember, if you combine these operators with http/database requests, change the scheduler to IO before executing this request. Otherwise, pass the subscription scheduler as a parameter directly to timer/delay/interval.

8. subscribe{ doAction() }
Performs an action on computation, returns Disposable. By using Disposable, you can manage the life of the entire subscription.

Conclusion

• All operators are created from top to bottom and subscribe to each other one by one from bottom to top. Thus they form a singly LinkedList.
  Emission goes from top to bottom starting from the root Observable.
• There are operators like startWith/doOnSubscribe that are executed during the subscription process. What’s more, they are executed as many times as you’ve written them. Their execution threads may be affected by the subscribeOn written below.
• subscribeOn is executed during the subscription process and affects both upper, and lower streams. It also can change the thread as many times as you write it. The one closest to the top of the chain is applied.
• observeOn affects only the lower streams, and is executed during emission. It changes the thread as many times as you write it.
• flatMap starts the chain only during root chain data emission.
  No actions are performed during the root stream subscription process.
• Operators interval/delay/timer subscribe to computation under the hood, by default. Remember to change the scheduler to IO when you use these operators with http/db requests.

In one of my next articles, I’m going to analyze Kotlin Flow in detail in the same way. Stay tuned!