test: refactor test setup

Author: darrachequesne

test/index.ts

@@ -1,81 +1,23 @@
-import { createServer } from "http";
-import { Server, Socket as ServerSocket } from "socket.io";
-import { io as ioc, Socket as ClientSocket } from "socket.io-client";
-import expect = require("expect.js");
-import { createAdapter } from "../lib";
-import type { AddressInfo } from "net";
-import { times, sleep, shouldNotHappen } from "./util";
-import { Pool } from "pg";
-
-const NODES_COUNT = 3;
+import { type Server, type Socket as ServerSocket } from "socket.io";
+import { type Socket as ClientSocket } from "socket.io-client";
+import expect from "expect.js";
+import { setup, times, sleep, shouldNotHappen } from "./util";
 
 describe("@socket.io/postgres-adapter", () => {
   let servers: Server[],
     serverSockets: ServerSocket[],
     clientSockets: ClientSocket[],
-    pool: Pool;
-
-  beforeEach((done) => {
-    servers = [];
-    serverSockets = [];
-    clientSockets = [];
-    pool = new Pool({
-      user: "postgres",
-      host: "localhost",
-      database: "postgres",
-      password: "changeit",
-      port: 5432,
-    });
-
-    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);
-      io.adapter(
-        createAdapter(pool, {
-          tableName: "events",
-        })
-      );
-      httpServer.listen(() => {
-        const port = (httpServer.address() as AddressInfo).port;
-        const clientSocket = ioc(`http://localhost:${port}`);
-
-        io.on("connection", async (socket) => {
-          clientSockets.push(clientSocket);
-          serverSockets.push(socket);
-          servers.push(io);
-          if (servers.length === NODES_COUNT) {
-            await sleep(200);
-
-            // ensure all nodes know each other
-            servers[0].emit("ping");
-            servers[1].emit("ping");
-            servers[2].emit("ping");
-
-            await sleep(200);
-
-            done();
-          }
-        });
-      });
-    }
+    cleanup: () => void;
+
+  beforeEach(async () => {
+    const testContext = await setup();
+    servers = testContext.servers;
+    serverSockets = testContext.serverSockets;
+    clientSockets = testContext.clientSockets;
+    cleanup = testContext.cleanup;
   });
 
-  afterEach((done) => {
-    servers.forEach((server) => server.close());
-    clientSockets.forEach((socket) => socket.disconnect());
-    pool.end(done);
-  });
+  afterEach(() => cleanup());
 
   describe("broadcast", function () {
     it("broadcasts to all clients", (done) => {
@@ -320,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

@@ -1,3 +1,10 @@
+import { createServer } from "http";
+import { AddressInfo } from "net";
+import { Server, type Socket as ServerSocket } from "socket.io";
+import { io as ioc, type Socket as ClientSocket } from "socket.io-client";
+import { Pool } from "pg";
+import { createAdapter } from "../lib";
+
 export function times(count: number, done: (err?: Error) => void) {
   let i = 0;
   return () => {
@@ -17,3 +24,82 @@ export function sleep(duration: number) {
 export function shouldNotHappen(done: (err?: Error) => void) {
   return () => done(new Error("should not happen"));
 }
+
+function createServerAndClient(pool: Pool) {
+  const httpServer = createServer();
+  const io = new Server(httpServer, {
+    adapter: createAdapter(pool, {
+      tableName: "events",
+    }),
+  });
+
+  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,
+        });
+      });
+    });
+  });
+}
+
+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",
+  });
+
+  await pool.query(
+    `
+      CREATE TABLE IF NOT EXISTS events (
+          id          bigserial UNIQUE,
+          created_at  timestamptz DEFAULT NOW(),
+          payload     bytea
+      );
+    `,
+    () => {}
+  );
+
+  const results = await Promise.all([
+    createServerAndClient(pool),
+    createServerAndClient(pool),
+    createServerAndClient(pool),
+  ]);
+
+  const servers = results.map(({ io }) => io);
+  const serverSockets = results.map(({ socket }) => socket);
+  const clientSockets = results.map(({ clientSocket }) => clientSocket);
+
+  servers.forEach((server) => server.of("/").adapter.init());
+
+  while (!isInitComplete(servers)) {
+    await sleep(20);
+  }
+
+  return {
+    servers,
+    serverSockets,
+    clientSockets,
+    cleanup: () => {
+      servers.forEach((server) => server.close());
+      clientSockets.forEach((socket) => socket.disconnect());
+      pool.end();
+    },
+  };
+}