RxJava Filtering Operators
In this tutorial, you will learn about a group of powerful RxJava operators that will allow you to work only with the data you require at a certain step of the data transformation process: the RxJava Filtering Operators. By Andres Torres.
Sign up/Sign in
With a free Kodeco account you can download source code, track your progress, bookmark, personalise your learner profile and more!
Create accountAlready a member of Kodeco? Sign in
Sign up/Sign in
With a free Kodeco account you can download source code, track your progress, bookmark, personalise your learner profile and more!
Create accountAlready a member of Kodeco? Sign in
Sign up/Sign in
With a free Kodeco account you can download source code, track your progress, bookmark, personalise your learner profile and more!
Create accountAlready a member of Kodeco? Sign in
Contents
RxJava Filtering Operators
15 mins
- Getting Started
- What is RxJava?
- What are RxJava Filtering Operators?
- Using the Take Operator
- The Take Operator Variations
- Using the SkipWhile Operator
- The SkipWhile Operator Variations
- Using the IgnoreElements Operator
- Using the Filter Operator
- Using the Debounce Operator
- Using the DistinctUntilChanged Operator
- Where to Go From Here?
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!
To start learning RxJava read this Reactive Programming with RxAndroid tutorial.
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:
- You create an extension function for showing the results for any
Observable
emittingRestaurant
values. - As soon as the
Observable
completes, you create a list with all the emitted values. - Then you subscribe to the corresponding scheduler.
- You wrap the resulting list in a
Resource
object with a type ofSuccess
. - Next, you send all the
Resource
values to the correspondingLiveData
. - Finally, you add the resulting disposable to a
CompositeDisposable
which is disposed of when theViewModel
is removed. This avoids memory leaks. Remember to always clean up after yourself. :]
restaurantsSource
, is the direct result from the query to your Room database. The second one, restaurantSource
, is the previous observable but its elements are flatten so you can work directly with each element to assure the stream completes.
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 anObservable
. This operator needs anObservable
that completes to know which elements are last. Otherwise, it won’t emit them. -
takeUntil
: Emits items until a secondaryObservable
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 thetake
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 anObservable
that completes and lets it know which elements are last. It skips the final n items emmitted by anObservable
. -
skipUntil
: Skips items until a secondaryObservable
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:
- Any
Observable
useing this extension also triggers aCompletable
. - The completable subscribes to the corresponding scheduler.
- Then the completable observes the stream of data on the main thread provided by RxAndroid.
- Everytime the stream of data completes, the completable sends a
Resource
of typeSuccess
to the activity through the correspondingLiveData
. - Finally, the completable adds the resulting disposable to a
CompositeDisposable
that disposes when theViewModel
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:
- Changes in the search query come through the
queryObservable
. - On every query change, you send those values emitted to
filterSource()
, a helper function that will filter the values. - This function receives the new query with which you filter your source of truth by returning either true or false.
- If the query is empty, it doesn’t filter. In other words, it emits all values.
- 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.
throttle
. The throttle
operator emits only one item from a group of emitted values within periodic time intervals. Here are two for reference:
-
throttleFirst
: Emits the first element within periodic time intervals. -
throttleLast
: Emits the last element within periodic time intervals.
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!
distinct
operator. Instead of suppressing sequential duplicates, as distinctUntilChanged
does, it suppresses duplicates anywhere on your data stream.
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!