Web Streams API

May 6, 2023

MDN: https://developer.mozilla.org/docs/Web/API/Streams_API
WHATWG: https://streams.spec.whatwg.org/

ReadableStream

interface ReadableStream<R = any> {
    // Returns a boolean indicating whether or not the readable stream is locked to a reader.
    readonly locked: boolean;
    // Returns a Promise that resolves when the stream is canceled. Calling this method signals a loss of interest in the stream by a consumer. The supplied reason argument will be given to the underlying source, which may or may not use it.
    cancel(reason?: any): Promise<void>;
    // Creates a reader and locks the stream to it. While the stream is locked, no other reader can be acquired until this one is released.
    getReader(options: { mode: "byob" }): ReadableStreamBYOBReader;
    getReader(): ReadableStreamDefaultReader<R>;
    getReader(options?: ReadableStreamGetReaderOptions): ReadableStreamReader<R>;
    // Provides a chainable way of piping the current stream through a transform stream or any other writable/readable pair.
    pipeThrough<T>(transform: ReadableWritablePair<T, R>, options?: StreamPipeOptions): ReadableStream<T>;
    // Pipes the current ReadableStream to a given WritableStream and returns a Promise that fulfills when the piping process completes successfully, or rejects if any errors were encountered.
    pipeTo(destination: WritableStream<R>, options?: StreamPipeOptions): Promise<void>;
    // The tee method tees this readable stream, returning a two-element array containing the two resulting branches as new ReadableStream instances. Each of those streams receives the same incoming data.
    tee(): [ReadableStream<R>, ReadableStream<R>];
}

declare var ReadableStream: {
    prototype: ReadableStream;
    new (underlyingSource: UnderlyingByteSource, strategy?: { highWaterMark?: number }): ReadableStream<Uint8Array>;
    new <R = any>(underlyingSource: UnderlyingDefaultSource<R>, strategy?: QueuingStrategy<R>): ReadableStream<R>;
    new <R = any>(underlyingSource?: UnderlyingSource<R>, strategy?: QueuingStrategy<R>): ReadableStream<R>;
};

读取， getReader

目前只有两种 reader

reader 可通过异步迭代器读取，参见此处

ReadableStreamDefaultReader

const reader = readableStream.getReader();
const { done, value } = await reader.read();

定义

declare var ReadableStreamDefaultReader: {
    prototype: ReadableStreamDefaultReader;
    new <R = any>(stream: ReadableStream<R>): ReadableStreamDefaultReader<R>;
};

interface ReadableStreamDefaultReader<R = any> extends ReadableStreamGenericReader {
    read(): Promise<ReadableStreamReadResult<R>>;
    releaseLock(): void;
}

type ReadableStreamReadResult<T> = ReadableStreamReadValueResult<T> | ReadableStreamReadDoneResult<T>;

interface ReadableStreamReadDoneResult<T> {
    done: true;
    value?: T;
}

interface ReadableStreamReadValueResult<T> {
    done: false;
    value: T;
}

ReadableStreamBYOBReader

BYOB: bing you own data 即自带缓冲区

const reader = readableStream.getReader({ mode: "byob" });

迭代

ReadableStream 实现了异步可迭代协议

const response = await fetch("https://www.example.org");
const readableStream: ReadableStream<Uint8Array> = response.body;
let total = 0;
for await (const chunk of readableStream) total += chunk.length;
console.log(total);

构造 ReadableStream， UnderlyingSource

https://developer.mozilla.org/docs/Web/API/ReadableStream/ReadableStream

interface UnderlyingDefaultSource<R = any> {
    cancel?: UnderlyingSourceCancelCallback;
    pull?: (controller: ReadableStreamDefaultController<R>) => void | PromiseLike<void>;
    start?: (controller: ReadableStreamDefaultController<R>) => any;
    type?: undefined;
}

interface UnderlyingSourceCancelCallback {
    (reason?: any): void | PromiseLike<void>;
}

interface ReadableStreamDefaultController<R = any> {
    // the desired size required to fill the stream's internal queue.
    readonly desiredSize: number | null;
    // Closes the associated stream.
    close(): void;
    // Enqueues a given chunk in the associated stream.
    enqueue(chunk?: R): void;
    // Causes any future interactions with the associated stream to error.
    error(e?: any): void;
}

declare var ReadableStreamDefaultController: {
    prototype: ReadableStreamDefaultController;
    new (): ReadableStreamDefaultController;
};

演示：

let interval: number;

const underlyingSource: UnderlyingDefaultSource<string> = {
    start(controller: ReadableStreamDefaultController<string>) {
        let counter = 0;

        interval = setInterval(() => {
            // 随机生成一些字符串
            let string = Math.random().toString().slice(2);

            // Add the string to the stream
            controller.enqueue(string);

            if (++counter >= 10) {
                clearInterval(interval);
                controller.close();
            }
        }, 300);
    },
    pull(controller) {
        // We don't really need a pull in this example
    },
    cancel() {
        // This is called if the reader cancels,
        // so we should stop generating strings
        clearInterval(interval);
        // 在本演示中是可选的，因为调用controller.close之前已经clearInterval了
    },
};

const stream = new ReadableStream<string>(underlyingSource);

for await (const chunk of stream) {
    console.log(chunk);
}

WriteableStream

interface WritableStream<W = any> {
    // A boolean indicating whether the WritableStream is locked to a writer.
    readonly locked: boolean;
    // Aborts the stream, signaling that the producer can no longer successfully write to the stream and it is to be immediately moved to an error state, with any queued writes discarded.
    abort(reason?: any): Promise<void>;
    // Closes the stream.
    close(): Promise<void>;
    // Returns a new instance of WritableStreamDefaultWriter and locks the stream to that instance. While the stream is locked, no other writer can be acquired until this one is released.
    getWriter(): WritableStreamDefaultWriter<W>;
}

declare var WritableStream: {
    prototype: WritableStream;
    new <W = any>(underlyingSink?: UnderlyingSink<W>, strategy?: QueuingStrategy<W>): WritableStream<W>;
};

写入， getWriter

https://developer.mozilla.org/docs/Web/API/WritableStreamDefaultWriter

目前只有 WritableStreamDefaultWriter，直接调用 getWriter 方法获取

const writer = writeableStream.getWriter();
await writer.write(chunk);

定义：

interface WritableStreamDefaultWriter<W = any> {
    readonly closed: Promise<undefined>;
    readonly desiredSize: number | null;
    readonly ready: Promise<undefined>;
    abort(reason?: any): Promise<void>;
    close(): Promise<void>;
    releaseLock(): void;
    write(chunk?: W): Promise<void>;
}

declare var WritableStreamDefaultWriter: {
    prototype: WritableStreamDefaultWriter;
    new <W = any>(stream: WritableStream<W>): WritableStreamDefaultWriter<W>;
};

构造 WritableStream， UnderlyingSink

https://developer.mozilla.org/docs/Web/API/WritableStream/WritableStream

interface UnderlyingByteSource {
    autoAllocateChunkSize?: number;
    cancel?: UnderlyingSourceCancelCallback;
    pull?: (controller: ReadableByteStreamController) => void | PromiseLike<void>;
    start?: (controller: ReadableByteStreamController) => any;
    type: "bytes";
}

interface UnderlyingDefaultSource<R = any> {
    cancel?: UnderlyingSourceCancelCallback;
    pull?: (controller: ReadableStreamDefaultController<R>) => void | PromiseLike<void>;
    start?: (controller: ReadableStreamDefaultController<R>) => any;
    type?: undefined;
}

interface UnderlyingSink<W = any> {
    abort?: UnderlyingSinkAbortCallback;
    close?: UnderlyingSinkCloseCallback;
    start?: UnderlyingSinkStartCallback;
    type?: undefined;
    write?: UnderlyingSinkWriteCallback<W>;
}

演示：

let result = "";

const writableStream = new WritableStream(
    {
        // Implement the sink
        write(chunk) {
            const buffer = new ArrayBuffer(1);
            const view = new Uint8Array(buffer);
            view[0] = chunk;
            const decoded = new TextDecoder("utf-8").decode(view, { stream: true });

            console.log(`[WritableStream write()]: ${decoded}`);
            result += decoded;
        },
        close() {
            console.log(`[WritableStream close()]: ${result}`);
        },
        abort(err) {
            console.log("Sink error:", err);
        },
    },
    new CountQueuingStrategy({ highWaterMark: 1 }),
);

async function sendMessage(message: string, writableStream: WritableStream) {
    // defaultWriter is of type WritableStreamDefaultWriter
    const defaultWriter = writableStream.getWriter();
    const encoded = new TextEncoder().encode(message);

    try {
        for (const chunk of encoded) {
            await defaultWriter.ready;
            await defaultWriter.write(chunk);
            console.log("Chunk written to sink.");
        }
        // Call ready again to ensure that all chunks are written
        // before closing the writer.
        await defaultWriter.ready;
        await defaultWriter.close();
        console.log("All chunks written");
    } catch (err) {
        console.log("Error:", err);
    }
}

sendMessage("Hello, world.", writableStream);

TransformStream

interface TransformStream<I = any, O = any> {
    readonly readable: ReadableStream<O>;
    readonly writable: WritableStream<I>;
}

declare var TransformStream: {
    prototype: TransformStream;
    new <I = any, O = any>(
        transformer?: Transformer<I, O>,
        writableStrategy?: QueuingStrategy<I>,
        readableStrategy?: QueuingStrategy<O>,
    ): TransformStream<I, O>;
};

interface TransformStreamDefaultController<O = any> {
    readonly desiredSize: number | null;
    enqueue(chunk?: O): void;
    error(reason?: any): void;
    terminate(): void;
}

declare var TransformStreamDefaultController: {
    prototype: TransformStreamDefaultController;
    new (): TransformStreamDefaultController;
};

QueuingStrategy

指定如何（自动）将数据块拆分为大小不等的块。

参见：背压（backpressure）机制

ByteLengthQueuingStrategy

ByteLengthQueuingStrategy 构造函数接受一个参数 chunkSize。表示传输中的数据块最大字节数，即，给定一个块流，在施加背压之前，这些块中可以包含多少字节。
当传输的数据块大小超过 chunkSize 时，ReadableStream 会自动将其拆分为大小不超过 chunkSize 的块。

该策略通常用于处理二进制数据，例如音频或视频流。在这种情况下，数据块的大小往往是不确定的，因此使用 ByteLengthQueuingStrategy 可以更好地平衡传输速度和内存使用。

CountQueuingStrategy

CountQueuingStrategy 构造函数接受一个参数 highWaterMark。表示给定一个块流，在施加背压之前，可以包含在内部队列中的块的总数。
当队列中的数据块数量超过 highWaterMark 时，ReadableStream 将停止读取数据，直到队列中的数据量下降到 highWaterMark 以下。

该策略通常用于处理文本数据，例如日志文件或 JSON 数据。在这种情况下，数据块的大小可能是固定的，但是数据块的数量可能会非常大，因此使用 CountQueuingStrategy 可更好地控制内存使用和传输速度。

需要注意的是，CountQueuingStrategy 的 highWaterMark 参数是一个建议值，实际上可能会被忽略，具体取决于底层实现。

另见

https://vercel.com/docs/functions/streaming/streaming-examples

←Back to home

This work is licensed under CC BY-NC-SA 4.0