Rx Js 6 загрузить больше потока пагинации

Question

Я пытаюсь создать одну из тех более современных страниц, где нет выделенных ссылок для отдельных страниц, а одна, где автоматически загружается больше результатов, когда пользователь прокручивает страницу вниз. На веб-странице есть несколько виджетов, которые позволяют изменять параметры поиска. Когда параметры изменяются, следует получать дополнительные результаты через ajax, начиная с первой страницы. Я довольно новичок в Rx Js, и у меня возникают проблемы, связанные с тем, как определить наблюдаемые / предметы, которые мне нужны, и как составить их для достижения описанного поведения.

Вот специфика c flow Я имею в виду:

При первой загрузке страницы берется начальный набор параметров и используется для загрузки первой страницы. Когда происходит событие «load more», следующая страница должна быть извлечена и отображена на странице.

Когда параметры изменяются, страница должна быть загружена снова, начиная со страницы 1.

Когда сервер сигнализирует о том, что больше нет результатов для загрузки, я должен получить уведомление об этом через наблюдаемый объект. Если дальнейшие события «load more» запускаются после того, как больше нет доступных страниц, не следует делать запрос ajax для экономии полосы пропускания на мобильных устройствах.

Наконец, пока открыт сетевой запрос, я хочу быть возможность отображать загрузчик, поэтому мне нужна заметка, которая информирует меня о том, есть ли открытые запросы или нет.

В качестве бонуса: в настоящее время я реализовал сигнализацию не более результатов, возвращая 404 из бэкэнда, когда запрашивается страница, которая больше последней. Я хотел бы использовать catchError на ajax, наблюдаемом таким образом, что он изящно останавливает запрос ajax, не нарушая подписку.

Вот что я смог придумать так далеко, но у него есть несколько проблем (описанных ниже):

import { BehaviorSubject, Subject, fromEvent } from 'rxjs'; 
import { map, mergeMap, switchMap, takeUntil, tap } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';
import { stringify } from 'qs';

const paramsEl = document.querySelector<HTMLTextAreaElement>('#params');
const paramsChangedBtn = document.querySelector<HTMLButtonElement>('#paramsSubmit');
const loadNextPageBtn = document.querySelector<HTMLButtonElement>('#loadNextPage');

const getParams = () => JSON.parse(paramsEl.value);

const params$ = new BehaviorSubject(getParams());
const page$ = new BehaviorSubject(1);
const noMoreResults$ = new Subject<void>();       // <- public
const connections$ = new BehaviorSubject(0);
const loading$ = new BehaviorSubject(false);      // <- public

// for the sake of this example we're not using an IntersectionObserver etc. but a plain button to fire a "load more" event
const loadNextPage$ = fromEvent(loadNextPageBtn, 'click');
// same for params changed event. In my real app I've got a working stream fed from the widgets
fromEvent(paramsChangedBtn, 'click').subscribe(e => params$.next(getParams()));

// when the params change, reset page to 1
params$.subscribe(() => page$.next(1));

// update loading$ observable for displaying/hiding a loader
connections$.subscribe(connections => {
  if(connections > 0 && loading$.getValue() === false) loading$.next(true);
  if(connections <= 0 && loading$.getValue() === true) loading$.next(false);
});

// when we need to load the next page, increment the page observable
loadNextPage$.subscribe(e => page$.next(page$.getValue() + 1));

//////////////

// whenever a new page should be requested, get the current parameters and fetch data for this page
page$
  .pipe(
    takeUntil(noMoreResults$),
    tap(() => connections$.next(connections$.getValue() + 1)),
    mergeMap(page => {
      const qs = stringify({
        ...params$.getValue(),
        page,
      });
      return ajax.getJSON<any>(`https://httpbin.org/get?${qs}`)
        // this doesn't seem to do anything
        // furthermore the ajax request would already have been made at this point
        // .pipe(
        //   takeUntil(noMoreResults$)
        // );
    }),
    tap(() => connections$.next(connections$.getValue() - 1)),
  )
  .subscribe(data => {
    console.log(data.args);
    // for testing purposes pretend we have no more data at page 5
    if(data.args.page === "5") noMoreResults$.next();
  });

// for debugging purposes
loading$.subscribe(loading => console.log('loading: ', loading));
noMoreResults$.subscribe(() => console.warn('no more results'));

Вы можете найти работающую версию этого здесь на stackblitz .

Вот проблемы с код:

1. Текущий темп takeUntil(noMoreResults$) прерывает подписку, когда срабатывает noMoreResults$, а затем params$ не выдает больше никаких загруженных страниц. (См. Комментарий в коде для другого местоположения в ajax трубе).
2. с использованием params$.getValue(), когда mergeMapping к наблюдаемой ajax чувствует себя неправильно, однако я не знаю, как передать оба номер страницы, а также параметры в одном потоке.
3. В общем, я думаю, что я слишком много использовал Subjects / BehaviorSubjects, но я не уверен. Вы можете подтвердить или опровергнуть это?
4. Состав наблюдаемых ощущается очень грязным и за ним трудно следить. Основано ли это на том, что я пытаюсь сделать, или есть место для улучшения этой проблемы?

Можете ли вы привести рабочий пример, а также рассказать о самых больших ошибках, которые я допустил.