24

First: This is the first project in which I am using RxJs, I thought I will learn best by using it.

I found this answer: Turning paginated requests into an Observable stream with RxJs But it says in the comments:

You're exceeding the maximum call stack still. At around 430 pages returned. I think recursion might not be the best solution here

I want to query the Youtube Data API, the results come back in pages and I need to paginate through them. I imagined a work flow like this could work: 1)Initiate a call 2)Check if the response has a 'nextPageToken' 3)If it has, do another request to the Youtube API 4)If not, finish

So to do this I could Imagine the following Observables / streams:
FirstRequestStream -A-X--------------->
ResponseStream     -A-A-A-A--X-------->
RequestStream      -I-A-I-A----------->
A = Action
I = Info from upper stream
X = Termination

(Not sure if this diagram is correct the way I made it)

So the ResponseStream depends on FirstRequestStream and RequestStream(using the merge function). The RequestStream depends on the ResponseStream( is this called a circulating observable ?)

-Is this the right approach ?

-Are 'circulating observables' a good thing, are they even possible ?(I had problems creating one).

-Any other way I should try first?

-Is it possible to create interdependent observable streams ?

Thank you for your help.

4 Answers 4

36

You are overcomplicating this problem, it can be solved a lot easier using defer operator.

Idea is that you are creating deferred observable (so it will be created and start fetching data only after subscription) and concatenate it with the same observable but for the next page, which will be also concatenated with the next page, and so on ... . And all of that can be done without recursion.

Here is how the code looks:

const { defer, from, concat, EMPTY, timer } = rxjs; // = require("rxjs")
const { mergeMap, take, mapTo, tap } = rxjs.operators; // = require("rxjs/operators")

// simulate network request
function fetchPage(page=0) {
  return timer(100).pipe(
    tap(() => console.log(`-> fetched page ${page}`)),
    mapTo({
      items: Array.from({ length: 10 }).map((_, i) => page * 10 + i),
      nextPage: page + 1,
    })
  );
}

const getItems = page => defer(() => fetchPage(page)).pipe(
  mergeMap(({ items, nextPage }) => {
    const items$ = from(items);
    const next$ = nextPage ? getItems(nextPage) : EMPTY;
    return concat(items$, next$);
  })
);

// process only first 30 items, without fetching all of the data
getItems()
 .pipe(take(30))
 .subscribe(e => console.log(e));
<script src="https://unpkg.com/[email protected]/bundles/rxjs.umd.min.js"></script>

8
  • But as I understand it this will make all the paginated call seven if no-one is interested in the results--eg, the user has not pressed the "next page" button, done "inifinite scrolling". How can this logic be modified to allow something like that?
    – user663031
    Jun 27, 2016 at 18:05
  • @torazaburo no this observable is deferred ("lazy"), so it will start making requests only after someone has subscribed to it, and stop when you unsubscribe. So if you "take" just several items from the stream, only needed pages will be fetched (not all of them), see the code examples in the answer. Jun 27, 2016 at 18:29
  • 1
    "And all of that can be done without recursion" From what I know calling fetchItems inside fetchItems IS a recursion. Can you explain?
    – kayo
    Aug 24, 2018 at 10:25
  • 1
    @hasanain I have updated code snippet to rxjs v6 (sorry for the late reply) Aug 24, 2018 at 23:14
  • 1
    This may be also a case for expand - see blog.angularindepth.com/rxjs-understanding-expand-a5f8b41a3602
    – Picci
    Sep 12, 2018 at 6:01
21

Here is my solution using the rxjs operators expand, reduce and empty using the HttpClient module:

Suppose your API response is an object containing shaped like the following

interface Response {
  data: items[]; // array of result items
  next: string|null; // url of next page, or null if there are no more items
}

You could use expand and reduce like so

getAllResults(url) {
  return this.http.get(url).pipe(
    expand((res) => res.next ? this.http.get(res.next) : EMPTY),
    reduce((acc, res) => acc.concat(res.data), [])
  );
}
0
12

I shamelessly reuse the code snippet from Oles Savluk, with its good fetchPage function, and I apply the ideas explained in the blog article linked to by Picci (in the comments), using expand.

Article on expand by Nicholas Jamieson

It gives a slightly simpler code, with recursion hidden in the expand call (and comments of the article show how to linearize it, if needed).

const { timer, EMPTY } = rxjs; // = require("rxjs")
const { concatMap, expand, mapTo, tap, toArray } = rxjs.operators; // = require("rxjs/operators")

// simulate network request
const pageNumber = 3;
function fetchPage(page = 0) {
  return timer(1000).pipe(
    tap(() => console.log(`-> fetched page ${page}`)),
    mapTo({
      items: Array.from({ length: 10 }).map((_, i) => page * 10 + i),
      nextPage: ++page === pageNumber ? undefined : page,
    }),
  );
}

const list = fetchPage().pipe(
  expand(({ nextPage }) => nextPage ? fetchPage(nextPage) : EMPTY),
  concatMap(({ items }) => items),
  // Transforms the stream of numbers (Observable<number>)
  // to a stream with only an array of numbers (Observable<number[]>).
  // Remove if you want a stream of numbers, not waiting for all requests to complete.
  toArray(),
);

list.subscribe(console.log);
<script src="https://unpkg.com/[email protected]/bundles/rxjs.umd.min.js"></script>

1
  • Thanks for the link to Nicholas Jamieson's article - it had a solution for the exact problem I was trying to solve!
    – decocijo
    Jun 26, 2019 at 6:54
0

LuJaks is definitively the simplest approach !

For a one line example, suppose you have a function that make a http request for a given page and that returns a (partial) array of data. We call that function until server returns empty array :

import { Observable, EMPTY, of } from "rxjs";
import { expand, reduce } from "rxjs/operators";

// Mock a request that returns only 5 pages... 
function httpGet(p): Observable<number[]> {
  if (p > 5) { return of([]); }
  return of(new Array(10).fill(0).map((_, i) => p * 10 + i));
}

httpGet(0).pipe( // get the fist page
    expand((value, index) => (value.length > 0 ? httpGet(index + 1) : EMPTY)), // other pages
    reduce((a, v) => [...a, ...v], []), // optional if you want only one emit
  ).subscribe((x) => console.log(x));

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.