concatMap only subscries to the first inner observable, ignoring the rest

userwBXl1 0
Viewed 27

I am trying to run multiple request in batches of 3 using the windowCount operator and joining the results using the concatMap operator:

import { range, windowCount, map, concatMap, mergeMap } from 'rxjs';
import { ajax } from 'rxjs/ajax';

const ids = range(1, 10);
const result = ids.pipe(
  windowCount(3),
  concatMap((win) =>
    win.pipe(
      mergeMap((id) =>
        ajax(`https://jsonplaceholder.typicode.com/posts/${id}`).pipe(
          map((res) => res.response)
        )
      )
    )
  )
);
result.subscribe((x) => console.log(x));

But the concatMap operator sunscribes only to the first window and ignores the rest. This is the result i get:

>> {id: 2, ...}
>> {id: 3, ...}
>> {id: 1, ...}

What's going on?

javascript rxjs
1 Answers

Each window created is a Subject that emits its values when the source emits.

In this case you have a source that emits 10 values synchronously which means that all 4 windows also emit its values synchronously.

Since the process of each window is async, when the 2nd, 3rd and 4th window are subscribed to in the concatMap, the internal Subject has already emitted its values and you end up with an EMPTY observable for those 3 cases.

So for this use case you are better of using bufferCount that works the same but returns an Array with the accumulated values instead of an observable.

const result = ids.pipe(
  bufferCount(3), // [1,2,3], [4,5,6], ...
  concatMap((batch) =>
    from(batch).pipe( // use from to create an Observable that emit each value of the array
      mergeMap((id) =>
        ajax(`https://jsonplaceholder.typicode.com/posts/${id}`).pipe(
          map((res) => res.response)
        )
      )
    )
  )
);

Cheers

akotech0
Related