21. RxJava & Jetpack

Written by Alex Sullivan

Android Jetpack is a suite of libraries provided by the Android team to make developing Android apps a breeze (well, maybe not quite a breeze…). You’ve already been working with two of the libraries provided as part of Jetpack throughout the book: LiveData and ViewModel. In this chapter, you’re going to explore two more libraries that every Android developer should know about, and how they interact with RxJava.

The first library you’re going to utilize is the Room database library. Interacting with a database has typically been a painful process when writing an Android app. In the beginning, a developer would usually use a custom instance of the SQLiteOpenHelper class to manually create tables and run updates using SQL.

This approach worked, but came with a lot of downsides. It was cumbersome to keep all of the SQL statements you were writing in code and it was very easy to have the objects you were trying to store in the database and the tables representing those objects get out of sync. To top it all off, you needed a lot of boilerplate to turn those objects into ContentValues to then be inserted into the database. Luckily, Room provides an easy to use abstraction on top of SQLiteOpenHlper that makes storing data a much simpler task.

The second library you’re going to explore is the Paging library. Another common task for app developers is to implement a kind of infinitely scrollable list, like Instagram or Facebook has. The Paging library provides simple hooks for you to use to load new data as a user scrolls down in a list. It even ties together with Room to give you an easy way to pipe data from your database into your app.

Best of all, both Room and the Paging library come with first-class RxJava integrations!

In this chapter, you’ll explore both libraries by creating a Lord of the Rings-based book collector app, which allows a user to fetch a list of books from the Open Library API, scroll through the books, favorite some of them and mark others as read.

Getting started

Open the starter project in Android Studio and run the app. You should see the following screen:

The BookCollector app displays a list of books fetched from the Open Library API. A user can then either favorite the book by clicking the star icon, or mark it as a book they’ve already read by clicking the envelope icon.

There are three pages in the app. The first page is the screen you see in the screenshot above and the starting screen for the app, which displays the entire list of books. The second page is a favorites page, which displays the books the user has favorited. The third page displays all of the books the user has marked as read.

Each page is controlled by a different Fragment in a ViewPager. Each fragment is backed by the same view model, which is called MainViewModel. Open the MainViewModel class now and take a look around.

The first thing you’ll notice is that this view model follows a familiar pattern: There are three LiveData objects that govern what’s shown on an individual page. Then, in the init block, the view model queries the Open Library API and uses the cache operator to cache the result. Then, the view model subscribes to the resulting Observable three times, once for each live data object, filtering and mapping the results according to what that live data should emit.

There are also two stubbed out methods:

fun favoriteClicked(book: Book) {
  TODO()
}

fun readClicked(book: Book) {
  TODO()
}

You’ll update these two methods governing what happens when a user clicks the favorite icon and the read icon later on in the chapter. Before you go any further, it’s a good idea to get a quick refresher on how Room works.

There are three core components to Room:

1. The Entity: An Entity is a model object annotated with the @Entity annotation, and represents the data that will reside in the database. Room will typically create a table under the hood for each class marked with the @Entity annotation.
2. The Dao: A Dao is an interface marked with the @Dao annotation. This interface typically exposes high-level methods to insert and query items from the database. You can think of the Dao as being akin to a Retrofit interface.
3. The Database: The Database class is a class that you create that extends the RoomDatabase object, and it is annotated with an @Database annotation, wherein you list all of your Entities and expose the version of the database.

Open Book.kt to see an example of an Entity:

@Entity
data class Book(@PrimaryKey val title: String,
                val authorName: String,
                val publisher: String,
                val subject: String,
                val isFavorited: Boolean = false,
                val isAlreadyRead: Boolean = false)

The Book class is marked with the @Entity annotation to signify that it can be inserted and retrieved from a Room database.

Each Entity needs an instance variable marked with the @PrimaryKey annotation. The @PrimaryKey annotation signifies to Room that this instance variable can determine uniqueness for an object. That means that, in the example above, you could never have two Books with the same title in a Room database, since that would violate a primary key constraint on uniqueness.

Now, open BookDao.kt to see an example of a Dao:

@Dao
interface BookDao

As you can see, the BookDao class is empty. For now. :]

Last but not least, open BookDatabase.kt to see an example Database:

@Database(entities = [Book::class], version = 1)
abstract class BookDatabase : RoomDatabase() {
  abstract fun bookDao(): BookDao
}

It outlines the entities that will live in the database and the version of the database. For this app, you’ll only have one object residing in the database: the Book class. Your only exposed dao will be the BookDao.

RxJava and Room

Now that you’re familiar with Room, it’s time to sprinkle some Rx goodness on top of it.

The team behind Room has exposed a very helpful extension that allows you to utilize RxJava reactive types in a similar manner to the Retrofit library.

Add the following dependency to your build.gradle file:

implementation "androidx.room:room-rxjava2:$room_version"

Adding this extension will allow you to return any reactive type you want in your Dao object. Unfortunately the Room library has yet to update to using RxJava3, so you’ll have to use the bridge library you learned about in previous chapters to transition between the types returned by the RxJava2 and 3 libraries.

Database philosophy

Before you start getting your hands dirty, take a minute to discuss what the strategy is going to be moving forward for dealing with the network and the database.

Sometimes, utilizing both a network and a database as a source of information can be a frustrating experience. If one gets out of sync with the other, it can be confusing trying to reconcile which you should believe.

To attempt to mitigate the above issue, you’re going to be using the database as the primary source of truth in the app. The goal for the BookCollector app is going to be to pull data from the network when the app is started and then immediately insert it into the database. Once the database has been updated with new content you’ll use it to populate the information that the user sees.

By utilizing this database first philosophy, you’ll be able to sidestep the chaos of choosing who to believe and you’ll be able to implement a more reactive pattern of interacting with your data layer.

Now that you’ve got the philosophy down, it’s time to get this project started!

Open BookDao.kt and add the following method to the interface importing the io.reactivex.* version of Completable:

@Insert(onConflict = OnConflictStrategy.IGNORE)
fun insertBooks(books: List<Book>): Completable

The insertBooks method will insert a List<Book> into the database. It will ignore any conflicts when the books are inserted. Since all you care about is a confirmation that the items were added into the database, it makes sense to use Completable as the return type of insertBooks.

If you were instead interested in a list of IDs for the newly inserted books you could’ve specified a return type of Single<List<Long>> instead of Completable. Just like the Retrofit library, Room will look at the return type you provide and adjust the behavior of the library accordingly. You’ll see several examples, later on, utilizing the different return types in the BookDao.

What good is inserting items into a database if you can’t retrieve them? Add the following method below insertBooks importing the rxjava2 version of Observable:

@Query("SELECT * from book ORDER BY title")
fun bookStream(): Observable<List<Book>>

The bookStream method will pull all of the books out of the database and return them as an Observable<List<Book>>. Hold onto your seat, because this is where things start to get interesting.

If you specify a return type of Observable or Flowable for your Room queries, the Observable or Flowable you get back will emit a new list of books every time you insert or update a book. That means that, by subscribing to your Observable, you’ll get constant updates as the database changes. That allows you to keep your UI perfectly up-to-date and enables that reactive flow you saw outlined earlier.

The reactive return type you specify in your DAO methods have different meanings depending on what type of operation you’re executing. Here’s a quick breakdown of what the different types mean if you’re inserting an item, updating or deleting an item, or retrieving an item from the database.

Inserting an item

If you’re inserting an item into the database, you can use the following return types:

1. Completable: If your database insertion was successful, your completable will complete as expected. If it wasn’t successful, an error will be emitted that will filter into your onError block of your subscription.
2. Single or Maybe: Single and Maybe work the same when inserting with Room. If the insertion was successful, onSuccess will be emitted with a Long value representing the ID of the newly inserted object. If you’re inserting a list or array of items, you can instead specify the return type as Single<List<Long>>, in which each Long in the list represents the id of the correspondingly inserted item in the list.

That’s it — you can’t declare a return type of Observable or Flowable, since an insertion into a database will only ever return up to one value – the ID of the object that was inserted. Specifying a return type of Observable would be confusing in this scenario since you could use Single or Maybe, which better describe the situation.

Updating and deleting an item

If you’re updating or deleting an item in the database, you can use the same return types as inserting but with a slightly different meaning:

1. Completable: Just like before, you can use a Completable if all you care about is that the update or delete finished or failed.
2. Single or Maybe: Similarly to inserting an object, you can return a Single or Maybe when updating or deleting an object in the database. However, instead of returning Single<Long> or Maybe<Long>, you need to return a Single<Int> or Maybe<Int>. The Int value represents the number of rows affected by this update, so even if you update a list of objects you’ll still use Single<Int> or Maybe<Int>.

Querying

Querying is where all the magic happens, and you can use the full suite of reactive types, other than Completable, which doesn’t make much sense in a querying context:

1. Single: If you declare a query’s return type as Single, it will return a single instance of whatever object you’re querying. If, however, the database doesn’t contain the object your querying for, your single will emit an error, since a Single needs to emit exactly one value.
2. Maybe: Similarly to Single, a query returning a Maybe will return a single object if it exists in the database. However, if it doesn’t exist, the Maybe will complete normally.
3. Observable/Flowable: If you specify your return type as Observable or Flowable, you’ll receive the items that correspond to that query in the Observable or Flowable. Whenever the underlying data in the database changes, the Observable or Flowable will emit again.

Note: When using a query of type Observable or Flowable, Room will make sure that your query runs off the main thread without you needing to use the subscribeOn operator.

Reacting to database changes

Now that you’ve exposed methods to insert and retrieve books from the database, it’s time to update the app to utilize those methods.

Open the MainViewModel class and replace the observable declaration at the top of the init block with the following:

// 1
val observable = OpenLibraryApi.searchBooks("Lord of the Rings")
  .subscribeOn(Schedulers.io())
  // 2
  .flatMapCompletable {
    database.bookDao().insertBooks(it).toV3Completable()
  }
  // 3
  .andThen(database.bookDao().bookStream().toV3Observable())
  .share()

Here’s a breakdown of the changes:

1. Just like before, you’re using the OpenLibraryApi class to search for books.
2. This time, instead of just returning those books, you’re inserting all of them into the database using the insertBooks method you wrote earlier. Since Room is still using the RxJava2 library, you’re using the toV3Completable extension method to convert an RxJava2 Completable to an RxJava3 Completable. Don’t forget that after this operator the chain will now be an instance of Completable.
3. Once you’ve inserted the books you’re using, the andThen operator on Completable to transition to the Observable<List<Book>> returned by the bookStream method. Just like before you’re using the toV3Observable method to transition from an RxJava2 Observable to an RxJava3 Observable.

You’re now fetching books from the server and saving them to the database. You’re then querying and observing the database for books, and powering the views based off that Observable.

Since the type of observable switched from Single to Observable, you’ll need to update all of the subscribeBy calls in the class to use the onNext parameter instead of onSuccess, since onSuccess is specific to Single.

Here’s an example of the updated Rx chain that pushes new values to the allBooksLiveData object:

observable
  .subscribeBy(
    onNext = { item -> allBooksLiveData.postValue(item) },
    onError = { print("Error: $it") }
  )
.addTo(disposables)

Once you’ve made those changes, run the app. Everything should work the same and you should see the full list of Lord of the Rings books on the initial page.

Updating individual items

The next thing you need to do to get the BookCollector app up and running is to fill out the details of the favoriteClicked and readClicked methods in the MainViewModel. However, before you do that, you’ll need a way to insert a single updated book into the database.

Head back to the BookDao class and add the following method to the interface:

@Update(onConflict = OnConflictStrategy.REPLACE)
fun updateBook(book: Book): Single<Int>

updateBook takes a book and inserts it into the database, replacing any existing value that’s already there. It returns a Single<Int>, where the Int value represents the number of rows updated.

Now that you’ve got a way to update an individual book, it’s time to fill out the favoriteClicked and readClicked methods to utilize that new method.

Back in MainViewModel replace the favoriteClicked method with the following:

fun favoriteClicked(book: Book) {
  database.bookDao()
    .updateBook(book.copy(isFavorited = !book.isFavorited))
    .toV3Single()
    .subscribeOn(Schedulers.io())
    .subscribe()
    .addTo(disposables)
}

You’re using the new updateBook method and passing through a new instance of Book with the isFavorited flag toggled. Note that you’re not actually doing anything with the return value. Instead, the app relies on the bookStream Observable to emit a new list of items since the data has been updated.

Run the app and click the star icon on one of the books. You’ll notice the outlined star icon fills in. Swipe to the right to get to the list of favorites for the book, and you’ll notice the items you starred show up. You can even “unstar” one of the books from this list and it will disappear. All of this behavior is being driven by the bookStream observable, which is emitting a new list of books every time you favorite or unfavorite a book!

Now, replace the readClicked method with a similar body:

fun readClicked(book: Book) {
  database.bookDao()
    .updateBook(book.copy(isAlreadyRead = !book.isAlreadyRead))
    .toV3Single()
    .subscribeOn(Schedulers.io())
    .subscribe()
    .addTo(disposables)
}

Just like before, you’re using the updateBook method and sending through a Book with the isAlreadyRead flag toggled.

Run the app again. Click the envelope icon on a few books and then swipe to the third page. You should see all of the books you marked as read, and just like on the favorites page you can click the read icon here and see them disappear!

By utilizing Room’s reactive types, you’ve created a fully reactive app that observes a database and listens for new values, which are emitted every time the saved list of books is updated. Pretty magical, right?

Starting the app with cached data

The app is working great, but it’s not fully utilizing the fact that it’s using a database. Specifically, when the app starts, it’s immediately making a network request and not showing any information until that request finishes. That’s a bummer since you’ve got the data at your fingertips!

Luckily, you can make short work of that issue by using the startWith operator. In MainViewModel, replace the existing observable declaration at the top of init with the following:

val observable = OpenLibraryApi.searchBooks("Lord of the Rings")
  // 1
  .retryWhen { it.delay(5, TimeUnit.SECONDS) }
  .subscribeOn(Schedulers.io())
  .flatMapCompletable {
      database.bookDao().insertBooks(it).toV3Completable()
   }
  .andThen(database.bookDao().bookStream().toV3Observable())
  // 2
  .startWith(database.bookDao().bookStream().toV3Observable()
    .take(1))
  .share()

There’re two new operators at play, here:

1. You’re using the retryWhen operator to retry a failed network request after five seconds. For more information on how retryWhen works, check out Chapter 12, “Error Handling in Practice.”
2. You’re using the startWith operator to kick off the Observable with the books already in the database. You’re using the take operator to take the first List<Book> from the database since you only care about what’s initially in the database.

Put the phone in Airplane mode and run the app. You should immediately see books from the database populate the app. Once you turn off Airplane mode and wait a few seconds, the app will be populated with the latest and greatest from the network.

Paging data in

Now that you’ve explored the Room libraries Rx integration, it’s time to implement infinite paging using the paging library.

Open OpenLibraryService.kt and update the searchBooks method to take in a page number:

@GET("search.json")
fun searchBooks(
  @Query("q") searchTerm: String,
  @Query("page") page: Int
): Single<OpenLibraryResponse>

Now, open OpenLibraryApi.kt and update the searchBooks method to take in a page number:

fun searchBooks(searchTerm: String, page: Int = 1): Single<List<Book>> {
  return service.searchBooks(searchTerm, page)
  ...
}

By default, you’ll start at the first page, so you can use a default parameter there.

The paging library utilizes a different type of adapter to handle the special type of lists it works with. Open BookAdapter.kt and update the type of adapter BookAdapter to extending the following:

PagedListAdapter<Book, BookViewHolder>(getDiffUtil())

Getting an item from the underlying list in a PagedListAdapter can return a null item, so replace the book declaration in onBindViewHolder with the following:

val book = getItem(position) ?: return

Now that your API is ready to go and your adapter is all set up, it’s time to hook into the Paging library and start paging some content in!

Just like before, the strategy is going to be to load directly from the database and populate it with more data as the user scrolls in the RecyclerView. To do that, you’ll need a DataSource. A DataSource is a class specific to the Paging library that aids in loading data from some source. Apt name, right?

Now normally you’d have to create a new class that extends DataSource. But, since you’re using Room, you can do something that kind of feels like cheating in the programming world.

Open BookDao.kt and replace the existing bookStream method with the following:

@Query("SELECT * from book ORDER BY title")
fun bookStream(): DataSource.Factory<Int, Book>

Room can actually generate DataSource.Factory objects for you just by changing the return type of your query! This factory class will create a DataSource, so you don’t have to create your own custom data source. The Factory in this case is of type Factory<Int, Book>, since the object type you’re operating on is a Book and the page type is an Int.

The next step in the Paging library journey is to create a PagedListBuilder. A PagedListBuilder is a class in charge of creating new instances of PagedList, which is that special type of list you updated the BookAdapter to handle.

Now, normally, you’d use the LivePagedListBuilder class to get a LiveData<PagedList<Book>>, but luckily the wonderful team in charge of building the Paging library has created Rx bindings for the library, just like they did for Room!

Add the following to your build.gradle file:

implementation "androidx.paging:paging-rxjava2-ktx:$paging_version"

Now, open MainViewModel and delete everything in the init block. That’s right everything. The Paging library handles so much behind the scenes that you won’t need the code to manually hit the API anymore.

At the top of the now empty init block, add the following:

val config = PagedList.Config.Builder()
  .setEnablePlaceholders(false)
  .setPageSize(20)
  .build()

A PagedList.Config is a configuration object that tells the Paging library how you want to page data in. In this config, you’re setting the following configs:

1. You’re setting enablePlaceholders to false. If you know the size of your dataset before you start paging data in, you can set placeholder values so the user has an accurate scroll bar on their RecyclerView. Since you’re querying an API, you don’t have this information so you can just set it to false.
2. You’re setting the page size to 20. That means the paged list will load in 20 objects at a time from the database, which will take up several scrollable pages in the RecyclerView.

Below the config declaration, use the RxPagedListBuilder class to create an Observable<PagedList<Book>>:

RxPagedListBuilder<Int, Book>(
    database.bookDao().bookStream(), config)
  .buildObservable()
  .toV3Observable()  
  .subscribe(allBooksLiveData::postValue)
  .addTo(disposables)

RxPagedListBuilder is one of the classes exposed by RxJava paging integration. It allows you build an Observable<PagedList<Book>>, which you can then utilize however you want. In the above example you’re passing in the DataSource.Factory returned from the BookDao.bookStream method and the config you just created.

Before you can run the app, you’ll need to update the LiveData objects at the top of the class to be of type MutableLiveData<PagedList<Book>>:

val allBooksLiveData = MutableLiveData<PagedList<Book>>()
val favoriteBooksLiveData = MutableLiveData<PagedList<Book>>()
val alreadyReadBooksLiveData =
  MutableLiveData<PagedList<Book>>()

Now, all the components of your app are speaking the same language.

Run the app. Assuming you haven’t uninstalled the app since the last time you ran it as part of this chapter, you should see content paging in from the database on the main page. If you scroll to the other two pages, they’ll be empty. Time to fix that.

Open BookDao.kt again and add the following two methods below bookStream:

@Query("SELECT * from book WHERE isFavorited = 1 ORDER BY title")
fun favoritesStream(): DataSource.Factory<Int, Book>

@Query("SELECT * from book WHERE isAlreadyRead = 1 ORDER BY title")
fun alreadyReadStream(): DataSource.Factory<Int, Book>

Remember how you used to use filter and map operators to pull out the favorited and already-read books? You’re moving that logic into the database and exposing two new queries: one for favorited books and one for books that have already been read. Just like in the bookStream method, you’re returning a DataSource.Factory<Int, Book> to be used with the Paging library.

Open MainViewModel.kt again and add matching RxPagedListBuilder statements for your new queries:

RxPagedListBuilder<Int, Book>(
    database.bookDao().favoritesStream(), config)
  .buildObservable()
  .toV3Observable()
  .subscribe(favoriteBooksLiveData::postValue)
  .addTo(disposables)

RxPagedListBuilder<Int, Book>(
    database.bookDao().alreadyReadStream(), config)
  .buildObservable()
  .toV3Observable()
  .subscribe(alreadyReadBooksLiveData::postValue)
  .addTo(disposables)

Now, run the app again. If you scroll to the right, you’ll see the favorited and already-read book lists being populated as expected. Toggle the favorited status of a book to confirm that everything works.

Paging in from the network

This app is looking beautiful, but there’s one problem: it’s not pulling anything from the server! Right now the app is only serving up cached data from the database; it’s never actually fetching anything new. If you were to uninstall and reinstall the app, you wouldn’t see any content because nothing would actually be downloaded.

Here’s a diagram of what’s happening right now:

And here’s the end goal for the app:

To achieve the desired flow of the app pulling down new books from the server when it runs out of content, you’ll need to use another utility that the Paging library provides: a BoundaryCallback.

A BoundaryCallback is a handy utility that you can use to execute some action whenever you’ve run out of cached content to display. It’s use case is specifically oriented towards apps that want to load data into a database and then fetch new data from a server once the app has displayed all of the data already loaded in the database.

Open the BookBoundaryCallback.kt class. It has two unimplemented methods that you’ll fill out: onZeroItemsLoaded and onItemAtEndLoaded.

onZeroItemsLoaded is called when there’s no items yet loaded into the database. That would usually be the scenario after the first time the app runs. You’ll use this callback to load the first page of data from the network.

onItemAtEndLoaded is called when the last item is loaded from the database. You’ll use this callback to load the latest page of data from the network.

Both methods will have very similar implementations, so you’re going to create a helper function. Add the following method at the bottom of the class:

private fun loadItems(requestType: PagingRequestHelper.RequestType) {
  // 1
  helper.runIfNotRunning(requestType) { callback ->
    // 2
    OpenLibraryApi.searchBooks(searchTerm, currentPage)
      // 3
      .flatMapCompletable {
        db.bookDao().insertBooks(it).toV3Completable()
      }
      .subscribeOn(Schedulers.io())
      // 4
      .subscribe {
        currentPage++
        callback.recordSuccess()
      }
  }
}

Here’s a breakdown of the loadItems method:

1. Sometimes onZeroItemsLoaded or onItemAtEndLoaded can be called multiple times. That can be a bit of a problem because you don’t want to kick off multiple network requests to load a single page of data. The paging team has provided a handy PagingRequestHelper class to help coordinate running these asynchronous tasks. Here, you’re using the runIfNotRunning method to run a block of code depending on if the passed in RequestType has been started or not.
2. You’re using the searchBooks method to search for a list of books, passing through the current page, which starts at one.
3. You’re then inserting the books you get back from the server into the database.
4. Finally, you’re incrementing the page count so the next time you make a network request you fetch the next page. You’re also letting the PagingRequestHelper know that the initial fetch was finished by calling the recordSuccess method on the callback object runIfNotRunning provides.

Now, replace the onZeroItemsLoaded and onItemAtEndLoaded methods with the following:

override fun onZeroItemsLoaded() {
  loadItems(PagingRequestHelper.RequestType.INITIAL)
}

override fun onItemAtEndLoaded(itemAtEnd: Book) {
  loadItems(PagingRequestHelper.RequestType.AFTER)
}

You’re passing a RequestType of INITIAL for the first download and AFTER for subsequent downloads.

Now that you’ve got a BoundaryCallback ready to go, you can set it on the RxPagedListBuilder objects you set up earlier.

Back in MainViewModel, add the following line after each RxPagedListBuilder declaration:

.setBoundaryCallback(
    BookBoundaryCallback("The lord of the rings", database))

Run the app again. You should be able to scroll to your heart’s content. If by some magic you manage to get to the end of the (astoundingly large) list of Lord of the Rings books, congratulations! You’ve officially learned everything there is to know about how many cool Lord of the Rings books there are!

Key points

• Room allows you to specify reactive types in your Dao objects.
• You can use Completable or Single or Maybe when inserting, updating or deleting items from the database.
• You can use Observable or Flowable when querying items from the database.
• Your query Observable will keep emitting as data changes in the database!
• The Paging library comes with an RxJava extension that allows you to stream PagedList objects.
• Room and the Paging library make for a fantastic reactive combination!

Where to go from here?

The Room and Paging libraries are great examples of how a library can effectively integrate Rx into its API. Given these libraries are written by Google, its nice to know that you’re getting first party support for Rx from these libraries.

If you’re interested in learning more about either libraries and the extend of their Rx support, you can checkout the documentation (medium.com/androiddevelopers/room-rxjava-acb0cd4f3757)(developer.android.com/topic/libraries/architecture/paging#ex-observe-rxjava2) for a deeper look at the integration.