Higher-order Observables and Flattening Operators

Higher Order Observables

The term higher order (something) probably sounds a lot fancier than it really is. It is also really common to hear the term higher order functions in programming. The basic idea of a higher order function is that it does either of the following (or both):

• Accepts a function as a parameter
• Returns a function

A higher order function is a function that returns a function, or accepts a function as a parameter. A higher order observable is an observable that emits observables.

You may find this supplementary video helpful to provide some context:

Play

Most of what we have looked at so far deals with observables that return simple values. For example, this:

const myObservable$ = from([1,2,3,4,5]);

Is an observable that emits numbers:

1
2
3
4
5

Let’s imagine a different scenario, we are going to borrow the general concept of the example in the documentation. Let’s say we have an observable of APIs we want to make a request to in order to fetch some data:

const myObservable$ = from([
    'https://api-one.com/data',
    'https://api-two.com/data',
    'https://api-three.com/data',
    'https://api-four.com/data'
]);

Everything is looking pretty normal so far — we have an observable that emits four strings:

https://api-one.com/data
https://api-two.com/data
https://api-three.com/data
https://api-four.com/data

That’s not of much use to us — we want the actual data from those URLs. This is a good scenario for our pipe — we want to transform this stream in some way. We want to transform this stream from being an observable of URLs into an observable of data fetched from those URLs.

The first thing that might spring to mind is to use the map operator:

myObservable$.pipe(
    map(url => this.http.get(url))
)

Now we are taking the URL that is passed into the map and making an HTTP request with Angular’s HttpClient to fetch the data for us. This looks promising, but there is a problem…

A get request made with the HttpClient will return an observable. What we want is the data from the HTTP request being emitted on this stream, but if we subscribe to the stream above what we actually get is:

Observable
Observable
Observable
Observable

We still need to subscribe to those observables to execute the HTTP request and to get the data out of them. So, how should we handle that? The most natural step might then be to… just subscribe to them. Maybe we do something like this:

myObservable$.pipe(
    map(url => this.http.get(url))
).subscribe((result) => {
    result.subscribe((data) => console.log(data))
})

This would work… but it’s getting a bit ugly with nested subscribes. Not only that, we probably want to do something more than just console.log this data — but how do we use it? Maybe we take each emission and push it into an array:

const results = [];

myObservable$.pipe(
    map(url => this.http.get(url))
).subscribe((result) => {
    result.subscribe((data) => results.push(data))
})

In case my tone hasn’t been obvious you absolutely should not do this. It works, results will now contain the results of all the HTTP requests, but it is messy and is not coding in a reactive way (since we explicitly need to subscribe to pull the data out of streams in order to work with it). We are making things way harder than they need to be.

Flattening Operators

The whole point of the example above was to show why the flattening operators we are about to talk about are useful. These operators solve this situation for us, in different ways depending on what exactly we want to do. When we have a higher order observable, or an observable that returns an observable, we can use these operators to subscribe to the inner observable (the inner observable is the observable being returned by the observable) for us. They “flatten” the inner observables by subscribing to all of them for us, and just giving us the simple values from within those streams.

To demonstrate these operators, we are going to continue using the same scenario we just discussed:

const myObservable$ = from([
    'https://api-one.com/data',
    'https://api-two.com/data',
    'https://api-three.com/data',
    'https://api-four.com/data'
]);

We want to transform this stream by taking these values and making HTTP requests to the urls.

switchMap

Let’s take a look at switchMap first, as it is probably the most common.

import { switchMap } from 'rxjs/operators';

myObservable$.pipe(
    switchMap(url => this.http.get(url))
).subscribe(val => console.log(val));

This example might seem oddly familiar. It is exactly the same as our “bad” example, except instead of the normal map operator we swapped it out with switchMap. This will give us the following result:

(data from api-four)

We kind of get the result we want — this will return the actual data from the URL rather than an observable for the HTTP request to get that data. But… it only gives us the last one. The result from making a GET request to: https://api-four.com/data.

This is often the behaviour we want, but not for this scenario. What switchMap will do is the following:

1. Receives a new url value from the outer observable (myObservable$)
2. Passes the url into the get call, and subscribes to it (the inner observable)
3. Emits the value from the inner observable

If the http.get() call returned instantly/synchronously, then switchMap would indeed return all the data for us:

(data from api-one)
(data from api-two)
(data from api-three)
(data from api-four)

But, an HTTP request might take some time to complete, maybe even a few seconds. What happens with switchMap is that it will receive its first value, and subscribe to the inner http.get() observable. However, if it receives its second value from the outer observable (myObservable$) before it gets a result from the inner observable (http.get), it will cancel the request and launch a new request for the second value instead. The process will go more like this:

1. Receives a api-one value from the outer observable (myObservable$)
2. Passes the api-one URL into the get call, and subscribes to it (the inner observable)
3. Receives a new api-two value from the outer observable
4. Unsubscribes from the http.get() for api-one and subscribes to http.get() for api-two instead
5. Receives a new api-three value from the outer observable
6. Unsubscribes from the http.get() for api-two and subscribes to http.get() for api-three instead
7. Receives a new api-four value from the outer observable
8. Unsubscribes from the http.get() for api-three and subscribes to http.get() for api-four instead
9. Emits the value from the inner observable for api-four

All of the values get emitted from myObservable$ before it has time to finish its http.get() request for each value, so it only has time to fetch and emit the value for the last value (because no more values are coming in that cause it to be cancelled before it can finish).

This is the special behaviour of switchMap: whenever it receives a new value it will cancel the previous subscription.

Although this is not suited to this particular situation, it often is the appropriate flattening operator to use. For example, imagine the scenario of a user entering a search term into a search box. When they enter a value, we make an API request to get the results. If the user enters:

socks

We will take that value and use it to launch an HTTP request to get the results for socks. But, if the user enters a new search term before that one returns:

pants