Наблюдаемые RxJs делают запросы параллельно, пока это не терпит неудачу

Question

Я хочу получить API, который возвращает данные на страницах по 50 элементов, но я не знаю, сколько их элементов (и, следовательно, страниц).

Моя идея - отправлять 20 запросов параллельно, каждыйодин будет запрашивать i-ую страницу, а затем i-ю + 20-ю страницу и т. д., пока страница не станет пустой, и в этом случае я заканчиваю.

При таком подходе я выполняю не более 20 ненужных запросов.

Дело в том, что я не знаю, как структурировать этот цикл с помощью наблюдаемых.

Я представил что-то вроде этого:

return Observable.from(_.range(0, 20))
  .map((pageNo) => fetchPage(pageNo))
  .while((page) => isValid(page));

, но этот метод while или аналогичныене существует / работа

Я нашел этот похожий вопрос, но он использует interval, который кажется неэффективным RxJs Наблюдаемый интервал до достижения желаемого значения

Насколько я понимаю, я могуне используйте takeWhile, потому что он проверяет условие, которое уже выполнено, а не ответ на запрос, который еще предстоит сделать.

Answer 1

Вы можете создать 20 запросов и подождать, пока все они завершатся с forkJoin, а затем использовать takeWhile для завершения цепочки, когда массив результатов пуст:

const fetchPage = page => {
  ...
  return forkJoin(...);
};

range(0, 20).pipe(
  concatMap(page => fetchPage(page)),
  takeWhile(arr => arr.length > 0),
)
.subscribe(console.log);

Завершить демонстрацию:https://stackblitz.com/edit/rxjs-zw1sr2?devtoolsheight=60

Answer 2

Это может помочь

return Observable.from(_.range(0, 20)).pipe(
    mergeMap(pageNo => ajax.getJSON(`/api/fetchPage/${pageNo}`).pipe(
        mergeMap(result => 
            of(addPersonFulfilled(result), secondFunc(foo)),
            retryWhen(error => tap(console.log('error on page', error)))
        )           
    ))
)