7. Transforming Operators

Written by Alex Sullivan & Scott Gardner

Before you decided to buy this book and commit to learning RxJava, you might have felt that RxJava was some esoteric library; elusive, yet strangely compelling you to master it. And maybe that reminds you of when you first started learning Android or Kotlin. Now that you’re up to Chapter 7, you’ve come to realize that RxJava isn’t magic. It’s a carefully constructed API that does a lot of heavy lifting for you and streamlines your code. You should be feeling good about what you’ve learned so far.

In this chapter, you’re going to learn about one of the most important categories of operators in RxJava: transforming operators. You’ll use transforming operators all the time, to prepare data coming from an Observable for use by your Subscriber. Once again, there are parallels between transforming operators in RxJava and the Kotlin standard library, such as map() and flatMap(). By the end of this chapter, you’ll be transforming everything!

Getting started

This chapter will use a normal IntelliJ project, so go ahead and open the starter project now.

Transforming elements

Observables emit elements individually, but you will frequently want to work with collections. One typical use case is when you’re emitting a list of items to show in a RecyclerView.

A convenient way to transform an Observable of individual elements into a list of all those elements is by using toList.

As depicted in this marble diagram, toList will convert an observable sequence of elements into a list of those elements, and emit a next event containing that array to the subscribers.

Add this new example to your project:

exampleOf("toList") {

  val subscriptions = CompositeDisposable()
  // 1
  val items = Observable.just("A", "B", "C")

  subscriptions.add(
      items
          // 2
          .toList()
          .subscribeBy {
            println(it)
          }
  )
}

Here’s what you just did:

1. Create an Observable of letters.
2. Use toList to transform the elements in a list.

A list of the letters is printed.

--- Example of: toList ---
[A, B, C]

map operator

RxJava’s map operator works just like Kotlin’s standard map function, except it operates on observables instead of a collection. In the marble diagram, map takes a lambda that multiplies each element by 2.

Add this new example to your project:

exampleOf("map") {

  val subscriptions = CompositeDisposable()

  subscriptions.add(
      // 1
      Observable.just("M", "C", "V", "I")
          // 2
          .map {
            // 3
            it.romanNumeralIntValue()
          }
          // 4
          .subscribeBy {
            println(it)
          })
}

Here’s the play-by-play:

1. You create an Observable of Roman numerals, in this case M which stands for 1000, C which stands for 100, V which stands for 5, and I which stands for 1.
2. You use map to transform the Observable, passing in a lambda.
3. You take each of the Roman numeral items emitted by the observable and then use a romanNumeralIntValue method to convert it into its corresponding integer value.
4. You subscribe to the Observable to print the transformed values.

Note: The romanNumeralIntValue method is defined in the SupportingCode.kt file. The implementation is pretty straightforward, but feel free to have a look if you are curious.

Go ahead and run the code. You should see the following output:

--- Example of: map ---
1000
100
5
1

Using the map operator, you have mapped each element of the original Observable to a new value as it passes through the stream.

Transforming inner observables

You may have wondered at some point, “How do I work with observables that are properties of observables?” Get ready to get your mind blown.

In the SupportCode.kt file in your project, add the following class which you’ll use in the upcoming examples:

class Student(val score: BehaviorSubject<Int>)

Student is a class which has a score property that is a BehaviorSubject<Int>. RxJava includes a few operators in the flatMap family that allow you to reach into an Observable and work with its observable properties. You’re going to learn how to use the two most common ones here.

Note: A heads up before you begin: These operators have elicited more than their fair share of questions (and groans and moans) from the RxJava’s newcomers. They may seem complex at first but you are going to walk through detailed explanations of each one. By the end of the section you’ll be ready to put these operators into action with confidence!

flatMap operator

The first one you’ll learn about is flatMap. The documentation for flatMap states that it “Projects each element of an observable sequence to an observable sequence and merges the resulting observable sequences into one observable sequence.” Makes perfect sense, right?

Not!

That description, and the following marble diagram, may feel a bit overwhelming at first. Read through the play-by-play explanation that follows, referring back to the marble diagram, and you’ll get it.

The easiest way to follow what’s happening in this marble diagram is to take each path from the source observable (the top line) all the way through to the target Observable. The target Observable is represented by the bottom line, and it delivers elements to the Subscriber. The source observable is a type of object that has a value property that itself is an observable of type Int. To put it another way, the source observable emits observables. The initial value of each emitted observable is the number of the object: O1’s initial value is 1, O2’s is 2, and O3’s is 3.

Starting with O1, flatMap receives the object and reaches in to access its value property and multiply it by 10. It then projects the transformed elements from O1 onto a new Observable. This is just what a regular map would do.

The first line below flatMap on the diagram is just for O1. That Observable is flattened down to the target Observable that will deliver elements to the Subscriber (the bottom line).

Later, O1’s value property changes to 4, which is not visually represented in the marble diagram (otherwise the diagram would become even more congested).

But the evidence that O1’s value has changed is that it is transformed to 40 and then projected onto the existing Observable for O1. As with the initial value, it is then flattened down to the target observable. This all happens in a time-linear fashion.

The next value in the source observable, O2, is received by flatMap. Now its initial value 2 is transformed to 20, projected onto a new observable for O2, and then flattened down to the target Observable. Later, O2’s value is changed to 5. It is transformed to 50, projected, and flattened to the target Observable.

Finally, O3 is received by flatMap, its initial value of 3 is transformed, projected, and flattened.

flatMap transforms and projects all the values from all the Observables that it receives. It then flattens them all down to a target Observable. Simple, isn’t it? Time to go hands-on with flatMap and really see how to use it. Add this example to your project:

exampleOf("flatMap") {

  val subscriptions = CompositeDisposable()
  // 1
  val ryan = Student(BehaviorSubject.createDefault(80))
  val charlotte = Student(BehaviorSubject.createDefault(90))
  // 2
  val student = PublishSubject.create<Student>()

  student
      // 3
      .flatMap { it.score }
      // 4
      .subscribe { println(it) }
      .addTo(subscriptions)
}

Here’s the play-by-play:

1. You create two instances of Student, ryan and charlotte.
2. You create a source subject of type Student.
3. You use flatMap to reach into the student subject and access its score. You don’t modify score in any way. Just pass it through.
4. You print out next event elements in the Subscription.

There’s nothing in the console, yet. Add this code to the example:

student.onNext(ryan)

ryan’s score is now printed out.

--- Example of: flatMap ---
80

Now change ryan’s score by adding this code to the example:

ryan.score.onNext(85)

ryan’s new score is printed.

--- Example of: flatMap ---
80
85

Next, add a different Student instance, charlotte, onto the source subject by adding the following code:

student.onNext(charlotte)

flatMap does its thing and charlotte’s score is printed.

--- Example of: flatMap ---
80
85
90

Here’s where it gets interesting. Change ryan’s score by adding this line of code:

ryan.score.onNext(95)

ryan’s new score is printed.

--- Example of: flatMap ---
80
85
90
95

This is because flatMap keeps up with each and every Observable it creates, one for each element added onto the source observable.

Now change charlotte’s score by adding the following code, just to verify that flatMap monitors both Observables and projects the changes:

charlotte.score.onNext(100)

Sure enough, her new score is printed out.

--- Example of: flatMap ---
80
85
90
95
100

To recap, flatMap keeps projecting changes from each Observable. There will be times when you want this behavior and there will be times when you only want to keep up with the latest element in the source observable. Luckily, RxJava has an operator just for that situation called switchMap.

switchMap operator

According to the documentation, switchMap: “Applies the given io.reactivex.functions.Function to each item emitted by a reactive source, where that function returns a reactive source, and emits the items emitted by the most recently projected of these reactive sources.”

So basically, switchMap takes a function which returns some type of reactive source (a Completable, Observable, Single and so on) and applies that function to each item emitted by some source observable. The observable returned by switchMap then emits only the items from whatever reactive source was the last emitted. Take a look at the following marble diagram:

The top line represents the source observable that emits three separate items - O1, O2, and O3.

O1 is received by switchMap, it transforms its value by a factor of 10, projects it onto a new observable for O1, and flattens it down to the target observable. Just like before.

But then switchMap receives O2 and does its thing, switching to O2’s observable because it’s now the latest. When O1 emits a value that is transformed to 40, that value does not get emitted by the target observable, since it has been switched to O2.

The process repeats when O3 is received by switchMap: it switches to the O3 stream and ignores the previous one (O2). So when O2 emits a value that is transformed to 50, the 50 is not emitted by the overall stream.

In summary, the result of using switchMap is that the target observable only receives elements from the latest source observable that has emitted. It’s ok if things are still confusing—flatMap and switchMap tend to be some of the hardest operators for people to understand. But another example will help clear things up!

Add the following example to your project, which is a clone of the previous example except for changing flatMap to switchMap:

exampleOf("switchMap") {

  val ryan = Student(BehaviorSubject.createDefault(80))
  val charlotte = Student(BehaviorSubject.createDefault(90))

  val student = PublishSubject.create<Student>()

  student
      .switchMap { it.score }
      .subscribe { println(it) }

  student.onNext(ryan)

  ryan.score.onNext(85)

  student.onNext(charlotte)

  ryan.score.onNext(95)

  charlotte.score.onNext(100)
}

Now run the example. You should see the following output:

--- Example of: switchMap---
80
85
90
100

The only thing that’s “missing” here compared to the flatMap example is that the last call to the ryan subject, i.e. ryan.score.onNext(95), isn’t being emitted. That’s because the charlotte subject has already emitted and now the switchMap only emits its values! Since charlotte is a BehaviorSubject it will immediately emit its latest value, which is 90 in this case.

So you may be wondering when would you use flatMap or switchMap? Probably the most common use case for using switchMap is with networking operations. You will go through examples of this later in the book, but for a simple example, imagine that you’re implementing a type-ahead search. As the user types each letter, e.g. k, o, t, l, i,n, you’ll want to execute a new search and ignore results from the previous one. switchMap is how you do that.

Observing events

There may be times when you want to convert an Observable into an Observable of its events. One typical scenario where this is useful is when you do not have control over an Observable that has Observable properties, and you want to handle error events to avoid terminating outer sequences. Don’t worry, it will get clearer in a couple of moments, just hang in there.

materialize operator

The materialize operator can do exactly that. It takes a normal Observable and turns it into an Observable that emits Notification objects that wrap the event type - whether it’s onNext, onComplete or onError.

Enter this new example into the project:

exampleOf("materialize/dematerialize") {

  val subscriptions = CompositeDisposable()

  val ryan = Student(BehaviorSubject.createDefault(80))
  val charlotte = Student(BehaviorSubject.createDefault(90))

  val student = BehaviorSubject.create<Student>(ryan)
}

This code should look pretty familiar—just like before you’re creating two new Student objects, ryan and charlotte, each of which contain a BehaviorSubject with an initial value. You’re then also creating a BehaviorSubject named student of type Student with the initial value of ryan.

Similar to the previous two examples, you want to subscribe to the inner score property of Student. Add this code to the example:

// 1
val studentScore = student
    .switchMap { it.score }
// 2
subscriptions.add(studentScore
    .subscribe {
      println(it)
    })
// 3
ryan.score.onNext(85)

ryan.score.onError(RuntimeException("Error!"))

ryan.score.onNext(90)
// 4
student.onNext(charlotte)

Continuing this example, you:

1. Create a studentScore observable using switchMap to reach into the student Observable and access its scoreObservable property.
2. Subscribe and print out each score as it’s emitted.
3. Add a score, an error, and another score onto the current student.
4. Add the second student charlotte onto the student Observable. Because you used switchMap, this will switch to this new student and subscribe to her score.

The error you added is unhandled. As a result, the studentScore observable terminates, and you get a very gnarly stack trace in the console.

--- Example of: materialize and dematerialize ---
80
85
io.reactivex.exceptions.OnErrorNotImplementedException: Error!

Using the materialize operator, you can wrap each event emitted by an Observable in a Notification.

In the marble diagram, Int elements emitted by an observable are transformed to Notification<Int> values when emitted.

Change the studentScore implementation to the following:

val studentScore = student
    .switchMap { it.score.materialize() }

If you check the type of studentScore you’ll see it is now an Observable<Notification<Int>>. And the Subscription to it now emits notifications. The error still causes the studentScore to terminate, but not the outer student Observable.

This way, when you switch to the new student, its score is successfully received and printed.

--- Example of: materialize/dematerialize ---
OnNextNotification[80]
OnNextNotification[85]
OnErrorNotification[java.lang.RuntimeException: Error!]
OnNextNotification[90]

However, now you’re dealing with Notifications, not the Int elements of the original Observables.

dematerialize operator

That’s where dematerialize comes in. It will convert a materialized Observable back into its original form.

In the marble diagram, Notification<Int> values are transformed back into Int elements.

Change the Subscription in the example to the following:

studentScore
    // 1
    .filter {
      if (it.error != null) {
        println(it.error)
        false
      } else {
        true
      }
    }
    // 2
    .dematerialize { it }
    .subscribe {
      println(it)
    }
    .addTo(subscriptions)

Wrapping things up:

1. You print and filter out any errors.
2. You use dematerialize to return the studentScore Observable to its original form, emitting scores and stop events, not notifications of scores and stop events. Since this Observable is emitting Notifications directly you’re simply returning it to dematerialize.

As a result, your student Observable is protected by errors on its inner score Observable. The error is printed and ryan’s score Observable is terminated, so adding a new score onto him does nothing.

But when you add charlotte onto the student subject, her score is printed.

--- Example of: materialize/dematerialize ---
80
85
java.lang.RuntimeException: Error!
90

Challenge

Challenge: Sending alpha-numeric characters

In Chapter 5’s challenge, you created a phone number lookup using filtering operators. You added the code necessary to look up a contact based on a 10-digit number entered by the user.

Your goal for this challenge is to modify the implementation to be able to take letters as well, and convert them to their corresponding number based on a standard phone keypad (abc is 2, def is 3, and so on).

The starter project includes a helper lambda to do the conversion:

val convert: (String) -> Int = { value ->
  val number = try {
    value.toInt()
  } catch (e: NumberFormatException) {
    val keyMap = mapOf(
        "abc" to 2, "def" to 3, "ghi" to 4, "jkl" to 5,
        "mno" to 6, "pqrs" to 7, "tuv" to 8, "wxyz" to 9)

    keyMap.filter { it.key.contains(value.toLowerCase()) }
        .map { it.value }.first()
  }

  if (number < 10) {
    number
  } else {
    // RxJava 2 does not allow null in stream, so return
    // sentinel value
    sentinel
  }
}

Since RxJava 3 doesn’t allow null in an Observable stream, we will use a sentinel value of -1 to mark a number that exceeds 10 digits.

And there are lambdas to format and “dial” the contact if found (really, just print it out):

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

val dial: (String) -> String = { phone ->
  val contact = contacts[phone]
  if (contact != null) {
    "Dialing $contact ($phone)..."
  } else {
    "Contact not found"
  }
}

These lambda values allow you to move the logic out of the Subscription, where it really doesn’t belong. So what’s left to do then? You’ll use multiple maps to perform each transformation along the way. You’ll use skipWhile just like you did in Chapter 5 to skip 0s at the beginning.

The starter project also includes code to test your solution. Just add your solution right below the comment // Add your code here.

Key points

• Transforming operators let you transform observable items from their original type to another type or value.
• You can use toList to turn a normal observable into an observable that emits a single list.
• The map operator will transform individual elements in an observable to some other value or type.
• You can use flatMap to flatten an observable stream of observables into one stream of items.
• Similarly, switchMap will also flatten a stream of observables, but this time only listening to the observable in the source that has most recently emitted.
• You use materialize to make observables emit notifications of events rather than the events themselves, and dematerialize to transform from the notification type back to the original type.

Where to go from here?

Just like for the earlier chapters on filtering operators, you’ll want to gain experience using transforming operators in a real Android app project. That’s up next!