Codex
64 lines
2.7 KiB
TypeScript
64 lines
2.7 KiB
TypeScript
export const DEFAULT_READ_ONLY_LONG_POLL_TIMEOUT_MS = 120_000;
|
|
export const READ_ONLY_LONG_POLL_TRANSPORT_MARGIN_MS = 15_000;
|
|
|
|
export interface ReadOnlyLongPollClock {
|
|
now(): number;
|
|
}
|
|
|
|
export interface ReadOnlyLongPollResult<T> {
|
|
value: T;
|
|
completed: boolean;
|
|
timedOut: boolean;
|
|
cancelled: boolean;
|
|
elapsedMs: number;
|
|
observations: number;
|
|
cursor: string;
|
|
}
|
|
|
|
export function parseStrictDuration(value: string, label = "duration"): number {
|
|
const match = /^(\d+)(ms|s|m)$/u.exec(value);
|
|
if (match === null) throw new Error(`${label} must match <integer>ms|s|m`);
|
|
const amount = Number.parseInt(match[1]!, 10);
|
|
if (!Number.isSafeInteger(amount) || amount <= 0) throw new Error(`${label} must be greater than zero`);
|
|
const multiplier = match[2] === "ms" ? 1 : match[2] === "s" ? 1_000 : 60_000;
|
|
const durationMs = amount * multiplier;
|
|
if (!Number.isSafeInteger(durationMs) || durationMs > 3_600_000) throw new Error(`${label} must not exceed 60m`);
|
|
return durationMs;
|
|
}
|
|
|
|
export function readOnlyLongPollTransportTimeoutMs(timeoutMs: number): number {
|
|
if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0 || timeoutMs > 3_600_000) throw new Error("long-poll timeoutMs must be between 1ms and 60m");
|
|
return timeoutMs + READ_ONLY_LONG_POLL_TRANSPORT_MARGIN_MS;
|
|
}
|
|
|
|
export async function runReadOnlyLongPoll<T>(options: {
|
|
observe: () => Promise<T>;
|
|
completed: (value: T) => boolean;
|
|
waitForChange: (remainingMs: number, signal?: AbortSignal) => Promise<void>;
|
|
cursor: (value: T) => string;
|
|
timeoutMs?: number;
|
|
clock?: ReadOnlyLongPollClock;
|
|
signal?: AbortSignal;
|
|
}): Promise<ReadOnlyLongPollResult<T>> {
|
|
const timeoutMs = options.timeoutMs ?? DEFAULT_READ_ONLY_LONG_POLL_TIMEOUT_MS;
|
|
if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) throw new Error("timeoutMs must be a positive integer");
|
|
const clock = options.clock ?? { now: () => Date.now() };
|
|
const startedAt = clock.now();
|
|
let observations = 0;
|
|
let value = await options.observe();
|
|
observations += 1;
|
|
while (!options.completed(value)) {
|
|
if (options.signal?.aborted === true) {
|
|
return { value, completed: false, timedOut: false, cancelled: true, elapsedMs: clock.now() - startedAt, observations, cursor: options.cursor(value) };
|
|
}
|
|
const remainingMs = timeoutMs - (clock.now() - startedAt);
|
|
if (remainingMs <= 0) {
|
|
return { value, completed: false, timedOut: true, cancelled: false, elapsedMs: clock.now() - startedAt, observations, cursor: options.cursor(value) };
|
|
}
|
|
await options.waitForChange(remainingMs, options.signal);
|
|
value = await options.observe();
|
|
observations += 1;
|
|
}
|
|
return { value, completed: true, timedOut: false, cancelled: false, elapsedMs: clock.now() - startedAt, observations, cursor: options.cursor(value) };
|
|
}
|