target sender

Author: darrachequesne

lib/index.ts

@@ -5,6 +5,7 @@ import {
   ClusterMessage,
   ServerId,
   ClusterResponse,
+  MessageType,
 } from "socket.io-adapter";
 import debugModule from "debug";
 
@@ -32,13 +33,15 @@ export class NodeClusterAdapter extends ClusterAdapterWithHeartbeat {
   constructor(nsp: any, opts: ClusterAdapterOptions = {}) {
     super(nsp, opts);
     process.on("message", (message: any) => {
-      if (message?.source !== MESSAGE_SOURCE) {
-        debug("ignore unknown source");
+      const isValidSource = message?.source === MESSAGE_SOURCE;
+      if (!isValidSource) {
+        debug("[%s] ignore unknown source", this.uid);
+        return;
       }
 
-      // note: this check should be done in the onMessage handler
+      // note: this check should be done in the onMessage() handler
       if (message.nsp !== this.nsp.name) {
-        debug("ignore other namespace");
+        debug("[%s] ignore other namespace", this.uid);
         return;
       }
 
@@ -49,42 +52,65 @@ export class NodeClusterAdapter extends ClusterAdapterWithHeartbeat {
     this.init();
   }
 
-  override doPublish(message: ClusterMessage & { source: string }) {
+  protected override doPublish(message: ClusterMessage & { source: string }) {
     message.source = MESSAGE_SOURCE;
 
     process.send(message, null, {}, ignoreError);
 
     return Promise.resolve(""); // connection state recovery is not supported
   }
 
-  override doPublishResponse(
+  protected override doPublishResponse(
     requesterUid: ServerId,
-    response: ClusterResponse & { source: string },
+    response: ClusterResponse & { source: string; requesterUid: string },
   ) {
     response.source = MESSAGE_SOURCE;
+    response.requesterUid = requesterUid;
 
     process.send(response, null, {}, ignoreError);
 
     return Promise.resolve();
   }
 }
 
+const UIDS = Symbol("uids");
+
 export function setupPrimary() {
   cluster.on("message", (worker, message) => {
     const isValidSource = message?.source === MESSAGE_SOURCE;
     if (!isValidSource) {
       return;
     }
 
-    const emitterIdAsString = String(worker.id);
-    // emit to all workers but the requester
-    for (const workerId in cluster.workers) {
-      if (
-        hasOwnProperty.call(cluster.workers, workerId) &&
-        workerId !== emitterIdAsString
-      ) {
-        cluster.workers[workerId].send(message, null, ignoreError);
-      }
+    // store the requester's uids (one per namespace) so that the response can be sent specifically to them
+    worker[UIDS] = worker[UIDS] || new Set();
+    worker[UIDS].add(message.uid);
+
+    switch (message.type) {
+      case MessageType.FETCH_SOCKETS_RESPONSE:
+      case MessageType.SERVER_SIDE_EMIT_RESPONSE:
+        const requesterUid = message.requesterUid;
+        for (const workerId in cluster.workers) {
+          if (
+            hasOwnProperty.call(cluster.workers, workerId) &&
+            cluster.workers[workerId][UIDS]?.has(requesterUid)
+          ) {
+            cluster.workers[workerId].send(message, null, ignoreError);
+            break;
+          }
+        }
+        break;
+      default:
+        const emitterIdAsString = String(worker.id);
+        // emit to all workers but the requester
+        for (const workerId in cluster.workers) {
+          if (
+            hasOwnProperty.call(cluster.workers, workerId) &&
+            workerId !== emitterIdAsString
+          ) {
+            cluster.workers[workerId].send(message, null, ignoreError);
+          }
+        }
     }
   });
 }