[ws] Isolation tests

Author: jamesgpearce

test/unit/common.ts

@@ -1281,11 +1281,13 @@ export const describeSchemaTests = <DB, Depth extends number>(
 export const createClients = async (
   number: number,
   connect: (path: string) => Promise<{status: number; webSocket: any}>,
+  getPath?: (index: number) => string,
 ) => {
   const webSockets: any[] = [];
   const received: [string, string][][] = Array.from({length: number}, () => []);
   for (let i = 0; i < number; i++) {
-    const {webSocket} = await connect('');
+    const path = getPath ? getPath(i) : '';
+    const {webSocket} = await connect(path);
     if (!webSocket) {
       throw new Error('failed to obtain WebSocket');
     }

test/unit/ws/broker-basics.test.ts

@@ -10,7 +10,7 @@ import {
   pause,
 } from '../common.ts';
 
-let serverSynclet: Synclet<1>;
+let synclet: Synclet<1>;
 
 const connect = async (
   port: number,
@@ -30,31 +30,30 @@ const connect = async (
 
 describeCommonBrokerTests(
   async () => {
-    serverSynclet = await createSynclet({
+    synclet = await createSynclet({
       transport: createWsBrokerTransport({
         webSocketServer: new WebSocketServer({
           port: allocatePort(),
         }).setMaxListeners(0),
       }),
     });
-    await serverSynclet.start();
+    await synclet.start();
 
     return [
       (path: string) =>
         connect(
-          (
-            serverSynclet.getTransport()[0] as WsBrokerTransport
-          ).getWebSocketServer().options.port!,
+          (synclet.getTransport()[0] as WsBrokerTransport).getWebSocketServer()
+            .options.port!,
           path,
         ),
       async () =>
-        (serverSynclet.getTransport()[0] as WsBrokerTransport).getClientIds(),
+        (synclet.getTransport()[0] as WsBrokerTransport).getClientIds(),
     ] as const;
   },
 
   async () => {
-    serverSynclet.destroy();
-    (serverSynclet.getTransport()[0] as WsBrokerTransport)
+    synclet.destroy();
+    (synclet.getTransport()[0] as WsBrokerTransport)
       .getWebSocketServer()
       .close();
   },
@@ -79,24 +78,23 @@ describe('external httpServer', () => {
     });
   });
 
-  test('single webSocketServer serving both paths', async () => {
-    serverSynclet = await createSynclet({
+  test('single webSocketServer, two paths', async () => {
+    const synclet = await createSynclet({
       transport: createWsBrokerTransport({
         webSocketServer: new WebSocketServer({
           server: httpServer,
         }).setMaxListeners(0),
       }),
     });
-    await serverSynclet.start();
+    await synclet.start();
 
     const [[ws1, ws2], [received1, received2]] = await createClients(
       2,
       (path: string) => connect(port, path),
     );
 
     expect(
-      (serverSynclet.getTransport()[0] as WsBrokerTransport).getClientIds()
-        .length,
+      (synclet.getTransport()[0] as WsBrokerTransport).getClientIds().length,
     ).toEqual(2);
 
     ws1.send('* from1To*');
@@ -112,9 +110,134 @@ describe('external httpServer', () => {
 
     ws1.close();
     ws2.close();
-    await serverSynclet.destroy();
-    (serverSynclet.getTransport()[0] as WsBrokerTransport)
+    await synclet.destroy();
+    (synclet.getTransport()[0] as WsBrokerTransport)
       .getWebSocketServer()
       .close();
   });
+
+  test('single webSocketServer, manual upgrade, two paths', async () => {
+    const wss = new WebSocketServer({noServer: true}).setMaxListeners(0);
+    const synclet = await createSynclet({
+      transport: createWsBrokerTransport({
+        webSocketServer: wss,
+      }),
+    });
+    await synclet.start();
+
+    httpServer.on('upgrade', (request, socket, head) =>
+      wss.handleUpgrade(request, socket, head, (ws) =>
+        wss.emit('connection', ws, request),
+      ),
+    );
+
+    const [[ws1, ws2], [received1, received2]] = await createClients(
+      2,
+      (path: string) => connect(port, path),
+    );
+
+    expect(
+      (synclet.getTransport()[0] as WsBrokerTransport).getClientIds().length,
+    ).toEqual(2);
+
+    ws1.send('* from1To*');
+    await pause(transportPause);
+
+    ws2.send('* from2To*');
+    await pause(transportPause);
+
+    expect(received1.length).toEqual(1);
+    expect(received1[0][1]).toEqual('from2To*');
+    expect(received2.length).toEqual(1);
+    expect(received2[0][1]).toEqual('from1To*');
+
+    ws1.close();
+    ws2.close();
+    await synclet.destroy();
+    wss.close();
+  });
+
+  test('two webSocketServers, manual upgrade, two paths', async () => {
+    const wss1 = new WebSocketServer({noServer: true}).setMaxListeners(0);
+    const synclet1 = await createSynclet({
+      transport: createWsBrokerTransport({
+        webSocketServer: wss1,
+      }),
+    });
+    await synclet1.start();
+
+    const wss2 = new WebSocketServer({noServer: true}).setMaxListeners(0);
+    const synclet2 = await createSynclet({
+      transport: createWsBrokerTransport({
+        webSocketServer: wss2,
+      }),
+    });
+    await synclet2.start();
+
+    httpServer.on('upgrade', (request, socket, head) => {
+      const pathname = new URL(request.url!, 'ws://localhost').pathname;
+      if (pathname === '/p1') {
+        wss1.handleUpgrade(request, socket, head, (ws) =>
+          wss1.emit('connection', ws, request),
+        );
+      } else if (pathname === '/p2') {
+        wss2.handleUpgrade(request, socket, head, (ws) =>
+          wss2.emit('connection', ws, request),
+        );
+      }
+    });
+
+    const [[ws1p1, ws2p1], [received1p1, received2p1]] = await createClients(
+      2,
+      (path: string) => connect(port, path),
+      () => 'p1',
+    );
+
+    const [[ws1p2, ws2p2], [received1p2, received2p2]] = await createClients(
+      2,
+      (path: string) => connect(port, path),
+      () => 'p2',
+    );
+
+    expect(
+      (synclet1.getTransport()[0] as WsBrokerTransport).getClientIds().length,
+    ).toEqual(2);
+    expect(
+      (synclet2.getTransport()[0] as WsBrokerTransport).getClientIds().length,
+    ).toEqual(2);
+
+    ws1p1.send('* from1p1To*');
+    await pause(transportPause);
+    ws2p1.send('* from2p1To*');
+    await pause(transportPause);
+
+    expect(received1p1.length).toEqual(1);
+    expect(received1p1[0][1]).toEqual('from2p1To*');
+    expect(received2p1.length).toEqual(1);
+    expect(received2p1[0][1]).toEqual('from1p1To*');
+    expect(received1p2.length).toEqual(0);
+    expect(received1p2.length).toEqual(0);
+
+    ws1p2.send('* from1p2To*');
+    await pause(transportPause);
+    ws2p2.send('* from2p2To*');
+    await pause(transportPause);
+
+    expect(received1p1.length).toEqual(1);
+    expect(received1p1.length).toEqual(1);
+    expect(received1p2.length).toEqual(1);
+    expect(received1p2[0][1]).toEqual('from2p2To*');
+    expect(received2p2.length).toEqual(1);
+    expect(received2p2[0][1]).toEqual('from1p2To*');
+
+    ws1p1.close();
+    ws2p1.close();
+    ws1p2.close();
+    ws2p2.close();
+
+    await synclet1.destroy();
+    await synclet2.destroy();
+    wss1.close();
+    wss2.close();
+  });
 });