/**
 * 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}. */
    private readonly stream: ReadableStream<T>;

    /** 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.
     *
     * 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);
    }

    /**
     * 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.
     */
    [Symbol.asyncIterator](): AsyncIterableIterator<T> {
        return this.stream.values({ preventCancel: true });
    }
}