9. Combining Operators

Written by Alex Sullivan & Florent Pillet

In earlier chapters, you learned how to create, filter and transform Observable sequences. RxJava filtering and transformation operators behave much like Kotlin’s standard collection operators. You got a glimpse into the true power of RxJava with flatMap, the workhorse operator that lets you perform a lot of tasks with very little code.

This chapter will show you several different ways to assemble sequences, and how to combine the data within each sequence. Some operators you’ll work with are very similar to Kotlin collection functions. They help combine elements from asynchronous sequences, just as you do with Kotlin lists.

Getting started

This chapter uses IntelliJ to demonstrate some of the concepts. It also uses the exampleOf method you’ve become so familiar with. Open the starter project and run the Main.kt file. It’s empty, so you won’t see any output other than a “process finished” message in the run tab.

RxJava is all about working with and mastering asynchronous sequences. But you’ll often need to make order out of chaos! There is a lot you can accomplish by combining Observables.

Prefixing and concatenating

One of the more obvious needs when working with Observables is to guarantee that an observer receives an initial value. There are situations where you’ll need the “current state” first. Good use cases for this are “current location” and “network connectivity status.” These are some Observables you’ll want to prefix with the current state.

Using startWith

The diagram below should make it clear what this operator does:

Add the following code to the main() function:

exampleOf("startWith") {

  val subscriptions = CompositeDisposable()
  // 1
  val missingNumbers = Observable.just(3, 4, 5)
  // 2
  val completeSet =
    missingNumbers.startWithIterable(listOf(1, 2))

  completeSet
    .subscribe { number ->
      println(number)
    }
    .addTo(subscriptions)
}

The startWithIterable() and startWithItem() operators prefix an Observable sequence with the given initial value. This value must be of the same type as the Observable elements. For startWithItem(), this is a single item, while startWithIterable() can be a list of initial items that will the stream will emit individually.

Here’s what’s going on in the code above:

1. Create an Observable of numbers.
2. Create an Observable starting with the missing values 1 and 2, then continue with the original sequence of numbers.

Don’t get fooled by the position of the startWithIterable() operator! Although you chain it to the missingNumbers stream, the Observable it creates emits the initial values, followed by the values from the original missingNumbers Observable.

Run the code and look at the run area in the project to confirm this:

--- Example of: startWith ---
1
2
3
4
5

This is a handy tool you’ll use in many situations. It fits in well with the deterministic nature of RxJava and guarantees observers they’ll get an initial value right away, and any updates later.

Using concat

As it turns out, the startWith operators are a simple variant of the more general concat family of operators. Your initial value is a stream of one or more elements, to which RxJava appends the sequence that startWith chains to. The Observable.concat static function chains two sequences.

Have a look:

Add this code to the main() function:

exampleOf("concat") {

  val subscriptions = CompositeDisposable()
  // 1
  val first = Observable.just(1, 2, 3)
  val second = Observable.just(4, 5, 6)
  // 2
  Observable.concat(first, second)
    .subscribe { number ->
    println(number)
  }
  .addTo(subscriptions)
}

Written this way, the concatenation order is more obvious to the untrained reader than when using one of the startWith operators. Run the example to see elements from the first stream: 1 2 3, followed by elements of the second stream: 4 5 6.

The Observable.concat static function takes a vararg number of Observables (i.e. an array). It subscribes to the first Observable of the collection, relays its elements until it completes, then moves to the next one. The process repeats until it uses all the Observables in the collection. If at any point an inner Observable emits an error, the concatenated Observable in turns emits the error and terminates.

Using concatWith

Another way to append sequences together is the concatWith operator (an instance method of Observable, not a class method). Add this code to the function:

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

  val germanCities =
    Observable.just("Berlin", "Münich", "Frankfurt")
  val spanishCities =
    Observable.just("Madrid", "Barcelona", "Valencia")

  germanCities
    .concatWith(spanishCities)
    .subscribe { number ->
      println(number)
    }
    .addTo(subscriptions)
}

This variant applies to an existing Observable. It waits for the source Observable to complete, then subscribes to the parameter Observable. Aside from instantiation, it works just like Observable.concat(). Run the code and check the output; you’ll see a list of German cities followed by a list of Spanish cities.

Note: Observable sequences are strongly typed. You can only concatenate sequences whose elements are of the same type!

If you try to concatenate sequences of different types, brace yourself for compiler errors. The Kotlin compiler knows when one sequence is an Observable<String> and the other an Observable<Int>, so it will not allow you to mix them up.

Using concatMap

A final operator of interest is concatMap, closely related to flatMap which you learned about in Chapter 7, “Transforming Operators.” The lambda you pass to flatMap returns an Observable sequence which is subscribed to, and the emitted Observables are all merged. concatMap guarantees that each sequence produced by the lambda will run to completion before the next is subscribed to. concatMap is therefore a handy way to guarantee sequential order.

Try it in the project:

exampleOf("concatMap") {
  val subscriptions = CompositeDisposable()
  // 1
  val countries = Observable.just("Germany", "Spain")
  // 2
  val observable = countries
    .concatMap {
    when (it) {
      "Germany" ->
        Observable.just("Berlin", "Münich", "Frankfurt")
      "Spain" ->
        Observable.just("Madrid", "Barcelona", "Valencia")
      else -> Observable.empty<String>()
      }
    }
  // 3
  observable
    .subscribe { city ->
      println(city)
    }
    .addTo(subscriptions)
}

This example:

1. Creates an Observable of two country names.
2. Uses concatMap to produce another Observable depending on what country name it receives.
3. Outputs the full sequence of cities for a given country before starting to consider the next one.

Run the project. You should see this output:

--- Example of: concatMap ---
Berlin
Münich
Frankfurt
Madrid
Barcelona
Valencia

The German cities are all printed out, followed by the cities from Spain.

Now that you know how to append sequences together using the various concatenating operators, it’s time to move on to combining elements from multiple sequences.

Merging

RxJava offers several ways to combine sequences. The easiest to start with is merge.

Using merge

Can you picture what merge does from the diagram below?

Your next task is to add a new exampleOf block, and prepare two subjects to which you can push values. You learned about Subject in Chapter 3, “Subjects”. Start by adding this block:

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

  val left = PublishSubject.create<Int>()
  val right = PublishSubject.create<Int>()
}

You’ll now merge left and right together. Add the following to the example:

Observable.merge(left, right)
  .subscribe {
    println(it)
  }
  .addTo(subscriptions)

Now it’s time to start emitting items. Add the following below the above code, within the example block:

left.onNext(0)
left.onNext(1)
right.onNext(3)
left.onNext(4)
right.onNext(5)
right.onNext(6)

You emit 0 and 1 from the left subject, then 3 from the right subject, and so on. If you were using concat here, you’d expect to see the following (assuming you called onComplete on both of the subjects): 0, 1, 4, 3, 5, 6. But since you’re using merge, you see the following:

--- Example of: merge ---
0
1
3
4
5
6

Merge emits the items in the order that they come in. Pretty handy, right?

A merge() Observable subscribes to each of the sequences it receives and emits the elements as soon as they arrive — there’s no predefined order.

You may be wondering when and how merge() completes. Good question! As with everything in RxJava, the rules are well-defined:

• merge() completes after its source sequence completes and all inner sequences have completed.
• The order in which the inner sequences complete is irrelevant.
• If any of the sequences emit an error, the merge() Observable immediately relays the error, then terminates.

Using mergeWith

Just like for concat and concatWith, there’s also a mergeWith method you can use instead of the statically resolved Observable.merge method. Add the following example:

exampleOf("mergeWith") {

  val subscriptions = CompositeDisposable()

  val germanCities = PublishSubject.create<String>()
  val spanishCities = PublishSubject.create<String>()

  germanCities.mergeWith(spanishCities)
    .subscribe {
      println(it)
    }
    .addTo(subscriptions)
}

You’re again using city names, this time via the mergeWith operator.

Now add the following:

germanCities.onNext("Frankfurt")
germanCities.onNext("Berlin")
spanishCities.onNext("Madrid")
germanCities.onNext("Münich")
spanishCities.onNext("Barcelona")
spanishCities.onNext("Valencia")

Just like before you’re sending cities through on the different subjects in a mixed manner.

Run the project and you should see the following output:

--- Example of: mergeWith ---
Frankfurt
Berlin
Madrid
Münich
Barcelona
Valencia

The cities are received in the same order they are emitted by the merged subjects.

Combining elements

Using combineLatest

An essential operator in RxJava is the combineLatest operator. It combines values from several sequences:

Every time one of the inner (combined) sequences emits a value, it calls a lambda you provide. You receive the last value from each of the inner sequences. This has many concrete applications, such as observing several text fields at once and combining their values, watching the status of multiple sources, and so on.

Does this sound complicated? It’s actually quite simple! You’ll break it down by working through an example.

First, create two subjects to push values to. Add this example to your main() function:

exampleOf("combineLatest") {

  val subscriptions = CompositeDisposable()

  val left = PublishSubject.create<String>()
  val right = PublishSubject.create<String>()
}

Next, create an Observable that combines the latest value from both sources. Don’t worry; you’ll understand how the code works once you’ve finished adding everything together:

Observables
  .combineLatest(left, right) { leftString, rightString ->
    "$leftString $rightString"
}.subscribe {
  println(it)
}.addTo(subscriptions)

Now add the following code to start pushing values to the Observables:

left.onNext("Hello")
right.onNext("World")
left.onNext("It’s nice to")
right.onNext("be here!")
left.onNext("Actually, it’s super great to")

Run the complete example from above. You’ll see four sentences show up in the output of the project:

--- Example of: combineLatest ---
Hello World
It’s nice to World
It’s nice to be here!
Actually, it’s super great to be here!

A few notable points about this example:

1. You combine Observables using a lambda receiving the latest value of each sequence as arguments. In this example, the combination is the concatenated string of both left and right values. It could be anything else that you need, as the type of the elements emitted by the combined Observable is the return type of the lambda.
2. In practice, this means you can combine sequences of heterogeneous types. combineLatest is the only core operator that permits using Observables of differing types.
3. Nothing happens until each of the combined Observables emits one value. After that, each time one of the combined observables emits a new value, the lambda receives the latest value of each of the Observables and produces its element.

Note: Remember that combineLatest waits for all its Observables to emit one element before starting to call your lambda. It’s a frequent source of confusion! It’s also a good opportunity to use the startWith operator to provide an initial value for the sequences, which could take time to update. Like the map operator covered in Chapter 7, “Transforming Operators”, combineLatest creates an Observable whose type is the lambda return type. You can use this to switch to a new type alongside a chain of operators!

A common pattern is to combine values to a tuple then pass them down the chain. For example, you’ll often want to combine values and then call filter on them like so:

val observable = Observables
  .combineLatest(left, right) {
    leftString: String, rightString: String ->
   
    leftString to rightString
  }
  .filter { !it.first.isEmpty() }

One other interesting thing here is that you’re actually using the Observables.combineLatest method exposed by RxKotlin here, not the one exposed by RxJava. RxKotlin provides several convenience methods that make them easier to call from Kotlin. For example, if you didn’t have RxKotlin, the combineLatest call from the example would instead have to look like this:

Observable.combineLatest<String, String, String>(left, right,
      BiFunction { leftString, rightString ->
  "$leftString $rightString"
})

There are several variants in the combineLatest family of operators. They take between two and eight Observable sequences as parameters. As mentioned above, sequences don’t need to have the same element type.

Note: Last but not least, combineLatest completes only when the last of its inner sequences completes. Before that, it keeps sending combined values. If some sequences terminate, it uses the last value emitted to combine with new values from other sequences.

Using zip

Another combination operator is the zip family of operators. Like the combineLatest family, it comes in several variants:

Add a new example:

exampleOf("zip") {

  val subscriptions = CompositeDisposable()

  val left = PublishSubject.create<String>()
  val right = PublishSubject.create<String>()
}

Then create a zipped Observable of both sources. Note that you’re again using the RxKotlin version of the zip method. You can tell because it’s namespaced with Observables rather than Observable:

Observables.zip(left, right) { weather, city ->
  "It’s $weather in $city"
}.subscribe {
  println(it)
}.addTo(subscriptions)

Finally, feed some values into your subjects:

left.onNext("sunny")
right.onNext("Lisbon")
left.onNext("cloudy")
right.onNext("Copenhagen")
left.onNext("cloudy")
right.onNext("London")
left.onNext("sunny")
right.onNext("Madrid")
right.onNext("Vienna")

Run the code and check the output:

--- Example of: zip ---
It’s sunny in Lisbon
It’s cloudy in Copenhagen
It’s cloudy in London
It’s sunny in Madrid

Here’s what zip did for you:

• Subscribed to the Observables you provided.
• Waited for each to emit a new value.
• Called your lambda with both new values.

Did you notice how Vienna didn’t show up in the output? Why is that?

The explanation lies in the way zip operators work. They wait until each of the inner Observables emits a new value. If one of them completes, zip completes as well. It doesn’t wait until all of the inner Observables are done! This is called indexed sequencing, which is a way to walk though sequences in lockstep.

Note: Kotlin also has a zip collection operator. It creates a new collection of pairs with items from both collections.

Triggers

Apps have diverse needs and must manage multiple input sources. You’ll often need to accept input from several Observables at once. Some will simply trigger actions in your code, while others will provide data. RxJava has you covered with powerful operators that will make your life easier. Well, your coding life at least!

Using withLatestFrom

You’ll first look at withLatestFrom. Often overlooked by beginners, it’s a useful companion tool when dealing with user interfaces, among other things.

Add this code to the main() function. You may need to import withLatestFrom using io.reactivex.rxkotlin.withLatestFrom:

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

  // 1
  val button = PublishSubject.create<Unit>()
  val editText = PublishSubject.create<String>()

  // 2
  button.withLatestFrom(editText) { _: Unit, value: String ->
    value
  }.subscribe {
    println(it)
  }.addTo(subscriptions)

  // 3
  editText.onNext("Par")
  editText.onNext("Pari")
  editText.onNext("Paris")
  button.onNext(Unit)
  button.onNext(Unit)
}

This example simulates an Android EditText and Button.

Run this example and you’ll see this output:

--- Example of: withLatestFrom ---
Paris
Paris

Let’s go through what you just did:

1. Create two subjects simulating button presses and edit text input. Since the button carries no real data, you can use Unit as an element type.
2. When button emits a value, ignore it but instead emit the latest value received from the simulated EditText. The Button is acting as a trigger for getting values from the EditText.
3. Simulate successive inputs to the EditText, with values that are then emitted by the two successive button presses.

Simple and straightforward! withLatestFrom is useful in all situations where you want the current (latest) value emitted from an Observable, but only when a particular trigger occurs.

Using sample

A close relative to withLatestFrom is the sample operator.

It does nearly the same thing with just one variation: each time the trigger Observable emits a value, sample emits the latest value from the “other” Observable, but only if it arrived since the last “tick”. If no new data arrived, sample won’t emit anything.

Try it in the project. Duplicate the previous example of withLatestFrom, using sample instead:

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

  val button = PublishSubject.create<Unit>()
  val editText = PublishSubject.create<String>()

  editText.sample(button)
    .subscribe {
      println(it)
    }.addTo(subscriptions)

  editText.onNext("Par")
  editText.onNext("Pari")
  editText.onNext("Paris")
  button.onNext(Unit)
  button.onNext(Unit)
}

Run the project.

Notice that "Paris" now prints only once! This is because no new value was emitted by the text field between your two fake button presses. You could have achieved the same behavior by adding a distinctUntilChanged to the withLatestFrom Observable, but the smallest possible operator chains are the Zen of Rx™.

Note: Don’t forget that withLatestFrom takes the data observable as a parameter, while sample takes the trigger observable as a parameter. This can easily be a source of mistakes — so be careful!

Waiting for triggers is a great help when doing UI work. In some cases your “trigger” may come in the form of a sequence of observables (I know, it’s Inception once again). Or maybe you want to wait on a pair of observables and only keep one. No matter — RxJava has operators for this!

Switches

Using amb

RxJava comes with one main so-called “switching” operator: amb. It allows you to produce an Observable sequence by switching between the events of the combined source sequences. This allows you to decide which sequence’s events the subscriber will receive at runtime.

Think of “amb” as in “ambiguous”.

Add this code to the project:

exampleOf("amb") {

  val subscriptions = CompositeDisposable()

  val left = PublishSubject.create<String>()
  val right = PublishSubject.create<String>()

  // 1
  left.ambWith(right)
    .subscribe {
      println(it)
    }
    .addTo(subscriptions)

  // 2
  left.onNext("Lisbon")
  right.onNext("Copenhagen")
  left.onNext("London")
  left.onNext("Madrid")
  right.onNext("Vienna")
}

If you run the project, you’ll notice that the output only shows items from the left subject. Here’s what you did:

1. Create an Observable using ambWith which resolves ambiguity between left and right.
2. Have both Observables send data.

The ambWith operator combines the left and right Observables. It waits for any of them to emit an element, then unsubscribes subscriptions from the other one. After that, it only relays elements from the first active Observable. It really does draw its name from the term ambiguous: at first, you don’t know which sequence you’re interested in, and want to decide only when one fires.

This operator is often overlooked. It has a few select practical applications, like connecting to redundant servers and sticking with the one that responds first.

Combining elements within a sequence

All cooks know that the more you reduce, the tastier your sauce will be. Although not aimed at chefs, RxJava has the tools to reduce your sauce to its most flavorful components!

Using reduce

Through your coding adventures in Kotlin, you may already know about its reduce collection operator. If you don’t, here’s a great opportunity to learn about it, as this knowledge applies to pure Kotlin collections as well.

To get started, add this code to the project:

exampleOf("reduce") {

  val subscriptions = CompositeDisposable()

  val source = Observable.just(1, 3, 5, 7, 9)
  source
    .reduce(0) { a, b -> a + b }
    .subscribeBy(onSuccess = {
      println(it)
    })
    .addTo(subscriptions)
}

This is much like what you’d do with Kotlin collections, but instead with Observable sequences. The code above uses a lambda to add two items together. Run the code and see this reflected in the result:

--- Example of: reduce ---
25

The reduce operator “accumulates” a summary value. It starts with the initial value you provide (in this example, you start with 0). Each time the source Observable emits an item, reduce calls your lambda to produce a new summary by combining the current value with the newly emitted value per the lambda. When the source Observable completes, reduce emits the summary value, then completes.

Note: reduce produces its summary (accumulated) value only when the source Observable completes. Applying this operator to sequences that never complete won’t emit anything. This is a frequent source of confusion and hidden problems.

Using scan

A close relative to reduce is the scan operator. Can you spot the difference in the diagram below, comparing to the last one above?

Add some code to the project to experiment:

exampleOf("scan") {

  val subscriptions = CompositeDisposable()

  val source = Observable.just(1, 3, 5, 7, 9)

  source
    .scan(0) { a, b -> a + b }
    .subscribe {
      println(it)
    }
    .addTo(subscriptions)
}

Now run it and look at the output:

--- Example of: scan ---
1
4
9
16
25

You get one output value per input value. As you may have guessed, this value is the running total accumulated by the lambda. Each time the source Observable emits an element, scan invokes your lambda. It passes the running value along with the new element, and the lambda returns the new accumulated value. Like reduce, the resulting Observable type is the lambda return type.

The range of use cases for scan is quite large; you can use it to compute running totals, statistics, states and so on. Encapsulating state information within a scan Observable is a good idea; you won’t need to use local variables, and it goes away when the source Observable completes.

Challenge: The zip case

You learned a great deal about many combining operators in this chapter. But there is so much more to learn (and more fun to be had) about sequence combination!

You’ve learned about the zip family of operators that let you go through sequences in lockstep — it’s time to start using them.

Take the code from the scan example above and improve it so as to display both the current value and the running total at the same time.

There are several ways to do this — and not necessarily with zip. Bonus points if you can find more than one method.

The solutions to this challenge, found in the project files for this chapter, show two possible implementations. Can you find them both?

Key points

• You can prepend or append Observable sequences to one another using operators like startWith, concatWith, and concatMap.
• The merge family of operators lets you merge sequences together so that items are received in the order that they are emitted.
• The combineLatest operator lets you combine heterogeneous observables into a type that gets emitted each time one of the inner observables emits.
• The zip operators emit only when each of the inner Observables have all emitted a new value, called indexed sequencing; the overall Observable completes when any of the inner Observables complete.
• In combined sequences, if an inner sequence emits an error, then generally the overall Observable emits the error and the sequence terminates.
• Triggering operators like withLatestFrom and sample let you limit the emitting of elements to only when certain triggering events occur.
• The amb or “ambiguous” operator lets you switch between multiple Observables by sticking to the first one that is active.
• The reduce and scan operators let you combine the elements in a sequence based on an input lambda; reduce only emits the final value when it receives the complete event, whereas scan emits intermediate accumulated values.

Where to go from here?

Having been introduced to combining operators, in the next chapter you’ll see them in action in an Android app. The app project will retrieve data from a NASA API that you will combine in various ways. Despite being Earth-based data, it’s sure to be out of this world!