Add tests. Add exponential backoff contexts. Update error handling. Imrove Event Parser compatibility. Simplify SSE Session. Simplify Async Iterator.

src/utils/async-push-iterator.ts

@@ -1,8 +1,8 @@
 /**
  * An async iterable queue that bridges push-based producers and pull-based consumers.
  *
- * Values are pushed from outside the iteration loop (for example, from an SSE
- * read callback) and consumed with standard async iteration:
+ * 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>();
@@ -16,92 +16,67 @@
  * }
  * ```
  *
- * When a consumer is already waiting on {@link AsyncPushIterator.prototype.next},
- * {@link push} delivers immediately. Otherwise values are buffered in FIFO order
- * until consumed. Call {@link close} to signal end-of-stream; further
- * {@link push} calls are ignored.
- *
- * Implements `Symbol.asyncDispose` so instances can be closed with `using` when
- * the runtime supports explicit resource management.
+ * {@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> implements AsyncIterable<T> {
-  /** Values pushed before a consumer was waiting to read them. */
-  private queue: T[] = [];
+export class AsyncPushIterator<T> {
+  /** ReadableStream backing the async iterator returned from {@link Symbol.asyncIterator}. */
+  private readonly stream: ReadableStream<T>;
 
-  /** Pending `next()` calls waiting for a pushed value or close. */
-  private resolvers: ((result: IteratorResult<T>) => void)[] = [];
+  /** Controller used to enqueue values and close the stream from {@link push} and {@link close}. */
+  private controller: ReadableStreamDefaultController<T> | undefined;
 
   /** When true, no more values are accepted and iteration eventually completes. */
   public 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;
+      },
+    });
+  }
+
   /**
    * Enqueues a value for the consumer.
    *
-   * If a consumer is blocked on `next()`, the value is delivered immediately and
-   * the queue is bypassed. After {@link close}, pushes are silently dropped.
+   * 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;
 
-    if (this.resolvers.length > 0) {
-      // Someone is waiting for a value, resolve immediately
-      const resolve = this.resolvers.shift()!;
-      resolve({ value, done: false });
-    } else {
-      // No one waiting, buffer the value
-      this.queue.push(value);
-    }
+    this.controller?.enqueue(value);
   }
 
   /**
    * Ends the stream.
    *
-   * Marks the iterator closed so future {@link push} calls are ignored. Any
-   * consumer currently waiting on `next()` receives `{ done: true }`. Buffered
-   * values are still yielded before iteration completes.
+   * Marks the iterator closed so future {@link push} calls are ignored.
+   * Buffered values are still yielded before iteration completes.
    */
   close(): void {
     this.closed = true;
-    for (const resolve of this.resolvers) {
-      resolve({ value: undefined as T, done: true });
+
+    try {
+      this.controller?.close();
+    } catch {
+      // The reader may already have released or cancelled the stream.
     }
-    this.resolvers = [];
   }
 
   /**
-   * Returns an async iterator that reads from this instance's shared queue.
+   * Returns an async iterator over the composed stream.
    *
-   * Buffered values are returned first, then the iterator waits for pushes or
-   * for {@link close}. Intended for a single consumer per instance.
+   * Uses `preventCancel: true` so early `break` from `for await...of` does not
+   * close the stream and block later pushes.
    */
-  [Symbol.asyncIterator](): AsyncIterator<T> {
-    return {
-      next: (): Promise<IteratorResult<T>> => {
-        // If we have buffered values, return immediately
-        if (this.queue.length > 0) {
-          return Promise.resolve({ value: this.queue.shift()!, done: false });
-        }
-
-        // If closed and no buffered values, we're done
-        if (this.closed) {
-          return Promise.resolve({ value: undefined as T, done: true });
-        }
-
-        // Wait for a value to be pushed
-        return new Promise((resolve) => {
-          this.resolvers.push(resolve);
-        });
-      },
-    };
-  }
-
-  /**
-   * Closes the iterator when used with explicit resource management (`using`).
-   */
-  [Symbol.asyncDispose](): Promise<void> {
-    this.close();
-    return Promise.resolve();
+  [Symbol.asyncIterator](): AsyncIterableIterator<T> {
+    return this.stream.values({ preventCancel: true });
   }
 }