RxJs Observable Pagination

问题:

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.

回答1:

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/rxjs@6.2.2/bundles/rxjs.umd.min.js"></script>