Operator fusion in RxJava 2

November 2, 2019


RxJava is a very powerful library, though it has some issues as well. Specifically performance and memory issues, which come from problems library tries to solve and how the solution is designed from technical perspective. In order to minimize overhead in RxJava there are a number of optimizations, which are called “operator fusion”. And we’ll talk about them in this article. But first let’s recap how RxJava reactive types work and what issues they have.


When working with Observable there are three major pieces: Observable, Observer and Disposable. We all know Observable and how it is created (e.g. Observable.just(“Hello, World!”)). Observable is a building block of each Rx-chain. To make Observable work, we need to subscribe to it with passing Observer to the subscribe(…) method. Observer is basically an interface with callbacks onSubscribe, onNext, onError and onComplete. When we subscribe to Observable with some Observer, Observable creates Disposable object and passes it to Observer (so Observer can later dispose Rx-chain if needed) via onSubscribe callback. After that is done — communication is established and Observable can emit some values via onNext without any additional waiting. Therefore Observable doesn’t support backpressure — there is no way for Observer to notify Observable that it should wait and not emit more values.


With Flowable everything is similar, but instead of Observer and Disposable we have Subscriber and Subscription. Subscription has additional request(n) method, using which Subscriber can explicitly request Flowable to emit requested amount of items. Without requesting values Flowable won’t emit anything, that is why Flowable supports backpressure.

Assembly and subscribe

When working with RxJava reactive types there are two important stages: assembly and subscribe. On assembly Rx-chain is built, on subscribe — we “start” Rx-chain.

Consider following example:

In this case we go from top to bottom and the following happens:

So you can see that quite a lot of things happen with this short Rx-chain. If this chain was of type Flowable then additional communication with request(n) would also take place making this even more complicated.

Queues and synchronization

Operators under the hood might have their internal Queues for processing events. Access to this queue should be serialized (that means that it should be accessed with proper synchronization). RxJava2 has non-blocking synchronization based on Atomics (e.g. AtomicInteger) and infinite loops with compareAndSet method (for better performance). So inside the library it is common to see some code like this:

Queue in an operator along with Atomic objects also provides an overhead if we consider that each operator in a chain can have its own Queue.


So considering all the above, issues which RxJava has are:

Operator fusion

To resolve some of the performance and memory issues there is “operator fusion”.

There are two types of fusion:

Macro fusion on Assembly

On Assembly

Macro fusion on assembly is focused on minimizing number of Observables and objects created during assembly. When we talk about “assembly” we mean this place:

Basic on Assembly fusion

The simplest way to optimize some Observables is to add checks for special cases, so that created Observables are simpler in terms of implementation than general one. As an example we can look at Observable.fromArray, which can downgrade to Observable.empty or Observable.just if there are 0 or 1 items provided respectively:


First “advanced” optimization from fuseable package we’ll look onto is ScalarCallable interface:

It is basically that same interface as common java Callable with the difference that it doesn’t throw exceptions. ScallarCallable is an interface which reactive type can implement to say that it holds constant value which can be safely extracted during assembly time. Specifically such reactive type can either contain exactly one item or no items at all. Therefore when we call call method — we check returned value: if it is null — then reactive type doesn’t have any value, if it returned non-null value — then it has only that value. Based on the described contract only empty and just operators on Observable, Flowable and Maybe can be marked with this interface.

Then for example in xMap operators (flatMap, switchMap, concatMap) we can apply the optimization if source is marked with this interface:

In case source is marked with ScalarCallable interface we can switch to simplified version of xMap instead of full implementation (which is quite heavy).


In the fuseable package there are three interfaces:

Let’s take a closer look at FuseToObservable, for other interfaces everything will be similar.

Consider we have the following Rx-chain:

Here we create some range and want to count number of emitted items. count operator returns Single, but we would like to have an Observable (for example we want to merge this result with some other Observables). Then we add additional toObservable operator to our Rx-chain making it more complex and heavy.

FuseToObservable comes with a help here. What this interface does is saying that some operators which return reactive type other than Observable under the hood have some implementation which returns Observable, and that implementation can be extracted inside toObservable call.

Considering our example with count:

By default it returns ObservableCountSingle, but if we look into implementation of that operator we’ll see that it implements FuseToObservable interface and can provide different implementation when called in fused mode:

And when we call toObservable that implementation will be extracted which effectively means that toObservable Observable won’t be created:

Macro fusion on subscribe

Macro fusion on subscribe is focused on the same type of optimizations which are done on assembly, but they are done inside subscribeActual method:

Sometimes it is just not possible to apply optimization during assembly, because data is unknown before subscription, sometimes it is just more convenient to do some optimization on subscribe rather than on assembly.

Basic on subscribe fusion

Same as with assembly there is simple optimization which checks for some special conditions to use simplified implementation instead of general one. For example Observable.amb checks number of sources provided to decide whether heavy AmbCoordinator should be instantiated or not:


During on assembly we have some optimization with ScalarCallable interface. For subscribe time we have similar optimizations with Callable interface.

NOTE: as ScalarCallable extends Callable — any optimization which can be applied during assembly on ScalarCallable also can be applied to Callable on subscribe

For example in xMap operators during subscribe for Observables marked with Callable interface it is possible to switch to simplified implementation:

Micro fusion

Micro fusion is aimed to minimize overhead by reducing some synchronization or sharing internal structures as queues.


Let’s look at the example of what happens when operator like Flowable.filter is used:

We have Upstream, our filter operator and the downstream. Let’s imagine that our filter function checks whether or not value is less than 5. After subscription is established downstream has to request some number of items from upstream:

As you can see there is a lot of communication between operators, and more importantly each event comes in a serialized way which means that it implies some overhead.

Consider we have two Filter operators in between — communication can be increased significantly so as the overhead:

This is where ConditionalSubscriber comes to a rescue:

Usually onNext callback in Subscriber returns no value, as upstream just provides values via that callback and waits for new requests from downstream. ConditionalSubscriber has additional method tryOnNext which is similar to onNext with a difference that it immediately returns boolean on whether value was accepted or rejected. This can minimize number of required request(n) calls as upstream receives direct response.

If we look for example at the Flowable.filter implementation we can see that basically upstream filter can directly access predicate of downstream filter predicate:

And this can save us some request calls:

But it would not be that awesome if this optimization aimed to make less overhead with chained filter operators (as it seems anyway can be written inside one filter operator). The good thing is that Flowable.map also implements ConditionalSubscriber, which makes less overhead when have multiple filter and map operators chained together:

Queue fuseable

The most complicated micro fusion is based on sharing internal queues between operators. The whole optimization is based on the QueueSubscription interface

It is basically just Queue and Subscription under one interface. But Queue interface is not just simple interface from Java, instead it has additional method requestFusion in it:

The idea is that comparing with usual communication between Flowable and Subscriber via onXXX callbacks, upstream can provide not just Subscription, but QueueSubscription allowing downstream to directly access internal queue.

The mechanics is as following. First of all during onSubscribe upstream and downstream should agree on the fusion and select fusion mode in which they will work. If they agree on the fusion — then new communication implementation will be used, if they don’t — traditional communication via onXXX callbacks will be established.

Generally after fusion is established downstream will get items directly by calling poll() method on the upstream queue:

There are three fusion modes:

ANY — is just SYNC or ASYNC (what exactly will be established is based on what mode upstream supports).

SYNC fusion

SYNC fusion is available only when upstream values are either already statically available or are generated when poll() is called synchronously.

If upstream and downstream agree to use sync fusion mode, this will lead to the following contract:

Example of operator which supports SYNC fusion mode is Flowable.range:

ASYNC fusion

ASYNC fusion mode is available when upstream values may become available to poll() eventually.

If upstream and downstream agree to use async fusion mode, this will lead to the following contract:

Yes, it is possible to have null value in onNext in RxJava

Example of operator which supports ASYNC fusion mode is Flowable.filter:

We’ve looked at some examples of operators which support fusion modes, but to enable this mode one should request fusion first. For example fusion mode is requested inside Flowable.flatMap for InnerSubscriber:

Here you see that during on subscribe, when source implements QueueSubscription fusion mode ANY is requested. And depending on what mode was accepted by source different strategies are applied.

QueueSubscription threading

With queue fusion it is important to pay attention to threading issue. If we allow downstream to access upstream’s queue it may lead to an issue if upstream and downstream are working on different threads:

It might happen that inside map we have some heavy computation, which (in case of direct polling) may leak that computation to a different thread. To resolve this issue there is additional marker option BOUNDARY, which says that caller of the poll method might do that on a different thread. Then operators should either ignore BOUNDARY option and allow accessing its queue from another thread or decline fusion in case BOUNDARY option is requested.

If we look at the Observable.map implementation we can see that it uses transitiveBoundaryFusion helper function:

Which inside states that BOUNDARY mode is not allowed:


In this article we had an overview of some optimizations in RxJava and found some interesting things:

So, don’t think that under the hood RxJava does everything effectively and now you can write long Rx-chains. Benchmark your code, profile important chains and try to find out how to optimize each separately.

Further reading: Operator-fusion (Part 1) Operator fusion (part 2 - final)

Happy coding