6. RxJava

Written by Aldo Olivares

In this chapter, you won’t be writing any code. Instead, you’ll learn the theory behind one of the most popular open source libraries for Java: RxJava.

You may be wondering: Why learn about RxJava if this is a book written in Kotlin?

Even though RxJava is a Java library, it’ll work with your Kotlin code. This is because Kotlin was designed with Java interoperability built in.

You might also be wondering: I thought this was a book about architecture patterns!

True, this is a book about architecture patterns for Android, but RxJava makes use of important elements that are present amongst most of the architecture patterns you’ll see in this book, such as Observables and Reactive Programming. Also, RxJava is popular, and a lot of source code that you’ll encounter online use RxJava. So, it’s important that you understand how this library works, at least at a high level.

What is the Observer pattern?

Before you dive into the RxJava world, it’s critical that you understand the Observer pattern. When most Android developers hear the word pattern, they immediately think about MVC, MVP and MVVM. In reality, those are not technically patterns; they’re compound patterns.

What is a compound pattern?

A compound pattern is a pattern made up of multiple patterns. In other words, a pattern of patterns.

MVC, for example, relies on the Strategy pattern to configure the interaction between the Views and the Controller. The controller provides a strategy for the View, and the View is only concerned about displaying the UI. The View also relies on the Composite pattern under the hood to manage all the UI widgets that you see on the screen, such as buttons or lists. The models rely on the Observer pattern to make sure the controllers and Views have the latest data updates.

How does the Observer pattern work?

The Observer pattern helps your objects know when something happens — such as a database update or a network response — so they can react accordingly.

To understand the Observer pattern, imagine how a subscription to your favorite YouTube channel works:

• A content creator publishes a new YouTube video every other day.
• You like the channel’s content, so you subscribe to the channel for new video notifications.
• As soon as a new video is published, you get a notification. You’ll continue to receive notifications as long as you’re a subscriber. But that doesn’t mean you have to watch the video.
• If you unsubscribe, you no longer receive notifications from that channel.
• As long as the user keeps posting videos, all subscribers to the channel will get a notification when a new video is posted.

The Observer pattern works similarly to the YouTube channel subscription described above. The analogy is that the YouTube channel is the Observable and the subscribers as the Observers. When the state of the Observable changes, all of the Observers who subscribed to it get a notification about the update, creating a one-to-many relationship between them.

Now, suppose you want to display a list of items as soon as the user presses a button in your app. You can define your button as the Observable and your list as the Observer. As soon as the button changes its state, it emits an event to the Observers. In this case, the list is automatically updated to display a list of items.

In Android, there are many ways to implement the Observer pattern in your apps, one of which is RxJava.

Getting to know RxJava

RxJava is an open-source library for Java and Android that helps you create reactive code. It offers the possibility to implement the Observer pattern for Observable/Observer callbacks and gives you a range of operators that allow you to handle asynchronous and event-based programs.

Programming reactively

RxJava helps you create reactive code. In imperative programming, you often evaluate an expression linearly or line-by-line. If you want to calculate the area of a rectangle, your code would look like this:

var width = 2
var height = 3
var area = width * height // z is 6

In this code example, the area is calculated once at runtime and never changes.

Reactive programming is a different paradigm in which your code dynamically reacts to changes. The changes can be almost anything you want, such as value changes or state changes.

The truth is, you might have done a bit of reactive programming in the past without even noticing it. Every time you add an onClickListener to one of your buttons, for example, you did something like this:

button.setOnClickListener {
  //Some code
}

In this case, you’re reacting to a change in the button state. Once the state changes, you can react to it by updating a list, displaying a notification or adding any action to it.

Observing events

RxJava implements the Observer pattern via two main interfaces: Observable and Observer. One of the most important methods in the Observable interface is subscribe().

The Observer interface, on the other hand, has three methods that the Observable calls when it changes the appropriate state:

• onNext(T value): This gets called by the Observable when it emits a new item of type T to the Observer.
• onComplete(): This notifies all of the Observers that the Observable is done with its task.
• onError(Throwable e): This notifies the Observer that the Observable has experienced an error.

As a rule of thumb, every Observable may emit one or more items that can be followed by a completion or an error.

The diagram above represents an event emitted by an Observable. The green dot represents an item that was emitted by the Observable (such as a new video notification), and the vertical black line represents a completion or an error.

However, there can be continuous events with no completion or errors. Here’s how an event caused by a mouse Observable looks:

As you can see, there are many events on this diagram that can represent mouse movements: right-clicks, left-clicks, middle-button clicks and much more.

But keep in mind: Events can’t be emitted after an Observable completes its tasks. Here’s an example of an Observable that’s not following the appropriate flow:

This flow violates the Observable contract by emitting an event after signaling completion.

Creating an Observable

There are many libraries in Java that help you create an Observable from almost any type of event. But sometimes it’s better to create your own. You can create an Observable using Observable.create(). Here’s its signature:

Observable<T> create(ObservableOnSubscribe<T> source)

That’s nice and concise, but what does it mean? What is the source? To understand that signature, you need to know what about ObservableOnSubscribe. ObservableOnSubscribe an interface, with this contract:

public interface ObservableOnSubscribe<T> {
  void subscribe(ObservableEmitter<T> e) throws Exception;
}

RxJava’s Emitter interface is similar to the Observer:

public interface Emitter<T> {
  void onNext(T value);
  void onError(Throwable error);
  void onComplete();
}

An ObservableEmitter also provides a means to cancel the subscription.

The best way to understand Observables is with simple examples that illustrate the entire process. Once you understand the basics, the complicated stuff becomes a little easier. It’s kind of like solving a jigsaw puzzle: Once you have the corners, everything else fits into place.

To create a simple method that returns an Observable, you can use this code:

//1
fun createYoutuber(): Observable<String>{
  //2  
  return Observable.create{emitter ->
    //3                       
    emitter.onNext("How to breed llamas")
  }
}

The Observable emits a string (here, new video title) for its subscribers. Here’s a closer look:

1. Declare a method that returns an Observable of type String. Since Observables are generic, you can create an Observable for almost anything you want such as Strings, Doubles or Network Responses.
2. Use create() to create a new Observable.
3. Make the Observable (The Youtuber) emit the new video title to its Observers (The subscribers).

The following example declares a method that creates a new Observer object:

//1
private fun createSubscriber(): Observer<String> {
  //2
  return object : Observer<String> {
    //3
    override fun onSubscribe(d: Disposable) {
      Log.d(TAG, " Im subscribed")
    }
     //4
    override fun onNext(value: String) {
      Log.d(TAG, "New Video : $value")
    }
    //5
    override fun onError(e: Throwable) {
      //Define onError() Action
    }
    //6
    override fun onComplete() {
      //Define onComplete() Action
    }
  }
}

Taking each commented section in turn

1. This method returns an Observer of type String. Just like the Observables, the Observers can observe any type of values.
2. Create a new Observer of type String.
3. onSubscribe() declares the action to take when your Observer is attached to the Observable.
4. onNext() declares the action to take when the emitter emits a new value. The parameter represents the value emitted by your Observable.
5. onError() declares the action to take when the Observable emits an error.
6. onComplete() declares the action to take when the Observable completes its tasks.

With these methods, you can easily create a new Observable and subscribe an Observer, like so:

createYoutuber().subscribe(createSubscriber())

When you execute that code, you’ll get these logs in the console:

D/MainActivity: Im subscribed
D/MainActivity: New Video : How to breed llamas

If you want to emit some values to your Observers, there’s an easier way to create your Observables using just().

Here’s how you can rewrite createYoutuber() to create an Observable that emits a single value to its observers:

fun createYoutuber(): Observable<String>{
  return Observable.just("How to breed Llamas")
}

The resulting log is the same:

D/MainActivity: Im subscribed
D/MainActivity: New Video : How to breed llamas

Asynchronous tasks

One common misconception about RxJava is that the tasks executed with this library are executed on a background thread. By default, RxJava does all of the tasks in the same thread from which it was called. For an Android app, this means that an Observable will usually emit all its data using the UI thread unless told otherwise.

You can, however, execute code on a background thread using observeOn() and subscribeOn(). observeOn() subscribes Observers to their Observable on the specified scheduler. subscribeOn() modifies the Observable to emit its events and notifications on the specified scheduler.

Suppose you want the Observers from previous examples to subscribe to their Observable on a background thread, but you still want to emit notifications on the main thread. This is how you can rewrite the subscribe/observe model to do it:

createYoutuber()
    // Subscribe on a background thread
    .subscribeOn(Schedulers.io())
    // Observe on the main thread
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(createSubscriber())

The resulting output:

D/MainActivity: Im subscribed
D/MainActivity: New Video : How to breed llamas

Although the result is the same for this example, having the ability to execute different tasks on different threads is useful — especially when you’re dealing with background operations such as database updates and network calls.

Operators

YouTube is great, but there are other video streaming services available. Perhaps you’d like to subscribe to more than just a YouTube Observable, but apply the same logic to each.

You can create a method that returns a new Netflix Observable like this:

fun createNetflixChannel(): Observable<String> {
  return Observable.just("House of Cards")
}

Then, you can subscribe to both the Youtube Observable and the Netflix Observable, like so:

val subscriber = createSubscriber()
createYoutuber()
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(subscriber)

createNetflixChannel()
   .subscribeOn(Schedulers.io())
   .observeOn(AndroidSchedulers.mainThread())
   .subscribe(subscriber)

That’s a lot of duplicated code! Fortunately, there’s an easier way to do this without duplicating the code. You can merge your Observables into a single Observable using the merge operator.

Operators in RxJava are a way to modify your Observables and/or their data to facilitate your development workflow without the need to create your own methods. Most operators operate on an Observable and return an Observable, making it possible to chain them one after the other.

You’ve already learned about two operators: SubscribeOn and ObserveOn. Both of them are utility operators that operate on an Observable and return an Observable — that’s why you were able to chain them in the previous examples.

On the other hand, the merge operator lets you merge two different Observables into one by using the merge() method from RxJava. This is especially useful if you want to apply the same operations to both of them.

You can rewrite the above code using the merge() method like this:

Observable.merge(createYoutuber(), createNetflixChannel())
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(createSubscriber())

This is much simpler and easier to read. Here, you’re creating a Youtuber Observable, a Netflix Observable and merging them into a new Observable using merge().

After that, you can apply the necessary operations to the new Observable just like you would to each of them separately.

The result is the following:

D/MainActivity:  Im subscribed
D/MainActivity:  Im subscribed
D/MainActivity: New Video : How to breed Llamas
D/MainActivity: New Video : House of Cards

Awesome! You just made your code easier to understand.

There are many other useful operators at your disposal in the RxJava library. They’re not all covered in this chapter because there are too many. If you’re interested in learnig more, check out the official documentation: http://reactivex.io/documentation/operators.html.

Frequently Not Asked RxJava Questions

Q. Can I use the usual listeners/callbacks instead of RxJava?

You certainly can! But using RxJava offers several benefits that can heavily pay off in the long run by reducing the amount of boilerplate code and making it more concise. You can even apply several operations to your Observables to reduce duplicate code.

Q. Are you sure that RxJava is 100% compatible with my Kotlin code?

Yes. No need to worry about compatibility issues, Kotlin was designed with Java interoperability in mind, and the RxJava library is no exception.

Q. I just read about RxKotlin, what is that?

RxKotlin is a lightweight library that takes advantage of Kotlin’s extension functions to make RxJava code more Kotlin-y. Of course, you can use RxJava with Kotlin out-of-the-box, but this library makes things even easier.

Key points

• The Observer pattern helps your objects know when something interesting happens so they can react accordingly.
• The Observer pattern defines a one to many relationship between an Observable and the Observers.
• An Observable is an object that emits notifications about its state to one or more Observers.
• Reactive Programming is a declarative programming paradigm in which you dynamically react to changes.
• RxJava is an open source library for Java and Android that helps you create reactive code. It’s heavily inspired by functional programming.
• RxJava implements the Observer pattern with two main interfaces: Observable and Observer.
• RxJava executes all tasks on the same thread from which it was called.

Where to go from here?

RxJava is a powerful and popular library that lets you write reactive code in your Android Projects. Even though it was written in Java, it’s 100% interoperable with Kotlin. If you want to learn more about RxJava, check out Irina Galata’s on the site Reactive Programming with RxAndroid in Kotlin raywenderlich.com/384-reactive-programming-with-rxandroid-in-kotlin-an-introduction