9. Combining Operators

Written by Florent Pillet

In earlier chapters, you learned how to create, filter and transform observable sequences. RxSwift filtering and transformation operators behave much like Swift’s standard collection operators. You got a glimpse into the true power of RxSwift 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 Swift collection operators. They help combine elements from asynchronous sequences, just as you do with Swift arrays.

Getting started

For this chapter, you will be using an Xcode Playground set up with the basic building blocks you need to go through the chapter tasks.

To get started, open the macOS Terminal application (found in your Mac‘s Applications > Utilities folder), navigate to the current chapter’s starter project folder, then run the bootstrap script like so:

$ ./bootstrap.sh

You will again use the example(of:) construct to wrap your code in distinct blocks. Remember to show the Debug Area in Xcode (under the View and Debug Area menus), as this is where playground print(_:) statements display their output.

RxSwift 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

The first and most obvious need 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.

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

Add the following code to the playground:

example(of: "startWith") {
  // 1
  let numbers = Observable.of(2, 3, 4)

  // 2
  let observable = numbers.startWith(1)
  _ = observable.subscribe(onNext: { value in
    print(value)
  })
}

The startWith(_:) operator prefixes an observable sequence with the given initial value. This value must be of the same type as the observable elements.

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

1. Create a sequence of numbers.
2. Create a sequence starting with the value 1, then continue with the original sequence of numbers.

Don’t get fooled by the position of the startWith(_:) operator! Although you chain it to the numbers sequence, the observable it creates emits the initial value, followed by the values from the numbers sequence.

Look at the debug area in the playground to confirm this:

——— Example of: startWith ———
1
2
3
4

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

Note: We purposely don‘t keep the Disposable returned by the subscription, because the Observable here immediately completes after emitting its two items. Therefore, our subscription will automatically end. You will use this form in further examples when it‘s safe to do so.

As it turns out, startWith(_:) is the simple variant of the more general concat family of operators. Your initial value is a sequence of one element, to which RxSwift appends the sequence that startWith(_:) chains to. The Observable.concat(_:) static function chains two sequences.

Have a look:

Add this code to the playground:

example(of: "Observable.concat") {
  // 1
  let first = Observable.of(1, 2, 3)
  let second = Observable.of(4, 5, 6)

  // 2
  let observable = Observable.concat([first, second])

  observable.subscribe(onNext: { value in
    print(value)
  })
}

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

The Observable.concat(_:) static method takes either an ordered collection of observables (i.e. an array), or a variadic list of observables. It subscribes to the first sequence of the collection, relays its elements until it completes, then moves to the next one. The process repeats until all the observables in the collection have been used. If at any point an inner observable emits an error, the concatenated observable in turn emits the error and terminates.

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

example(of: "concat") {
  let germanCities = Observable.of("Berlin", "Münich", "Frankfurt")
  let spanishCities = Observable.of("Madrid", "Barcelona", "Valencia")

  let observable = germanCities.concat(spanishCities)
  _ = observable.subscribe(onNext: { value in
    print(value)
  })
}

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(_:). Check the playground 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 Swift compiler knows when one sequence is an Observable<String> and the other an Observable<Int> so it will not allow you to mix them.

A final operator of interest is concatMap(_:), closely related to flatMap(_:) which you learned about in Chapter 7, “Transforming Operators”. The closure you pass to concatMap(_:) returns an Observable sequence which the operator first subscribes to, then relays the values it emits into the resulting sequence. concatMap(_:) guarantees that each sequence the closure produces runs to completion before subscribing to the next one. It‘s a handy way to guarantee sequential order while giving you the power of flatMap(_:).

Try it in the playground:

example(of: "concatMap") {
  // 1
  let sequences = [
    "German cities": Observable.of("Berlin", "Münich", "Frankfurt"),
    "Spanish cities": Observable.of("Madrid", "Barcelona", "Valencia")
  ]

  // 2
  let observable = Observable.of("German cities", "Spanish cities")
    .concatMap { country in sequences[country] ?? .empty() }

  // 3
  _ = observable.subscribe(onNext: { string in
      print(string)
    })
}

This example:

1. Prepares two sequences producing German and Spanish city names.
2. Has a sequence emit country names, each in turn mapping to a sequence emitting city names for this country.
3. Outputs the full sequence for a given country before starting to consider the next one.

Now that you know how to append sequences together, it’s time to move on and combine elements from multiple sequences.

Merging

RxSwift offers several ways to combine sequences. The easiest to start with is merge. Can you picture what it does from the diagram below?

Switch to the playground; your task is to add a new example(of:) block, and prepare two subjects to which you can push values. You learned about Subject in Chapter 3, “Subjects”.

example(of: "merge") {
  // 1
  let left = PublishSubject<String>()
  let right = PublishSubject<String>()

Next, create a source observable of observables — it’s like Inception! To keep things simple, make it a fixed list of your two subjects:

  // 2
  let source = Observable.of(left.asObservable(), right.asObservable())

Next, create a merge observable from the two subjects, as well as a subscription to print the values it emits:

  // 3
  let observable = source.merge()
  _ = observable.subscribe(onNext: { value in
    print(value)
  })

Then you need to randomly pick and push values to either observable. The loop uses up all values from leftValues and rightValues arrays then exits.

  // 4
  var leftValues = ["Berlin", "Munich", "Frankfurt"]
  var rightValues = ["Madrid", "Barcelona", "Valencia"]
  repeat {
      switch Bool.random() {
      case true where !leftValues.isEmpty:
          left.onNext("Left:  " + leftValues.removeFirst())
      case false where !rightValues.isEmpty:
          right.onNext("Right: " + rightValues.removeFirst())
      default:
          break
      }
  } while !leftValues.isEmpty || !rightValues.isEmpty

One last bit before you’re done is calling onCompleted on both left and right PublishSubjects:

  // 5
  left.onCompleted()
  right.onCompleted()
}

Whoa, that was a lot of code, so if you don’t see any warnings, pat yourself on the shoulder — good job!

Run the code (it might have run automatically after you saved your work) and look at the debug output. Results will be different each time you run this code, but they should look similar to this:

——— Example of: merge ———
Right: Madrid
Left:  Berlin
Right: Barcelona
Right: Valencia
Left:  Munich
Left:  Frankfürt

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 RxSwift, 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.

Take a second to look at the code. Notice that merge() takes a source observable, which itself emits observables sequences of the element type. This means that you could send a lot of sequences for merge() to subscribe to!

To limit the number of sequences subscribed to at once, you can use merge(maxConcurrent:). This variant keeps subscribing to incoming sequences until it reaches the maxConcurrent limit. After that, it puts incoming observables in a queue. It will subscribe to them in order, as soon as one of the active sequences completes.

Note: You might end up using this limiting variant less often than merge() itself. Keep it in mind, though, as it can be handy in resource-intensive situations. You could use it in scenarios such as when making a lot of network requests to limit the number of concurrent outgoing connections.

Combining elements

An essential group of operators in RxSwift is the combineLatest family. They combine values from several sequences:

Every time one of the inner (combined) sequences emits a value, it calls a closure you provide. You receive the last value emitted by 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 a few examples.

First, create two subjects to push values to:

example(of: "combineLatest") {
  let left = PublishSubject<String>()
  let right = PublishSubject<String>()

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

  // 1
  let observable = Observable.combineLatest(left, right) {
    lastLeft, lastRight in
    "\(lastLeft) \(lastRight)"
  }

  _ = observable.subscribe(onNext: { value in
    print(value)
  })

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

  // 2
  print("> Sending a value to Left")
  left.onNext("Hello,")
  print("> Sending a value to Right")
  right.onNext("world")
  print("> Sending another value to Right")
  right.onNext("RxSwift")
  print("> Sending another value to Left")
  left.onNext("Have a good day,")

Finally, don’t forget to complete both of your subjects and close the example(of:) trailing closure:

  left.onCompleted()
  right.onCompleted()
}

Run the complete example from above. You’ll see three sentences show up in the debug output of the Playground, plus information about when you send values to the combined observable. These help make it clear as to when your closure receives values.

A few notable points about this example:

1. You combine observables using a closure 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 the combined observable emits is the return type of the closure. In practice, this means you can combine sequences of heterogeneous types. It is one of the rare core operators that permit this, the other being withLatestFrom(_:) you‘ll learn about in a short while.
2. Nothing happens until each of the observables emit one value. After that, each time one emits a new value, the closure receives the latest value of each of the observables and produces its result.

Note: Remember that combineLatest(_:_:resultSelector:) waits for all its observables to emit one element before starting to call your closure. It’s a frequent source of confusion and a good opportunity to use the startWith(_:) operator to provide an initial value for the sequences which may not immediately delive a value.

Like the map(_:) operator covered in Chapter 7, “Transforming Operators,” combineLatest(_:_:resultSelector:) creates an observable whose type is the closure return type. This is a great opportunity 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:

let observable = Observable
  .combineLatest(left, right) { ($0, $1) }
  .filter { !$0.0.isEmpty }

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.

Let’s look at another example. Add this code to your playground:

example(of: "combine user choice and value") {
  let choice: Observable<DateFormatter.Style> = Observable.of(.short, .long)
  let dates = Observable.of(Date())

  let observable = Observable.combineLatest(choice, dates) {
    format, when -> String in
    let formatter = DateFormatter()
    formatter.dateStyle = format
    return formatter.string(from: when)
  }

  _ = observable.subscribe(onNext: { value in
    print(value)
  })
}

This example demonstrates automatic updates of on-screen values when the user settings change. Think about all the manual updates you’ll remove with such patterns!

A final variant of the combineLatest family takes a collection of observables and a combining closure, which receives latest values in an array. Since it’s a collection, all observables carry elements of the same type.

Since it’s less flexible than the multiple parameter variants, it is seldom-used but still handy to know about The string observable in your first combineLatest(_:_:resultSelector:) example could be rewritten as:

  // 1
  let observable = Observable.combineLatest([left, right]) {
    strings in strings.joined(separator: " ")
  }

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.

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

To get started, create a Weather enum and a couple of observables:

example(of: "zip") {
  enum Weather {
    case cloudy
    case sunny
  }
  let left: Observable<Weather> = Observable.of(.sunny, .cloudy, .cloudy, .sunny)
  let right = Observable.of("Lisbon", "Copenhagen", "London", "Madrid", "Vienna")

Then create a zipped observable of both sources. Note that you’re using the zip(_:_:resultSelector:) variant. Use the shorter form as shown below, with the closure after the last parenthesis, for improved readability.

  let observable = Observable.zip(left, right) { weather, city in
    return "It's \(weather) in \(city)"
  }
  _ = observable.subscribe(onNext: { value in
    print(value)
  })
}

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(_:_:resultSelector:) did for you:

• Subscribed to the observables you provided.
• Waited for each to emit a new value.
• Called your closure 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 pair each next value of each observable at the same logical position (1st with 1st, 2nd with 2nd, etc.). This implies that if no next value from one of the inner observables is available at the next logical position (i.e. because it completed, like in the example above), zip won‘t emit anything anymore. This is called indexed sequencing, which is a way to walk sequences in lockstep. But while zip may stop emitting values early, it won‘t itself complete until all its inner observables complete, making sure each can complete its work.

Note: Swift also has a zip(_:_:) collection operator. It creates a new collection of tuples with items from both collections. But this is its only implementation. RxSwift offers variants for two to eight observables, plus a variant for collections, like combineLatest does.

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. RxSwift has you covered with powerful operators that will make your life easier. Well, your coding life at least!

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 playground:

example(of: "withLatestFrom") {
  // 1
  let button = PublishSubject<Void>()
  let textField = PublishSubject<String>()

  // 2
  let observable = button.withLatestFrom(textField)
  _ = observable.subscribe(onNext: { value in
    print(value)
  })

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

This example simulates a text field and a button. In Chapter 12, “Beginning RxCocoa,” you’ll learn about RxCocoa, a framework that helps bind your UI with RxSwift. The last two lines are duplicated on purpose!

Run this example and you’ll see this output in the debug area:

Paris
Paris

Let’s go through what you just did:

1. Create two subjects simulating button taps and text field input. Since the button carries no real data, you can use Void as an element type.
2. When button emits a value, ignore it but instead emit the latest value received from the simulated text field.
3. Simulate successive inputs to the text field, which is done by the two successive button taps.

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.

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 trigger. If no new data arrived, sample(_:) won’t emit anything.

Try it in the playground. Replace withLatestFrom(_:) with sample(_:):

  // 2
  let observable = textField.sample(button)

Notice that "Paris" now prints only once! This is because the text field didn‘t emit a new value between your two fake button taps. You could have achieved the same behavior by adding a distinctUntilChanged() to the withLatestFrom(_:) observable, but 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 — RxSwift has operators for this!

Switches

RxSwift comes with two main so-called “switching” operators: amb(_:) and switchLatest(). They both allow you to produce an observable sequence by switching between the events of the combined or source sequences. This allows you to decide which sequence’s events will the subscriber receive at runtime.

Let’s look at amb(_:) first. Think of “amb” as in “ambiguous”.

Add this code to the playground:

example(of: "amb") {
  let left = PublishSubject<String>()
  let right = PublishSubject<String>()

  // 1
  let observable = left.amb(right)
  _ = observable.subscribe(onNext: { value in
    print(value)
  })

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

  left.onCompleted()
  right.onCompleted()
}

You’ll notice that the debug output only shows items from the left subject. Here’s what you did:

1. Create an observable which resolves ambiguity between left and right.
2. Have both observables send data.

The amb(_:) operator subscribes to the left and right observables. It waits for any of them to emit an element, then unsubscribes 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, such as connecting to redundant servers and sticking with the one that responds first.

A more popular option is the switchLatest() operator:

To try it out, first create three subjects and a source subject. You’ll push observable sequences to this one.

example(of: "switchLatest") {
  // 1
  let one = PublishSubject<String>()
  let two = PublishSubject<String>()
  let three = PublishSubject<String>()

  let source = PublishSubject<Observable<String>>()

Next, create an observable with the switchLatest() operator and print its output.

  // 2
  let observable = source.switchLatest()
  let disposable = observable.subscribe(onNext: { value in
    print(value)
  })

Start feeding the source with observables, and feed observables with values.

  // 3
  source.onNext(one)
  one.onNext("Some text from sequence one")
  two.onNext("Some text from sequence two")

  source.onNext(two)
  two.onNext("More text from sequence two")
  one.onNext("and also from sequence one")

  source.onNext(three)
  two.onNext("Why don't you see me?")
  one.onNext("I'm alone, help me")
  three.onNext("Hey it's three. I win.")

  source.onNext(one)
  one.onNext("Nope. It's me, one!")

Finally dispose the subscription when you‘re done.

  disposable.dispose()
}

Note: It can be difficult to form a mental model of an observable of observables. Don’t worry; you’ll get used to it. Practice is key to a fluid understanding of sequences. Don’t hesitate to review the examples as your experience grows! You’ll learn more about putting this to good use in the next chapter.

The previous code produces this output:

——— Example of: switchLatest ———
Some text from sequence one
More text from sequence two
Hey it's three. I win.
Nope. It's me, one!

Notice the few output lines. Your subscription only prints items from the latest sequence pushed to the source observable. This is the purpose of switchLatest().

Note: Did you notice any similarity between switchLatest() and another operator? You learned about its cousin flatMapLatest(_:) in Chapter 7, “Transforming Operators”. They do pretty much the same thing: flatMapLatest maps the latest value to an observable, then subscribes to it. It keeps only the latest subscription active, just like switchLatest.

Combining elements within a sequence

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

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

To get started, add this code to the playground:

example(of: "reduce") {
  let source = Observable.of(1, 3, 5, 7, 9)

  // 1
  let observable = source.reduce(0, accumulator: +)
  _ = observable.subscribe(onNext: { value in
    print(value)
  })
}

This is much like what you’d do with Swift collections, but with observable sequences. The code above uses a shortcut form (using the + operator) to accumulate values. This by itself is not very self-explanatory. To get a grasp on how it works, replace the observable creation above with the following code:

  // 1
  let observable = source.reduce(0) { summary, newValue in
    return summary + newValue
  }

The 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 closure to produce a new summary. 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.

A close relative to reduce(_:_:) is the scan(_:accumulator:) operator. Can you spot the difference in the schema below, comparing to the last one above?

Add some code to the playground to experiment:

example(of: "scan") {
  let source = Observable.of(1, 3, 5, 7, 9)

  let observable = source.scan(0, accumulator: +)
  _ = observable.subscribe(onNext: { value in
    print(value)
  })
}

Now 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 closure.

Each time the source observable emits an element, scan(_:accumulator:) invokes your closure. It passes the running value along with the new element, and the closure returns the new accumulated value.

Like reduce(_:_:), the resulting observable type is the closure return type.

The range of use cases for scan(_:accumulator:) is quite large; you can use it to compute running totals, statistics, states and so on.

Encapsulating state information within a scan(_:accumulator:) observable is a good idea; you won’t need to use local variables, and it goes away when the source observable completes. You’ll see a couple of neat examples of scan in action in Chapter 20, “RxGesture.”

Challenge

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

Challenge: The zip case

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

Take the code from the scan(_:accumulator:) 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 show two possible implementations. Can you find them both?