diff --git a/source/sse-session/async-push-iterator.ts b/source/sse-session/async-push-iterator.ts new file mode 100644 index 0000000..a09a60f --- /dev/null +++ b/source/sse-session/async-push-iterator.ts @@ -0,0 +1,82 @@ +/** + * 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(); + * + * // 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 { + /** ReadableStream backing the async iterator returned from {@link Symbol.asyncIterator}. */ + private readonly stream: ReadableStream; + + /** Controller used to enqueue values and close the stream from {@link push} and {@link close}. */ + private controller: ReadableStreamDefaultController | 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) => { + 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]() { + return this.stream.values({ preventCancel: true }); + } +} diff --git a/source/sse-session/index.ts b/source/sse-session/index.ts new file mode 100644 index 0000000..d5b61f5 --- /dev/null +++ b/source/sse-session/index.ts @@ -0,0 +1 @@ +export * from './async-push-iterator.ts'; diff --git a/test/sse-session/async-push-iterator.test.ts b/test/sse-session/async-push-iterator.test.ts new file mode 100644 index 0000000..ac5c36c --- /dev/null +++ b/test/sse-session/async-push-iterator.test.ts @@ -0,0 +1,188 @@ +import { expect, test, vi } from 'vitest'; + +import { AsyncPushIterator } from '../../source/sse-session/async-push-iterator.ts'; + +/** + * Collects every value from the iterator into an array. + * + * @param iterator - Iterator under test. + */ +const collectAll = async (iterator: AsyncPushIterator): Promise => { + const results: T[] = []; + + for await (const value of iterator) { + results.push(value); + } + + return results; +}; + +/** + * Tests that values pushed while a consumer is already waiting are delivered in order. + */ +const testPushComposedPushAndConsume = async (): Promise => { + const iterator = new AsyncPushIterator(); + + const result = new Promise((resolve) => { + void (async () => { + resolve(await collectAll(iterator)); + })(); + }); + + iterator.push(1); + iterator.push(2); + iterator.push(3); + iterator.close(); + + await expect(result).resolves.toEqual([ 1, 2, 3 ]); +}; + +/** + * Tests that values pushed before `for await...of` starts are buffered and yielded + * once the consumer begins reading. + */ +const testPushComposedBuffersValuesPushedBeforeLoopStarts = async (): Promise => { + const iterator = new AsyncPushIterator(); + + iterator.push(1); + iterator.push(2); + iterator.push(3); + + const result = collectAll(iterator); + + iterator.close(); + + await expect(result).resolves.toEqual([ 1, 2, 3 ]); +}; + +/** + * Tests that the iterator completes with no values when nothing was pushed. + */ +const testPushComposedResolvesWithNoValues = async (): Promise => { + const iterator = new AsyncPushIterator(); + + const result = new Promise((resolve) => { + void (async () => { + resolve(await collectAll(iterator)); + })(); + }); + + iterator.close(); + + await expect(result).resolves.toEqual([]); +}; + +/** + * Tests that values pushed after {@link AsyncPushIterator.close} are ignored. + */ +const testPushComposedIgnoresValuesAfterClose = async (): Promise => { + const iterator = new AsyncPushIterator(); + + const result = new Promise((resolve) => { + void (async () => { + resolve(await collectAll(iterator)); + })(); + }); + + iterator.push(1); + iterator.push(2); + iterator.push(3); + iterator.close(); + iterator.push(4); + + await expect(result).resolves.toEqual([ 1, 2, 3 ]); +}; + +/** + * Tests that only one async consumer can read from the composed ReadableStream at a time. + * + * Unlike the hand-rolled async-push-iterator, the second consumer fails with a + * stream lock error rather than TooManyAsyncIteratorsError. + */ +const testPushComposedRejectsMultipleConsumers = async (): Promise => { + const iterator = new AsyncPushIterator(); + + const failureFlag = vi.fn(); + + const successfulIterator = (): Promise => + new Promise((resolve) => { + void (async () => { + resolve(await collectAll(iterator)); + })(); + }); + + const failedIterator = (): Promise => + new Promise((resolve, reject) => { + void (async () => { + try { + /* eslint-disable-next-line */ + for await (const _value of iterator) { + } + } catch (error) { + failureFlag(); + reject(error); + } + + resolve(); + })(); + }); + + const promises = [ successfulIterator(), failedIterator().catch(() => {}) ]; + + iterator.close(); + + await Promise.all(promises); + + expect(failureFlag).toHaveBeenCalledOnce(); +}; + +/** + * Tests that closing before iteration starts lets the loop finish immediately. + */ +const testPushComposedResolvesWhenClosedBeforeLoop = async (): Promise => { + const iterator = new AsyncPushIterator(); + + iterator.close(); + + await expect(collectAll(iterator)).resolves.toEqual([]); +}; + +/** + * Tests that breaking out of `for await...of` early does not cancel the stream. + * + * {@link AsyncPushIterator} uses `preventCancel: true` so producers can keep pushing + * and a later consumer can read the remaining values. + */ +const testPushComposedAllowsPushingAfterEarlyBreak = async (): Promise => { + const iterator = new AsyncPushIterator(); + + iterator.push(1); + + const firstPass: number[] = []; + + for await (const value of iterator) { + firstPass.push(value); + break; + } + + iterator.push(2); + iterator.push(3); + iterator.close(); + + const secondPass = await collectAll(iterator); + + expect(firstPass).toEqual([ 1 ]); + expect(secondPass).toEqual([ 2, 3 ]); +}; + +const runTests = async (): Promise => { + test('AsyncPushIterator (composed): pushes and consumes values', testPushComposedPushAndConsume); + test('AsyncPushIterator (composed): buffers values pushed before the for-await loop starts', testPushComposedBuffersValuesPushedBeforeLoopStarts); + test('AsyncPushIterator (composed): resolves with no values when nothing was pushed', testPushComposedResolvesWithNoValues); + test('AsyncPushIterator (composed): ignores values pushed after close', testPushComposedIgnoresValuesAfterClose); + test('AsyncPushIterator (composed): rejects multiple consumers', testPushComposedRejectsMultipleConsumers); + test('AsyncPushIterator (composed): resolves immediately when closed before the loop starts', testPushComposedResolvesWhenClosedBeforeLoop); + test('AsyncPushIterator (composed): keeps the stream open after an early break', testPushComposedAllowsPushingAfterEarlyBreak); +}; + +await runTests();