refactor multiplexer to typescript

package.json

@@ -17,6 +17,8 @@
     "@babel/eslint-parser": "^7.21.3",
     "@babel/eslint-plugin": "^7.19.1",
     "@babel/preset-react": "^7.18.6",
+    "@types/lodash.isempty": "^4.4.9",
+    "@types/meteor": "^2.9.8",
     "@types/node": "^18.16.18",
     "@typescript-eslint/eslint-plugin": "^5.56.0",
     "@typescript-eslint/parser": "^5.56.0",

packages/mongo/mongo_driver.js

@@ -24,6 +24,7 @@ import {
 } from "meteor/minimongo/constants";
 import { Meteor } from "meteor/meteor";
 import { ObserveHandle } from './observe_handle';
+import { ObserveMultiplexer } from './observe_multiplex';
 
 MongoInternals = {};
 

packages/mongo/observe_handle.ts

@@ -1,37 +1,46 @@
+import { ObserveHandleCallback, ObserveMultiplexer } from './observe_multiplex';
+
 let nextObserveHandleId = 1;
 
+export type ObserveHandleCallbackInternal = '_added' | '_addedBefore' | '_changed' | '_movedBefore' | '_removed';
+
 // When the callbacks do not mutate the arguments, we can skip a lot of data clones
 export class ObserveHandle {
-    _stopped: boolean;
-    _id: number;
-    _multiplexer: any;
-    nonMutatingCallbacks: boolean;
-
-    constructor(multiplexer: any, callbacks: any, nonMutatingCallbacks: boolean) {
-        this._multiplexer = multiplexer;
-
-        multiplexer.callbackNames().forEach((name) => {
-            if (callbacks[name]) {
-                this['_' + name] = callbacks[name];
-            } else if (name === "addedBefore" && callbacks.added) {
-                this._addedBefore = async function (id, fields, before) {
-                    await callbacks.added(id, fields);
-                };
-            }
-        });
-
-        this._stopped = false;
-        this._id = nextObserveHandleId++;
-        this.nonMutatingCallbacks = nonMutatingCallbacks;
-    }
-
-    async stop() {
-        if (this._stopped) return;
-        this._stopped = true;
-        await this._multiplexer.removeHandle(this._id);
-    }
-
-    _addedBefore(id: any, fields: any, before: any) {
-        throw new Error("Method not implemented.");
-    }
+  _id: number;
+  _multiplexer: ObserveMultiplexer;
+  nonMutatingCallbacks: boolean;
+  _stopped: boolean;
+
+  _added?: (...args: any[]) => void;
+  _addedBefore?: (...args: any[]) => void;
+  _changed?: (...args: any[]) => void;
+  _movedBefore?: (...args: any[]) => void;
+  _removed?: (...args: any[]) => void;
+
+  constructor(multiplexer: any, callbacks: Record<ObserveHandleCallback, any>, nonMutatingCallbacks: boolean) {
+    this._multiplexer = multiplexer;
+
+    multiplexer.callbackNames().forEach((name: ObserveHandleCallback) => {
+      if (callbacks[name]) {
+        this[`_${name}` as ObserveHandleCallbackInternal] = callbacks[name];
+        return;
+      }
+
+      if (name === "addedBefore" && callbacks.added) {
+        this._addedBefore = async function (id, fields, before) {
+          await callbacks.added(id, fields);
+        };
+      }
+    });
+
+    this._stopped = false;
+    this._id = nextObserveHandleId++;
+    this.nonMutatingCallbacks = nonMutatingCallbacks;
+  }
+
+  async stop() {
+    if (this._stopped) return;
+    this._stopped = true;
+    await this._multiplexer.removeHandle(this._id);
+  }
 }