import {ErrorAborted, TimeoutError} from "./errors.js"; import {sleep} from "./sleep.js"; import {ArrayToTuple, NonEmptyArray} from "./types.js"; /** * While promise t is not finished, call function `fn` per `interval` */ export async function callFnWhenAwait( p: Promise, fn: () => void, interval: number, signal?: AbortSignal ): Promise { let done = false; const logFn = async (): Promise => { while (!done) { await sleep(interval, signal); if (!done) fn(); } }; const t = await Promise.race([p, logFn()]).finally(() => { done = true; }); return t as T; } /** * Create a deferred promise */ export function defer() { let resolve!: (v: T) => void; let reject!: (e: unknown) => void; const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); return {promise, resolve, reject}; } export type PromiseResult = { promise: Promise; } & ( | { status: "pending"; } | { status: "fulfilled"; value: T; durationMs: number; } | { status: "rejected"; reason: Error; durationMs: number; } ); export type PromiseFulfilledResult = PromiseResult & {status: "fulfilled"}; export type PromiseRejectedResult = PromiseResult & {status: "rejected"}; /** * Wrap a promise to an object to track the status and value of the promise */ export function wrapPromise(promise: PromiseLike): PromiseResult { const startedAt = Date.now(); const result = { promise: promise.then( (value) => { result.status = "fulfilled"; (result as PromiseFulfilledResult).value = value; (result as PromiseFulfilledResult).durationMs = Date.now() - startedAt; return value; }, (reason: unknown) => { result.status = "rejected"; (result as PromiseRejectedResult).reason = reason as Error; (result as PromiseRejectedResult).durationMs = Date.now() - startedAt; throw reason; } ), status: "pending", } as PromiseResult; return result; } type ReturnPromiseWithTuple>> = { [Index in keyof ArrayToTuple]: PromiseResult>; }; /** * Two phased approach for resolving promises: * - first wait `resolveTimeoutMs` or until all promises settle * - then wait `raceTimeoutMs - resolveTimeoutMs` or until at least a single promise resolves * * Returns a list of promise results, see `PromiseResult` */ export async function resolveOrRacePromises>>( promises: T, { resolveTimeoutMs, raceTimeoutMs, signal, }: { resolveTimeoutMs: number; raceTimeoutMs: number; signal?: AbortSignal; } ): Promise> | never { if (raceTimeoutMs <= resolveTimeoutMs) { throw new Error("Race time must be greater than resolve time"); } const resolveTimeoutError = new TimeoutError( `Given promises can't be resolved within resolveTimeoutMs=${resolveTimeoutMs}` ); const raceTimeoutError = new TimeoutError( `Not a any single promise be resolved in given raceTimeoutMs=${raceTimeoutMs}` ); const promiseResults = promises.map((p) => wrapPromise(p)) as ReturnPromiseWithTuple; // We intentionally want an array of promises here promises = (promiseResults as PromiseResult[]).map((p) => p.promise) as unknown as T; try { await Promise.race([ Promise.allSettled(promises), sleep(resolveTimeoutMs, signal).then(() => { throw resolveTimeoutError; }), ]); return promiseResults; } catch (err) { if (err instanceof ErrorAborted) { return promiseResults; } if (err !== resolveTimeoutError) { throw err; } } try { await Promise.race([ Promise.any(promises), sleep(raceTimeoutMs - resolveTimeoutMs, signal).then(() => { throw raceTimeoutError; }), ]); return promiseResults; } catch (err) { if (err instanceof ErrorAborted) { return promiseResults; } if (err !== raceTimeoutError && !(err instanceof AggregateError)) { throw err; } } return promiseResults; }