Impaginazione osservabile RxJs

Question

Primo: Questo è il primo progetto in cui sto usando RxJs, ho pensato che avrei imparato meglio usandolo.

ho trovato questa risposta: Turning paginated requests into an Observable stream with RxJs Ma si dice nei commenti:

Stai superiore dello stack massima chiamata ancora. A circa 430 pagine restituite. Penso che la ricorsione potrebbe non essere la soluzione migliore qui

Voglio interrogare l'API di dati di YouTube, i risultati tornano nelle pagine e ho bisogno di impaginare attraverso di loro. Ho immaginato che un flusso di lavoro come questo potesse funzionare: 1) Avvia una chiamata 2) Controlla se la risposta ha un 'nextPageToken' 3) Se lo ha, fai un'altra richiesta all'API di YouTube 4) Altrimenti, termina

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 

(Non so se questo schema è corretto il modo in cui l'ho fatta)

Così il responseStream dipende FirstRequestStream e RequestStream (utilizzando la funzione di fusione). RequestStream dipende da ResponseStream (si tratta di un osservabile circolante?)

-È questo l'approccio giusto?

-Le 'osservabili circolanti' sono una buona cosa, sono anche possibili? (Avevo problemi a crearne una).

-Qualsiasi altro modo dovrei provare prima?

-È possibile creare flussi osservabili interdipendenti?

Grazie per il vostro aiuto.

Answer 1

Si sta complicando questo problema, può essere risolto molto più facilmente utilizzando Rx.Observable.defer.

L'idea è che si sta creando un osservabile differito (quindi verrà creato e inizierà il recupero dei dati solo dopo la sottoscrizione) e lo si concatena con lo stesso osservabile ma per la pagina successiva, che sarà anche concatenata con la pagina successiva, e presto ... . E tutto ciò può essere fatto senza ricorsione.

Ecco come il codice è:

function fetchItems(params, pageToken) { 
    return Observable.defer(
    () => fetchSomething({ 
     params, 
     pageToken, 
    }) 
     .flatMap(({ items, nextPageToken }) => { 
     const items$ = Observable.fromArray(items); 
     const next$ = nextPageToken 
      ? fetchItems(params, nextPageToken) 
      : Observable.empty(); 

     return Observable.concat(
      items$, 
      next$ 
     ); 
     }) 
); 
} 

const items$ = fetchItems(params) 
    // process all items until end 

const firstTenItems$ = fetchItems(params) 
    .take(10); 
    // process only first 10 items, without fetching all of the data 

Dove fetchSomething è funzione che accetta alcuni params e tornare osservabile di risultato. Esso può essere creato da metodi di libreria come segue:

import {Observable} from 'rx'; 
import google from 'googleapis'; 

const yt = google.youtube({ 
    version: 'v3', 
    auth: // your apiKey here, 
}); 

const getChannel = Observable.fromNodeCallback(yt.channels.list, yt, r=>r);