diff --git a/.vscode/launch.json b/.vscode/launch.json new file mode 100644 index 0000000..0c341a3 --- /dev/null +++ b/.vscode/launch.json @@ -0,0 +1,26 @@ +{ + // Use IntelliSense to learn about possible attributes. + // Hover to view descriptions of existing attributes. + // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387 + "version": "0.2.0", + "configurations": [ + { + "request": "launch", + "name": "Lauch File (Deno)", + "type": "node", + "program": "${file}", + "cwd": "${workspaceFolder}", + "env": { + "DENO_FUTURE": "1" + }, + "runtimeExecutable": "/home/gitpod/.deno/bin/deno", + "runtimeArgs": [ + "run", + "--unstable", + "--inspect-wait", + "--allow-all" + ], + "attachSimplePort": 9229 + } + ] +} \ No newline at end of file diff --git a/_assert.ts b/_assert.ts new file mode 100644 index 0000000..40ca3cb --- /dev/null +++ b/_assert.ts @@ -0,0 +1,32 @@ +import { AssertionError } from "@std/assert/assertion-error"; + +/** + * Make an assertion that `actual` and `expected` are not equal, deeply. + * If not then throw. + * + * Type parameter can be specified to ensure values under comparison have the same type. + * + * @example Usage + * ```ts ignore + * import { assertNotEquals } from "./_assert.ts"; + * + * assertNotEquals(1, 2); // Doesn't throw + * assertNotEquals(1, 1); // Throws + * ``` + * + * @typeParam T The type of the values to compare. + * @param actual The actual value to compare. + * @param expected The expected value to compare. + * @param msg The optional message to display if the assertion fails. + */ +export function assertNotEquals(actual: T, expected: T, msg?: string) { + if (actual !== expected) { + return; + } + const actualString = String(actual); + const expectedString = String(expected); + const msgSuffix = msg ? `: ${msg}` : "."; + throw new AssertionError( + `Expected actual: ${actualString} not to be: ${expectedString}${msgSuffix}`, + ); +} \ No newline at end of file diff --git a/_repl.ts b/_repl.ts new file mode 100644 index 0000000..37002bc --- /dev/null +++ b/_repl.ts @@ -0,0 +1,47 @@ +// Create a new consumer `ReadableStream` +const consumer = new ReadableStream({ + async pull(controller) { + controller.enqueue("Hello, World!"); + controller.close(); + }, + + async cancel(reason) { + console.log("Consumer canceled:", reason); + }, +}); + +const reader1 = consumer.getReader(); + +await reader1.releaseLock(); +await consumer.cancel("Consumer canceled early"); + + +const reader2 = consumer.getReader(); + +// await reader.cancel("Consumer canceled"); + +let { value, done } = await reader2.read(); +console.log({ + value, + done, +}); + + +// await reader.cancel("Consumer canceled"); +// reader.releaseLock(); + +({ value, done } = await reader2.read()); +console.log({ + value, + done, +}); + +console.log({ + reader2, + consumer, + + locked: consumer.locked, + closed: await reader2.closed, +}) + +await reader1.cancel("Done early"); \ No newline at end of file diff --git a/types.ts b/_types.ts similarity index 100% rename from types.ts rename to _types.ts diff --git a/channel.ts b/channel.ts index 1565fcc..d24fdb7 100644 --- a/channel.ts +++ b/channel.ts @@ -1,5 +1,5 @@ -import type { EnhancedReadableStream } from "./stream.ts"; -import { enhanceReadableStream } from "./stream.ts"; +import type { MulticastReadableStream } from "./multicast.ts"; +import { createMulticastStream } from "./multicast.ts"; /** * Creates a unidirectional communication channel built on Web Streams, allowing data to flow from one or more writers to multiple independent readers. @@ -136,7 +136,7 @@ export function createChannel(): Channel { const sharedWriter = transformStream.writable.getWriter(); const readableStream = transformStream.readable; - const enhancedReadableStream = enhanceReadableStream(readableStream); + const multicastStream = createMulticastStream(readableStream); return { /** @@ -150,7 +150,7 @@ export function createChannel(): Channel { * * @returns A new readable stream with disposal support. */ - readable: enhancedReadableStream, + readable: multicastStream, /** * Method to get the shared writer for direct writing. @@ -168,7 +168,7 @@ export function createChannel(): Channel { async [Symbol.asyncDispose]() { await Promise.all([ sharedWriter.close(), // Close the writable stream - enhancedReadableStream.cancel(), + multicastStream.cancel(), ]); }, }; @@ -393,7 +393,7 @@ export interface Channel { * * @returns A new readable stream with disposal support. */ - readonly readable: EnhancedReadableStream; + readonly readable: MulticastReadableStream; /** * Asynchronously disposes of the channel resources using the Symbol.asyncDispose protocol. @@ -415,7 +415,7 @@ export interface BidirectionalChannel { */ readonly endpointA: { readonly writer: WritableStreamDefaultWriter; - readonly readable: EnhancedReadableStream; + readonly readable: MulticastReadableStream; }; /** @@ -423,7 +423,7 @@ export interface BidirectionalChannel { */ readonly endpointB: { readonly writer: WritableStreamDefaultWriter; - readonly readable: EnhancedReadableStream; + readonly readable: MulticastReadableStream; }; /** diff --git a/deno.jsonc b/deno.jsonc index f6b786b..66f6501 100644 --- a/deno.jsonc +++ b/deno.jsonc @@ -4,20 +4,28 @@ "exports": { ".": "./mod.ts", "./split": "./split.ts", - "./channel": "./channel.ts", - "./events": "./events.ts", "./utils": "./utils.ts", - "./stream": "./stream.ts", - "./types": "./types.ts" + "./types": "./_types.ts", + "./events": "./events.ts", + "./channel": "./channel.ts", + "./multicast": "./multicast.ts" }, "tasks": { "test": "deno test -RW --allow-run=deno,bun,node,npx --clean --trace-leaks", - "dev": "deno task test --filter='/^\\[deno\\]/'" + "dev": "deno task test --filter='/^DENO/'" }, "license": "MIT", "imports": { "@libs/testing": "jsr:@libs/testing@^3.0.1", "@std/assert": "jsr:@std/assert@1", "@std/expect": "jsr:@std/expect@^1.0.5" + }, + "test:permissions": { + "run": [ + "deno", + "node", + "bun", + "npx" + ] } } diff --git a/events.ts b/events.ts index 3897d1f..c25d41f 100644 --- a/events.ts +++ b/events.ts @@ -5,7 +5,7 @@ * @module */ -import type { EnhancedReadableStream } from "./stream.ts"; +import type { MulticastReadableStream } from "./multicast.ts"; import { createChannel } from "./channel.ts"; /** @@ -128,7 +128,7 @@ export function createEventDispatcher>(): * * @returns A new readable stream of StatusEvents. */ - get events(): EnhancedReadableStream { + get events(): MulticastReadableStream { return channel.readable; }, @@ -287,7 +287,7 @@ export interface EventDispatcher> { * * @returns A new readable stream of StatusEvents. */ - readonly events: EnhancedReadableStream; + readonly events: MulticastReadableStream; /** * Disposes of the dispatcher asynchronously, releasing resources. diff --git a/mod.ts b/mod.ts index aefedb7..8fd17f1 100644 --- a/mod.ts +++ b/mod.ts @@ -1,6 +1,6 @@ -export * from "./stream.ts"; +export * from "./multicast.ts"; export * from "./channel.ts"; export * from "./split.ts"; export * from "./utils.ts"; -export type * from "./types.ts"; \ No newline at end of file +export type * from "./_types.ts"; \ No newline at end of file diff --git a/multicast.ts b/multicast.ts new file mode 100644 index 0000000..2bce36a --- /dev/null +++ b/multicast.ts @@ -0,0 +1,401 @@ +/** + * @module multicast + * + * This module enhances the native `ReadableStream` by enabling multiple consumers to + * read from a single source stream without duplicating the data. Normally, `ReadableStream` allows + * only one reader at a time, and once a stream is locked by a reader, it can't be read by others + * simultaneously. This module solves that limitation by allowing multiple consumers (readers) + * to access the stream's data without conflicts. + * + * ### Core Idea + * The main goal of this module is to allow multiple independent consumers to access and read + * from the same stream without competing for stream locks. This is achieved by managing multiple + * readers and stream controllers through various internal maps and weak maps. + * + * Each time a consumer reads data, the stream distributes the data chunk to all registered + * consumers. The system maintains a registry of consumers and controllers, ensuring that + * when one consumer pulls data, the same data is made available to all consumers. + * + * ### Why is this necessary? + * In many scenarios, different subsystems or components need to process the same stream of data. + * For example: + * - In distributed systems, multiple logging systems may need to capture the same data stream for + * monitoring purposes. + * - In data pipelines, multiple workers may need to consume the same stream of data for parallel + * processing. + * + * However, the native `ReadableStream` only allows a single reader at a time, creating the need for + * a system to manage multiple consumers that can read the same data stream independently. + * + * ### Key considerations + * - **Stream Locking**: Only one reader can lock a stream at a time. This module tracks each reader + * independently to avoid lock conflicts. + * - **Resource Disposal**: Ensures that resources, such as readers and locks, are properly released + * when consumers are done reading from the stream. + * - **Asynchronous Disposal**: Handles async operations like canceling readers and releasing locks, + * ensuring proper resource management even in async scenarios. + * + * @example + * ```ts + * const sourceStream = new ReadableStream({ + * // stream initialization + * }); + * const multicastStream = createMulticastStream(sourceStream); + * + * // Creating multiple consumers + * const streamA = createConsumer(multicastStream); + * const streamB = createConsumer(multicastStream); + * + * // Each consumer can now read independently + * const readerA = streamA.getReader(); + * const readerB = streamB.getReader(); + * + * // Both readers will receive the same data chunks + * readerA.read().then(console.log); // Consumer 1 reads + * readerB.read().then(console.log); // Consumer 2 reads + * ``` + */ + +import type { DualDisposable } from "./_types.ts"; +import { assertNotEquals } from "./_assert.ts"; + +/** + * Global symbol to identify a `MulticastReadableStream` and distinguish it from a regular `ReadableStream`. + */ +export const MULTICAST_STREAM_SYMBOL = Symbol.for("MulticastReadableStream"); + +/** + * Global constant to track the original `getReader` method of `ReadableStream`. + * This is important to retain the default behavior while enhancing it. + */ +const originalReadableStreamGetReader = ReadableStream.prototype.getReader; + +/** + * WeakMaps to track various relationships between source streams, consumers, and readers. + * + * @remarks + * - **SourceStreamReaders**: Keeps track of the reader that is currently locked on the source stream. + * - **ConsumerStreams**: Stores a set of all consumer streams that are associated with the source stream. + * - **ConsumerStreamControllers**: Stores the controllers for the consumer streams. + * - **ConsumerStreamReaders**: Keeps track of the active readers for each consumer stream. + */ +export const SourceStreamReader = new WeakMap, ReadableStreamDefaultReader>(); +export const ConsumerStreams = new WeakMap, Set>>(); +export const ConsumerStreamControllers = new WeakMap, Set>>(); +export const ConsumerStreamReaders = new WeakMap, Map, ReadableStreamReaderWithDisposal>>>(); +export const ClosedConsumerStreamReaders = new WeakSet>>(); + +/** + * Interface for an enhanced `ReadableStream` with disposal capabilities. + * This allows for proper resource management, ensuring that streams can be canceled and + * disposed of both synchronously and asynchronously. + */ +export interface ReadableStreamWithDisposal extends ReadableStream, AsyncDisposable {} + +/** + * Represents a `MulticastReadableStream`, which extends the native `ReadableStream` by adding + * support for multiple consumers and custom disposal behavior. + * + * @template T - The type of data that the stream produces. + */ +export type MulticastReadableStream = Omit, "getReader"> & { + /** + * A symbol identifying this as an enhanced stream. + */ + [MULTICAST_STREAM_SYMBOL]: typeof MULTICAST_STREAM_SYMBOL; + + /** + * Retrieves a reader for the stream, ensuring proper disposal of resources. + * @returns The reader with disposal capabilities. + */ + getReader(): ReadableStreamReaderWithDisposal, T>; + + /** + * Overloaded `getReader` that allows passing additional options. + * @returns The reader with disposal capabilities. + */ + getReader(...args: Parameters["getReader"]> | []): ReadableStreamReaderWithDisposal, T>; +}; + +/** + * Represents a `ReadableStreamReader` that includes disposal capabilities, allowing the reader + * to be synchronously or asynchronously disposed of when no longer needed. + * + * @template R - The type of the reader. + * @template T - The type of data being read. + */ +export type ReadableStreamReaderWithDisposal, T = unknown> = R & DualDisposable; + +/** + * Enhances a `ReadableStream` to support multiple consumers and adds disposal capabilities. + * + * @template T - The type of data in the `ReadableStream`. + * @param stream - The `ReadableStream` to enhance. + * @returns A new `MulticastReadableStream` with disposal capabilities and support for multiple consumers. + * + * @example + * ```ts + * const sourceStream = new ReadableStream({ + * // stream initialization + * }); + * const multicastStream = createMulticastStream(sourceStream); + * + * const reader = multicastStream.getReader(); + * reader.read().then(console.log); // Consumer 1 reads + * + * const anotherReader = multicastStream.getReader(); + * anotherReader.read().then(console.log); // Consumer 2 reads + * ``` + */ +export function createMulticastStream( + stream: ReadableStream, +): MulticastReadableStream { + assertNotEquals( + (stream as unknown as Record)[MULTICAST_STREAM_SYMBOL], + MULTICAST_STREAM_SYMBOL, + "Stream is already enhanced" + ); + + // Enhancing the provided stream with custom methods for handling multiple consumers + const multicastStream = Object.assign(stream, { + [MULTICAST_STREAM_SYMBOL]: MULTICAST_STREAM_SYMBOL, + + /** + * Custom `getReader` method that manages readers and ensures disposal. + * It tracks the reader and ensures the proper management of the consumer stream. + * + * @param args - Optional parameters for the reader. + * @returns A reader with disposal support. + */ + getReader>( + this: ReadableStream, + ...args: Parameters["getReader"]> | [] + ) { + // Create a new consumer stream + const consumer = createConsumer(this); + + // Use the original `getReader` method to create a reader for the consumer stream + const rawReader = consumer.getReader(...args); + + // Store the original `cancel` and `releaseLock` methods + const originalCancel = rawReader.cancel; + const originalReleaseLock = rawReader.releaseLock; + + const reader = Object.assign(rawReader as ReadableStreamReaderWithDisposal, { + /** + * `releaseLock` method to handle the release of the reader lock. + * It works the same way the normal reader `releaseLock` works but it also keep track of the reader which was just released + * to ensure `System.asyncDisposal`, doesn't try to dispose of the same reader multiple times. + * + * This basically means that the errors you'd expect to see from releaseLocks are still there, + * e.g. you can't release a reader that has already been released. + */ + releaseLock(...args: Parameters["releaseLock"]>) { + const result = originalReleaseLock.apply(rawReader, args); + ClosedConsumerStreamReaders.add(reader); + return result; + }, + + /** + * `cancel` method to handle the cancellation of the reader. + * It works the same way the normal reader `cancel` works but it also keep track of the reader which was just canceled + * to ensure `System.asyncDisposal`, doesn't try to dispose of the same reader multiple times. + * + * This basically means that the errors you'd expect to see from cancel are still there, + * e.g. you can't cancel a reader that has already been canceled. + */ + cancel(...args: Parameters["cancel"]>) { + const result = originalCancel.apply(rawReader, args); + ClosedConsumerStreamReaders.add(reader); + return result; + }, + + /** + * Asynchronously dispose of the reader when done by cancelling the reader. + * This ensures that the reader is properly cleaned up when no longer needed. + * It also checks if the reader has already been released to avoid duplicate disposal. + */ + [Symbol.asyncDispose]() { + if (ClosedConsumerStreamReaders.has(reader)) { + return Promise.resolve(); + } + return originalCancel.call(rawReader, "Symbol.asyncDispose"); + } + }); + + // Track the reader in the ConsumerStreamReaders map for disposal management + if (!ConsumerStreamReaders.has(this)) { + ConsumerStreamReaders.set(this, new Map()); + } + ConsumerStreamReaders.get(this)?.set(consumer, reader); + + return reader; + }, + + /** + * Custom `cancel` method to handle the cancellation and disposal of the stream. + * Ensures that all readers and consumers are properly canceled and cleaned up. + * + * @param args - Optional arguments for cancellation. + * @returns A promise that resolves when the stream is fully canceled. + */ + async cancel(this: ReadableStream, ...args: Parameters["cancel"]>) { + const readers = ConsumerStreamReaders.get(this); + Array.from(readers?.values() ?? [], reader => reader?.releaseLock()); + + const reader = SourceStreamReader.get(this); + await reader?.cancel(...args); + reader?.releaseLock(); + + // Clean up the associated controllers and consumer streams + ConsumerStreamControllers.get(this)?.clear(); + ConsumerStreams.get(this)?.clear(); + readers?.clear(); + + SourceStreamReader.delete(this); + ConsumerStreamReaders.delete(this); + }, + + /** + * Provides an asynchronous iterator over the stream, allowing consumers to iterate + * over the data chunks. It locks and unlocks the reader properly. + * + * @yields {Promise} Each chunk of data from the stream. + */ + async *[Symbol.asyncIterator](this: ReadableStream) { + const reader = this.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + yield value; + } + } finally { + reader.releaseLock(); // Release the lock when done + } + }, + + /** + * Asynchronously disposes of the stream by canceling it and releasing resources. + * Ensures that resources are properly cleaned up when the stream is no longer needed. + * + * @returns A promise that resolves when the stream is fully disposed of. + */ + [Symbol.asyncDispose](this: ReadableStream) { + return this.cancel("Sybmol.asyncDispose"); + }, + }); + + return multicastStream as MulticastReadableStream; +} + +/** + * Creates a new `ReadableStream` for a consumer from a source stream. + * + * This function manages the relationship between the source stream and consumer streams, + * ensuring that multiple consumers can independently access the data. + * + * @template T - The type of data emitted by the stream. + * @param sourceStream - The source `ReadableStream` to create a consumer stream from. + * @param opts - (Optional) queuing strategy for the consumer stream. + * @returns A new `ReadableStream` for the consumer. + * + * @example + * ```ts + * const sourceStream = createInfiniteStream(); + * const streamA = createConsumer(sourceStream); + * const streamB = createConsumer(sourceStream); + * + * // streamA and streamB are independent consumers of the sourceStream + * ``` + */ +export function createConsumer(sourceStream: ReadableStream, opts?: QueuingStrategy): ReadableStream { + let currentCtrlr: ReadableStreamDefaultController | null = null; + + // Create a new consumer `ReadableStream` + const consumer = new ReadableStream({ + start(controller) { + // Track the controller for the consumer stream + const consumerCtrlrs = ConsumerStreamControllers.get(sourceStream) || new Set(); + consumerCtrlrs.add(currentCtrlr = controller); + ConsumerStreamControllers.set(sourceStream, consumerCtrlrs); + + // Create a reader for the source stream if not already created + if (!SourceStreamReader.has(sourceStream)) { + const sourceStreamReader = originalReadableStreamGetReader.apply(sourceStream) as ReadableStreamDefaultReader; + SourceStreamReader.set(sourceStream, sourceStreamReader); + } + }, + + async pull() { + const sourceStreamReader = SourceStreamReader.get(sourceStream); + if (!sourceStreamReader) { + return; + } + + const consumerStreamCtrlrs = ConsumerStreamControllers.get(sourceStream); + const consumers = ConsumerStreams.get(sourceStream); + + const { done, value } = await sourceStreamReader.read(); + if (done) { + // If the stream is done, close the controllers and release locks + consumerStreamCtrlrs?.forEach(ctrlr => ctrlr.close()); + currentCtrlr = null; + + sourceStreamReader.releaseLock(); + + consumers?.clear(); + consumerStreamCtrlrs?.clear(); + + SourceStreamReader.delete(sourceStream); + ConsumerStreamControllers.delete(sourceStream); + ConsumerStreams.delete(sourceStream); + return; + } + + // Distribute the value to all consumer controllers + consumerStreamCtrlrs?.forEach(ctrlr => ctrlr.enqueue(value)); + }, + + async cancel(reason) { + const consumerCtrlrs = ConsumerStreamControllers.get(sourceStream); + const sourceStreamReader = SourceStreamReader.get(sourceStream); + const consumers = ConsumerStreams.get(sourceStream); + + console.log({ + reason, + consumerCtrlrs, + sourceStreamReader, + consumers, + }); + + if (consumerCtrlrs && consumerCtrlrs.size > 0 && currentCtrlr) { + consumerCtrlrs.delete(currentCtrlr!); + currentCtrlr = null; + } + + if (consumers && consumers.size > 0 && consumer) { + consumers.delete(consumer!); + } + + if (sourceStreamReader && consumerCtrlrs && consumers && (consumerCtrlrs.size <= 0 && consumers.size <= 0)) { + await sourceStreamReader.cancel(reason); + sourceStreamReader.releaseLock(); + + consumers?.clear(); + consumerCtrlrs?.clear(); + + SourceStreamReader.delete(sourceStream); + ConsumerStreamControllers.delete(sourceStream); + ConsumerStreams.delete(sourceStream); + } + }, + }, opts); + + // Track the consumer stream for the source stream + const consumers = ConsumerStreams.get(sourceStream) || new Set(); + ConsumerStreams.set(sourceStream, consumers.add(consumer)); + + // Return the new consumer stream + return consumer; +} diff --git a/multicast_test.ts b/multicast_test.ts new file mode 100644 index 0000000..80a7efa --- /dev/null +++ b/multicast_test.ts @@ -0,0 +1,727 @@ +import { test } from "@libs/testing"; +import { expect } from "@std/expect"; + +import { createMulticastStream, ConsumerStreamReaders } from "./multicast.ts"; +import { iterableFromStream } from "./utils.ts"; + +function createInfiniteStream(delay = 10) { + let intervalId: ReturnType; + return new ReadableStream({ + start(controller) { + // Infinite stream for testing + let count = 0; + intervalId = setInterval(() => { + controller.enqueue(count++); + }, delay); + }, + cancel() { + console.log("Stream canceled"); + // Clean up when stream is canceled + clearInterval(intervalId); + }, + }); +} + +// Test Case ERS1: Basic reading functionality +test("deno")("createMulticastStream - basic reading functionality", async () => { + // Create a simple ReadableStream emitting [1, 2, 3] + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + controller.close(); + }, + }); + + // Enhance the stream + await using multicastStream = createMulticastStream(stream); + + const values = []; + for await (const value of multicastStream) { + values.push(value); + } + + expect(values).toEqual([1, 2, 3]); +}); + +// Test Case ERS2: Disposal using Symbol.asyncDispose +test("deno")("createMulticastStream - disposal using Symbol.asyncDispose", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + // Do not close the stream to simulate an ongoing stream + }, + }); + + await using multicastStream = createMulticastStream(stream); + + const values = []; + await Promise.all([ + // Dispose the stream while it is locked + new Promise(resolve => setTimeout(resolve, 0)) + .then(() => multicastStream[Symbol.asyncDispose]()), + + // Read from the stream + (async () => { + try { + for await (const value of multicastStream) { + values.push(value); + } + + console.log({ + stream, + multicastStream + }) + + expect(values.length).toBe(3); + } catch (error) { + // Expected to throw an error because the stream is canceled + expect(error).toBeDefined(); + expect(error).toBeInstanceOf(TypeError); + expect((error as TypeError)?.message).toContain("The reader was released."); + } + })(), + ]); + + expect(values.length).toBe(3); +}); + +// Test Case ERS5: Disposing a locked stream +test("deno")("createMulticastStream - disposing a locked stream", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + // Do not close the stream to keep it open + }, + }); + + await using multicastStream = createMulticastStream(stream); + + // Get the reader, which normally locks the stream + const reader = multicastStream.getReader(); + + // Dispose the stream while it is locked + await multicastStream[Symbol.asyncDispose](); + + try { + // Attempt to read from the reader + const { value, done } = await reader.read(); + expect(value).toBe(1); + expect(done).toBe(false); // Should error out since it's been disposed + } catch (error) { + // Expected behavior; the reader should be canceled + expect(error).toBeDefined(); + expect(error).toBeInstanceOf(TypeError); + expect((error as TypeError)?.message).toContain("Reader has no associated stream."); + } + + // Ensure that the reader's lock is released + expect(multicastStream.locked).toBe(false); +}); + +// Test Case ERS5: Disposing a locked stream +test("deno")("createMulticastStream - recover from a disposing a locked stream and then dispose it again", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + // Do not close the stream to keep it open + }, + }); + + await using multicastStream = createMulticastStream(stream); + + // Get the reader, which normally locks the stream + // You cannot `await using` here because it will cause the reader to + // try to dispose the consumer stream directly, but unfortuneately + // there is a call to dispose the source stream, this plays together to cause an issue. + // When a source stream is disposed, it will also dispose the consumer streams, + // but will not dispose the readers of the consumer stream per-se, so be careful + // of trying to dispose of the reader after as it will cause an error + await using reader1 = multicastStream.getReader(); + + console.log("Cool") + + // Dispose the stream while it is locked + await multicastStream[Symbol.asyncDispose](); + + // Get the reader, which normally locks the stream + // using `await using` here is fine because getReader creates a new consumer stream + await using reader2 = multicastStream.getReader(); + + try { + // Attempt to read from the reader + const { value, done } = await reader2.read(); + expect(value).toBeUndefined(); + expect(done).toBe(true); + } catch (error) { + // Expected behavior; the reader should be canceled + expect(error).toBeDefined(); + expect(error).toBeInstanceOf(TypeError); + expect((error as TypeError)?.message).toContain("Reader has no associated stream."); + } + + // Ensure that the reader's lock is released + expect(multicastStream.locked).toBe(false); +}); + + +// Test Case ERS5: Disposing a locked stream +test("deno")("createMulticastStream - recover from a disposing a locked stream and then cancel it this time", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + // Do not close the stream to keep it open + }, + }); + + await using multicastStream = createMulticastStream(stream); + + // Get the reader, which normally locks the stream + // You cannot `await using` here because it will cause the reader to + // try to dispose the consumer stream directly, but unfortuneately + // there is a call to dispose the source stream, this plays together to cause an issue. + // When a source stream is disposed, it will also dispose the consumer streams, + // but will not dispose the readers of the consumer stream per-se, so be careful + // of trying to dispose of the reader after as it will cause an error + const reader1 = multicastStream.getReader(); + + // Dispose the stream while it is locked + await multicastStream[Symbol.asyncDispose](); + + // Get the reader, which normally locks the stream + // using `await using` here is fine because getReader creates a new consumer stream + await using reader2 = multicastStream.getReader(); + + try { + // Attempt to read from the reader + const { value, done } = await reader2.read(); + expect(value).toBeUndefined(); + expect(done).toBe(true); + } catch (error) { + // Expected behavior; the reader should be canceled + expect(error).toBeDefined(); + expect(error).toBeInstanceOf(TypeError); + expect((error as TypeError)?.message).toContain("Reader has no associated stream."); + } + + // Ensure that the reader's lock is released + expect(multicastStream.locked).toBe(false); + + // Cancel the reader + try { + await reader1.cancel("Dispose of it"); + } catch (error) { + // Expected behavior; the reader should error keeping with the original behavior of cancel + expect(error).toBeDefined(); + expect(error).toBeInstanceOf(TypeError); + expect((error as TypeError)?.message).toContain("Reader has no associated stream."); + } +}); + +// Test Case ERS6: Read multiple streams simultaneously +test("deno")("createMulticastStream - read multiple streams simultaneously", async () => { + const createStream = (id: number) => { + return new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(`${id}-${value}`)); + controller.close(); + }, + }); + } + + await using stream1 = createMulticastStream(createStream(1)); + await using stream2 = createMulticastStream(createStream(2)); + + const results1: string[] = []; + const results2: string[] = []; + + await Promise.all([ + (async () => { + for await (const value of stream1) { + results1.push(value); + } + })(), + (async () => { + for await (const value of stream2) { + results2.push(value); + } + })(), + ]); + + expect(results1).toEqual(["1-1", "1-2", "1-3"]); + expect(results2).toEqual(["2-1", "2-2", "2-3"]); +}); + +// Test Case ERS7: Consume stream using while loop with .read() +test("deno")("createMulticastStream - consume using while loop with .read()", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + controller.close(); + }, + }); + + await using multicastStream = createMulticastStream(stream); + await using reader = multicastStream.getReader(); + + const values = []; + let result; + while (!(result = await reader.read()).done) { + values.push(result.value); + } + + expect(values).toEqual([1, 2, 3]); +}); + +// Test Case ERS8: Attempt to get a second reader when the stream is already locked +test("deno")("createMulticastStream - attempt to get a second reader when locked", async () => { + const stream = createInfiniteStream(); + + await using multicastStream = createMulticastStream(stream); + + await using reader1 = multicastStream.getReader(); + await using reader2 = multicastStream.getReader(); + + // Attempt to get a second reader + console.log({ + value1: await reader1.read(), + value2: await reader2.read(), + reader1, + reader2 + }) + expect(true).toBe(true); + + await multicastStream.cancel(); +}); + +// Test Case ERS9: Dispose the stream while it's being read +test("deno")("createMulticastStream - dispose the stream while it's being read", async () => { + const stream = createInfiniteStream(50); + await using multicastStream = createMulticastStream(stream); + const values: number[] = []; + + const readPromise = (async () => { + for await (const value of multicastStream) { + console.log({ value }) + values.push(value); + if (value >= 3) { + break; + } + } + })(); + + await readPromise; + await multicastStream[Symbol.asyncDispose](); + + // Ensure that only values up to 3 are read + expect(values).toEqual([0, 1, 2, 3]); + + // Ensure that the stream is properly disposed + expect(multicastStream.locked).toBe(false); + expect(ConsumerStreamReaders.has(multicastStream)).toBe(false); +}); + +// Test Case ERS10: Attempt to have a second reader with parallel reads when the stream is already locked +test("deno")("createMulticastStream - attempt to have a second reader with parallel reads when locked", async () => { + const stream = createInfiniteStream(); + await using multicastStream = createMulticastStream(stream); + + await Promise.race([ + (async () => { + try { + for await (const value of iterableFromStream(multicastStream)) { + console.log("Reader 1:", value); + } + } catch (_) { console.warn(_) } + })(), + (async () => { + try { + for await (const value of iterableFromStream(multicastStream)) { + console.log("Reader 2:", value); + } + } catch (_) { console.warn(_) } + })(), + + new Promise(resolve => setTimeout(resolve, 1000)) + ]); + + await multicastStream.cancel(); + + // await stream.cancel(); +}); + +// Test Case: Complex Stream Teeing and Disposal +test("deno")("createMulticastStream - complex teeing and disposal", async () => { + // Create a source ReadableStream that emits numbers every 500ms + const sourceStream = new ReadableStream({ + start(controller) { + let count = 0; + const intervalId = setInterval(() => { + if (count > 10) { + clearInterval(intervalId); + controller.close(); + return; + } else { + controller.enqueue(count++); + } + }, 500); + }, + }); + + // Split the source stream into two branches + await using multicastStream = createMulticastStream(sourceStream); + + // Enhance both branches + await using reader1 = multicastStream.getReader(); + await using reader2 = multicastStream.getReader(); + + const resultsBranch1: number[] = []; + const resultsBranch2: number[] = []; + const resultsSubBranch1: number[] = []; + const resultsSubBranch2: number[] = []; + + // Start reading from the parent branches + const readParentBranches = Promise.all([ + (async () => { + for await (const value of iterableFromStream(reader1)) { + resultsBranch1.push(value); + if (value === 3) { + // After reading some values, split branch1 into two sub-branches + const subBranch1 = multicastStream.getReader(); + const subBranch2 = multicastStream.getReader(); + + // Start reading from the sub-branches after a delay + setTimeout(() => { + (async () => { + try { + for await (const subValue of iterableFromStream(subBranch1)) { + resultsSubBranch1.push(subValue); + } + } catch (_) { + console.warn(_); + } + })(); + }, 2000); // Delay of 2 seconds + + setTimeout(() => { + (async () => { + try { + for await (const subValue of iterableFromStream(subBranch2)) { + resultsSubBranch2.push(subValue); + } + } catch (_) { + console.warn(_); + } + })(); + }, 2000); // Delay of 2 seconds + } + + if (value === 5) { + break; + } + } + })(), + (async () => { + for await (const value of iterableFromStream(reader2)) { + resultsBranch2.push(value); + if (value === 5) { + break; + } + } + })(), + ]); + + // Wait for the parent branches to finish reading + await readParentBranches; + + // Dispose of the parent branch after reading value 5 + // await enhancedBranch2[Symbol.asyncDispose](); + + // Wait for the sub-branches to read remaining values + await new Promise((resolve) => setTimeout(resolve, 6000)); // Wait longer to allow sub-branches to read all values + + // Output the results + console.log("Parent Branch 1:", resultsBranch1); + console.log("Parent Branch 2:", resultsBranch2); + console.log("Sub Branch 1:", resultsSubBranch1); + console.log("Sub Branch 2:", resultsSubBranch2); + + // // Assertions + expect(resultsBranch1).toEqual([0, 1, 2, 3, 4, 5]); + expect(resultsBranch2).toEqual([0, 1, 2, 3, 4, 5]); + + // The sub-branches should have started reading from value 3 onwards + expect(resultsSubBranch1[0]).toBe(4); + expect(resultsSubBranch2[0]).toBe(4); + + // The sub-branches should continue to read values even after parent branch is disposed + expect(resultsSubBranch1).toEqual([4, 5, 6, 7, 8, 9, 10]); + expect(resultsSubBranch2).toEqual([4, 5, 6, 7, 8, 9, 10]); +}); + + +// ==== + +// Test Case ERR1: Basic reading from enhanced reader +test("deno")("enhanceReaderWithDisposal - basic reading functionality", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + controller.close(); + }, + }); + + await using enhancedReader = createMulticastStream(stream).getReader(); + + const values = []; + let result: ReadableStreamReadResult; + do { + result = await enhancedReader.read(); + if (!result.done) { + values.push(result.value); + } + } while (!result.done); + + expect(values).toEqual([1, 2, 3]); +}); + +// Test Case ERR2: Disposal using Symbol.asyncDispose +test("deno")("enhanceReaderWithDisposal - disposal using Symbol.asyncDispose", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + // Do not close the stream + }, + }); + + const multicastStream = createMulticastStream(stream); + const reader = multicastStream.getReader(); + + // Dispose the reader + await reader[Symbol.asyncDispose](); + + try { + // Attempt to read from the reader + const value = await reader.read(); + console.log(value) + + // Should reach here + expect(true).toBe(true); + } catch (error) { + // Should not throw an error + expect(error).toBeDefined(); + expect(error).toBeInstanceOf(TypeError); + expect((error as TypeError)?.message).toContain("Reader has no associated stream."); + } + + expect(stream.locked).toBe(false); +}); + +test("deno")("enhanceReaderWithDisposal - cancel the reader early", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + // Do not close the stream + }, + }); + + const multicastStream = createMulticastStream(stream); + const reader = multicastStream.getReader(); + + // Dispose the reader + await reader.cancel("Duh"); + + try { + // Attempt to read from the reader + const value = await reader.read(); + console.log(value) + + // Should reach here + expect(true).toBe(true); + } catch (error) { + // Should not throw an error + expect(error).toBeDefined(); + expect(error).toBeInstanceOf(TypeError); + expect((error as TypeError)?.message).toContain("Reader has no associated stream."); + } + + expect(stream.locked).toBe(false); +}); + +test("deno")("enhanceReaderWithDisposal - cancel source stream, then try getting reader", async () => { + const stream = new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(value)); + // Do not close the stream + }, + }); + + await using multicastStream = createMulticastStream(stream); + + // Dispose the multistream early + await multicastStream.cancel("Duh"); + + try { + const reader = multicastStream.getReader(); + + // Attempt to read from the reader + const value = await reader.read(); + expect(value).toEqual({ value: 1, done: false }); + console.log(value) + + // Should reach here + expect(true).toBe(true); + } catch (error) { + // console.log({ error }) + // Should not throw an error + expect(error).toBeDefined(); + expect(error).toBeInstanceOf(TypeError); + expect((error as TypeError)?.message).toContain("The reader was released."); + } + + expect(stream.locked).toBe(true); +}); + +// Test Case ERR3: Multiple readers from different streams +test("deno")("enhanceReaderWithDisposal - multiple readers from different streams", async () => { + const createStream = (id: number) => { + return new ReadableStream({ + start(controller) { + [1, 2, 3].forEach((value) => controller.enqueue(`${id}-${value}`)); + controller.close(); + }, + }); + } + + const stream1 = createStream(1); + const stream2 = createStream(2); + + await using multicastStream1 = createMulticastStream(stream1); + await using multicastStream2 = createMulticastStream(stream2); + + await using reader1 = multicastStream1.getReader(); + await using reader2 = multicastStream2.getReader(); + + const values2: string[] = []; + const values1: string[] = []; + + await Promise.all([ + (async () => { + let result; + while (!(result = await reader1.read()).done) { + values1.push(result.value); + } + })(), + (async () => { + let result; + while (!(result = await reader2.read()).done) { + values2.push(result.value); + } + })(), + ]); + + expect(values1).toEqual(["1-1", "1-2", "1-3"]); + expect(values2).toEqual(["2-1", "2-2", "2-3"]); +}); + +// Test Case ERR4: Reader cancellation during read operations +test("deno")("enhanceReaderWithDisposal - reader cancellation during read", async () => { + let timeout: ReturnType; + const stream = new ReadableStream({ + start(controller) { + let count = 0; + function push() { + controller.enqueue(count++); + timeout = setTimeout(push, 10); + } + push(); + }, + cancel() { + clearTimeout(timeout); + }, + }); + + await using multicastStream = createMulticastStream(stream); + await using reader1 = multicastStream.getReader(); + await using reader2 = multicastStream.getReader(); + + const values: number[] = []; + + const read1Promise = (async () => { + while (true) { + const result = await reader1.read(); + if (result.done) break; + + values.push(result.value); + if (result.value >= 5) { + // Cancel the reader + await reader1.cancel("No longer needed"); + // break; + } + } + })(); + + const read2Promise = (async () => { + while (true) { + const result = await reader2.read(); + if (result.done) break; + + console.log(result) + } + })(); + + await read1Promise; + await Promise.race([ + read2Promise, + new Promise(resolve => setTimeout(resolve, 1000)) + ]); + + expect(values).toEqual([0, 1, 2, 3, 4, 5]); + expect(stream.locked).toBe(true); + + await reader2.cancel("No longer needed"); + expect(stream.locked).toBe(false); +}); + +// Test Case ERR5: Dispose reader while it's in the middle of reading +test("deno")("enhanceReaderWithDisposal - dispose reader during read", async () => { + let timeout: ReturnType; + const stream = new ReadableStream({ + start(controller) { + let count = 0; + function push() { + controller.enqueue(count++); + timeout = setTimeout(push, 20); + } + push(); + }, + cancel() { + clearTimeout(timeout); + }, + }); + + await using multicastStream = createMulticastStream(stream); + await using reader = multicastStream.getReader(); + + const values: number[] = []; + + const readPromise = (async () => { + while (true) { + const result = await reader.read(); + if (result.done) break; + values.push(result.value); + + if (result.value >= 3) { + // Dispose the reader + await reader[Symbol.asyncDispose](); + break; + } + } + })(); + + await readPromise; + + expect(values).toEqual([0, 1, 2, 3]); + + // Ensure the reader is released + expect(stream.locked).toBe(false); +}); diff --git a/split-by.ts b/split-by.ts new file mode 100644 index 0000000..121effd --- /dev/null +++ b/split-by.ts @@ -0,0 +1,57 @@ +import { splitStream } from "./split.ts"; + +// Create a source stream that produces numbers with varying delays +const sourceStream = new ReadableStream({ + async start(controller) { + for (let i = 0; i < 10; i++) { + // await new Promise(resolve => setTimeout(resolve, Math.random() * 500)); + controller.enqueue(i); + console.log(`Source produced: ${i}`); + } + controller.close(); + } +}); + +// Split the stream into even and odd numbers +const [evenStream, oddStream] = splitStream(sourceStream, num => num % 2 === 0); + +// const evenStreamReader = evenStream.getReader(); +// const oddStreamReader = oddStream.getReader(); + +// Helper function to read from a stream +async function readStream(stream: ReadableStream, name: string) { + const reader = stream.getReader(); + try { + while (true) { + const { value, done } = await reader.read(); + console.log({ name, value, done}) + if (done) break; + console.log(`${name} received: ${value}`); + await new Promise(resolve => setTimeout(resolve, 300)); + } + console.log(`${name} done`); + } finally { + reader.releaseLock(); + } +} + +// Read from both streams concurrently +Promise.all([ + readStream(evenStream, "Even Stream"), + // readStream(oddStreamReader, "Odd Stream") +]).then(() => console.log("All streams processed")); + +// Demonstrate cancellation after a delay +setTimeout(async () => { + // console.log("Cancelling streams"); + // await Promise.all([ + // evenStreamReader.cancel("Demo cancellation"), + // oddStreamReader.cancel("Demo cancellation") + // ]); +}, 2000); + +// // Use AsyncDisposable feature +// (async () => { +// await using streams = splitStream(sourceStream, num => num % 2 === 0); +// // ... use streams ... +// })(); \ No newline at end of file diff --git a/split.ts b/split.ts index 72a1ff0..ea04054 100644 --- a/split.ts +++ b/split.ts @@ -1,94 +1,15 @@ -import type { EnhancedReadableStream } from "./stream.ts"; -import { createChannel } from "./channel.ts"; - /** - * Splits a source ReadableStream into two separate ReadableStreams: one for valid values and one for errors encountered during stream processing. - * - * ### Error Handling: - * - This function catches errors during the enqueueing of values and routes them to the error stream. - * - Errors occurring during the transformation process will be sent to the error stream. - * - Valid values will continue to flow into the valid stream, ensuring that processing can continue even in the presence of errors. - * - * ### Disposal: - * - Each resulting stream supports multiple readers, and they will automatically handle resource cleanup once all consumers have finished reading. - * - The channels used to manage the streams are properly disposed of when the streams are closed or no longer needed. - * - * @template V The type of data contained in the source stream. - * @template E The type of errors encountered during stream processing. - * @param source The original source ReadableStream to be split. - * @returns An array containing two ReadableStreams: - * - The first stream for valid values. - * - The second stream for errors encountered during stream processing. - * - * @example - * ```ts - * import { splitStream } from "./stream.ts" - * - * // Example source stream with valid values and an error - * const sourceStream = new ReadableStream({ - * start(controller) { - * controller.enqueue("Valid value 1"); - * controller.enqueue("Valid value 2"); - * controller.error(new Error("Something went wrong")); - * controller.close(); - * } - * }); - * - * const [validStream, errorStream] = splitStream(sourceStream); - * - * // Reading from the valid stream - * const validReader = validStream.getReader(); - * validReader.read().then(({ value }) => console.log("Valid:", value)); // Logs: "Valid value 1" - * - * // Reading from the error stream - * const errorReader = errorStream.getReader(); - * errorReader.read().then(({ value }) => console.error("Error:", value)); // Logs: Error: Something went wrong - * ``` + * Context object to hold shared state between splitStream and readFromSource. */ -export function splitStream( - source: ReadableStream, -): - & readonly [EnhancedReadableStream, EnhancedReadableStream] - & AsyncDisposable { - // Create channels for valid values and errors - const validChannel = createChannel(); - const errorChannel = createChannel(); - - // Transformer to manage the splitting logic - const transformer = new TransformStream({ - async transform(chunk) { - try { - // Attempt to enqueue the chunk into the valid channel - await validChannel.getWriter().write(chunk); - } catch (error) { - // If an error occurs, enqueue the error into the error channel - await errorChannel.getWriter().write(error as E); - } - }, - async flush() { - // Close both channels when the stream is done - await Promise.all([ - validChannel[Symbol.asyncDispose](), - errorChannel[Symbol.asyncDispose]() - ]); - }, - }); - - // Pipe the source through the transformer - source.pipeThrough(transformer); - - // Return the two readable streams wrapped with disposables - return Object.assign( - [validChannel.readable, errorChannel.readable] as const, - { - async [Symbol.asyncDispose]() { - await Promise.all([ - validChannel[Symbol.asyncDispose](), - errorChannel[Symbol.asyncDispose]() - ]); - }, - }, - ); +export interface SplitStreamContext { + source: ReadableStreamDefaultReader | null; + predicate: ((chunk: T | F) => boolean) | null; + queues: [ + { readable: ReadableStream | null, writer: WritableStreamDefaultWriter | null, size: number }, + { readable: ReadableStream | null, writer: WritableStreamDefaultWriter | null, size: number }, + ]; + readInProgress: boolean; + sourceDone: boolean; } /** @@ -137,45 +58,154 @@ export function splitStream( * oddReader.read().then(({ value }) => console.log("Odd:", value)); // Logs: 1 * ``` */ -export function splitByStream( - source: ReadableStream, - predicate: (chunk: T | F) => boolean | PromiseLike, +export function splitStream( + sourceStream: ReadableStream, + predicate: (chunk: T | F) => boolean ): - & readonly [EnhancedReadableStream, EnhancedReadableStream] + & readonly [ReadableStream + // , ReadableStream + ] & AsyncDisposable { - // Create channels for true and false predicate results - const trueChannel = createChannel(); - const falseChannel = createChannel(); - - // Transformer to manage the splitting logic based on the predicate - const transformer = new TransformStream({ - async transform(chunk) { - // Route chunks based on the predicate - if (await predicate(chunk)) { - trueChannel.getWriter().write(chunk as T); - } else { - falseChannel.getWriter().write(chunk as F); - } - }, - async flush() { - // Close both channels when the stream is done - await Promise.all([ - trueChannel[Symbol.asyncDispose](), - falseChannel[Symbol.asyncDispose](), - ]); - }, - }); + // Create TransformStreams to act as minimal queues + const { writable: queue1, readable: readable1 } = new TransformStream(); + const { writable: queue2, readable: readable2 } = new TransformStream(); - // Pipe the source through the transformer - source.pipeThrough(transformer); + const context: SplitStreamContext = { + source: sourceStream.getReader(), + predicate, + queues: [ + { readable: readable1, writer: queue1.getWriter(), size: 0 }, + { readable: readable2, writer: queue2.getWriter(), size: 0 }, + ], + readInProgress: false, + sourceDone: false, + }; - // Return the two readable streams wrapped with disposables - return Object.assign([trueChannel.readable, falseChannel.readable] as const, { + const trueStream = createPullStream(context, { index: 0 }); + // const falseStream = createPullStream(context, { index: 1 }); + + return Object.assign([trueStream, + // falseStream + + ] as const, { async [Symbol.asyncDispose]() { await Promise.all([ - trueChannel[Symbol.asyncDispose](), - falseChannel[Symbol.asyncDispose](), + trueStream.cancel("Symbol.asyncDispose"), + // falseStream.cancel("Symbol.asyncDispose"), ]); }, + }) +} + +function createPullStream(context: SplitStreamContext, { index: currentQueueIndex }: { index: number }): ReadableStream { + const queue = context.queues[currentQueueIndex]; + const readable = queue.readable as ReadableStream | null; + + return new ReadableStream({ + async pull(controller) { + if (context.sourceDone) { + controller.close(); + return; + } + + // Always trigger a read after pulling, whether we got a value or not + // if (!context.readInProgress && !context.sourceDone) { + while (queue.size === 0 && !context.sourceDone) { + await readFromSource(context); + } + // } + + const reader = readable?.getReader(); + try { + const { value, done } = await reader?.read() ?? { value: undefined, done: true, error: "Something went wrong" }; + + console.log({ value, done }) + if (done) { + if (!context.sourceDone) { + // If this queue is empty but source isn't done, trigger another read + if (!context.readInProgress) { + await readFromSource(context); + } + + return; // Don't close the controller, wait for more data + } + + controller.close(); + } else { + controller.enqueue(value as V); + queue.size--; + } + + // Always trigger a read after pulling, whether we got a value or not + // if (!context.readInProgress && !context.sourceDone) { + // readFromSource(context); + // } + } catch (error) { + controller.error(error); + } finally { + reader?.releaseLock(); + } + }, + async cancel(reason) { + // Handle cancellation if needed + await readable?.cancel(reason); + + const allQueuesCancelled = context.queues.every((queue, currentQueueIndex) => { + if ((queue?.readable as V) === readable) { + context.queues[currentQueueIndex].writer?.releaseLock(); + + context.queues[currentQueueIndex].readable = null; + context.queues[currentQueueIndex].writer = null; + } + + return queue?.readable === null; + }); + + if (allQueuesCancelled && context.source) { + if (!context.sourceDone) { + await context.source?.cancel(reason); + } + + context.source?.releaseLock(); + + context.sourceDone = true; + context.source = null; + } + } }); } + +async function readFromSource(context: SplitStreamContext) { + const { source, predicate } = context; + if (context.readInProgress || context.sourceDone) return; + + try { + while (!context.sourceDone) { + const { value, done } = await source?.read() ?? { value: undefined, done: true }; + console.log({ value, done, readFromSource: true }); + + if (done) { + context.sourceDone = true; + await Promise.all([ + context.queues.map((queue) => queue?.writer?.close()) + ]); + break; + } + + if (predicate?.(value)) { + await context.queues[0]?.writer?.write(value as T); + context.queues[0].size++; + break; + } else { + await context.queues[1]?.writer?.write(value as F); + context.queues[1].size++; + break; + } + } + } catch (error) { + console.log("Error reading from source", error); + await Promise.all([ + context.queues.map((queue) => queue?.writer?.abort(error)) + ]); + } +} diff --git a/stream.ts b/stream.ts deleted file mode 100644 index 84f3d5d..0000000 --- a/stream.ts +++ /dev/null @@ -1,554 +0,0 @@ -import type { DualDisposable } from "./types.ts"; - -/** - * Metadata interface for tracking streams and their relationships. - */ -export interface StreamMetadata { - id?: string; // Optional identifier for debugging. - available: Promise; // Promise that resolves when the stream is available. - parent: ReadableStream | null; // Parent stream, if any. - children: Set>; // Set of child streams. -} - -/** - * Global constants for original ReadableStream methods. - */ -const originalReadableStreamTee = ReadableStream.prototype.tee; -const originalReadableStreamCancel = ReadableStream.prototype.cancel; -const originalReadableStreamGetReader = ReadableStream.prototype.getReader; - -// Registry to keep track of streams and their metadata. -export const ReadableStreamRegistry = new WeakMap, StreamMetadata>(); -// Map to track the current inactive streams associated with each source stream. -export const AvailableReadableStream = new WeakMap, ReadableStream>(); -// Map to keep count of streams created from each source stream. -export const ReadableStreamCounter = new WeakMap, number>(); - -/** - * A WeakMap that stores the `ReadableStreamReader` for a given `ReadableStream`. - * - * This map is used to track readers associated with specific streams, ensuring that each - * stream's reader can be managed and disposed of properly. - */ -// Weak -export const ReadableStreamReader = new WeakMap< - ReadableStream, - Map, ReadableStreamReaderWithDisposal>> ->(); - -/** - * Interface representing an enhanced `ReadableStream` with disposal capabilities. - */ -export interface ReadableStreamWithDisposal extends ReadableStream, AsyncDisposable { } - -/** - * Enhanced ReadableStream type with overloaded `getReader` method - */ -export type EnhancedReadableStream = Omit, "getReader"> & { - // Overload for default reader - getReader(): ReadableStreamReaderWithDisposal, T>; - - // Overload for BYOB mode - // getReader(options: { mode: "byob" }): ReadableStreamReaderWithDisposal; - - // Overload for general ReadableStreamGetReaderOptions - // getReader(options?: ReadableStreamGetReaderOptions): ReadableStreamReaderWithDisposal, T>; - - // Fallback using Parameters of the original getReader method - getReader(...args: Parameters["getReader"]> | []): ReadableStreamReaderWithDisposal, T>; -}; - -/** - * Interface representing an enhanced `ReadableStreamReader` with disposal capabilities. - */ -export type ReadableStreamReaderWithDisposal< - R extends ReadableStreamReader, - T = unknown -> = R & DualDisposable & { - stream: ReadableStream -}; - -/** - * Enhances the `ReadableStream` by adding disposal capabilities. - * - * @template T - The type of data in the `ReadableStream`. - * @param stream - The `ReadableStream` to wrap and enhance with disposal support. - * @returns A new `ReadableStream` that includes methods for synchronous and asynchronous disposal. - */ -export function enhanceReadableStream( - stream: ReadableStream, -): EnhancedReadableStream { - const enhancedStream = Object.assign(stream, { - /** - * Overrides the default `getReader` method to track and manage the reader. - * Ensures that the reader is properly associated with the stream and can be disposed of. - * - * @template R - The type of the reader. - * @param this - The enhanced `ReadableStream` instance. - * @param args - Arguments passed to the original `getReader` method. - * @returns The `ReadableStreamReader` associated with the stream. - */ - getReader>( - this: ReadableStream, - ...args: Parameters["getReader"]> | [] - ) { - const stream = createReadable(this); - const rawReader = stream.getReader(...args); - const reader = Object.assign(rawReader as ReadableStreamReaderWithDisposal, { - stream, - [Symbol.dispose]() { - return rawReader.releaseLock(); - }, - [Symbol.asyncDispose]() { - return Promise.resolve(rawReader.releaseLock()); - } - }); - - // Track the reader in the ReadableStreamReader to manage disposal - if (!ReadableStreamReader.has(this)) { - ReadableStreamReader.set(this, new Map()); - } - - // Return the enhanced reader - ReadableStreamReader.get(this)?.set(stream, reader); - return reader; - }, - - /** - * Overrides the default `getReader` method to track and manage the reader. - * Ensures that the reader is properly associated with the stream and can be disposed of. - * - * @param this - The `ReadableStream` instance for which the reader is being requested. - * @param args - Arguments passed to the original `getReader` method. - * @returns The `ReadableStreamReader` associated with the stream. - */ - async cancel(this: ReadableStream, ...args: Parameters["cancel"]>) { - const readers = ReadableStreamReader.get(this); - Array.from( - readers?.values() ?? [], - reader => reader?.releaseLock() - ); - await cancelAll(this, ...args); - readers?.clear(); - ReadableStreamReader.delete(this); - }, - - /** - * Provides an async iterator over the `ReadableStream`. - * - * Note: This method only works with the default reader, not the BYOB reader. - * - * @param this - The enhanced `ReadableStream` instance. - */ - async *[Symbol.asyncIterator](this: ReadableStream) { - const reader = this.getReader(); - - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - yield value; - } - } finally { - reader.releaseLock(); // Release the lock when done - } - }, - - /** - * Asynchronous disposal of the `ReadableStream`. - * - * This method cancels the stream and releases resources asynchronously. If the stream is - * locked, the lock is explicitly released before the stream is canceled. - * - * @param this - The enhanced `ReadableStream` instance. - * @returns A promise that resolves when the disposal is complete. - */ - [Symbol.asyncDispose](this: ReadableStream) { - return this.cancel(); - }, - }); - - return enhancedStream as EnhancedReadableStream; -} - -/** - * Custom `tee` function that splits a `ReadableStream` into two branches with additional control and tracking. - * - * This function acts similarly to the native `ReadableStream.tee()` method but provides enhanced functionality - * such as tracking availability and supporting asynchronous disposal. - * - * @typeParam T - The type of data chunks emitted by the stream. - * @param stream - The original `ReadableStream` to split. - * @returns A tuple containing two new `ReadableStreams`, a `Promise` that resolves when the streams are available, - * and implements `AsyncDisposable` for proper cleanup. - * - * @example - * ```ts - * const [branch1, branch2, available] = enhancedTee(originalStream); - * // Use branch1 and branch2 independently - * ``` - * - * @remarks - * This function is designed to work with streams that need enhanced control over cancellation and resource management. - * It ensures that both branches are properly handled in case of errors or cancellations. - */ -export function enhancedTee( - stream: ReadableStream, - ..._args: Parameters["tee"]> | [] -): [ReadableStream, ReadableStream, Promise] & AsyncDisposable { - // Create two TransformStreams to act as branches. - // TransformStream allows us to write to its writable end and read from its readable end. - const branch1 = new TransformStream(); - const branch2 = new TransformStream(); - - // Create a promise that will be resolved when the branches are fully set up. - const { promise: available, resolve } = Promise.withResolvers(); - - // Start an async function to read from the original stream and write to both branches. - (async () => { - // Get a reader from the original stream to read data chunks. - const reader = originalReadableStreamGetReader.apply(stream) as ReadableStreamDefaultReader; - // Get writers for both branches to write data into them. - const writer1 = branch1.writable.getWriter(); - const writer2 = branch2.writable.getWriter(); - - try { - while (true) { - // Read a chunk from the original stream. - const { done, value } = await reader.read(); - if (done) { - // If the original stream is done, close both writers. - // Use Promise.any to proceed as soon as one writer is closed successfully. - await Promise.any([ - writer1.close(), - writer2.close() - ]); - break; - } - - // Write the chunk to both branches. - // Use Promise.any to proceed as soon as one write is successful. - // This prevents the read loop from being blocked if one of the branches is slow. - await Promise.any([ - writer1.write(value), - writer2.write(value) - ]); - } - } catch (error) { - // If an error occurs, abort both writers. - // Use Promise.all to ensure both writers are aborted. - await Promise.all([ - writer1.abort(error), - writer2.abort(error) - ]); - } finally { - // Release the locks on the reader and writers. - reader.releaseLock(); - writer1.releaseLock(); - writer2.releaseLock(); - // Resolve the available promise to signal that the branches are available. - resolve(true); - } - })(); - - // Get the readable ends of the TransformStreams to return as the branches. - const stream1 = branch1.readable; - const stream2 = branch2.readable; - - // Prepare the result tuple with the two branches and the availability promise. - const result: [ - ReadableStream, - ReadableStream, - Promise - ] = [stream1, stream2, available]; - - // Implement AsyncDisposable to allow for proper cleanup. - return Object.assign(result, { - async [Symbol.asyncDispose]() { - const err = new Error("Cancelled"); - // Cancel both branches and wait for them to be available. - await Promise.all([ - stream1.cancel(err), - stream2.cancel(err), - available - ]); - } - }); -} - -/** - * Creates a new `ReadableStream` from a source stream, allowing dynamic branching. - * - * This function maintains a registry of streams and their relationships, enabling the creation of multiple - * readable streams from a single source stream. It uses an enhanced tee function to split the current inactive - * stream into active and inactive branches. - * - * @typeParam T - The type of data chunks emitted by the stream. - * @param sourceStream - The original `ReadableStream` to create a new readable from. - * @returns A new `ReadableStream` that reads data from the source stream. - * - * @example - * ```ts - * const sourceStream = createInfiniteStream(); - * const streamA = createReadable(sourceStream); - * const streamB = createReadable(sourceStream); - * // Now streamA and streamB are independent readers of the sourceStream. - * ``` - * - * @remarks - * This function keeps track of the current inactive stream associated with the source stream. - * Each time `createReadable` is called, it splits the current inactive stream into active and inactive branches. - * The active branch is returned, and the inactive branch becomes the new current inactive stream. - * This allows for dynamic creation of new readables from the source stream at any time. - */ -export function createReadable(sourceStream: ReadableStream): ReadableStream { - // Initialize the count for the sourceStream if not already set - if (!ReadableStreamCounter.has(sourceStream)) ReadableStreamCounter.set(sourceStream, 0); - - // Get the current inactive stream associated with the sourceStream, or default to the sourceStream - const currentInactiveStream = AvailableReadableStream.get(sourceStream) as ReadableStream || sourceStream; - - // Use enhancedTee to split the currentInactiveStream into active and inactive branches - const [active, inactive, available] = enhancedTee(currentInactiveStream); - - // Retrieve the metadata for the currentInactiveStream from the registry - let sourceNode = ReadableStreamRegistry.get(currentInactiveStream); - - // If the sourceNode doesn't exist, initialize it - if (!sourceNode) { - const count = ReadableStreamCounter.get(sourceStream)!; - // Set metadata for the currentInactiveStream in the registry - ReadableStreamRegistry.set(currentInactiveStream, (sourceNode = { - id: `${count}-source`, - available, - parent: null, - children: new Set([active, inactive]), - })); - - // Optionally mark the sourceStream for debugging purposes - Object.assign(sourceStream, { source: true }); - // Increment the count for the sourceStream - ReadableStreamCounter.set(sourceStream, count + 1); - } else { - // If the sourceNode exists, update its available promise and add the new branches to its children - sourceNode.available = available; - sourceNode.children.add(active); - sourceNode.children.add(inactive); - } - - // Get the updated count for naming purposes - const count = ReadableStreamCounter.get(sourceStream)!; - - // Create metadata for the active branch (the new readable to return) - ReadableStreamRegistry.set(active, { - id: `${count}-active`, - available, - parent: currentInactiveStream, - children: new Set(), - }); - // Optionally assign an id to the active stream for debugging - Object.assign(active, { id: `${count}-active` }); - - // Create metadata for the inactive branch (the new current inactive stream) - ReadableStreamRegistry.set(inactive, { - id: `${count}-inactive`, - available, - parent: currentInactiveStream, - children: new Set(), - }); - // Optionally assign an id to the inactive stream for debugging - Object.assign(inactive, { id: `${count}-inactive` }); - - // Update the currentStreams map with inactive as the new current inactive stream - AvailableReadableStream.set(sourceStream, inactive); - // Increment the count for the sourceStream - ReadableStreamCounter.set(sourceStream, count + 1); - - // Return the active branch as the new readable stream - return active; -} - -/** - * Cancels all streams starting from the given `sourceStream`, recursively canceling all its children. - * - * This function traverses the stream tree starting from the `sourceStream` and cancels each stream, - * ensuring that resources are properly cleaned up. - * - * @typeParam T - The type of data chunks emitted by the streams. - * @param sourceStream - The original `ReadableStream` from which to start cancellation. - * @returns A `Promise` that resolves when all streams have been canceled. - * - * @example - * await cancelAll(sourceStream); - * - * @remarks - * This function uses a stack to perform a depth-first traversal of the stream tree. - * It keeps track of visited streams to prevent processing the same stream multiple times. - * After canceling each stream, it updates the registry and currentStreams maps accordingly. - */ -export async function cancelAll(sourceStream: ReadableStream, ...args: Parameters["cancel"]>): Promise { - // Initialize the stack with the sourceStream - const stack = [[sourceStream]]; - // Initialize a set to keep track of visited streams - const visited = new WeakSet>(); - - // Perform a depth-first traversal of the stream tree - for (let i = 0; i < stack.length; i++) { - const queue = stack[i]; - const len = queue.length; - - for (let j = 0; j < len; j++) { - const stream = queue[j]; - - if (!visited.has(stream)) { - // Get the metadata for the stream - const node = ReadableStreamRegistry.get(stream); - if (node?.children?.size) { - // If the stream has children, add them to the stack for later processing - stack.push(Array.from(node.children) as ReadableStream[]); - } - - // Mark the stream as visited - visited.add(stream); - } - } - } - - // Cancel streams in reverse order to ensure proper cleanup - while (stack.length > 0) { - // Get the streams at the current level - const streams = stack.pop(); - - // Cancel each stream at this level - const cancellations = Array.from(streams ?? [], async substream => { - // Get the metadata for the substream - const substreamMetadata = ReadableStreamRegistry.get(substream); - // Cancel the substream, passing its id as a reason (optional) - await originalReadableStreamCancel.apply(substream, args ?? [substreamMetadata?.id]); - - // Get the parent stream - const parent = substreamMetadata?.parent!; - // Get the metadata for the parent - const metadata = ReadableStreamRegistry.get(parent); - // Wait for the parent's available promise to ensure it's ready - await metadata?.available; - - // Remove the substream from the parent's children - metadata?.children.delete(substream); - // Remove the substream from the registry - ReadableStreamRegistry.delete(substream); - - // Return the substream's metadata for debugging or logging - return Object.assign({}, substreamMetadata, substream); - }); - - // Wait for all cancellations at this level to complete - await Promise.all(cancellations); - } - - // Clean up the currentStreams map - AvailableReadableStream.delete(sourceStream); - ReadableStreamCounter.delete(sourceStream); -} - -/** - * Custom pipeTo function that works with enhanced streams. - * Pipes the stream to a writable destination. - */ -export async function enhancedPipeTo( - stream: ReadableStream, - destination: WritableStream, - options?: StreamPipeOptions -): Promise { - options = options ?? {}; - const { preventClose = false, preventAbort = false, preventCancel = false, signal } = options; - - const reader = stream.getReader(); - const writer = destination.getWriter(); - - let shuttingDown = false; - let currentWrite: Promise = Promise.resolve(); - - // Handle abort signal - if (signal) { - if (signal.aborted) { - await abort(); - throw new DOMException("Aborted", "AbortError"); - } - signal.addEventListener("abort", () => { - abort().catch(() => { }); - }, { once: true }); - } - - async function abort() { - if (shuttingDown) return; - shuttingDown = true; - - const actions = []; - - if (!preventAbort) { - actions.push(writer.abort(new DOMException("Aborted", "AbortError"))); - } else { - actions.push(writer.releaseLock()); - } - - if (!preventCancel) { - actions.push(reader.cancel(new DOMException("Aborted", "AbortError"))); - } else { - actions.push(reader.releaseLock()); - } - - await Promise.all(actions); - } - - async function pipeLoop(): Promise { - while (true) { - let readResult: ReadableStreamReadResult; - try { - readResult = await reader.read(); - if (readResult.done) { - break; - } - } catch (readError) { - if (!preventAbort) { - try { - await writer.abort(readError); - } catch { - // Ignore errors during abort - } - } - throw readError; - } - - try { - currentWrite = writer.write(readResult.value); - await currentWrite; - } catch (writeError) { - if (!preventCancel) { - try { - await reader.cancel(writeError); - } catch { - // Ignore errors during cancel - } - } - throw writeError; - } - } - } - - try { - await pipeLoop(); - - if (!preventClose) { - await writer.close(); - } else { - writer.releaseLock(); - } - } catch (error) { - // Handle errors already handled in pipeLoop - throw error; - } finally { - reader.releaseLock(); - writer.releaseLock(); - } -} diff --git a/stream_test.ts b/stream_test.ts deleted file mode 100644 index a35c530..0000000 --- a/stream_test.ts +++ /dev/null @@ -1,511 +0,0 @@ -import { test } from "@libs/testing"; -import { expect } from "@std/expect"; - -import { enhanceReadableStream, ReadableStreamReader } from "./stream.ts"; - -function createInfiniteStream(delay = 10) { - let intervalId: ReturnType; - return new ReadableStream({ - start(controller) { - // Infinite stream for testing - let count = 0; - intervalId = setInterval(() => { - controller.enqueue(count++); - }, delay); - }, - cancel() { - console.log("Stream canceled"); - // Clean up when stream is canceled - clearInterval(intervalId); - }, - }); -} - -// Test Case ERS1: Basic reading functionality -test("all")("enhanceReadableStream - basic reading functionality", async () => { - // Create a simple ReadableStream emitting [1, 2, 3] - const stream = new ReadableStream({ - start(controller) { - [1, 2, 3].forEach((value) => controller.enqueue(value)); - controller.close(); - }, - }); - - // Enhance the stream - const enhancedStream = enhanceReadableStream(stream); - - const values = []; - for await (const value of enhancedStream) { - values.push(value); - } - - expect(values).toEqual([1, 2, 3]); -}); - -// Test Case ERS2: Disposal using Symbol.asyncDispose -test("all")("enhanceReadableStream - disposal using Symbol.asyncDispose", async () => { - const stream = new ReadableStream({ - start(controller) { - [1, 2, 3].forEach((value) => controller.enqueue(value)); - // Do not close the stream to simulate an ongoing stream - }, - }); - - const enhancedStream = enhanceReadableStream(stream); - - // Dispose the stream before reading - enhancedStream[Symbol.asyncDispose](); - - const values = []; - try { - for await (const value of enhancedStream) { - values.push(value); - } - } catch (error) { - // Expected to throw an error because the stream is canceled - expect(error).toBeDefined(); - } - - expect(values.length).toBe(0); -}); - -// Test Case ERS5: Disposing a locked stream -test("all")("enhanceReadableStream - disposing a locked stream", async () => { - const stream = new ReadableStream({ - start(controller) { - [1, 2, 3].forEach((value) => controller.enqueue(value)); - // Do not close the stream to keep it open - }, - }); - - const enhancedStream = enhanceReadableStream(stream); - - // Get the reader, which locks the stream - const reader = enhancedStream.getReader(); - - // Dispose the stream while it is locked - enhancedStream[Symbol.asyncDispose](); - - // Attempt to read from the reader - try { - const { value, done } = await reader.read(); - expect(done).toBe(true); // Should be done because the stream was canceled - } catch (error) { - // Expected behavior; the reader should be canceled - expect(error).toBeDefined(); - } - - // Ensure that the reader's lock is released - expect(enhancedStream.locked).toBe(false); -}); - -// Test Case ERS6: Read multiple streams simultaneously -test("all")("enhanceReadableStream - read multiple streams simultaneously", async () => { - const createStream = (id: number) => - new ReadableStream({ - start(controller) { - [1, 2, 3].forEach((value) => controller.enqueue(`${id}-${value}`)); - controller.close(); - }, - }); - - const stream1 = enhanceReadableStream(createStream(1)); - const stream2 = enhanceReadableStream(createStream(2)); - - const results1: string[] = []; - const results2: string[] = []; - - await Promise.all([ - (async () => { - for await (const value of stream1) { - results1.push(value); - } - })(), - (async () => { - for await (const value of stream2) { - results2.push(value); - } - })(), - ]); - - expect(results1).toEqual(["1-1", "1-2", "1-3"]); - expect(results2).toEqual(["2-1", "2-2", "2-3"]); -}); - -// Test Case ERS7: Consume stream using while loop with .read() -test("all")("enhanceReadableStream - consume using while loop with .read()", async () => { - const stream = new ReadableStream({ - start(controller) { - [1, 2, 3].forEach((value) => controller.enqueue(value)); - controller.close(); - }, - }); - - const enhancedStream = enhanceReadableStream(stream); - const reader = enhancedStream.getReader(); - - const values = []; - let result; - while (!(result = await reader.read()).done) { - values.push(result.value); - } - - expect(values).toEqual([1, 2, 3]); -}); - -// Test Case ERS8: Attempt to get a second reader when the stream is already locked -test("all")("enhanceReadableStream - attempt to get a second reader when locked", async () => { - const stream = createInfiniteStream(); - - const enhancedStream = enhanceReadableStream(stream); - - const reader1 = enhancedStream.getReader(); - const reader2 = enhancedStream.getReader(); - - // Attempt to get a second reader - console.log({ - value1: await reader1.read(), - value2: await reader2.read(), - reader1, - reader2 - }) - expect(true).toBe(true); - - await enhancedStream.cancel(); -}); - -// Test Case ERS9: Dispose the stream while it's being read -test("all")("enhanceReadableStream - dispose the stream while it's being read", async () => { - const stream = createInfiniteStream(50); - const enhancedStream = enhanceReadableStream(stream); - const values: number[] = []; - - const readPromise = (async () => { - for await (const value of enhancedStream) { - values.push(value); - if (value >= 3) { - break; - } - } - })(); - - await readPromise; - await enhancedStream[Symbol.asyncDispose](); - - // Ensure that only values up to 3 are read - expect(values).toEqual([0, 1, 2, 3]); - - // Ensure that the stream is properly disposed - expect(enhancedStream.locked).toBe(false); - expect(ReadableStreamReader.has(enhancedStream)).toBe(false); -}); - -// Test Case ERS10: Attempt to have a second reader with parallel reads when the stream is already locked -test("all")("enhanceReadableStream - attempt to have a second reader with parallel reads when locked", async () => { - const stream = createInfiniteStream(); - const enhancedStream = enhanceReadableStream(stream); - - await Promise.race([ - (async () => { - try { - for await (const value of enhancedStream) { - console.log("Reader 1:", value); - } - } catch (_) { console.warn(_) } - })(), - (async () => { - try { - for await (const value of enhancedStream) { - console.log("Reader 2:", value); - } - } catch (_) { console.warn(_) } - })(), - - new Promise(resolve => setTimeout(resolve, 1000)) - ]); - - await enhancedStream.cancel(); - - // await stream.cancel(); -}); - -// Test Case: Complex Stream Teeing and Disposal -test.only("deno")("enhanceReadableStream - complex teeing and disposal", async () => { - // Create a source ReadableStream that emits numbers every 500ms - const sourceStream = new ReadableStream({ - start(controller) { - let count = 0; - const intervalId = setInterval(() => { - if (count > 10) { - clearInterval(intervalId); - controller.close(); - return; - } else { - controller.enqueue(count++); - } - }, 500); - }, - }); - - // Split the source stream into two branches - const enhacnedBranch = enhanceReadableStream(sourceStream); - - // Enhance both branches - const branch1 = enhacnedBranch.getReader().stream; - const branch2 = enhacnedBranch.getReader().stream; - - const enhancedBranch1 = enhanceReadableStream(branch1); - const enhancedBranch2 = enhanceReadableStream(branch2); - - const resultsBranch1: number[] = []; - const resultsBranch2: number[] = []; - const resultsSubBranch1: number[] = []; - const resultsSubBranch2: number[] = []; - - // Start reading from the parent branches - const readParentBranches = Promise.all([ - (async () => { - for await (const value of enhancedBranch1) { - resultsBranch1.push(value); - if (value === 3) { - // After reading some values, split branch1 into two sub-branches - const subBranch1 = enhancedBranch1.getReader().stream; - const subBranch2 = enhancedBranch1.getReader().stream; - - const enhancedSubBranch1 = enhanceReadableStream(subBranch1); - const enhancedSubBranch2 = enhanceReadableStream(subBranch2); - - // Start reading from the sub-branches after a delay - setTimeout(() => { - (async () => { - try { - for await (const subValue of enhancedSubBranch1) { - resultsSubBranch1.push(subValue); - } - } catch (_) { - console.warn(_); - } - })(); - }, 2000); // Delay of 2 seconds - - setTimeout(() => { - (async () => { - try { - for await (const subValue of enhancedSubBranch2) { - resultsSubBranch2.push(subValue); - } - } catch (_) { - console.warn(_); - } - })(); - }, 2000); // Delay of 2 seconds - } - - if (value === 5) { - break; - } - } - })(), - (async () => { - for await (const value of enhancedBranch2) { - resultsBranch2.push(value); - if (value === 5) { - break; - } - } - })(), - ]); - - // Wait for the parent branches to finish reading - await readParentBranches; - - // Dispose of the parent branch after reading value 5 - // await enhancedBranch2[Symbol.asyncDispose](); - - // Wait for the sub-branches to read remaining values - await new Promise((resolve) => setTimeout(resolve, 6000)); // Wait longer to allow sub-branches to read all values - - // Output the results - // console.log("Parent Branch 1:", resultsBranch1); - // console.log("Parent Branch 2:", resultsBranch2); - // console.log("Sub Branch 1:", resultsSubBranch1); - // console.log("Sub Branch 2:", resultsSubBranch2); - - // // Assertions - // expect(resultsBranch1).toEqual([0, 1, 2, 3, 4, 5]); - // expect(resultsBranch2).toEqual([0, 1, 2, 3, 4, 5]); - - // // The sub-branches should have started reading from value 3 onwards - // expect(resultsSubBranch1[0]).toBe(3); - // expect(resultsSubBranch2[0]).toBe(3); - - // // The sub-branches should continue to read values even after parent branch is disposed - // expect(resultsSubBranch1).toEqual([3, 4, 5, 6, 7, 8, 9, 10]); - // expect(resultsSubBranch2).toEqual([3, 4, 5, 6, 7, 8, 9, 10]); -}); - - -// ==== - -// Test Case ERR1: Basic reading from enhanced reader -test("all")("enhanceReaderWithDisposal - basic reading functionality", async () => { - const stream = new ReadableStream({ - start(controller) { - [1, 2, 3].forEach((value) => controller.enqueue(value)); - controller.close(); - }, - }); - - const enhancedReader = enhanceReadableStream(stream).getReader(); - - const values = []; - let result: ReadableStreamReadResult; - do { - result = await enhancedReader.read(); - if (!result.done) { - values.push(result.value); - } - } while (!result.done); - - expect(values).toEqual([1, 2, 3]); -}); - -// Test Case ERR2: Disposal using Symbol.asyncDispose -test("all")("enhanceReaderWithDisposal - disposal using Symbol.asyncDispose", async () => { - const stream = new ReadableStream({ - start(controller) { - [1, 2, 3].forEach((value) => controller.enqueue(value)); - // Do not close the stream - }, - }); - - const enhancedReader = enhanceReadableStream(stream).getReader(); - - // Dispose the reader - await enhancedReader[Symbol.asyncDispose](); - - // Attempt to read from the reader - try { - await enhancedReader.read(); - // Should not reach here - expect(true).toBe(false); - } catch (error) { - // Expected to throw an error - expect(error).toBeDefined(); - } -}); - -// Test Case ERR3: Multiple readers from different streams -test("all")("enhanceReaderWithDisposal - multiple readers from different streams", async () => { - const createStream = (id: number) => - new ReadableStream({ - start(controller) { - [1, 2, 3].forEach((value) => controller.enqueue(`${id}-${value}`)); - controller.close(); - }, - }); - - const stream1 = createStream(1); - const stream2 = createStream(2); - - const reader1 = enhanceReadableStream(stream1).getReader(); - const reader2 = enhanceReadableStream(stream2).getReader(); - - const values2: string[] = []; - const values1: string[] = []; - - await Promise.all([ - (async () => { - let result; - while (!(result = await reader1.read()).done) { - values1.push(result.value); - } - })(), - (async () => { - let result; - while (!(result = await reader2.read()).done) { - values2.push(result.value); - } - })(), - ]); - - expect(values1).toEqual(["1-1", "1-2", "1-3"]); - expect(values2).toEqual(["2-1", "2-2", "2-3"]); -}); - -// Test Case ERR4: Reader cancellation during read operations -test("all")("enhanceReaderWithDisposal - reader cancellation during read", async () => { - const stream = new ReadableStream({ - start(controller) { - let count = 0; - function push() { - controller.enqueue(count++); - setTimeout(push, 10); - } - push(); - }, - }); - - const reader = enhanceReadableStream(stream).getReader(); - - const values: number[] = []; - - const readPromise = (async () => { - while (true) { - const result = await reader.read(); - if (result.done) break; - values.push(result.value); - if (result.value >= 5) { - // Cancel the reader - await reader.cancel("No longer needed"); - break; - } - } - })(); - - await readPromise; - - expect(values).toEqual([0, 1, 2, 3, 4, 5]); - - // Ensure the reader is released - reader.releaseLock(); - expect(stream.locked).toBe(false); -}); - -// Test Case ERR5: Dispose reader while it's in the middle of reading -test("all")("enhanceReaderWithDisposal - dispose reader during read", async () => { - const stream = new ReadableStream({ - start(controller) { - let count = 0; - function push() { - controller.enqueue(count++); - setTimeout(push, 20); - } - push(); - }, - }); - - const reader = enhanceReadableStream(stream).getReader(); - - const values: number[] = []; - - const readPromise = (async () => { - while (true) { - const result = await reader.read(); - if (result.done) break; - values.push(result.value); - if (result.value >= 3) { - // Dispose the reader - reader[Symbol.asyncDispose](); - break; - } - } - })(); - - await readPromise; - - expect(values).toEqual([0, 1, 2, 3]); - - // Ensure the reader is released - expect(stream.locked).toBe(false); -}); diff --git a/utils.ts b/utils.ts index 3e43e4b..a3a44d5 100644 --- a/utils.ts +++ b/utils.ts @@ -1,14 +1,27 @@ /** - * Converts a ReadableStream into an Async Generator. + * Converts a ReadableStream or ReadableStreamDefaultReader into an Async Generator. * - * This allows you to iterate over the chunks of data in the stream using `for await...of` syntax. + * This function allows you to iterate over the chunks of data from a stream reader using the `for await...of` syntax. + * It supports both ReadableStream and ReadableStreamDefaultReader as input and provides options to configure its behavior. * - * @param stream The ReadableStream to convert. + * @param stream The ReadableStream or ReadableStreamDefaultReader to convert into an Async Generator. + * If a ReadableStream is passed, a reader is automatically obtained from it. + * @param options Configuration options for the function. + * @param options.autoRelease Whether to automatically release the lock on the reader when done. + * Defaults to `true` if a ReadableStream is passed, otherwise `false`. + * When set to `true`, the lock on the reader will be released automatically + * after the iteration is complete, ensuring that the stream can be read by other consumers. + * When set to `false`, the caller is responsible for releasing the lock. + * @param options.throw Whether to rethrow any errors encountered while reading from the stream. + * Defaults to `false`, which logs a warning instead. + * When set to `true`, any errors encountered during the reading process will be rethrown, + * allowing the caller to handle them explicitly. When set to `false`, errors are caught, + * a warning is logged, and the iteration stops gracefully. * @returns An Async Iterator that yields the chunks of data from the stream. * * @example * ```ts - * import { streamToAsyncGenerator } from "./stream.ts" + * import { iterableFromStream } from "./stream.ts" * const readableStream = new ReadableStream({ * start(controller) { * controller.enqueue("chunk1"); @@ -17,17 +30,43 @@ * } * }); * - * const asyncIterator = streamToAsyncGenerator(readableStream); + * const reader = readableStream.getReader(); + * const asyncIterator = iterableFromStream(reader); * * for await (const chunk of asyncIterator) { * console.log(chunk); // Logs: "chunk1", "chunk2" * } * ``` + * + * @example + * ```ts + * import { iterableFromStream } from "./stream.ts" + * const readableStream = new ReadableStream({ + * start(controller) { + * controller.enqueue("chunk1"); + * controller.enqueue("chunk2"); + * controller.close(); + * } + * }); + * + * const asyncIterator = iterableFromStream(readableStream, { autoRelease: true, throw: true }); + * + * try { + * for await (const chunk of asyncIterator) { + * console.log(chunk); // Logs: "chunk1", "chunk2" + * } + * } catch (error) { + * console.error("Stream error:", error); + * } + * ``` */ -export async function* streamToAsyncGenerator( - stream: ReadableStream, -): AsyncGenerator { - const reader = stream.getReader(); +export async function* iterableFromStream( + stream: ReadableStreamDefaultReader | ReadableStream, + { autoRelease: _autoRelease, throw: _throw = false }: { autoRelease?: boolean, throw?: boolean } = {}, +): AsyncIterable { + const isReadableStream = stream instanceof ReadableStream; + const reader = isReadableStream ? stream.getReader() : stream; + _autoRelease = _autoRelease ?? isReadableStream; try { while (true) { @@ -35,28 +74,40 @@ export async function* streamToAsyncGenerator( if (done) break; yield value; } + } catch (error) { + // If the stream is cancelled or there is an error, we handle it gracefully + if (_throw) throw error; + return; } finally { - reader.releaseLock(); + // Release the lock on the stream reader + if (_autoRelease) { + reader.releaseLock(); + } } } /** - * Converts an Async Iterator into a ReadableStream. + * Converts an Iterable, Async Iterable, Iterator, and/or Async Iterator into a ReadableStream. + * + * This allows you to produce a stream of data from a iterable. * - * This allows you to produce a stream of data from an async iterator. + * If the produced iterator (`iterable[Symbol.asyncIterator]()` or + * `iterable[Symbol.iterator]()`) is a generator, or more specifically is found + * to have a `.throw()` method on it, that will be called upon + * `readableStream.cancel()`. This is the case for the second input type above: * - * @param iterator The Async Iterator to convert. - * @returns A ReadableStream that streams the data produced by the async iterator. + * @param iterable The iterable to convert. + * @returns A ReadableStream that streams the data produced by the iterable. * * @example * ```ts - * import { asyncIteratorToStream } from "./stream.ts" + * import { iterableToStream } from "./stream.ts" * async function* asyncGenerator() { * yield "chunk1"; * yield "chunk2"; * } * - * const readableStream = asyncIteratorToStream(asyncGenerator()); + * const readableStream = iterableToStream(asyncGenerator()); * * const reader = readableStream.getReader(); * reader.read().then(({ value, done }) => { @@ -64,61 +115,28 @@ export async function* streamToAsyncGenerator( * }); * ``` */ -export function asyncIteratorToStream( - iterator: AsyncIterator, +export function iterableToStream( + iterable: Iterable | AsyncIterable | Iterator | AsyncIterator, ): ReadableStream { + const iterator: Iterator | AsyncIterator = + (iterable as AsyncIterable)[Symbol.asyncIterator]?.() ?? + (iterable as Iterable)[Symbol.iterator]?.() ?? + iterable; + return new ReadableStream({ async pull(controller) { - try { - const { value, done } = await iterator.next(); - if (done) { - controller.close(); - } else { - controller.enqueue(value); - } - } catch (err) { - controller.error(err); + const { value, done } = await iterator.next(); + if (done) { + controller.close(); + } else { + controller.enqueue(value); } }, - }); -} - -/** - * Converts a Synchronous Iterator into a ReadableStream. - * - * This allows you to produce a stream of data from a synchronous iterator. - * - * @param iterator The Synchronous Iterator to convert. - * @returns A ReadableStream that streams the data produced by the iterator. - * - * @example - * ```ts - * import { iteratorToStream } from "./stream.ts" - * function* syncGenerator() { - * yield "chunk1"; - * yield "chunk2"; - * } - * - * const readableStream = iteratorToStream(syncGenerator()); - * - * const reader = readableStream.getReader(); - * reader.read().then(({ value, done }) => { - * console.log(value); // Logs: "chunk1" - * }); - * ``` - */ -export function iteratorToStream(iterator: Iterator): ReadableStream { - return new ReadableStream({ - pull(controller) { - try { - const { value, done } = iterator.next(); - if (done) { - controller.close(); - } else { - controller.enqueue(value); - } - } catch (err) { - controller.error(err); + async cancel(reason) { + if (typeof iterator.throw == "function") { + try { + await iterator.throw(reason); + } catch { /* `iterator.throw()` always throws on site. We catch it. */ } } }, });