| 1 | import { createOperationOwner, type OperationTarget } from "./operationOwner"; |
| 2 | |
| 3 | export type PollClock = { setTimeout(callback: () => void, delay: number): unknown; clearTimeout(handle: unknown): void }; |
| 4 | type PollInput<T> = { |
| 5 | target: OperationTarget; periodMs: number; clock: PollClock; |
| 6 | read(): Promise<T>; publish(value: T): void; failed(error: unknown): void; |
| 7 | }; |
| 8 | type PollState<T> = { |
| 9 | input?: PollInput<T>; owner: ReturnType<typeof createOperationOwner>; |
| 10 | epoch: number; timer?: unknown; pending?: Promise<void>; |
| 11 | }; |
| 12 | |
| 13 | async function sample<T>(state: PollState<T>): Promise<void> { |
| 14 | if (!state.input) return; |
| 15 | const identity = state.owner.begin(state.input.target, undefined, "poll"); |
| 16 | const read = state.input.read; |
| 17 | let status: "completed" | "failed" = "completed"; |
| 18 | try { |
| 19 | const value = await read(); |
| 20 | if (!state.owner.owns(identity)) return; |
| 21 | state.input?.publish(value); |
| 22 | } catch (error) { |
| 23 | status = "failed"; |
| 24 | if (!state.owner.owns(identity)) return; |
| 25 | state.input?.failed(error); |
| 26 | } finally { state.owner.finish(identity, status); } |
| 27 | } |
| 28 | function bindRefresh<T>(state: PollState<T>): () => Promise<void> { |
| 29 | const refresh = (): Promise<void> => { |
| 30 | if (!state.input) return Promise.resolve(); |
| 31 | if (state.pending) return state.pending; |
| 32 | if (state.timer !== undefined) { state.input.clock.clearTimeout(state.timer); state.timer = undefined; } |
| 33 | const pending = sample(state).finally(() => { |
| 34 | if (state.pending !== pending) return; |
| 35 | state.pending = undefined; |
| 36 | if (state.input) state.timer = state.input.clock.setTimeout(() => { state.timer = undefined; void refresh(); }, state.input.periodMs); |
| 37 | }); |
| 38 | state.pending = pending; |
| 39 | return pending; |
| 40 | }; |
| 41 | return refresh; |
| 42 | } |
| 43 | |
| 44 | /** Single-flight polling. Disposal releases sinks and cancels queued delivery synchronously. */ |
| 45 | export function createPollingOwner<T>(input: PollInput<T>, track?: (delta: 1 | -1) => void) { |
| 46 | const owner = createOperationOwner(track); |
| 47 | const state: PollState<T> = { input, owner, epoch: owner.mount() }; |
| 48 | const refresh = bindRefresh(state); |
| 49 | return { |
| 50 | refresh, |
| 51 | dispose() { |
| 52 | if (!state.input) return; |
| 53 | if (state.timer !== undefined) state.input.clock.clearTimeout(state.timer); |
| 54 | state.timer = undefined; |
| 55 | state.input = undefined; |
| 56 | state.owner.unmount(state.epoch); |
| 57 | }, |
| 58 | }; |
| 59 | } |
| 60 |