Angular + mergeMap

By Arvind Rai, January 25, 2024

RxJS mergeMap operator projects each source value to an Observable and finally they are merged into output Observable using RxJS mergeAll operator. If the source Observable has emitted more than one element to mergeMap and if inner Observable of mergeMap has not completed yet for the previous element then mergeMap will wait to execute all inner Observable and then merge them in one Observable as output.
Suppose source Observable is emitting 3 elements consecutively to mergeMap and every element is processed by an inner Observable inside mergeMap. Now suppose all the three elements from source Observable has entered into mergeMap before inner Observable for first element is not completed. Then mergeMap will wait to complete the inner Observable for all the three elements and then their Observable responses will be merged into one Observable response and returned it by mergeMap as output.

1. Using RxJS mergeMap

RxJS mergeMap is imported as following.

import { mergeMap } from 'rxjs/operators';

Find sample examples.
Example-1:

of('x', 'y', 'z').pipe(
   mergeMap(el => of(1, 2).pipe(
	   delay(2000),
	   map(num => el+num)        
	 )
   )
).subscribe(res => console.log(res));

The 'x', 'y', 'z' elements from the source Observable will be emitted to mergeMap. The inner Observable will emit 1, 2 number for every elements of source Observable. We are delaying inner Observable process by 2 seconds and hence before completing request for 'x' element, the 'y' and 'z' element will also enter into mergeMap. So mergeMap will wait for completion of inner Observable for all the three elements and finally will merge them into one Observable as output of mergeMap.
We can understand merging of inner Observable into one as following.

of(x1) + of(x2) + of(y1) + of(y2) + of(z1) + of(z2) = of(x1, x2, y1, y2, z1, z2)

When we subscribe, we will get output x1, x2, y1, y2, z1, z2.
If inner Observable of mergeMap, is responding for any emitted element from source Observable before getting request for new element, then there is nothing to merge.

Example-2:

id = 102;
addMoreBooks() {
 let book1 = new Book(++this.id, "Book-"+ this.id);
 let book2 = new Book(++this.id, "Book-"+ this.id);
 let book3 = new Book(++this.id, "Book-"+ this.id);

 of(book1, book2, book3).pipe(
	mergeMap(book => {
	  delay(2000);
	  return this.bookService.addBook(book);
	})
 ).subscribe(book=>{
	console.log(book.id+":"+book.name);
  });
}

In the above code, before addBook returns response for book1, source Observable will emit book2, and book3 to mergeMap. So the responses of all the three inner Observable will be merged as final Observable output of mergeMap. On subscribe of above code, output in console would be as following.

103:Book-103
104:Book-104
105:Book-105

2. mergeMap + catchError

To handle the error thrown by mergeMap, we can use catchError as following.

of(book1, book2, book3).pipe(
	mergeMap(book => {
	  delay(2000);
	  return this.bookService.addBook(book);
	}),
	catchError(err => {
	  console.error(err.message);
	  return of(new Book(100, "Default Book"));
	})
).subscribe(book=>{
	console.log(book.id+":"+book.name);
});

3. mergeMap vs switchMap vs concatMap vs exhaustMap

1. mergeMap merges the output of inner Observable into one Observable and returns it as response. Output order is not fixed.

of('x', 'y', 'z').pipe(
   mergeMap(el => of(1, 2).pipe(
	   delay(2000),
	   map(num => el+num)        
	 )
   )
).subscribe(res => console.log(res));

Output
x1, x2, y1, y2, z1, z2

2. switchMap returns the response of only latest inner Observable output.

of('x', 'y', 'z').pipe(
   switchMap(el => of(1, 2).pipe(
	   delay(2000),
	   map(num => el+num)        
	 )
   )
).subscribe(res => console.log(res));

Output
z1, z2

3. concatMap is same as mergeMap operator but inner Observable are not merged but they are concatenated and keeps the order.

of('x', 'y', 'z').pipe(
   concatMap(el => of(1, 2).pipe(
	   delay(2000),
	   map(num => el+num)        
	 )
   )
).subscribe(res => console.log(res));

Output
x1, x2 (after 2 seconds) y1, y2 (after 2 seconds) z1, z2

4. exhaustMap returns the response of oldest inner Observable. It is opposite to switchMap operator.

of('x', 'y', 'z').pipe(
   exhaustMap(el => of(1, 2).pipe(
	   delay(2000),
	   map(num => el+num)        
	 )
   )
).subscribe(res => console.log(res));

Output
x1, x2