changelog.md

@@ -1,13 +1,5 @@
 # Changelog
 
-## 2025-04-25 - 5.16.0 - feat(watcher)
-Enhance change stream watchers with buffering and EventEmitter support; update dependency versions
-
-- Bumped smartmongo from ^2.0.11 to ^2.0.12 and smartrx from ^3.0.7 to ^3.0.10
-- Upgraded @tsclass/tsclass to ^9.0.0 and mongodb to ^6.16.0
-- Refactored the watch API to accept additional options (bufferTimeMs, fullDocument) for improved change stream handling
-- Modified SmartdataDbWatcher to extend EventEmitter and support event notifications
-
 ## 2025-04-24 - 5.15.1 - fix(cursor)
 Improve cursor usage documentation and refactor getCursor API to support native cursor modifiers
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartdata",
-  "version": "5.16.0",
+  "version": "5.15.1",
   "private": false,
   "description": "An advanced library for NoSQL data organization and manipulation using TypeScript with support for MongoDB, data validation, collections, and custom data types.",
   "main": "dist_ts/index.js",
@@ -26,23 +26,23 @@
     "@push.rocks/lik": "^6.0.14",
     "@push.rocks/smartdelay": "^3.0.1",
     "@push.rocks/smartlog": "^3.0.2",
-    "@push.rocks/smartmongo": "^2.0.12",
+    "@push.rocks/smartmongo": "^2.0.11",
     "@push.rocks/smartpromise": "^4.0.2",
-    "@push.rocks/smartrx": "^3.0.10",
+    "@push.rocks/smartrx": "^3.0.7",
     "@push.rocks/smartstring": "^4.0.15",
     "@push.rocks/smarttime": "^4.0.6",
     "@push.rocks/smartunique": "^3.0.8",
     "@push.rocks/taskbuffer": "^3.1.7",
-    "@tsclass/tsclass": "^9.0.0",
+    "@tsclass/tsclass": "^8.2.0",
-    "mongodb": "^6.16.0"
+    "mongodb": "^6.15.0"
   },
   "devDependencies": {
     "@git.zone/tsbuild": "^2.3.2",
     "@git.zone/tsrun": "^1.2.44",
     "@git.zone/tstest": "^1.0.77",
     "@push.rocks/qenv": "^6.0.5",
-    "@push.rocks/tapbundle": "^5.6.3",
+    "@push.rocks/tapbundle": "^5.6.2",
-    "@types/node": "^22.15.2"
+    "@types/node": "^22.14.0"
   },
   "files": [
     "ts/**/*",

test/test.watch.ts

@@ -60,43 +60,6 @@ tap.test('should watch a collection', async (toolsArg) => {
   await done.promise;
 });
 
-// ======= New tests for EventEmitter and buffering support =======
-tap.test('should emit change via EventEmitter', async (tools) => {
-  const done = tools.defer();
-  const watcher = await House.watch({});
-  watcher.on('change', async (houseArg) => {
-    // Expect a House instance
-    expect(houseArg).toBeDefined();
-    // Clean up
-    await watcher.stop();
-    done.resolve();
-  });
-  // Trigger an insert to generate a change event
-  const h = new House();
-  await h.save();
-  await done.promise;
-});
-
-tap.test('should buffer change events when bufferTimeMs is set', async (tools) => {
-  const done = tools.defer();
-  // bufferTimeMs collects events into arrays every 50ms
-  const watcher = await House.watch({}, { bufferTimeMs: 50 });
-  let received: House[];
-  watcher.changeSubject.subscribe(async (batch: House[]) => {
-    if (batch && batch.length > 0) {
-      received = batch;
-      await watcher.stop();
-      done.resolve();
-    }
-  });
-  // Rapidly insert multiple docs
-  const docs = [new House(), new House(), new House()];
-  for (const doc of docs) await doc.save();
-  await done.promise;
-  // All inserts should be in one buffered batch
-  expect(received.length).toEqual(docs.length);
-});
-
 // =======================================
 // close the database connection
 // =======================================

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartdata',
-  version: '5.16.0',
+  version: '5.15.1',
   description: 'An advanced library for NoSQL data organization and manipulation using TypeScript with support for MongoDB, data validation, collections, and custom data types.'
 }

ts/classes.collection.ts

@@ -256,40 +256,24 @@ export class SmartdataCollection<T> {
   }
 
   /**
-   * Watches the collection, returning a SmartdataDbWatcher with RxJS and EventEmitter support.
+   * watches the collection while applying a filter
-   * @param filterObject match filter for change stream
-   * @param opts optional MongoDB ChangeStreamOptions & { bufferTimeMs } to buffer events
-   * @param smartdataDbDocArg document class for instance creation
    */
   public async watch(
     filterObject: any,
-    opts: (plugins.mongodb.ChangeStreamOptions & { bufferTimeMs?: number }) = {},
+    smartdataDbDocArg: typeof SmartDataDbDoc,
-    smartdataDbDocArg?: typeof SmartDataDbDoc,
   ): Promise<SmartdataDbWatcher> {
     await this.init();
-    // Extract bufferTimeMs from options
-    const { bufferTimeMs, fullDocument, ...otherOptions } = opts || {};
-    // Determine fullDocument behavior: default to 'updateLookup'
-    const changeStreamOptions: plugins.mongodb.ChangeStreamOptions = {
-      ...otherOptions,
-      fullDocument:
-        fullDocument === undefined
-          ? 'updateLookup'
-          : fullDocument === true
-          ? 'updateLookup'
-          : fullDocument,
-    } as any;
-    // Build pipeline with match if provided
-    const pipeline = filterObject ? [{ $match: filterObject }] : [];
     const changeStream = this.mongoDbCollection.watch(
-      pipeline,
+      [
-      changeStreamOptions,
+        {
-    );
+          $match: filterObject,
-    const smartdataWatcher = new SmartdataDbWatcher(
+        },
-      changeStream,
+      ],
-      smartdataDbDocArg,
+      {
-      { bufferTimeMs },
+        fullDocument: 'updateLookup',
+      },
     );
+    const smartdataWatcher = new SmartdataDbWatcher(changeStream, smartdataDbDocArg);
     await smartdataWatcher.readyDeferred.promise;
     return smartdataWatcher;
   }

ts/classes.doc.ts

@@ -305,20 +305,13 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
    * @param filterArg
    * @param forEachFunction
    */
-  /**
-   * Watch the collection for changes, with optional buffering and change stream options.
-   * @param filterArg MongoDB filter to select which changes to observe
-   * @param opts optional ChangeStreamOptions plus bufferTimeMs
-   */
   public static async watch<T>(
     this: plugins.tsclass.typeFest.Class<T>,
     filterArg: plugins.tsclass.typeFest.PartialDeep<T>,
-    opts?: plugins.mongodb.ChangeStreamOptions & { bufferTimeMs?: number },
+  ) {
-  ): Promise<SmartdataDbWatcher<T>> {
     const collection: SmartdataCollection<T> = (this as any).collection;
     const watcher: SmartdataDbWatcher<T> = await collection.watch(
       convertFilterForMongoDb(filterArg),
-      opts || {},
       this as any,
     );
     return watcher;

ts/classes.watcher.ts

@@ -1,73 +1,37 @@
 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>;
-  private rawSubject: plugins.smartrx.rxjs.Subject<T>;
+
-  /** Emits change documents (or arrays of documents if buffered) */
+  public changeSubject = new plugins.smartrx.rxjs.Subject<T>();
-  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) => {
-      let docInstance: T = null;
+      if (!item.fullDocument) {
-      if (item.fullDocument) {
+        this.changeSubject.next(null);
-        docInstance = smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(
+        return;
-          item.fullDocument
-        ) as any as T;
       }
-      // Notify subscribers
+      this.changeSubject.next(
-      this.rawSubject.next(docInstance);
+        smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(item.fullDocument) as any as T,
-      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();
   }
 }