RxSwift course - flatMap, flatMapLatest and async tasks

Working with async tasks using RxSwift is easy and straightforward. For me, the hardest thing was to grasp the concept of the flatMap operator.

FlatMap - general idea

First, let us focus on flatMap what we can use on collections. It is converting an array of arrays to one flat array:

[[1, 2, 3], [4, 5, 6], [7, 8, 9]] will be transformed to [1, 2, 3, 4, 5, 6, 7, 8, 9] - flattened.

FlatMap - RxSwift

RxSwift flatMap does a very similar thing, it flattens a series of observables into one observable. Think of the observable sequence as an array of events, it will take an array of observables (a.k.a array of events) and deliver it as one array of events.

Consider the following code:

let disposeBag = DisposeBag()
// 1
func createAsyncTask(for input: Int) -> Observable<String> {
    return Observable.create { sink in
        for i in 1...4  {
            DispatchQueue.main.asyncAfter(deadline: .now() + Double(i)) {
                sink.onNext("task for input \(input) element \(i)")
            }
        }
        return Disposables.create()
    }
}

let subject = PublishSubject<Int>()

// 2
subject.flatMap {
    createAsyncTask(for: $0)
}.subscribe(onNext: {
    print($0)
}).disposed(by: disposeBag)

for i in 1...4 {
    DispatchQueue.main.asyncAfter(deadline: .now() + Double(i)) {
        subject.onNext(i)
    }
}
// 3
RunLoop.main.run()

Creating observable
Observable has an additional factory method - it gives you a way to provide subscribe method, it will have one parameter - sink that you can use to send events. In my case, it will periodically send onNext events with a string value.
Observable sequence
Now we are ready to use flatMap. Every time subject emits a value, code will create a new observable sequence using the method we created above. It will be added to flatMaps inner observables - it will then flatten them down into one stream of events.
Infinite execution
Our code will run for a longer time and shouldn't end just after the last line of code - RunLoop.main.run() will add our program to the run loop. This is a console application, not an iOS app, so we have to do it manually.

Run the code, and you will see similar output:

I added colors to better see what is going on. We start as expected and the first observable is emitting strings. Then, input 2 comes, a new observable is created and starts emitting values, then 3 and 4. FlatMap is getting all these observables and combining its events into one stream of events as they come, no matter which inner observable is emitting the value.

I know this can be confusing at first glance, take your time, experiment with the code, see what you will get as an output. After grasping this concept, working with RxSwift is a blast!

FlatMapLatest

Most of the time, we care only for the newest observable elements being emitted. Imagine this - we have a search function in our code, every time user puts the next letter into the search box, we make calls to API for the results. This means we want to get results only for the newest search query. This is where flatMapLatest comes to play, It will flatten results, but only forward events from the latest observable created:

let disposeBag = DisposeBag()

func createAsyncTask(for input: Int) -> Observable<String> {
    return Observable.create { sink in
        for i in 1...4  {
            DispatchQueue.main.asyncAfter(deadline: .now() + Double(i)) {
                sink.onNext("task for input \(input) element \(i)")
            }
        }
        return Disposables.create()
    }
}

let subject = PublishSubject<Int>()


subject.flatMapLatest {
    createAsyncTask(for: $0)
}.subscribe(onNext: {
    print($0)
}).disposed(by: disposeBag)

for i in 0...3 {
    DispatchQueue.main.asyncAfter(deadline: .now() + 2.0 * Double(i)) {
        subject.onNext(i)
    }
}

RunLoop.main.run()

And the output: