Formatting

source/services/broadcaster.ts

@@ -6,6 +6,7 @@ import { ApplicationError } from '../errors/index.ts';
 
 /** Request-scoped view from which the broadcaster obtains a stable connection. */
 export interface BroadcastStream {
+
     /** Connection identity shared by every request on the same transport session. */
     readonly connection: BaseStream;
 
@@ -15,6 +16,7 @@ export interface BroadcastStream {
 
 /** One pending subscribe call and the topics whose removal will resolve it. */
 interface SubscriptionWaiter {
+
     /** Only topics newly introduced by this particular subscribe call. */
     readonly remainingTopics: Set<string>;
 

test/services/broadcaster.test.ts

@@ -1,199 +1,180 @@
-import { describe, expect, it, vi } from "vitest";
+import { describe, expect, it, vi } from 'vitest';
 
-import { ApplicationError } from "../../source/errors/index.js";
+import { Broadcaster } from '../../source/services/broadcaster.ts';
-import { Broadcaster } from "../../source/services/broadcaster.js";
+import { ApplicationRouteStream } from '../../source/services/route-stream.ts';
-import { ApplicationRouteStream } from "../../source/services/route-stream.js";
+import { Logger } from '../../source/utils/logger.ts';
-import { Logger } from "../../source/utils/logger.js";
+import { TestConnection } from '../helpers/test-connection.ts';
-import { TestConnection } from "../helpers/test-connection.js";
 
-function createBroadcaster(): Broadcaster {
+const createBroadcaster = (): Broadcaster => {
-  return new Broadcaster(new Logger("broadcaster-test"));
+    return new Broadcaster(new Logger('broadcaster-test'));
-}
+};
 
-function routeStream(connection: TestConnection): ApplicationRouteStream {
+const routeStream = (connection: TestConnection): ApplicationRouteStream => {
-  return new ApplicationRouteStream(connection, undefined);
+    return new ApplicationRouteStream(connection, undefined);
-}
+};
 
-async function expectPending(promise: Promise<void>): Promise<void> {
+const expectPending = async (promise: Promise<void>): Promise<void> => {
-  const settled = vi.fn();
+    const settled = vi.fn();
-  void promise.then(settled);
+    void promise.then(settled);
-  await Promise.resolve();
+    await Promise.resolve();
-  expect(settled).not.toHaveBeenCalled();
+    expect(settled).not.toHaveBeenCalled();
-}
+};
 
-describe("Broadcaster subscriptions", () => {
+describe('Broadcaster subscriptions', () => {
-  it("delivers events and resolves after a later request removes the topic", async () => {
+    it('delivers events and resolves after a later request removes the topic', async (): Promise<void> => {
-    const broadcaster = createBroadcaster();
+        const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, true);
+        const connection = new TestConnection(true, true);
-    const subscribed = broadcaster.subscribe(routeStream(connection), [
+        const subscribed = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
-      "items",
-    ]);
 
-    await expectPending(subscribed);
+        await expectPending(subscribed);
-    await broadcaster.publish("items", {
+        await broadcaster.publish('items', {
-      type: "item-changed",
+            type: 'item-changed',
-      data: { id: "a" },
+            data: { id: 'a' },
+        });
+
+        expect(connection.messages).toEqual([
+            expect.objectContaining({
+                type: 'item-changed',
+                data: { id: 'a' },
+            }),
+        ]);
+
+        // A different request-scoped facade still resolves the connection's
+        // original subscription.
+        await broadcaster.unsubscribe(routeStream(connection), [ 'items' ]);
+        await expect(subscribed).resolves.toBeUndefined();
     });
 
-    expect(connection.messages).toEqual([
+    it('resolves fully duplicate subscriptions immediately', async (): Promise<void> => {
-      expect.objectContaining({
+        const broadcaster = createBroadcaster();
-        type: "item-changed",
+        const connection = new TestConnection(true, true);
-        data: { id: "a" },
+        const first = broadcaster.subscribe(routeStream(connection), [ 'items', 'items' ]);
-      }),
+        const duplicate = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
-    ]);
 
-    // A different request-scoped facade still resolves the connection's
+        await expect(duplicate).resolves.toBeUndefined();
-    // original subscription.
+        await expectPending(first);
-    await broadcaster.unsubscribe(routeStream(connection), ["items"]);
+        expect(connection.closeCallbacks).toHaveLength(1);
-    await expect(subscribed).resolves.toBeUndefined();
-  });
 
-  it("resolves fully duplicate subscriptions immediately", async () => {
+        await broadcaster.unsubscribe(routeStream(connection), [ 'items' ]);
-    const broadcaster = createBroadcaster();
+        await first;
-    const connection = new TestConnection(true, true);
-    const first = broadcaster.subscribe(routeStream(connection), [
-      "items",
-      "items",
-    ]);
-    const duplicate = broadcaster.subscribe(routeStream(connection), ["items"]);
-
-    await expect(duplicate).resolves.toBeUndefined();
-    await expectPending(first);
-    expect(connection.closeCallbacks).toHaveLength(1);
-
-    await broadcaster.unsubscribe(routeStream(connection), ["items"]);
-    await first;
-  });
-
-  it("waits only for topics newly added by a partially overlapping call", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, true);
-    const first = broadcaster.subscribe(routeStream(connection), ["a"]);
-    const second = broadcaster.subscribe(routeStream(connection), ["a", "b"]);
-
-    await broadcaster.unsubscribe(routeStream(connection), ["a"]);
-    await expect(first).resolves.toBeUndefined();
-    await expectPending(second);
-
-    await broadcaster.unsubscribe(routeStream(connection), ["b"]);
-    await expect(second).resolves.toBeUndefined();
-  });
-
-  it("resolves every pending subscription and removes topics on close", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, false);
-    const first = broadcaster.subscribe(routeStream(connection), ["a"]);
-    const second = broadcaster.subscribe(routeStream(connection), ["b"]);
-
-    connection.close();
-    await Promise.all([first, second]);
-    await broadcaster.publish("a", { type: "changed", data: null });
-    await broadcaster.publish("b", { type: "changed", data: null });
-
-    expect(connection.messages).toEqual([]);
-  });
-
-  it("immediately resolves registration against an already-closed connection", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, false);
-    connection.close();
-
-    await expect(
-      broadcaster.subscribe(routeStream(connection), ["items"]),
-    ).resolves.toBeUndefined();
-  });
-
-  it("rejects subscriptions on a non-streaming connection", () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(false, false);
-
-    expect(() =>
-      broadcaster.subscribe(routeStream(connection), ["items"]),
-    ).toThrowError(
-      expect.objectContaining({ statusCode: 406 }),
-    );
-  });
-
-  it("treats an empty subscription and repeated unsubscribe as no-ops", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, true);
-
-    await expect(
-      broadcaster.subscribe(routeStream(connection), []),
-    ).resolves.toBeUndefined();
-    await broadcaster.unsubscribe(routeStream(connection), ["missing"]);
-    await broadcaster.unsubscribe(routeStream(connection));
-
-    expect(connection.closeCallbacks).toHaveLength(0);
-  });
-
-  it("fans out concurrently to independent connections", async () => {
-    const broadcaster = createBroadcaster();
-    const first = new TestConnection(true, false);
-    const second = new TestConnection(true, false);
-    const originalFirstSend = first.send.bind(first);
-    let releaseFirst: () => void = () => undefined;
-    const firstReleased = new Promise<void>((resolve) => {
-      releaseFirst = resolve;
-    });
-    let markSecondSent: () => void = () => undefined;
-    const secondSent = new Promise<void>((resolve) => {
-      markSecondSent = resolve;
     });
 
-    first.send = async (message) => {
+    it('waits only for topics newly added by a partially overlapping call', async (): Promise<void> => {
-      await firstReleased;
+        const broadcaster = createBroadcaster();
-      await originalFirstSend(message);
+        const connection = new TestConnection(true, true);
-    };
+        const first = broadcaster.subscribe(routeStream(connection), [ 'a' ]);
-    second.send = async (message) => {
+        const second = broadcaster.subscribe(routeStream(connection), [ 'a', 'b' ]);
-      await TestConnection.prototype.send.call(second, message);
-      markSecondSent();
-    };
 
-    const firstSubscription = broadcaster.subscribe(routeStream(first), [
+        await broadcaster.unsubscribe(routeStream(connection), [ 'a' ]);
-      "items",
+        await expect(first).resolves.toBeUndefined();
-    ]);
+        await expectPending(second);
-    const secondSubscription = broadcaster.subscribe(routeStream(second), [
+
-      "items",
+        await broadcaster.unsubscribe(routeStream(connection), [ 'b' ]);
-    ]);
+        await expect(second).resolves.toBeUndefined();
-    const publication = broadcaster.publish("items", {
-      type: "item-changed",
-      data: {},
     });
 
-    await secondSent;
+    it('resolves every pending subscription and removes topics on close', async (): Promise<void> => {
-    releaseFirst();
+        const broadcaster = createBroadcaster();
-    await publication;
+        const connection = new TestConnection(true, false);
+        const first = broadcaster.subscribe(routeStream(connection), [ 'a' ]);
+        const second = broadcaster.subscribe(routeStream(connection), [ 'b' ]);
 
-    expect(first.messages).toHaveLength(1);
+        connection.close();
-    expect(second.messages).toHaveLength(1);
+        await Promise.all([ first, second ]);
+        await broadcaster.publish('a', { type: 'changed', data: null });
+        await broadcaster.publish('b', { type: 'changed', data: null });
 
-    first.close();
+        expect(connection.messages).toEqual([]);
-    second.close();
-    await Promise.all([firstSubscription, secondSubscription]);
-  });
-
-  it("closes and removes a connection whose event delivery fails", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, true);
-    connection.send = vi.fn().mockRejectedValue(new Error("socket failed"));
-    const subscribed = broadcaster.subscribe(routeStream(connection), [
-      "items",
-    ]);
-
-    await broadcaster.publish("items", {
-      type: "item-changed",
-      data: {},
     });
-    await subscribed;
 
-    expect(connection.closed).toBe(true);
+    it('immediately resolves registration against an already-closed connection', async (): Promise<void> => {
-    expect(connection.send).toHaveBeenCalledOnce();
+        const broadcaster = createBroadcaster();
+        const connection = new TestConnection(true, false);
+        connection.close();
 
-    await broadcaster.publish("items", {
+        await expect(broadcaster.subscribe(routeStream(connection), [ 'items' ])).resolves.toBeUndefined();
-      type: "item-changed",
+    });
-      data: {},
+
+    it('rejects subscriptions on a non-streaming connection', (): void => {
+        const broadcaster = createBroadcaster();
+        const connection = new TestConnection(false, false);
+
+        expect(() => broadcaster.subscribe(routeStream(connection), [ 'items' ])).toThrowError(expect.objectContaining({ statusCode: 406 }));
+    });
+
+    it('treats an empty subscription and repeated unsubscribe as no-ops', async (): Promise<void> => {
+        const broadcaster = createBroadcaster();
+        const connection = new TestConnection(true, true);
+
+        await expect(broadcaster.subscribe(routeStream(connection), [])).resolves.toBeUndefined();
+        await broadcaster.unsubscribe(routeStream(connection), [ 'missing' ]);
+        await broadcaster.unsubscribe(routeStream(connection));
+
+        expect(connection.closeCallbacks).toHaveLength(0);
+    });
+
+    it('fans out concurrently to independent connections', async (): Promise<void> => {
+        const broadcaster = createBroadcaster();
+        const first = new TestConnection(true, false);
+        const second = new TestConnection(true, false);
+        const originalFirstSend = first.send.bind(first);
+        let releaseFirst: () => void = () => undefined;
+        const firstReleased = new Promise<void>((resolve) => {
+            releaseFirst = resolve;
+        });
+        let markSecondSent: () => void = () => undefined;
+        const secondSent = new Promise<void>((resolve) => {
+            markSecondSent = resolve;
+        });
+
+        first.send = async (message): Promise<void> => {
+            await firstReleased;
+            await originalFirstSend(message);
+        };
+
+        second.send = async (message): Promise<void> => {
+            await TestConnection.prototype.send.call(second, message);
+            markSecondSent();
+        };
+
+        const firstSubscription = broadcaster.subscribe(routeStream(first), [ 'items' ]);
+        const secondSubscription = broadcaster.subscribe(routeStream(second), [ 'items' ]);
+        const publication = broadcaster.publish('items', {
+            type: 'item-changed',
+            data: {},
+        });
+
+        await secondSent;
+        releaseFirst();
+        await publication;
+
+        expect(first.messages).toHaveLength(1);
+        expect(second.messages).toHaveLength(1);
+
+        first.close();
+        second.close();
+        await Promise.all([ firstSubscription, secondSubscription ]);
+    });
+
+    it('closes and removes a connection whose event delivery fails', async (): Promise<void> => {
+        const broadcaster = createBroadcaster();
+        const connection = new TestConnection(true, true);
+        connection.send = vi.fn().mockRejectedValue(new Error('socket failed'));
+        const subscribed = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
+
+        await broadcaster.publish('items', {
+            type: 'item-changed',
+            data: {},
+        });
+        await subscribed;
+
+        expect(connection.closed).toBe(true);
+        expect(connection.send).toHaveBeenCalledOnce();
+
+        await broadcaster.publish('items', {
+            type: 'item-changed',
+            data: {},
+        });
+        expect(connection.send).toHaveBeenCalledOnce();
     });
-    expect(connection.send).toHaveBeenCalledOnce();
-  });
 });

test/subscription-flow.test.ts

@@ -1,76 +1,67 @@
-import { describe, expect, it, vi } from "vitest";
+import { describe, expect, it, vi } from 'vitest';
 
-import type { RouteDefinition, RouteModule } from "../source/routes/types.js";
+import type { RouteDefinition, RouteModule } from '../source/routes/types.ts';
-import { ApplicationRouter } from "../source/services/router.js";
+import { ApplicationRouter } from '../source/services/router.ts';
-import { Broadcaster } from "../source/services/broadcaster.js";
+import { Broadcaster } from '../source/services/broadcaster.ts';
-import { Logger } from "../source/utils/logger.js";
+import { Logger } from '../source/utils/logger.ts';
-import { TestConnection } from "./helpers/test-connection.js";
+import { TestConnection } from './helpers/test-connection.ts';
 
-function moduleWith(routes: RouteDefinition[]): RouteModule {
+const moduleWith = (routes: RouteDefinition[]): RouteModule => {
-  return {
+    return {
-    async getRoutes() {
+        async getRoutes(): Promise<RouteDefinition[]> {
-      return routes;
+            return routes;
-    },
-  };
-}
-
-async function expectPending(promise: Promise<void>): Promise<void> {
-  const settled = vi.fn();
-  void promise.then(settled);
-  await Promise.resolve();
-  expect(settled).not.toHaveBeenCalled();
-}
-
-describe("long-lived subscription dispatch", () => {
-  it("allows duplicate subscribe and unsubscribe requests while the original request waits", async () => {
-    const broadcaster = new Broadcaster(new Logger("subscription-flow-test"));
-    const router = await ApplicationRouter.create([
-      moduleWith([
-        {
-          url: "/items/subscribe",
-          handler: async (stream) => {
-            await broadcaster.subscribe(stream, ["items"]);
-          },
         },
-        {
+    };
-          url: "/items/unsubscribe",
+};
-          handler: async (stream) => {
-            await broadcaster.unsubscribe(stream, ["items"]);
-            await stream.send({});
-          },
-        },
-      ]),
-    ]);
-    const connection = new TestConnection(true, true);
 
-    const original = router.dispatch(
+const expectPending = async (promise: Promise<void>): Promise<void> => {
-      { path: "/items/subscribe", requestId: "subscribe-1" },
+    const settled = vi.fn();
-      connection,
+    void promise.then(settled);
-    );
+    await Promise.resolve();
-    await vi.waitFor(() => expect(connection.closeCallbacks).toHaveLength(1));
+    expect(settled).not.toHaveBeenCalled();
-    await expectPending(original);
+};
 
-    // This request uses a different ApplicationRouteStream over the same
+describe('long-lived subscription dispatch', (): void => {
-    // connection. Since the topic already exists, its dispatch completes.
+    it('allows duplicate subscribe and unsubscribe requests while the original request waits', async (): Promise<void> => {
-    await router.dispatch(
+        const broadcaster = new Broadcaster(new Logger('subscription-flow-test'));
-      { path: "/items/subscribe", requestId: "subscribe-2" },
+        const router = await ApplicationRouter.create([
-      connection,
+            moduleWith([
-    );
+                {
-    await expectPending(original);
+                    url: '/items/subscribe',
+                    handler: async (stream): Promise<void> => {
+                        await broadcaster.subscribe(stream, [ 'items' ]);
+                    },
+                },
+                {
+                    url: '/items/unsubscribe',
+                    handler: async (stream): Promise<void> => {
+                        await broadcaster.unsubscribe(stream, [ 'items' ]);
+                        await stream.send({});
+                    },
+                },
+            ]),
+        ]);
+        const connection = new TestConnection(true, true);
 
-    await router.dispatch(
+        const original = router.dispatch({ path: '/items/subscribe', requestId: 'subscribe-1' }, connection);
-      { path: "/items/unsubscribe", requestId: "unsubscribe-1" },
+        await vi.waitFor(() => expect(connection.closeCallbacks).toHaveLength(1));
-      connection,
+        await expectPending(original);
-    );
-    await original;
 
-    expect(connection.messages).toEqual([
+        // This request uses a different ApplicationRouteStream over the same
-      {
+        // connection. Since the topic already exists, its dispatch completes.
-        id: "unsubscribe-1",
+        await router.dispatch({ path: '/items/subscribe', requestId: 'subscribe-2' }, connection);
-        type: "response",
+        await expectPending(original);
-        statusCode: 200,
+
-        body: {},
+        await router.dispatch({ path: '/items/unsubscribe', requestId: 'unsubscribe-1' }, connection);
-      },
+        await original;
-    ]);
+
-  });
+        expect(connection.messages).toEqual([
+            {
+                id: 'unsubscribe-1',
+                type: 'response',
+                statusCode: 200,
+                body: {},
+            },
+        ]);
+    });
 });