fix tests

packages/socket.io-clustered-engine/lib/engine.ts

@@ -1,5 +1,5 @@
 import { Server, type ServerOptions, Socket, type Transport } from "engine.io";
-import { randomUUID } from "node:crypto";
+import { randomBytes } from "node:crypto";
 import { setTimeout, clearTimeout } from "node:timers";
 import { type IncomingMessage } from "node:http";
 import { type Packet } from "engine.io-parser";
@@ -18,7 +18,11 @@ type Brand<K, T> = K & { __brand: T };
 
 type NodeId = Brand<string, "NodeId">;
 type SessionId = Brand<string, "SessionId">;
-type RequestId = Brand<string, "RequestId">;
+type RequestId = Brand<number, "RequestId">;
+
+function randomId() {
+  return randomBytes(3).toString("hex");
+}
 
 enum MessageType {
   ACQUIRE_LOCK = 0,
@@ -149,9 +153,10 @@ interface ClusterEngineOptions {
 // @ts-expect-error onWebSocket() method is private in parent class
 export abstract class ClusterEngine extends Server {
   private readonly _opts: Required<ClusterEngineOptions>;
-  protected readonly _nodeId = randomUUID() as NodeId;
+  protected readonly _nodeId = randomId() as NodeId;
   private readonly _requests = new Map<RequestId, ClusterRequest>();
   private readonly _remoteTransports = new Map<SessionId, Transport>();
+  private _requestCount = 0;
 
   constructor(opts?: ServerOptions & ClusterEngineOptions) {
     super(opts);
@@ -426,7 +431,7 @@ export abstract class ClusterEngine extends Server {
     onSuccess: (senderId: NodeId) => void,
     onError: () => void
   ) {
-    const requestId = randomUUID() as RequestId;
+    const requestId = ++this._requestCount as RequestId;
 
     const timer = setTimeout(() => {
       this._requests.delete(requestId);
@@ -571,7 +576,7 @@ export abstract class ClusterEngine extends Server {
     debug("upgrade success");
     this._hookTransport(sid, transport, "read", senderId);
 
-    const requestId = randomUUID() as RequestId;
+    const requestId = ++this._requestCount as RequestId;
 
     const onSuccess = (takeOver: boolean, packets: Packet[]) => {
       if (takeOver) {

packages/socket.io-clustered-engine/lib/redis.ts

@@ -1,5 +1,5 @@
 import { ClusterEngine, type Message } from "./engine";
-import { encode, decode } from "notepack.io";
+import { encode, decode } from "@msgpack/msgpack";
 import { type ServerOptions } from "engine.io";
 import cluster from "node:cluster";
 import { randomUUID } from "node:crypto";
@@ -41,7 +41,7 @@ export function setupPrimaryWithRedis(
   SUBSCRIBE(subClient, channels, (buffer: Buffer) => {
     let message: Message & { _source?: string; _primaryId?: string };
     try {
-      message = decode(buffer);
+      message = decode(buffer) as Message;
     } catch (e) {
       debug("ignore malformed buffer");
       return;
@@ -147,7 +147,7 @@ export class RedisEngine extends ClusterEngine {
     SUBSCRIBE(subClient, channels, (buffer: Buffer) => {
       let message: Message & { _source?: string; _primaryId?: string };
       try {
-        message = decode(buffer);
+        message = decode(buffer) as Message;
       } catch (e) {
         debug("ignore malformed buffer");
         return;
@@ -170,7 +170,7 @@ export class RedisEngine extends ClusterEngine {
     message._source = MESSAGE_SOURCE;
 
     debug("publish message to channel %s", channel);
-    this._pubClient.publish(channel, encode(message));
+    this._pubClient.publish(channel, Buffer.from(encode(message)));
   }
 }
 

packages/socket.io-clustered-engine/package.json

@@ -6,11 +6,10 @@
   "license": "MIT",
   "description": "",
   "dependencies": {
-    "@msgpack/msgpack": "^3.0.0-beta2",
+    "@msgpack/msgpack": "~2.8.0",
     "debug": "~4.3.3",
     "engine.io": "~6.6.0",
-    "engine.io-parser": "~5.2.3",
-    "notepack.io": "^3.0.1"
+    "engine.io-parser": "~5.2.3"
   },
   "scripts": {
     "compile": "rimraf ./dist && tsc",

packages/socket.io-clustered-engine/test/cluster.ts

@@ -23,7 +23,16 @@ describe("cluster", () => {
   });
 
   afterEach((done) => {
-    cluster.disconnect(done);
+    for (const worker of Object.values(cluster.workers)) {
+      worker.kill();
+    }
+    function onExit() {
+      if (Object.keys(cluster.workers).length === 0) {
+        cluster.off("exit", onExit);
+        done();
+      }
+    }
+    cluster.on("exit", onExit);
   });
 
   it("should ping/pong", (done) => {

packages/socket.io-clustered-engine/test/in-memory.ts

@@ -51,8 +51,11 @@ describe("in-memory", () => {
     engine2.close();
     engine3.close();
     httpServer1.close();
+    httpServer1.closeAllConnections();
     httpServer2.close();
+    httpServer2.closeAllConnections();
     httpServer3.close();
+    httpServer3.closeAllConnections();
   });
 
   it("should work (read)", (done) => {

packages/socket.io-clustered-engine/test/redis.ts

@@ -48,8 +48,11 @@ describe("redis", () => {
         engine2.close();
         engine3.close();
         httpServer1.close();
+        httpServer1.closeAllConnections();
         httpServer2.close();
+        httpServer2.closeAllConnections();
         httpServer3.close();
+        httpServer3.closeAllConnections();
 
         return Promise.all([
           pubClient.disconnect(),
@@ -143,20 +146,21 @@ describe("redis", () => {
       engine3.attach(httpServer3);
       httpServer3.listen(3002);
 
-      cleanup = () => {
+      cleanup = async () => {
         engine1.close();
         engine2.close();
         engine3.close();
         httpServer1.close();
+        httpServer1.closeAllConnections();
         httpServer2.close();
+        httpServer2.closeAllConnections();
         httpServer3.close();
+        httpServer3.closeAllConnections();
 
-        return Promise.all([
-          pubClient.disconnect(),
-          subClient1.disconnect(),
-          subClient2.disconnect(),
-          subClient3.disconnect(),
-        ]).then();
+        pubClient.disconnect();
+        subClient1.disconnect();
+        subClient2.disconnect();
+        subClient3.disconnect();
       };
     });