RxJava Cheat Sheet, With a Pinch of Android

Today we're talking about RxJava, the library that implements the ideas of the reactive extensions, which is a library for composing asynchronous and event-based programs using observable sequences.

Download and print out this cheat sheet so you can use it whenever you need. To get fuller explanations and more detailed content in the cheat sheet, continue reading this blog post!

Get The Cheat Sheet

Basic RxJava Classes

RxJava is an implementation of the reactive extensions, a library for composing asynchronous and event-based programs using observable sequences. As such it offers the implementations of the abstractions to create streams of events and items that you can subscribe to, in your code and react to the data flowing through those streams. Note, these are different from the streams you may be familiar with in the Java 8 Streams API.

Here are the basic RxJava classes you should know about:

• Observable - emits 0 or n items of T and terminates with complete or an error.
• Flowable - an addition that came in RxJava 2. It acts as an Observable, and emits 0 or n items. It terminates with complete or an error. But it also supports back pressure, which lets you control how fast the source emits items. It's really handy when you're building complex systems, since a fast producer of data cannot flood the system with items waiting to be processed.
• Single, an Observable, which either emits a single item or throws an error. Think of it as the reactive version of a method call. In general you can think of it also as a performance optimization of Observable, when you know that only a single item can pass through, you might cut some corners in the implementation and drop some checks to make it work faster.
• Maybe succeeds with either an item, no item, or errors. The reactive version of an Optional.
• Completable - either completes or returns an error. It never return items. Think of it as a reactive version of a Runnable. 

There are more classes in the API you need to know, but these are responsible for creating the streams and producing the data. On the other side, data is processed via Observers. We'll talk about them in a moment.

Observables in RxJava

Most of the time in a system, you'll have some way to get the observables from the API. Here's how you can create an observable from a value, just call the just method. There's also a way to create an Observable from a collection, then it'll emit all the items in the collection one by one and the observers will have a chance to process them.

Observable.just("RebelLabs");
Observable.fromIterable(iterable);
Observable.fromCallable(callable);

The attentive reader might have nodded knowingly at our third example in the snippet above. The one where you can create an Observable from a callable. This comes really handy when you're building an asynchronous API.

Subscribing to Observables

Now we have the building blocks to create reactive streams of data. The missing piece is processing the data that flows through them. This is exactly what Observers allow you to do. Observers provide a mechanism for receiving data and notifications from Observables using the following API:

• onNext(T t) - provides the Observer with a new item to observe.
• onError(Throwable e) - notifies the Observer that the Observable has experienced an error condition.
• onComplete() - notifies the Observer that the Observable has finished sending push-based notifications.

To subscribe to the observables you just implement the Observer interface, put your business logic into the onNext method while using the other two manage the lifecycle notifications from the observable.

Publishing to Multiple Observers in RxJava

Sometimes, you want to have more than a single observer being notified of the items emitted from an observable. Imagine for example that you're building a UI and need to process user actions. You might want to, for example, log the action, start a computation or change the UI to convey that an operation is in progress. All these actions are independent and it'd be smart not to mix them up with business logic. So you'd want multiple observers to receive an item saying something happened, so the logger would log, the UI updater would update the UI, and so one.

Luckily, there's a way to do that. You're need to use publish() to subscribe multiple Observers to an Observable. After that the observavle would be able to receive multiple observers and in fact it won't emit any items until you call the connect method. Then it'll start notifying all the observers of the items.

Observable topic = Observable.publish().autoConnect(2); 
topic.subscribe(subscriber1); 
topic.subscribe(subscriber2);

You can also use the autoConnect(int) method to say how many observers you're going to connect, then you don't need to call the connect method manually and the items will flow automatically when all the observers are attached.

Data Processing Functions in RxJava

In this section we look at the most frequently used functions used process the data flowing through the observables. In a nutshell, by applying these on an observable you'll get another observable which you can subscribe to.

You can recognize some of these from the Java 8 Stream API since most stream processing libraries have the basic functions that are very similar.

• map(Function<? super T,? extends R> mapper) - applies a function to each of the items, and emits the returned values.
• filter(Predicate<? super T> predicate) - emits only the items satisfying a predicate.
• buffer(int count) - emits lists of the items of the specified size.
• zip(ObservableSource s1, ObservableSource s2, BiFunction<T1, T2, R> f) - applies a function to the items from multiple observables and emits the returned value.
• flatMap(Function<? super T,? extends ObservableSource<? extends R>> mapper) - takes a function from items to an Observable, emits the items of the resulting Observables
• groupBy(Function<? super T,? extends K> keySelector) - emits items grouped by a specified key selector function
• timeout(long timeout, TimeUnit timeUnit) - emits items of the original observable. If the next item isn't emitted within the specified timeout, a TimeoutException occurs.

RxJava documentation comes with the marble diagrams -- pictures that illustrate what happens with the data when you apply a function. They look amazing and are quite intuitive. In the cheat sheet we illustrated the flatmap behavior with a marble diagram:

Testing Observables in RxJava

A great thing about RxJava is that it's production ready and developer friendly. For example they have a way to test the observables and the data transformation functions you use.

TestSubscriber is a subscriber that records events that you can make assertions upon.
TestObserver is an Observer that records events that you can make assertions upon.

Here's a basic example of using a TestSubscriber to check if the range function creates an observable that emits N items.

TestSubscriber ts = Flowable.range(1, 5).test();
// assert properties of the flow
assertThat(ts.values()).hasSize(5); 

Next, let's look at some of the libraries that can help you use RxJava on Android. One reason we're interested in using RxJava on Android is because ZeroTurnaround develops JRebel for Android, an Android Studio plugin to skip the build and install cycle when developing, and just reload your Android application code right in the device or meulator on the fly.

RxLifecycle

One problem with RxJava on the Android platform is the complicated lifecycle of the Android framework components. Activities and fragments can come and go as the framework pleases, and that means you always have to be sure you clean up all the resources that you attached to them. In the Rx world, it's a bit more complicated, since you don't build the graph of objects in the normal way, but you attach the data processing functions in the reverse order. So either you always cancel the subscriptions manually or use something like the RxLifecycle library.

RxLifecycle allows you to bind subscription lifecycles to the lifecycle of the Android components. It will automatically destroy the subscriptions and avoid memory leaks when the Android component is destroyed.

myObservable.compose(
  RxLifecycleAndroid.bindActivity(lifecycle)).subscribe();

Just compose your observable and call one static method from the library, it doesn't get much easier. There's also a more verbose API to manually specify when to cancel the subscription, but for most cases you're good with the default one.

RxBinding

One very Android specific use case for RxJava is managing UI events and actions with Observables. To simplify doing it, RxBinding offers a API to unify all the UI event handlers.

You can use the RxView class to turn UI events into RxJava observables:

Button button = (Button) findViewById(R.id.button);
RxView.clicks(button).subscribe(x -> {
  // do work here
});

And it'll work the same way for all the UI elements: buttons, text views, and so on.

RxAndroid

Another thing you should be careful of when developing Android apps is on which thread do you process the data. You cannot do any long running computations on the UI thread, as it'll render the UI unresponsive. So you want your observers to operate on background threads. To make it seamless, use RxAndoid library, with which you can easily control on which threads you subscribe to observables.

Observable.just("RebelLabs")   
  .subscribeOn(Schedulers.newThread()) 
  .observeOn(AndroidSchedulers.mainThread())
.subscribe(/* an Observer */);

In the snippet above we’re observing the events on the main Android looper thread, and the subscription actions will be executed on the new background thread. The library offers more customizations, but for the most part, the defaults will work nicely.

Whenever you use RxJava, you'll use Observables, Observers and transform the data, and this cheat sheet can be a great reminder of the most frequently used parts of Rx Java.

Additional Resources

Ever since Anton wrote a post about RxJava, I wanted to create an RxJava cheat sheet. With RxJava API, data processing functions, some lesser known tricks and utilities, we have the perfect mix for a great cheat sheet -- trust me, we've done a bunch of them. Here's a selection of our most popular ones:

Java Regular Expressions
Java 8 Best Practices
Java 8 Streams
Git Commands
SQL 

And many more. You can see all of them on our cheat sheets page.

Get The Cheat Sheet