feat(infra): op call with signal (#11567)

This commit is contained in:
EYHN
2025-04-10 03:38:49 +00:00
parent bf293d8dca
commit 62b9422834
2 changed files with 24 additions and 5 deletions

View File

@@ -11,7 +11,13 @@ import {
type SubscribeMessage, type SubscribeMessage,
type UnsubscribeMessage, type UnsubscribeMessage,
} from './message'; } from './message';
import type { OpInput, OpNames, OpOutput, OpSchema } from './types'; import type {
OpInput,
OpInputWithSignal,
OpNames,
OpOutput,
OpSchema,
} from './types';
export interface CancelablePromise<T> extends Promise<T> { export interface CancelablePromise<T> extends Promise<T> {
cancel(): void; cancel(): void;
@@ -107,10 +113,14 @@ export class OpClient<Ops extends OpSchema> extends AutoMessageHandler {
call<Op extends OpNames<Ops>>( call<Op extends OpNames<Ops>>(
op: Op, op: Op,
...args: OpInput<Ops, Op> ...args: OpInputWithSignal<Ops, Op>
): CancelablePromise<OpOutput<Ops, Op>> { ): CancelablePromise<OpOutput<Ops, Op>> {
const promiseWithResolvers = Promise.withResolvers<any>(); const promiseWithResolvers = Promise.withResolvers<any>();
const payload = args[0]; const abortSignal =
args[args.length - 1] instanceof AbortSignal
? (args.pop() as AbortSignal)
: undefined;
const payload = args.pop();
const msg = { const msg = {
type: 'call', type: 'call',
@@ -121,7 +131,7 @@ export class OpClient<Ops extends OpSchema> extends AutoMessageHandler {
const promise = promiseWithResolvers.promise as CancelablePromise<any>; const promise = promiseWithResolvers.promise as CancelablePromise<any>;
const raise = (reason: string) => { const raise = (reason: any) => {
const pending = this.pendingCalls.get(msg.id); const pending = this.pendingCalls.get(msg.id);
if (!pending) { if (!pending) {
return; return;
@@ -130,11 +140,15 @@ export class OpClient<Ops extends OpSchema> extends AutoMessageHandler {
type: 'cancel', type: 'cancel',
id: msg.id, id: msg.id,
} satisfies CancelMessage); } satisfies CancelMessage);
promiseWithResolvers.reject(new Error(reason)); promiseWithResolvers.reject(reason);
clearTimeout(pending.timeout); clearTimeout(pending.timeout);
this.pendingCalls.delete(msg.id); this.pendingCalls.delete(msg.id);
}; };
abortSignal?.addEventListener('abort', () => {
raise(abortSignal.reason);
});
promise.cancel = () => { promise.cancel = () => {
raise('canceled'); raise('canceled');
}; };

View File

@@ -26,6 +26,11 @@ export type OpInput<
: never : never
: never; : never;
export type OpInputWithSignal<Ops extends OpSchema, Type extends OpNames<Ops>> =
OpInput<Ops, Type> extends [infer In]
? [In, AbortSignal | undefined] | [In]
: [AbortSignal | undefined] | [];
export type OpOutput< export type OpOutput<
Ops extends OpSchema, Ops extends OpSchema,
Type extends OpNames<Ops>, Type extends OpNames<Ops>,