Making Sense of FlatMap: An Rx Story

We’ve been using Rx for a while now and across a variety of projects. Yet we continue to learn new things.

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 developerson 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.

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

https://gist.github.com/sundeepgupta/7be5fddc7dbb8472e10dc0313132de4b

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

https://gist.github.com/sundeepgupta/294bbf0314239dfd7f4d72920b3d12d6

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.

https://gist.github.com/sundeepgupta/c7773c8215bd630913676b2f2776afce

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?

https://gist.github.com/sundeepgupta/10dba2787d2c794dd06b899a06b4f9da

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?

https://gist.github.com/sundeepgupta/bb2581b9ff3fc502024ff2b92d51eb24

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:

https://gist.github.com/sundeepgupta/e2cc4a4f3adeadcb6433c681c22ff4ac

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

I’d like to thank Eli Burnstein for helping shape and edit this article, Deyu Wang for creating the visual animation, and everyone on my project team for helping me better understand FlatMap.