In the previous section, we learned how to transform and combine observables with operations such as map, reduce, and zip. However, the two observables above—musicians and bands—were perfectly capable of producing values on their own. They did not need any extra input.

In this section, we examine a different scenario: we'll learn how we can combine observables, where the output of one is the input of another. We encountered flatmap before in Chapter 1, What is Reactive Programming? If you have been wondering what its role is, this section addresses exactly that.

Here's what we are going to do: given an observable representing a list of all positive integers, we'll calculate the factorial for all even numbers in that list. Since the list is too big, we'll take five items from it. The end result should be the factorials of 0, 2, 4, 6, and 8, respectively.

The first thing we need is a function to calculate the factorial of a number n, as well as our observable:

(defn factorial [n]
  (reduce * (range 1 (inc n))))

(defn all-positive-integers []
  (Observable/interval 1 TimeUnit/MICROSECONDS))

Using some type of visual aid will be helpful in this section, so we'll start with a marble diagram representing the previous observable:

The middle arrow represents time and it flows from left to right. This diagram represents an infinite Observable sequence, as indicated by the use of ellipsis at the end of it.

Since we're combining all the observables now, we'll create one that, given a number, emits its factorial using the helper function defined earlier. We'll use Rx's create method for this purpose:

(defn fact-obs [n]
  (rx/observable*
   (fn [observer]
     (rx/on-next observer (factorial n))
     (rx/on-completed observer))))

This is very similar to the just-obs observable we created earlier in this chapter, except that it calculates the factorial of its argument and emits the result/factorial instead, ending the sequence immediately thereafter. The following diagram illustrates how it works:

We feed the number 5 to the observable, which in turn emits its factorial, 120. The vertical bar at the end of the time line indicates the sequence terminates then.

Running the code confirms that our function is correct:

(rx/subscribe (fact-obs 5) prn-to-repl)

;; 120

So far so good. Now, we need to combine both observables in order to achieve our goal. This is where flatmap of Rx comes in. We'll first see it in action and then get into the explanation:

(rx/subscribe (->> (all-positive-integers)
                   (rx/filter  even?)
                   (rx/flatmap fact-obs)
                   (rx/take 5))
              prn-to-repl)

If we run the preceding code, it will print the factorials for 0, 2, 4, 6, and 8, just like we wanted:

1
2
24
720
40320

Most of the preceding code snippet should look familiar. The first thing we do is filter all even numbers from all-positive-numbers. This leaves us with the following observable sequence:

Much like all-positive-integers, this, too, is an infinite observable.

However, the next line of our code looks a little odd. We call flatmap and give it the fact-obs function. A function we know itself returns another observable. flatmap will call fact-obs with each value it emits. fact-obs will, in turn, return a single-value observable for each number. However, our subscriber doesn't know how to deal with observables! It's simply interested in the factorials!

This is why, after calling fact-obs to obtain an observable, flatmap flattens all of them into a single observable we can subscribe to. This is quite a mouthful, so let's visualize what this means:

As you can see in the preceding diagram, throughout the execution of flatmap, we end up with a list of observables. However, we don't care about each observable but rather about the values they emit. Flatmap, then, is the perfect tool as it combines—flattens—all of them into the observable sequence shown at the bottom of the figure.

You can think of flatmap as mapcat for observable sequences.

The rest of the code is straightforward. We simply take the first five elements from this observable and subscribe to it, as we have been doing so far.

One more flatmap for the road

You might be wondering what would happen if the observable sequence we're flatmapping emitted more than one value. What then?

We'll see one last example before we begin the next section in order to illustrate the behavior of flatMap in such cases.

Here's an observable that emits its argument twice:

(defn repeat-obs [n]
  (rx/seq->o (repeat 2 n)))

Using it is straightforward:

(-> (repeat-obs 5)
    (rx/subscribe prn-to-repl))

;; 5
;; 5

As previously, we'll now combine this observable with the one we created earlier, all-positive-integers. Before reading on, think about what you expect the output to be for, say, the first three positive integers.

The code is as follows:

(rx/subscribe (->> (all-positive-integers)
                   (rx/flatmap repeat-obs)
                   (rx/take 6))
              prn-to-repl)

And the output is as follows:

0
0
1
1
2
2

The result might be unexpected for some readers. Let's have a look at the marble diagram for this example and make sure we understand how it works:

Each time repeat-obs gets called, it emits two values and terminates. flatmap then combines them all in a single observable, making the previous output clearer.

One last thing worth mentioning about flatmap—and the title of this section—is that its "friends" refer to the several names by which flatmap is known.

For instance, Rx.NET calls it selectMany. RxJava and Scala call it flatMap—though RxJava has an alias for it called mapMany. The Haskell community calls it bind. Though they have different names, these functions semantics are the same and are part of a higher-order abstraction called a Monad. We don't need to know anything about Monads to proceed.

The important thing to keep in mind is that when you're sitting at the bar talking to your friends about Compositional Event Systems, all these names mean the same thing.