reverts to the double-queue approach to minimize setTimeout() calls

jacoscaz · Aug 22, 2023 · b58c187 · b58c187
1 parent 29baab5
commit b58c187
Show file tree

Hide file tree

Showing 4 changed files with 21 additions and 25 deletions.
diff --git a/src/InstantRelay.spec.ts b/src/InstantRelay.spec.ts
@@ -89,8 +89,7 @@ describe('instant-relay', () => {
         return (message) => new Promise(() => {});
       }, opts);
       for (let i = 0; i < 4; i += 1) {
-        // @ts-ignore
-        ir.nodeMap.get('r')!.push({ id: i + '', type: 'msg' });
+        ir.nodeMap.get('r')!.queue.push({ id: i + '', type: 'msg' });
       }
     });
 
@@ -105,8 +104,7 @@ describe('instant-relay', () => {
       let currDelta = 0;
       let count = 0;
       const loop = () => {
-        // @ts-ignore
-        ir.nodeMap.get('r')!.push({ id: count + '', type: 'msg' }).then(() => {
+        ir.nodeMap.get('r')!.queue.push({ id: count + '', type: 'msg' }).then(() => {
           currTstmp = Date.now();
           currDelta = currTstmp - prevTstmp;
           if (count > 0) {
@@ -150,8 +148,7 @@ describe('instant-relay', () => {
       }, {});
 
       setImmediate(() => {
-        // @ts-ignore
-        ir.nodeMap.get('a')!.push({ id: '0', type: 'greeting' });
+        ir.nodeMap.get('a')!.queue.push({ id: '0', type: 'greeting' });
       });
 
     });
@@ -181,8 +178,7 @@ describe('instant-relay', () => {
       }, {});
 
       setImmediate(() => {
-        // @ts-ignore
-        ir.nodeMap.get('a')!.push({ id: '0', type: 'greeting' }, () => {});
+        ir.nodeMap.get('a')!.queue.push({ id: '0', type: 'greeting' });
       });
 
     });

diff --git a/src/InstantRelay.ts b/src/InstantRelay.ts
@@ -4,8 +4,8 @@ import { makeNode } from './node';
 
 export class InstantRelay<M> {
 
-  private readonly nodeMap: Map<string, InternalNode<M>>;
-  private readonly nodeArr: InternalNode<M>[];
+  readonly nodeMap: Map<string, InternalNode<M>>;
+  readonly nodeArr: InternalNode<M>[];
 
   constructor() {
     this.nodeMap = new Map();

diff --git a/src/node.ts b/src/node.ts
@@ -20,13 +20,13 @@ const makeSend = <M>(nodeMap: Map<string, InternalNode<M>>, senderId: string): S
       crashWithError(new Error(`Unknown node with id "${recipientId}"`));
       return;
     }
-    await recipient.push(message);
+    await recipient.queue.push(message);
   };
 };
 
 const makeBroadcast = <M>(nodeArr: InternalNode<M>[], senderId: string): BroadcastMessage<M> => {
   return async (msg: M) => {
-    await Promise.all(nodeArr.map(node => node.id !== senderId ? node.push(msg) : null));
+    await Promise.all(nodeArr.map(node => node.id !== senderId ? node.queue.push(msg) : null));
   };
 };
 
@@ -51,27 +51,25 @@ export const makeNode = <M, O extends {}>(
   const broadcast = makeBroadcast(nodeArr, id);
   const handleMessage = factory(send, broadcast, { ...opts, id });
 
-  let queueLength = 0;
+  const handlingQueue = fastq.promise(handleMessage, concurrency);
 
-  const queue = fastq.promise(async (msg: M) => {
-    await handleMessage(msg);
-  }, concurrency);
-
-  queue.error((err, task) => {
+  handlingQueue.error((err, task) => {
     if (err !== null) {
       crashWithError(err);
       return;
     }
   });
 
-  const push = async (msg: M) => {
-    queue.push(msg);
-    if ((queueLength = queue.length()) >= highWaterMark) {
-      await wait(throttle(queueLength));
+  let handlingQueueLength = 0;
+
+  const incomingQueue = fastq.promise(async (msg: M) => {
+    handlingQueue.push(msg);
+    if ((handlingQueueLength = handlingQueue.length()) >= highWaterMark) {
+      await wait(throttle(handlingQueueLength));
     }
-  };
+  }, 1);
 
-  const node: InternalNode<M> = { id, push };
+  const node: InternalNode<M> = { id, queue: incomingQueue };
 
   nodeMap.set(id, node);
   nodeArr.push(node);

diff --git a/src/types.ts b/src/types.ts
@@ -1,4 +1,6 @@
 
+import fastq from 'fastq';
+
 export interface HandleMessage<M> {
   (message: M): Promise<void>;
 }
@@ -23,5 +25,5 @@ export interface AddNodeOpts {
 
 export interface InternalNode<M> {
   readonly id: string;
-  readonly push: (msg: M) => Promise<void>;
+  readonly queue: fastq.queueAsPromised<M>;
 }