Harvmaster/sync-server-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch '6-add-websocket-transport' into 7-add-resources-route

Browse Source
This commit is contained in:
Harvmaster
2026-08-03 03:41:09 +00:00
parent f25ec076b0 c9845d0f28
commit 6febaf327a
21 changed files with 816 additions and 886 deletions
Download Patch File Download Diff File Expand all files Collapse all files
source/app.ts → source/index.ts
+3 -1
View File
@@ -39,8 +39,9 @@ export class App {
// Both transports share one ApplicationRouter. Routes use the shared
// Broadcaster directly, while HTTP and WebSocket remain protocol adapters.
const http = new HttpTransportRouter(router, debug);
const ws = new WsTransportRouter(router, debug, config.server.maxRequestBodyBytes);
const host = new ServerHost(config, debug, [http, ws]);
const host = new ServerHost(config, debug, [ http, ws ]);
return new App(host, database);
}
@@ -53,6 +54,7 @@ export class App {
) {}
async start(): Promise<void> {
await this.database.start();
await this.host.start();
}
source/routes/resources.ts
+7 -7
View File
@@ -42,7 +42,7 @@ const resourceIdsSchema = z.object({
resourceId: z
.array(z.string().min(1))
.min(1, 'At least one resourceId is required')
.transform((ids) => [...new Set(ids)]),
.transform((ids) => [ ...new Set(ids) ]),
});
type WriteResource = z.infer<typeof writeResource>;
@@ -95,7 +95,7 @@ export class DataRoute implements RouteModule {
const resourceIds = this.getResourceIds(stream);
// Remove duplicates.
const uniqueIds = [...new Set(resourceIds)];
const uniqueIds = [ ...new Set(resourceIds) ];
// If there are no resource ids, return an empty array.
if (uniqueIds.length === 0) {
@@ -107,7 +107,7 @@ export class DataRoute implements RouteModule {
// Read the data from the database.
const rows = await this.database.db
.selectFrom('resource_data')
.select(['resource_id', 'public_key', 'blob', 'timestamp'])
.select([ 'resource_id', 'public_key', 'blob', 'timestamp' ])
.where('resource_id', 'in', uniqueIds)
.orderBy('timestamp', 'asc')
.execute();
@@ -155,11 +155,10 @@ export class DataRoute implements RouteModule {
timestamp,
})
.onConflict((oc) =>
oc.columns(['resource_id', 'public_key']).doUpdateSet({
oc.columns([ 'resource_id', 'public_key' ]).doUpdateSet({
blob,
timestamp,
}),
)
}))
.execute();
}
});
@@ -276,7 +275,8 @@ export class DataRoute implements RouteModule {
const signature = hexToBin(signatureHex);
// Create a SHA-256 hash of the payload.
const messageHash = createHash('sha256').update(payload).digest();
const messageHash = createHash('sha256').update(payload)
.digest();
// Low-S normalization rejects malleable signature encodings.
return secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash);
source/routes/types.ts
+3
View File
@@ -1,6 +1,7 @@
import type { BaseStream } from '../services/stream/base-stream.js';
export type RouteSendOptions = {
/** Defaults to `response`; any other value sends an application event. */
type?: string;
@@ -15,6 +16,7 @@ export type RouteSendOptions = {
* connection lifetime are shared with other requests on the same connection.
*/
export interface RouteStream {
/** Connection shared by every request on the same transport session. */
readonly connection: BaseStream;
@@ -29,6 +31,7 @@ export type RouteHandler = (stream: RouteStream) => void | Promise<void>;
/** An exact application route with no transport-specific metadata. */
export type RouteDefinition = {
/** Exact route name. Parameter and wildcard syntax are not supported. */
url: string;
handler: RouteHandler;
source/services/broadcaster.ts
+2
View File
@@ -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>;
source/services/router.ts
+1
View File
@@ -5,6 +5,7 @@ import type { BaseStream } from './stream/base-stream.ts';
/** Canonical request produced by every transport adapter. */
export type ApplicationRequest = {
/** Exact application route name. */
path: string;
source/services/server-host.ts
+4 -4
View File
@@ -40,8 +40,8 @@ export class ServerHost {
const corsMiddleware = cors({
origin: corsConfig.origin ?? '*',
allowMethods: corsConfig.methods ?? ['POST', 'OPTIONS'],
allowHeaders: corsConfig.allowedHeaders ?? ['Content-Type', 'Accept'],
allowMethods: corsConfig.methods ?? [ 'POST', 'OPTIONS' ],
allowHeaders: corsConfig.allowedHeaders ?? [ 'Content-Type', 'Accept' ],
});
this.app.use('*', corsMiddleware);
@@ -77,7 +77,7 @@ export class ServerHost {
throw new Error('ServerHost supports only one WebSocket upgrade server');
}
const [upgradeTransport] = upgradeTransports;
const [ upgradeTransport ] = upgradeTransports;
this.server = serve({
fetch: this.app.fetch,
@@ -124,7 +124,7 @@ export class ServerHost {
const closeTransports = this.transports.map((transport) => Promise.resolve(transport.stop?.()));
// Create a promise that resolves when the server and transports are closed
this.stopPromise = Promise.all([closeServer, ...closeTransports]).then(() => {
this.stopPromise = Promise.all([ closeServer, ...closeTransports ]).then(() => {
this.stopPromise = undefined;
});
source/services/storage/database.ts
+11 -6
View File
@@ -6,6 +6,7 @@ import type { Logger } from '../../utils/logger.ts';
/** Options required to open a SQLite database connection. */
export type DatabaseOptions = {
/** Filesystem path to the SQLite database file. */
path: string;
@@ -39,9 +40,6 @@ export class Database {
this.kysely = new Kysely<DatabaseTables>({
dialect: this.dialect,
});
// Configure the SQLite pragmas.
this.configurePragmas();
}
/**
@@ -53,6 +51,13 @@ export class Database {
return this.kysely;
}
async start(): Promise<void> {
this.debug('starting database connection');
// Configure the SQLite pragmas.
await this.configurePragmas();
}
/**
* Destroys the database connection.
*/
@@ -66,10 +71,10 @@ export class Database {
*
* WAL improves write concurrency; foreign keys enforce referential integrity.
*/
private configurePragmas(): void {
private async configurePragmas(): Promise<void> {
this.debug('configuring SQLite pragmas');
this.kysely.executeQuery(CompiledQuery.raw('PRAGMA journal_mode = WAL'));
this.kysely.executeQuery(CompiledQuery.raw('PRAGMA foreign_keys = ON'));
await this.kysely.executeQuery(CompiledQuery.raw('PRAGMA journal_mode = WAL'));
await this.kysely.executeQuery(CompiledQuery.raw('PRAGMA foreign_keys = ON'));
}
}
source/services/storage/migrations/001-resources.ts
+3 -2
View File
@@ -23,7 +23,7 @@ export const up = async (db: Kysely<DatabaseTables>): Promise<void> => {
.addColumn('public_key', 'text', (col) => col.notNull())
.addColumn('blob', 'blob', (col) => col.notNull())
.addColumn('timestamp', 'integer', (col) => col.notNull().defaultTo(millisecondTime))
.addPrimaryKeyConstraint('pk_resource_data', ['resource_id', 'public_key'])
.addPrimaryKeyConstraint('pk_resource_data', [ 'resource_id', 'public_key' ])
.execute();
};
@@ -33,5 +33,6 @@ export const up = async (db: Kysely<DatabaseTables>): Promise<void> => {
* @param db - Kysely database to apply the rollback against.
*/
export const down = async (db: Kysely<DatabaseTables>): Promise<void> => {
await db.schema.dropTable('resource_data').ifExists().execute();
await db.schema.dropTable('resource_data').ifExists()
.execute();
};
source/services/storage/tables.ts
+1
View File
@@ -10,6 +10,7 @@ export type BlobColumn = ColumnType<Buffer, Buffer | Uint8Array, Buffer>;
* One row per (resource_id, public_key). Each publicKey owns a slot within a shared resource.
*/
export interface ResourceDataTable {
/** Shared resource identifier grouping related instances. */
resource_id: string;
source/services/stream/base-stream.ts
+2
View File
@@ -1,5 +1,6 @@
/** A normal request/response result before transport encoding. */
export type StreamResponse = {
/** Optional correlation ID for multiplexed transports. */
id?: string;
@@ -15,6 +16,7 @@ export type StreamResponse = {
/** An application event before a transport applies its wire encoding. */
export type StreamEvent = {
/** Optional event or correlation ID. */
id?: string;
source/services/transport/transport-router.ts
+3
View File
@@ -4,6 +4,7 @@ import type { Hono } from 'hono';
/** Hono variables populated by transport-boundary middleware. */
export type AppEnv = {
Variables: {
/** Decoded Extended JSON request body, when present. */
parsedBody?: unknown;
@@ -19,6 +20,7 @@ export type AppEnv = {
* not application routing. Implementations remain unaware of route modules.
*/
export interface TransportRouter {
/**
* Attach wire endpoints and middleware to the shared Hono application.
*
@@ -32,6 +34,7 @@ export interface TransportRouter {
/** A transport which also supplies the WebSocket server used during upgrade. */
export interface UpgradeTransportRouter extends TransportRouter {
/** WebSocket server instance passed to the Node HTTP listener. */
readonly websocketServer: WebSocketServerLike;
}
source/services/transport/ws-transport.ts
+4 -5
View File
@@ -20,7 +20,8 @@ const WS_ROUTE = '/ws';
// Strict validation prevents legacy or protocol-specific fields reaching routes.
const wsRequestSchema = z
.object({
id: z.string().min(1).optional(),
id: z.string().min(1)
.optional(),
path: z.string().min(1),
body: z.unknown().optional(),
})
@@ -178,13 +179,11 @@ export class WsTransportRouter implements UpgradeTransportRouter {
* @param error - Failure to normalize into the public error shape.
*/
private sendError(ws: WSContext, requestId: string | undefined, error: unknown): void {
ws.send(
toExtendedJson({
ws.send(toExtendedJson({
...(requestId === undefined ? {} : { id: requestId }),
type: 'error',
...normalizePublicError(error),
}),
);
}));
}
/**
test/helpers/test-connection.ts
+3 -5
View File
@@ -1,7 +1,4 @@
import {
BaseStream,
type StreamMessage,
} from "../../source/services/stream/base-stream.js";
import { BaseStream, type StreamMessage } from '../../source/services/stream/base-stream.ts';
/** Minimal observable connection used by application and broadcaster tests. */
export class TestConnection extends BaseStream {
@@ -18,7 +15,7 @@ export class TestConnection extends BaseStream {
async send(message: StreamMessage): Promise<void> {
if (this.closed) {
throw new Error("connection is closed");
throw new Error('connection is closed');
}
this.messages.push(message);
@@ -37,6 +34,7 @@ export class TestConnection extends BaseStream {
onClose(callback: () => void): void {
if (this.closed) {
callback();
return;
}
test/routes/resources.test.ts
+32 -49
View File
@@ -1,24 +1,24 @@
import { describe, expect, it, vi } from "vitest";
import { describe, expect, it, vi } from 'vitest';
import { DataRoute } from "../../source/routes/resources.js";
import { UnauthorizedError } from "../../source/errors/index.js";
import { type BaseBroadcaster } from "../../source/services/broadcaster.js";
import { ApplicationRouteStream } from "../../source/services/route-stream.js";
import { Database } from "../../source/services/storage/database.js";
import { TestConnection } from "../helpers/test-connection.js";
import { HTTP_STATUS_CODE_NOT_ACCEPTED } from "../../source/constants.js";
import { DataRoute } from '../../source/routes/resources.ts';
import { UnauthorizedError } from '../../source/errors/index.ts';
import { type BaseBroadcaster } from '../../source/services/broadcaster.ts';
import { ApplicationRouteStream } from '../../source/services/route-stream.ts';
import type { Database } from '../../source/services/storage/database.ts';
import { TestConnection } from '../helpers/test-connection.ts';
import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../../source/constants.ts';
function createBroadcasterStub() {
const createBroadcasterStub = (): BaseBroadcaster => {
return {
subscribe: vi.fn(),
unsubscribe: vi.fn().mockResolvedValue(undefined),
publish: vi.fn(),
sendEvent: vi.fn(),
} as unknown as BaseBroadcaster;
}
};
describe("DataRoute subscriptions", () => {
it("subscribes to future resource changes until removal", async () => {
describe('DataRoute subscriptions', (): void => {
it('subscribes to future resource changes until removal', async (): Promise<void> => {
let resolveRemoved: () => void = () => undefined;
const removed = new Promise<void>((resolve) => {
resolveRemoved = resolve;
@@ -32,30 +32,22 @@ describe("DataRoute subscriptions", () => {
vi.mocked(broadcaster.subscribe).mockReturnValue(removed);
const connection = new TestConnection(true, false);
const stream = new ApplicationRouteStream(connection, {
resourceId: ["a", "b"],
resourceId: [ 'a', 'b' ],
});
const route = new DataRoute(storage, broadcaster, 0);
const execution = route.subscribeData(stream);
expect(broadcaster.subscribe).toHaveBeenCalledWith(
stream,
["resource:a", "resource:b"],
);
expect(broadcaster.subscribe).toHaveBeenCalledWith(stream, [ 'resource:a', 'resource:b' ]);
expect(storage.db.transaction).not.toHaveBeenCalled();
expect(connection.messages).toEqual([]);
await expect(
Promise.race([
execution.then(() => "settled"),
Promise.resolve("pending"),
]),
).resolves.toBe("pending");
await expect(Promise.race([ execution.then(() => 'settled'), Promise.resolve('pending') ])).resolves.toBe('pending');
resolveRemoved();
await execution;
});
it("unsubscribes a bidirectional connection and acknowledges the request", async () => {
it('unsubscribes a bidirectional connection and acknowledges the request', async (): Promise<void> => {
const storage = {
db: {
transaction: vi.fn(),
@@ -63,29 +55,23 @@ describe("DataRoute subscriptions", () => {
} as unknown as Database;
const broadcaster = createBroadcasterStub();
const connection = new TestConnection(true, true);
const stream = new ApplicationRouteStream(
connection,
{ resourceId: ["a"] },
"unsubscribe-1",
);
const stream = new ApplicationRouteStream(connection, { resourceId: [ 'a' ] }, 'unsubscribe-1');
const route = new DataRoute(storage, broadcaster, 0);
await route.unsubscribeData(stream);
expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [
"resource:a",
]);
expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [ 'resource:a' ]);
expect(connection.messages).toEqual([
{
id: "unsubscribe-1",
type: "response",
id: 'unsubscribe-1',
type: 'response',
statusCode: 200,
body: {},
},
]);
});
it("rejects selective unsubscribe on a one-way connection", async () => {
it('rejects selective unsubscribe on a one-way connection', async (): Promise<void> => {
const storage = {
db: {
transaction: vi.fn(),
@@ -93,7 +79,7 @@ describe("DataRoute subscriptions", () => {
} as unknown as Database;
const broadcaster = createBroadcasterStub();
const stream = new ApplicationRouteStream(new TestConnection(true, false), {
resourceId: ["a"],
resourceId: [ 'a' ],
});
const route = new DataRoute(storage, broadcaster, 0);
@@ -102,19 +88,18 @@ describe("DataRoute subscriptions", () => {
});
});
describe("DataRoute resource write auth", () => {
it("rejects an invalid resource signature before writing the batch", async () => {
describe('DataRoute resource write auth', (): void => {
it('rejects an invalid resource signature before writing the batch', async (): Promise<void> => {
const storage = {
db: {
transaction: vi.fn(),
},
} as unknown as Database;
const broadcaster = createBroadcasterStub()
const broadcaster = createBroadcasterStub();
const route = new DataRoute(storage, broadcaster, 0);
await expect(
route.writeData({
await expect(route.writeData({
connection: new TestConnection(true, true),
streaming: true,
bidirectional: true,
@@ -122,19 +107,17 @@ describe("DataRoute resource write auth", () => {
body: {
resources: [
{
id: "resource-a",
publicKey: "not-a-public-key",
id: 'resource-a',
publicKey: 'not-a-public-key',
timestamp: Date.now(),
signature: "not-a-signature",
value: new Uint8Array([1, 2, 3]),
signature: 'not-a-signature',
value: new Uint8Array([ 1, 2, 3 ]),
},
],
},
} as unknown as ApplicationRouteStream),
).rejects.toBeInstanceOf(UnauthorizedError);
} as unknown as ApplicationRouteStream)).rejects.toBeInstanceOf(UnauthorizedError);
expect(storage.db.transaction).not.toHaveBeenCalled()
expect(storage.db.transaction).not.toHaveBeenCalled();
expect(broadcaster.publish).not.toHaveBeenCalled();
});
});
test/services/broadcaster.test.ts
+59 -78
View File
@@ -1,133 +1,119 @@
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 {
const routeStream = (connection: TestConnection): ApplicationRouteStream => {
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();
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 () => {
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",
]);
const subscribed = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
await expectPending(subscribed);
await broadcaster.publish("items", {
type: "item-changed",
data: { id: "a" },
await broadcaster.publish('items', {
type: 'item-changed',
data: { id: 'a' },
});
expect(connection.messages).toEqual([
expect.objectContaining({
type: "item-changed",
data: { id: "a" },
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 broadcaster.unsubscribe(routeStream(connection), [ 'items' ]);
await expect(subscribed).resolves.toBeUndefined();
});
it("resolves fully duplicate subscriptions immediately", async () => {
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"]);
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 broadcaster.unsubscribe(routeStream(connection), [ 'items' ]);
await first;
});
it("waits only for topics newly added by a partially overlapping call", async () => {
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 first = broadcaster.subscribe(routeStream(connection), [ 'a' ]);
const second = broadcaster.subscribe(routeStream(connection), [ 'a', 'b' ]);
await broadcaster.unsubscribe(routeStream(connection), ["a"]);
await broadcaster.unsubscribe(routeStream(connection), [ 'a' ]);
await expect(first).resolves.toBeUndefined();
await expectPending(second);
await broadcaster.unsubscribe(routeStream(connection), ["b"]);
await broadcaster.unsubscribe(routeStream(connection), [ 'b' ]);
await expect(second).resolves.toBeUndefined();
});
it("resolves every pending subscription and removes topics on close", async () => {
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"]);
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 });
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 () => {
it('immediately resolves registration against an already-closed connection', async (): Promise<void> => {
const broadcaster = createBroadcaster();
const connection = new TestConnection(true, false);
connection.close();
await expect(
broadcaster.subscribe(routeStream(connection), ["items"]),
).resolves.toBeUndefined();
await expect(broadcaster.subscribe(routeStream(connection), [ 'items' ])).resolves.toBeUndefined();
});
it("rejects subscriptions on a non-streaming connection", () => {
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 }),
);
expect(() => broadcaster.subscribe(routeStream(connection), [ 'items' ])).toThrowError(expect.objectContaining({ statusCode: 406 }));
});
it("treats an empty subscription and repeated unsubscribe as no-ops", async () => {
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 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 () => {
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);
@@ -141,23 +127,20 @@ describe("Broadcaster subscriptions", () => {
markSecondSent = resolve;
});
first.send = async (message) => {
first.send = async (message): Promise<void> => {
await firstReleased;
await originalFirstSend(message);
};
second.send = async (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",
const firstSubscription = broadcaster.subscribe(routeStream(first), [ 'items' ]);
const secondSubscription = broadcaster.subscribe(routeStream(second), [ 'items' ]);
const publication = broadcaster.publish('items', {
type: 'item-changed',
data: {},
});
@@ -170,19 +153,17 @@ describe("Broadcaster subscriptions", () => {
first.close();
second.close();
await Promise.all([firstSubscription, secondSubscription]);
await Promise.all([ firstSubscription, secondSubscription ]);
});
it("closes and removes a connection whose event delivery fails", async () => {
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",
]);
connection.send = vi.fn().mockRejectedValue(new Error('socket failed'));
const subscribed = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
await broadcaster.publish("items", {
type: "item-changed",
await broadcaster.publish('items', {
type: 'item-changed',
data: {},
});
await subscribed;
@@ -190,8 +171,8 @@ describe("Broadcaster subscriptions", () => {
expect(connection.closed).toBe(true);
expect(connection.send).toHaveBeenCalledOnce();
await broadcaster.publish("items", {
type: "item-changed",
await broadcaster.publish('items', {
type: 'item-changed',
data: {},
});
expect(connection.send).toHaveBeenCalledOnce();
test/services/router.test.ts
+40 -63
View File
@@ -1,47 +1,37 @@
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 {
const moduleWith = (routes: RouteDefinition[]): RouteModule => {
return {
async getRoutes() {
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 () => {
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) => {
url: '/echo',
handler: async (stream): Promise<void> => {
expect(stream.connection).toBe(connection);
await stream.send(stream.body);
},
@@ -49,32 +39,27 @@ describe("ApplicationRouter dispatch", () => {
]),
]);
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",
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 () => {
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) => {
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 });
@@ -84,51 +69,43 @@ describe("ApplicationRouter dispatch", () => {
]);
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")?.();
completions.get('B')?.();
await second;
completions.get("A")?.();
completions.get('A')?.();
await first;
expect(connection.messages).toEqual([
{
id: "B",
type: "response",
id: 'B',
type: 'response',
statusCode: 200,
body: { key: "B" },
body: { key: 'B' },
},
{
id: "A",
type: "response",
id: 'A',
type: 'response',
statusCode: 200,
body: { key: "A" },
body: { key: 'A' },
},
]);
});
it("propagates route failures without infrastructure-specific cleanup", async () => {
const error = new Error("route failed");
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: () => {
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);
});
});
test/services/stream/stream.test.ts
+16 -20
View File
@@ -1,34 +1,32 @@
import { describe, expect, it, vi } from "vitest";
import { describe, expect, it, vi } from 'vitest';
import { HonoSSEStream } from "../../../source/services/stream/hono-sse-stream.js";
import { HttpRequestStream } from "../../../source/services/stream/http-request-stream.js";
import { WSStream } from "../../../source/services/stream/ws-stream.js";
import { HonoSSEStream } from '../../../source/services/stream/hono-sse-stream.ts';
import { HttpRequestStream } from '../../../source/services/stream/http-request-stream.ts';
import { WSStream } from '../../../source/services/stream/ws-stream.ts';
describe("stream lifecycle observers", () => {
it("buffers exactly one normal HTTP response", async () => {
describe('stream lifecycle observers', (): void => {
it('buffers exactly one normal HTTP response', async (): Promise<void> => {
const stream = new HttpRequestStream();
await stream.send({
type: "response",
type: 'response',
statusCode: 200,
body: { ok: true },
});
expect(stream.getResponse()).toEqual({
type: "response",
type: 'response',
statusCode: 200,
body: { ok: true },
});
await expect(
stream.send({
type: "response",
await expect(stream.send({
type: 'response',
statusCode: 200,
body: { second: true },
}),
).rejects.toThrow("only send one response");
})).rejects.toThrow('only send one response');
});
it("notifies WebSocket observers registered after remote closure", () => {
it('notifies WebSocket observers registered after remote closure', (): void => {
const stream = new WSStream({
send: vi.fn(),
close: vi.fn(),
@@ -42,19 +40,17 @@ describe("stream lifecycle observers", () => {
expect(onClose).toHaveBeenCalledOnce();
});
it("notifies SSE observers registered after local closure", () => {
it('notifies SSE observers registered after local closure', async (): Promise<void> => {
const streamApi = {
writeSSE: vi.fn(),
close: vi.fn(),
};
const stream = new HonoSSEStream(
streamApi as unknown as ConstructorParameters<typeof HonoSSEStream>[0],
);
const stream = new HonoSSEStream(streamApi as unknown as ConstructorParameters<typeof HonoSSEStream>[0]);
const onClose = vi.fn();
stream.close();
await stream.close();
stream.onClose(onClose);
stream.close();
await stream.close();
expect(streamApi.close).toHaveBeenCalledOnce();
expect(onClose).toHaveBeenCalledOnce();
test/services/transports/http-transport.test.ts
+90 -95
View File
@@ -1,27 +1,26 @@
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(
const createApp = async (
routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]),
maxRequestBodyBytes = 1024 * 1024,
): Promise<Hono<AppEnv>> {
const debug = new Logger("http-transport-test");
): Promise<Hono<AppEnv>> => {
const debug = new Logger('http-transport-test');
const broadcaster = new Broadcaster(debug);
const resolvedRoutes =
typeof routes === "function" ? routes(broadcaster) : routes;
const resolvedRoutes = typeof routes === 'function' ? routes(broadcaster) : routes;
const router = await ApplicationRouter.create([
{
async getRoutes() {
async getRoutes(): Promise<RouteDefinition[]> {
return resolvedRoutes;
},
},
@@ -30,25 +29,26 @@ async function createApp(
const app = new Hono<AppEnv>();
app.onError(HttpTransportRouter.createErrorHandler(debug));
app.use("*", ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug));
app.use("*", HttpTransportRouter.createExtJsonMiddleware(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 () => {
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) => stream.send(stream.body),
url: '/echo',
handler: async (stream): Promise<void> => stream.send(stream.body),
},
]);
const value = new Uint8Array([1, 2, 3]);
const value = new Uint8Array([ 1, 2, 3 ]);
const response = await app.request("/echo", {
method: "POST",
headers: { "content-type": "application/json" },
const response = await app.request('/echo', {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: toExtendedJson({ value }),
});
@@ -56,102 +56,100 @@ describe("HttpTransportRouter", () => {
expect(fromExtendedJson(await response.text())).toEqual({ value });
});
it("returns 204 when a normal HTTP route sends nothing", async () => {
it('returns 204 when a normal HTTP route sends nothing', async (): Promise<void> => {
const app = await createApp([
{
url: "/nothing",
handler: () => undefined,
url: '/nothing',
handler: (): void => undefined,
},
]);
const response = await app.request("/nothing", { method: "POST" });
const response = await app.request('/nothing', { method: 'POST' });
expect(response.status).toBe(204);
expect(await response.text()).toBe("");
expect(await response.text()).toBe('');
});
it("returns normalized errors for non-streaming requests", async () => {
it('returns normalized errors for non-streaming requests', async (): Promise<void> => {
const app = await createApp([]);
const missing = await app.request("/missing", { method: "POST" });
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",
error: 'No route found for /missing',
});
const invalid = await app.request("/missing", {
method: "POST",
headers: { "content-type": "application/json" },
body: "{",
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",
error: 'Invalid JSON in request body',
});
});
it("rejects subscribe when normal HTTP has no streaming capability", async () => {
it('rejects subscribe when normal HTTP has no streaming capability', async (): Promise<void> => {
const app = await createApp((broadcaster) => [
{
url: "/items/subscribe",
handler: async (stream) => {
await broadcaster.subscribe(stream, ["items"]);
url: '/items/subscribe',
handler: async (stream): Promise<void> => {
await broadcaster.subscribe(stream, [ 'items' ]);
},
},
]);
const response = await app.request("/items/subscribe", { method: "POST" });
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 () => {
it('sends SSE route errors as events and closes only that stream', async (): Promise<void> => {
const app = await createApp([
{
url: "/items/subscribe",
handler: () => {
throw new Error("private storage failure");
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 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");
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 () => {
it('sends a normal route as one SSE response event and then closes', async (): Promise<void> => {
const app = await createApp([
{
url: "/echo",
handler: (stream) => stream.send({ ok: true }),
url: '/echo',
handler: (stream): Promise<void> => stream.send({ ok: true }),
},
]);
const response = await app.request("/echo", {
method: "POST",
headers: { accept: "text/event-stream" },
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('event: response');
expect(events).toContain('data: {"ok":true}');
});
it("keeps SSE open until the route's subscription promise resolves", async () => {
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) => {
@@ -159,10 +157,10 @@ describe("HttpTransportRouter", () => {
});
const app = await createApp((broadcaster) => [
{
url: "/items/subscribe",
handler: async (stream) => {
const topics = ["items"];
removeSubscription = () => broadcaster.unsubscribe(stream, topics);
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();
@@ -171,9 +169,9 @@ describe("HttpTransportRouter", () => {
},
]);
const response = await app.request("/items/subscribe", {
method: "POST",
headers: { accept: "text/event-stream" },
const response = await app.request('/items/subscribe', {
method: 'POST',
headers: { accept: 'text/event-stream' },
});
const body = response.text();
const completed = vi.fn();
@@ -185,58 +183,55 @@ describe("HttpTransportRouter", () => {
await removeSubscription();
expect(await body).toBe("");
expect(await body).toBe('');
expect(completed).toHaveBeenCalledOnce();
});
it("rejects unsubscribe over non-bidirectional SSE", async () => {
it('rejects unsubscribe over non-bidirectional SSE', async (): Promise<void> => {
const app = await createApp((broadcaster) => [
{
url: "/items/unsubscribe",
handler: async (stream) => {
url: '/items/unsubscribe',
handler: async (stream): Promise<void> => {
if (!stream.bidirectional) {
throw new ApplicationError(
400,
"This route requires an existing bidirectional stream",
);
throw new ApplicationError(400, 'This route requires an existing bidirectional stream');
}
await broadcaster.unsubscribe(stream, ["items"]);
await broadcaster.unsubscribe(stream, [ 'items' ]);
},
},
]);
const response = await app.request("/items/unsubscribe", {
method: "POST",
headers: { accept: "text/event-stream" },
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('event: error');
expect(events).toContain('"statusCode":400');
});
it("rejects HTTP bodies larger than the configured byte limit", async () => {
it('rejects HTTP bodies larger than the configured byte limit', async (): Promise<void> => {
const app = await createApp(
[
{
url: "/echo",
handler: async (stream) => stream.send(stream.body),
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) }),
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",
error: 'Request body exceeds the 32 byte limit',
});
});
});
test/services/transports/ws-transport.test.ts
+27 -37
View File
@@ -1,53 +1,43 @@
import { describe, expect, it } from "vitest";
import { z } from "zod";
import { WebSocketServer } from "ws";
import { describe, expect, it } from 'vitest';
import { z } from 'zod';
import type { WebSocketServer } from 'ws';
import { ApplicationRouter } from "../../../source/services/router.js";
import {
WsTransportRouter,
} from "../../../source/services/transport/ws-transport.js";
import { Logger } from "../../../source/utils/logger.js";
import { toExtendedJson } from "@xo-cash/utils";
import { ApplicationRouter } from '../../../source/services/router.ts';
import { WsTransportRouter } from '../../../source/services/transport/ws-transport.ts';
import { Logger } from '../../../source/utils/logger.ts';
import { toExtendedJson } from '@xo-cash/utils';
describe("WebSocket request decoding", () => {
it("decodes the minimal route-agnostic envelope and Extended JSON body", async () => {
await expect(
WsTransportRouter.decodeWebSocketRequest(
toExtendedJson({
id: "request-1",
path: "/data/write",
body: { value: new Uint8Array([1, 2, 3]) },
}),
),
).resolves.toEqual({
requestId: "request-1",
path: "/data/write",
body: { value: new Uint8Array([1, 2, 3]) },
describe('WebSocket request decoding', (): void => {
it('decodes the minimal route-agnostic envelope and Extended JSON body', async (): Promise<void> => {
await expect(WsTransportRouter.decodeWebSocketRequest(toExtendedJson({
id: 'request-1',
path: '/data/write',
body: { value: new Uint8Array([ 1, 2, 3 ]) },
}))).resolves.toEqual({
requestId: 'request-1',
path: '/data/write',
body: { value: new Uint8Array([ 1, 2, 3 ]) },
});
});
it.each([
"{}",
'{"path":42}',
'{"path":"/data/get","id":1}',
'{"path":"/data/get","method":"POST"}',
])("rejects an invalid envelope: %s", async (payload) => {
await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(
z.ZodError,
it.each([ '{}', '{"path":42}', '{"path":"/data/get","id":1}', '{"path":"/data/get","method":"POST"}' ])(
'rejects an invalid envelope: %s',
async (payload) => {
await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError);
},
);
});
it("rejects malformed JSON", async () => {
await expect(WsTransportRouter.decodeWebSocketRequest("{")).rejects.toMatchObject({
it('rejects malformed JSON', async (): Promise<void> => {
await expect(WsTransportRouter.decodeWebSocketRequest('{')).rejects.toMatchObject({
statusCode: 400,
message: "Invalid JSON in WebSocket message",
message: 'Invalid JSON in WebSocket message',
});
});
});
describe("WsTransportRouter payload limits", () => {
describe('WsTransportRouter payload limits', (): void => {
it("configures Hono's ws server with the requested maxPayload", async () => {
const debug = new Logger("ws-transport-test");
const debug = new Logger('ws-transport-test');
const router = await ApplicationRouter.create([]);
const transport = new WsTransportRouter(router, debug, 1024);
const wsServer = transport.websocketServer as unknown as WebSocketServer;
test/subscription-flow.test.ts
+25 -34
View File
@@ -1,41 +1,41 @@
import { describe, expect, it, vi } from "vitest";
import { describe, expect, it, vi } from 'vitest';
import type { RouteDefinition, RouteModule } from "../source/routes/types.js";
import { ApplicationRouter } from "../source/services/router.js";
import { Broadcaster } from "../source/services/broadcaster.js";
import { Logger } from "../source/utils/logger.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 { Broadcaster } from '../source/services/broadcaster.ts';
import { Logger } from '../source/utils/logger.ts';
import { TestConnection } from './helpers/test-connection.ts';
function moduleWith(routes: RouteDefinition[]): RouteModule {
const moduleWith = (routes: RouteDefinition[]): RouteModule => {
return {
async getRoutes() {
async getRoutes(): Promise<RouteDefinition[]> {
return routes;
},
};
}
};
async function expectPending(promise: Promise<void>): Promise<void> {
const expectPending = async (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"));
describe('long-lived subscription dispatch', (): void => {
it('allows duplicate subscribe and unsubscribe requests while the original request waits', async (): Promise<void> => {
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/subscribe',
handler: async (stream): Promise<void> => {
await broadcaster.subscribe(stream, [ 'items' ]);
},
},
{
url: "/items/unsubscribe",
handler: async (stream) => {
await broadcaster.unsubscribe(stream, ["items"]);
url: '/items/unsubscribe',
handler: async (stream): Promise<void> => {
await broadcaster.unsubscribe(stream, [ 'items' ]);
await stream.send({});
},
},
@@ -43,31 +43,22 @@ describe("long-lived subscription dispatch", () => {
]);
const connection = new TestConnection(true, true);
const original = router.dispatch(
{ path: "/items/subscribe", requestId: "subscribe-1" },
connection,
);
const original = router.dispatch({ path: '/items/subscribe', requestId: 'subscribe-1' }, connection);
await vi.waitFor(() => expect(connection.closeCallbacks).toHaveLength(1));
await expectPending(original);
// This request uses a different ApplicationRouteStream over the same
// connection. Since the topic already exists, its dispatch completes.
await router.dispatch(
{ path: "/items/subscribe", requestId: "subscribe-2" },
connection,
);
await router.dispatch({ path: '/items/subscribe', requestId: 'subscribe-2' }, connection);
await expectPending(original);
await router.dispatch(
{ path: "/items/unsubscribe", requestId: "unsubscribe-1" },
connection,
);
await router.dispatch({ path: '/items/unsubscribe', requestId: 'unsubscribe-1' }, connection);
await original;
expect(connection.messages).toEqual([
{
id: "unsubscribe-1",
type: "response",
id: 'unsubscribe-1',
type: 'response',
statusCode: 200,
body: {},
},
tsconfig.json
+1 -1
View File
@@ -15,5 +15,5 @@
"declarationMap": true,
"types": ["node"]
},
"exclude": ["node_modules/**/*", "dist/**/*", "test"]
"exclude": ["node_modules/**/*", "dist/**/*"]
}