How to properly flatten an stream of observables?

Question

Here's what I trying to do:

Given a paginated API, get all the resources using parallel requests.

The API returns a limited number of resources per call. So, you need to use an offset parameter to get to the next set of data until all data is extracted.

Here's my idea (but getting some warning because I'm using flat on the response), so maybe there's a better way to do this.

Get the total count of items.
Given the count and the limit, calculate how many requests are needed to get all data.
Trigger all requests in parallel and combine all data into a flattened array.

Here's an example:

https://stackblitz.com/edit/paginated-api?embed=1&file=index.ts&hideExplorer=1&devtoolsheight=100

getCount().pipe(
  mergeMap(count => range(0, Math.ceil(count / limit))),
  map(offset => getDevices(offset, limit)),
  combineAll(),
).subscribe(res => {
  const a = res.flat(); // <--- warning: Property 'flat' does not exist on type '{ name: string; }[][]'.
  console.log(JSON.stringify(a));
});

I feel this solution is a little hacky. It's flattening the response in the subscription. I'd like to know if there's an RXJS operator that I can use on the pipe to flatten the response, so I don't have to in the subscription?

Then have you read e.g. https://stackoverflow.com/q/53556409/3001761? — jonrsharpe, Jun 18 '20 at 19:24
@jonrsharpe that's good to silence the warning. However, I feel it is a little hacky to flatten array in the subscription. I'd like to know if there's an RXJS operator that I can use on the pipe to flatten the response, so I don't have to in the subscription? — Adrian, Jun 18 '20 at 19:37
RxJS doesn't care about arrays, it deals with observables. But if you're at a point where you have working code that you think could be improved, see [codereview.se]. — jonrsharpe, Jun 18 '20 at 19:41
Ok, so going back to manipulating observables, how to merge/flatten multiple observables responses? I think there must an operator that does that I'm not aware of. — Adrian, Jun 18 '20 at 19:55

DeborahK · Accepted Answer · 2020-06-18T23:00:44.007

For every inner Observable, we need another flattening operator.

So something like this would work:

getCount().pipe(

  mergeMap(count => range(0, Math.ceil(count / limit))),

  mergeMap(offset => getDevices(offset, limit)),

  mergeAll(),
  toArray()

).subscribe(res => {
  console.log('result', JSON.stringify(res));
});

The first mergeMap flattens the inner range Observable. The second mergeMap flattens the getDevices, which I assume returns an Observable.

The mergeAll() merges all of the individual values, which are the objects.

The toArray() then adds all of the objects to a single array.

This is the result:

result
[{"name":"dev-1"},{"name":"dev-2"},{"name":"dev-3"},{"name":"dev-4"},{"name":"dev-5"},{"name":"dev-6"},{"name":"dev-7"},{"name":"dev-8"},{"name":"dev-9"},{"name":"dev-10"},{"name":"dev-11"},{"name":"dev-12"},{"name":"dev-13"},{"name":"dev-14"},{"name":"dev-15"},{"name":"dev-16"},{"name":"dev-17"},{"name":"dev-18"},{"name":"dev-19"},{"name":"dev-20"}]

Hope this helps.

Thanks a lot for your answer @DeborahK! Exactly what I wanted. BTW, I went to your talk in ng-conf! Definitely taking your RXJS course in Pluralsight! :) — Adrian, Jun 19 '20 at 01:00

score 0 · Answer 2 · answered Jun 19 '20 at 18:14

3 ways to do this:

Using mergeMap, all requests are triggered in parallel. However, the end result will be based on the order of arrival. Meaning that if your API is sorted this might break it.

const getAllOffsets = () => pipe(
  mergeMap((count: number) => range(0, Math.ceil(count / limit))),
  toArray(),
  // all requests in parallel and results in order (order of creation)
  concatMap(r => forkJoin(...r.map(offset => getDevices(offset, limit)))),
  map(a => a.flat())
);

Using concatMap, this will guarantee the order. However, each request is executed one after another. This will be a performance issue

const getAllOffsets2 = () => pipe(
  mergeMap((count: number) => range(0, Math.ceil(count / limit))),
  // all requests in parallel but order is not guarantee (response order)
  concatMap(offset => getDevices(offset, limit)),
  mergeAll(),
  toArray()
);

Finally, using forkJoin. This will execute all requests in parallel and keep them in the order of creation. Like Promise.all. This is the most optimal for getting all resources from paginated APIs.

const getAllOffsets3 = () => pipe(
  mergeMap((count: number) => range(0, Math.ceil(count / limit))),
  toArray(),
  // all requests in parallel and results in order (order of creation)
  concatMap(r => forkJoin(...r.map(offset => getDevices(offset, limit)))),
  map(a => a.flat())
);

Full working example: https://stackblitz.com/edit/paginated-api-ozcgwg?file=index.ts&hideExplorer=1&devtoolsheight=100

getCount().pipe(getAllOffsets3()).subscribe(res => {
  console.log({res, size: res.length || 0});
  console.log(JSON.stringify(res))
  // console.log('end', res.map(d => d.name));
});

How to properly flatten an stream of observables?

2 Answers2