From c93ed7c09cb82db553eb1716a15dd86eda98eec8 Mon Sep 17 00:00:00 2001 From: Harvey Zuccon Date: Fri, 24 Jul 2026 04:51:58 +0000 Subject: [PATCH] Composed push iterator --- package-lock.json | 60 +++--- source/sse-session/async-push-iterator.ts | 105 +++++++++ source/sse-session/index.ts | 1 + test/sse-session/async-push-iterator.test.ts | 214 +++++++++++++++++++ 4 files changed, 350 insertions(+), 30 deletions(-) create mode 100644 source/sse-session/async-push-iterator.ts create mode 100644 test/sse-session/async-push-iterator.test.ts diff --git a/package-lock.json b/package-lock.json index 2d1a9ee..813ede6 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1105,9 +1105,9 @@ "license": "MIT" }, "node_modules/@eslint/config-array/node_modules/brace-expansion": { - "version": "1.1.15", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz", - "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==", + "version": "1.1.16", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz", + "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==", "dev": true, "license": "MIT", "dependencies": { @@ -1186,9 +1186,9 @@ "license": "MIT" }, "node_modules/@eslint/eslintrc/node_modules/brace-expansion": { - "version": "1.1.15", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz", - "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==", + "version": "1.1.16", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz", + "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==", "dev": true, "license": "MIT", "dependencies": { @@ -3435,9 +3435,9 @@ } }, "node_modules/body-parser": { - "version": "1.20.5", - "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-1.20.5.tgz", - "integrity": "sha512-3grm+/2tUOvu2cjJkvsIxrv/wVpfXQW4PsQHYm7yk4vfpu7Ekl6nEsYBoJUL6qDwZUx8wUhQ8tR2qz+ad9c9OA==", + "version": "1.20.6", + "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-1.20.6.tgz", + "integrity": "sha512-p5tAzS57i5MV9fZFDj9LeIiTZEufbSe2eDozP+ElheSUq1m74CRq1jI4mYNDdVs9vQztXFLuk/Gd6BWTdwRJ5g==", "dev": true, "license": "MIT", "dependencies": { @@ -3503,9 +3503,9 @@ } }, "node_modules/brace-expansion": { - "version": "5.0.6", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.6.tgz", - "integrity": "sha512-kLpxurY4Z4r9sgMsyG0Z9uzsBlgiU/EFKhj/h91/8yHu0edo7XuixOIH3VcJ8kkxs6/jPzoI6U9Vj3WqbMQ94g==", + "version": "5.0.7", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.7.tgz", + "integrity": "sha512-7oFy703dxfY3/NLxC1fh2SUCQ0H9rmAY+5EpDVfXjUTTs+HEwR2nYaqLv+GWcTsumwxPfiz6CzCNkwXwBUwqCA==", "dev": true, "license": "MIT", "dependencies": { @@ -5292,9 +5292,9 @@ "peer": true }, "node_modules/eslint-plugin-import/node_modules/brace-expansion": { - "version": "1.1.15", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz", - "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==", + "version": "1.1.16", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz", + "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==", "dev": true, "license": "MIT", "peer": true, @@ -5379,9 +5379,9 @@ "peer": true }, "node_modules/eslint-plugin-jsx-a11y/node_modules/brace-expansion": { - "version": "1.1.15", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz", - "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==", + "version": "1.1.16", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz", + "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==", "dev": true, "license": "MIT", "peer": true, @@ -5468,9 +5468,9 @@ "peer": true }, "node_modules/eslint-plugin-react/node_modules/brace-expansion": { - "version": "1.1.15", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz", - "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==", + "version": "1.1.16", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz", + "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==", "dev": true, "license": "MIT", "peer": true, @@ -5542,9 +5542,9 @@ "license": "MIT" }, "node_modules/eslint/node_modules/brace-expansion": { - "version": "1.1.15", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.15.tgz", - "integrity": "sha512-EwOCDEex4quD37XhqM3omwtMoJjr//isUZz1JopUNWms+4Z2ViyM/k1YIRePpoVNnQhENnxtFjLaxNHrT7xIUg==", + "version": "1.1.16", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.16.tgz", + "integrity": "sha512-IDw48K2/2kRkg9LdJxurvq3lV3aBgq0REY89duEqFRthjlPdXHKMj7EnQOXVckxzgisinf3nHfrcE2FufFLXMw==", "dev": true, "license": "MIT", "dependencies": { @@ -7129,9 +7129,9 @@ "license": "MIT" }, "node_modules/js-yaml": { - "version": "4.2.0", - "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-4.2.0.tgz", - "integrity": "sha512-ePWsvanv0DWuDRsW8dnt+R4jQ31SCRCQ7hhNcPXZPsoBZiemuZNYGf7adZdqX2D86j6rvKp3RpCxVTSb8WQlOw==", + "version": "4.3.0", + "resolved": "https://registry.npmjs.org/js-yaml/-/js-yaml-4.3.0.tgz", + "integrity": "sha512-1td788aAnnZ5qs7V2QIRl1owjtYpbKt749Y3xauqQgwIIGF/xXWz1wMTEBx5O3LK3lXLVuqXPdPxj2BoFHaW9Q==", "dev": true, "funding": [ { @@ -7524,9 +7524,9 @@ } }, "node_modules/linkify-it": { - "version": "5.0.1", - "resolved": "https://registry.npmjs.org/linkify-it/-/linkify-it-5.0.1.tgz", - "integrity": "sha512-wVoTjP4Q6R0NW5hiZkVJaFZPWgtXfoGF+6LucL3/FtiNjmcHhYjEr5f1Kqjirc1nBW07J/ZuRFumqr2oqccEWg==", + "version": "5.0.2", + "resolved": "https://registry.npmjs.org/linkify-it/-/linkify-it-5.0.2.tgz", + "integrity": "sha512-ONTm2jCMAVZjgQa/Fy1kScXsuOoF5NPTsoFBdE1KVIZ2vAh/r9+Bqo+0jINCBYnavTPQZz38QzFTme79ENoN3Q==", "dev": true, "funding": [ { diff --git a/source/sse-session/async-push-iterator.ts b/source/sse-session/async-push-iterator.ts new file mode 100644 index 0000000..949b3d9 --- /dev/null +++ b/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(); + * + * // 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}. */ + #stream: ReadableStream; + + /** Controller used to enqueue values and close the stream from {@link push} and {@link close}. */ + #controller: ReadableStreamDefaultController | 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): 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 { + return this.#stream.values({ preventCancel: true }); + } +} diff --git a/source/sse-session/index.ts b/source/sse-session/index.ts index 563fb3e..48842e2 100644 --- a/source/sse-session/index.ts +++ b/source/sse-session/index.ts @@ -1 +1,2 @@ +export * from './async-push-iterator.ts'; export * from './sse-event-parser.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..82a1940 --- /dev/null +++ b/test/sse-session/async-push-iterator.test.ts @@ -0,0 +1,214 @@ +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 = (): Promise => 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 = async (): Promise => 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 = async (): Promise => 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 => collectAll(iterator); + + const failedIterator = async (): Promise => { + try { + /* eslint-disable-next-line */ + for await (const _value of iterator) { + } + } catch (error) { + failureFlag(); + } + }; + + const promises = [ successfulIterator(), failedIterator() ]; + + 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 ]); +}; + +/** + * Tests that the iterator rejects after {@link AsyncPushIterator.error} is called. + */ +const testPushIteratorRejectsAfterError = async (): Promise => { + const iterator = new AsyncPushIterator(); + + iterator.error(new Error('Stream has been closed for a test')); + + await expect(collectAll(iterator)).rejects.toThrow('Stream has been closed for a test'); +}; + +/** + * Tests that the iterator closes the stream when error() is called. + */ +const testPushIteratorClosesWhenErrorIsCalled = async (): Promise => { + const iterator = new AsyncPushIterator(); + + iterator.error(new Error('Stream has been closed for a test')); + expect(iterator.closed).toBe(true); +}; + +/** + * Tests that subsequent calls to error() are ignored. + */ +const testPushIteratorIgnoresSubsequentErrorCalls = async (): Promise => { + const iterator = new AsyncPushIterator(); + + iterator.error(new Error('Stream has been closed for a test')); + expect(iterator.closed).toBe(true); + + expect(iterator.error(new Error('Second error'))).toBe(undefined); +}; + +/** + * Tests that a consumer can check if the iterator is closed. + */ +const testPushIteratorCanCheckIfClosed = async (): Promise => { + const iterator = new AsyncPushIterator(); + + expect(iterator.closed).toBe(false); + + iterator.close(); + expect(iterator.closed).toBe(true); +}; + +const runTests = async (): Promise => { + test('AsyncPushIterator: pushes and consumes values', testPushComposedPushAndConsume); + test('AsyncPushIterator: buffers values pushed before the for-await loop starts', testPushComposedBuffersValuesPushedBeforeLoopStarts); + test('AsyncPushIterator: resolves with no values when nothing was pushed', testPushComposedResolvesWithNoValues); + test('AsyncPushIterator: ignores values pushed after close', testPushComposedIgnoresValuesAfterClose); + test('AsyncPushIterator: rejects multiple consumers', testPushComposedRejectsMultipleConsumers); + test('AsyncPushIterator: resolves immediately when closed before the loop starts', testPushComposedResolvesWhenClosedBeforeLoop); + test('AsyncPushIterator: keeps the stream open after an early break', testPushComposedAllowsPushingAfterEarlyBreak); + test('AsyncPushIterator: rejects after error', testPushIteratorRejectsAfterError); + test('AsyncPushIterator: closes the stream when error() is called', testPushIteratorClosesWhenErrorIsCalled); + test('AsyncPushIterator: ignores subsequent error() calls', testPushIteratorIgnoresSubsequentErrorCalls); + test('AsyncPushIterator: can check if closed', testPushIteratorCanCheckIfClosed); +}; + +await runTests();