Operators are a key component of RxSwift and the ReactiveX API. They help make reactive programming powerful and flexible. In the previous episode, we explored operators that transform and filter the values emitted by an observable. This episode zooms in on operators that combine multiple observables into one observable. Combining observables is a common pattern and it can greatly simplify the implementation of a range of tedious tasks.
Merging Observables
One of the simplest operators to combine multiple observables into one observable is the merge operator. Because it doesn't operate on a single observable, it is defined as a static method. Let's give it a try.
We invoke the merge(_:) method on the Observable class. The merge(_:) method accepts an array of observables and returns a new observable. The observables that are passed to the merge(_:) method are not affected by the operation. We covered that in the previous episode.
We define two observables, intervalA and intervalB. The first observable emits an integer every second. The second observable emits an integer every three seconds. Notice that we use the map operator to convert the integers emitted by the observables to strings. We prefix the integers with A and B. Why we do that becomes clear in a moment.
import RxSwift
import Foundation
let intervalA = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).map { "A \($0)" }
let intervalB = Observable<Int>.interval(.seconds(3), scheduler: MainScheduler.instance).map { "B \($0)" }
To merge the observables into a new observable, we invoke the merge(_:) method on the Observable class, passing in the source observables as an array.
import RxSwift
import Foundation
let intervalA = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).map { "A \($0)" }
let intervalB = Observable<Int>.interval(.seconds(3), scheduler: MainScheduler.instance).map { "B \($0)" }
let observable = Observable.merge([intervalA, intervalB])
To see the result of the merge operator, we subscribe to the resulting observable by invoking the subscribe(onNext:onError:onCompleted:onDisposed:) method. We are only interested in the next events. In the onNext event handler, we print the values emitted by the observable to the console. Because we are working in a playground, there is no need to add the subscription to a dispose bag.
import RxSwift
import Foundation
let intervalA = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).map { "A \($0)" }
let intervalB = Observable<Int>.interval(.seconds(3), scheduler: MainScheduler.instance).map { "B \($0)" }
let observable = Observable.merge([intervalA, intervalB])
observable
.subscribe(onNext: { (value) in
print(value)
})
Run the contents of the playground. The output in the console confirms that the resulting observable emits the values emitted by both observables.
A 0
A 1
A 2
B 0
A 3
A 4
A 5
B 1
A 6
A 7
A 8
B 2
A 9
The output shows why we converted the integers emitted by the source observables to strings. By prefixing the original values with the letters A and B, it's easier to understand which values originate from which source observable.
Let me show you the marble diagram for this example to avoid any confusion. It's worth pointing out that the resulting observable emits two values at the same time every three seconds.
Because the elements of an observable are of a particular type, it isn't possible to merge observables of different types. If we remove the map operator from the first observable, then intervalA produces integers and it's no longer possible to merge intervalA and intervalB.
import RxSwift
import Foundation
let intervalA = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
let intervalB = Observable<Int>.interval(.seconds(3), scheduler: MainScheduler.instance).map { "B \($0)" }
let observable = Observable.merge([intervalA, intervalB])
observable
.subscribe(onNext: { (value) in
print(value)
})
Combining Observables
Another operator I use frequently is combineLatest. Its name reveals its function. It takes the last value of two or more observables to create a new observable. Let me illustrate this with an example.
We create two observables. The first observable emits a sequence of strings, a, b, and c. The second observable emits an integer every second. This should look familiar.
import RxSwift
import Foundation
let strings = Observable<String>.of("a", "b", "c")
let integers = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
Like the merge operator, the combineLatest operator is defined as a static method. We invoke the combineLatest method on the Observable class. RxSwift defines a range of variations of the combineLatest method, depending on the number of observables you want to combine. We invoke the combineLatest variation that accepts two observables, passing in references to strings and integers.
import RxSwift
import Foundation
let strings = Observable<String>.of("a", "b", "c")
let integers = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
let observable = Observable.combineLatest(strings, integers)
observable
.subscribe(onNext: { (value) in
print(value)
})
We subscribe to the resulting observable to find out which values it emits. Run the contents of the playground to see the result.
("c", 0)
("c", 1)
("c", 2)
("c", 3)
("c", 4)
("c", 5)
The result is quite interesting and it may be surprising if you are new to the combineLatest operator. Let's start with the type of the values emitted by the resulting observable, (String, Int). The resulting observable wraps the values emitted by the source observables in a tuple.
It may surprise you that the first and second values emitted by the strings observable are not emitted by the resulting observable. Let's take a look at the marble diagram of this example to better understand what is happening.
The strings observable immediately emits three values. The integers observable emits its first value, 0, after one second. What you need to understand about the combineLatest operator is that the resulting observable emits its first value after each source observable has emitted a value. That is why the first next event emits a tuple that includes the last value of the strings observable and the first value of the integers observable. The marble diagram illustrates that.
The combineLatest method optionally accepts a closure as its last argument. The closure accepts the values emitted by the source observables and returns a type of your choosing. You can see it as a built-in map operator. The closure is executed every time one of the source observables emits a value. Let's use string interpolation to return a string from the closure.
import RxSwift
import Foundation
let strings = Observable<String>.of("a", "b", "c")
let integers = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
let observable = Observable.combineLatest(strings, integers) { (string, integer) -> String in
string + " \(integer)"
}
observable
.subscribe(onNext: { (value) in
print(value)
})
The resulting observable no longer emits tuples. It now emits values of type String.
c 0
c 1
c 2
c 3
c 4
c 5
Zipping Observables
The zip operator is in some ways similar to the combineLatest operator, but there is a key difference. Like the combineLatest operator, the zip operator can be used to combine observables and it starts emitting values after each source observable has emitted a value. The difference is that the zip operator emits a value every time each source observable emits a value.
Let me illustrate this with two observables that emit integers at regular time intervals. The first observable emits an integer every second. The second observable emits an integer every three seconds.
We invoke the zip method on the Observable class, passing in the source observables. Subscribe to the resulting observable and run the contents of the playground.
import RxSwift
import Foundation
let integers1 = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
let integers2 = Observable<Int>.interval(.seconds(3), scheduler: MainScheduler.instance)
let observable = Observable.zip(integers1, integers2)
observable
.subscribe(onNext: { (value) in
print(value)
})
The first time I used the zip operator I was a bit surprised and confused so don't worry if the output doesn't make sense.
(0, 0)
(1, 1)
(2, 2)
(3, 3)
(4, 4)
(5, 5)
There are a few details I want to point out. Notice that the resulting observable emits a value every three seconds. The second source observable, integers2, shows the same behavior. Even though the first source observable emits a value every second, the values emitted by the resulting observable suggest that the source observables are synchronized in some way. Let's take a look at the marble diagram to clear up any confusion.
The zip operator combines two or more observables. What's unique about the zip operator is that the values it emits are combinations of the source observables. This simply means that the first value is the combination of the first item of the first source observable and the first item of second source observable, the second value is the combination of the second item of first source observable and the second item of second source observable, and so on.
Combining Operators
I would like to end this episode with a more advanced example to illustrate the power of operators. We start with an observable that emits a number of integers. I would like to create an observable that emits the difference between two subsequent values. This is surprisingly simple using the zip operator and the skip operator.
import RxSwift
import Foundation
let integers = Observable.of(1, 5, 3, 6, 8)
Remember that the zip operator combines the values of multiple observables. This means that we need another observable. The idea is simple. We use the skip operator to create an observable that is always one value ahead of the original observable because it skips the first value emitted by the original observable. We pass both observables to the zip method.
import RxSwift
import Foundation
let integers = Observable.of(1, 5, 3, 6, 8)
let observable = Observable.zip(integers, integers.skip(1))
We subscribe to the resulting observable and calculate the difference between the values emitted by the source observables. Run the contents of the playground to see the result.
import RxSwift
import Foundation
let integers = Observable.of(1, 5, 3, 6, 8)
let observable = Observable.zip(integers, integers.skip(1))
observable
.subscribe(onNext: { (value) in
print(abs(value.0 - value.1))
})
4
2
3
2
Let's bring up the marble diagram for this example. Notice that the source observables emit the same values with the exception of the first value.
This solution is useful if you need to populate a table or collection view. When you update a table or collection view, you need to compute the differences between the old data and the new data. By using this pattern, you can extract the information you need to decide which rows or items to insert, delete, update, or move.
What's Next?
The past two episodes have given you a glimpse of the power and flexibility of operators. Observables and operators are at the heart of RxSwift so it's important that you understand what we have covered so far.