From 967f09bbf884a65cba218002df7512b0e50a89d0 Mon Sep 17 00:00:00 2001 From: Okiki Date: Tue, 15 Oct 2024 05:42:00 +0000 Subject: [PATCH] feat: add code Signed-off-by: Okiki --- .gitignore | 149 +++++++++++++ channel.ts | 435 ++++++++++++++++++++++++++++++++++++++ deno.jsonc | 23 ++ events.ts | 323 ++++++++++++++++++++++++++++ mod.ts | 6 + split.ts | 181 ++++++++++++++++ stream.ts | 554 +++++++++++++++++++++++++++++++++++++++++++++++++ stream_test.ts | 511 +++++++++++++++++++++++++++++++++++++++++++++ types.ts | 81 ++++++++ utils.ts | 125 +++++++++++ 10 files changed, 2388 insertions(+) create mode 100644 .gitignore create mode 100644 channel.ts create mode 100644 deno.jsonc create mode 100644 events.ts create mode 100644 mod.ts create mode 100644 split.ts create mode 100644 stream.ts create mode 100644 stream_test.ts create mode 100644 types.ts create mode 100644 utils.ts diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7b49742 --- /dev/null +++ b/.gitignore @@ -0,0 +1,149 @@ +# Logs +logs +*.log +npm-debug.log* +yarn-debug.log* +yarn-error.log* +lerna-debug.log* +.pnpm-debug.log* + +# Diagnostic reports (https://nodejs.org/api/report.html) +report.[0-9]*.[0-9]*.[0-9]*.[0-9]*.json + +# Runtime data +pids +*.pid +*.seed +*.pid.lock + +# Directory for instrumented libs generated by jscoverage/JSCover +lib-cov + +# Coverage directory used by tools like istanbul +coverage +*.lcov + +# nyc test coverage +.nyc_output + +# Grunt intermediate storage (https://gruntjs.com/creating-plugins#storing-task-files) +.grunt + +# Bower dependency directory (https://bower.io/) +bower_components + +# node-waf configuration +.lock-wscript + +# Compiled binary addons (https://nodejs.org/api/addons.html) +build/Release + +# Dependency directories +node_modules/ +jspm_packages/ + +# Snowpack dependency directory (https://snowpack.dev/) +web_modules/ + +# TypeScript cache +*.tsbuildinfo + +# Optional npm cache directory +.npm + +# Optional eslint cache +.eslintcache + +# Optional stylelint cache +.stylelintcache + +# Microbundle cache +.rpt2_cache/ +.rts2_cache_cjs/ +.rts2_cache_es/ +.rts2_cache_umd/ + +# Optional REPL history +.node_repl_history + +# Output of 'npm pack' +*.tgz + +# Yarn Integrity file +.yarn-integrity + +# dotenv environment variable files +.env +.env.development.local +.env.test.local +.env.production.local +.env.local + +# parcel-bundler cache (https://parceljs.org/) +.cache +.parcel-cache + +# Next.js build output +.next +out + +# Nuxt.js build / generate output +.nuxt +dist + +# Gatsby files +.cache/ +# Comment in the public line in if your project uses Gatsby and not Next.js +# https://nextjs.org/blog/next-9-1#public-directory-support +# public + +# vuepress build output +.vuepress/dist + +# vuepress v2.x temp and cache directory +.temp +.cache + +# Docusaurus cache and generated files +.docusaurus + +# Serverless directories +.serverless/ + +# FuseBox cache +.fusebox/ + +# DynamoDB Local files +.dynamodb/ + +# TernJS port file +.tern-port + +# Stores VSCode versions used for testing VSCode extensions +.vscode-test + +# yarn v2 +.yarn/cache +.yarn/unplugged +.yarn/build-state.yml +.yarn/install-state.gz +.pnp.* + +# Deno +coverage + +# NPM packages +.npmrc +*.mjs +node_modules +package.json +package-lock.json +pnpm-lock.yaml +bun.lockb +deno.lock + +# Vercel +.vercel + +# Docs +docs \ No newline at end of file diff --git a/channel.ts b/channel.ts new file mode 100644 index 0000000..1565fcc --- /dev/null +++ b/channel.ts @@ -0,0 +1,435 @@ +import type { EnhancedReadableStream } from "./stream.ts"; +import { enhanceReadableStream } from "./stream.ts"; + +/** + * Creates a unidirectional communication channel built on Web Streams, allowing data to flow from one or more writers to multiple independent readers. + * + * The writable stream is accessible for piping, and readers can be accessed via a getter. The channel can be closed, and the shared writer is accessible via a method. + * The channel also supports disposal using `Symbol.dispose`. + * + * ## Overview + * + * A **channel** in this context refers to a communication pathway where data can be sent by one or more producers (writers) and received by one or more consumers (readers). + * The channel concept is built on top of the **Web Streams API**, which provides the foundational `ReadableStream`, `WritableStream`, and `TransformStream` constructs. + * + * The Web Streams API was introduced to standardize streaming data handling in JavaScript, particularly for use cases involving large or continuous data flows, like processing files or handling network data. + * It allows for more efficient memory usage and better control over data flow compared to older approaches like Promises or callbacks. + * + * ### Key Components: + * - **ReadableStream**: Represents a source of data that you can read from. In this channel, it allows multiple readers to independently consume the same data stream. + * - **WritableStream**: Represents a destination for data that you can write to. Here, multiple writers can push data into the channel. + * - **TransformStream**: Combines a `WritableStream` and a `ReadableStream`, allowing data to be modified as it passes through. This stream powers the internal mechanics of the channel. + * + * ### Why Use a Channel? + * Channels built on Web Streams offer a flexible way to manage data flow between multiple producers and consumers. Unlike traditional methods like Promises or callbacks, streams provide built-in support for backpressure and memory efficiency, making them ideal for scenarios where data is being produced or consumed at different rates. + * + * ### Weaknesses of Web Streams + * While powerful, Web Streams can be complex to manage, especially when dealing with multiple readers and writers. The native API does not inherently support multiple readers consuming the same data stream, which is where this `createChannel` utility comes in. + * It simplifies these tasks by providing a more user-friendly interface for common patterns, such as setting up multiple consumers or sharing a writer among different components. + * + * ### Internals of `createChannel` + * Internally, `createChannel` uses a `TransformStream` to handle data writing and reading. The writable part of this stream is exposed for data input, while the readable part is split into multiple branches using the `ReadableStream.tee()` method. + * This allows each reader to independently consume the same data stream, with the channel managing the complexity of coordinating these operations. + * + * ### Why Not Use Iterators or Generators? + * While iterators, iterables, and generators are powerful tools in JavaScript, they are not ideal for scenarios involving multiple independent consumers of the same data stream. + * Web Streams are designed specifically for handling streaming data with built-in support for backpressure, which ensures that producers do not overwhelm consumers, and vice versa. This makes them more suitable for scenarios where data production and consumption rates may vary, as is often the case in real-time applications. + * + * ### Methods & Properties: + * - **writable**: The writable stream where data can be pushed. This stream supports piping and direct writing. + * - **getWriter()**: Provides access to the shared writer, allowing direct writes to the channel. This is useful when you need explicit control over writing, such as when coordinating between multiple components. + * - **readable**: A getter that returns a new readable stream each time it is accessed. This allows multiple readers to independently consume the same data stream. + * - **close()**: Closes the channel by shutting down the writable stream and canceling all active readable branches, ensuring that no more data can be written or read. + * - **[Symbol.dispose]()**: Implements the disposal protocol, allowing the channel to be cleanly disposed of using the `using` keyword or equivalent patterns in resource management. + * + * > [!WARNING] + * > _Do not pipe to the writable stream and write to it directly at the same time._ + * > + * > Piping data into the writable stream and manually writing to it simultaneously can lead to conflicts and unpredictable behavior. + * > Choose one method of writing to the stream to maintain consistent and reliable data flow. + * + * @template T - The type of data transmitted through the channel. + * @returns An object containing the writable stream, a method to get the shared writer, a getter for the readable stream, and methods to close or dispose of the channel. + * + * @example Basic Usage + * ```typescript + * // Create a channel + * const channel = createChannel(); + * + * // Access the shared writer and write data to the channel + * const writer = channel.getWriter(); + * writer.write("Message 1"); + * writer.write("Message 2"); + * + * // Set up multiple readers to consume the data from the channel + * (async () => { + * for await (const value of channel.readable) { + * console.log("Reader 1 received:", value); + * } + * })(); + * + * (async () => { + * for await (const value of channel.readable) { + * console.log("Reader 2 received:", value); + * } + * })(); + * + * // Close the channel after operations are done + * channel.close(); + * ``` + * + * @example Using the `using` Keyword + * ```typescript + * // Automatically managing the channel's lifecycle + * using channel = createChannel(); + * + * // Writer and reader operations... + * + * // The channel is automatically closed and disposed of when the block scope ends + * ``` + * + * @example Piping Data into the Channel + * ```typescript + * // Piping data into the channel + * const readable = new ReadableStream({ + * start(controller) { + * controller.enqueue("Piped data 1"); + * controller.enqueue("Piped data 2"); + * controller.close(); + * }, + * }); + * + * // Pipe the data into the channel's writable stream + * readable.pipeTo(channel.writable); + * + * // Reader consuming the piped data + * (async () => { + * for await (const value of channel.readable) { + * console.log("Reader received:", value); + * } + * })(); + * ``` + * + * @example Using Channels Between Web Workers + * ```ts + * // Main thread + * const channel = createChannel(); + * const worker = new Worker("worker.js"); + * worker.postMessage({ readable: channel.readable }, [channel.readable]); + * + * const writer = channel.getWriter(); + * writer.write("Hello from main thread!"); + * + * channel.close(); + * + * // worker.js + * self.addEventListener("message", async (event) => { + * const { readable } = event.data; + * for await (const message of readable) { + * console.log(message); // Logs "Hello from main thread!" + * } + * }); + * ``` + */ +export function createChannel(): Channel { + const transformStream = new TransformStream(); + const sharedWriter = transformStream.writable.getWriter(); + const readableStream = transformStream.readable; + + const enhancedReadableStream = enhanceReadableStream(readableStream); + + return { + /** + * The writable stream that can be used for piping or writing directly. + */ + writable: transformStream.writable, + + /** + * Getter for the readable stream, which supports multiple readers via tee. + * Each time it's accessed, it provides a new reader wrapped with disposal support. + * + * @returns A new readable stream with disposal support. + */ + readable: enhancedReadableStream, + + /** + * Method to get the shared writer for direct writing. + * @returns WritableStreamDefaultWriter + */ + getWriter(): WritableStreamDefaultWriter { + return sharedWriter; + }, + + /** + * Asynchronously disposes of the channel resources using the Symbol.asyncDispose protocol. + * This ensures that all readers are properly canceled and cleaned up asynchronously. + * @returns A promise that resolves when the disposal is complete. + */ + async [Symbol.asyncDispose]() { + await Promise.all([ + sharedWriter.close(), // Close the writable stream + enhancedReadableStream.cancel(), + ]); + }, + }; +} + +/** + * Creates a bidirectional channel that supports full-duplex communication between two endpoints, leveraging Web Streams. + * + * Each endpoint can read and write independently, allowing for full-duplex communication. + * The channel supports disposal using `Symbol.dispose`. + * + * ## Overview + * + * A **bidirectional channel** allows for two-way communication between two distinct entities, often referred to as endpoints A and B. + * This type of communication is essential in scenarios like client-server architectures, where both sides need to send and receive data. + * + * The **Web Streams API** is the foundation of this bidirectional channel, utilizing `ReadableStream`, `WritableStream`, and `TransformStream` to handle data flow. + * + * ### Key Concepts: + * - **Full-Duplex Communication**: Both endpoints can send and receive data independently, without blocking each other. This is achieved by using two unidirectional channels internally, one for each direction of communication. + * - **ReadableStream and WritableStream**: Each endpoint has its own readable and writable streams, allowing for independent data flow in both directions. + * - **TransformStream**: Used internally to manage the flow of data and ensure that each endpoint's streams are connected appropriately. + * + * ### Why Use a Bidirectional Channel? + * Bidirectional channels are crucial for real-time applications, peer-to-peer communication, and any scenario where two entities need to exchange data continuously. + * By building on Web Streams, these channels benefit from efficient data handling, backpressure management, and compatibility with other stream-based APIs. + * + * ### Weaknesses of Channels and Web Streams + * While streams provide a powerful abstraction for managing data flow, they can be complex to implement correctly, especially when dealing with multiple readers and writers. + * Channels, while simplifying some of these complexities, introduce additional overhead in coordinating multiple streams and managing their lifecycle. + * + * ### Internals of `createBidirectionalChannel` + * This function builds on `createChannel`, effectively combining two unidirectional channels to support duplex communication. + * Each endpoint in the channel has its own writable stream for sending data and a readable stream for receiving data from the other endpoint. + * This setup ensures that data flows smoothly in both directions without interference. + * + * ### Methods & Properties: + * - **endpointA.writer**: The writable stream for endpoint A, used to send data to endpoint B. + * - **endpointA.readable**: The readable stream for endpoint A, used to receive data from endpoint B. + * - **endpointB.writer**: The writable stream for endpoint B, used to send data to endpoint A. + * - **endpointB.readable**: The readable stream for endpoint B, used to receive data from endpoint A. + * - **[Symbol.dispose]()**: Implements the disposal protocol, ensuring both endpoints' resources are released when the channel is no longer needed. + * + * > [!WARNING] + * > _Do not pipe to the writable stream and write to it directly at the same time._ + * > + * > Piping data into the writable stream and manually writing to it simultaneously can lead to conflicts and unpredictable behavior. + * > Choose one method of writing to the stream to maintain consistent and reliable data flow. + * + * @template TRequest - The type of data sent from endpoint A to B. + * @template TResponse - The type of data sent from endpoint B to A. + * @returns An object containing methods to access the writable and readable streams for both endpoints and a method to dispose of the channel. + * + * @example Basic Bidirectional Communication + * ```typescript + * // Create a bidirectional channel for communication between two endpoints + * const bidirectionalChannel = createBidirectionalChannel(); + * + * // Endpoint A writes a request and reads a response + * const writerA = bidirectionalChannel.endpointA.writer; + * const readerA = bidirectionalChannel.endpointA.readable; + * + * writerA.write("Request from A"); + * + * (async () => { + * for await (const value of readerA) { + * console.log("Endpoint A received:", value); + * } + * })(); + * + * // Endpoint B reads the request and writes a response + * const writerB = bidirectionalChannel.endpointB.writer; + * const readerB = bidirectionalChannel.endpointB.readable; + * + * (async () => { + * for await (const value of readerB) { + * console.log("Endpoint B received:", value); + * writerB.write(`Response to ${value}`); + * } + * })(); + * + * // Dispose of the bidirectional channel when done + * bidirectionalChannel[Symbol.dispose](); + * ``` + * + * @example Using the `using` Keyword + * ```typescript + * // Automatically managing the bidirectional channel's lifecycle + * using bidirectionalChannel = createBidirectionalChannel(); + * + * // Endpoint A writes and reads data + * bidirectionalChannel.endpointA.writer.write("Request from A"); + * + * (async () => { + * for await (const value of bidirectionalChannel.endpointA.readable) { + * console.log("Endpoint A received:", value); + * } + * })(); + * + * // Endpoint B reads and writes data + * (async () => { + * for await (const value of bidirectionalChannel.endpointB.readable) { + * bidirectionalChannel.endpointB.writer.write(`Response to ${value}`); + * } + * })(); + * + * // The channel is automatically closed and disposed of when the block scope ends + * ``` + * + * @example Full-Duplex Communication Example + * ```typescript + * // Complex scenario with full-duplex communication between two endpoints + * const duplexChannel = createBidirectionalChannel(); + * + * // Endpoint A sending multiple requests + * duplexChannel.endpointA.writer.write("First request"); + * duplexChannel.endpointA.writer.write("Second request"); + * + * // Endpoint B processing and responding + * (async () => { + * for await (const request of duplexChannel.endpointB.readable) { + * console.log("Endpoint B processing:", request); + * duplexChannel.endpointB.writer.write(`Response to ${request}`); + * } + * })(); + * + * // Endpoint A receiving responses + * (async () => { + * for await (const response of duplexChannel.endpointA.readable) { + * console.log("Endpoint A received:", response); + * } + * })(); + * + * // Dispose of the duplex channel when complete + * duplexChannel[Symbol.dispose](); + * ``` + */ +export function createBidirectionalChannel< + TRequest, + TResponse, +>(): BidirectionalChannel { + const channelAtoB = createChannel(); + const channelBtoA = createChannel(); + + return { + /** + * Endpoint A can write requests and read responses. + */ + endpointA: { + get writer() { + return channelAtoB.getWriter(); + }, + get readable() { + return channelBtoA.readable; + }, + }, + + /** + * Endpoint B can write responses and read requests. + */ + endpointB: { + get writer() { + return channelBtoA.getWriter(); + }, + get readable() { + return channelAtoB.readable; + }, + }, + + /** + * Asynchronously disposes of the bidirectional channel resources using the Symbol.asyncDispose protocol. + * This ensures that all streams are closed and resources are released asynchronously. + * @returns A promise that resolves when the disposal is complete. + */ + async [Symbol.asyncDispose]() { + await Promise.all([ + channelAtoB[Symbol.asyncDispose](), + channelBtoA[Symbol.asyncDispose](), + ]); + }, + }; +} + + + +/** + * This module provides utility functions for creating both unidirectional and bidirectional communication channels built on top of Web Streams. + * The channels enable data flow between one or more producers (writers) and multiple consumers (readers), with support for full-duplex communication in the bidirectional case. + * + * The unidirectional `createChannel` function allows multiple independent readers to access the same stream of data written by one or more producers. + * The bidirectional `createBidirectionalChannel` function enables two endpoints to communicate in a full-duplex manner, each with its own readable and writable streams. + * + * Both types of channels support proper resource disposal via `Symbol.dispose`, and the writable streams are accessible for piping or direct writing, while the readable streams allow for multiple independent consumers. + * + * The Web Streams API serves as the foundation for these channels, offering efficient handling of continuous or large data flows with built-in backpressure management. This makes them ideal for scenarios such as real-time communication, file processing, or network data streaming. + * + * @see {@link https://developer.mozilla.org/en-US/docs/Web/API/Streams_API Streams API Documentation} for more details on Web Streams. + * + * @module + */ + +/** + * Interface representing a unidirectional communication channel. + * + * @template T - The type of data transmitted through the channel. + */ +export interface Channel { + /** + * The writable stream that can be used for piping or writing directly. + */ + readonly writable: WritableStream; + + /** + * Method to get the shared writer for direct writing. + * @returns WritableStreamDefaultWriter + */ + getWriter(): WritableStreamDefaultWriter; + + /** + * Getter for the readable stream, which supports multiple readers via tee. + * Each time it's accessed, it provides a new reader wrapped with disposal support. + * + * @returns A new readable stream with disposal support. + */ + readonly readable: EnhancedReadableStream; + + /** + * Asynchronously disposes of the channel resources using the Symbol.asyncDispose protocol. + * This ensures that all readers are properly canceled and cleaned up asynchronously. + * @returns A promise that resolves when the disposal is complete. + */ + [Symbol.asyncDispose](): Promise; +} + +/** + * Interface representing a bidirectional communication channel. + * + * @template TRequest - The type of data sent from endpoint A to B. + * @template TResponse - The type of data sent from endpoint B to A. + */ +export interface BidirectionalChannel { + /** + * Endpoint A can write requests and read responses. + */ + readonly endpointA: { + readonly writer: WritableStreamDefaultWriter; + readonly readable: EnhancedReadableStream; + }; + + /** + * Endpoint B can write responses and read requests. + */ + readonly endpointB: { + readonly writer: WritableStreamDefaultWriter; + readonly readable: EnhancedReadableStream; + }; + + /** + * Asynchronously disposes of the bidirectional channel resources using the Symbol.asyncDispose protocol. + * This ensures that all streams are closed and resources are released asynchronously. + * @returns A promise that resolves when the disposal is complete. + */ + [Symbol.asyncDispose](): Promise; +} \ No newline at end of file diff --git a/deno.jsonc b/deno.jsonc new file mode 100644 index 0000000..f6b786b --- /dev/null +++ b/deno.jsonc @@ -0,0 +1,23 @@ +{ + "name": "@okikio/streams", + "version": "0.1.0", + "exports": { + ".": "./mod.ts", + "./split": "./split.ts", + "./channel": "./channel.ts", + "./events": "./events.ts", + "./utils": "./utils.ts", + "./stream": "./stream.ts", + "./types": "./types.ts" + }, + "tasks": { + "test": "deno test -RW --allow-run=deno,bun,node,npx --clean --trace-leaks", + "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" + } +} diff --git a/events.ts b/events.ts new file mode 100644 index 0000000..3897d1f --- /dev/null +++ b/events.ts @@ -0,0 +1,323 @@ +/** + * This module provides functionality for creating and managing status event dispatchers using the Web Streams API. + * It allows for efficient broadcasting of status events to multiple listeners with proper handling of backpressure and synchronization. + * + * @module + */ + +import type { EnhancedReadableStream } from "./stream.ts"; +import { createChannel } from "./channel.ts"; + +/** + * Creates a status event dispatcher that allows dispatching and listening to status events + * using the Web Streams API. This function utilizes a channel internally to manage event + * dispatching and multiple listeners via `for await...of`. + * + * ## What is a Channel? + * A channel is a communication mechanism built on top of Web Streams, allowing data to flow + * from one or more producers (writers) to multiple independent consumers (readers). Channels + * efficiently manage data flow, backpressure, and synchronization across multiple consumers. + * + * In the context of `createStatusEventDispatcher`, the channel provides the infrastructure for + * broadcasting status events to multiple listeners while ensuring that all listeners receive + * the events as they occur. For more details, see the documentation for `createChannel`. + * + * @see {@link createChannel} for more information on how channels are implemented and their benefits. + * + * ## Key Features: + * - Allows multiple listeners to concurrently listen to status events using `for await...of`. + * - Ensures that all listeners receive events, though their processing order may be influenced by backpressure. + * - Uses `ReadableStream.tee()` to create multiple branches, enabling independent consumption of the same data stream. + * + * ## Important Considerations: + * + * ### Event Timing and Backpressure + * - **Event Dispatch Timing**: Unlike `EventTarget`, where all listeners receive the event immediately, + * `ReadableStreams` introduce backpressure. This means that if one listener is slower in consuming + * the stream, it could delay the delivery of events to other listeners. + * - **Synchronization**: All branches created via `tee()` must be ready to consume data before the + * underlying source produces more data. This can result in a slower processing speed if there is + * a discrepancy in consumption rates between different listeners. + * + * ### Behavior of `ReadableStream.tee()` + * - **Mid-Stream `tee()`**: When `tee()` is called on a `ReadableStream` mid-event and a new listener + * is added via `for await...of`, the new stream will only start consuming events from the next + * unconsumed event onward. It will not receive any previously consumed events from before the `tee()` + * was called. + * - **Pre-existing Streams**: The original stream will continue consuming data as normal, but its + * consumption pace might be affected by the newly created branches. Specifically, if the new branches + * are slower, the original stream will be paused until all branches are ready to consume the next event. + * - **No Retroactive Events**: The newly created branches do not loop through or receive events that were + * already consumed by the original stream. They only process new events that have not yet been consumed. + * + * ### Performance Impact + * - **Backpressure Management**: The system will introduce natural backpressure if one of the branches + * is slower, ensuring data consistency across all branches. However, this could lead to performance + * degradation if one listener significantly lags behind others. + * - **Use Case Suitability**: This method works best when streams are set up at the start and have + * similar consumption speeds. If frequent dynamic listener addition is required, a custom event + * multiplexing solution might be preferable. + * + * @template T - The specific status type being dispatched and listened for. + * + * @returns An object with methods to dispatch status events, listen to them, and manage the stream lifecycle. + * + * @remarks + * This function is particularly useful for scenarios where status events need to be broadcasted to multiple + * listeners that might consume the events at different rates. The use of Web Streams ensures efficient + * backpressure management, though developers should be aware of the potential impact on performance if + * listeners consume data at different speeds. + * + * @example + * ```typescript + * // Create a status event dispatcher + * const statusDispatcher = createStatusEventDispatcher(); + * + * // Example listeners using `for await...of` + * (async () => { + * const reader = statusDispatcher.events; + * for await (const event of reader) { + * console.log("Listener 1 received:", event.status); + * } + * })(); + * + * (async () => { + * const reader = statusDispatcher.events; + * for await (const event of reader) { + * console.log("Listener 2 received:", event.status); + * } + * })(); + * + * // Dispatching events + * (async () => { + * const runningEvent = new StatusEvent(Status.Running, { data: { jobId: 123 } }); + * const pausedEvent = new StatusEvent(Status.Paused); + * + * await statusDispatcher.dispatch(runningEvent); + * await statusDispatcher.dispatch(pausedEvent); + * + * // Close the dispatcher when done + * statusDispatcher.close(); + * })(); + * ``` + * + * @see CustomEvent + * @see ReadableStream + * @see WritableStream + * @see TransformStream + * @see {@link createChannel} for more information on how channels are implemented and their benefits. + * + * @public + */ +export function createEventDispatcher>(): EventDispatcher { + const channel = createChannel(); + + return { + /** + * Dispatches a status event to the channel. + * @param event - The status event to dispatch. + */ + async dispatch(event: E): Promise { + const writer = channel.getWriter(); + await writer.write(event); + }, + + /** + * Provides a new readable stream that can be used with `for await...of` + * to listen to status events. + * + * @returns A new readable stream of StatusEvents. + */ + get events(): EnhancedReadableStream { + return channel.readable; + }, + + /** + * Disposes of the dispatcher asynchronously, releasing resources. + */ + [Symbol.asyncDispose]() { + return channel[Symbol.asyncDispose](); + }, + }; +} + +/** + * Listens for a specific event type from a `ReadableStream` and returns the first matching event. + * Supports aborting the operation using an `AbortSignal`. + * + * This function continuously reads from the provided `ReadableStream` until it finds an event + * that matches the specified type. It also supports cancellation through an `AbortSignal`, + * allowing the operation to be aborted if necessary. If the stream ends without a matching + * event and `throwOnNoMatch` is set to `true`, the function will throw an error. + * + * ## AbortSignal Support: + * If an `AbortSignal` is provided and the operation is aborted: + * - The function will resolve with the reason for the abortion. + * - The reader lock will be released before the stream is canceled. + * + * ## Lock Handling: + * The function automatically releases the lock on the stream's reader when the operation completes, + * either by finding a matching event, reaching the end of the stream, or encountering an abort signal. + * + * ## Edge Cases: + * - If the stream ends without finding a matching event and `throwOnNoMatch` is `false`, the function will return `undefined`. + * - If the stream is aborted, the function will return the abort reason. + * - If `throwOnNoMatch` is `true` and no matching event is found, the function will throw an error. + * + * @template T - The type of events in the `ReadableStream`. + * @template K - The specific event type to listen for. + * @param stream - The `ReadableStream` or `ReadableStreamDefaultReader` to listen to. + * @param type - The event type to match against the stream's events. + * @param options.signal - An optional `AbortSignal` to cancel the operation. + * @param options.throwOnNoMatch - A boolean that, when true, throws an error if none of the events match. Defaults to `false`. + * @returns A promise that resolves with the first event that matches the specified type, the abort reason, or `undefined` if no match is found and `throwOnNoMatch` is `false`. + * + * @example + * ```typescript + * const statusEventStream = createStatusEventDispatcher().readable; + * + * // Listen for the "Running" status event + * const runningEvent = await waitForEvent(statusEventStream, Status.Running, { + * signal: abortSignal + * }); + * + * if (runningEvent) { + * console.log('Received running event:', runningEvent); + * } else { + * console.log('No running event received or operation was aborted'); + * } + * ``` + * + * @example + * ```typescript + * const statusEventStream = createStatusEventDispatcher().readable; + * + * // Listen for the "Paused" status event with error handling if no match is found + * try { + * const pausedEvent = await waitForEvent(statusEventStream, Status.Paused, { + * signal: abortSignal, + * throwOnNoMatch: true + * }); + * console.log('Received paused event:', pausedEvent); + * } catch (error) { + * console.error('Error:', error.message); + * } + * ``` + */ +export async function waitForEvent< + T extends Event = CustomEvent, + K extends unknown = unknown +>( + stream: ReadableStream | ReadableStreamDefaultReader, + type: K, + { signal, throwOnNoMatch = false }: WaitForEventOptions = {}, +): Promise { + const reader = "getReader" in stream ? stream.getReader() : stream; + let result: ReadableStreamReadResult | AbortedReadableStreamReadDoneResult; + + // Promise that resolves if the signal is aborted + const abortable = new Promise( + (resolve) => { + if (signal?.aborted) { + resolve({ + done: true, + value: { aborted: true, reason: signal.reason }, + }); + } else { + signal?.addEventListener?.("abort", () => { + resolve({ + done: true, + value: { aborted: true, reason: signal.reason }, + }); + }, { once: true }); + } + }, + ); + + try { + do { + if (signal?.aborted) return signal.reason; + + // Wait for either a read result or an abort signal + result = await Promise.race([ + reader.read(), + abortable, + ]); + + // Check if the operation was aborted + if ((result as AbortedReadableStreamReadDoneResult)?.value?.aborted) { + return (result as AbortedReadableStreamReadDoneResult)?.value?.reason as + | Error + | Event; + } + + // Check if the event matches the desired type + if ((result as ReadableStreamReadDoneResult)?.value?.type === type) { + return (result as ReadableStreamReadDoneResult).value; + } + } while (!result?.done); + } finally { + // Release the reader's lock before attempting to cancel the stream + reader?.releaseLock?.(); + } + + if (throwOnNoMatch) { + throw new Error(`Stream ended without receiving event of type: ${type}`); + } +} + +/** + * Represents the return type of the `createStatusEventDispatcher` function. + * + * This interface defines the structure of the dispatcher object, which includes methods for dispatching events, + * accessing the readable stream of events, and managing the lifecycle of the dispatcher. + */ +export interface EventDispatcher> { + /** + * Dispatches a status event to the channel. + * + * @param event - The status event to dispatch. + * @returns A promise that resolves when the event has been dispatched. + */ + dispatch(event: E): Promise; + + /** + * Provides a new readable stream that can be used with `for await...of` + * to listen to status events. + * + * @returns A new readable stream of StatusEvents. + */ + readonly events: EnhancedReadableStream; + + /** + * Disposes of the dispatcher asynchronously, releasing resources. + * + * @returns A promise that resolves when the disposal is complete. + */ + [Symbol.asyncDispose](): Promise; +} + +/** + * Options for the `waitForEvent` function. + */ +export interface WaitForEventOptions { + /** + * Whether to throw an error if no matching event is found. + * @defaultValue false + */ + throwOnNoMatch?: boolean; + + /** + * An optional `AbortSignal` to cancel the operation. + */ + signal?: AbortSignal; +} + +/** + * Represents the result of a `ReadableStream` read operation that was aborted. + * This type is used internally by the `waitForEvent` function to manage abort signals. + * @internal + */ +export type AbortedReadableStreamReadDoneResult = ReadableStreamReadDoneResult< + { aborted: boolean; reason: unknown } +>; diff --git a/mod.ts b/mod.ts new file mode 100644 index 0000000..aefedb7 --- /dev/null +++ b/mod.ts @@ -0,0 +1,6 @@ +export * from "./stream.ts"; +export * from "./channel.ts"; +export * from "./split.ts"; +export * from "./utils.ts"; + +export type * from "./types.ts"; \ No newline at end of file diff --git a/split.ts b/split.ts new file mode 100644 index 0000000..72a1ff0 --- /dev/null +++ b/split.ts @@ -0,0 +1,181 @@ +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 + * ``` + */ +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]() + ]); + }, + }, + ); +} + +/** + * Splits a source ReadableStream into two separate ReadableStreams based on a predicate function. + * + * ### Predicate-Based Splitting: + * - The source stream is evaluated chunk by chunk using the provided predicate function. + * - Chunks that satisfy the predicate are routed to the first stream. + * - Chunks that do not satisfy the predicate are routed to the second stream. + * + * ### Disposal: + * - Each resulting stream supports multiple readers and 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 T The type of data contained in the source stream. + * @param source The original source ReadableStream to be split. + * @param predicate A function that evaluates each chunk and returns a boolean. `true` means the chunk goes to the first stream, `false` means it goes to the second stream. + * @returns An array containing two ReadableStreams: + * - The first stream contains chunks that satisfied the predicate. + * - The second stream contains chunks that did not satisfy the predicate. + * + * @example + * ```ts + * import { splitByStream } from "./stream.ts" + * + * // Example source stream with numbers + * const sourceStream = new ReadableStream({ + * start(controller) { + * controller.enqueue(1); + * controller.enqueue(2); + * controller.enqueue(3); + * controller.enqueue(4); + * controller.close(); + * } + * }); + * + * const isEven = (value: number) => value % 2 === 0; + * const [evenStream, oddStream] = splitByStream(sourceStream, isEven); + * + * // Reading from the even stream + * const evenReader = evenStream.getReader(); + * evenReader.read().then(({ value }) => console.log("Even:", value)); // Logs: 2 + * + * // Reading from the odd stream + * const oddReader = oddStream.getReader(); + * oddReader.read().then(({ value }) => console.log("Odd:", value)); // Logs: 1 + * ``` + */ +export function splitByStream( + source: ReadableStream, + predicate: (chunk: T | F) => boolean | PromiseLike, +): + & readonly [EnhancedReadableStream, EnhancedReadableStream] + & 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](), + ]); + }, + }); + + // Pipe the source through the transformer + source.pipeThrough(transformer); + + // Return the two readable streams wrapped with disposables + return Object.assign([trueChannel.readable, falseChannel.readable] as const, { + async [Symbol.asyncDispose]() { + await Promise.all([ + trueChannel[Symbol.asyncDispose](), + falseChannel[Symbol.asyncDispose](), + ]); + }, + }); +} diff --git a/stream.ts b/stream.ts new file mode 100644 index 0000000..84f3d5d --- /dev/null +++ b/stream.ts @@ -0,0 +1,554 @@ +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 new file mode 100644 index 0000000..a35c530 --- /dev/null +++ b/stream_test.ts @@ -0,0 +1,511 @@ +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/types.ts b/types.ts new file mode 100644 index 0000000..16b8e4e --- /dev/null +++ b/types.ts @@ -0,0 +1,81 @@ +/** + * `DualDisposable` interface combines the capabilities of both `Disposable` and `AsyncDisposable` interfaces. + * + * This interface is used for objects that require explicit resource management, typically for cleaning up + * resources such as file handles, database connections, or any other resources that need to be disposed + * of when no longer in use. + * + * @example + * ```typescript + * class MyResource implements DualDisposable { + * [Symbol.dispose]() { + * // Synchronous cleanup logic + * } + * + * async [Symbol.asyncDispose]() { + * // Asynchronous cleanup logic + * } + * } + * + * const resource = new MyResource(); + * + * // Ensure resource is disposed synchronously + * using (resource) { + * // Work with resource + * } + * + * // Ensure resource is disposed asynchronously + * await using (await resource) { + * // Work with resource asynchronously + * } + * ``` + * + * The `Symbol.dispose` method will be invoked automatically when the scope in which the `using` keyword is used + * is exited, ensuring that resources are properly released. Similarly, `Symbol.asyncDispose` will be called + * for asynchronous disposal. + * + * @interface + * @extends Disposable + * @extends AsyncDisposable + */ +export interface DualDisposable extends Disposable, AsyncDisposable { } + + +/** + * A `PromiseWithDisposable` is an extension of the standard `Promise` interface, designed to include + * the ability to clean up resources once the promise is no longer needed or has completed its operation. + * + * ## What is a Disposable? + * + * A **disposable** is an object that implements the `Disposable` and/or `AsyncDisposable` interfaces, + * providing a standard way to release or clean up resources, such as memory or file handles, when they + * are no longer needed. This is particularly important in scenarios where failing to release resources + * can lead to memory leaks or other performance issues. + * + * The `Disposable` interface typically includes a `dispose` method, which can be called to perform + * synchronous cleanup. The `AsyncDisposable` interface includes an `asyncDispose` method, which is + * used for asynchronous cleanup operations. + * + * When using a `PromiseWithDisposable`, you can be confident that any associated resources will be + * properly cleaned up once the promise is settled (resolved or rejected) or when it's manually disposed of. + * This makes it particularly useful in scenarios where promises represent operations tied to external + * resources, such as file I/O, network requests, or UI components. + * + * @template T - The type of the value that the promise resolves to. + * + * @example + * ```typescript + * // Create a disposable promise + * const disposablePromise: PromiseWithDisposable = someAsyncOperation(); + * + * // Use the promise as you would any other promise + * disposablePromise.then(result => console.log(result)); + * + * // When done, dispose of the promise to clean up resources + * disposablePromise[Symbol.dispose](); + * ``` + * + * @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise Promise Documentation} + * @see {@link https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Symbol/dispose Symbol.dispose Documentation} + */ +export interface PromiseWithDisposal extends Promise, DualDisposable { } \ No newline at end of file diff --git a/utils.ts b/utils.ts new file mode 100644 index 0000000..3e43e4b --- /dev/null +++ b/utils.ts @@ -0,0 +1,125 @@ +/** + * Converts a ReadableStream into an Async Generator. + * + * This allows you to iterate over the chunks of data in the stream using `for await...of` syntax. + * + * @param stream The ReadableStream to convert. + * @returns An Async Iterator that yields the chunks of data from the stream. + * + * @example + * ```ts + * import { streamToAsyncGenerator } from "./stream.ts" + * const readableStream = new ReadableStream({ + * start(controller) { + * controller.enqueue("chunk1"); + * controller.enqueue("chunk2"); + * controller.close(); + * } + * }); + * + * const asyncIterator = streamToAsyncGenerator(readableStream); + * + * for await (const chunk of asyncIterator) { + * console.log(chunk); // Logs: "chunk1", "chunk2" + * } + * ``` + */ +export async function* streamToAsyncGenerator( + stream: ReadableStream, +): AsyncGenerator { + const reader = stream.getReader(); + + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + yield value; + } + } finally { + reader.releaseLock(); + } +} + +/** + * Converts an Async Iterator into a ReadableStream. + * + * This allows you to produce a stream of data from an async iterator. + * + * @param iterator The Async Iterator to convert. + * @returns A ReadableStream that streams the data produced by the async iterator. + * + * @example + * ```ts + * import { asyncIteratorToStream } from "./stream.ts" + * async function* asyncGenerator() { + * yield "chunk1"; + * yield "chunk2"; + * } + * + * const readableStream = asyncIteratorToStream(asyncGenerator()); + * + * const reader = readableStream.getReader(); + * reader.read().then(({ value, done }) => { + * console.log(value); // Logs: "chunk1" + * }); + * ``` + */ +export function asyncIteratorToStream( + iterator: AsyncIterator, +): ReadableStream { + 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); + } + }, + }); +} + +/** + * 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); + } + }, + }); +} -- 2.51.2