RxJS is a popular library nowadays and it’s used in a wide range of applications. And it even becomes part of popular frameworks like Angular.
It’s not unknown that the library comes with many powerful functions that can be used together to solve complex problems with only a few lines of code. A couple of days ago, I saw the following question on a developer website:
How can you use flatMap operator to process a list of data as an Observable and then process the data one by one as Observables too?
Of course, there could be several ways to solve this problem. However, we will address a solution with reactive thinking to come up with a solution that could be used in many similar problems.
The Problem
Let's suppose you're building an application that needs to handle a series of asynchronous calls to display the users and their respective addresses.
However, in a real-world scenario, the system can provide a RESTful service to get the list of the Users(ie. getUsers()).
Then, you'll need to perform another HTTP call to get the address for each of the previous users (ie. getAddress(userId)).
Let's solve this problem in a Reactive Approach using RxJS operators and TypeScript.
Define the Data Model
Let's rely on TypeScript interfaces and powerful static typing to have the model ready.
// user.ts
export interface User {
id: number;
name: string;
address?: Address;
}
The User interface defines the address attribute as optional. That means it can be undefined if the user doesn't "contain" a valid address value yet.
// address.ts
export interface Address {
country: string;
state: string;
city: string;
street: string;
zipCode: number;
}
The Address interface displays a set of attributes that represents a complete address.
The Data Example
To make testing easier, let's define a dataset for the entities we defined already in the model.
// index.t
import { Address, User } from "./address";
import { User } from "./user";
const users: User[] = [
{
id: 0,
name: "Donald Mayfield"
},
{
id: 1,
name: "Jill J. Fritz"
},
{
id: 2,
name: "Terry Buttram"
}
];
const address: Address[] = [
{
street: "2180 BELLFLOWER",
country: "USA",
state: "AL",
city: "Madison",
zipCode: 35064
},
{
street: "845 ODOM ROAD, SUITE 200",
country: "USA",
state: "CA",
city: "Los Angeles",
zipCode: 90720
},
{
street: "9025 QUEENS BLVD",
country: "USA",
state: "NY",
city: "Queens",
zipCode: 11355
}
];
Looks like we'll need a function to get an Address given a User identifier.
// index.ts
const getAddress = (userId: number): Observable<Address> => of(address[userId]);
The previous function getAddress will return an Address as an Observable: Observable<Address>. This is possible with the use of the of operator from RxJS, which is a "Creation Operator" that converts the arguments (an Address object) to an observable sequence.
In a real-world scenario, an asynchronous operation could take a variable time according to the request. So let's add a random delay for every request to "simulate" such a situation.
// index.ts
const getAddress = (userId: number): Observable<Address> =>
of(address[userId]).pipe(
tap(() => console.log(`getAddress(${userId})]. Started`)),
delay(randomDelay()),
tap(() => console.log(`getAddress(${userId}). Finished`))
);
If the function keyword is your friend in JavaScript, then you can write it as follows:
// index.ts
function getAddress(userId: number): Observable<Address> {
return of(address[userId]).pipe(
tap(() => console.log(`getAddress(${userId})]. Started`)),
delay(randomDelay()),
tap(() => console.log(`getAddress(${userId}). Finished`))
);
}
Processing the Data as Observables
Since we have the data model defined, it is time to create some variables that allow to "contain" the set of users and addresses as Observables too:
// index.ts
let users$: Observable<User[]>;
let address$: Observable<Address[]>;
Next, let's assign the appropriate value to the users$variable:
users$ = of(users);
Again, the previous line is using the of operator to create an observable from the users array.
In the real world, the "users" data will come from a RESTful endpoint, a JSON file, or any other function that can perform an asynchronous call. You may expect an Observable as a result of that function or you can create it using an RxJS Operator.
Using flatMap and mergeMap Operators
Now it is time to process the set of Users and perform the getAddress(id) call for every User:
// index.ts
address$ = users$.pipe(
// "flat" the Array: User[] -> User
flatMap(users => users)
);
Let's describe what's happening at this point.
- users$.pipe(). This function call provides a readable way to use RxJS operators together (
flatMapoperator goes first). - flatMap() operator allows to process the data array(
User[]) which comes from the observable (Observable<User[]>). Also, we can think that this operator allows transforming the users emitted by a single observable into many observables (one per user).
So far, we're able to process every user separately. However, we're not doing anything useful yet with the new observables. Let's "map" every user and perform a new request to get their addresses:
// index.ts
address$ = users$.pipe(
flatMap(users => users),
// process the User and return the Address as an Observable
mergeMap(user => {
const address$: Observable<Address> = getAddress(user.id);
// You can even apply other operators to address$ here...
return address$;
})
);
- mergeMap() operator takes the user output from the previous operator(flatMap). The user identifier is read to perform a request to obtain the respective address. As you may understand, this operator maps each value to an Observable.
Some reasons why using the mergeMap operator in this solution:
- Since we have a set of users, it's required to get the Address of all of them.
- The order of the final result doesn't matter.
As a final step, you'll need to subscribe as follows.
// Subscription
address$.subscribe((address: Address[]) => {
console.log({ address });
});
After running this code, you may see something similar to the following output:
getAddress(0)]. Started
getAddress(1)]. Started
getAddress(2)]. Started
getAddress(1). Finished
{address: {…}}
getAddress(2). Finished
{address: {…}}
getAddress(0). Finished
{address: {…}}
Surprise! We are getting one object at a time. That is totally expected since we flattened the original array and then the mergeMap operator is emitting an observable at a time.
We got an initial array as an input: how can we get an array as an output?
That would be one of your questions at this time. Let's change the code using the toArray operator:
// index.ts
address$ = users$.pipe(
flatMap(users => users),
mergeMap(user => {
const address$: Observable<Address> = getAddress(user.id);
return address$;
}),
// Get the values inside an array at the end
toArray()
);
With the help of toArray operator, once the source Observable is done, it will return an array containing all the emitted values(In this case an array of Address[]).
The final result may be a JSON object as follows:
{
"address": [
{
"street": "2180 BELLFLOWER",
"country": "USA",
"state": "AL",
"city": "Madison",
"zipCode": 35064
},
{
"street": "845 ODOM ROAD, SUITE 200",
"country": "USA",
"state": "CA",
"city": "Los Angeles",
"zipCode": 90720
},
{
"street": "9025 QUEENS BLVD",
"country": "USA",
"state": "NY",
"city": "Queens",
"zipCode": 11355
}
]
}
flatMap vs mergeMap
According to RxJS documentation, flatMap is an alias of mergeMap. You can verify it here.
However, it's usual to find some questions about flatMap operator and doubts about how it works. Also, some time ago, people around RxJS community suggested renaming mergeMap back to flatMap.
I personally like the idea of using a flatMap operator because it resembles the Array.prototype.flatMap() function in JavaScript. I prefer to use it in that way.
Alternative Solution
Would you like to see an alternative solution to this problem? Do not miss the Understanding switchMap and forkJoin operators in RxJS article.
Demo Project
Find the complete project running in StackBlitz. Don't forget to open the browser's console to see the results.
You can follow me on Twitter and GitHub to see more about my work.
