smartdata/ts/smartdata.classes.watcher.ts

import { SmartDataDbDoc } from './smartdata.classes.doc.js';
import * as plugins from './smartdata.plugins.js';

/**
 * a wrapper for the native mongodb cursor. Exposes better
 */
export class SmartdataDbWatcher<T = any> {
  // STATIC
  public readyDeferred = plugins.smartpromise.defer();

  // INSTANCE
  private changeStream: plugins.mongodb.ChangeStream<T>;

  public changeSubject = new plugins.smartrx.rxjs.Subject<T>();
  constructor(
    changeStreamArg: plugins.mongodb.ChangeStream<T>,
    smartdataDbDocArg: typeof SmartDataDbDoc
  ) {
    this.changeStream = changeStreamArg;
    this.changeStream.on('change', async (item: any) => {
      if (!item.fullDocument) {
        this.changeSubject.next(null);
        return;
      }
      this.changeSubject.next(
        smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(item.fullDocument) as any as T
      );
    });
    plugins.smartdelay.delayFor(0).then(() => {
      this.readyDeferred.resolve();
    });
  }

  public async close() {
    await this.changeStream.close();
  }
}