5. Filtering Operators

Written by Alex Sullivan & Scott Gardner

Learning a new technology stack is a bit like building a skyscraper: You’ve got to build a solid foundation before you can kiss the sky. By now, you’ve established a solid RxJava foundation, and it’s time to start building up your knowledge base and skill set, one floor at a time.

This chapter will teach you about RxJava’s filtering operators that you can use to apply conditional constraints to next events, so that the subscriber only receives the elements it wants to deal with. If you’ve ever used the filter method in the Kotlin Standard Library, you’re already half way there. But if not, no worries; you’re going to be an expert at this filtering business by the end of this chapter.

Getting started

The starter project for this chapter is an IntelliJ project. Open it up and give it a build. You won’t see anything in the console yet.

Ignoring operators

Without further ado, you’re going to jump right in and look at some useful filtering operators in RxJava, beginning with ignoreElements. As depicted in the following marble diagram, ignoreElements will do just that: ignore next event elements. It will, however, allow through stop events, i.e., complete or error events. Allowing through stop events is usually implied in marble diagrams.

It’s just explicitly called out this time by the dashed line because that’s all ignoreElements will let through.

Note: Up until now you’ve seen marble diagrams used for observable types. The marble diagram shown here instead helps to visualize how operators work. The top line is the observable that is being subscribed to. The box represents the operator and its parameters, and the bottom line is the subscriber, or more specifically, what the subscriber will receive after the operator does its thing.

See one, now do one, by adding this example to your main function:

exampleOf("ignoreElements") {

  val subscriptions = CompositeDisposable()
  // 1
  val strikes = PublishSubject.create<String>()
  // 2
  subscriptions.add(
      strikes.ignoreElements() 
          // 3
          .subscribeBy {
            println("You’re out!")
          })
}

Here’s what you’re doing:

1. Create a strikes subject.
2. Subscribe to all strikes events, but ignore all next events by using ignoreElements.
3. Since this observable now has no elements, ignoreElements converts it into a Completable. There is no onNext in subscribeBy for a Completable.

ignoreElements is useful when you only want to be notified when an observable has terminated, via a complete or error event. Add this code to the example:

strikes.onNext("X")
strikes.onNext("X")
strikes.onNext("X")

Even though this batter can’t seem to hit the broad side of a barn and has clearly struck out, nothing is printed, because you’re ignoring all next events. It’s up to you to add a complete event to this subject in order to let the subscriber be notified. Add this code to do that:

strikes.onComplete()

Now, the subscriber will receive the complete event, and print that catchphrase no batter ever wants to hear.

--- Example of: ignoreElements ---
You’re out!

Note: If you don’t happen to know much about strikes, batters and the game of baseball in general, you can read up on those when you decide to take a little break from programming: https://simple.wikipedia.org/wiki/Baseball.

elementAt operator

There may be times when you only want to handle the nth (ordinal) element emitted by an observable, such as the third strike. For that you can use elementAt, which takes the index of the element you want to receive, and it ignores everything else.

In the marble diagram, elementAt is passed an index of 1, so it only allows through the second element. Remember: observables, just like lists, are zero-indexed.

Add this new example:

exampleOf("elementAt") {

  val subscriptions = CompositeDisposable()
  // 1
  val strikes = PublishSubject.create<String>()
  // 2
  subscriptions.add(
      strikes.elementAt(2) 
          // 3
          .subscribeBy(
              onSuccess = { println("You’re out!") }
          ))
}

Here’s the play-by-play:

1. You create a strikes subject.
2. You subscribe to the strikes observable, ignoring every element other than the third item.
3. Since this observable may not have a third item, elementAt returns a Maybe. Because it’s a Maybe will subscribe with onSuccess instead of onNext.

Now you can simply add new strikes onto the subject, and your subscription will take care of letting you know when the batter has struck out. Add this code inside the example block:

strikes.onNext("X")
strikes.onNext("X")
strikes.onNext("X")

“Hey batta, batta, batta, swing batta!”

Now you can build and run and imagine the game:

--- Example of: elementAt ---
You’re out!

filter operator

ignoreElements and elementAt are filtering elements emitted by an observable. When your filtering needs go beyond all or one, there’s filter. filter takes a predicate lambda, which it applies to each element, allowing through only those elements for which the predicate resolves to true.

Check out this marble diagram, where only 1 and 2 are let through, because the filter’s predicate only allows elements that are less than 3.

Add this example to your main function:

exampleOf("filter") {

  val subscriptions = CompositeDisposable()

  subscriptions.add(
      // 1
      Observable.fromIterable(
        listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
          // 2
          .filter { number ->
            number > 5
          }.subscribe {
            // 3
            println(it)
          })
}

From the top:

1. You create an observable of some predefined integers.
2. You use the filter operator to apply a conditional constraint to prevent any number less than five from getting through. filter takes a predicate that returns a Bool. Return true to let the element through or false to prevent it. filter will filter elements for the life of the subscription.
3. You subscribe and print out the elements that passed the filter predicate.

The result of applying this filter is that only numbers greater than five are printed:

--- Example of: filter ---
6
7
8
9
10

Skipping operators

It might be that you need to skip a certain number of elements. Consider observing a weather forecast, where maybe you don’t want to start receiving hourly forecast data until later in the day, because you’re stuck in a cubicle until then anyway. The skip operator allows you to ignore from the first to the number you pass as its parameter. All subsequent elements will then pass through.

This marble diagram shows skip being passed 2, so it ignores the first 2 elements.

Enter this new example in your main function:

exampleOf("skip") {
  val subscriptions = CompositeDisposable()

  subscriptions.add(
      // 1
      Observable.just("A", "B", "C", "D", "E", "F")
          // 2
          .skip(3)
          .subscribe {
            println(it)
          })
}

With this code, you:

1. Create an observable of letters.
2. Use skip to skip the first 3 elements and then subscribe to next events.

After skipping the first three elements, only D, E, and F are printed like so:

--- Example of: skip ---
D
E
F

skipWhile operator

There’s a small family of skip operators. Like filter, skipWhile lets you include a predicate to determine what should be skipped. However, unlike filter, which will filter elements for the life of the subscription, skipWhile will only skip up until something is not skipped, and then it will let everything else through from that point on. Also, with skipWhile, returning true will cause the element to be skipped, and returning false will let it through: it uses the return value in the opposite way to filter.

In this marble diagram, 1 is prevented because 1 % 2 equals 1, but then 2 is allowed through because it fails the predicate, and 3 (and everything else going forward) gets through because skipWhile is no longer skipping.

Add this new example to your main function:

exampleOf("skipWhile") {

  val subscriptions = CompositeDisposable()

  subscriptions.add(
      // 1
      Observable.just(2, 2, 3, 4)
          // 2
          .skipWhile { number ->
            number % 2 == 0
          }.subscribe {
            println(it)
          })

}

Here’s what you did:

1. Create an observable of integers.
2. Use skipWhile with a predicate that skips elements until an odd integer is emitted.

skipWhile only skips elements up until the first element is let through, and then all remaining elements are allowed through.

--- Example of: skipWhile ---
3
4

If you were developing an insurance claims app, you could use skipWhile to deny coverage until the deductible is met. If only the insurance industry were that straightforward here in the United States.

skipUntil operator

So far, the filtering has been based on some static condition. What if you wanted to dynamically filter elements based on some other observable? There are a couple of operators that you’ll learn about here that can do this. The first is skipUntil, which will keep skipping elements from the source observable (the one you’re subscribing to) until some other trigger observable emits.

In this marble diagram, skipUntil ignores elements emitted by the source observable (the top line) until the trigger observable (second line) emits a next event.

Then it stops skipping and lets everything through from that point on.

Add this example to see how skipUntil works in code:

exampleOf("skipUntil") {

  val subscriptions = CompositeDisposable()
  // 1
  val subject = PublishSubject.create<String>()
  val trigger = PublishSubject.create<String>()

  subscriptions.add(
      // 2
      subject.skipUntil(trigger)
          .subscribe {
            println(it)
          })
}

In this code, you:

1. Create a subject to model the data you want to work with, and another subject to model a trigger to change how you handle things in the first subject.
2. Use skipUntil, passing the trigger subject. When trigger emits, skipUntil will stop skipping.

Add a couple of next events onto subject:

subject.onNext("A")
subject.onNext("B")

Nothing is printed out, because you’re skipping. Now add a new next event onto trigger:

trigger.onNext("X")

Doing so causes skipUntil to stop skipping. From this point onward, all elements will be let through. Add another next event onto subject:

subject.onNext("C")

Sure enough, it’s printed out.

--- Example of: skipUntil ---
C

Taking operators

Taking is the opposite of skipping. When you want to only take certain elements, RxJava has you covered. The first taking operator you’ll learn about is take. As shown in this marble diagram, the result will take the first of the number of elements you specified and ignore everything that follows.

Add this example to your main function to explore the first of the take operators:

exampleOf("take") {
  val subscriptions = CompositeDisposable()

  subscriptions.add(
      // 1
      Observable.just(1, 2, 3, 4, 5, 6)
          // 2
          .take(3)
          .subscribe {
            println(it)
          })
}

Here’s what you did:

1. Create an observable of integers.
2. Take the first three elements using take.

What you take is what you get. The output this time is:

--- Example of: take ---
1
2
3

takeWhile operator

There’s also a takeWhile operator that works similarly to skipWhile, except you’re taking instead of skipping. takeWhile works like take, but uses a predicate instead of a number of next events, as in this marble diagram:

Enter this new example in your main function:

exampleOf("takeWhile") {
  val subscriptions = CompositeDisposable()

  subscriptions.add(
      // 1
      Observable.fromIterable(
        listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 1))
          // 2
          .takeWhile { number ->
            number < 5
          }.subscribe {
            println(it)
          })
}

From the top:

1. Create an observable of integers counting up from 1 to 10, and then finally emitting another 1 value.
2. Use the takeWhile operator and take any number that’s less than 5.

The output from the takeWhile example is:

--- Example of: takeWhile ---
1
2
3
4

You only receive integers that are less than five and came before any integer that was greater than 5. The 1 value at the end isn’t emitted because the takeWhile operator already hit a value greater than 5.

takeUntil operator

Like skipUntil, there’s also a takeUntil operator, shown in the next marble diagram. It takes from the source observable until the trigger observable emits an element.

Add this new example, which is just like the skipUntil example you created earlier:

exampleOf("takeUntil") {
  val subscriptions = CompositeDisposable()
  // 1
  val subject = PublishSubject.create<String>()
  val trigger = PublishSubject.create<String>()

  subscriptions.add(
      // 2
      subject.takeUntil(trigger)
          .subscribe {
            println(it)
          })
  // 3
  subject.onNext("1")
  subject.onNext("2")
}

Here’s what you did:

1. Create a primary subject and a trigger subject.
2. Use takeUntil, passing the trigger that will cause takeUntil to stop taking once it emits.
3. Add a couple of elements onto subject.

The console shows those elements, but takeUntil is in taking mode because the trigger did not emmit yet.

--- Example of: takeUntil ---
1
2

Now add an element onto trigger, followed by another element onto subject:

trigger.onNext("X")

subject.onNext("3")

The X stops the taking, so 3 is not allowed through and nothing more is printed.

Distinct operators

The next couple of operators you’re going to learn about let you prevent duplicate items one-after-another from getting through. As shown in this marble diagram, distinctUntilChanged only prevents duplicates that are right next to each other. The second 2 does not emit but second 1 gets through since it is a change relative to what came before it.

Distinct operators can be visualized like this:

Add this new example to your main function:

exampleOf("distinctUntilChanged") {
  val subscriptions = CompositeDisposable()

  subscriptions.add(
      // 1
      Observable.just("Dog", "Cat", "Cat", "Dog")
          // 2
          .distinctUntilChanged()
          .subscribe {
            println(it)
          })
}

What you’re doing, here:

1. Create an observable of our fluffy friends.
2. Use distinctUntilChanged to prevent sequential duplicates from getting through.

distinctUntilChanged only prevents contiguous duplicates. So the third element is prevented because it’s the same as the second, but the last item, a Dog, is allowed through, because it comes after a different pet (Cat).

The resulting printout only includes the first Dog, first Cat, and then the Dog at the end:

--- Example of: distinctUntilChanged ---
Dog
Cat
Dog

The default behavior of distinctUntilChanged uses the equals method to determine that two items are equal. That may not be what you want, so you can use a variant of distinctUntilChanged that accepts a predicate comparing two items that are emitted one after another.

In the following marble diagram, objects with a property named value are being compared for distinctness based on value:

Add this new example to your project to use the new version of distinctUntilChanged in a slightly more elaborate way:

exampleOf("distinctUntilChangedPredicate") {
  val subscriptions = CompositeDisposable()

  subscriptions.add(
      // 1
      Observable.just(
        "ABC", "BCD", "CDE", "FGH", "IJK", "JKL", "LMN")
          // 2
          .distinctUntilChanged { first, second ->
            // 3
            second.any { it in first }
          }
          // 4
          .subscribe {
            println(it)
          }
  )
}

From the top, you:

1. Create an observable of strings representing chunks of the English alphabet.
2. Use distinctUntilChanged(comparer: BiPredicate), which takes a lambda that receives each sequential pair of elements.
3. Return true if any character in the second string is also in the first string.
4. Subscribe and print out elements that are considered distinct based on the comparing logic you provided.

As a result, only distinct letters are printed in each pair of next events; that is, in each pair of strings, one does not contain any of the characters of the other:

--- Example of: distinctUntilChangedPredicate ---
ABC
FGH
IJK

So, this version of distinctUntilChanged is useful when you want to distinctly prevent duplicates for types that do not have a useful equals implementation.

Challenge

Challenge: Create a phone number lookup

Open the challenge starter project and have a look at what’s to be found inside!

Breaking down this challenge, you’ll need to use several filter operators. Here are the requirements, along with a suggested operator to use:

1. Phone numbers can’t begin with 0 — use skipWhile.
2. You an only input a single-digit number at a time; use filter to only allow elements that are less than 10.
3. This is limited to U.S. phone numbers, which are 10 digits, so take only the first 10 numbers inputted; use take and toList.

Review the setup code in the starter project. There’s a simple contacts dictionary:

val contacts = mapOf(
    "603-555-1212" to "Florent",
    "212-555-1212" to "Junior",
    "408-555-1212" to "Marin",
    "617-555-1212" to "Scott")

There’s a utility function that will return a formatted phone number for the list of 10 values you pass to it:

fun phoneNumberFrom(inputs: List<Int>): String {
  val phone = inputs.map { it.toString() }.toMutableList()
  phone.add(3, "-")
  phone.add(7, "-")
  return phone.joinToString("")
}

There’s a PublishSubject to start you off:

val input = PublishSubject.create<Int>()

And there’s a series of onNext calls to test that your solution works:

input.onNext(0)
input.onNext(603)

input.onNext(2)
input.onNext(1)

// Confirm that 7 results in "Contact not found", and then
// change to 2 and confirm that Junior is found
input.onNext(2)

"5551212".forEach {
  // Need toString() or else Char conversion is done
  input.onNext(it.toString().toInt())
}

input.onNext(9)

Because this challenge focuses on using the filter operators, here is code that you can use in the subscription’s next event handler. It takes the result from phoneNumberFrom and print out the contact if found or else "Contact not found":

if (contact != null) {
  println("Dialing $contact ($phone)...")
} else {
  println("Contact not found")
}

Add your code right below the comment // Add your code here.

Once you’ve implemented your solution, follow the instructions in the comment beginning // Confirm that 7 results in... to test that your solution works.

Key points

• Ignoring operators like ignoreElements, elementAt, and filter let you remove certain elements from an observable stream.
• Skipping operators let you skip certain elements and then begin emitting.
• Conversely, taking operators let you take certain elements and then stop emitting.
• Distinct operators let you prevent duplicates from being emitted back-to-back in an observable stream.

Where to go from here?

You’ve seen the theory behind filttering operators in an IntelliJ project. Next up, transfer that knowledge into a real Android app by going back to the Combinestagram photo collage app.