fix(companion): reject pending event waits on teardown
This commit is contained in:
@@ -339,6 +339,30 @@ describe('CompanionRuntimeClient', () => {
|
||||
await awaited;
|
||||
});
|
||||
|
||||
it('waitForEvent rejects immediately when event subscriptions are cleared', async () => {
|
||||
const client = new CompanionRuntimeClient({
|
||||
url: 'ws://127.0.0.1:1',
|
||||
});
|
||||
|
||||
const awaited = expect(
|
||||
client.waitForEvent('agent.stream', { timeoutMs: 10_000 }),
|
||||
).rejects.toThrow('Event subscriptions cleared');
|
||||
client.clearEventSubscriptions();
|
||||
await awaited;
|
||||
});
|
||||
|
||||
it('waitForEvent rejects immediately on disconnect', async () => {
|
||||
const client = new CompanionRuntimeClient({
|
||||
url: 'ws://127.0.0.1:1',
|
||||
});
|
||||
|
||||
const awaited = expect(
|
||||
client.waitForEvent('agent.stream', { timeoutMs: 10_000 }),
|
||||
).rejects.toThrow('Disconnected');
|
||||
client.disconnect();
|
||||
await awaited;
|
||||
});
|
||||
|
||||
it('waitForAgentStream resolves on agent.stream events', async () => {
|
||||
const client = new CompanionRuntimeClient({
|
||||
url: 'ws://127.0.0.1:1',
|
||||
|
||||
@@ -274,6 +274,7 @@ export class CompanionRuntimeClient {
|
||||
private nextId = 1;
|
||||
private pending = new Map<number, PendingRequest>();
|
||||
private readonly eventHandlers = new Set<CompanionEventHandler>();
|
||||
private readonly pendingEventWaits = new Set<(error: Error) => void>();
|
||||
|
||||
constructor(options: CompanionRuntimeClientOptions) {
|
||||
const requestTimeoutMs = options.requestTimeoutMs ?? 15_000;
|
||||
@@ -354,12 +355,14 @@ export class CompanionRuntimeClient {
|
||||
|
||||
disconnect(code?: number, reason?: string): void {
|
||||
if (!this.ws) {
|
||||
this.rejectEventWaits(new Error('Disconnected'));
|
||||
return;
|
||||
}
|
||||
|
||||
const ws = this.ws;
|
||||
this.ws = null;
|
||||
this.rejectAllPending(new Error('Disconnected'));
|
||||
this.rejectEventWaits(new Error('Disconnected'));
|
||||
ws.close(code, reason);
|
||||
}
|
||||
|
||||
@@ -377,6 +380,7 @@ export class CompanionRuntimeClient {
|
||||
|
||||
clearEventSubscriptions(): void {
|
||||
this.eventHandlers.clear();
|
||||
this.rejectEventWaits(new Error('Event subscriptions cleared'));
|
||||
}
|
||||
|
||||
subscribeEvent<TData = unknown>(
|
||||
@@ -436,9 +440,15 @@ export class CompanionRuntimeClient {
|
||||
abortCleanup();
|
||||
abortCleanup = null;
|
||||
}
|
||||
this.pendingEventWaits.delete(cancelWait);
|
||||
fn();
|
||||
};
|
||||
|
||||
const cancelWait = (error: Error) => {
|
||||
finish(() => reject(error));
|
||||
};
|
||||
this.pendingEventWaits.add(cancelWait);
|
||||
|
||||
const unsubscribe = this.subscribeEvent<TData>(eventName, (data) => {
|
||||
if (predicate && !predicate(data)) {
|
||||
return;
|
||||
@@ -452,7 +462,7 @@ export class CompanionRuntimeClient {
|
||||
|
||||
if (signal) {
|
||||
const onAbort = () => {
|
||||
finish(() => reject(new Error(`Aborted while waiting for event ${eventName}`)));
|
||||
cancelWait(new Error(`Aborted while waiting for event ${eventName}`));
|
||||
};
|
||||
signal.addEventListener('abort', onAbort, { once: true });
|
||||
abortCleanup = () => {
|
||||
@@ -697,6 +707,13 @@ export class CompanionRuntimeClient {
|
||||
}
|
||||
this.pending.clear();
|
||||
}
|
||||
|
||||
private rejectEventWaits(error: Error): void {
|
||||
for (const cancel of this.pendingEventWaits) {
|
||||
cancel(error);
|
||||
}
|
||||
this.pendingEventWaits.clear();
|
||||
}
|
||||
}
|
||||
|
||||
function withToken(url: string, token?: string): string {
|
||||
|
||||
Reference in New Issue
Block a user