Class DuplexStreamFactory

libraries/stream-extra/src/duplex.ts:50–143

 * when any of them is closed, all other streams will be closed as well.
 */
export class DuplexStreamFactory<R, W> {
    #readableControllers: ReadableStreamDefaultController<R>[] = [];
    #writers: WritableStreamDefaultWriter<W>[] = [];

    #writableClosed = false;
    get writableClosed() {
        return this.#writableClosed;
    }

    #closed = new PromiseResolver<void>();
    get closed() {
        return this.#closed.promise;
    }

    readonly #options: DuplexStreamFactoryOptions;

    constructor(options?: DuplexStreamFactoryOptions) {
        this.#options = options ?? {};
    }

    wrapReadable(
        readable: ReadableStream<R>,
        strategy?: QueuingStrategy<R>,
    ): WrapReadableStream<R> {
        return new WrapReadableStream<R>(
            {
                start: (controller) => {
                    this.#readableControllers.push(controller);
                    return readable;
                },
                cancel: async () => {
                    // cancel means the local peer wants to close the connection.
                    await this.close();
                },
                close: async () => {
                    // stream end means the remote peer closed the connection first.
                    await this.dispose();
                },
            },
            strategy,
        );
    }

    createWritable(stream: WritableStream<W>): WritableStream<W> {
        const writer = stream.getWriter();
        this.#writers.push(writer);

        // `WritableStream` has no way to tell if the remote peer has closed the connection.
        // So it only triggers `close`.
        return new WritableStream<W>({
            write: async (chunk) => {
                await writer.write(chunk);
            },
            abort: async (reason) => {
                await writer.abort(reason);
                await this.close();
            },
            close: async () => {