78 lines
3.0 KiB
TypeScript
78 lines
3.0 KiB
TypeScript
import { describe, expect, it, vi } from 'vitest';
|
|
|
|
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';
|
|
|
|
import { createControlledRequest } from './helpers/controlled-request.ts';
|
|
import { createMockAuth, toRoutes } from './helpers/misc.ts';
|
|
|
|
/**
|
|
* A mock of the AuthSecp256k1 service
|
|
*/
|
|
const auth = createMockAuth();
|
|
|
|
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({
|
|
auth,
|
|
},
|
|
[
|
|
toRoutes([
|
|
{
|
|
url: '/items/subscribe',
|
|
handler: async (stream): Promise<void> => {
|
|
const { signalStarted, released } = stream.body as {
|
|
signalStarted: () => void;
|
|
released: Promise<void>;
|
|
};
|
|
|
|
signalStarted();
|
|
|
|
await broadcaster.subscribe(stream, [ 'items' ]);
|
|
await released;
|
|
},
|
|
},
|
|
{
|
|
url: '/items/unsubscribe',
|
|
handler: async (stream): Promise<void> => {
|
|
await broadcaster.unsubscribe(stream, [ 'items' ]);
|
|
await stream.send({});
|
|
},
|
|
},
|
|
]),
|
|
]);
|
|
const connection = new TestConnection(true, true);
|
|
|
|
const original = createControlledRequest({ router, connection, path: '/items/subscribe', requestId: 'subscribe-1' });
|
|
await vi.waitFor(() => expect(connection.closeCallbacks).toHaveLength(1));
|
|
await original.started;
|
|
|
|
// This request uses a different ApplicationRouteStream over the same
|
|
// connection. Since the topic already exists, its dispatch completes.
|
|
const second = createControlledRequest({ router, connection, path: '/items/subscribe', requestId: 'subscribe-2' });
|
|
second.release();
|
|
await second.request;
|
|
|
|
// Unsubscribe the original request stream
|
|
const third = createControlledRequest({ router, connection, path: '/items/unsubscribe', requestId: 'unsubscribe-1' });
|
|
third.release();
|
|
await third.request;
|
|
|
|
// Release the original request stream
|
|
original.release();
|
|
await original.request;
|
|
|
|
expect(connection.messages).toEqual([
|
|
{
|
|
id: 'unsubscribe-1',
|
|
type: 'response',
|
|
statusCode: 200,
|
|
body: {},
|
|
},
|
|
]);
|
|
});
|
|
});
|