106 lines
3.4 KiB
TypeScript
106 lines
3.4 KiB
TypeScript
/**
|
|
* An async iterable queue that bridges push-based producers and pull-based consumers.
|
|
*
|
|
* Composes an internal {@link ReadableStream} instead of extending it, so producers
|
|
* call {@link push} while consumers use standard async iteration (`for await...of`).
|
|
*
|
|
* ```ts
|
|
* const messages = new AsyncPushIterator<SSEvent>();
|
|
*
|
|
* // Producer (elsewhere)
|
|
* messages.push(event);
|
|
*
|
|
* // Consumer
|
|
* for await (const event of messages) {
|
|
* handle(event);
|
|
* }
|
|
* ```
|
|
*
|
|
* {@link Symbol.asyncIterator} returns `stream.values({ preventCancel: true })` so
|
|
* breaking out of `for await...of` does not cancel the underlying stream. That
|
|
* matters for long-lived sessions where the producer keeps pushing after a consumer
|
|
* stops reading early (for example, test helpers that only collect a fixed count).
|
|
*/
|
|
export class AsyncPushIterator<T> {
|
|
/** ReadableStream backing the async iterator returned from {@link Symbol.asyncIterator}. */
|
|
#stream: ReadableStream<T>;
|
|
|
|
/** Controller used to enqueue values and close the stream from {@link push} and {@link close}. */
|
|
#controller: ReadableStreamDefaultController<T> | undefined;
|
|
|
|
/** When true, no more values are accepted and iteration eventually completes. */
|
|
#closed = false;
|
|
|
|
public constructor() {
|
|
// `start`'s `this` is the underlying source object when using a plain method.
|
|
// An arrow function captures the class instance so the controller is stored here.
|
|
this.#stream = new ReadableStream({
|
|
start: (controller: ReadableStreamDefaultController<T>): void => {
|
|
this.#controller = controller;
|
|
},
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Flag indicating if the iterator is closed.
|
|
*/
|
|
public get closed(): boolean {
|
|
return this.#closed;
|
|
}
|
|
|
|
/**
|
|
* Enqueues a value for the consumer.
|
|
*
|
|
* After {@link close}, pushes are silently dropped.
|
|
*
|
|
* @param value - The next value to yield from the iterator.
|
|
*/
|
|
push(value: T): void {
|
|
if (this.#closed) return;
|
|
|
|
this.#controller?.enqueue(value);
|
|
}
|
|
|
|
/**
|
|
* Causes any future interactions with the associated stream to error with {@link error}.
|
|
* Calling this will also clear the pending values immediately, so iterators that were listening will not receive them.
|
|
*
|
|
* @param error - The error to throw from the stream.
|
|
*/
|
|
error(error: Error): void {
|
|
if (this.#closed) return;
|
|
|
|
this.#closed = true;
|
|
this.#controller?.error(error);
|
|
}
|
|
|
|
/**
|
|
* Ends the stream.
|
|
*
|
|
* Marks the iterator closed so future {@link push} calls are ignored.
|
|
* Buffered values are still yielded before iteration completes.
|
|
*/
|
|
close(): void {
|
|
this.#closed = true;
|
|
|
|
try {
|
|
this.#controller?.close();
|
|
} catch {
|
|
// The reader may already have released or cancelled the stream.
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Returns an async iterator over the composed stream.
|
|
*
|
|
* Uses `preventCancel: true` so early `break` from `for await...of` does not
|
|
* close the stream and block later pushes.
|
|
*
|
|
* Because values are discarded after being read, only a single consumer is supported.
|
|
* Additional consumers will receive a stream lock error. Unread values will be preserved until {@link close} is called.
|
|
*/
|
|
[Symbol.asyncIterator](): AsyncIterableIterator<T> {
|
|
return this.#stream.values({ preventCancel: true });
|
|
}
|
|
}
|