Reactive Modelling on Android

This post was originally published on Toptal.

Concurrency and asynchronicity are inherent to mobile programming.

Dealing with concurrency through imperative-style programming, which is what programming on Android generally involves, can be the cause of many problems. Using Reactive Programming with RxJava, you can avoid potential concurrency problems by providing a cleaner and less error-prone solution.

Aside from simplifying concurrent, asynchronous tasks, RxJava also provides the ability to perform functional style operations that transform, combine, and aggregate emissions from an Observable until we achieve our desired result.

By combining RxJava’s reactive paradigm and functional style operations, we can model a wide range of concurrency constructs in a reactive way, even in Android’s non-reactive world. In this article, you will learn how you can do exactly that. You will also learn how to adopt RxJava into an existing project incrementally.

If you are new to RxJava, I recommend reading the post here which talks about some of the fundamentals of RxJava.

Bridging Non-Reactive into the Reactive World

One of the challenges of adding RxJava as one of the libraries to your project is that it fundamentally changes the way that you reason about your code.

RxJava requires you to think about data as being pushed rather than being pulled. While the concept itself is simple, changing a full codebase that is based on a pull paradigm can be a bit daunting. Although consistency is always ideal, you might not always have the privilege to make this transition throughout your entire code base all at once, so more of an incremental approach may be needed.

Consider the following code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14

/**
 * @return a list of users with blogs
 */
public List<User> getUsersWithBlogs() {
   final List<User> allUsers = UserCache.getAllUsers();
   final List<User> usersWithBlogs = new ArrayList<>();
   for (User user : allUsers) {
       if (user.blog != null && !user.blog.isEmpty()) {
           usersWithBlogs.add(user);
       }
   }
   Collections.sort(usersWithBlogs, (user1, user2) -> user1.name.compareTo(user2.name));
   return usersWithBlogs;
}

This function gets a list of User objects from the cache, filters each one based on whether or not the user has a blog, sorts them by the user’s name, and finally returns them to the caller. Looking at this snippet, we notice that many of these operations can take advantage of RxJava operators; e.g., filter() and sorted().

Rewriting this snippet then gives us:

1
2
3
4
5
6
7
8

/**
 * @return a list of users with blogs
 */
public Observable<User> getUsersWithBlogs() {
   return Observable.fromIterable(UserCache.getAllUsers())
                    .filter(user -> user.blog != null && !user.blog.isEmpty())
                    .sorted((user1, user2) -> user1.name.compareTo(user2.name));
}

The first line of the function converts the List<User> returned by UserCache.getAllUsers() to an Observable<User> via fromIterable(). This is the first step into making our code reactive. Now that we are operating on an Observable, this enables us to perform any Observable operator in the RxJava toolkit – filter() and sorted() in this case.

There are a few other points to note about this change.

First, the method signature is no longer the same. This may not be a huge deal if this method call is only used in a few places and it’s easy to propagate the changes up to other areas of the stack; however, if it breaks clients relying on this method, that is problematic and the method signature should be reverted.

Second, RxJava is designed with laziness in mind. That is, no long operations should be performed when there are no subscribers to the Observable. With this modification, that assumption is no longer true since UserCache.getAllUsers() is invoked even before there are any subscribers.

Leaving the Reactive World

To address the first issue from our change, we can make use of any of the blocking operators available to an Observable such as blockingFirst() and blockingNext(). Essentially, both of these operators will block until an item is emitted downstream: blockingFirst() will return the first element emitted and finish, whereas blockingNext() will return an Iterable which allows you to perform a for-each loop on the underlying data (each iteration through the loop will block).

A side-effect of using a blocking operation that is important to be aware of, though, is that exceptions are thrown on the calling thread rather than being passed to an observer’s onError() method.

Using a blocking operator to change the method signature back to a List<User>, our snippet would now look like this:

1
2
3
4
5
6
7
8
9
10

/**
 * @return a list of users with blogs
 */
public List<User> getUsersWithBlogs() {
   return Observable.fromIterable(UserCache.getAllUsers())
           .filter(user -> user.blog != null && !user.blog.isEmpty())
           .sorted((user1, user2) -> user1.name.compareTo(user2.name))
           .toList()
           .blockingGet();
}

Before calling a blocking operator (i.e. blockingGet()) we first need to chain the aggregate operator toList() so that the stream is modified from an Observable<User> to a Single<List<User>> (a Single is a special type of Observable that only emits a single value in onSuccess(), or an error via onError()).

Afterwards, we can call the blocking operator blockingGet() which unwraps the Single and returns a List<User>.

Although RxJava supports this, as much as possible this should be avoided as this is not idiomatic reactive programming. When absolutely necessary though, blocking operators are a nice initial way of stepping out of the reactive world.

The Lazy Approach

As mentioned earlier, RxJava was designed with laziness in mind. That is, long-running operations should be delayed as long as possible (i.e., until a subscribe is invoked on an Observable). To make our solution lazy, we make use of the defer() operator.

defer() takes in an ObservableSource factory which creates an Observable for each new observer that subscribes. In our case, we want to return Observable.fromIterable(UserCache.getAllUser()) whenever an observer subscribes.

1
2
3
4
5
6
7
8

/**
 * @return a list of users with blogs
 */
public Observable<User> getUsersWithBlogs() {
   return Observable.defer(() -> Observable.fromIterable(UserCache.getAllUsers()))
                    .filter(user -> user.blog != null && !user.blog.isEmpty())
                    .sorted((user1, user2) -> user1.name.compareTo(user2.name));
}

Now that the long running operation is wrapped in a defer(), we have full control as to what thread this should run in simply by specifying the appropriate Scheduler in subscribeOn(). With this change, our code is fully reactive and subscription should only occur at the moment the data is needed.

1
2
3
4
5
6
7
8
9

/**
 * @return a list of users with blogs
 */
public Observable<User> getUsersWithBlogs() {
   return Observable.defer(() -> Observable.fromIterable(UserCache.getAllUsers()))
                    .filter(user -> user.blog != null && !user.blog.isEmpty())
                    .sorted((user1, user2) -> user1.name.compareTo(user2.name))
                    .subscribeOn(Schedulers.io());
}

Another quite useful operator for deferring computation is the fromCallable() method. Unlike defer(), which expects an Observable to be returned in the lambda function and in turn “flattens” the returned Observable, fromCallable() will invoke the lambda and return the value downstream.

1
2
3
4
5
6
7
8
9

/**
 * @return a list of users with blogs
 */
public Observable<User> getUsersWithBlogs() {
   final Observable<List<User>> usersObservable = Observable.fromCallable(() -> UserCache.getAllUsers());
   final Observable<User> userObservable = usersObservable.flatMap(users -> Observable.fromIterable(users));
   return userObservable.filter(user -> user.blog != null && !user.blog.isEmpty())
                        .sorted((user1, user2) -> user1.name.compareTo(user2.name));
}

Single using fromCallable() on a list would now return an Observable<List<User>>, we need to flatten this list using flatMap().

Reactive-everything

From the previous examples, we have seen that we can wrap any object in an Observable and jump between non-reactive and reactive states using blocking operations and defer()/fromCallable(). Using these constructs, we can start converting areas of an Android app to be reactive.

Long Operations

A good place to initially think of using RxJava is whenever you have a process that takes a while to perform, such as network calls (check out previous post for examples), disk reads and writes, etc. The following example illustrates a simple function that will write text to the file system:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

/**
 * Writes {@code text} to the file system.
 *
 * @param context a Context
 * @param filename the name of the file
 * @param text the text to write
 * @return true if the text was successfully written, otherwise, false
 */
public boolean writeTextToFile(Context context, String filename, String text) {
   FileOutputStream outputStream;
   try {
       outputStream = context.openFileOutput(filename, Context.MODE_PRIVATE);
       outputStream.write(text.getBytes());
       outputStream.close();
       return true;
   } catch (Exception e) {
       e.printStackTrace();
       return false;
   }
}

When calling this function, we need to make sure that it is done on a separate thread since this operation is blocking. Imposing such a restriction on the caller complicates things for the developer which increases the likelihood of bugs and can potentially slow down development.

Adding a comment to the function will of course help avoid errors by the caller, but that is still far from bulletproof.

Using RxJava, however, we can easily wrap this into an Observable and specify the Scheduler that it should run on. This way, the caller doesn’t need to be concerned at all with invoking the function in a separate thread; the function will take care of this itself.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

/**
 * Writes {@code text} to the filesystem.
 *
 * @param context a Context
 * @param filename the name of the file
 * @param text the text to write
 * @return An Observable emitting a boolean indicating whether or not the text was successfully written.
 */
public Observable<Boolean> writeTextToFile(Context context, String filename, String text) {
   return Observable.fromCallable(() -> {
       FileOutputStream outputStream;
       outputStream = context.openFileOutput(filename, Context.MODE_PRIVATE);
       outputStream.write(text.getBytes());
       outputStream.close();
       return true;
   }).subscribeOn(Schedulers.io());
}

Using fromCallable(), writing the text to file is deferred up until subscription time.

Since exceptions are first-class objects in RxJava, one other benefit of our change is that the we no longer need to wrap the operation in a try/catch block. The exception will simply be propagated downstream rather than being swallowed. This allows the caller to handle the exception a he/she sees fit (e.g. show an error to the user depending on what exception was thrown, etc.).

One other optimization we can perform is to return a Completable rather than an Observable. A Completable is essentially a special type of Observable — similar to a Single — that simply indicates if a computation succeeded, via onComplete(), or failed, via onError(). Returning a Completable seems to make more sense in this case since it seems silly to return a single true in an Observable stream.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

/**
 * Writes {@code text} to the filesystem.
 *
 * @param context a context
 * @param filename the name of the file
 * @param text the text to write
 * @return A Completable
 */
public Completable writeTextToFile(Context context, String filename, String text) {
   return Completable.fromAction(() -> {
       FileOutputStream outputStream;
       outputStream = context.openFileOutput(filename, Context.MODE_PRIVATE);
       outputStream.write(text.getBytes());
       outputStream.close();
   }).subscribeOn(Schedulers.io());
}

To complete the operation, we use the fromAction() operation of a Completable since the return value is no longer of interest to us. If needed, like an Observable, a Completable also supports the fromCallable() and defer() functions.

Replacing Callbacks

So far, all the examples that we’ve looked at emit either one value (i.e., can be modelled as a Single), or tell us if an operation succeeded or failed (i.e., can be modelled as a Completable).

How might we convert areas in our app, though, that receive continuous updates or events (such as location updates, view click events, sensor events, and so on)?

We will look at two ways to do this, using create() and using Subjects.

create() allows us to explicitly invoke an observer’s onNext()|onComplete()|onError() method as we receive updates from our data source. To use create(), we pass in an ObservableOnSubscribe which receives an ObservableEmitter whenever an observer subscribes. Using the received emitter, we can then perform all the necessary set-up calls to start receiving updates and then invoke the appropriate Emitter event.

In the case of location updates, we can register to receive updates in this place and emit location updates as received.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

public class LocationManager {

   /**
    * Call to receive device location updates.
    * @return An Observable emitting location updates
    */
   public Observable<Location> observeLocation() {
       return Observable.create(emitter -> {
           // Make sure that the following conditions apply and if not, call the emitter's onError() method
           // (1) googleApiClient is connected
           // (2) location permission is granted
           final LocationRequest locationRequest = new LocationRequest();
           locationRequest.setInterval(1000);
           locationRequest.setPriority(LocationRequest.PRIORITY_HIGH_ACCURACY);

           LocationServices.FusedLocationApi.requestLocationUpdates(googleApiClient, locationRequest, new LocationListener() {
               @Override public void onLocationChanged(Location location) {
                   if (!emitter.isDisposed()) {
                       emitter.onNext(location);
                   }
               }
           });
       });
   }
}

The function inside the create() call requests location updates and passes in a callback that gets invoked when the device’s location changes. As we can see here, we essentially replace the callback-style interface and instead emit the received location in the created Observable stream (for the sake of educational purposes, I skipped some of the details with constructing a location request, if you want to delve deeper into the details you can read it here).

One other thing to note about create() is that, whenever subscribe() is called, a new emitter is provided. In other words, create() returns a cold Observable. This means that, in the function above, we would potentially be requesting location updates multiple times, which is not what we want.

To work around this, we want to change the the function to return a hot Observable with the help of Subjects.

Enter Subjects

A Subject extends an Observable and implements Observer at the same time. This is particularly useful whenever we want to emit or cast the same event to multiple subscribers at the same time. Implementation-wise, we would want to expose the Subject as an Observable to clients, while keeping it as a Subject for the provider.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

public class LocationManager {

   private Subject<Location> locationSubject = PublishSubject.create();
   
   /**
    * Invoke this method when this LocationManager should start listening to location updates.
    */
   public void connect() {
       final LocationRequest locationRequest = new LocationRequest();
       locationRequest.setInterval(1000);
       locationRequest.setPriority(LocationRequest.PRIORITY_HIGH_ACCURACY);

       LocationServices.FusedLocationApi.requestLocationUpdates(googleApiClient, locationRequest, new LocationListener() {
           @Override public void onLocationChanged(Location location) {
               locationSubject.onNext(location);
           }
       });
   }
   
   /**
    * Call to receive device location updates.
    * @return An Observable emitting location updates
    */
   public Observable<Location> observeLocation() {
       return locationSubject;
   }
}

In this new implementation, the subtype PublishSubject is used which emits events as they arrive starting from the time of subscription. Accordingly, if a subscription is performed at a point when location updates have already been emitted, past emissions will not be received by the observer, only subsequent ones. If this behavior is not desired, there are a couple of other Subject subtypes in the RxJava toolkit that can be used.

In addition, we also created a separate connect() function which starts the request to receive location updates. The observeLocation() can still do the connect() call, but we refactored it out of the function for clarity/simplicity.

Summary

We’ve looked at a number of mechanisms and techniques:

• defer() and its variants to delay execution of a computation until subscription
• cold Observables generated through create()
• hot Observables using Subjects
• blockingX() operations when we want to leave the reactive world

Hopefully, the examples provided in this article inspired some ideas regarding different areas in your app that can be converted to be reactive. We’ve covered a lot and if you have any questions, suggestions, or if anything is not clear, feel free to leave a comment below!

If you are interested in learning more about RxJava, I am working on an in-depth book that explains how to view problems the reactive way using Android examples. If you’d like to receive updates on it, please subscribe here.

Introduction to RxJava for Android

This post was originally published on Toptal under my Toptal account.

If you’re an Android developer, chances are you’ve heard of RxJava. It’s one of the most discussed libraries for enabling Reactive Programming in Android development. It’s touted as the go-to framework for simplifying concurrency/asynchronous tasks inherent in mobile programming.

But… what is RxJava and how does it “simplify” things?

While there are lots of resources already available online explaining what RxJava is, in this article my goal is to give you a basic introduction to RxJava and specifically how it fits into Android development. I’ll also give some concrete examples and suggestions on how you can integrate it in a new or existing project.

Why Consider RxJava

At its core, RxJava simplifies development because it raises the level of abstraction around threading. That is, as a developer you don’t have to worry too much about the details of how to perform operations that should occur on different threads. This is particularly attractive since threading is challenging to get right and, if not correctly implemented, can cause some of the most difficult bugs to debug and fix.

Granted, this doesn’t mean RxJava is bulletproof when it comes to threading and it is still important to understand what’s happening behind the scenes; however, RxJava can definitely make your life easier.

Let’s look at an example.

Network Call – RxJava vs AsyncTask

Say we want to obtain data over the network and update the UI as a result. One way to do this is to (1) create an inner AsyncTask subclass in our Activity/Fragment, (2) perform the network operation in the background, and (3) take the result of that operation and update the UI in the main thread.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

public class NetworkRequestTask extends AsyncTask<Void, Void, User> {

    private final int userId;

    public NetworkRequestTask(int userId) {
        this.userId = userId;
    }

    @Override protected User doInBackground(Void... params) {
        return networkService.getUser(userId);
    }

    @Override protected void onPostExecute(User user) {
        nameTextView.setText(user.getName());
        // ...set other views
    }
}
   
private void onButtonClicked(Button button) {
   new NetworkRequestTask(123).execute()
}

Harmless as this may seem, this approach has some issues and limitations. Namely, memory/context leaks are easily created since NetworkRequestTask is an inner class and thus holds an implicit reference to the outer class. Also, what if we want to chain another long operation after the network call? We’d have to nest two AsyncTasks which can significantly reduce readability.

In contrast, an RxJava approach to performing a network call might look something like this:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

private Subscription subscription;

private void onButtonClicked(Button button) {
   subscription = networkService.getObservableUser(123)
                 .subscribeOn(Schedulers.io())
                 .observeOn(AndroidSchedulers.mainThread())
                 .subscribe(new Action1<User>() {
                     @Override public void call(User user) {
                         nameTextView.setText(user.getName());
                         // ... set other views
                     }
                 });
}

@Override protected void onDestroy() {
   if (subscription != null && !subscription.isUnsubscribed()) {
       subscription.unsubscribe();
   }
   super.onDestroy();
}

Using this approach, we solve the problem (of potential memory leaks caused by a running thread holding a reference to the outer context) by keeping a reference to the returned Subscription object. This Subscription object is then tied to the Activity/Fragment object’s #onDestroy() method to guarantee that the Action1#call operation does not execute when the Activity/Fragment needs to be destroyed.

Also, notice that that the return type of #getObservableUser(...) (i.e. an Observable<User>) is chained with further calls to it. Through this fluid API, we’re able to solve the second issue of using an AsyncTask which is that it allows further network call/long operation chaining. Pretty neat, huh?

Let’s dive deeper into some RxJava concepts.

Observable, Observer, and Operator – The 3 O’s of RxJava Core

In the RxJava world, everything can be modeled as streams. A stream emits item(s) over time, and each emission can be consumed/observed.

If you think about it, a stream is not a new concept: click events can be a stream, location updates can be a stream, push notifications can be a stream, and so on.

The stream abstraction is implemented through 3 core constructs which I like to call “the 3 O’s”; namely: the Observable, Observer, and the Operator. The Observable emits items (the stream); and the Observer consumes those items. Emissions from Observable objects can further be modified, transformed, and manipulated by chaining Operator calls.

Observable

An Observable is the stream abstraction in RxJava. It is similar to an Iterator in that, given a sequence, it iterates through and produces those items in an orderly fashion. A consumer can then consume those items through the same interface, regardless of the underlying sequence.

Say we wanted to emit the numbers 1, 2, 3, in that order. To do so, we can use the Observable<T>#create(OnSubscribe<T>) method.

1
2
3
4
5
6
7
8

Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
   @Override public void call(Subscriber<? super Integer> subscriber) {
       subscriber.onNext(1);
       subscriber.onNext(2);
       subscriber.onNext(3);
       subscriber.onCompleted();
   }
});

Invoking subscriber.onNext(Integer) emits an item in the stream and, when the stream is finished emitting, subscriber.onCompleted() is then invoked.

This approach to creating an Observable is fairly verbose. For this reason, there are convenience methods for creating Observable instances which should be preferred in almost all cases.

The simplest way to create an Observable is using Observable#just(...). As the method name suggests, it just emits the item(s) that you pass into it as method arguments.

1

Observable.just(1, 2, 3); // 1, 2, 3 will be emitted, respectively

Observer

The next component to the Observable stream is the Observer (or Observers) subscribed to it. Observers are notified whenever something “interesting” happens in the stream. Observers are notified via the following events:

• Observer#onNext(T) – invoked when an item is emitted from the stream
• Observable#onError(Throwable) – invoked when an error has occurred within the stream
• Observable#onCompleted() – invoked when the stream is finished emitting items.

To subscribe to a stream, simply call Observable#subscribe(…) and pass in an Observer instance.

1
2
3
4
5
6
7
8
9
10
11
12
13
14

Observable<Integer> observable = Observable.just(1, 2, 3);
observable.subscribe(new Observer<Integer>() {
   @Override public void onCompleted() {
       Log.d("Test", "In onCompleted()");
   }

   @Override public void onError(Throwable e) {
       Log.d("Test", "In onError()");
   }

   @Override public void onNext(Integer integer) {
       Log.d("Test", "In onNext():" + integer);
   }
});

The above code will emit the following in Logcat:

1
2
3
4
5

In onNext(): 1
In onNext(): 2
In onNext(): 3
In onNext(): 4
In onCompleted()

There may also be some instances where we are no longer interested in the emissions of an Observable. This is particularly relevant in Android when, for example, an Activity/Fragment needs to be reclaimed in memory.

To stop observing items, we simply need to call Subscription#unsubscribe() on the returned Subscription object.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

Subscription subscription = someInfiniteObservable.subscribe(new Observer<Integer>() {
   @Override public void onCompleted() {
       // ...
   }

   @Override public void onError(Throwable e) {
       // ...
   }

   @Override public void onNext(Integer integer) {
       // ...
   }
});

// Call unsubscribe when appropriate
subscription.unsubscribe();

As seen in the code snippet above, upon subscribing to an Observable, we hold the reference to the returned Subscription object and later invoke subscription#unsubscribe() when necessary. In Android, this is best invoked within Activity#onDestroy() or Fragment#onDestroy().

Operator

Items emitted by an Observable can be transformed, modified, and filtered through Operators before notifying the subscribed Observer object(s). Some of the most common operations found in functional programming (such as map, filter, reduce, etc.) can also be applied to an Observable stream. Let’s look at map as an example:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

Observable.just(1, 2, 3, 4, 5).map(new Func1<Integer, Integer>() {
   @Override public Integer call(Integer integer) {
       return integer * 2;
   }
}).subscribe(new Observer<Integer>() {
   @Override public void onCompleted() {
       // ...
   }

   @Override public void onError(Throwable e) {
       // ...
   }

   @Override public void onNext(Integer integer) {
       // ...
   }
});

The code snippet above would take each emission from the Observable and multiply each by 2, producing the stream 2, 4, 6, 8, 10, respectively. Applying an Operator typically returns another Observable as a result, which is convenient as this allows us to chain multiple operations to obtain a desired result.

Given the stream above, say we wanted to only receive even numbers. This can be achieved by chaining a filter operation.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

Observable.just(1, 2, 3, 4, 5).map(new Func1<Integer, Integer>() {
   @Override public Integer call(Integer integer) {
       return integer * 2;
   }
}).filter(new Func1<Integer, Boolean>() {
   @Override public Boolean call(Integer integer) {
       return integer % 2 == 0;
   }
}).subscribe(new Observer<Integer>() {
   @Override public void onCompleted() {
       // ...
   }

   @Override public void onError(Throwable e) {
       // ...
   }

   @Override public void onNext(Integer integer) {
       // ...
   }
});

There are many operators built-in the RxJava toolset that modify the Observable stream; if you can think of a way to modify the stream, chances are, there’s an Operator for it. Unlike most technical documentation, reading the RxJava/ReactiveX docs is fairly simple and to-the-point. Each operator in the documentation comes along with a visualization on how the Operator affects the stream. These visualizations are called “marble diagrams.”

Here’s how a hypothetical Operator called flip might be modeled through a marble diagram:

Multithreading with RxJava

Controlling the thread within which operations occur in the Observable chain is done by specifying the Scheduler within which an operator should occur. Essentially, you can think of a Scheduler as a thread pool that, when specified, an operator will use and run on. By default, if no such Scheduler is provided, the Observable chain will operate on the same thread where Observable#subscribe(...) is called. Otherwise, a Scheduler can be specified via Observable#subscribeOn(Scheduler) and/or Observable#observeOn(Scheduler) wherein the scheduled operation will occur on a thread chosen by the Scheduler.

The key difference between the two methods is that Observable#subscribeOn(Scheduler) instructs the source Observable which Scheduler it should run on. The chain will continue to run on the thread from the Scheduler specified in Observable#subscribeOn(Scheduler) until a call to Observable#observeOn(Scheduler) is made with a different Scheduler. When such a call is made, all observers from there on out (i.e., subsequent operations down the chain) will receive notifications in a thread taken from the observeOn Scheduler.

Here’s a marble diagram that demonstrates how these methods affect where operations are run:

In the context of Android, if a UI operation needs to take place as a result of a long operation, we’d want that operation to take place on the UI thread. For this purpose, we can use AndroidScheduler#mainThread(), one of the Schedulers provided in the RxAndroid library.

RxJava on Android

Now that we’ve got some of the basics under our belt, you might be wondering — what’s the best way to integrate RxJava in an Android application? As you might imagine, there are many use cases for RxJava but, in this example, let’s take a look at one specific case: using Observable objects as part of the network stack.

In this example, we will look at Retrofit, an HTTP client open sourced by Square which has built-in bindings with RxJava to interact with GitHub’s API. Specifically, we’ll create a simple app that presents all the starred repositories for a user given a GitHub username. If you want to jump ahead, the source code is available here.

Create a New Android Project

• Start by creating a new Android project and naming it GitHubRxJava.

• In the Target Android Devices screen, keep Phone and Tablet selected and set the minimum SDK level of 17. Feel free to set it to a lower/higher API level but, for this example, API level 17 will suffice.

• Select Empty Activity in the next prompt.

• In the last step, keep the Activity Name as MainActivity and generate a layout file activity_main.

Project Set-Up

Include RxJava, RxAndroid, and the Retrofit library in app/build.gradle. Note that including RxAndroid implicitly also includes RxJava. It is best practice, however, to always include those two libraries explicitly since RxAndroid does not always contain the most up-to-date version of RxJava. Explicitly including the latest version of RxJava guarantees use of the most up-to-date version.

1
2
3
4
5
6
7
8

dependencies {
    compile 'com.squareup.retrofit2:adapter-rxjava:2.1.0'
    compile 'com.squareup.retrofit2:converter-gson:2.1.0'
    compile 'com.squareup.retrofit2:retrofit:2.1.0'
    compile 'io.reactivex:rxandroid:1.2.0'
    compile 'io.reactivex:rxjava:1.1.8'
    // ...other dependencies
}

Create Data Object

Create the GitHubRepo data object class. This class encapsulates a repository in GitHub (the network response contains more data but we’re only interested in a subset of that).

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

public class GitHubRepo {

    public final int id;
    public final String name;
    public final String htmlUrl;
    public final String description;
    public final String language;
    public final int stargazersCount;

    public GitHubRepo(int id, String name, String htmlUrl, String description, String language, int stargazersCount) {
        this.id = id;
        this.name = name;
        this.htmlUrl = htmlUrl;
        this.description = description;
        this.language = language;
        this.stargazersCount = stargazersCount;
    }
}

Set-Up Retrofit

• Create the GitHubService interface. We will pass this interface into Retrofit and Retrofit will create an implementation of GitHubService.

1
2
3

public interface GitHubService {
    @GET("users/{user}/starred") Observable<List<GitHubRepo>> getStarredRepositories(@Path("user") String userName);
}

• Create the GitHubClient class. This will be the object we will interact with to make network calls from the UI level.
  • When constructing an implementation of GitHubService through Retrofit, we need to pass in an RxJavaCallAdapterFactory as the call adapter so that network calls can return Observable objects (passing a call adapter is needed for any network call that returns a result other than a Call).
  • We also need to pass in a GsonConverterFactory so that we can use Gson as a way to marshal JSON objects to Java objects.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

public class GitHubClient {

    private static final String GITHUB_BASE_URL = "https://api.github.com/";

    private static GitHubClient instance;
    private GitHubService gitHubService;

    private GitHubClient() {
        final Gson gson =
            new GsonBuilder().setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES).create();
        final Retrofit retrofit = new Retrofit.Builder().baseUrl(GITHUB_BASE_URL)
                                                        .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                                                        .addConverterFactory(GsonConverterFactory.create(gson))
                                                        .build();
        gitHubService = retrofit.create(GitHubService.class);
    }

    public static GitHubClient getInstance() {
        if (instance == null) {
            instance = new GitHubClient();
        }
        return instance;
    }

    public Observable<List<GitHubRepo>> getStarredRepos(@NonNull String userName) {
        return gitHubService.getStarredRepositories(userName);
    }
}

Set-Up Layouts

Next, create a simple UI that displays the retrieved repos given an input GitHub username. Create activity_home.xml – the layout for our activity – with something like the following:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout
    xmlns:android="http://schemas.android.com/apk/res/android"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    android:orientation="vertical">

    <ListView
        android:id="@+id/list_view_repos"
        android:layout_width="match_parent"
        android:layout_height="0dp"
        android:layout_weight="1"/>

    <LinearLayout
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:orientation="horizontal">

        <EditText
            android:id="@+id/edit_text_username"
            android:layout_width="0dp"
            android:layout_height="wrap_content"
            android:layout_weight="1"
            android:hint="@string/username"/>

        <Button
            android:id="@+id/button_search"
            android:layout_width="wrap_content"
            android:layout_height="wrap_content"
            android:text="@string/search"/>

    </LinearLayout>

</LinearLayout>

Create item_github_repo.xml – the ListView item layout for GitHub repository object – with something like the following:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

<?xml version="1.0" encoding="utf-8"?>
<RelativeLayout
    xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:tools="http://schemas.android.com/tools"
    android:orientation="vertical"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    android:padding="6dp">

    <TextView
        android:id="@+id/text_repo_name"
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:textSize="24sp"
        android:textStyle="bold"
        tools:text="Cropper"/>

    <TextView
        android:id="@+id/text_repo_description"
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:lines="2"
        android:ellipsize="end"
        android:textSize="16sp"
        android:layout_below="@+id/text_repo_name"
        tools:text="Android widget for cropping and rotating an image."/>

    <TextView
        android:id="@+id/text_language"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_below="@+id/text_repo_description"
        android:layout_alignParentLeft="true"
        android:textColor="?attr/colorPrimary"
        android:textSize="14sp"
        android:textStyle="bold"
        tools:text="Language: Java"/>

    <TextView
        android:id="@+id/text_stars"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_below="@+id/text_repo_description"
        android:layout_alignParentRight="true"
        android:textColor="?attr/colorAccent"
        android:textSize="14sp"
        android:textStyle="bold"
        tools:text="Stars: 1953"/>

</RelativeLayout>

Glue Everything Together

Create a ListAdapter that is in charge of binding GitHubRepo objects into ListView items. The process essentially involves inflating item_github_repo.xml into a View if no recycled View is provided; otherwise, a recycled View is reused to prevent overinflating too many View objects.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

public class GitHubRepoAdapter extends BaseAdapter {

    private List<GitHubRepo> gitHubRepos = new ArrayList<>();

    @Override public int getCount() {
        return gitHubRepos.size();
    }

    @Override public GitHubRepo getItem(int position) {
        if (position < 0 || position >= gitHubRepos.size()) {
            return null;
        } else {
            return gitHubRepos.get(position);
        }
    }

    @Override public long getItemId(int position) {
        return position;
    }

    @Override public View getView(int position, View convertView, ViewGroup parent) {
        final View view = (convertView != null ? convertView : createView(parent));
        final GitHubRepoViewHolder viewHolder = (GitHubRepoViewHolder) view.getTag();
        viewHolder.setGitHubRepo(getItem(position));
        return view;
    }

    public void setGitHubRepos(@Nullable List<GitHubRepo> repos) {
        if (repos == null) {
            return;
        }
        gitHubRepos.clear();
        gitHubRepos.addAll(repos);
        notifyDataSetChanged();
    }

    private View createView(ViewGroup parent) {
        final LayoutInflater inflater = LayoutInflater.from(parent.getContext());
        final View view = inflater.inflate(R.layout.item_github_repo, parent, false);
        final GitHubRepoViewHolder viewHolder = new GitHubRepoViewHolder(view);
        view.setTag(viewHolder);
        return view;
    }

    private static class GitHubRepoViewHolder {

        private TextView textRepoName;
        private TextView textRepoDescription;
        private TextView textLanguage;
        private TextView textStars;

        public GitHubRepoViewHolder(View view) {
            textRepoName = (TextView) view.findViewById(R.id.text_repo_name);
            textRepoDescription = (TextView) view.findViewById(R.id.text_repo_description);
            textLanguage = (TextView) view.findViewById(R.id.text_language);
            textStars = (TextView) view.findViewById(R.id.text_stars);
        }

        public void setGitHubRepo(GitHubRepo gitHubRepo) {
            textRepoName.setText(gitHubRepo.name);
            textRepoDescription.setText(gitHubRepo.description);
            textLanguage.setText("Language: " + gitHubRepo.language);
            textStars.setText("Stars: " + gitHubRepo.stargazersCount);
        }
    }
}

Glue everything together in MainActivity. This is essentially the Activity that gets displayed when we first launch the app. In here, we ask the user to enter their GitHub username, and finally, display all the starred repositories by that username.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

public class MainActivity extends AppCompatActivity {

    private static final String TAG = MainActivity.class.getSimpleName();
    private GitHubRepoAdapter adapter = new GitHubRepoAdapter();
    private Subscription subscription;

    @Override protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        final ListView listView = (ListView) findViewById(R.id.list_view_repos);
        listView.setAdapter(adapter);

        final EditText editTextUsername = (EditText) findViewById(R.id.edit_text_username);
        final Button buttonSearch = (Button) findViewById(R.id.button_search);
        buttonSearch.setOnClickListener(new View.OnClickListener() {
            @Override public void onClick(View v) {
                final String username = editTextUsername.getText().toString();
                if (!TextUtils.isEmpty(username)) {
                    getStarredRepos(username);
                }
            }
        });
    }

    @Override protected void onDestroy() {
        if (subscription != null && !subscription.isUnsubscribed()) {
            subscription.unsubscribe();
        }
        super.onDestroy();
    }

    private void getStarredRepos(String username) {
        subscription = GitHubClient.getInstance()
                                   .getStarredRepos(username)
                                   .subscribeOn(Schedulers.io())
                                   .observeOn(AndroidSchedulers.mainThread())
                                   .subscribe(new Observer<List<GitHubRepo>>() {
                                       @Override public void onCompleted() {
                                           Log.d(TAG, "In onCompleted()");
                                       }

                                       @Override public void onError(Throwable e) {
                                           e.printStackTrace();
                                           Log.d(TAG, "In onError()");
                                       }

                                       @Override public void onNext(List<GitHubRepo> gitHubRepos) {
                                           Log.d(TAG, "In onNext()");
                                           adapter.setGitHubRepos(gitHubRepos);
                                       }
                                   });
    }
}

Run the App

Running the app should present a screen with an input box to enter a GitHub username. Searching should then present the list of all starred repos.

Conclusion

I hope this serves as a useful introduction to RxJava and an overview of its basic capabilities. There are a ton of powerful concepts in RxJava and I urge you to explore them by digging more deeply into the well-documented RxJava wiki.

Feel free to leave any questions or comments in the comment box below. You can also follow me on Twitter at @arriolachris where I tweet a lot about RxJava and all things Android.

Getting Attached To Your State

Configuration changes occur in runtime and are caused by various events such as: when keyboard visibility changes, when language changes, or when orientation changes. This in turn causes any visible Activity to be reconstructed once the change finishes. For state to be recovered, it needs to be explicitly tied to Android’s parcelling mechanism via Bundle.

Saving State, The Wrong Way

Upon encountering state loss caused by a configuration change, say on an orientation change, 2 poor solutions are:

1. Locking the orientation mode

1
2
3

<activity 
    android:name="com.app.app.MainActivity"
    android:screenOrientation="portrait" />

This is an anti-solution and simply prevents the configuration change from occurring. This will not guard against the other 13 configuration changes as of API level 23.

Orientation should only be locked as a result of a UI/UX decision, and not for retaining state.

2. Handling configuration changes manually

1
2
3

<activity 
    android:name="com.app.app.MainActivity"
    android:configChanges="orientation" />

Again, this is a band-aid and not a proper solution. What if new configuration changes are introduced? Also this might cause some unintended consequences. Say for example you want to declare a landscape/portrait-specific resource, that resource will not be loaded automatically anymore and you need to explicitly load and apply it in Activity#onConfigurationChanged() instead. Not knowing this consequence may be a pain to debug.

Quoting Roman Guy:

“…it is sometimes confusing for new Android developers who wonder why their activity is destroyed and recreated. Facing this “issue,” some developers choose to handle configuration changes themselves which is, in my opinion, a short-term solution that will complicate their life when other devices come out or when the application becomes more complex. The automatic resource handling is a very efficient and easy way to adapt your application’s user interface to various devices and devices configurations.”

Indeed, it is good practice to allow the system to do what it was designed to do as it allows your application to behave correctly on varying devices especially as your application gets more complex.

Manually handling configuration changes should only be done sparingly and as a result of some constraint (e.g. performance reasons).

Saving State, The Right Way

Luckily, framework Views will automatically be saved and recovered for you—assuming all your Views have unique IDs. In other instances you need to explicitly retain state.

Retaining State In An Activity

1
2
3
4
5
6
7
8
9
10
11
12
13

@Override
public void onSaveInstanceState(Bundle savedInstanceState) {
    // This is called by the system so that any instance that can be recovered
    // when the Activity is recreated.
    savedInstanceState.putInt(SOME_VALUE, someIntValue);
    super.onSaveInstanceState(savedInstanceState);
}

public void onRestoreInstanceState(Bundle savedInstanceState) {
    // This is called by the system when the Activity is reconstructed.
    super.onRestoreInstanceState(savedInstanceState);
    someIntValue = savedInstanceState.getInt(SOME_VALUE);
}

There are other lifecycle events where state can be restored such as in Activity#onCreate(). Retaining state in a Fragment can be done in a similar way via Fragment#onSaveInstanceState(). Here’s a good resource by CodePath for learning more about how this works.

Retaining State in A Custom View

Retaining state in a custom View is a bit more involved but nonetheless also fairly straightforward.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

package com.operator.android;

import android.content.Context;
import android.os.Parcel;
import android.os.Parcelable;
import android.view.View;

/**
 * A custom {@link View} that demonstrates how to save/restore instance state.
 */
public class CustomView extends View {

    private boolean someState;

    public CustomView(Context context) {
        super(context);
    }

    public boolean isSomeState() {
        return someState;
    }

    public void setSomeState(boolean someState) {
        this.someState = someState;
    }

    @Override protected Parcelable onSaveInstanceState() {
        final Parcelable superState = super.onSaveInstanceState();
        final CustomViewSavedState customViewSavedState = new CustomViewSavedState(superState);
        customViewSavedState.someState = this.someState;
        return customViewSavedState;
    }

    @Override protected void onRestoreInstanceState(Parcelable state) {
        final CustomViewSavedState customViewSavedState = (CustomViewSavedState) state;
        setSomeState(customViewSavedState.someState);
        super.onRestoreInstanceState(customViewSavedState.getSuperState());
    }

    private static class CustomViewSavedState extends BaseSavedState {

        boolean someState;

        public static final Parcelable.Creator<CustomViewSavedState> CREATOR = new Creator<CustomViewSavedState>() {
            @Override public CustomViewSavedState createFromParcel(Parcel source) {
                return new CustomViewSavedState(source);
            }

            @Override public CustomViewSavedState[] newArray(int size) {
                return new CustomViewSavedState[size];
            }
        };

        public CustomViewSavedState(Parcelable superState) {
            super(superState);
        }

        private CustomViewSavedState(Parcel source) {
            super(source);
            someState = source.readInt() == 1;
        }

        @Override public void writeToParcel(Parcel out, int flags) {
            super.writeToParcel(out, flags);
            out.writeInt(someState ? 1 : 0);
        }
    }

}

TL;DR

• Don’t save your state the wrong way 😅 Save your state the right way 😁