Readable Streams

/**
 * A readable stream represents a source of data, from which you can read.
 */
class ReadableStream<R = any> implements AsyncIterable<R> {
  constructor(underlyingSource?: UnderlyingSource<R>, strategy?: QueuingStrategy<R>);
  constructor(underlyingSource: UnderlyingByteSource, strategy?: { highWaterMark?: number; size?: undefined });
  
  /** Whether or not the readable stream is locked to a reader */
  readonly locked: boolean;
  
  /** Cancels the stream, signaling a loss of interest in the stream by a consumer */
  cancel(reason?: any): Promise<void>;
  
  /** Creates a ReadableStreamDefaultReader and locks the stream to the new reader */
  getReader(): ReadableStreamDefaultReader<R>;
  
  /** Creates a ReadableStreamBYOBReader and locks the stream to the new reader */
  getReader(options: { mode: 'byob' }): ReadableStreamBYOBReader;
  
  /** Provides a convenient, chainable way of piping this readable stream through a transform stream */
  pipeThrough<RS extends ReadableStream>(
    transform: ReadableWritablePair<R, RS>, 
    options?: StreamPipeOptions
  ): RS;
  
  /** Pipes this readable stream to a given writable stream */
  pipeTo(destination: WritableStream<R>, options?: StreamPipeOptions): Promise<void>;
  
  /** Tees this readable stream, returning a two-element array containing the two resulting branches */
  tee(): [ReadableStream<R>, ReadableStream<R>];
  
  /** Asynchronously iterates over the chunks in the stream's internal queue */
  values(options?: ReadableStreamIteratorOptions): ReadableStreamAsyncIterator<R>;
  
  /** Asynchronously iterates over the chunks in the stream's internal queue */
  [Symbol.asyncIterator](options?: ReadableStreamIteratorOptions): ReadableStreamAsyncIterator<R>;
  
  /** Creates a new ReadableStream wrapping the provided iterable or async iterable */
  static from<R>(asyncIterable: Iterable<R> | AsyncIterable<R> | ReadableStreamLike<R>): ReadableStream<R>;
}

import { ReadableStream } from "web-streams-polyfill";

// Create a simple readable stream
const stream = new ReadableStream({
  start(controller) {
    controller.enqueue("chunk 1");
    controller.enqueue("chunk 2");
    controller.close();
  }
});

// Read using async iteration
for await (const chunk of stream) {
  console.log(chunk);
}

// Create from an async iterable
async function* generateData() {
  yield "item 1";
  yield "item 2";
  yield "item 3";
}

const streamFromGenerator = ReadableStream.from(generateData());

/**
 * Default reader for readable streams
 */
class ReadableStreamDefaultReader<R> {
  constructor(stream: ReadableStream<R>);
  
  /** Promise that will be fulfilled when the stream becomes closed or rejected if the stream ever errors */
  readonly closed: Promise<undefined>;
  
  /** Cancel the associated stream */
  cancel(reason?: any): Promise<void>;
  
  /** Read the next chunk from the stream's internal queue */
  read(): Promise<ReadableStreamDefaultReadResult<R>>;
  
  /** Release the reader's lock on the corresponding stream */
  releaseLock(): void;
}

import { ReadableStream } from "web-streams-polyfill";

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue("Hello");
    controller.enqueue("World");
    controller.close();
  }
});

const reader = stream.getReader();

try {
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    console.log(value);
  }
} finally {
  reader.releaseLock();
}

/**
 * BYOB (Bring Your Own Buffer) reader for byte streams
 */
class ReadableStreamBYOBReader {
  constructor(stream: ReadableStream<Uint8Array>);
  
  /** Promise that will be fulfilled when the stream becomes closed or rejected if the stream ever errors */
  readonly closed: Promise<undefined>;
  
  /** Cancel the associated stream */
  cancel(reason?: any): Promise<void>;
  
  /** Read data into the provided buffer view */
  read<T extends ArrayBufferView>(
    view: T, 
    options?: { min?: number }
  ): Promise<ReadableStreamBYOBReadResult<T>>;
  
  /** Release the reader's lock on the corresponding stream */
  releaseLock(): void;
}

import { ReadableStream } from "web-streams-polyfill";

// Create a byte stream
const byteStream = new ReadableStream({
  type: 'bytes',
  start(controller) {
    const chunk = new Uint8Array([72, 101, 108, 108, 111]); // "Hello"
    controller.enqueue(chunk);
    controller.close();
  }
});

const reader = byteStream.getReader({ mode: 'byob' });
const buffer = new Uint8Array(1024);

try {
  const { done, value } = await reader.read(buffer);
  if (!done) {
    console.log(new TextDecoder().decode(value));
  }
} finally {
  reader.releaseLock();
}

/**
 * Controller for default readable streams
 */
class ReadableStreamDefaultController<R> {
  /** Returns the desired size to fill the controlled stream's internal queue */
  readonly desiredSize: number | null;
  
  /** Close the controlled readable stream */
  close(): void;
  
  /** Enqueue a chunk into the controlled readable stream */
  enqueue(chunk: R): void;
  
  /** Error the controlled readable stream */
  error(error: any): void;
}

/**
 * Controller for byte readable streams
 */
class ReadableByteStreamController {
  /** Returns the current BYOB pull request, or null if there isn't one */
  readonly byobRequest: ReadableStreamBYOBRequest | null;
  
  /** Returns the desired size to fill the controlled stream's internal queue */
  readonly desiredSize: number | null;
  
  /** Close the controlled readable stream */
  close(): void;
  
  /** Enqueue a chunk into the controlled readable stream */
  enqueue(chunk: ArrayBufferView): void;
  
  /** Error the controlled readable stream */
  error(error: any): void;
}

/**
 * Represents a BYOB (Bring Your Own Buffer) read request
 */
class ReadableStreamBYOBRequest {
  /** The ArrayBufferView to be filled, or null if the request was already responded to */
  readonly view: ArrayBufferView<ArrayBuffer> | null;
  
  /** Indicates to the associated stream that bytesWritten bytes were written into the view */
  respond(bytesWritten: number): void;
  
  /** Indicates to the associated stream that the request is being responded to with a new view */
  respondWithNewView(view: ArrayBufferView): void;
}

interface UnderlyingSource<R = any> {
  /** Called immediately during construction of the ReadableStream */
  start?: (controller: ReadableStreamDefaultController<R>) => void | PromiseLike<void>;
  
  /** Called when the stream's internal queue of chunks becomes not full */
  pull?: (controller: ReadableStreamDefaultController<R>) => void | PromiseLike<void>;
  
  /** Called when the stream is canceled */
  cancel?: (reason: any) => void | PromiseLike<void>;
  
  /** Must be undefined for default streams */
  type?: undefined;
}

interface UnderlyingByteSource {
  /** Called immediately during construction of the ReadableStream */
  start?: (controller: ReadableByteStreamController) => void | PromiseLike<void>;
  
  /** Called when the stream's internal queue of chunks becomes not full */
  pull?: (controller: ReadableByteStreamController) => void | PromiseLike<void>;
  
  /** Called when the stream is canceled */
  cancel?: (reason: any) => void | PromiseLike<void>;
  
  /** Must be 'bytes' for byte streams */
  type: 'bytes';
  
  /** Size of chunks to automatically allocate when using default readers */
  autoAllocateChunkSize?: number;
}

type ReadableStreamDefaultReadResult<T> = 
  | { done: false; value: T }
  | { done: true; value: undefined };

type ReadableStreamBYOBReadResult<T extends ArrayBufferView> = 
  | { done: false; value: T }
  | { done: true; value: T | undefined };

interface ReadableStreamIteratorOptions {
  /** Prevent canceling the stream when the iterator's return() method is called */
  preventCancel?: boolean;
}

interface ReadableStreamAsyncIterator<R> extends AsyncIterableIterator<R> {
  next(): Promise<IteratorResult<R, undefined>>;
  return(value?: any): Promise<IteratorResult<any>>;
}

interface ReadableStreamLike<R = any> {
  readonly locked: boolean;
  getReader(): ReadableStreamDefaultReaderLike<R>;
}

interface ReadableStreamDefaultReaderLike<R = any> {
  readonly closed: Promise<undefined>;
  cancel(reason?: any): Promise<void>;
  read(): Promise<ReadableStreamDefaultReadResult<R>>;
  releaseLock(): void;
}

type ReadableStreamReader<R> = ReadableStreamDefaultReader<R> | ReadableStreamBYOBReader;

interface ReadableStreamGetReaderOptions {
  /** The type of reader to create ('byob' for BYOB reader, undefined for default reader) */
  mode?: 'byob';
}

interface ReadableStreamBYOBReaderReadOptions {
  /** Minimum number of bytes to read */
  min?: number;
}

type UnderlyingSourceStartCallback<R> = (
  controller: ReadableStreamDefaultController<R>
) => void | PromiseLike<void>;

type UnderlyingSourcePullCallback<R> = (
  controller: ReadableStreamDefaultController<R>
) => void | PromiseLike<void>;

type UnderlyingSourceCancelCallback = (reason: any) => void | PromiseLike<void>;

type UnderlyingByteSourceStartCallback = (
  controller: ReadableByteStreamController
) => void | PromiseLike<void>;

type UnderlyingByteSourcePullCallback = (
  controller: ReadableByteStreamController
) => void | PromiseLike<void>;