#flatmap

My team and I recently discovered a bug in one of our projects, and the culprit turned out to be the FlatMap operator—or rather, our misuse of it. I don’t know why, but FlatMap was a recurring source of confusion. (And the official Rx docs didn’t really shed much light on the subject.) Fortunately for us, we eventually gained clarity around the FlatMap operator using a real-life analogy that I thought would be worth sharing:

FlatMap is like the combining or flattening of commits pushed by a growing team of developers on a project.

Think of a team of developers on a project that uses a Continuous Integration (CI) service to build each pushed commit. The CI is interested in all of the commits pushed by all of the developers, including newly added ones, on the project.

FlatMap Analogy. Animation by Deyu Wang

We can dig into this more with code—in this case, Swift (and RxSwift). First, let’s define the objects we’ll need.

class Project { private let developerSubject = PublishSubject<developer>() var developerStream: Observable<developer> { return developerSubject.asObservable() } func addDeveloper(_ developer: Developer) { developerSubject.onNext(developer) } func stop() { developerSubject.onCompleted() } } class Developer { private let commitSubject = PublishSubject<commit>() let name: String init(_ name: String) { self.name = name } func startCoding() -> Observable<commit> { return commitSubject.asObservable() } func stopCoding() { commitSubject.onCompleted() } // Helper to externally simulate coding activity. func pushCommit(_ hash: String) { commitSubject.onNext(Commit(author: name, hash: hash)) } } struct Commit { let author: String let hash: String } class CI: ObserverType { func on(_ event: Event<commit>) { switch event { case .next(let commit): print("CI is building \(commit).") case .completed: print("CI stopped.") case .error(let error): print("CI errored: \(error).") } } }

Type Declarations

Now let’s instantiate things and hook it all up.

let project = Project() let jim = Developer("Jim") let anna = Developer("Anna") let bob = Developer("Bob") let ci = CI() project.developerStream .flatMap { developer -> Observable<commit> in print("\(developer.name) started coding...") return developer.startCoding() } .subscribe(ci)

Setup Objects

We flatMap the developerStream onto each developer’s commits. The resulting stream is subscribed to by the CI.

Looking closer, in the flatMap closure, for each developer emitted on developerStream, we return the developer’s stream of commits via the startCoding() function. This allows the flatMap to observe the commits emitted by all developers and then flatten them into a single output stream. The CI subscribes to this output stream so it can build each commit.

Let’s play with this to see what happens when we start adding developers and pushing commits. The trailing comments are the output from the print()s.

project.addDeveloper(jim) // Jim started coding... jim.pushCommit("1") // CI is building Commit(author: "Jim", hash: "1"). project.addDeveloper(anna) // Anna started coding... jim.pushCommit("2") // CI is building Commit(author: "Jim", hash: "2"). anna.pushCommit("3") // CI is building Commit(author: "Anna", hash: "3"). jim.pushCommit("4") // CI is building Commit(author: "Jim", hash: "4").

Normal Operation

Notice how even after Anna is added to the project, Jim remains “active” and the CI continues to see Jim’s additional commits. Jim’s commit stream returned from startCoding() doesn’t get replaced by Anna’s. Further, the order in which each developer is added doesn’t matter. The CI only sees the commits in the sequence they are pushed. The CI doesn’t care about the developers themselves, it only cares about their commits.

Taking a closer look, our flatMap subscribes to the commits from each developer received from developerStream. Even when new developers are received, those subscriptions continue to live on. This enables the commits from both Jim and Anna to be combined and flattened into a single output stream.

Wait — how is this different from **Merge?**

Instead of FlatMap, we could of course use the Merge operator to combine commits from multiple developers into a single stream. If the project’s developers are known and fixed, this would be fine, as Merge takes a static list of observables.

Observable.merge([jim.startCoding(), anna.startCoding()]).subscribe(ci)

However, if Bob came along to join the project at a later point, we’d have to handle that somehow. With FlatMap, since we subscribe to new developers being added, it’s handled for us already. Bob’s commits would automatically be taken into account and added to the flatMap’s output stream. Thus, we could say that Merge is for a static list of observables, while FlatMap is for a dynamic list of observables.

So, that’s most of it. But let’s get a more complete (Rx joke) understanding of how FlatMap works by exploring complete and error events.

Completions

It’s easy to start things, hard to complete them.

This saying is very true for FlatMap. In a nutshell, the CI will receive the completed event once all “active” observables in the chain have completed.

Let’s look at an example where, after some normal operation, only the project stops. Will the CI stop?

project.addDeveloper(jim) // Jim started coding... project.stop() project.addDeveloper(bob) bob.pushCommit("1") jim.pushCommit("2") // CI is building Commit(author: "Jim", hash: "2").

Completing the Project (Parent Observable)

No, the CI doesn’t stop. After the project stops, adding Bob has no effect because developerStream has already completed so he’s never “activated” in the flatMap. Thus, when Bob pushes a commit, the CI doesn’t see it. However, existing project members like Jim are free to continue pounding away and pushing commits which do get built by the CI.

Now let’s examine when, instead of the project stopping, every developer on the project stops. Will the CI stop?

project.addDeveloper(jim) // Jim started coding... project.addDeveloper(anna) // Anna started coding... jim.stopCoding() anna.stopCoding() jim.pushCommit("1") project.addDeveloper(bob) // Bob started coding... bob.pushCommit("2") // CI is building Commit(author: "Bob", hash: "2").

Completing the Developers (Child Observables)

No, the CI doesn’t stop. As shown, when all the project’s developers stop, it affects neither the CI nor the project. Bob is still able to join the project afterwards and push commits which the CI builds.

The only way to stop the CI’s subscription is to stop the project and all existing developers:

project.addDeveloper(jim) // Jim started coding... project.addDeveloper(anna) // Anna started coding... project.stop() jim.stopCoding() anna.stopCoding() // CI stopped. project.addDeveloper(bob) bob.pushCommit("1") jim.pushCommit("2")

Completing the CI’s Subscription

It’s worth noting here that if no developers are ever added to the project, calling project.stop() would also stop the CI.

Errors

error events work as expected. When either the project or any added developer errors, the CI will receive the error event and the whole chain stops working.

Conclusion

Now that we know how FlatMap actually behaves, we’re able to use it with more confidence and in the right places.

The easiest way to play with this code (or any RxSwift code) on your own is to use the RxSwift repo’s playground. To get a closer look at things, including when the isDisposed event is being emitted, you may want to add this code to the playground along with some debug operators.

Acknowledgements