Бесконечная прокрутка и мгновенные снимки обеспечивает отображение нескольких дубликатов данных в хранилище

Question

Я пытаюсь реализовать бесконечную прокрутку, а также адаптировать обновления в реальном времени к пользовательскому интерфейсу.

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

Я изменил предоставляемые услуги в соответствии с потребностями, приведенными ниже.

pagination.service.ts

import { Injectable } from "@angular/core";
import { AngularFirestore, AngularFirestoreCollection } from "angularfire2/firestore";
import { BehaviorSubject } from "rxjs/BehaviorSubject";
import { Observable } from "rxjs/Observable";
import "rxjs/add/operator/do";
import "rxjs/add/operator/scan";
import "rxjs/add/operator/take";
import { QueryConfig } from "../interfaces/QueryConfig";

@Injectable()
export class GeneralPaginationService {

  // Source data
  private _done = new BehaviorSubject(false);
  private _loading = new BehaviorSubject(false);
  private _data = new BehaviorSubject([]);

  private query: QueryConfig;

  // Observable data
  data: Observable<any>;
  done: Observable<boolean> = this._done.asObservable();
  loading: Observable<boolean> = this._loading.asObservable();
  reset: boolean;

  constructor(private afs: AngularFirestore) { }

  // Initial query sets options and defines the Observable
  // passing opts will override the defaults
  init(path: string, field: string, limit: number,  reset: boolean, opts?: any, where?: any) {
    this.reset = reset;
    if (reset){
      this._data = new BehaviorSubject([]);
      this._done.next(false);
    }
    this.query = {
      path,
      field,
      where,
      limit: limit,
      reverse: false,
      prepend: false,
      ...opts
    };
    const first = this.afs.collection<any>(this.query.path, ref => {
      let query: any = ref;
      if (this.query.where !== undefined && this.query.where !== null && this.query.where.length > 0) {
        this.query.where.forEach(element => {
          query = query.where(element.whereField, element.whereOp, element.whereVal);
        });
      }
      // query = query.orderBy(this.query.field, this.query.reverse ? 'desc' : 'asc');
      query = query.limit(this.query.limit);
      return query;
    });
    this.mapAndUpdate(first);

    // Create the observable array for consumption in components
    this.data = this._data.asObservable().scan( (acc, val) => {
        return this.query.prepend ? val.concat(acc) : acc.concat(val);
    });
  }

  // Retrieves additional data from firestore
  more() {
    if (this._done.value || this._loading.value) { return; }
    const cursor = this.getCurrentCursor();
    const more = this.afs.collection<any>(this.query.path, ref => {
      let query: any = ref;
      if (this.query.where !== undefined && this.query.where !== null && this.query.where.length > 0) {
        this.query.where.forEach(element => {
          query = query.where(element.whereField, element.whereOp, element.whereVal);
        });
      }
      // query = query.orderBy(this.query.field, 'desc');
      query = query.limit(this.query.limit);
      query = query.startAfter(cursor);
      return query;
    });
    this.mapAndUpdate(more);
  }


  // Determines the doc snapshot to paginate query
  private getCurrentCursor() {
    const current = this._data.value;
    if (current.length) {
      return this.query.prepend ? current[0].doc : current[current.length - 1].doc;
    }
    return null;
  }

  // Maps the snapshot to usable format the updates source
  private mapAndUpdate(col: AngularFirestoreCollection<any>) {
    if (this._done.value || this._loading.value) { return; }
    // loading
    this._loading.next(true);

    // Map snapshot with doc ref (needed for cursor)
    return col.snapshotChanges().
      do(arr => {
        let values = arr.map(snap => {
          const data = snap.payload.doc.data();
          const doc = snap.payload.doc;
          return { ...data, doc };
        });
        // If prepending, reverse the batch order
        values = this.query.prepend ? values.reverse() : values;
        // update source with new values, done loading
        this._data.next(values);
        this._loading.next(false);
        // no more values, mark done
        if (!values.length) {
          this._done.next(true);
        }
    })
    .take(1)
    .subscribe();
  }
}

Все работает хорошо, но если произойдут какие-либо обновления, это не отразится на внешнем интерфейсе. Ниже приведен фрагмент кода для snapshotChanges() с take(1)

  // Maps the snapshot to usable format the updates source
  private mapAndUpdate(col: AngularFirestoreCollection<any>) {
    if (this._done.value || this._loading.value) { return; }
    // loading
    this._loading.next(true);

    // Map snapshot with doc ref (needed for cursor)
    return col.snapshotChanges().
      do(arr => {
        let values = arr.map(snap => {
          const data = snap.payload.doc.data();
          const doc = snap.payload.doc;
          return { ...data, doc };
        });
        // If prepending, reverse the batch order
        values = this.query.prepend ? values.reverse() : values;
        // update source with new values, done loading
        this._data.next(values);
        this._loading.next(false);
        // no more values, mark done
        if (!values.length) {
          this._done.next(true);
        }
    })
    .take(1)
    .subscribe();
  }

Теперь, если я удалю .take(1), он подпишется на изменения, но затем продолжит добавлять дублирующиеся данные в пользовательский интерфейс.

Как я могу преодолеть это? Как правильно это реализовать?