Composed push iterator

source/sse-session/async-push-iterator.ts

@@ -0,0 +1,105 @@
+/**
+ * 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 });
+    }
+}