RxJs Наблюдаемая разбивка на страницы

Question

Во-первых: Это первый проект, в котором я использую RxJs. Я думал, что я буду учиться лучше всего, используя его.

Я нашел этот ответ: Turning paginated requests into an Observable stream with RxJs Но это говорит в комментариях:

Вы превышая максимальный стек вызовов до сих пор. Пришло около 430 страниц. Я думаю, recurion может быть и не лучшим решением.

Я хочу запросить API данных Youtube, результаты возвращаются на страницы, и мне нужно их прокручивать. Я предположил, что такой рабочий процесс может работать: 1) Инициировать вызов 2) Проверьте, имеет ли ответ «nextPageToken» 3) Если он есть, выполните другой запрос API Youtube 4) Если нет, завершите

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 

(Не уверен, что если эта схема верна, как я сделал это)

Так ResponseStream зависит от FirstRequestStream и RequestStream (с помощью функции слияния). RequestStream зависит от ResponseStream (это называется циркулирующим наблюдаемым?)

-Это правильный подход?

-Возможно, что «циркулирующие наблюдаемые» - это даже возможно (у меня были проблемы с созданием одного).

-Другой путь Я должен попробовать в первую очередь?

-Возможно создать взаимозависимые наблюдаемые потоки?

Благодарим за помощь.

Answer 1

Вы преувеличиваете эту проблему, ее можно решить намного проще, используя Rx.Observable.defer.

Идея заключается в том, что вы создаете отложенную наблюдаемую (поэтому она будет создана и начнет получать данные только после подписки) и объединяет ее с тем же наблюдаемым, но для следующей страницы, которая также будет объединена со следующей страницей, и скоро ... . И все это можно сделать без рекурсии.

Вот как выглядит код:

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 

Где fetchSomething это функция, которая принимает некоторые Params и возврата наблюдаемым результата. Он может быть создан из библиотечных методов следующим образом:

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);