bis

Author: darrachequesne

test/index.ts

@@ -262,7 +262,7 @@ describe("@socket.io/postgres-adapter", () => {
       servers[0].disconnectSockets();
     });
 
-    it.skip("sends a packet before all socket instances disconnect", (done) => {
+    it("sends a packet before all socket instances disconnect", (done) => {
       const partialDone = times(3, done);
 
       clientSockets.forEach((clientSocket) => {

test/util.ts

@@ -25,72 +25,81 @@ export function shouldNotHappen(done: (err?: Error) => void) {
   return () => done(new Error("should not happen"));
 }
 
-interface TestContext {
-  servers: Server[];
-  serverSockets: ServerSocket[];
-  clientSockets: ClientSocket[];
-  cleanup: () => void;
-}
+function createServerAndClient(pool: Pool) {
+  const httpServer = createServer();
+  const io = new Server(httpServer, {
+    adapter: createAdapter(pool, {
+      tableName: "events",
+    }),
+  });
 
-const NODES_COUNT = 3;
+  return new Promise<{
+    io: Server;
+    socket: ServerSocket;
+    clientSocket: ClientSocket;
+  }>((resolve) => {
+    httpServer.listen(() => {
+      const port = (httpServer.address() as AddressInfo).port;
+      const clientSocket = ioc(`http://localhost:${port}`);
+
+      io.on("connection", (socket) => {
+        resolve({
+          io,
+          socket,
+          clientSocket,
+        });
+      });
+    });
+  });
+}
 
-export function setup() {
-  const servers: Server[] = [];
-  const serverSockets: ServerSocket[] = [];
-  const clientSockets: ClientSocket[] = [];
+function isInitComplete(servers: Server[]) {
+  return servers.every((server) => {
+    return server.of("/").adapter.nodesMap.size === servers.length - 1;
+  });
+}
 
+export async function setup() {
   const pool = new Pool({
     user: "postgres",
     password: "changeit",
   });
 
-  return new Promise<TestContext>(async (resolve) => {
-    await pool.query(
-      `
+  await pool.query(
+    `
       CREATE TABLE IF NOT EXISTS events (
           id          bigserial UNIQUE,
           created_at  timestamptz DEFAULT NOW(),
           payload     bytea
       );
     `,
-      () => {}
-    );
-
-    for (let i = 1; i <= NODES_COUNT; i++) {
-      const httpServer = createServer();
-      const io = new Server(httpServer, {
-        adapter: createAdapter(pool, {
-          tableName: "events",
-        }),
-      });
+    () => {}
+  );
 
-      io.of("/").adapter.init();
+  const results = await Promise.all([
+    createServerAndClient(pool),
+    createServerAndClient(pool),
+    createServerAndClient(pool),
+  ]);
 
-      httpServer.listen(() => {
-        const port = (httpServer.address() as AddressInfo).port;
-        const clientSocket = ioc(`http://localhost:${port}`);
+  const servers = results.map(({ io }) => io);
+  const serverSockets = results.map(({ socket }) => socket);
+  const clientSockets = results.map(({ clientSocket }) => clientSocket);
 
-        io.on("connection", async (socket) => {
-          clientSockets.push(clientSocket);
-          serverSockets.push(socket);
-          servers.push(io);
+  servers.forEach((server) => server.of("/").adapter.init());
 
-          if (servers.length === NODES_COUNT) {
-            await sleep(100);
+  while (!isInitComplete(servers)) {
+    await sleep(20);
+  }
 
-            resolve({
-              servers,
-              serverSockets,
-              clientSockets,
-              cleanup: () => {
-                servers.forEach((server) => server.close());
-                clientSockets.forEach((socket) => socket.disconnect());
-                pool.end();
-              },
-            });
-          }
-        });
-      });
-    }
-  });
+  return {
+    servers,
+    serverSockets,
+    clientSockets,
+    cleanup: () => {
+      servers.forEach((server) => server.close());
+      clientSockets.forEach((socket) => socket.disconnect());
+      pool.end();
+    },
+  };
 }