Merge branch '4-add-broadcaster' into 5-add-http-and-sse

This commit is contained in:
2026-08-03 03:36:02 +00:00
13 changed files with 366 additions and 403 deletions
+158 -177
View File
@@ -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.js";
import { ApplicationRouteStream } from "../../source/services/route-stream.js";
import { Logger } from "../../source/utils/logger.js";
import { TestConnection } from "../helpers/test-connection.js";
import { Broadcaster } from '../../source/services/broadcaster.ts';
import { ApplicationRouteStream } from '../../source/services/route-stream.ts';
import { Logger } from '../../source/utils/logger.ts';
import { TestConnection } from '../helpers/test-connection.ts';
function createBroadcaster(): Broadcaster {
return new Broadcaster(new Logger("broadcaster-test"));
}
const createBroadcaster = (): Broadcaster => {
return new Broadcaster(new Logger('broadcaster-test'));
};
function routeStream(connection: TestConnection): ApplicationRouteStream {
return new ApplicationRouteStream(connection, undefined);
}
const routeStream = (connection: TestConnection): ApplicationRouteStream => {
return new ApplicationRouteStream(connection, undefined);
};
async function expectPending(promise: Promise<void>): Promise<void> {
const settled = vi.fn();
void promise.then(settled);
await Promise.resolve();
expect(settled).not.toHaveBeenCalled();
}
const expectPending = async (promise: Promise<void>): Promise<void> => {
const settled = vi.fn();
void promise.then(settled);
await Promise.resolve();
expect(settled).not.toHaveBeenCalled();
};
describe("Broadcaster subscriptions", () => {
it("delivers events and resolves after a later request removes the topic", async () => {
const broadcaster = createBroadcaster();
const connection = new TestConnection(true, true);
const subscribed = broadcaster.subscribe(routeStream(connection), [
"items",
]);
describe('Broadcaster subscriptions', () => {
it('delivers events and resolves after a later request removes the topic', async (): Promise<void> => {
const broadcaster = createBroadcaster();
const connection = new TestConnection(true, true);
const subscribed = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
await expectPending(subscribed);
await broadcaster.publish("items", {
type: "item-changed",
data: { id: "a" },
await expectPending(subscribed);
await broadcaster.publish('items', {
type: 'item-changed',
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([
expect.objectContaining({
type: "item-changed",
data: { id: "a" },
}),
]);
it('resolves fully duplicate subscriptions immediately', async (): Promise<void> => {
const broadcaster = createBroadcaster();
const connection = new TestConnection(true, true);
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
// original subscription.
await broadcaster.unsubscribe(routeStream(connection), ["items"]);
await expect(subscribed).resolves.toBeUndefined();
});
await expect(duplicate).resolves.toBeUndefined();
await expectPending(first);
expect(connection.closeCallbacks).toHaveLength(1);
it("resolves fully duplicate subscriptions immediately", async () => {
const broadcaster = createBroadcaster();
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;
await broadcaster.unsubscribe(routeStream(connection), [ 'items' ]);
await first;
});
first.send = async (message) => {
await firstReleased;
await originalFirstSend(message);
};
second.send = async (message) => {
await TestConnection.prototype.send.call(second, message);
markSecondSent();
};
it('waits only for topics newly added by a partially overlapping call', async (): Promise<void> => {
const broadcaster = createBroadcaster();
const connection = new TestConnection(true, true);
const first = broadcaster.subscribe(routeStream(connection), [ 'a' ]);
const second = broadcaster.subscribe(routeStream(connection), [ 'a', 'b' ]);
const firstSubscription = broadcaster.subscribe(routeStream(first), [
"items",
]);
const secondSubscription = broadcaster.subscribe(routeStream(second), [
"items",
]);
const publication = broadcaster.publish("items", {
type: "item-changed",
data: {},
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();
});
await secondSent;
releaseFirst();
await publication;
it('resolves every pending subscription and removes topics on close', async (): Promise<void> => {
const broadcaster = createBroadcaster();
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);
expect(second.messages).toHaveLength(1);
connection.close();
await Promise.all([ first, second ]);
await broadcaster.publish('a', { type: 'changed', data: null });
await broadcaster.publish('b', { type: 'changed', data: null });
first.close();
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: {},
expect(connection.messages).toEqual([]);
});
await subscribed;
expect(connection.closed).toBe(true);
expect(connection.send).toHaveBeenCalledOnce();
it('immediately resolves registration against an already-closed connection', async (): Promise<void> => {
const broadcaster = createBroadcaster();
const connection = new TestConnection(true, false);
connection.close();
await broadcaster.publish("items", {
type: "item-changed",
data: {},
await expect(broadcaster.subscribe(routeStream(connection), [ 'items' ])).resolves.toBeUndefined();
});
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();
});
});
+94 -117
View File
@@ -1,134 +1,111 @@
import { describe, expect, it } from "vitest";
import { describe, expect, it } from 'vitest';
import type { RouteDefinition, RouteModule } from "../../source/routes/types.js";
import { ApplicationError } from "../../source/errors/index.js";
import { ApplicationRouter } from "../../source/services/router.js";
import { TestConnection } from "../helpers/test-connection.js";
import type { RouteDefinition, RouteModule } from '../../source/routes/types.ts';
import { ApplicationRouter } from '../../source/services/router.ts';
import { TestConnection } from '../helpers/test-connection.ts';
function moduleWith(routes: RouteDefinition[]): RouteModule {
return {
async getRoutes() {
return routes;
},
};
}
const moduleWith = (routes: RouteDefinition[]): RouteModule => {
return {
async getRoutes(): Promise<RouteDefinition[]> {
return routes;
},
};
};
describe("ApplicationRouter initialization", () => {
it("rejects duplicate exact paths during startup", async () => {
const route = { url: "/echo", handler: () => undefined };
describe('ApplicationRouter initialization', (): void => {
it('rejects duplicate exact paths during startup', async (): Promise<void> => {
const route = { url: '/echo', handler: (): void => undefined };
await expect(
ApplicationRouter.create([moduleWith([route]), moduleWith([route])]),
).rejects.toThrow("Duplicate application route: /echo");
});
await expect(ApplicationRouter.create([ moduleWith([ route ]), moduleWith([ route ]) ])).rejects.toThrow('Duplicate application route: /echo');
});
it.each(["echo", "/", "/items/:id", "/items?active=true", "/items#active"])(
"rejects the invalid route path %s",
async (url) => {
await expect(
ApplicationRouter.create([
moduleWith([{ url, handler: () => undefined }]),
]),
).rejects.toThrow("Invalid application route");
},
);
it.each([ 'echo', '/', '/items/:id', '/items?active=true', '/items#active' ])('rejects the invalid route path %s', async (url): Promise<void> => {
await expect(ApplicationRouter.create([ moduleWith([{ url, handler: (): void => undefined }]) ])).rejects.toThrow('Invalid application route');
});
});
describe("ApplicationRouter dispatch", () => {
it("binds the connection, body, and request ID to one route stream", async () => {
const connection = new TestConnection(false, false);
const router = await ApplicationRouter.create([
moduleWith([
{
url: "/echo",
handler: async (stream) => {
expect(stream.connection).toBe(connection);
await stream.send(stream.body);
},
},
]),
]);
describe('ApplicationRouter dispatch', (): void => {
it('binds the connection, body, and request ID to one route stream', async (): Promise<void> => {
const connection = new TestConnection(false, false);
const router = await ApplicationRouter.create([
moduleWith([
{
url: '/echo',
handler: async (stream): Promise<void> => {
expect(stream.connection).toBe(connection);
await stream.send(stream.body);
},
},
]),
]);
await router.dispatch(
{ path: "/echo", body: { value: 1 }, requestId: "request-1" },
connection,
);
await router.dispatch({ path: '/echo', body: { value: 1 }, requestId: 'request-1' }, connection);
expect(connection.messages).toEqual([
{
id: "request-1",
type: "response",
statusCode: 200,
body: { value: 1 },
},
]);
expect(connection.messages).toEqual([
{
id: 'request-1',
type: 'response',
statusCode: 200,
body: { value: 1 },
},
]);
await expect(
router.dispatch({ path: "/echo/other", body: {} }, connection),
).rejects.toMatchObject({ statusCode: 404 });
});
await expect(router.dispatch({ path: '/echo/other', body: {} }, connection)).rejects.toMatchObject({ statusCode: 404 });
});
it("preserves correlation when concurrent requests finish out of order", async () => {
const completions = new Map<string, () => void>();
const router = await ApplicationRouter.create([
moduleWith([
{
url: "/delayed",
handler: async (stream) => {
const key = (stream.body as { key: string }).key;
await new Promise<void>((resolve) => completions.set(key, resolve));
await stream.send({ key });
},
},
]),
]);
const connection = new TestConnection(true, true);
it('preserves correlation when concurrent requests finish out of order', async (): Promise<void> => {
const completions = new Map<string, () => void>();
const router = await ApplicationRouter.create([
moduleWith([
{
url: '/delayed',
handler: async (stream): Promise<void> => {
const key = (stream.body as { key: string }).key;
await new Promise<void>((resolve) => completions.set(key, resolve));
await stream.send({ key });
},
},
]),
]);
const connection = new TestConnection(true, true);
const first = router.dispatch(
{ path: "/delayed", body: { key: "A" }, requestId: "A" },
connection,
);
const second = router.dispatch(
{ path: "/delayed", body: { key: "B" }, requestId: "B" },
connection,
);
const first = router.dispatch({ path: '/delayed', body: { key: 'A' }, requestId: 'A' }, connection);
const second = router.dispatch({ path: '/delayed', body: { key: 'B' }, requestId: 'B' }, connection);
completions.get("B")?.();
await second;
completions.get("A")?.();
await first;
completions.get('B')?.();
await second;
completions.get('A')?.();
await first;
expect(connection.messages).toEqual([
{
id: "B",
type: "response",
statusCode: 200,
body: { key: "B" },
},
{
id: "A",
type: "response",
statusCode: 200,
body: { key: "A" },
},
]);
});
expect(connection.messages).toEqual([
{
id: 'B',
type: 'response',
statusCode: 200,
body: { key: 'B' },
},
{
id: 'A',
type: 'response',
statusCode: 200,
body: { key: 'A' },
},
]);
});
it("propagates route failures without infrastructure-specific cleanup", async () => {
const error = new Error("route failed");
const router = await ApplicationRouter.create([
moduleWith([
{
url: "/failure",
handler: () => {
throw error;
},
},
]),
]);
it('propagates route failures without infrastructure-specific cleanup', async (): Promise<void> => {
const error = new Error('route failed');
const router = await ApplicationRouter.create([
moduleWith([
{
url: '/failure',
handler: (): void => {
throw error;
},
},
]),
]);
await expect(
router.dispatch({ path: "/failure" }, new TestConnection(false, false)),
).rejects.toBe(error);
});
await expect(router.dispatch({ path: '/failure' }, new TestConnection(false, false))).rejects.toBe(error);
});
});