RxJava Filtering Operators

RxJava is a powerful library for handling asynchronous code based on events. It avoids callbacks by providing a composable style of programming. RxJava provides many versatile operators you can use to process your data in straightforward ways.

In this tutorial, you’ll learn about RxJava Filtering Operators, a group of powerful operators that let you work with the data you need at a specific step of the data transformation process.

In the process you’ll learn how to:

• Take a select number of elements from your data stream.
• Let your data stream skip elements based on a condition.
• Filter your data based on a predicate.
• Take an element only after the user finishes giving input.
• Act on sequential elements only when they differ.

These operators will form the base of your RxJava vocabulary. Time to dive in!

Getting Started

Download the starter project by clicking on the Download Materials button at the top or bottom of the tutorial. Then, open the starter project in Android Studio, where you’ll find Foodeat, a restaurant-searching app.

Build and run. You’ll see the following screen:

You’ll see a list of restaurants ordered from highest to lowest rated. The app stores the restaurants in a Room database. You’ll also see a search field and a menu with some items that represent filtering criteria on the toolbar.

You’ll develop features to filter information based on the criteria given by the search field and menu items. With this information, you’ll quickly pick somewhere to eat. :]

Before you start coding, take a moment to review some theory about what RxJava is.

What is RxJava?

RxJava is a library that extends the observer pattern for processing asynchronous code by using sequences and functional style operators. RxJava provides several advantages. With it, you can:

• React to new data in a sequential and isolated manner.
• Simplify multithreading.
• Avoid callback hell.
• Use operators that let you declare how to process data in each step of the chain in a simple way.

To make the most of these advantages, RxJava provides several helpful operators.

What are RxJava Filtering Operators?

RxJava comes with several operators to process data streams. You’ll use RxJava Filtering Operators extensively throughout your RxJava career. These filtering operators apply conditional constraints to the events you receive on your data stream.

You can apply these constraints to limit the values emitted by your stream, ensuring you work with only the elements you need. For example, the image above shows you can use a filter operator to get only even values from your stream.

The most common filtering operators are:

• debounce
• distinct
• elementAt
• filter
• ignoreElements
• last
• skip
• skipLast
• take
• takeLast
• takeUntil

While you’ll only work with a handful of these operators in this tutorial, you can use what you learn here to implement the other filtering operators. The first operator you’ll use is take.

Using the Take Operator

The take operator emits only the first n items emitted and then completes.

You’ll use this operator to take and display the highest rated restaurants. Open RestaurantViewModel.kt and replace getTopRestaurants() with:

fun getTopRestaurants() {
  _restaurantsLiveData.value = Resource.Loading

  restaurantSource
      .take(5)
      .showResults()
}

Here you get the stream of restaurants, let through only the first five emitted values, and then show the results. Look at the bottom of RestaurantViewModel.kt. You’ll see showResults():

//1
private fun Observable<Restaurant>.showResults() {
  this.toList() //2
      .subscribeOn(Schedulers.io()) //3
      .map { Resource.Success(it) } //4
      .subscribe(_restaurantsLiveData::postValue) //5
      .addTo(disposables) //6
}

This handy extension function helps you keep your code clean and DRY, or Don’t Repeat Yourself. Here’s a detailed breakdown:

1. You create an extension function for showing the results for any Observable emitting Restaurant values.
2. As soon as the Observable completes, you create a list with all the emitted values.
3. Then you subscribe to the corresponding scheduler.
4. You wrap the resulting list in a Resource object with a type of Success.
5. Next, you send all the Resource values to the corresponding LiveData.
6. Finally, you add the resulting disposable to a CompositeDisposable which is disposed of when the ViewModel is removed. This avoids memory leaks. Remember to always clean up after yourself. :]

Build and run. Go to the toolbar menu and tap the Top 5 Rated item. You’ll now see the filtered list showing the highest-rated restaurants.

The Take Operator Variations

There are many overloads and variations of take you can choose from to best suit your needs:

• takeLast: Only emits the final n items emmitted by an Observable. This operator needs an Observable that completes to know which elements are last. Otherwise, it won’t emit them.
• takeUntil: Emits items until a secondary Observable triggers by emitting an item or terminating.
• takeWhile: Emits items while a specified condition is true.
• elementAt: This is not necesarily a direct descendant of the take operator. But, by specifing an index, you can take that particular element in the corresponding index, achieving a very similar behavior.

Next, you’ll work with the skipwhile operator.

Using the SkipWhile Operator

Next, you’ll filter your restaurant list to show only restaurants rated three stars or fewer. You can start reducing your short-list for possible restaurants the next time you go out. :]

The skipWhile operator lets you skip values while a certain condition is true. When the condition is no longer met, the Observable starts emitting the rest of the values.

In RestaurantViewModel.kt, replace getLowestRatedRestaurants() with:

fun getLowestRatedRestaurants() {
  _restaurantsLiveData.value = Resource.Loading

  restaurantSource
      .skipWhile { it.rating > 3 }
      .showResults()
}

You get the restaurant stream of values, then skip all the values while the rating is above three.

Build and run. Go to the toolbar menu, but this time tap the Rated 3 and Below item. Now, you’ll see a list with all the restaurants with a rating of three stars and below. Instead of taking or skipping a specific number of items, you’re filtering for a particular condition.

Now that you’re more comfortable filtering your data, it’s time to go further.

The SkipWhile Operator Variations

skipWhile also has a few varients, each with their own conditions:

• skip: Skips the first n values and then emits the rest.
• skipLast: This operator won’t emit unless it has an Observable that completes and lets it know which elements are last. It skips the final n items emmitted by an Observable.
• skipUntil: Skips items until a secondary Observable triggers by either emitting an item or terminating.

Sometimes you’ll only care about emitted values when the data stream completes. The ignoreElements operator is the perfect candidate for these occasions.

Using the IgnoreElements Operator

ignoreElements transforms the Observable in a Completable. A Completable behaves similary to an Observable except it only emits terminal events, either onError or onComplete, and no values.

You’ll use ignoreElements to tell the user when data was successfully fetched and filtered. In RestaurantViewModel.kt, add the following code to the bottom of showResults:

this.ignoreElements() //1
    .subscribeOn(Schedulers.io()) //2
    .observeOn(AndroidSchedulers.mainThread()) //3
    .subscribe {
      _uiLiveData.value = Resource.Success(Unit) //4
    }
    .addTo(disposables) //5

Here’s a code breakdown:

1. Any Observable useing this extension also triggers a Completable.
2. The completable subscribes to the corresponding scheduler.
3. Then the completable observes the stream of data on the main thread provided by RxAndroid.
4. Everytime the stream of data completes, the completable sends a Resource of type Success to the activity through the corresponding LiveData.
5. Finally, the completable adds the resulting disposable to a CompositeDisposable that disposes when the ViewModel is removed.

Your whole extension will look like this:

private fun Observable<Restaurant>.showResults() {
  this.toList()
      .subscribeOn(Schedulers.io())
      .map { Resource.Success(it) }
      .subscribe(_restaurantsLiveData::postValue)
      .addTo(disposables)

  this.ignoreElements()
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe {
        _uiLiveData.value = Resource.Success(Unit)
      }
      .addTo(disposables)
}

Since both your getTopRestaurants() and getLowestRatedRestaurants() use this extension for displaying results, your Activity is notified on a separate LiveData when those streams have successfully completed.

Build and run. Go to the toolbar menu and tap either Top 5 Rated or Rated 3 and Below. A toast will notify you as soon as the data processing completes.

Next, you’ll learn about the filter operator.

Using the Filter Operator

The aptly-named filter is one of the most important and versatile filtering operators. It only passes values that match a particular predicate you declare.

The filter operator is a perfect fit for shortening your list to only the restaurants that match your specific search query. Open your RestaurantViewModel.kt and replace setQueryListener() with:

//1
fun setQueryListener(queryObservable: Observable<String>) {
  queryObservable
      .observeOn(AndroidSchedulers.mainThread())
      .map(::filterSource) //2
      .subscribe()
}


private fun filterSource(query: String) {
  Log.d(this::class.java.simpleName, "Search query: $query")

  _restaurantsLiveData.value = Resource.Loading

  restaurantSource //3
      .filter { restaurant ->
        if (query.isEmpty()) return@filter true //4
        restaurant.name.contains(query, true)//5
      }
      .showResults()
}

Here’s a breakdown:

1. Changes in the search query come through the queryObservable.
2. On every query change, you send those values emitted to filterSource(), a helper function that will filter the values.
3. This function receives the new query with which you filter your source of truth by returning either true or false.
4. If the query is empty, it doesn’t filter. In other words, it emits all values.
5. When the query is not empty, it compares each restaurant name with the query. If the query is part of the restaurant name, it emits the value. If the name is not, it skips that name.

Build and run. Tap the search icon, input Isl and notice your list is filtered with every letter you type.

With the filter operator, you can tap every element of your stream and sequentially and simply filter it with any conditional logic you see fit.

Next, you’ll learn about the debounce operator.

Using the Debounce Operator

Look at your Logcat and filter it to show logs from RestaurantViewModel. Notice how the filtering occurs with every letter you enter.

If you were using a remote server, this would have a significant impact on your app's performance. It could raise your server costs for unnecessary transactions. Ideally, you'll only call your data source when you're relatively sure the query you received is the one the user wants to search for.

This problem seems like a big headache, but RxJava has your back with debounce.

The debounce operator only emits an item from an Observable if a particular time has passed without emitting another item.

This isn't as confusing as it may seem. debounce lets through only the last item in a certain timespan. Open RestaurantViewModel.kt and replace setQueryListener() with:

fun setQueryListener(queryObservable: Observable<String>) {
  queryObservable
      .debounce(300, TimeUnit.MILLISECONDS)
      .observeOn(AndroidSchedulers.mainThread())
      .map(::filterSource)
      .subscribe()
}

All you did is add the debounce operator. Here, debounce only lets values through after a pause of 300 milliseconds. From the app perspective, this means the Observable waits to emit the last item, which will be the most recent query, until the user stops typing characters for at least 300 milliseconds.

As a result, you get considerable performance improvement with very little code.

Build and run. Again, search for Isl and try to write without stopping much between characters. In your Logcat, you'll see the Observable only let through and filtered one query as expected, Isl.

Without the help of RxJava, adding this performance-boosting functionality will be cumbersome.

It's time for the last operator, distinctUntilChanged.

Using the DistinctUntilChanged Operator

You'll use the last operator to resolve a small edge case. When you search for a specific query and quickly add and remove a letter, you might be fast enough to beat your debounce operator. If you do, you'll search and filter your list based on the same search query.

Build and run, then search for Isl. As soon as the list filters go on, add an a and remove it within your debounce 300 milliseconds time period.

Check your Logcat and see the list filters based on the same search query. Since that query already filtered the list, this search is unnecessary.

This is an easy fix with the distintctUntilChanged operator. This operator only lets through values that are different from their predecessors. Any sequential duplicated items are suppressed.

Open RestaurantViewModel.kt and replace setQueryListener with:

fun setQueryListener(queryObservable: Observable<String>) {
  queryObservable
      .debounce(300, TimeUnit.MILLISECONDS)
      .distinctUntilChanged()
      .observeOn(AndroidSchedulers.mainThread())
      .map(::filterSource)
      .subscribe()
}

Here you added the distintUntilChanged operator. Build and run, search for Isl and then add and remove the letter a within the 300 milliseconds debounce period. Check you Logcat and you'll see your data was only fetched and filtered once with the query Isl!

Where to Go From Here?

You can download the final version of this project using the Download Materials button at the top or bottom of this tutorial.

Congratulations on making it all the way through! Now you can work with filtering operators and understand their full potential. There are many ways of achieving the same result. The path you take depends on what you need and how you compose your operators.

If you want to continue your journey in RxJava, check out the ReactiveX Filtering Operators documentation. Another great resource is our awesome book, Reactive Programming with Kotlin.

If you have any questions, comments, or want to show off what other filtering operators you used in this project, feel free to join the discussion below!