10. Combining Operators in Practice

Written by Alex Sullivan & Florent Pillet

In the previous chapter, you learned about combining operators and worked through increasingly more detailed exercises on some rather mind-bending concepts. Some operators may have left you wondering about the real-world application of these reactive concepts.

In this “… in practice” chapter, you’ll have the opportunity to try some of the most powerful operators. You’ll learn how to solve problems similar to those you’ll face in your own applications.

Note: This chapter assumes you’ve already worked your way through Chapter 9, “Combining Operators,” and are familiar with both filtering (Chapter 5) and transforming operators (Chapter 7). At this point in the book, it is important that you are familiar with these concepts, so make sure to review these chapters if necessary!

You’ll start with a new project for this chapter and build a small application with an ambitious name: Our Planet.

Getting started

The project will tap into the wealth of public data exposed by NASA. You’ll target EONET, NASA’s Earth Observatory Natural Event Tracker. It is a near real-time, curated repository of natural events of all types occurring on the planet. Check out https://eonet.sci.gsfc.nasa.gov/ to learn more!

To get started with Our Planet, open the starter project for this chapter in Android Studio 4.0 or newer.

Build and run the starter application; the default screen is empty.

Your tasks with this application are as follows:

• Gather the event categories from the EONET public API https://eonet.sci.gsfc.nasa.gov/docs/v2.1 and display them on the first screen.
• Download events and show a count for each category.
• When user taps a category, display a list of events for this category.

You’ll learn how useful combineLatest can be in several situations, but you’ll also exercise concat, merge, and scan. Of course, you’ll also rely on operators you are already familiar with, like map and flatMap.

Preparing the EONET API class

Good applications have a clear architecture with well-defined roles. The code that talks with the EONET API shouldn’t live in any of the activities. Instead, it will live in an object that you’ll reference from a ViewModel.

Expand the model package in the OurPlanet project; you’ll find a Retrofit interface ready for the app to utilize in EONETApi.kt. That interface will be used by the EONET object. You’ll also find EOCategoriesResponse, EOCategory and EOEvent classes that map to the content delivered by the API.

Open the EONET object; it’s already fleshed out with the basic structure of the class, including the API URL, endpoints, and some date formats you’ll use later on.

Fetching categories

Now open the EONETApi file and add a fetchCategories method to the bottom of the class, after the companion object block.

@GET(EONET.CATEGORIES_ENDPOINT)
fun fetchCategories(): Observable<EOCategoriesResponse>

fetchCategories will fetch the different event categories from the EONET API. It returns an Observable<EOCategoriesResponse>. EOCategoriesResponse is a simple wrapper class for a list of categories, which are represented here via the AnyMap typealias.

Just like in previous projects, you’re using AnyMap to represent a simple Map<String, Any> for objects returned from the network. You’ll deserialize that AnyMap later on.

Next, open the EONET object and add the following method:

fun fetchCategories(): Observable<EOCategoriesResponse> {
  return eonet.fetchCategories()
}

This exposes the fetchCategories method you just created to other consumers of the EONET object.

Updating the CategoriesViewModel

Open CategoriesViewModel and add the following to the empty startDownload method:

// 1
EONET.fetchCategories()
  // 2
  .map { response ->
    val categories = response.categories
      categories.mapNotNull { EOCategory.fromJson(it) }
  }
  // 3
  .share()
  // 4
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  // 5
  .subscribe {
    categoriesLiveData.value = it
  }
  .addTo(disposables)

In this block you’re doing the following:

1. Use the new fetchCategories method you added above to query the EONET API for event categories
2. Use the map operator to turn the response object you received from the previous call into a List<EOCategory>
3. Use the share operator so you can re-use this Observable later on.
4. Use the subscribeOn and observeOn methods to make sure you’re querying the network off the main thread and observing the results on the main thread (you’ll learn more about these operators in Chapter 13: “Intro to Schedulers”)
5. Finally, subscribe to the Observable and update the categoriesLiveData with the data coming through. 8 Run the app. You should see a list of events, each of which show as having zero events listed.

Add events into the mix

Now that you’ve got the categories loading, it’s time to update the app to actually display the number of events in the category.

Add a new EOEventsResponse class in the model package:

class EOEventsResponse(val events: List<AnyMap>)

EOEventsResponse will model the top level response you get back from the EONET API.

Open the EONETApi class and add the following method below the fetchCategories method:

@GET(EONET.EVENTS_ENDPOINT)
fun fetchEvents(
  @Query("days") forLastDays: Int,
  @Query("status") status: String
): Observable<EOEventsResponse>

fetchEvents queries the EONET API for events that have happened within a certain number of days, and that have a status of either opened or closed.

Now that you have your Retrofit method set up, it’s time to expose it in the EONET object. But before you expose the fetchEvents method publicly, you’ll write a small helper method to help ease the process of returning both open and closed events. Open EONET and add the following events method:

// 1
private fun events(forLastDays: Int, closed: Boolean):
  Observable<List<EOEvent>> {
  // 2
  val status = if (closed) "closed" else "open"
  // 3
  return EONET.eonet.fetchEvents(forLastDays, status)
    //4
    .map { response ->
      val events = response.events
      events.mapNotNull { EOEvent.fromJson(it) }
    }
}

The events helper method works as follows:

1. It takes two parameters. The first parameter represents how many days back you want to go when fetching events from the EONET API. The second is a Boolean representing whether you want to look up closed or open events.

2. You’re transforming the Boolean closed variable into a String that the EONET API can understand.

3. You’re then using the fetchEvents method you just wrote to fetch events for a set number of days back and with the status that you just determined.

4. You’re then taking the response from that network call and using the map operator to transform it into a List<EOEvent>, similar to what you did earlier with the EOCategory.

Now that you have the helper method out of the way, you can create the public fetchEvents method that will expose the events API to a consumer.

Add the following empty method above the events method:

fun fetchEvents(forLastDays: Int = 360): Observable<List<EOEvent>> {

}

You’re supplying a default number of days to make consumption of this API easier.

When you fetch events from the EONET API, you want to fetch both the closed and open events, so you get the full picture.

Add the following code at the top of the empty fetchEvents method:

val openEvents = events(forLastDays, false)
val closedEvents = events(forLastDays, true)

You’re using the helper events method you just implemented to create two new Observables - one for the closed events and one for the open events.

Now that you have two Observables you want to combine them together. Funny how that would be the case in this chapter! You’re going to use a concat method to combine openEvents and closedEvents. Add the following code below the two Observable declarations:

return openEvents.concatWith(closedEvents)

Here’s what’s going on in the code you just added:

concatWith combines the openEvents Observable with the closedEvents Observable by first emitting all the events from the openEvents Observable and then following up with the events from the closedEvents Observable. If either of them emit an error, the whole Observable will terminate and emit that error. This is an OK solution for now, but you’ll improve on it later on.

Combining events and categories

you’ve got a fancy fetchEvents method that fetches all of your events, so now it’s time to utilize it in the CategoriesViewModel class.

You’re going to update the startDownload method to combine the categories you’re currently fetching with the events associated with each category.

First off, you’re going to stop subscribing to the Observable produced by the fetchCategories method and instead save it off in a variable named eoCategories, making sure to remove all the operators after the map operator. You’re new trimmed down method should look like this:

fun startDownload() {
  val eoCategories = EONET.fetchCategories()
    .map { response ->
      val categories = response.categories
      categories.mapNotNull { EOCategory.fromJson(it) }
    }
}

Now you’re going to create a new Observable utilizing the fetchEvents method you just added. Add the following to the bottom of the startDownload method:

val downloadedEvents = EONET.fetchEvents()

You’re using the default number of days to create your downloadedEvents Observable.

Next you’re going to use another combining trick you learned about in the last chapter to combine downloadedEvents and eoCategories: the combineLatest method. Add the following to the startDownload method below the downloadedEvents declaration:

// 1
val updatedCategories = Observables
  .combineLatest(eoCategories, downloadedEvents)
    { categoriesResponse, eventsResponse ->
      // 2
      categoriesResponse.map { category ->
        // 3
        val cat = category.copy()
        // 4
        cat.events.addAll(eventsResponse.filter {
          it.categories.contains(category.id)
        })
        // 5
        cat
      }
    }

Wow, that’s a hearty block of code! Here’s a breakdown:

1. You’re using the combineLatest method to combine the latest emission from eoCategories with the latest emission from downloadedEvents. Take a look at the diagram below this explanation for some visual help on making sense of combineLatest.

2. combineLatest takes a function that receives one item from the first Observable and one item from the second Observable as an argument. So in this lambda block you have categoriesResponse, which is a List<Category>, and eventsResponse, which is a List<EOEvent>.

   You’re using the Kotlin standard library map method on the categoriesResponse list to transform each element.

3. Within the map lambda, you’re creating a new copy of the current EOCategory you’re looping through.

4. You’re then adding every event from the eventsResponse list that has an associated category with the same id as the current EOCategory into the new cat copy. In other words, you’re adding every event that belongs to category to your copy cat.

5. Finally, you’re using Kotlin lambda implicit returns to return the new category copy. Remember though that the cat line at the bottom of the block is the return value for the categoriesResponse.map line, which is itself the return value for the combineLatest lambda. There’s some lambda inception going on here!

9

Last but not least, you’re going to combine the Observable of empty categories, i.e. eoCategories, with the new Observable of populated categories, updatedCategories. To do this, you’ll use the concatWith operator. Finally, you will subscribe to the resulting Observable:

eoCategories.concatWith(updatedCategories)
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe({
    categoriesLiveData.value = it
  }, {
    Log.e("CategoriesViewModel", it.localizedMessage)
  })
  .addTo(disposables)

Just like before you’re using subscribeOn and observeOn to handle your threading needs. By using concatWith to combine the empty and populated categories, you’ll be able to quickly show some data, the empty categories, while downloading new events to then combine with the categories.

There’s one more small update to make before you can run the app. Open the CategoriesAdapter class and add the following line to the top of the updateCategories method:

this.categories.clear()

This will clear out the existing list of categories when you get a new list. Build and run the app.

You should see the categories listed and quickly updated to show the real count of events associated with that category.

Downloading in parallel

Recall that the app is currently calling the EONET events endpoint twice. Once for closed events and once for open events. Since you’re using the concat operator, it first downloads the open events and then the closed events.

Wouldn’t it be great if you could parallelize that and download both events at once? With RxJava, not only can you achieve that parallelization, but you can do it without touching any of your UI code.

The EONET API allows you to download events in two ways. The first approach, which the app is currently taking, is to download all of the events at once. The second approach is to download events by category.

You’ll refactor the app to download events by category rather than all at once. It’ll be more complicated than the current approach, but you’re already an RxJava ninja, so you’ll get through it fine!

First, open the EONETApi class and update the fetchEvents method to take in its GET end point as a parameter:

@GET("{endpoint}")
fun fetchEvents(
  @Path("endpoint", encoded = true) endpoint: String,
  @Query("days") forLastDays: Int,
  @Query("status") status: String
): Observable<EOEventsResponse>

Retrofit makes this easy, since you can use the @Path annotation to update the endpoint in the @GET annotation.

Next, update the private events method in the EONET object to utilize the changed fetchEvents method:

private fun events(
  forLastDays: Int,
  closed: Boolean,
  endpoint: String
): Observable<List<EOEvent>> {
  val status = if (closed) "closed" else "open"
  return EONET.eonet.fetchEvents(endpoint, forLastDays, status)
    .map { response ->
    val events = response.events
    events.mapNotNull { EOEvent.fromJson(it) }
    }
}

You’re now passing in an endpoint, which is a String, into the events helper method. You’re then passing the endpoint through to the fetchEvents method you updated earlier.

Now update the public fetchEvents method in the EONET object to take in an EOCategory and pass its endpoint value into the open and closed events Observables:

fun fetchEvents(category: EOCategory, forLastDays: Int = 360):
    Observable<List<EOEvent>> {
  val openEvents =
    EONET.events(forLastDays, false, category.endpoint)
  val closedEvents =
    EONET.events(forLastDays, true, category.endpoint)

  return Observable.concat(openEvents, closedEvents)
}

Since you want the two network calls to be made in parallel, you should use the Observable.merge method to merge the two observables together instead of concat. Update the return statement of the fetchEvents method to use merge:

return Observable.merge(openEvents, closedEvents)

Incrementally updating events

You’ve done a lot of great work to parallelize downloading closed and open events, but there’s still a bit farther to go.

Open CategoriesViewModel. In the startDownload method you’ll notice that there’s an error on the line declaring downloadedEvents. That’s because you’re not passing in an EOCategory to the fetchEvents method.

You may not have an instance of EOCategory lying around, but you’ve got the next best thing. The eoCategories Observable defined at the top of the startDownload method emits lists of EOCategory objects. All you need to do is tie into the eoCategories Observable to get your EOCategory instance!

Replace the original downloadedEvents Observable with the following two Observables:

// 1
val eventsObservables = eoCategories.flatMap { categories ->
  // 2
  val categoryEventObservables = categories.map { category ->
    EONET.fetchEvents(category)
  }
  // 3
  Observable.fromIterable(categoryEventObservables)
}
// 4
val downloadedEvents = Observable.merge(eventsObservables, 2)

That’s a dense chunk of code. Here’s a breakdown:

1. You’re calling flatMap on the eoCategories Observable. eoCategories emits values of type List<EOCategory>, so categories here is a list of categories. Remember that flatMap expects you to return an Observable from its lambda.

2. You’re then using the Kotlin standard library map method and the EONET.fetchEvents method to transform each EOCategory in categories into an Observable<List<EOEvent>>. That means the type of categoryEventObservables is actually List<Observable<List<EOEvent>>>. That’s an intimidating type signature!

3. Next up you’re using Observable.fromIterable to transform your List<Observable<List<EOEvent>>> into an Observable<Observable<List<EOEvent>>>. This is the ultimate type of eventsObservables. It’s hard to reason about Observables inside Observables, but luckily you can flatten things out…

4. By using the merge operator! merge has a plethora of helpful versions. Rather than providing merge with a set number of Observables, you can instead give it an Observable that emits Observables and it will combine them all together. Merge them, if you will :].

   Merge also allows you to pass in a maximum number of concurrent subscriptions to those Observables. You don’t want your app to make too many network requests at once, so by passing in an upper limit on the number of Observables being subscribed to you can help lessen the load.

The end result of the above code is that you get an Observable<List<EOEvent>> just like before, except now each category that comes through the eoCategories Observable triggers a call to get that category’s associated EOEvent objects.

Head back to the EONET object and add the following function at the bottom of the object:

fun filterEventsForCategory(
    events: List<EOEvent>,
    category: EOCategory
): List<EOEvent> {
  // 1
  return events.filter { event ->
    // 2
    event.categories.contains(category.id) &&
        // 3
        !category.events.map { it.id }.contains(event.id)
  // 4
  }.sortedWith(EOEvent.compareByDates)
}

The above method is used to get a list of EOEvent objects that are associated with, but not already added to, an EOCategory. Here’s a breakdown:

1. Call the Kotlin standard library filter operator on the passed in list of events. The filter lambda expects either true or false to determine if it should include the object in the list it returns.
2. Check if the list of category ids on the event contains the id of the category passed into this filterEventsForCategory function.
3. Also check to see if the passed in category already contains the event that filter is currently operating on. If the event belongs to the passed in category, and that category doesn’t already contain the event, return true so the filter will include this event in its return list.
4. Last but not least, sort the resulting List<EOEvent> based off their dates.

Back in the startDownload method in the CategoriesViewModel, replace the updatedCategories declaration with the following:

// 1
val updatedCategories = eoCategories.flatMap { categories ->
  // 2
  downloadedEvents.scan(categories) { updated, events ->
    // 3
    updated.map { category ->
      val eventsForCategory =
        EONET.filterEventsForCategory(events, category)

      if (!eventsForCategory.isEmpty()) {
        val cat = category.copy()
        cat.events.addAll(eventsForCategory.filter {
          it.closeDate != null
        })
        cat
      } else {
        category
      }
    }
  }
}

Broken down, the above code:

1. Again calls flatMap on the eoCategories Observable. Since eoCategories emits objects of type List<EOCategory>, the categories argument to the flatMap lambda is also of type List<EOCategory>.

2. Then use the scan operator on the downloadedEvents Observable to progressively build up fully populated instances of EOCategory. Recall that the scan operator takes in an initial value and then runs a function you provide every time an item is emitted from the Observable, updating that initial value you provided as it goes. In this case the initial value that you provide is a List<EOCategory>. Each EOCategory in the list, however, has no associated EOEvents.

3. In the scan operators accumulator function, you’re mapping through the list of EOCategorys and creating a new list of events that are associated with that category, using the filterEventsForCategory method you wrote earlier. You’re then creating a copy of the category and adding all of its associated events to its own internal list of events and returning that copy. If there are no new events for the category, you’re just returning the original category without modifying its events.

By using the scan combining operator, you’re able to slowly but surely get a fully populated list of EOCategory objects without waiting for every network call to finish.

Phew! That was a lot of very complex Rx code.

Build and run the app now. You’ll see that the counts for the categories increase as new network calls are made to populate each category.

Click into one of the categories and you’ll see a screen that shows your progress.

Interacting with the slider doesn’t do anything. Yet. You’ll change that next.

Wiring up the days seek bar

Open the EventsActivity class and add the following at the top of the class:

private val days = BehaviorSubject.createDefault(360)

private val subscriptions = CompositeDisposable()

You’ll be using a BehaviorSubject to represent the number of days currently selected in the days slider.

Next add the following to the bottom of the onCreate function:

seekBar.setOnSeekBarChangeListener(
  object : SeekBar.OnSeekBarChangeListener {
    override fun onProgressChanged(
      seekBar: SeekBar?, progress: Int, fromUser: Boolean) {
        days.onNext(progress)
      }

      override fun onStartTrackingTouch(seekBar: SeekBar?) {}
      override fun onStopTrackingTouch(seekBar: SeekBar?) {}
    })

You’re using an OnSeekBarChangeListener to forward the current seek bar position to the days BehaviorSubject. Now whenever you scroll, the subject days will emit the new seekbar progress.

Next add the following below the OnSeekBarChangeListener code:

val allEvents = intent
  .getParcelableExtra<EOCategory>(CATEGORY_KEY).events
val eventsObservable = Observable.just(allEvents)

You’re retrieving the EOCategory from the Activity intent and creating a new eventsObservable from the category events.

Now it’s time for the Rx magic. Add the following below the eventsObservable declaration:

// 1
Observables
  .combineLatest(days, eventsObservable) { days, events ->
    // 2
    val maxInterval = (days.toLong() * 24L * 3600000L)
    // 3
    events.filter { event ->
      val date = event.closeDate
      if (date != null) {
        abs(date.time - Date().time) < maxInterval
      } else {
        true
      }
    }
  }
  // 4
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe {
    adapter.updateEvents(it)
  }
  .addTo(subscriptions)

Here’s a breakdown of the code:

1. You’re using the combineLatest factory method to combine the days subject with the eventsObservable. Recall that combineLatest combines the last emitted value from each passed in Observable. Since eventsObservable only emits one item, that means that every time the days subject emits, its value will be combined with the full list of events.
2. Now you’re taking the number of days emitted by the days subject and converting it into a number of milliseconds. Specifically, you’re multiplying the number of days by 24 hours and then the number of hours by 3,600,000 milliseconds.
3. You’re then filtering out any events that haven’t happened within the time period entered on the seek bar.
4. Last but not least, you’re applying the appropriate schedulers and sending the events that passed the filter to the RecyclerView adapter for this page.

Now add the following below the code you just wrote:

days
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe {
      daysTextView.text =
        String.format(getString(R.string.last_days_format), it)
    }
    .addTo(subscriptions)

You’re again observing the days subject and updating daysTextView to show the proper value.

There’s only one thing missing. You need to clean up after yourself! Implement the onDestroy method and dispose of your subscriptions:

override fun onDestroy() {
  subscriptions.dispose()
  super.onDestroy()
}

Now run the app and click into a category. As you slide the seek bar you’ll notice the events displayed become filtered down to only those that have happened in the time frame you’ve selected.

Magic, right?

Challenge: Adding a progress bar

Start from the final project in this chapter. Place an indeterminate horizontal progress bar below the toolbar and above the list of categories on the main screen. The progress bar should show while the categories and events are being downloaded and be hidden as soon as the downloads finish.

You’ll need to update the layout in the activity_categories.xml file to include a horizontal progress bar. You’ll also need to add a few constraints to it to get everything working correctly within the containing ConstraintLayout.

Hiding the progress bar will be a side effect of your reactive code. You should also use a LiveData object in your CategoriesViewModel class to communicate hiding and showing the progress bar with the CategoriesActivity.

Key points

• The concatWith method can be used to combine two Observables to emit one after the other. Watch out for your error handling though, since one Observable encountering an error will end the whole chain!
• If you need to parallelize multiple Observables, you can use the merge method to interweave the Observables. You can also limit the number of concurrent subscriptions happening!
• combineLatest can be effectively used to combine the last values of multiple Observables. It’s particularly useful if you have one Observable that may not update often and another that updates frequently. Combining the two Observables with combineLatest can save you from writing a lot of stateful code!
• The merge method has a ton of overloads. If you have a collection of Observables, there’s almost certainly a merge overload out there to merge your collection together. It even works if you have an Observable of Observables!
• The scan operator can be used to continuously emit items as you build up progress in some process. For this chapter, the progress was fetching events for a certain type of category. If you need to build up to a final product, scan or reduce are both great options.

Where to go from here?

That wraps up our chapters focusing on filtering, transforming and combining operators. You’ve seen them all in action in Android apps.

Before moving on to Section III of the book, you’ll spend a chapter learning about another type of operator: time-based operators. And you’ll do so while working within another Android app project.