can1357
476a689c6f
- Extracted `raceWithAbort()` utility to pi-utils package for reuse across projects. - Replaced local `raceAbort()` implementation with imported `raceWithAbort()` from pi-utils. - Refactored cleanup timer to use async iterator pattern with `timers.setInterval()` instead of callback-based `setInterval()`. - Converted `#cleanupIdleSessions()` and `#disposeSession()` from async to synchronous methods with async cleanup loop delegated to `#runCleanupLoop()`.
87 lines
2.5 KiB
TypeScript
87 lines
2.5 KiB
TypeScript
import assert from "node:assert/strict";
|
|
|
|
export class AbortError extends Error {
|
|
constructor(signal: AbortSignal) {
|
|
assert(signal.aborted, "Abort signal must be aborted");
|
|
|
|
const message = signal.reason instanceof Error ? signal.reason.message : "Cancelled";
|
|
super(`Aborted: ${message}`, { cause: signal.reason });
|
|
this.name = "AbortError";
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sleep for a given number of milliseconds, respecting abort signal.
|
|
*
|
|
* Uses setTimeout (not Bun.sleep) so that vitest fake timers can intercept it in tests.
|
|
*/
|
|
export function abortableSleep(ms: number, signal?: AbortSignal): Promise<void> {
|
|
return untilAborted(signal, () => {
|
|
const { promise, resolve } = Promise.withResolvers<void>();
|
|
setTimeout(resolve, ms);
|
|
return promise;
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Creates an abortable stream from a given stream and signal.
|
|
*
|
|
* @param stream - The stream to make abortable
|
|
* @param signal - The signal to abort the stream
|
|
* @returns The abortable stream
|
|
*/
|
|
export function createAbortableStream<T>(stream: ReadableStream<T>, signal?: AbortSignal): ReadableStream<T> {
|
|
if (!signal) return stream;
|
|
return stream.pipeThrough(new TransformStream<T, T>(), { signal });
|
|
}
|
|
|
|
/**
|
|
* Runs a promise-returning function (`pr`). If the given AbortSignal is aborted before or during
|
|
* execution, the promise is rejected with a standard error.
|
|
*
|
|
* @param signal - Optional AbortSignal to cancel the operation
|
|
* @param pr - Function returning a promise to run
|
|
* @returns Promise resolving as `pr` would, or rejecting on abort
|
|
*/
|
|
export function untilAborted<T>(
|
|
signal: AbortSignal | undefined | null,
|
|
pr: Promise<T> | (() => Promise<T>),
|
|
): Promise<T> {
|
|
if (!signal) return typeof pr === "function" ? pr() : pr;
|
|
if (signal.aborted) return Promise.reject(new AbortError(signal));
|
|
|
|
const { promise, resolve, reject } = Promise.withResolvers<T>();
|
|
const onAbort = () => reject(new AbortError(signal));
|
|
signal.addEventListener("abort", onAbort, { once: true });
|
|
|
|
void (async () => {
|
|
try {
|
|
resolve(await (typeof pr === "function" ? pr() : pr));
|
|
} catch (err) {
|
|
reject(err);
|
|
} finally {
|
|
signal.removeEventListener("abort", onAbort);
|
|
}
|
|
})();
|
|
|
|
return promise;
|
|
}
|
|
|
|
/**
|
|
* Memoizes a function with no arguments, calling it once and caching the result.
|
|
*
|
|
* @param fn - Function to be called once
|
|
* @returns A function that returns the cached result of `fn`
|
|
*/
|
|
export function once<T>(fn: () => T): () => T {
|
|
let store = undefined as { value: T } | undefined;
|
|
return () => {
|
|
if (store) {
|
|
return store.value;
|
|
}
|
|
const value = fn();
|
|
store = { value };
|
|
return value;
|
|
};
|
|
}
|