move things around, allow multiple server transports

__tests__/bandwidth.bench.ts

@@ -1,9 +1,9 @@
 import { afterAll, assert, bench, describe } from 'vitest';
 import { getClientSendFn, waitForMessage } from '../testUtil';
-import { TestServiceSchema } from './fixtures/services';
+import { TestServiceSchema } from '../testUtil/fixtures/services';
 import { createServer } from '../router/server';
 import { createClient } from '../router/client';
-import { transports } from './fixtures/transports';
+import { transports } from '../testUtil/fixtures/transports';
 import { nanoid } from 'nanoid';
 
 let n = 0;

__tests__/cancellation.test.ts

@@ -9,15 +9,15 @@ import {
   createClient,
   createServer,
 } from '../router';
-import { testMatrix } from './fixtures/matrix';
+import { testMatrix } from '../testUtil/fixtures/matrix';
 import {
   cleanupTransports,
   createPostTestCleanups,
   testFinishesCleanly,
   waitFor,
-} from './fixtures/cleanup';
+} from '../testUtil/fixtures/cleanup';
 import { CANCEL_CODE, UNCAUGHT_ERROR_CODE } from '../router/errors';
-import { TestSetupHelpers } from './fixtures/transports';
+import { TestSetupHelpers } from '../testUtil/fixtures/transports';
 
 function makeMockHandler<T extends ValidProcType>(
   _type: T,

__tests__/cleanup.test.ts

@@ -10,7 +10,7 @@ import {
   SubscribableServiceSchema,
   TestServiceSchema,
   UploadableServiceSchema,
-} from './fixtures/services';
+} from '../testUtil/fixtures/services';
 import {
   Ok,
   Procedure,
@@ -26,9 +26,9 @@ import {
   createPostTestCleanups,
   testFinishesCleanly,
   waitFor,
-} from './fixtures/cleanup';
-import { testMatrix } from './fixtures/matrix';
-import { TestSetupHelpers } from './fixtures/transports';
+} from '../testUtil/fixtures/cleanup';
+import { testMatrix } from '../testUtil/fixtures/matrix';
+import { TestSetupHelpers } from '../testUtil/fixtures/transports';
 import { ControlFlags } from '../transport/message';
 import { Type } from '@sinclair/typebox';
 import { nanoid } from 'nanoid';

__tests__/context.test.ts

@@ -1,7 +1,10 @@
 import { beforeEach, describe, expect, test } from 'vitest';
-import { cleanupTransports, createPostTestCleanups } from './fixtures/cleanup';
-import { testMatrix } from './fixtures/matrix';
-import { TestSetupHelpers } from './fixtures/transports';
+import {
+  cleanupTransports,
+  createPostTestCleanups,
+} from '../testUtil/fixtures/cleanup';
+import { testMatrix } from '../testUtil/fixtures/matrix';
+import { TestSetupHelpers } from '../testUtil/fixtures/transports';
 import {
   Ok,
   Procedure,

__tests__/disconnects.test.ts

@@ -9,7 +9,7 @@ import {
   SubscribableServiceSchema,
   TestServiceSchema,
   UploadableServiceSchema,
-} from './fixtures/services';
+} from '../testUtil/fixtures/services';
 import {
   createClient,
   createServer,
@@ -20,10 +20,10 @@ import {
   cleanupTransports,
   testFinishesCleanly,
   waitFor,
-} from './fixtures/cleanup';
-import { testMatrix } from './fixtures/matrix';
-import { TestSetupHelpers } from './fixtures/transports';
-import { createPostTestCleanups } from './fixtures/cleanup';
+} from '../testUtil/fixtures/cleanup';
+import { testMatrix } from '../testUtil/fixtures/matrix';
+import { TestSetupHelpers } from '../testUtil/fixtures/transports';
+import { createPostTestCleanups } from '../testUtil/fixtures/cleanup';
 
 describe.each(testMatrix())(
   'procedures should handle unexpected disconnects ($transport.name transport, $codec.name codec)',

__tests__/e2e.test.ts

@@ -17,15 +17,15 @@ import {
   UploadableServiceSchema,
   OrderingServiceSchema,
   NonObjectSchemas,
-} from './fixtures/services';
+} from '../testUtil/fixtures/services';
 import {
   advanceFakeTimersBySessionGrace,
   cleanupTransports,
   createPostTestCleanups,
   testFinishesCleanly,
   waitFor,
-} from './fixtures/cleanup';
-import { testMatrix } from './fixtures/matrix';
+} from '../testUtil/fixtures/cleanup';
+import { testMatrix } from '../testUtil/fixtures/matrix';
 import { Type } from '@sinclair/typebox';
 import {
   Procedure,
@@ -38,7 +38,7 @@ import {
   createClientHandshakeOptions,
   createServerHandshakeOptions,
 } from '../router/handshake';
-import { TestSetupHelpers } from './fixtures/transports';
+import { TestSetupHelpers } from '../testUtil/fixtures/transports';
 
 describe.each(testMatrix())(
   'client <-> server integration test ($transport.name transport, $codec.name codec)',
@@ -869,6 +869,7 @@ describe.each(testMatrix())(
         createClientHandshakeOptions(requestSchema, () => ({ data: 'foobar' })),
       );
       const serverTransport = getServerTransport(
+        'SERVER',
         createServerHandshakeOptions(requestSchema, (metadata) => {
           return {
             data: metadata.data,

__tests__/handler.test.ts

@@ -7,9 +7,9 @@ import {
   TestServiceSchema,
   SubscribableServiceSchema,
   UploadableServiceSchema,
-} from './fixtures/services';
+} from '../testUtil/fixtures/services';
 import { createClient, createServer, UNCAUGHT_ERROR_CODE } from '../router';
-import { createMockTransportNetwork } from './fixtures/mockTransport';
+import { createMockTransportNetwork } from '../testUtil/fixtures/mockTransport';
 
 describe('server-side test', () => {
   let mockTransportNetwork: ReturnType<typeof createMockTransportNetwork>;

__tests__/invalid-request.test.ts

@@ -9,16 +9,16 @@ import {
   createClient,
   createServer,
 } from '../router';
-import { testMatrix } from './fixtures/matrix';
+import { testMatrix } from '../testUtil/fixtures/matrix';
 import {
   cleanupTransports,
   createPostTestCleanups,
   waitFor,
-} from './fixtures/cleanup';
+} from '../testUtil/fixtures/cleanup';
 import { EventMap } from '../transport';
 import { INVALID_REQUEST_CODE } from '../router/errors';
 import { ControlFlags } from '../transport/message';
-import { TestSetupHelpers } from './fixtures/transports';
+import { TestSetupHelpers } from '../testUtil/fixtures/transports';
 import { nanoid } from 'nanoid';
 import { getClientSendFn } from '../testUtil';
 

__tests__/negative.test.ts

@@ -4,7 +4,7 @@ import {
   cleanupTransports,
   testFinishesCleanly,
   waitFor,
-} from './fixtures/cleanup';
+} from '../testUtil/fixtures/cleanup';
 import {
   createLocalWebSocketClient,
   createWebSocketServer,
@@ -23,7 +23,7 @@ import { Static } from '@sinclair/typebox';
 import { WebSocketClientTransport } from '../transport/impls/ws/client';
 import { ProtocolError } from '../transport/events';
 import NodeWs from 'ws';
-import { createPostTestCleanups } from './fixtures/cleanup';
+import { createPostTestCleanups } from '../testUtil/fixtures/cleanup';
 import { generateId } from '../transport/id';
 
 describe('should handle incompatabilities', async () => {

__tests__/serialize.test.ts

@@ -3,7 +3,7 @@ import {
   BinaryFileServiceSchema,
   FallibleServiceSchema,
   TestServiceSchema,
-} from './fixtures/services';
+} from '../testUtil/fixtures/services';
 import { serializeSchema } from '../router';
 import { Type } from '@sinclair/typebox';
 

__tests__/typescript-stress.test.ts

@@ -12,15 +12,15 @@ import {
   ResultUnwrapOk,
   unwrapOrThrow,
 } from '../router/result';
-import { TestServiceSchema } from './fixtures/services';
+import { TestServiceSchema } from '../testUtil/fixtures/services';
 import { readNextResult } from '../testUtil';
 import {
   createClientHandshakeOptions,
   createServerHandshakeOptions,
 } from '../router/handshake';
 import { flattenErrorType, ProcedureErrorSchemaType } from '../router/errors';
 import { ReadableImpl } from '../router/streams';
-import { createMockTransportNetwork } from './fixtures/mockTransport';
+import { createMockTransportNetwork } from '../testUtil/fixtures/mockTransport';
 
 const requestData = Type.Union([
   Type.Object({ a: Type.Number() }),

codec/codec.test.ts

@@ -1,5 +1,5 @@
 import { describe, test, expect } from 'vitest';
-import { codecs } from '../__tests__/fixtures/codec';
+import { codecs } from '../testUtil/fixtures/codec';
 
 describe.each(codecs)('codec -- $name', ({ codec }) => {
   test('empty object', () => {

__tests__/duplexPair.ts → testUtil/duplex/duplexPair.ts

@@ -1,5 +1,5 @@
 import { Duplex } from 'node:stream';
-import { assert } from 'vitest';
+import assert from 'assert';
 
 const kCallback = Symbol('Callback');
 const kInitOtherSide = Symbol('InitOtherSide');

__tests__/fixtures/cleanup.ts → testUtil/fixtures/cleanup.ts

@@ -8,7 +8,7 @@ import {
 } from '../../transport';
 import { Server } from '../../router';
 import { AnyServiceSchemaMap } from '../../router/services';
-import { numberOfConnections, testingSessionOptions } from '../../testUtil';
+import { numberOfConnections, testingSessionOptions } from '..';
 import { Value } from '@sinclair/typebox/value';
 import { ControlMessageAckSchema } from '../../transport/message';
 

__tests__/fixtures/mockTransport.ts → testUtil/fixtures/mockTransport.ts

@@ -2,15 +2,12 @@ import { TransportClientId } from '../../transport';
 import { ClientTransport } from '../../transport/client';
 import { Connection } from '../../transport/connection';
 import { ServerTransport } from '../../transport/server';
-import {
-  ClientHandshakeOptions,
-  ServerHandshakeOptions,
-} from '../../router/handshake';
-import { Observable } from '../observable';
+import { Observable } from '../observable/observable';
 import { ProvidedServerTransportOptions } from '../../transport/options';
-import { TestTransportOptions } from './transports';
+import { TestSetupHelpers, TestTransportOptions } from './transports';
 import { Duplex } from 'node:stream';
-import { duplexPair } from '../duplexPair';
+import { duplexPair } from '../duplex/duplexPair';
+import { nanoid } from 'nanoid';
 
 export class InMemoryConnection extends Connection {
   conn: Duplex;
@@ -54,8 +51,11 @@ export class InMemoryConnection extends Connection {
 }
 
 interface BidiConnection {
+  id: string;
   clientToServer: Duplex;
   serverToClient: Duplex;
+  clientId: TransportClientId;
+  serverId: TransportClientId;
   handled: boolean;
 }
 
@@ -67,34 +67,29 @@ interface BidiConnection {
 // difficult with real network connections (e.g. simulating a phantom
 // disconnect, .pause() vs .removeAllListeners('data'), congestion,
 // latency, differences in ws implementations between node and browsers, etc.)
-export function createMockTransportNetwork(opts?: TestTransportOptions): {
-  getClientTransport: (
-    id: TransportClientId,
-    handshakeOptions?: ClientHandshakeOptions,
-  ) => ClientTransport<InMemoryConnection>;
-  getServerTransport: (
-    handshakeOptions?: ServerHandshakeOptions,
-  ) => ServerTransport<InMemoryConnection>;
-  simulatePhantomDisconnect: () => void;
-  restartServer: () => Promise<void>;
-  cleanup: () => Promise<void> | void;
-} {
-  // client id -> [client->server, server->client]
-  const connections = new Observable<Record<TransportClientId, BidiConnection>>(
-    {},
-  );
+export function createMockTransportNetwork(
+  opts?: TestTransportOptions,
+): TestSetupHelpers {
+  // conn id -> [client->server, server->client]
+  const connections = new Observable<Record<string, BidiConnection>>({});
 
   const transports: Array<MockClientTransport | MockServerTransport> = [];
   class MockClientTransport extends ClientTransport<InMemoryConnection> {
-    async createNewOutgoingConnection(): Promise<InMemoryConnection> {
+    async createNewOutgoingConnection(
+      to: TransportClientId,
+    ): Promise<InMemoryConnection> {
       const [clientToServer, serverToClient] = duplexPair();
       await new Promise((resolve) => setImmediate(resolve));
 
+      const connId = nanoid();
       connections.set((prev) => ({
         ...prev,
-        [this.clientId]: {
+        [connId]: {
+          id: connId,
           clientToServer,
           serverToClient,
+          clientId: this.clientId,
+          serverId: to,
           handled: false,
         },
       }));
@@ -115,14 +110,14 @@ export function createMockTransportNetwork(opts?: TestTransportOptions): {
       this.subscribeCleanup = connections.observe((conns) => {
         // look for any unhandled connections
         for (const conn of Object.values(conns)) {
-          if (conn.handled) {
+          // if we've already handled this connection, skip it
+          // or if it's not for us, skip it
+          if (conn.handled || conn.serverId !== this.clientId) {
             continue;
           }
 
-          // if we find one, handle it
           conn.handled = true;
           const connection = new InMemoryConnection(conn.serverToClient);
-
           this.handleConnection(connection);
         }
       });
@@ -145,8 +140,8 @@ export function createMockTransportNetwork(opts?: TestTransportOptions): {
 
       return clientTransport;
     },
-    getServerTransport: (handshakeOptions) => {
-      const serverTransport = new MockServerTransport('SERVER', opts?.server);
+    getServerTransport: (id = 'SERVER', handshakeOptions) => {
+      const serverTransport = new MockServerTransport(id, opts?.server);
       if (handshakeOptions) {
         serverTransport.extendHandshake(handshakeOptions);
       }

__tests__/fixtures/services.ts → testUtil/fixtures/services.ts

@@ -1,7 +1,7 @@
 import { Type } from '@sinclair/typebox';
 import { ServiceSchema } from '../../router/services';
 import { Err, Ok, unwrapOrThrow } from '../../router/result';
-import { Observable } from '../observable';
+import { Observable } from '../observable/observable';
 import { Procedure } from '../../router';
 
 export const EchoRequest = Type.Object({

__tests__/fixtures/transports.ts → testUtil/fixtures/transports.ts

@@ -10,7 +10,7 @@ import {
   createWebSocketServer,
   getTransportConnections,
   onWsServerReady,
-} from '../../testUtil';
+} from '..';
 import {
   ClientTransportOptions,
   ServerTransportOptions,
@@ -36,6 +36,7 @@ export interface TestSetupHelpers {
     handshakeOptions?: ClientHandshakeOptions,
   ) => ClientTransport<Connection>;
   getServerTransport: (
+    id?: TransportClientId,
     handshakeOptions?: ServerHandshakeOptions,
   ) => ServerTransport<Connection>;
   simulatePhantomDisconnect: () => void;
@@ -92,10 +93,10 @@ export const transports: Array<TransportMatrixEntry> = [
 
           return clientTransport;
         },
-        getServerTransport(handshakeOptions) {
+        getServerTransport(id = 'SERVER', handshakeOptions) {
           const serverTransport = new WebSocketServerTransport(
             wss,
-            'SERVER',
+            id,
             opts?.server,
           );