nighthouse - npm Package Compare versions

Comparing version 2.1.6 to 3.0.0

out/common/lighthouse.d.ts

@@ -16,2 +16,6 @@ import { Transport } from "./transport";
     private responseHandlers;
+    /** Out-of-order received messages, e.g. if a response is faster than the response handler is registered. */
+    private outOfOrderMessages;
+    /** Active streams, keyed by the path array encoded as a JSON string. */
+    private streamsByPath;
     /** Whether the transport has been closed. */
@@ -25,3 +29,3 @@ private isClosed;
     /** Streams the user's model (including e.g. key/controller events). */
-    streamModel(user?: string): AsyncIterable<ServerMessage<unknown>>;
+    streamModel(user?: string): Promise<AsyncIterable<ServerMessage<unknown>>>;
     /** Fetches lamp server metrics. */
@@ -48,5 +52,5 @@ getLaserMetrics(): Promise<ServerMessage<LaserMetrics>>;
     /** Performs a single request to the given path with the given payload. */
-    perform<T>(verb: SingleVerb, path: string[], payload: T): Promise<ServerMessage<unknown>>;
+    perform<T>(verb: SingleVerb, path: string[], payload?: T): Promise<ServerMessage<unknown>>;
     /** Performs a streaming request to the given path with the given payload. */
-    stream<T>(path: string[], payload: T): AsyncIterable<ServerMessage<unknown>>;
+    stream<T>(path: string[], payload?: T): Promise<AsyncIterable<ServerMessage<unknown>>>;
     /** Sends a request. */
@@ -60,2 +64,3 @@ private sendRequest;
     private receiveStreaming;
+    private receiveResponse;
     /** Handles a server message. */
@@ -62,0 +67,0 @@ private handle;

out/common/lighthouse.js

@@ -20,2 +20,6 @@ "use strict";
         this.responseHandlers = new Map();
+        /** Out-of-order received messages, e.g. if a response is faster than the response handler is registered. */
+        this.outOfOrderMessages = new Map();
+        /** Active streams, keyed by the path array encoded as a JSON string. */
+        this.streamsByPath = new Map();
         /** Whether the transport has been closed. */
@@ -44,4 +48,4 @@ this.isClosed = false;
     /** Streams the user's model (including e.g. key/controller events). */
-    async *streamModel(user = this.auth.USER) {
-        yield* this.stream(['user', user, 'model'], {});
+    async streamModel(user = this.auth.USER) {
+        return this.stream(['user', user, 'model']);
     }
@@ -62,19 +66,19 @@ /** Fetches lamp server metrics. */
     async create(path) {
-        return await this.perform('CREATE', path, {});
+        return await this.perform('CREATE', path);
     }
     /** Deletes a resource at the given path. Requires DELETE permission. */
     async delete(path) {
-        return await this.perform('DELETE', path, {});
+        return await this.perform('DELETE', path);
     }
     /** Creates a directory at the given path. Requires CREATE permission. */
     async mkdir(path) {
-        return await this.perform('MKDIR', path, {});
+        return await this.perform('MKDIR', path);
     }
     /** Lists the directory tree at the given path. Requires READ permission. */
     async list(path) {
-        return await this.perform('LIST', path, {});
+        return await this.perform('LIST', path);
     }
     /** Gets the resource at the given path. Requires READ permission. */
     async get(path) {
-        return await this.perform('GET', path, {});
+        return await this.perform('GET', path);
     }
@@ -91,16 +95,45 @@ /** Links the given source to the given destination path. Requires WRITE permission for the destination and READ for the source. */
     async perform(verb, path, payload) {
-        const requestId = await this.sendRequest(verb, path, payload);
+        const requestId = await this.sendRequest(verb, path, payload !== null && payload !== void 0 ? payload : {});
         return await this.receiveSingle(requestId);
     }
     /** Performs a streaming request to the given path with the given payload. */
-    async *stream(path, payload) {
-        const requestId = await this.sendRequest('STREAM', path, payload);
-        try {
-            this.logger.debug(() => `Starting stream from ${JSON.stringify(path)}...`);
-            yield* this.receiveStreaming(requestId);
+    async stream(path, payload) {
+        const key = JSON.stringify(path);
+        let requestId;
+        if (this.streamsByPath.has(key) && this.streamsByPath.get(key).requestIds.length > 0) {
+            // This path is already being streamed, we only need to add a handler and
+            // don't send a `STREAM` request. This request id in this case is only
+            // for tracking our client-side handler and not sent to the server.
+            requestId = this.requestId++;
+            const stream = this.streamsByPath.get(key);
+            this.logger.trace(() => `Adding new demuxed stream ${requestId} for ${JSON.stringify(path)} (also streaming this resource: ${JSON.stringify(stream.requestIds)})...`);
+            this.streamsByPath.set(key, { ...stream, requestIds: [...stream.requestIds, requestId] });
         }
-        finally {
-            this.logger.debug(() => `Stopping stream from ${JSON.stringify(path)}...`);
-            await this.sendRequest('STOP', path, {});
+        else {
+            // This path has not been streamed yet.
+            requestId = await this.sendRequest('STREAM', path, payload !== null && payload !== void 0 ? payload : {});
+            this.logger.trace(() => `Registering new stream ${requestId} from ${JSON.stringify(path)}...`);
+            this.streamsByPath.set(key, { requestIds: [requestId] });
         }
+        return (async function* () {
+            try {
+                this.logger.debug(() => `Starting stream from ${JSON.stringify(path)}...`);
+                yield* this.receiveStreaming(requestId);
+            }
+            finally {
+                const sr = this.streamsByPath.get(key);
+                if (sr.requestIds.length > 1) {
+                    // This path is still being streamed by another consumer
+                    // TODO: Assert that sr.requestIds contains our request id (once)
+                    this.streamsByPath.set(key, { requestIds: sr.requestIds.filter(id => id !== requestId) });
+                }
+                else {
+                    // We were the last consumer to stream this path, so we can stop it
+                    // TODO: Assert that length === 1 and that this is exactly our request id
+                    this.logger.debug(() => `Stopping stream from ${JSON.stringify(path)}...`);
+                    await this.sendRequest('STOP', path, {});
+                    this.streamsByPath.delete(key);
+                }
+            }
+        }).bind(this)();
     }
@@ -135,7 +168,6 @@ /** Sends a request. */
         }
-        const deferred = new deferred_1.Deferred();
         this.logger.trace(`Registering handler for ${id}`);
-        this.responseHandlers.set(id, deferred);
+        const responsePromise = this.receiveResponse(id);
         try {
-            const message = await deferred.promise;
+            const message = await responsePromise;
             if (message.RNUM === 200) {
@@ -162,7 +194,6 @@ return message;
                 this.logger.trace(`Pushing promise for next response to request ${id}`);
-                const responseHandler = new deferred_1.Deferred();
-                this.responseHandlers.set(id, responseHandler);
+                const responsePromise = this.receiveResponse(id);
                 const nextDeferred = new deferred_1.Deferred();
                 nextPromises.push(nextDeferred.promise);
-                responseHandler.promise
+                responsePromise
                     .then(response => {
@@ -190,4 +221,22 @@ pushPromise();
     }
+    receiveResponse(id) {
+        var _a;
+        const responseHandler = new deferred_1.Deferred();
+        const outOfOrderMessage = (_a = this.outOfOrderMessages.get(id)) === null || _a === void 0 ? void 0 : _a.shift();
+        if (outOfOrderMessage) {
+            // Response was already received, return it
+            responseHandler.resolve(outOfOrderMessage);
+            if (this.outOfOrderMessages.get(id).length === 0) {
+                this.outOfOrderMessages.delete(id);
+            }
+        }
+        else {
+            // Register handler to await future response
+            this.responseHandlers.set(id, responseHandler);
+        }
+        return responseHandler.promise;
+    }
     /** Handles a server message. */
     async handle(message) {
+        var _a, _b, _c;
         const responseHandler = this.responseHandlers.get(message.REID);
@@ -199,4 +248,8 @@ if (responseHandler) {
         else {
-            // No response handler exists, warn about it.
-            this.logger.warning(`Got unhandled event for id ${message.REID}`);
+            // No response handler exists (yet?), warn about it.
+            const demuxedIds = (_b = (_a = [...this.streamsByPath.values()].find(s => s.requestIds.includes(message.REID))) === null || _a === void 0 ? void 0 : _a.requestIds) !== null && _b !== void 0 ? _b : [message.REID];
+            this.logger.warning(() => `Got out-of-order event for id ${message.REID}${demuxedIds.length > 1 ? ` (demuxed to ${JSON.stringify(demuxedIds)})` : ''}`);
+            for (const id of demuxedIds) {
+                this.outOfOrderMessages.set(id, [...((_c = this.outOfOrderMessages.get(id)) !== null && _c !== void 0 ? _c : []), message]);
+            }
         }
@@ -203,0 +256,0 @@ }

package.json

 {
   "name": "nighthouse",
-  "version": "2.1.6",
+  "version": "3.0.0",
   "description": "Lightweight Project Lighthouse client for JavaScript",
@@ -11,2 +11,3 @@ "workspaces": [
     "watch": "tsc -b tsconfig.json -w",
+    "test": "jest",
     "prepare": "npm run build",
@@ -22,2 +23,5 @@ "generate-docs": "typedoc"
     "@types/node": "^18.11.18",
+    "jest": "^29.7.0",
+    "ts-jest": "^29.2.5",
+    "tsx": "^4.19.1",
     "typedoc": "^0.24.8",
@@ -24,0 +28,0 @@ "typescript": "^5.1.6"

src/common/lighthouse.ts

@@ -9,2 +9,11 @@ import { Transport } from "./transport";
 
+/**
+ * Bookkeeping data for the stream of a resource. Each resource only has one
+ * such instance, we demultiplex multiple logical streams of the same resource.
+ */
+interface ResourceStream {
+  /** All request ids that we are listening on for this stream. */
+  requestIds: number[];
+}
+
 /** A connection to the lighthouse. */
@@ -18,2 +27,8 @@ export class Lighthouse {
 
+  /** Out-of-order received messages, e.g. if a response is faster than the response handler is registered. */
+  private outOfOrderMessages: Map<number, ServerMessage<unknown>[]> = new Map();
+
+  /** Active streams, keyed by the path array encoded as a JSON string. */
+  private streamsByPath: Map<string, ResourceStream> = new Map();
+
   /** Whether the transport has been closed. */
@@ -51,4 +66,4 @@ private isClosed = false;
   /** Streams the user's model (including e.g. key/controller events). */
-  async *streamModel(user: string = this.auth.USER): AsyncIterable<ServerMessage<unknown>> {
-    yield* this.stream(['user', user, 'model'], {});
+  async streamModel(user: string = this.auth.USER): Promise<AsyncIterable<ServerMessage<unknown>>> {
+    return this.stream(['user', user, 'model']);
   }
@@ -73,3 +88,3 @@
   async create(path: string[]): Promise<ServerMessage<unknown>> {
-    return await this.perform('CREATE', path, {});
+    return await this.perform('CREATE', path);
   }
@@ -79,3 +94,3 @@
   async delete(path: string[]): Promise<ServerMessage<unknown>> {
-    return await this.perform('DELETE', path, {});
+    return await this.perform('DELETE', path);
   }
@@ -85,3 +100,3 @@
   async mkdir(path: string[]): Promise<ServerMessage<unknown>> {
-    return await this.perform('MKDIR', path, {});
+    return await this.perform('MKDIR', path);
   }
@@ -91,3 +106,3 @@
   async list(path: string[]): Promise<ServerMessage<unknown>> {
-    return await this.perform('LIST', path, {});
+    return await this.perform('LIST', path);
   }
@@ -97,3 +112,3 @@
   async get(path: string[]): Promise<ServerMessage<unknown>> {
-    return await this.perform('GET', path, {});
+    return await this.perform('GET', path);
   }
@@ -112,4 +127,4 @@
   /** Performs a single request to the given path with the given payload. */
-  async perform<T>(verb: SingleVerb, path: string[], payload: T): Promise<ServerMessage<unknown>> {
-    const requestId = await this.sendRequest(verb, path, payload);
+  async perform<T>(verb: SingleVerb, path: string[], payload?: T): Promise<ServerMessage<unknown>> {
+    const requestId = await this.sendRequest(verb, path, payload ?? {});
     return await this.receiveSingle(requestId);
@@ -119,11 +134,40 @@ }
   /** Performs a streaming request to the given path with the given payload. */
-  async *stream<T>(path: string[], payload: T): AsyncIterable<ServerMessage<unknown>> {
-    const requestId = await this.sendRequest('STREAM', path, payload);
-    try {
-      this.logger.debug(() => `Starting stream from ${JSON.stringify(path)}...`);
-      yield* this.receiveStreaming(requestId);
-    } finally {
-      this.logger.debug(() => `Stopping stream from ${JSON.stringify(path)}...`);
-      await this.sendRequest('STOP', path, {});
+  async stream<T>(path: string[], payload?: T): Promise<AsyncIterable<ServerMessage<unknown>>> {
+    const key = JSON.stringify(path);
+    let requestId: number;
+
+    if (this.streamsByPath.has(key) && this.streamsByPath.get(key).requestIds.length > 0) {
+      // This path is already being streamed, we only need to add a handler and
+      // don't send a `STREAM` request. This request id in this case is only
+      // for tracking our client-side handler and not sent to the server.
+      requestId = this.requestId++;
+      const stream = this.streamsByPath.get(key)
+      this.logger.trace(() => `Adding new demuxed stream ${requestId} for ${JSON.stringify(path)} (also streaming this resource: ${JSON.stringify(stream.requestIds)})...`);
+      this.streamsByPath.set(key, { ...stream, requestIds: [...stream.requestIds, requestId] });
+    } else {
+      // This path has not been streamed yet.
+      requestId = await this.sendRequest('STREAM', path, payload ?? {});
+      this.logger.trace(() => `Registering new stream ${requestId} from ${JSON.stringify(path)}...`);
+      this.streamsByPath.set(key, { requestIds: [requestId] });
     }
+
+    return (async function* () {
+      try {
+        this.logger.debug(() => `Starting stream from ${JSON.stringify(path)}...`);
+        yield* this.receiveStreaming(requestId);
+      } finally {
+        const sr = this.streamsByPath.get(key);
+        if (sr.requestIds.length > 1) {
+          // This path is still being streamed by another consumer
+          // TODO: Assert that sr.requestIds contains our request id (once)
+          this.streamsByPath.set(key, { requestIds: sr.requestIds.filter(id => id !== requestId) });
+        } else {
+          // We were the last consumer to stream this path, so we can stop it
+          // TODO: Assert that length === 1 and that this is exactly our request id
+          this.logger.debug(() => `Stopping stream from ${JSON.stringify(path)}...`);
+          await this.sendRequest('STOP', path, {});
+          this.streamsByPath.delete(key);
+        }
+      }
+    }).bind(this)();
   }
@@ -162,9 +206,7 @@
 
-    const deferred = new Deferred<ServerMessage<unknown>>();
-
     this.logger.trace(`Registering handler for ${id}`);
-    this.responseHandlers.set(id, deferred);
+    const responsePromise = this.receiveResponse(id);
 
     try {
-      const message = await deferred.promise;
+      const message = await responsePromise;
       if (message.RNUM === 200) {
@@ -193,4 +235,3 @@ return message;
 
-        const responseHandler = new Deferred<ServerMessage<unknown>>();
-        this.responseHandlers.set(id, responseHandler);
+        const responsePromise = this.receiveResponse(id);
 
@@ -200,3 +241,3 @@ const nextDeferred = new Deferred<ServerMessage<unknown>>();
 
-        responseHandler.promise
+        responsePromise
           .then(response => {
@@ -225,2 +266,20 @@ pushPromise();
 
+  private receiveResponse(id: number): Promise<ServerMessage<unknown>> {
+    const responseHandler = new Deferred<ServerMessage<unknown>>();
+    const outOfOrderMessage = this.outOfOrderMessages.get(id)?.shift();
+
+    if (outOfOrderMessage) {
+      // Response was already received, return it
+      responseHandler.resolve(outOfOrderMessage);
+      if (this.outOfOrderMessages.get(id).length === 0) {
+        this.outOfOrderMessages.delete(id);
+      }
+    } else {
+      // Register handler to await future response
+      this.responseHandlers.set(id, responseHandler);
+    }
+
+    return responseHandler.promise;
+  }
+
   /** Handles a server message. */
@@ -233,4 +292,8 @@ private async handle(message: ServerMessage<unknown>): Promise<void> {
     } else {
-      // No response handler exists, warn about it.
-      this.logger.warning(`Got unhandled event for id ${message.REID}`);
+      // No response handler exists (yet?), warn about it.
+      const demuxedIds: number[] = [...this.streamsByPath.values()].find(s => s.requestIds.includes(message.REID))?.requestIds ?? [message.REID];
+      this.logger.warning(() => `Got out-of-order event for id ${message.REID}${demuxedIds.length > 1 ? ` (demuxed to ${JSON.stringify(demuxedIds)})` : ''}`);
+      for (const id of demuxedIds) {
+        this.outOfOrderMessages.set(id, [...(this.outOfOrderMessages.get(id) ?? []), message]);
+      }
     }
@@ -237,0 +300,0 @@ }

No alert changes

Improved metrics

Total package byte prevSize

147098

13.13%

Number of package files

82

5.13%

Lines of code

1793

14.94%

Worsened metrics

Dev dependency count

6

100%

No dependency changes