feat: implement the serverSideEmit functionality

This new API replaces the customRequest/customHook methods that were
removed in v6.

Syntax:

```js
// server A
io.serverSideEmit("hello", "world");

// server B
io.on("hello", (arg) => {
  console.log(arg); // prints "world"
});
```

With acknowledgements:

```js
// server A
io.serverSideEmit("hello", "world", (err, responses) => {
  console.log(responses); // prints ["hi"]
});

// server B
io.on("hello", (arg, callback) => {
  callback("hi");
});
```

Related: #370

Author: darrachequesne

lib/index.ts

@@ -17,6 +17,7 @@ enum RequestType {
   REMOTE_LEAVE = 3,
   REMOTE_DISCONNECT = 4,
   REMOTE_FETCH = 5,
+  SERVER_SIDE_EMIT = 6,
 }
 
 interface Request {
@@ -300,6 +301,37 @@ export class RedisAdapter extends Adapter {
         this.pubClient.publish(this.responseChannel, response);
         break;
 
+      case RequestType.SERVER_SIDE_EMIT:
+        if (request.uid === this.uid) {
+          debug("ignore same uid");
+          return;
+        }
+        const withAck = request.requestId !== undefined;
+        if (!withAck) {
+          this.nsp._onServerSideEmit(request.data);
+          return;
+        }
+        let called = false;
+        const callback = (arg) => {
+          // only one argument is expected
+          if (called) {
+            return;
+          }
+          called = true;
+          debug("calling acknowledgement with %j", arg);
+          this.pubClient.publish(
+            this.responseChannel,
+            JSON.stringify({
+              type: RequestType.SERVER_SIDE_EMIT,
+              requestId: request.requestId,
+              data: arg,
+            })
+          );
+        };
+        request.data.push(callback);
+        this.nsp._onServerSideEmit(request.data);
+        break;
+
       default:
         debug("ignoring unknown request type: %s", request.type);
     }
@@ -381,6 +413,23 @@ export class RedisAdapter extends Adapter {
         this.requests.delete(requestId);
         break;
 
+      case RequestType.SERVER_SIDE_EMIT:
+        request.responses.push(response.data);
+
+        debug(
+          "serverSideEmit: got %d responses out of %d",
+          request.responses.length,
+          request.numSub
+        );
+        if (request.responses.length === request.numSub) {
+          clearTimeout(request.timeout);
+          if (request.resolve) {
+            request.resolve(null, request.responses);
+          }
+          this.requests.delete(requestId);
+        }
+        break;
+
       default:
         debug("ignoring unknown request type: %s", request.type);
     }
@@ -733,6 +782,67 @@ export class RedisAdapter extends Adapter {
     this.pubClient.publish(this.requestChannel, request);
   }
 
+  public serverSideEmit(packet: any[]): void {
+    const withAck = typeof packet[packet.length - 1] === "function";
+
+    if (withAck) {
+      this.serverSideEmitWithAck(packet).catch(() => {
+        // ignore errors
+      });
+      return;
+    }
+
+    const request = JSON.stringify({
+      uid: this.uid,
+      type: RequestType.SERVER_SIDE_EMIT,
+      data: packet,
+    });
+
+    this.pubClient.publish(this.requestChannel, request);
+  }
+
+  private async serverSideEmitWithAck(packet: any[]) {
+    const ack = packet.pop();
+    const numSub = (await this.getNumSub()) - 1; // ignore self
+
+    debug('waiting for %d responses to "serverSideEmit" request', numSub);
+
+    if (numSub <= 0) {
+      return ack(null, []);
+    }
+
+    const requestId = uid2(6);
+    const request = JSON.stringify({
+      uid: this.uid,
+      requestId, // the presence of this attribute defines whether an acknowledgement is needed
+      type: RequestType.SERVER_SIDE_EMIT,
+      data: packet,
+    });
+
+    const timeout = setTimeout(() => {
+      const storedRequest = this.requests.get(requestId);
+      if (storedRequest) {
+        ack(
+          new Error(
+            `timeout reached: only ${storedRequest.responses.length} responses received out of ${storedRequest.numSub}`
+          ),
+          storedRequest.responses
+        );
+        this.requests.delete(requestId);
+      }
+    }, this.requestsTimeout);
+
+    this.requests.set(requestId, {
+      type: RequestType.SERVER_SIDE_EMIT,
+      numSub,
+      timeout,
+      resolve: ack,
+      responses: [],
+    });
+
+    this.pubClient.publish(this.requestChannel, request);
+  }
+
   /**
    * Get the number of subscribers of the request channel
    *

package-lock.json

@@ -21,7 +21,7 @@
   "dependencies": {
     "debug": "~4.3.1",
     "notepack.io": "~2.2.0",
-    "socket.io-adapter": "^2.2.0",
+    "socket.io-adapter": "~2.3.0",
     "uid2": "0.0.3"
   },
   "devDependencies": {
@@ -35,8 +35,8 @@
     "nyc": "^15.1.0",
     "prettier": "^2.1.2",
     "redis": "^3.1.2",
-    "socket.io": "^4.0.0",
-    "socket.io-client": "^4.0.0",
+    "socket.io": "^4.1.1",
+    "socket.io-client": "^4.1.1",
     "ts-node": "^9.1.1",
     "typescript": "^4.0.5"
   },