feat(stream): add WrapWritableStream to Consumable and MaybeConsumable (1d3d3c88) · Commits · e / tools / webadb

libraries/stream-extra/src/consumable.ts

+19 −151

Original line number	Diff line number	Diff line
		import { PromiseResolver, isPromiseLike } from "@yume-chan/async";

		import type {
		QueuingStrategy,
		WritableStreamDefaultController,
		WritableStreamDefaultWriter,
		} from "./stream.js";
		ConsumableReadableStreamController,
		ConsumableReadableStreamSource,
		ConsumableWritableStreamSink,
		} from "./consumable/index.js";
		import {
		ReadableStream as NativeReadableStream,
		WritableStream as NativeWritableStream,
		} from "./stream.js";
		ConsumableReadableStream,
		ConsumableWrapWritableStream,
		ConsumableWritableStream,
		} from "./consumable/index.js";
		import type { Task } from "./task.js";
		import { createTask } from "./task.js";

		// Workaround https://github.com/evanw/esbuild/issues/3923
		class WritableStream<in T> extends NativeWritableStream<Consumable<T>> {
		static async write<T>(
		writer: WritableStreamDefaultWriter<Consumable<T>>,
		value: T,
		) {
		const consumable = new Consumable(value);
		await writer.write(consumable);
		await consumable.consumed;
		}

		constructor(
		sink: Consumable.WritableStreamSink<T>,
		strategy?: QueuingStrategy<T>,
		) {
		let wrappedStrategy: QueuingStrategy<Consumable<T>> \| undefined;
		if (strategy) {
		wrappedStrategy = {};
		if ("highWaterMark" in strategy) {
		wrappedStrategy.highWaterMark = strategy.highWaterMark;
		}
		if ("size" in strategy) {
		wrappedStrategy.size = (chunk) => {
		return strategy.size!(
		chunk instanceof Consumable ? chunk.value : chunk,
		);
		};
		}
		}

		super(
		{
		start(controller) {
		return sink.start?.(controller);
		},
		async write(chunk, controller) {
		await chunk.tryConsume((chunk) =>
		sink.write?.(chunk, controller),
		);
		},
		abort(reason) {
		return sink.abort?.(reason);
		},
		close() {
		return sink.close?.();
		},
		},
		wrappedStrategy,
		);
		}
		}

		class ReadableStream<T> extends NativeReadableStream<Consumable<T>> {
		static async enqueue<T>(
		controller: { enqueue: (chunk: Consumable<T>) => void },
		chunk: T,
		) {
		const output = new Consumable(chunk);
		controller.enqueue(output);
		await output.consumed;
		}

		constructor(
		source: Consumable.ReadableStreamSource<T>,
		strategy?: QueuingStrategy<T>,
		) {
		let wrappedController:
		\| Consumable.ReadableStreamController<T>
		\| undefined;

		let wrappedStrategy: QueuingStrategy<Consumable<T>> \| undefined;
		if (strategy) {
		wrappedStrategy = {};
		if ("highWaterMark" in strategy) {
		wrappedStrategy.highWaterMark = strategy.highWaterMark;
		}
		if ("size" in strategy) {
		wrappedStrategy.size = (chunk) => {
		return strategy.size!(chunk.value);
		};
		}
		}

		super(
		{
		async start(controller) {
		wrappedController = {
		async enqueue(chunk) {
		await ReadableStream.enqueue(controller, chunk);
		},
		close() {
		controller.close();
		},
		error(reason) {
		controller.error(reason);
		},
		};

		await source.start?.(wrappedController);
		},
		async pull() {
		await source.pull?.(wrappedController!);
		},
		async cancel(reason) {
		await source.cancel?.(reason);
		},
		},
		wrappedStrategy,
		);
		}
		}

		export class Consumable<T> {
		static readonly WritableStream = WritableStream;

		static readonly ReadableStream = ReadableStream;
		static readonly WritableStream = ConsumableWritableStream;
		static readonly WrapWritableStream = ConsumableWrapWritableStream;
		static readonly ReadableStream = ConsumableReadableStream;

		readonly #task: Task;
		readonly #resolver: PromiseResolver<void>;
		@@ -176,35 +65,14 @@ export class Consumable<T> {
		}

		export namespace Consumable {
		export interface WritableStreamSink<in T> {
		start?(
		controller: WritableStreamDefaultController,
		): void \| PromiseLike<void>;
		write?(
		chunk: T,
		controller: WritableStreamDefaultController,
		): void \| PromiseLike<void>;
		abort?(reason: unknown): void \| PromiseLike<void>;
		close?(): void \| PromiseLike<void>;
		}
		export type WritableStreamSink<T> = ConsumableWritableStreamSink<T>;
		export type WritableStream<in T> = typeof ConsumableWritableStream<T>;

		export type WritableStream<in T> = typeof Consumable.WritableStream<T>;

		export interface ReadableStreamController<T> {
		enqueue(chunk: T): Promise<void>;
		close(): void;
		error(reason: unknown): void;
		}

		export interface ReadableStreamSource<T> {
		start?(
		controller: ReadableStreamController<T>,
		): void \| PromiseLike<void>;
		pull?(
		controller: ReadableStreamController<T>,
		): void \| PromiseLike<void>;
		cancel?(reason: unknown): void \| PromiseLike<void>;
		}
		export type WrapWritableStream<in T> =
		typeof ConsumableWrapWritableStream<T>;

		export type ReadableStream<T> = typeof Consumable.ReadableStream<T>;
		export type ReadableStreamController<T> =
		ConsumableReadableStreamController<T>;
		export type ReadableStreamSource<T> = ConsumableReadableStreamSource<T>;
		export type ReadableStream<T> = typeof ConsumableReadableStream<T>;
		}

libraries/stream-extra/src/consumable/index.ts

0 → 100644

+3 −0

Original line number	Diff line number	Diff line
		export * from "./readable.js";
		export * from "./wrap-writable.js";
		export * from "./writable.js";

libraries/stream-extra/src/consumable/readable.ts

0 → 100644

+80 −0

Original line number	Diff line number	Diff line
		import { Consumable } from "../consumable.js";
		import type { QueuingStrategy } from "../stream.js";
		import { ReadableStream } from "../stream.js";

		export interface ConsumableReadableStreamController<T> {
		enqueue(chunk: T): Promise<void>;
		close(): void;
		error(reason: unknown): void;
		}

		export interface ConsumableReadableStreamSource<T> {
		start?(
		controller: ConsumableReadableStreamController<T>,
		): void \| PromiseLike<void>;
		pull?(
		controller: ConsumableReadableStreamController<T>,
		): void \| PromiseLike<void>;
		cancel?(reason: unknown): void \| PromiseLike<void>;
		}

		export class ConsumableReadableStream<T> extends ReadableStream<Consumable<T>> {
		static async enqueue<T>(
		controller: { enqueue: (chunk: Consumable<T>) => void },
		chunk: T,
		) {
		const output = new Consumable(chunk);
		controller.enqueue(output);
		await output.consumed;
		}

		constructor(
		source: ConsumableReadableStreamSource<T>,
		strategy?: QueuingStrategy<T>,
		) {
		let wrappedController!: ConsumableReadableStreamController<T>;

		let wrappedStrategy: QueuingStrategy<Consumable<T>> \| undefined;
		if (strategy) {
		wrappedStrategy = {};
		if ("highWaterMark" in strategy) {
		wrappedStrategy.highWaterMark = strategy.highWaterMark;
		}
		if ("size" in strategy) {
		wrappedStrategy.size = (chunk) => {
		return strategy.size!(chunk.value);
		};
		}
		}

		super(
		{
		start(controller) {
		wrappedController = {
		enqueue(chunk) {
		return ConsumableReadableStream.enqueue(
		controller,
		chunk,
		);
		},
		close() {
		controller.close();
		},
		error(reason) {
		controller.error(reason);
		},
		};

		return source.start?.(wrappedController);
		},
		pull() {
		return source.pull?.(wrappedController);
		},
		cancel(reason) {
		return source.cancel?.(reason);
		},
		},
		wrappedStrategy,
		);
		}
		}

libraries/stream-extra/src/consumable/wrap-writable.spec.ts

0 → 100644

+41 −0

Original line number	Diff line number	Diff line
		import * as assert from "node:assert";
		import { describe, it } from "node:test";

		describe("Consumable", () => {
		describe("WritableStream", () => {
		it("should not pause the source stream while piping", async () => {
		let step = 0;

		const stream = new WritableStream<string>({
		write(chunk) {
		switch (step) {
		case 2:
		assert.strictEqual(chunk, "a");
		step += 1;
		break;
		case 3:
		assert.strictEqual(chunk, "b");
		step += 1;
		break;
		}
		},
		});

		const readable = new ReadableStream<string>({
		start(controller) {
		controller.enqueue("a");
		assert.strictEqual(step, 0);
		step += 1;

		controller.enqueue("b");
		assert.strictEqual(step, 1);
		step += 1;

		controller.close();
		},
		});

		await readable.pipeTo(stream);
		});
		});
		});

libraries/stream-extra/src/consumable/wrap-writable.ts

0 → 100644

+21 −0

Original line number	Diff line number	Diff line
		import type { Consumable } from "../consumable.js";
		import { WritableStream } from "../stream.js";

		export class ConsumableWrapWritableStream<in T> extends WritableStream<
		Consumable<T>
		> {
		constructor(stream: WritableStream<T>) {
		const writer = stream.getWriter();
		super({
		write(chunk) {
		return chunk.tryConsume((chunk) => writer.write(chunk));
		},
		abort(reason) {
		return writer.abort(reason);
		},
		close() {
		return writer.close();
		},
		});
		}
		}