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

Question

В моем приложении есть много таблиц, которые используют пейджинг и поиск на стороне сервера.Я хочу, чтобы в этих таблицах использовалась бесконечная прокрутка, а в нижней части таблицы я использую кнопку «Загрузить еще».Я хочу создать единый сервис, который можно повторно использовать во всех моих представлениях для всех этих таблиц, но я довольно новичок в rxjs и Observables и, похоже, не могу найти правильный способ выполнить все, что мне нужно.Вот что у меня есть:

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

// Setup our search stream that will trigger the $resources observable to be 
// refreshed upon search updating
this.searchStream = this.searchWatcher.pipe(
    // wait Xms after each keystroke before considering the term
    debounceTime(this.searchDebounceTime),

    // ignore new term if same as previous term
    distinctUntilChanged(),

    // switch to new search observable each time the term changes
    map((term: string) => {
        this.searchTerm = term;
        this.pageCurrent = 0;
        return new QueryOptions(this.pageSize, this.pageSize * this.pageCurrent, this.searchTerm);
    })
);

Наблюдаемый потокдля пейджинга, который также будет возвращать новый набор QueryOptions каждый раз, когда мы переходим на другую страницу (так как его бесконечная прокрутка всегда его this.pageCurrent++).Очевидно, что мы сохраним searchTerm на всех постраничных запросах.

// Setup our page stream that will trigger the $resources observable to be
// refreshed upon the page being changed
this.pageStream = this.pageWatcher.pipe(
    map(page => {
        return new QueryOptions(this.pageSize, this.pageSize * page, this.searchTerm);
    })
);

Теперь мне нужно объединить эти 2 потока в один optionsStream и сказать, чтобы он начинался со страницы 1, без запросакак это:

// This becomes a stream of the most recently updated options
let optionsStream = merge(this.searchStream, this.pageStream).pipe(
    // this line initializes the page at page 1, with `null` as the search term
    startWith(new QueryOptions(this.pageSize, 0))
);

Наконец, мой dataStream, который просто берет параметры, предоставленные апстримом, и отправляет запрос бэкэнд-серверу:

// This stream is the data stream triggered when the options are updated
this.dataStream = optionsStream.pipe(
    mergeMap(options => {
        // this.source(...) will return an observable of the items
        // I want to page through. e.g. returns a `Obserable<any[]>`
        return this.source(options);
    })
);

Пока все это работаетотлично, и я получаю правильные обновления, когда любой из этих двух потоков вносит изменения в параметры.Теперь моя проблема заключается в попытке выяснить, как сделать 2 вещи:

1. Мне нужно собрать все события, которые происходят из моего dataStream, так что если на странице 1 есть 10 элементов, то послеперейдя к странице 2, теперь должно быть 20 пунктов, затем на странице 3 будет 30 элементов и так далее.Я думаю, что могу использовать scan, чтобы сделать это, и даже написал небольшой тестовый сценарий, который накапливает «страницы», такие как:

let page1 = [1, 2, 3];
let page2 = [4, 5, 6];
let page3 = [7, 8, 9];

let pager = new BehaviorSubject([]);
let pagerObs = pager.asObservable();

pagerObs.pipe(
    scan((acc, val) => {
        return acc.concat(val);
    })
).subscribe(page => {
    console.log(page);
});

setTimeout(() => {
    pager.next(page1);
}, 2000);

setTimeout(() => {
    pager.next(page2);
}, 4000);

setTimeout(() => {
    pager.next(page3);
}, 6000);

// output looks like [1,2,3], then [1,2,3,4,5,6], then [1,2,3,4,5,6,7,8,9]
// which is exactly what I want!

Когда используется новый поисковый термин, мне нужно «сбросить» наблюдаемое, чтобы scan начинался с пустого набора [], однако я не могу найти правильный способ сделать это, используя чистые наблюдаемые.

Может ли кто-нибудь заполнить пробелы здесь о том, как я мог бы достичь такого рода функциональности с помощью наблюдаемых?

TL; DR

При использовании оператора scan вrxjs перебирать бэкэнд-сервер с большим количеством объектов, что является лучшим способом «перезагрузить» аккумулятор и начать его с нуля, если кто-то обновляет поисковый запрос, и поэтому подкачка должна начинаться со страницы 1.

ОБНОВЛЕНИЕ

Хорошо, я думаю, что я понял это, но у меня есть ощущение, что это немного хакерский и не использует наблюдаемые операторы, как они были предназначены:

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

Оставит вопрос открытым для тех, кто имеет более элегантное решение или предложения

this.$resources = this.dataStream.pipe(
    scan((acc, val) => {
        return this.pageCurrent == 0 ? val : acc.concat(val);
    }),
    share(),
    tap(_ => {
        this.isLoading = false;
        this.isSearching = false;
    })
);