Is there an rxjs operator to take a list of observables and subscribe to N at a time until all complete?

user_300242 0
Viewed 697

My app needs to make multiple separate calls to an API for data, but I don't want to dump them all at once for performance reasons. Is there an rxjs operator that would allow me to run just a few at a time then merge all their results into one?

So if I need to get results from ABC...XYZ and I want only 2 requests activate a time it would subscribe to A and B. Then if B completes it would subscribe to C. Then A completes and it subscribes to D. And so on until all are complete but only ever having 2 subscriptions active.

rxjs
2 Answers

Yes, merge() has an option to do just that. The last parameter to merge has to be the number (of concurrent requests).

Excerpt from the docs:

const timer1 = interval(1000).pipe(take(10));
const timer2 = interval(2000).pipe(take(6));
const timer3 = interval(500).pipe(take(10));
const concurrent = 2; // the argument
const merged = merge(timer1, timer2, timer3, concurrent);
merged.subscribe(x => console.log(x));
user_12532790

The closest I came to is a combination of concat for the two streams and merge to make them run parallel. Probably there are nicer way of doing this, but maybe it could be a starting point for you to fool around with:

import { delay, concatMap } from  'rxjs/operators';
import { of, concat, merge } from 'rxjs';

const reqA = of('a').pipe(delay(1000));
const reqB = of('b').pipe(delay(1000));
const reqC = of('c').pipe(delay(1000));
const reqX = of('x').pipe(delay(1000));
const reqY = of('y').pipe(delay(1000));
const reqZ = of('z').pipe(delay(1000));

const abc = [reqA, reqB, reqC];
const xyz = [reqX, reqY, reqZ];

merge(concat(...abc), concat(...xyz)).subscribe(
  console.log
);

Here the StackBlitz i made with the code above.

So using concat on an observable array (your stream) makes sure there is only one observable running and on complete it will subscribe to the next observable in the array. With merge you can merge two streams abc and xyz meaning there will be two streams active at the same time.

UPDATE

I didn't know about the concurrent parameter for the merge operator, so @lagoman his answer is the way to go. It was not clear for me from the documentation that it was possible. Not sure why they don't mention it explicitly, seems like a very useful feature.

user_16974590
Related