typescript - rxjs pagination med abonnement til hver side

Vi forespørger graphql over websocket for at have en abonnement på en liste af enheder. Vi bruger paginering, da listen forventes at kunne rulles uendeligt. Vi vil også gerne se ændringer foretaget i databasen i realtid i listen på frontenden.

Denne liste kan filtreres, sorteres og paginere.

Vi har klassen SelectParams, som hjælper os med at gøre det:

export class SelectParams {
    page = 0;
    query = '';
    sort: Sort = { sortBy: 'creationDate', sortOrder: 'DESC' };
    take = 30;
}

Så når vi ønsker at abonnere på en liste, giver vi bare en selectParams$=Observable<SelectParams> for at få listen, der er "reaktiv" og reagerer på ændringer i selectParams$. Dette bruges på følgende måde:

    this.items$ = this.featureSrv.selectMany(this.selectParams$)

Under motorhjelmen fungerer selectMany sådan her:

// when we call selectMany we just push the params to the pipeline
selectMany(params$: Observable<SelectParams> = of(new SelectParams())): Observable<T[]> {
    this.selectManyParams$.next(params$);
    return this.selectMany$;
}

// subject where we push params to sort, paginate and filter
selectManyParams$ = new ReplaySubject<Observable<SelectParams>>(1);
// when the params change then so does this observable, which is returned by the selectMany function
selectMany$ = this.selectManyParams$.asObservable().pipe(
    // retrieve params from their observable form
    flatMap(params$ => params$),
    // when the params haven't changed we shouldn't do anything
    distinctUntilChanged(),
    // then we query graphql to get a suscription to some part of the data
    switchMap(({ page, sort, query, take }: SelectParams) => {
    // the selectMany here is a subscription to some data on the server
        return this.apolloWrapper.selectMany(this.gql, page, sort, query, take ).pipe(
    // we add page data so we can use it in the scan
            map(data => ({ data, page }) as any)
        );
    }),
    // we append the result if page was incremented
    // else we just return the result
    scan((acc: any, curr: any) => curr.page === 0 ? curr.data : acc.concat(curr.data), [])
);

Metoden this.apolloWrapper.selectMany er en metode, vi har oprettet, som blot laver en abonnement over websocket til et bestemt dataslice. Lad os sige, at vi havde page=0, så vil this.apolloWrapper.selectMany returnere et abonnement til det første dataslice, og når page=1, returnerer this.apolloWrapper.selectMany et abonnement til det andet dataslice.

Problemet opstår, når vi indlæser den anden side og foretager en ændring af elementerne i det første udsnit, scan udløses derefter, og elementerne tilføjes igen til slutningen af listen.