feat(watcher): Enhance change stream watchers with buffering and EventEmitter support; update dependency versions

ts/classes.watcher.ts

@@ -1,37 +1,73 @@
 import { SmartDataDbDoc } from './classes.doc.js';
 import * as plugins from './plugins.js';
+import { EventEmitter } from 'events';
 
 /**
  * a wrapper for the native mongodb cursor. Exposes better
  */
-export class SmartdataDbWatcher<T = any> {
+/**
+ * Wraps a MongoDB ChangeStream with RxJS and EventEmitter support.
+ */
+export class SmartdataDbWatcher<T = any> extends EventEmitter {
   // STATIC
   public readyDeferred = plugins.smartpromise.defer();
 
   // INSTANCE
   private changeStream: plugins.mongodb.ChangeStream<T>;
-
-  public changeSubject = new plugins.smartrx.rxjs.Subject<T>();
+  private rawSubject: plugins.smartrx.rxjs.Subject<T>;
+  /** Emits change documents (or arrays of documents if buffered) */
+  public changeSubject: any;
+  /**
+   * @param changeStreamArg native MongoDB ChangeStream
+   * @param smartdataDbDocArg document class for instance creation
+   * @param opts.bufferTimeMs optional milliseconds to buffer events via RxJS
+   */
   constructor(
     changeStreamArg: plugins.mongodb.ChangeStream<T>,
     smartdataDbDocArg: typeof SmartDataDbDoc,
+    opts?: { bufferTimeMs?: number },
   ) {
+    super();
+    this.rawSubject = new plugins.smartrx.rxjs.Subject<T>();
+    // Apply buffering if requested
+    if (opts && opts.bufferTimeMs) {
+      this.changeSubject = this.rawSubject.pipe(plugins.smartrx.rxjs.ops.bufferTime(opts.bufferTimeMs));
+    } else {
+      this.changeSubject = this.rawSubject;
+    }
     this.changeStream = changeStreamArg;
     this.changeStream.on('change', async (item: any) => {
-      if (!item.fullDocument) {
-        this.changeSubject.next(null);
-        return;
+      let docInstance: T = null;
+      if (item.fullDocument) {
+        docInstance = smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(
+          item.fullDocument
+        ) as any as T;
       }
-      this.changeSubject.next(
-        smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(item.fullDocument) as any as T,
-      );
+      // Notify subscribers
+      this.rawSubject.next(docInstance);
+      this.emit('change', docInstance);
     });
+    // Signal readiness after one tick
     plugins.smartdelay.delayFor(0).then(() => {
       this.readyDeferred.resolve();
     });
   }
 
-  public async close() {
+  /**
+   * Close the change stream, complete the RxJS subject, and remove listeners.
+   */
+  public async close(): Promise<void> {
+    // Close MongoDB ChangeStream
     await this.changeStream.close();
+    // Complete the subject to teardown any buffering operators
+    this.rawSubject.complete();
+    // Remove all EventEmitter listeners
+    this.removeAllListeners();
+  }
+  /**
+   * Alias for close(), matching README usage
+   */
+  public async stop(): Promise<void> {
+    return this.close();
   }
 }