Merge branch '5-add-http-and-sse' into 6-add-websocket-transport
This commit is contained in:
+158
-177
@@ -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
@@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,242 +1,237 @@
|
||||
import { Hono } from "hono";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { Hono } from 'hono';
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import type { RouteDefinition } from "../../../source/routes/types.js";
|
||||
import { ApplicationError } from "../../../source/errors/index.js";
|
||||
import { ApplicationRouter } from "../../../source/services/router.js";
|
||||
import { Broadcaster } from "../../../source/services/broadcaster.js";
|
||||
import { HttpTransportRouter } from "../../../source/services/transport/http-transport.js";
|
||||
import type { AppEnv } from "../../../source/services/transport/transport-router.js";
|
||||
import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils";
|
||||
import { Logger } from "../../../source/utils/logger.js";
|
||||
import { ServerHost } from "../../../source/services/server-host.js";
|
||||
import type { RouteDefinition } from '../../../source/routes/types.ts';
|
||||
import { ApplicationError } from '../../../source/errors/index.ts';
|
||||
import { ApplicationRouter } from '../../../source/services/router.ts';
|
||||
import { Broadcaster } from '../../../source/services/broadcaster.ts';
|
||||
import { HttpTransportRouter } from '../../../source/services/transport/http-transport.ts';
|
||||
import type { AppEnv } from '../../../source/services/transport/transport-router.ts';
|
||||
import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
|
||||
import { Logger } from '../../../source/utils/logger.ts';
|
||||
import { ServerHost } from '../../../source/services/server-host.ts';
|
||||
|
||||
async function createApp(
|
||||
routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]),
|
||||
maxRequestBodyBytes = 1024 * 1024,
|
||||
): Promise<Hono<AppEnv>> {
|
||||
const debug = new Logger("http-transport-test");
|
||||
const broadcaster = new Broadcaster(debug);
|
||||
const resolvedRoutes =
|
||||
typeof routes === "function" ? routes(broadcaster) : routes;
|
||||
const router = await ApplicationRouter.create([
|
||||
{
|
||||
async getRoutes() {
|
||||
return resolvedRoutes;
|
||||
},
|
||||
},
|
||||
]);
|
||||
const transport = new HttpTransportRouter(router, debug);
|
||||
const app = new Hono<AppEnv>();
|
||||
|
||||
app.onError(HttpTransportRouter.createErrorHandler(debug));
|
||||
app.use("*", ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug));
|
||||
app.use("*", HttpTransportRouter.createExtJsonMiddleware(debug));
|
||||
transport.register(app);
|
||||
return app;
|
||||
}
|
||||
|
||||
describe("HttpTransportRouter", () => {
|
||||
it("runs normal HTTP through a non-streaming route stream", async () => {
|
||||
const app = await createApp([
|
||||
{
|
||||
url: "/echo",
|
||||
handler: async (stream) => stream.send(stream.body),
|
||||
},
|
||||
]);
|
||||
const value = new Uint8Array([1, 2, 3]);
|
||||
|
||||
const response = await app.request("/echo", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: toExtendedJson({ value }),
|
||||
});
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(fromExtendedJson(await response.text())).toEqual({ value });
|
||||
});
|
||||
|
||||
it("returns 204 when a normal HTTP route sends nothing", async () => {
|
||||
const app = await createApp([
|
||||
{
|
||||
url: "/nothing",
|
||||
handler: () => undefined,
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request("/nothing", { method: "POST" });
|
||||
|
||||
expect(response.status).toBe(204);
|
||||
expect(await response.text()).toBe("");
|
||||
});
|
||||
|
||||
it("returns normalized errors for non-streaming requests", async () => {
|
||||
const app = await createApp([]);
|
||||
|
||||
const missing = await app.request("/missing", { method: "POST" });
|
||||
expect(missing.status).toBe(404);
|
||||
expect(await missing.json()).toEqual({
|
||||
statusCode: 404,
|
||||
error: "No route found for /missing",
|
||||
});
|
||||
|
||||
const invalid = await app.request("/missing", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: "{",
|
||||
});
|
||||
expect(invalid.status).toBe(400);
|
||||
expect(await invalid.json()).toEqual({
|
||||
statusCode: 400,
|
||||
error: "Invalid JSON in request body",
|
||||
});
|
||||
});
|
||||
|
||||
it("rejects subscribe when normal HTTP has no streaming capability", async () => {
|
||||
const app = await createApp((broadcaster) => [
|
||||
{
|
||||
url: "/items/subscribe",
|
||||
handler: async (stream) => {
|
||||
await broadcaster.subscribe(stream, ["items"]);
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request("/items/subscribe", { method: "POST" });
|
||||
|
||||
expect(response.status).toBe(406);
|
||||
expect(await response.json()).toMatchObject({ statusCode: 406 });
|
||||
});
|
||||
|
||||
it("sends SSE route errors as events and closes only that stream", async () => {
|
||||
const app = await createApp([
|
||||
{
|
||||
url: "/items/subscribe",
|
||||
handler: () => {
|
||||
throw new Error("private storage failure");
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request("/items/subscribe", {
|
||||
method: "POST",
|
||||
headers: { accept: "text/event-stream" },
|
||||
});
|
||||
const events = await response.text();
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(events).toContain("event: error");
|
||||
expect(events).toContain(
|
||||
'data: {"statusCode":500,"error":"Internal Server Error"}',
|
||||
);
|
||||
expect(events).not.toContain("private storage failure");
|
||||
});
|
||||
|
||||
it("sends a normal route as one SSE response event and then closes", async () => {
|
||||
const app = await createApp([
|
||||
{
|
||||
url: "/echo",
|
||||
handler: (stream) => stream.send({ ok: true }),
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request("/echo", {
|
||||
method: "POST",
|
||||
headers: { accept: "text/event-stream" },
|
||||
});
|
||||
const events = await response.text();
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(events).toContain("event: response");
|
||||
expect(events).toContain('data: {"ok":true}');
|
||||
});
|
||||
|
||||
it("keeps SSE open until the route's subscription promise resolves", async () => {
|
||||
let removeSubscription: () => Promise<void> = async () => undefined;
|
||||
let markSubscribed: () => void = () => undefined;
|
||||
const subscribed = new Promise<void>((resolve) => {
|
||||
markSubscribed = resolve;
|
||||
});
|
||||
const app = await createApp((broadcaster) => [
|
||||
{
|
||||
url: "/items/subscribe",
|
||||
handler: async (stream) => {
|
||||
const topics = ["items"];
|
||||
removeSubscription = () => broadcaster.unsubscribe(stream, topics);
|
||||
|
||||
const removed = broadcaster.subscribe(stream, topics);
|
||||
markSubscribed();
|
||||
await removed;
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request("/items/subscribe", {
|
||||
method: "POST",
|
||||
headers: { accept: "text/event-stream" },
|
||||
});
|
||||
const body = response.text();
|
||||
const completed = vi.fn();
|
||||
void body.then(completed);
|
||||
|
||||
await subscribed;
|
||||
await Promise.resolve();
|
||||
expect(completed).not.toHaveBeenCalled();
|
||||
|
||||
await removeSubscription();
|
||||
|
||||
expect(await body).toBe("");
|
||||
expect(completed).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("rejects unsubscribe over non-bidirectional SSE", async () => {
|
||||
const app = await createApp((broadcaster) => [
|
||||
{
|
||||
url: "/items/unsubscribe",
|
||||
handler: async (stream) => {
|
||||
if (!stream.bidirectional) {
|
||||
throw new ApplicationError(
|
||||
400,
|
||||
"This route requires an existing bidirectional stream",
|
||||
);
|
||||
}
|
||||
|
||||
await broadcaster.unsubscribe(stream, ["items"]);
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request("/items/unsubscribe", {
|
||||
method: "POST",
|
||||
headers: { accept: "text/event-stream" },
|
||||
});
|
||||
const events = await response.text();
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(events).toContain("event: error");
|
||||
expect(events).toContain('"statusCode":400');
|
||||
});
|
||||
it("rejects HTTP bodies larger than the configured byte limit", async () => {
|
||||
const app = await createApp(
|
||||
[
|
||||
const createApp = async (
|
||||
routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]),
|
||||
maxRequestBodyBytes = 1024 * 1024,
|
||||
): Promise<Hono<AppEnv>> => {
|
||||
const debug = new Logger('http-transport-test');
|
||||
const broadcaster = new Broadcaster(debug);
|
||||
const resolvedRoutes = typeof routes === 'function' ? routes(broadcaster) : routes;
|
||||
const router = await ApplicationRouter.create([
|
||||
{
|
||||
url: "/echo",
|
||||
handler: async (stream) => stream.send(stream.body),
|
||||
async getRoutes(): Promise<RouteDefinition[]> {
|
||||
return resolvedRoutes;
|
||||
},
|
||||
},
|
||||
],
|
||||
32,
|
||||
);
|
||||
]);
|
||||
const transport = new HttpTransportRouter(router, debug);
|
||||
const app = new Hono<AppEnv>();
|
||||
|
||||
const response = await app.request("/echo", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ value: "x".repeat(64) }),
|
||||
app.onError(HttpTransportRouter.createErrorHandler(debug));
|
||||
app.use('*', ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug));
|
||||
app.use('*', HttpTransportRouter.createExtJsonMiddleware(debug));
|
||||
transport.register(app);
|
||||
|
||||
return app;
|
||||
};
|
||||
|
||||
describe('HttpTransportRouter', (): void => {
|
||||
it('runs normal HTTP through a non-streaming route stream', async (): Promise<void> => {
|
||||
const app = await createApp([
|
||||
{
|
||||
url: '/echo',
|
||||
handler: async (stream): Promise<void> => stream.send(stream.body),
|
||||
},
|
||||
]);
|
||||
const value = new Uint8Array([ 1, 2, 3 ]);
|
||||
|
||||
const response = await app.request('/echo', {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: toExtendedJson({ value }),
|
||||
});
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(fromExtendedJson(await response.text())).toEqual({ value });
|
||||
});
|
||||
|
||||
expect(response.status).toBe(413);
|
||||
expect(await response.json()).toEqual({
|
||||
statusCode: 413,
|
||||
error: "Request body exceeds the 32 byte limit",
|
||||
it('returns 204 when a normal HTTP route sends nothing', async (): Promise<void> => {
|
||||
const app = await createApp([
|
||||
{
|
||||
url: '/nothing',
|
||||
handler: (): void => undefined,
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request('/nothing', { method: 'POST' });
|
||||
|
||||
expect(response.status).toBe(204);
|
||||
expect(await response.text()).toBe('');
|
||||
});
|
||||
|
||||
it('returns normalized errors for non-streaming requests', async (): Promise<void> => {
|
||||
const app = await createApp([]);
|
||||
|
||||
const missing = await app.request('/missing', { method: 'POST' });
|
||||
expect(missing.status).toBe(404);
|
||||
expect(await missing.json()).toEqual({
|
||||
statusCode: 404,
|
||||
error: 'No route found for /missing',
|
||||
});
|
||||
|
||||
const invalid = await app.request('/missing', {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: '{',
|
||||
});
|
||||
expect(invalid.status).toBe(400);
|
||||
expect(await invalid.json()).toEqual({
|
||||
statusCode: 400,
|
||||
error: 'Invalid JSON in request body',
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects subscribe when normal HTTP has no streaming capability', async (): Promise<void> => {
|
||||
const app = await createApp((broadcaster) => [
|
||||
{
|
||||
url: '/items/subscribe',
|
||||
handler: async (stream): Promise<void> => {
|
||||
await broadcaster.subscribe(stream, [ 'items' ]);
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request('/items/subscribe', { method: 'POST' });
|
||||
|
||||
expect(response.status).toBe(406);
|
||||
expect(await response.json()).toMatchObject({ statusCode: 406 });
|
||||
});
|
||||
|
||||
it('sends SSE route errors as events and closes only that stream', async (): Promise<void> => {
|
||||
const app = await createApp([
|
||||
{
|
||||
url: '/items/subscribe',
|
||||
handler: (): void => {
|
||||
throw new Error('private storage failure');
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request('/items/subscribe', {
|
||||
method: 'POST',
|
||||
headers: { accept: 'text/event-stream' },
|
||||
});
|
||||
const events = await response.text();
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(events).toContain('event: error');
|
||||
expect(events).toContain('data: {"statusCode":500,"error":"Internal Server Error"}');
|
||||
expect(events).not.toContain('private storage failure');
|
||||
});
|
||||
|
||||
it('sends a normal route as one SSE response event and then closes', async (): Promise<void> => {
|
||||
const app = await createApp([
|
||||
{
|
||||
url: '/echo',
|
||||
handler: (stream): Promise<void> => stream.send({ ok: true }),
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request('/echo', {
|
||||
method: 'POST',
|
||||
headers: { accept: 'text/event-stream' },
|
||||
});
|
||||
const events = await response.text();
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(events).toContain('event: response');
|
||||
expect(events).toContain('data: {"ok":true}');
|
||||
});
|
||||
|
||||
it("keeps SSE open until the route's subscription promise resolves", async (): Promise<void> => {
|
||||
let removeSubscription: () => Promise<void> = async () => undefined;
|
||||
let markSubscribed: () => void = () => undefined;
|
||||
const subscribed = new Promise<void>((resolve) => {
|
||||
markSubscribed = resolve;
|
||||
});
|
||||
const app = await createApp((broadcaster) => [
|
||||
{
|
||||
url: '/items/subscribe',
|
||||
handler: async (stream): Promise<void> => {
|
||||
const topics = [ 'items' ];
|
||||
removeSubscription = (): Promise<void> => broadcaster.unsubscribe(stream, topics);
|
||||
|
||||
const removed = broadcaster.subscribe(stream, topics);
|
||||
markSubscribed();
|
||||
await removed;
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request('/items/subscribe', {
|
||||
method: 'POST',
|
||||
headers: { accept: 'text/event-stream' },
|
||||
});
|
||||
const body = response.text();
|
||||
const completed = vi.fn();
|
||||
void body.then(completed);
|
||||
|
||||
await subscribed;
|
||||
await Promise.resolve();
|
||||
expect(completed).not.toHaveBeenCalled();
|
||||
|
||||
await removeSubscription();
|
||||
|
||||
expect(await body).toBe('');
|
||||
expect(completed).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it('rejects unsubscribe over non-bidirectional SSE', async (): Promise<void> => {
|
||||
const app = await createApp((broadcaster) => [
|
||||
{
|
||||
url: '/items/unsubscribe',
|
||||
handler: async (stream): Promise<void> => {
|
||||
if (!stream.bidirectional) {
|
||||
throw new ApplicationError(400, 'This route requires an existing bidirectional stream');
|
||||
}
|
||||
|
||||
await broadcaster.unsubscribe(stream, [ 'items' ]);
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await app.request('/items/unsubscribe', {
|
||||
method: 'POST',
|
||||
headers: { accept: 'text/event-stream' },
|
||||
});
|
||||
const events = await response.text();
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(events).toContain('event: error');
|
||||
expect(events).toContain('"statusCode":400');
|
||||
});
|
||||
it('rejects HTTP bodies larger than the configured byte limit', async (): Promise<void> => {
|
||||
const app = await createApp(
|
||||
[
|
||||
{
|
||||
url: '/echo',
|
||||
handler: async (stream): Promise<void> => stream.send(stream.body),
|
||||
},
|
||||
],
|
||||
32,
|
||||
);
|
||||
|
||||
const response = await app.request('/echo', {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify({ value: 'x'.repeat(64) }),
|
||||
});
|
||||
|
||||
expect(response.status).toBe(413);
|
||||
expect(await response.json()).toEqual({
|
||||
statusCode: 413,
|
||||
error: 'Request body exceeds the 32 byte limit',
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user