Rxjs Expand - Reactively Retrieve All Rows From a Paginated Source

Let's say you have a data source which is paginated, so you can request this page or that, but you can't request all the entries at once. Now, regardless of whether it is right or not, how would you approach this task, if you wanted to do it anyway?

One approach would be imperative. In the imperative way, we use a loop, get one page, wait for the results, get the next one, and so on. When a page query returns no results, we return the collected results.

let pageNumber = 0;
let results = [];

do {
  const pageResults = await getPage(pageNumber++).toPromise();
  results = results.concat(pageResults);
} while(pageResults.length !== 0);

return results;

But imperative is so 2nd millennium, we want to do some reactive magic! Enter expand...

The expand Rxjs operator takes an input, and returns an observable. It then emits from that observable, but for each emitted item, it calls itself recursively. In our case the recursion is simple, call with page: n+1 until no more data arrives.

For this recursion to be valid, expand needs to return an observable of the same type of its input, and this input must contain the page number, so we must "annotate" our results with the page number as well.

const getPageAnnotated = pageNumber =>
  getPage(pageNumber)
    // Annotate with the page number
    .pipe(map(results => ({ results, page: pageNumber })));

return getPageAnnotated(0).pipe(
  expand(pr => getPageAnnotated(pr.page + 1)),
  takeWhile(pr => pr.results.length !== 0),
  map(pr => pr.results),
  scan((acc, v) => [...acc, ...v])
);

Although this works, the astute will notice that we're missing the termination condition in the expand call. So how come it works without it? The answer is takeUntil - because we are completing the entire stream as soon as a query returns with no results, the next query will get cancelled.

This might be harmless, but we can avoid it easily by changing the expand call to:

expand(pr => pr.results.length !== 0 
               ? getPageAnnotated(pr.page + 1) 
               : EMPTY)

This way we will not try to get the next page, if the result from the current page is empty.