refactor(stream): move streams to new package

.vscode/settings.json

@@ -67,5 +67,6 @@
             ],
             "url": "https://developer.microsoft.com/json-schemas/rush/v5/version-policies.schema.json"
         }
-    ]
+    ],
+    "typescript.preferences.quoteStyle": "single"
 }

common/config/rush/pnpm-lock.yaml

@@ -24,6 +24,7 @@ specifiers:
   '@rush-temp/demo': file:./projects/demo.tgz
   '@rush-temp/event': file:./projects/event.tgz
   '@rush-temp/scrcpy': file:./projects/scrcpy.tgz
+  '@rush-temp/stream-extra': file:./projects/stream-extra.tgz
   '@rush-temp/struct': file:./projects/struct.tgz
   '@rush-temp/ts-package-builder': file:./projects/ts-package-builder.tgz
   '@rush-temp/unofficial-adb-book': file:./projects/unofficial-adb-book.tgz
@@ -85,6 +86,7 @@ dependencies:
   '@rush-temp/demo': file:projects/demo.tgz_@mdx-js+react@1.6.22
   '@rush-temp/event': file:projects/event.tgz_@types+node@17.0.33
   '@rush-temp/scrcpy': file:projects/scrcpy.tgz_@types+node@17.0.33
+  '@rush-temp/stream-extra': file:projects/stream-extra.tgz_@types+node@17.0.33
   '@rush-temp/struct': file:projects/struct.tgz_@types+node@17.0.33
   '@rush-temp/ts-package-builder': file:projects/ts-package-builder.tgz
   '@rush-temp/unofficial-adb-book': file:projects/unofficial-adb-book.tgz_d45f1a34685929383f8ab73cab148e80
@@ -10692,7 +10694,7 @@ packages:
     dev: false
 
   /through/2.3.8:
-    resolution: {integrity: sha1-DdTJ/6q8NXlgsbckEV1+Doai4fU=}
+    resolution: {integrity: sha512-w89qg7PI8wAdvX60bMDP+bFoD5Dvhm9oLheFp5O4a2QF0cSBGsBX4qZmadPMvVqlLJBBci+WqGGOAPvcDeNSVg==}
     dev: false
 
   /thunky/1.1.0:
@@ -10700,7 +10702,7 @@ packages:
     dev: false
 
   /timed-out/4.0.1:
-    resolution: {integrity: sha1-8y6srFoXW+ol1/q1Zas+2HQe9W8=}
+    resolution: {integrity: sha512-G7r3AhovYtr5YKOWQkta8RKAPb+J9IsO4uVmzjl8AZwfhs8UcUwTiD6gcJYSgOtzyjvQKrKYn41syHbUWMkafA==}
     engines: {node: '>=0.10.0'}
     dev: false
 
@@ -10762,7 +10764,7 @@ packages:
     dev: false
 
   /trim-repeated/1.0.0:
-    resolution: {integrity: sha1-42RqLqTokTEr9+rObPsFOAvAHCE=}
+    resolution: {integrity: sha512-pkonvlKk8/ZuR0D5tLW8ljt5I8kmxp2XKymhepUeOdCEfKpZaktSArkLHZt76OB1ZvO9bssUsDty4SWhLvZpLg==}
     engines: {node: '>=0.10.0'}
     dependencies:
       escape-string-regexp: 1.0.5
@@ -11858,6 +11860,31 @@ packages:
       - ts-node
     dev: false
 
+  file:projects/stream-extra.tgz_@types+node@17.0.33:
+    resolution: {integrity: sha512-rJ7QINFBmWnprVVBMtJ4JF8i51LLk1oPvLLiVG642Y/LstYOFULaDJt1lytyBQSfWynM79Hv8JN+ZKaXcQDBhg==, tarball: file:projects/stream-extra.tgz}
+    id: file:projects/stream-extra.tgz
+    name: '@rush-temp/stream-extra'
+    version: 0.0.0
+    dependencies:
+      '@jest/globals': 28.1.0
+      '@yume-chan/async': 2.1.4
+      cross-env: 7.0.3
+      jest: 28.1.0_@types+node@17.0.33
+      ts-jest: 28.0.2_jest@28.1.0+typescript@4.7.2
+      tslib: 2.4.0
+      typescript: 4.7.2
+      web-streams-polyfill: 4.0.0-beta.3
+    transitivePeerDependencies:
+      - '@babel/core'
+      - '@types/jest'
+      - '@types/node'
+      - babel-jest
+      - esbuild
+      - node-notifier
+      - supports-color
+      - ts-node
+    dev: false
+
   file:projects/struct.tgz_@types+node@17.0.33:
     resolution: {integrity: sha512-N4tRna6p02qJC7klmbbeELeqARWvpb3GXHjZRRi5DzTumGngXNQyzYpG3VCNZnmZV9a4vgDVJX/CZOiOKbOKMQ==, tarball: file:projects/struct.tgz}
     id: file:projects/struct.tgz

common/config/rush/repo-state.json

@@ -1,4 +1,4 @@
 // DO NOT MODIFY THIS FILE MANUALLY BUT DO COMMIT IT. It is generated and used by Rush.
 {
-  "pnpmShrinkwrapHash": "3798fc5f6f6c673b1888e77923d3eeace5651923"
+  "pnpmShrinkwrapHash": "bc1c6912e108986d20555db6e1b3427c76999db1"
 }

libraries/adb-backend-direct-sockets/package.json

@@ -36,6 +36,7 @@
     },
     "dependencies": {
         "@yume-chan/adb": "^0.0.16",
+        "@yume-chan/stream-extra": "^0.0.16",
         "tslib": "^2.3.1"
     }
 }

libraries/adb-backend-direct-sockets/src/index.ts

@@ -1,4 +1,5 @@
-import { AdbBackend, AdbPacket, AdbPacketSerializeStream, pipeFrom, ReadableStream, StructDeserializeStream, WrapReadableStream, WrapWritableStream, WritableStream } from '@yume-chan/adb';
+import { AdbBackend, AdbPacket, AdbPacketSerializeStream } from '@yume-chan/adb';
+import { pipeFrom, ReadableStream, StructDeserializeStream, WrapReadableStream, WrapWritableStream, WritableStream } from '@yume-chan/stream-extra';
 
 declare global {
     interface TCPSocket {

libraries/adb-backend-webusb/package.json

@@ -33,6 +33,7 @@
     "dependencies": {
         "@types/w3c-web-usb": "^1.0.4",
         "@yume-chan/adb": "^0.0.16",
+        "@yume-chan/stream-extra": "^0.0.16",
         "@yume-chan/struct": "^0.0.16",
         "tslib": "^2.3.1"
     },

libraries/adb-backend-webusb/src/backend.ts

@@ -1,5 +1,6 @@
-import { AdbPacketHeader, AdbPacketSerializeStream, DuplexStreamFactory, pipeFrom, ReadableStream, WritableStream, type AdbBackend, type AdbPacketData, type AdbPacketInit, type ReadableWritablePair } from '@yume-chan/adb';
+import { AdbPacketHeader, AdbPacketSerializeStream, type AdbBackend, type AdbPacketData, type AdbPacketInit } from '@yume-chan/adb';
-import { EMPTY_UINT8_ARRAY, StructDeserializeStream } from "@yume-chan/struct";
+import { DuplexStreamFactory, pipeFrom, ReadableStream, WritableStream } from '@yume-chan/stream-extra';
+import { EMPTY_UINT8_ARRAY, StructDeserializeStream } from '@yume-chan/struct';
 
 export const ADB_DEVICE_FILTER: USBDeviceFilter = {
     classCode: 0xFF,
@@ -173,6 +174,6 @@ export class AdbWebUsbBackend implements AdbBackend {
             }
         }
 
-        throw new Error('Unknown error');
+        throw new Error('Can not find ADB interface');
     }
 }

libraries/adb-backend-ws/package.json

@@ -36,6 +36,7 @@
     },
     "dependencies": {
         "@yume-chan/adb": "^0.0.16",
+        "@yume-chan/stream-extra": "^0.0.16",
         "tslib": "^2.3.1"
     }
 }

libraries/adb-backend-ws/src/index.ts

@@ -1,4 +1,5 @@
-import { AdbPacket, AdbPacketSerializeStream, DuplexStreamFactory, pipeFrom, ReadableStream, StructDeserializeStream, WritableStream, type AdbBackend } from '@yume-chan/adb';
+import { AdbPacket, AdbPacketSerializeStream, type AdbBackend } from '@yume-chan/adb';
+import { DuplexStreamFactory, pipeFrom, ReadableStream, StructDeserializeStream, WritableStream } from '@yume-chan/stream-extra';
 
 export default class AdbWsBackend implements AdbBackend {
     public readonly serial: string;
@@ -12,7 +13,7 @@ export default class AdbWsBackend implements AdbBackend {
 
     public async connect() {
         const socket = new WebSocket(this.serial);
-        socket.binaryType = "arraybuffer";
+        socket.binaryType = 'arraybuffer';
 
         await new Promise((resolve, reject) => {
             socket.onopen = resolve;

libraries/adb/package.json

@@ -34,6 +34,7 @@
         "@yume-chan/async": "^2.1.4",
         "@yume-chan/dataview-bigint-polyfill": "^0.0.16",
         "@yume-chan/event": "^0.0.16",
+        "@yume-chan/stream-extra": "^0.0.16",
         "@yume-chan/struct": "^0.0.16",
         "tslib": "^2.3.1",
         "web-streams-polyfill": "^4.0.0-beta.3"

libraries/adb/src/adb.ts

@@ -1,13 +1,14 @@
 // cspell: ignore libusb
 
 import { PromiseResolver } from '@yume-chan/async';
+import { AbortController, DecodeUtf8Stream, GatherStringStream, WritableStream, type ReadableWritablePair } from '@yume-chan/stream-extra';
+
 import { AdbAuthenticationProcessor, ADB_DEFAULT_AUTHENTICATORS, type AdbCredentialStore } from './auth.js';
 import { AdbPower, AdbReverseCommand, AdbSubprocess, AdbSync, AdbTcpIpCommand, escapeArg, framebuffer, install, type AdbFrameBuffer } from './commands/index.js';
 import { AdbFeatures } from './features.js';
 import { AdbCommand, calculateChecksum, type AdbPacketData, type AdbPacketInit } from './packet.js';
 import { AdbIncomingSocketHandler, AdbPacketDispatcher, type AdbSocket, type Closeable } from './socket/index.js';
-import { AbortController, DecodeUtf8Stream, GatherStringStream, WritableStream, type ReadableWritablePair } from "./stream/index.js";
+import { decodeUtf8, encodeUtf8 } from './utils/index.js';
-import { decodeUtf8, encodeUtf8 } from "./utils/index.js";
 
 export enum AdbPropKey {
     Product = 'ro.product.name',

libraries/adb/src/auth.ts

@@ -1,6 +1,7 @@
 import { PromiseResolver } from '@yume-chan/async';
 import type { Disposable } from '@yume-chan/event';
 import type { ValueOrPromise } from '@yume-chan/struct';
+
 import { calculatePublicKey, calculatePublicKeyLength, sign } from './crypto.js';
 import { AdbCommand, type AdbPacketData } from './packet.js';
 import { calculateBase64EncodedLength, encodeBase64 } from './utils/index.js';

libraries/adb/src/backend.ts

@@ -1,6 +1,7 @@
+import type { ReadableWritablePair } from '@yume-chan/stream-extra';
 import type { ValueOrPromise } from '@yume-chan/struct';
-import type { AdbPacketData, AdbPacketInit } from "./packet.js";
+
-import type { ReadableWritablePair } from "./stream/index.js";
+import type { AdbPacketData, AdbPacketInit } from './packet.js';
 
 export interface AdbBackend {
     readonly serial: string;

libraries/adb/src/commands/base.ts

@@ -1,4 +1,5 @@
 import { AutoDisposable } from '@yume-chan/event';
+
 import type { Adb } from '../adb.js';
 
 export class AdbCommandBase extends AutoDisposable {

libraries/adb/src/commands/framebuffer.ts

@@ -1,6 +1,7 @@
-import Struct from "@yume-chan/struct";
+import { BufferedStream } from '@yume-chan/stream-extra';
+import Struct from '@yume-chan/struct';
+
 import type { Adb } from '../adb.js';
-import { AdbBufferedStream } from '../stream/index.js';
 
 const Version =
     new Struct({ littleEndian: true })
@@ -60,7 +61,7 @@ export type AdbFrameBuffer = AdbFrameBufferV1 | AdbFrameBufferV2;
 
 export async function framebuffer(adb: Adb): Promise<AdbFrameBuffer> {
     const socket = await adb.createSocket('framebuffer:');
-    const stream = new AdbBufferedStream(socket);
+    const stream = new BufferedStream(socket.readable);
     const { version } = await Version.deserialize(stream);
     switch (version) {
         case 1:

libraries/adb/src/commands/install.ts

@@ -1,7 +1,8 @@
-import type { Adb } from "../adb.js";
+import { WrapWritableStream, WritableStream } from '@yume-chan/stream-extra';
-import { WrapWritableStream, WritableStream } from "../stream/index.js";
+
-import { escapeArg } from "./subprocess/index.js";
+import type { Adb } from '../adb.js';
-import type { AdbSync } from "./sync/index.js";
+import { escapeArg } from './subprocess/index.js';
+import type { AdbSync } from './sync/index.js';
 
 export function install(
     adb: Adb,

libraries/adb/src/commands/power.ts

@@ -3,7 +3,7 @@
 // cspell: ignore keyevent
 // cspell: ignore longpress
 
-import { AdbCommandBase } from "./base.js";
+import { AdbCommandBase } from './base.js';
 
 export class AdbPower extends AdbCommandBase {
     public reboot(name: string = '') {

libraries/adb/src/commands/reverse.ts

@@ -1,11 +1,12 @@
 // cspell: ignore killforward
 
 import { AutoDisposable } from '@yume-chan/event';
+import { BufferedStream, BufferedStreamEndedError } from '@yume-chan/stream-extra';
 import Struct from '@yume-chan/struct';
-import type { Adb } from "../adb.js";
+
+import type { Adb } from '../adb.js';
 import type { AdbIncomingSocketHandler, AdbSocket } from '../socket/index.js';
-import { AdbBufferedStream, BufferedStreamEndedError } from '../stream/index.js';
+import { decodeUtf8 } from '../utils/index.js';
-import { decodeUtf8 } from "../utils/index.js";
 
 export interface AdbForwardListener {
     deviceSerial: string;
@@ -52,7 +53,7 @@ export class AdbReverseCommand extends AutoDisposable {
 
     private async createBufferedStream(service: string) {
         const socket = await this.adb.createSocket(service);
-        return new AdbBufferedStream(socket);
+        return new BufferedStream(socket.readable);
     }
 
     private async sendRequest(service: string) {

libraries/adb/src/commands/subprocess/command.ts

@@ -0,0 +1,131 @@
+import { GatherStringStream, DecodeUtf8Stream } from '@yume-chan/stream-extra';
+
+import { AdbCommandBase } from '../base.js';
+import { AdbSubprocessNoneProtocol, AdbSubprocessProtocol, AdbSubprocessProtocolConstructor, AdbSubprocessShellProtocol } from './protocols/index.js';
+
+export interface AdbSubprocessOptions {
+    /**
+     * A list of `AdbSubprocessProtocolConstructor`s to be used.
+     *
+     * Different `AdbSubprocessProtocol` has different capabilities, thus requires specific adaptations.
+     * Check their documentations for details.
+     *
+     * The first protocol whose `isSupported` returns `true` will be used.
+     * If no `AdbSubprocessProtocol` is supported, an error will be thrown.
+     *
+     * @default [AdbSubprocessShellProtocol, AdbSubprocessNoneProtocol]
+     */
+    protocols: AdbSubprocessProtocolConstructor[];
+}
+
+const DEFAULT_OPTIONS: AdbSubprocessOptions = {
+    protocols: [AdbSubprocessShellProtocol, AdbSubprocessNoneProtocol],
+};
+
+export interface AdbSubprocessWaitResult {
+    stdout: string;
+    stderr: string;
+    exitCode: number;
+}
+
+export class AdbSubprocess extends AdbCommandBase {
+    private async createProtocol(
+        mode: 'pty' | 'raw',
+        command?: string | string[],
+        options?: Partial<AdbSubprocessOptions>
+    ): Promise<AdbSubprocessProtocol> {
+        const { protocols } = { ...DEFAULT_OPTIONS, ...options };
+
+        let Constructor: AdbSubprocessProtocolConstructor | undefined;
+        for (const item of protocols) {
+            // It's async so can't use `Array#find`
+            if (await item.isSupported(this.adb)) {
+                Constructor = item;
+                break;
+            }
+        }
+
+        if (!Constructor) {
+            throw new Error('No specified protocol is supported by the device');
+        }
+
+        if (Array.isArray(command)) {
+            command = command.join(' ');
+        } else if (command === undefined) {
+            // spawn the default shell
+            command = '';
+        }
+        return await Constructor[mode](this.adb, command);
+    }
+
+    /**
+     * Spawns an executable in PTY (interactive) mode.
+     * @param command The command to run. If omitted, the default shell will be spawned.
+     * @param options The options for creating the `AdbSubprocessProtocol`
+     * @returns A new `AdbSubprocessProtocol` instance connecting to the spawned process.
+     */
+    public shell(
+        command?: string | string[],
+        options?: Partial<AdbSubprocessOptions>
+    ): Promise<AdbSubprocessProtocol> {
+        return this.createProtocol('pty', command, options);
+    }
+
+    /**
+     * Spawns an executable and pipe the output.
+     * @param command The command to run, or an array of strings containing both command and args.
+     * @param options The options for creating the `AdbSubprocessProtocol`
+     * @returns A new `AdbSubprocessProtocol` instance connecting to the spawned process.
+     */
+    public spawn(
+        command: string | string[],
+        options?: Partial<AdbSubprocessOptions>
+    ): Promise<AdbSubprocessProtocol> {
+        return this.createProtocol('raw', command, options);
+    }
+
+    /**
+     * Spawns a new process, waits until it exits, and returns the entire output.
+     * @param command The command to run
+     * @param options The options for creating the `AdbSubprocessProtocol`
+     * @returns The entire output of the command
+     */
+    public async spawnAndWait(
+        command: string | string[],
+        options?: Partial<AdbSubprocessOptions>
+    ): Promise<AdbSubprocessWaitResult> {
+        const shell = await this.spawn(command, options);
+
+        const stdout = new GatherStringStream();
+        const stderr = new GatherStringStream();
+
+        const [, , exitCode] = await Promise.all([
+            shell.stdout
+                .pipeThrough(new DecodeUtf8Stream())
+                .pipeTo(stdout),
+            shell.stderr
+                .pipeThrough(new DecodeUtf8Stream())
+                .pipeTo(stderr),
+            shell.exit
+        ]);
+
+        return {
+            stdout: stdout.result,
+            stderr: stderr.result,
+            exitCode,
+        };
+    }
+
+    /**
+     * Spawns a new process, waits until it exits, and returns the entire output.
+     * @param command The command to run
+     * @returns The entire output of the command
+     */
+    public async spawnAndWaitLegacy(command: string | string[]): Promise<string> {
+        const { stdout } = await this.spawnAndWait(
+            command,
+            { protocols: [AdbSubprocessNoneProtocol] }
+        );
+        return stdout;
+    }
+}

libraries/adb/src/commands/subprocess/index.ts

@@ -1,139 +1,3 @@
-import type { Adb } from '../../adb.js';
+export * from './command.js';
-import { DecodeUtf8Stream, GatherStringStream } from "../../stream/index.js";
-import { AdbSubprocessNoneProtocol, AdbSubprocessShellProtocol, type AdbSubprocessProtocol, type AdbSubprocessProtocolConstructor } from './protocols/index.js';
-
 export * from './protocols/index.js';
 export * from './utils.js';
-
-export interface AdbSubprocessOptions {
-    /**
-     * A list of `AdbSubprocessProtocolConstructor`s to be used.
-     *
-     * Different `AdbSubprocessProtocol` has different capabilities, thus requires specific adaptations.
-     * Check their documentations for details.
-     *
-     * The first protocol whose `isSupported` returns `true` will be used.
-     * If no `AdbSubprocessProtocol` is supported, an error will be thrown.
-     *
-     * @default [AdbSubprocessShellProtocol, AdbSubprocessNoneProtocol]
-     */
-    protocols: AdbSubprocessProtocolConstructor[];
-}
-
-const DEFAULT_OPTIONS: AdbSubprocessOptions = {
-    protocols: [AdbSubprocessShellProtocol, AdbSubprocessNoneProtocol],
-};
-
-export interface AdbSubprocessWaitResult {
-    stdout: string;
-    stderr: string;
-    exitCode: number;
-}
-
-export class AdbSubprocess {
-    public readonly adb: Adb;
-
-    public constructor(adb: Adb) {
-        this.adb = adb;
-    }
-
-    private async createProtocol(
-        mode: 'pty' | 'raw',
-        command?: string | string[],
-        options?: Partial<AdbSubprocessOptions>
-    ): Promise<AdbSubprocessProtocol> {
-        const { protocols } = { ...DEFAULT_OPTIONS, ...options };
-
-        let Constructor: AdbSubprocessProtocolConstructor | undefined;
-        for (const item of protocols) {
-            // It's async so can't use `Array#find`
-            if (await item.isSupported(this.adb)) {
-                Constructor = item;
-                break;
-            }
-        }
-
-        if (!Constructor) {
-            throw new Error('No specified protocol is supported by the device');
-        }
-
-        if (Array.isArray(command)) {
-            command = command.join(' ');
-        } else if (command === undefined) {
-            // spawn the default shell
-            command = '';
-        }
-        return await Constructor[mode](this.adb, command);
-    }
-
-    /**
-     * Spawns an executable in PTY (interactive) mode.
-     * @param command The command to run. If omitted, the default shell will be spawned.
-     * @param options The options for creating the `AdbSubprocessProtocol`
-     * @returns A new `AdbSubprocessProtocol` instance connecting to the spawned process.
-     */
-    public shell(
-        command?: string | string[],
-        options?: Partial<AdbSubprocessOptions>
-    ): Promise<AdbSubprocessProtocol> {
-        return this.createProtocol('pty', command, options);
-    }
-
-    /**
-     * Spawns an executable and pipe the output.
-     * @param command The command to run, or an array of strings containing both command and args.
-     * @param options The options for creating the `AdbSubprocessProtocol`
-     * @returns A new `AdbSubprocessProtocol` instance connecting to the spawned process.
-     */
-    public spawn(
-        command: string | string[],
-        options?: Partial<AdbSubprocessOptions>
-    ): Promise<AdbSubprocessProtocol> {
-        return this.createProtocol('raw', command, options);
-    }
-
-    /**
-     * Spawns a new process, waits until it exits, and returns the entire output.
-     * @param command The command to run
-     * @param options The options for creating the `AdbSubprocessProtocol`
-     * @returns The entire output of the command
-     */
-    public async spawnAndWait(
-        command: string | string[],
-        options?: Partial<AdbSubprocessOptions>
-    ): Promise<AdbSubprocessWaitResult> {
-        const shell = await this.spawn(command, options);
-
-        const stdout = new GatherStringStream();
-        const stderr = new GatherStringStream();
-
-        const [, , exitCode] = await Promise.all([
-            shell.stdout
-                .pipeThrough(new DecodeUtf8Stream())
-                .pipeTo(stdout),
-            shell.stderr
-                .pipeThrough(new DecodeUtf8Stream())
-                .pipeTo(stderr),
-            shell.exit
-        ]);
-
-        return {
-            stdout: stdout.result,
-            stderr: stderr.result,
-            exitCode,
-        };
-    }
-
-    /**
-     * Spawns a new process, waits until it exits, and returns the entire output.
-     * @param command The command to run
-     * @returns The entire output of the command
-     */
-    public async spawnAndWaitLegacy(command: string | string[]): Promise<string> {
-        const { stdout } = await this.spawnAndWait(
-            command,
-            { protocols: [AdbSubprocessNoneProtocol] }
-        );
-        return stdout;
-    }
-}

libraries/adb/src/commands/subprocess/protocols/none.ts

@@ -1,7 +1,8 @@
-import type { Adb } from "../../../adb.js";
+import { DuplexStreamFactory, ReadableStream } from '@yume-chan/stream-extra';
-import type { AdbSocket } from "../../../socket/index.js";
+
-import { DuplexStreamFactory, ReadableStream } from "../../../stream/index.js";
+import type { Adb } from '../../../adb.js';
-import type { AdbSubprocessProtocol } from "./types.js";
+import type { AdbSocket } from '../../../socket/index.js';
+import type { AdbSubprocessProtocol } from './types.js';
 
 /**
  * The legacy shell

libraries/adb/src/commands/subprocess/protocols/shell.ts

@@ -1,11 +1,12 @@
-import { PromiseResolver } from "@yume-chan/async";
+import { PromiseResolver } from '@yume-chan/async';
-import Struct, { placeholder, type StructValueType } from "@yume-chan/struct";
+import { pipeFrom, PushReadableStream, StructDeserializeStream, StructSerializeStream, TransformStream, WritableStream, type WritableStreamDefaultWriter, type PushReadableStreamController, type ReadableStream } from '@yume-chan/stream-extra';
-import type { Adb } from "../../../adb.js";
+import Struct, { placeholder, type StructValueType } from '@yume-chan/struct';
-import { AdbFeatures } from "../../../features.js";
+
-import type { AdbSocket } from "../../../socket/index.js";
+import type { Adb } from '../../../adb.js';
-import { pipeFrom, PushReadableStream, ReadableStream, StructDeserializeStream, StructSerializeStream, TransformStream, WritableStream, WritableStreamDefaultWriter, type PushReadableStreamController } from "../../../stream/index.js";
+import { AdbFeatures } from '../../../features.js';
-import { encodeUtf8 } from "../../../utils/index.js";
+import type { AdbSocket } from '../../../socket/index.js';
-import type { AdbSubprocessProtocol } from "./types.js";
+import { encodeUtf8 } from '../../../utils/index.js';
+import type { AdbSubprocessProtocol } from './types.js';
 
 export enum AdbShellProtocolId {
     Stdin,

libraries/adb/src/commands/subprocess/protocols/types.ts

@@ -1,7 +1,8 @@
-import type { ValueOrPromise } from "@yume-chan/struct";
+import type { ReadableStream, WritableStream } from '@yume-chan/stream-extra';
-import type { Adb } from "../../../adb.js";
+import type { ValueOrPromise } from '@yume-chan/struct';
-import type { AdbSocket } from "../../../socket/index.js";
+
-import type { ReadableStream, WritableStream } from "../../../stream/index.js";
+import type { Adb } from '../../../adb.js';
+import type { AdbSocket } from '../../../socket/index.js';
 
 export interface AdbSubprocessProtocol {
     /**

libraries/adb/src/commands/sync/index.ts

@@ -1,7 +1,7 @@
 export * from './list.js';
 export * from './pull.js';
+export * from './push.js';
 export * from './request.js';
 export * from './response.js';
-export * from './push.js';
 export * from './stat.js';
 export * from './sync.js';

libraries/adb/src/commands/sync/list.ts

@@ -1,5 +1,6 @@
+import type { BufferedStream, WritableStreamDefaultWriter } from '@yume-chan/stream-extra';
 import Struct from '@yume-chan/struct';
-import type { AdbBufferedStream, WritableStreamDefaultWriter } from '../../stream/index.js';
+
 import { AdbSyncRequestId, adbSyncWriteRequest } from './request.js';
 import { AdbSyncDoneResponse, adbSyncReadResponse, AdbSyncResponseId } from './response.js';
 import { AdbSyncLstatResponse, AdbSyncStatResponse, type AdbSyncStat } from './stat.js';
@@ -37,7 +38,7 @@ const LIST_V2_RESPONSE_TYPES = {
 };
 
 export async function* adbSyncOpenDir(
-    stream: AdbBufferedStream,
+    stream: BufferedStream,
     writer: WritableStreamDefaultWriter<Uint8Array>,
     path: string,
     v2: boolean,

libraries/adb/src/commands/sync/pull.ts

@@ -1,5 +1,6 @@
+import { BufferedStream, ReadableStream, WritableStreamDefaultWriter } from '@yume-chan/stream-extra';
 import Struct from '@yume-chan/struct';
-import { AdbBufferedStream, ReadableStream, WritableStreamDefaultWriter } from '../../stream/index.js';
+
 import { AdbSyncRequestId, adbSyncWriteRequest } from './request.js';
 import { AdbSyncDoneResponse, adbSyncReadResponse, AdbSyncResponseId } from './response.js';
 
@@ -15,7 +16,7 @@ const RESPONSE_TYPES = {
 };
 
 export function adbSyncPull(
-    stream: AdbBufferedStream,
+    stream: BufferedStream,
     writer: WritableStreamDefaultWriter<Uint8Array>,
     path: string,
 ): ReadableStream<Uint8Array> {

libraries/adb/src/commands/sync/push.ts

@@ -1,5 +1,6 @@
+import { BufferedStream, ChunkStream, pipeFrom, WritableStream, WritableStreamDefaultWriter } from '@yume-chan/stream-extra';
 import Struct from '@yume-chan/struct';
-import { AdbBufferedStream, ChunkStream, pipeFrom, WritableStream, WritableStreamDefaultWriter } from '../../stream/index.js';
+
 import { AdbSyncRequestId, adbSyncWriteRequest } from './request.js';
 import { adbSyncReadResponse, AdbSyncResponseId } from './response.js';
 import { LinuxFileType } from './stat.js';
@@ -15,7 +16,7 @@ const ResponseTypes = {
 export const ADB_SYNC_MAX_PACKET_SIZE = 64 * 1024;
 
 export function adbSyncPush(
-    stream: AdbBufferedStream,
+    stream: BufferedStream,
     writer: WritableStreamDefaultWriter<Uint8Array>,
     filename: string,
     mode: number = (LinuxFileType.File << 12) | 0o666,

libraries/adb/src/commands/sync/request.ts

@@ -1,6 +1,7 @@
+import type { WritableStreamDefaultWriter } from '@yume-chan/stream-extra';
 import Struct from '@yume-chan/struct';
-import type { WritableStreamDefaultWriter } from "../../stream/index.js";
+
-import { encodeUtf8 } from "../../utils/index.js";
+import { encodeUtf8 } from '../../utils/index.js';
 
 export enum AdbSyncRequestId {
     List = 'LIST',

libraries/adb/src/commands/sync/response.ts

@@ -1,6 +1,7 @@
+import type { BufferedStream } from '@yume-chan/stream-extra';
 import Struct, { type StructAsyncDeserializeStream, type StructLike, type StructValueType } from '@yume-chan/struct';
-import type { AdbBufferedStream } from '../../stream/index.js';
+
-import { decodeUtf8 } from "../../utils/index.js";
+import { decodeUtf8 } from '../../utils/index.js';
 
 export enum AdbSyncResponseId {
     Entry = 'DENT',
@@ -42,7 +43,7 @@ export const AdbSyncFailResponse =
         });
 
 export async function adbSyncReadResponse<T extends Record<string, StructLike<any>>>(
-    stream: AdbBufferedStream,
+    stream: BufferedStream,
     types: T,
     // When `T` is a union type, `T[keyof T]` only includes their common keys.
     // For example, let `type T = { a: string, b: string } | { a: string, c: string}`,

libraries/adb/src/commands/sync/stat.ts

@@ -1,5 +1,6 @@
+import type { BufferedStream, WritableStreamDefaultWriter } from '@yume-chan/stream-extra';
 import Struct, { placeholder } from '@yume-chan/struct';
-import type { AdbBufferedStream, WritableStreamDefaultWriter } from '../../stream/index.js';
+
 import { AdbSyncRequestId, adbSyncWriteRequest } from './request.js';
 import { adbSyncReadResponse, AdbSyncResponseId } from './response.js';
 
@@ -108,7 +109,7 @@ const LSTAT_V2_RESPONSE_TYPES = {
 };
 
 export async function adbSyncLstat(
-    stream: AdbBufferedStream,
+    stream: BufferedStream,
     writer: WritableStreamDefaultWriter<Uint8Array>,
     path: string,
     v2: boolean,
@@ -142,7 +143,7 @@ export async function adbSyncLstat(
 }
 
 export async function adbSyncStat(
-    stream: AdbBufferedStream,
+    stream: BufferedStream,
     writer: WritableStreamDefaultWriter<Uint8Array>,
     path: string,
 ): Promise<AdbSyncStatResponse> {

libraries/adb/src/commands/sync/sync.ts

@@ -1,10 +1,11 @@
 import { AutoDisposable } from '@yume-chan/event';
+import { BufferedStream, ReadableStream, WrapReadableStream, WrapWritableStream, WritableStream, WritableStreamDefaultWriter } from '@yume-chan/stream-extra';
+
 import type { Adb } from '../../adb.js';
 import { AdbFeatures } from '../../features.js';
 import type { AdbSocket } from '../../socket/index.js';
-import { AdbBufferedStream, ReadableStream, WrapReadableStream, WrapWritableStream, WritableStream, WritableStreamDefaultWriter } from '../../stream/index.js';
 import { AutoResetEvent } from '../../utils/index.js';
-import { escapeArg } from "../index.js";
+import { escapeArg } from '../subprocess/index.js';
 import { adbSyncOpenDir, type AdbSyncEntry } from './list.js';
 import { adbSyncPull } from './pull.js';
 import { adbSyncPush } from './push.js';
@@ -29,7 +30,7 @@ export function dirname(path: string): string {
 export class AdbSync extends AutoDisposable {
     protected adb: Adb;
 
-    protected stream: AdbBufferedStream;
+    protected stream: BufferedStream;
 
     protected writer: WritableStreamDefaultWriter<Uint8Array>;
 
@@ -56,7 +57,7 @@ export class AdbSync extends AutoDisposable {
         super();
 
         this.adb = adb;
-        this.stream = new AdbBufferedStream(socket);
+        this.stream = new BufferedStream(socket.readable);
         this.writer = socket.writable.getWriter();
     }
 

libraries/adb/src/features.ts

@@ -1,7 +1,7 @@
 // The order follows
 // https://android.googlesource.com/platform/packages/modules/adb/+/79010dc6d5ca7490c493df800d4421730f5466ca/transport.cpp#1252
 export enum AdbFeatures {
-    ShellV2 = "shell_v2",
+    ShellV2 = 'shell_v2',
     Cmd = 'cmd',
     StatV2 = 'stat_v2',
     ListV2 = 'ls_v2',

libraries/adb/src/index.ts

@@ -7,5 +7,4 @@ export * from './crypto.js';
 export * from './features.js';
 export * from './packet.js';
 export * from './socket/index.js';
-export * from './stream/index.js';
 export * from './utils/index.js';

libraries/adb/src/packet.ts

@@ -1,5 +1,5 @@
+import { TransformStream } from '@yume-chan/stream-extra';
 import Struct from '@yume-chan/struct';
-import { TransformStream } from "./stream/index.js";
 
 export enum AdbCommand {
     Auth = 0x48545541,    // 'AUTH'

libraries/adb/src/socket/dispatcher.ts

@@ -1,9 +1,9 @@
 import { AsyncOperationManager, PromiseResolver } from '@yume-chan/async';
 import type { RemoveEventListener } from '@yume-chan/event';
-import { EMPTY_UINT8_ARRAY, type ValueOrPromise } from "@yume-chan/struct";
+import { AbortController, WritableStream, WritableStreamDefaultWriter, type ReadableWritablePair } from '@yume-chan/stream-extra';
+import { EMPTY_UINT8_ARRAY, type ValueOrPromise } from '@yume-chan/struct';
 
 import { AdbCommand, calculateChecksum, type AdbPacketData, type AdbPacketInit } from '../packet.js';
-import { AbortController, WritableStream, WritableStreamDefaultWriter, type ReadableWritablePair } from '../stream/index.js';
 import { decodeUtf8, encodeUtf8 } from '../utils/index.js';
 import { AdbSocket, AdbSocketController } from './socket.js';
 

libraries/adb/src/socket/index.ts

@@ -1,2 +1,2 @@
-export * from './socket.js';
 export * from './dispatcher.js';
+export * from './socket.js';

libraries/adb/src/socket/socket.ts

@@ -1,7 +1,8 @@
-import { PromiseResolver } from "@yume-chan/async";
+import { PromiseResolver } from '@yume-chan/async';
-import type { Disposable } from "@yume-chan/event";
+import type { Disposable } from '@yume-chan/event';
+import { ChunkStream, DuplexStreamFactory, pipeFrom, PushReadableStream, WritableStream, type PushReadableStreamController, type ReadableStream, type ReadableWritablePair } from '@yume-chan/stream-extra';
+
 import { AdbCommand } from '../packet.js';
-import { ChunkStream, DuplexStreamFactory, pipeFrom, PushReadableStream, WritableStream, type PushReadableStreamController, type ReadableStream, type ReadableWritablePair } from '../stream/index.js';
 import type { AdbPacketDispatcher, Closeable } from './dispatcher.js';
 
 export interface AdbSocketInfo {

libraries/adb/src/stream/detect.ts

@@ -1,7 +0,0 @@
-// cspell: ignore vercel
-
-// Always use polyfilled version because
-// Vercel doesn't support Node.js 16 (`streams/web` module) yet
-export * from './detect.polyfill.js';
-
-// export * from './detect.native.js';

libraries/adb/src/stream/index.ts

@@ -1,3 +0,0 @@
-export * from './buffered.js';
-export * from './detect.js';
-export * from './transform.js';

libraries/adb/src/stream/transform.ts

@@ -1,474 +0,0 @@
-import { PromiseResolver } from "@yume-chan/async";
-import type Struct from "@yume-chan/struct";
-import type { StructValueType, ValueOrPromise } from "@yume-chan/struct";
-import { decodeUtf8 } from "../utils/index.js";
-import { BufferedStream, BufferedStreamEndedError } from "./buffered.js";
-import { AbortController, AbortSignal, ReadableStream, ReadableStreamDefaultReader, TransformStream, WritableStream, WritableStreamDefaultWriter, type QueuingStrategy, type ReadableStreamDefaultController, type ReadableWritablePair } from "./detect.js";
-
-export interface DuplexStreamFactoryOptions {
-    /**
-     * Callback when any `ReadableStream` is cancelled (the user doesn't need any more data),
-     * or `WritableStream` is ended (the user won't produce any more data),
-     * or `DuplexStreamFactory#close` is called.
-     *
-     * Usually you want to let the other peer know that the duplex stream should be clsoed.
-     *
-     * `dispose` will automatically be called after `close` completes,
-     * but if you want to wait another peer for a close confirmation and call
-     * `DuplexStreamFactory#dispose` yourself, you can return `false`
-     * (or a `Promise` that resolves to `false`) to disable the automatic call.
-     */
-    close?: (() => ValueOrPromise<boolean | void>) | undefined;
-
-    /**
-     * Callback when any `ReadableStream` is closed (the other peer doesn't produce any more data),
-     * or `WritableStream` is aborted (the other peer can't receive any more data),
-     * or `DuplexStreamFactory#abort` is called.
-     *
-     * Usually indicates the other peer has closed the duplex stream. You can clean up
-     * any resources you have allocated now.
-     */
-    dispose?: (() => void | Promise<void>) | undefined;
-}
-
-/**
- * A factory for creating a duplex stream.
- *
- * It can create multiple `ReadableStream`s and `WritableStream`s,
- * when any of them is closed, all other streams will be closed as well.
- */
-export class DuplexStreamFactory<R, W> {
-    private readableControllers: ReadableStreamDefaultController<R>[] = [];
-    private writers: WritableStreamDefaultWriter<W>[] = [];
-
-    private _writableClosed = false;
-    public get writableClosed() { return this._writableClosed; }
-
-    private _closed = new PromiseResolver<void>();
-    public get closed() { return this._closed.promise; }
-
-    private options: DuplexStreamFactoryOptions;
-
-    public constructor(options?: DuplexStreamFactoryOptions) {
-        this.options = options ?? {};
-    }
-
-    public wrapReadable(readable: ReadableStream<R>): WrapReadableStream<R> {
-        return new WrapReadableStream<R>({
-            start: (controller) => {
-                this.readableControllers.push(controller);
-                return readable;
-            },
-            cancel: async () => {
-                // cancel means the local peer closes the connection first.
-                await this.close();
-            },
-            close: async () => {
-                // stream end means the remote peer closed the connection first.
-                await this.dispose();
-            },
-        });
-    }
-
-    public 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.ready;
-                await writer.write(chunk);
-            },
-            abort: async (reason) => {
-                await writer.abort(reason);
-                await this.close();
-            },
-            close: async () => {
-                try { await writer.close(); } catch { }
-                await this.close();
-            },
-        });
-    }
-
-    public async close() {
-        if (this._writableClosed) {
-            return;
-        }
-        this._writableClosed = true;
-
-        // Call `close` first, so it can still write data to `WritableStream`s.
-        if (await this.options.close?.() !== false) {
-            // `close` can return `false` to disable automatic `dispose`.
-            await this.dispose();
-        }
-
-        for (const writer of this.writers) {
-            try { await writer.close(); } catch { }
-        }
-    }
-
-    public async dispose() {
-        this._writableClosed = true;
-        this._closed.resolve();
-
-        for (const controller of this.readableControllers) {
-            try { controller.close(); } catch { }
-        }
-
-        await this.options.dispose?.();
-    }
-}
-
-export class DecodeUtf8Stream extends TransformStream<Uint8Array, string>{
-    public constructor() {
-        super({
-            transform(chunk, controller) {
-                controller.enqueue(decodeUtf8(chunk));
-            },
-        });
-    }
-}
-
-export class GatherStringStream extends WritableStream<string>{
-    // Optimization: rope (concat strings) is faster than `[].join('')`
-    private _result = '';
-    public get result() { return this._result; }
-
-    public constructor() {
-        super({
-            write: (chunk) => {
-                this._result += chunk;
-            },
-        });
-    }
-}
-
-// TODO: StructTransformStream: Looking for better implementation
-export class StructDeserializeStream<T extends Struct<any, any, any, any>>
-    implements ReadableWritablePair<Uint8Array, StructValueType<T>>{
-    private _readable: ReadableStream<StructValueType<T>>;
-    public get readable() { return this._readable; }
-
-    private _writable: WritableStream<Uint8Array>;
-    public get writable() { return this._writable; }
-
-    public constructor(struct: T) {
-        // Convert incoming chunks to a `BufferedStream`
-        let incomingStreamController!: PushReadableStreamController<Uint8Array>;
-        const incomingStream = new BufferedStream(
-            new PushReadableStream<Uint8Array>(
-                controller => incomingStreamController = controller,
-            )
-        );
-
-        this._readable = new ReadableStream<StructValueType<T>>({
-            async pull(controller) {
-                try {
-                    const value = await struct.deserialize(incomingStream);
-                    controller.enqueue(value);
-                } catch (e) {
-                    if (e instanceof BufferedStreamEndedError) {
-                        controller.close();
-                        return;
-                    }
-                    throw e;
-                }
-            }
-        });
-
-        this._writable = new WritableStream({
-            async write(chunk) {
-                await incomingStreamController.enqueue(chunk);
-            },
-            abort() {
-                incomingStreamController.close();
-            },
-            close() {
-                incomingStreamController.close();
-            },
-        });
-    }
-}
-
-export class StructSerializeStream<T extends Struct<any, any, any, any>>
-    extends TransformStream<T['TInit'], Uint8Array>{
-    constructor(struct: T) {
-        super({
-            transform(chunk, controller) {
-                controller.enqueue(struct.serialize(chunk));
-            },
-        });
-    }
-}
-
-export type WrapWritableStreamStart<T> = () => ValueOrPromise<WritableStream<T>>;
-
-export interface WritableStreamWrapper<T> {
-    start: WrapWritableStreamStart<T>;
-    close?(): Promise<void>;
-}
-
-async function getWrappedWritableStream<T>(
-    wrapper: WritableStream<T> | WrapWritableStreamStart<T> | WritableStreamWrapper<T>
-) {
-    if ('start' in wrapper) {
-        return await wrapper.start();
-    } else if (typeof wrapper === 'function') {
-        return await wrapper();
-    } else {
-        // Can't use `wrapper instanceof WritableStream`
-        // Because we want to be compatible with any WritableStream-like objects
-        return wrapper;
-    }
-}
-
-export class WrapWritableStream<T> extends WritableStream<T> {
-    public writable!: WritableStream<T>;
-
-    private writer!: WritableStreamDefaultWriter<T>;
-
-    public constructor(wrapper: WritableStream<T> | WrapWritableStreamStart<T> | WritableStreamWrapper<T>) {
-        super({
-            start: async () => {
-                // `start` is invoked before `ReadableStream`'s constructor finish,
-                // so using `this` synchronously causes
-                // "Must call super constructor in derived class before accessing 'this' or returning from derived constructor".
-                // Queue a microtask to avoid this.
-                await Promise.resolve();
-
-                this.writable = await getWrappedWritableStream(wrapper);
-                this.writer = this.writable.getWriter();
-            },
-            write: async (chunk) => {
-                // Maintain back pressure
-                await this.writer.ready;
-                await this.writer.write(chunk);
-            },
-            abort: async (reason) => {
-                await this.writer.abort(reason);
-                if ('close' in wrapper) {
-                    await wrapper.close?.();
-                }
-            },
-            close: async () => {
-                // Close the inner stream first.
-                // Usually the inner stream is a logical sub-stream over the outer stream,
-                // closing the outer stream first will make the inner stream incapable of
-                // sending data in its `close` handler.
-                await this.writer.close();
-                if ('close' in wrapper) {
-                    await wrapper.close?.();
-                }
-            },
-        });
-    }
-}
-
-export type WrapReadableStreamStart<T> = (controller: ReadableStreamDefaultController<T>) => ValueOrPromise<ReadableStream<T>>;
-
-export interface ReadableStreamWrapper<T> {
-    start: WrapReadableStreamStart<T>;
-    cancel?(reason?: any): ValueOrPromise<void>;
-    close?(): ValueOrPromise<void>;
-}
-
-function getWrappedReadableStream<T>(
-    wrapper: ReadableStream<T> | WrapReadableStreamStart<T> | ReadableStreamWrapper<T>,
-    controller: ReadableStreamDefaultController<T>
-) {
-    if ('start' in wrapper) {
-        return wrapper.start(controller);
-    } else if (typeof wrapper === 'function') {
-        return wrapper(controller);
-    } else {
-        // Can't use `wrapper instanceof ReadableStream`
-        // Because we want to be compatible with any ReadableStream-like objects
-        return wrapper;
-    }
-}
-
-/**
- * This class has multiple usages:
- *
- * 1. Get notified when the stream is cancelled or closed.
- * 2. Synchronously create a `ReadableStream` by asynchronously return another `ReadableStream`.
- * 3. Convert native `ReadableStream`s to polyfilled ones so they can `pipe` between.
- */
-export class WrapReadableStream<T> extends ReadableStream<T>{
-    public readable!: ReadableStream<T>;
-
-    private reader!: ReadableStreamDefaultReader<T>;
-
-    public constructor(wrapper: ReadableStream<T> | WrapReadableStreamStart<T> | ReadableStreamWrapper<T>) {
-        super({
-            start: async (controller) => {
-                // `start` is invoked before `ReadableStream`'s constructor finish,
-                // so using `this` synchronously causes
-                // "Must call super constructor in derived class before accessing 'this' or returning from derived constructor".
-                // Queue a microtask to avoid this.
-                await Promise.resolve();
-
-                this.readable = await getWrappedReadableStream(wrapper, controller);
-                this.reader = this.readable.getReader();
-            },
-            cancel: async (reason) => {
-                await this.reader.cancel(reason);
-                if ('cancel' in wrapper) {
-                    await wrapper.cancel?.(reason);
-                }
-            },
-            pull: async (controller) => {
-                const result = await this.reader.read();
-                if (result.done) {
-                    controller.close();
-                    if ('close' in wrapper) {
-                        await wrapper.close?.();
-                    }
-                } else {
-                    controller.enqueue(result.value);
-                }
-            }
-        });
-    }
-}
-
-export class ChunkStream extends TransformStream<Uint8Array, Uint8Array>{
-    public constructor(size: number) {
-        super({
-            transform(chunk, controller) {
-                for (let start = 0; start < chunk.byteLength;) {
-                    const end = start + size;
-                    controller.enqueue(chunk.subarray(start, end));
-                    start = end;
-                }
-            }
-        });
-    }
-}
-
-function* splitLines(text: string): Generator<string, void, void> {
-    let start = 0;
-
-    while (true) {
-        const index = text.indexOf('\n', start);
-        if (index === -1) {
-            return;
-        }
-
-        const line = text.substring(start, index);
-        yield line;
-
-        start = index + 1;
-    }
-}
-
-export class SplitLineStream extends TransformStream<string, string> {
-    public constructor() {
-        super({
-            transform(chunk, controller) {
-                for (const line of splitLines(chunk)) {
-                    controller.enqueue(line);
-                }
-            }
-        });
-    }
-}
-
-/**
- * Create a new `WritableStream` that, when written to, will write that chunk to
- * `pair.writable`, when pipe `pair.readable` to `writable`.
- *
- * It's the opposite of `ReadableStream.pipeThrough`.
- *
- * @param writable The `WritableStream` to write to.
- * @param pair A `TransformStream` that converts chunks.
- * @returns A new `WritableStream`.
- */
-export function pipeFrom<W, T>(writable: WritableStream<W>, pair: ReadableWritablePair<W, T>) {
-    const writer = pair.writable.getWriter();
-    const pipe = pair.readable
-        .pipeTo(writable);
-    return new WritableStream<T>({
-        async write(chunk) {
-            await writer.ready;
-            await writer.write(chunk);
-        },
-        async close() {
-            await writer.close();
-            await pipe;
-        }
-    });
-}
-
-export class InspectStream<T> extends TransformStream<T, T> {
-    constructor(callback: (value: T) => void) {
-        super({
-            transform(chunk, controller) {
-                callback(chunk);
-                controller.enqueue(chunk);
-            }
-        });
-    }
-}
-
-export interface PushReadableStreamController<T> {
-    abortSignal: AbortSignal;
-
-    enqueue(chunk: T): Promise<void>;
-
-    close(): void;
-
-    error(e?: any): void;
-}
-
-export type PushReadableStreamSource<T> = (controller: PushReadableStreamController<T>) => void;
-
-export class PushReadableStream<T> extends ReadableStream<T> {
-    public constructor(source: PushReadableStreamSource<T>, strategy?: QueuingStrategy<T>) {
-        let waterMarkLow: PromiseResolver<void> | undefined;
-        const canceled: AbortController = new AbortController();
-
-        super({
-            start: (controller) => {
-                source({
-                    abortSignal: canceled.signal,
-                    async enqueue(chunk) {
-                        if (canceled.signal.aborted) {
-                            // If the stream is already cancelled,
-                            // throw immediately.
-                            throw canceled.signal.reason ?? new Error('Aborted');
-                        }
-
-                        // Only when the stream is errored, `desiredSize` will be `null`.
-                        // But since `null <= 0` is `true`
-                        // (`null <= 0` is evaluated as `!(null > 0)` => `!false` => `true`),
-                        // not handling it will cause a deadlock.
-                        if ((controller.desiredSize ?? 1) <= 0) {
-                            waterMarkLow = new PromiseResolver<void>();
-                            await waterMarkLow.promise;
-                        }
-
-                        // `controller.enqueue` will throw error for us
-                        // if the stream is already errored.
-                        controller.enqueue(chunk);
-                    },
-                    close() {
-                        controller.close();
-                    },
-                    error(e) {
-                        controller.error(e);
-                    },
-                });
-            },
-            pull: () => {
-                waterMarkLow?.resolve();
-            },
-            cancel: async (reason) => {
-                canceled.abort(reason);
-                waterMarkLow?.reject(reason);
-            },
-        }, strategy);
-    }
-}

libraries/adb/src/utils/base64.spec.ts

@@ -1,3 +1,5 @@
+import { describe, expect, it } from '@jest/globals';
+
 import { calculateBase64EncodedLength, decodeBase64, encodeBase64 } from './base64.js';
 
 describe('base64', () => {
@@ -117,7 +119,7 @@ describe('base64', () => {
                         let inputIndex = inputOffset + input.length - 1;
                         let outputIndex = outputOffset + correct.length - 1;
 
-                        const paddingLength = correct.filter(x => x === "=".charCodeAt(0)).length;
+                        const paddingLength = correct.filter(x => x === '='.charCodeAt(0)).length;
                         if (paddingLength !== 0) {
                             inputIndex -= (3 - paddingLength);
                             outputIndex -= 4;

libraries/android-bin/package.json

@@ -35,6 +35,7 @@
     },
     "dependencies": {
         "@yume-chan/adb": "^0.0.16",
+        "@yume-chan/stream-extra": "^0.0.16",
         "@yume-chan/struct": "^0.0.16",
         "tslib": "^2.3.1"
     }

libraries/android-bin/src/bug-report.ts

@@ -1,7 +1,8 @@
 // cspell: ignore bugreport
 // cspell: ignore bugreportz
 
-import { AdbCommandBase, AdbSubprocessShellProtocol, DecodeUtf8Stream, PushReadableStream, ReadableStream, SplitLineStream, WrapReadableStream, WritableStream } from "@yume-chan/adb";
+import { AdbCommandBase, AdbSubprocessShellProtocol } from '@yume-chan/adb';
+import { DecodeUtf8Stream, PushReadableStream, ReadableStream, SplitStringStream, WrapReadableStream, WritableStream } from '@yume-chan/stream-extra';
 
 export interface BugReportZVersion {
     major: number;
@@ -78,7 +79,7 @@ export class BugReportZ extends AdbCommandBase {
 
         await process.stdout
             .pipeThrough(new DecodeUtf8Stream())
-            .pipeThrough(new SplitLineStream())
+            .pipeThrough(new SplitStringStream('\n'))
             .pipeTo(new WritableStream<string>({
                 write(line) {
                     // `BEGIN:` and `PROGRESS:` only appear when `-p` is specified.

libraries/android-bin/src/logcat.ts

@@ -1,7 +1,8 @@
 // cspell: ignore logcat
 
-import { AdbCommandBase, AdbSubprocessNoneProtocol, BufferedStream, BufferedStreamEndedError, DecodeUtf8Stream, ReadableStream, SplitLineStream, WritableStream } from "@yume-chan/adb";
+import { AdbCommandBase, AdbSubprocessNoneProtocol } from '@yume-chan/adb';
-import Struct, { decodeUtf8, StructAsyncDeserializeStream } from "@yume-chan/struct";
+import { BufferedStream, BufferedStreamEndedError, DecodeUtf8Stream, ReadableStream, SplitStringStream, WritableStream } from '@yume-chan/stream-extra';
+import Struct, { decodeUtf8, StructAsyncDeserializeStream } from '@yume-chan/struct';
 
 // `adb logcat` is an alias to `adb shell logcat`
 // so instead of adding to core library, it's implemented here
@@ -144,7 +145,7 @@ export class Logcat extends AdbCommandBase {
         const result: LogSize[] = [];
         await stdout
             .pipeThrough(new DecodeUtf8Stream())
-            .pipeThrough(new SplitLineStream())
+            .pipeThrough(new SplitStringStream('\n'))
             .pipeTo(new WritableStream({
                 write(chunk) {
                     let match = chunk.match(Logcat.LOG_SIZE_REGEX_11);

libraries/scrcpy/package.json

@@ -38,6 +38,7 @@
         "@yume-chan/adb": "^0.0.16",
         "@yume-chan/async": "^2.1.4",
         "@yume-chan/event": "^0.0.16",
+        "@yume-chan/stream-extra": "^0.0.16",
         "@yume-chan/struct": "^0.0.16",
         "tslib": "^2.3.1"
     },

libraries/scrcpy/src/adb-client.ts

@@ -1,37 +1,8 @@
-import { AdbCommandBase, AdbSubprocessNoneProtocol, AdbSubprocessProtocol, AdbSync, DecodeUtf8Stream, ReadableStream, TransformStream, WrapWritableStream, WritableStream } from "@yume-chan/adb";
+import { AdbCommandBase, AdbSubprocessNoneProtocol, AdbSubprocessProtocol, AdbSync } from '@yume-chan/adb';
-import { ScrcpyClient } from "./client.js";
+import { DecodeUtf8Stream, ReadableStream, SplitStringStream, WrapWritableStream, WritableStream } from '@yume-chan/stream-extra';
-import { DEFAULT_SERVER_PATH, type ScrcpyOptions } from "./options/index.js";
 
-function* splitLines(text: string): Generator<string, void, void> {
+import { ScrcpyClient } from './client.js';
-    let start = 0;
+import { DEFAULT_SERVER_PATH, type ScrcpyOptions } from './options/index.js';
-
-    while (true) {
-        const index = text.indexOf('\n', start);
-        if (index === -1) {
-            return;
-        }
-
-        const line = text.substring(start, index);
-        yield line;
-
-        start = index + 1;
-    }
-}
-
-class SplitLinesStream extends TransformStream<string, string>{
-    constructor() {
-        super({
-            transform(chunk, controller) {
-                for (const line of splitLines(chunk)) {
-                    if (line === '') {
-                        continue;
-                    }
-                    controller.enqueue(line);
-                }
-            },
-        });
-    }
-}
 
 class ArrayToStream<T> extends ReadableStream<T>{
     private array!: T[];
@@ -135,7 +106,7 @@ export class AdbScrcpyClient extends AdbCommandBase {
 
             const stdout = process.stdout
                 .pipeThrough(new DecodeUtf8Stream())
-                .pipeThrough(new SplitLinesStream());
+                .pipeThrough(new SplitStringStream('\n'));
 
             // Read stdout, otherwise `process.exit` won't resolve.
             const output: string[] = [];

libraries/scrcpy/src/client.ts

@@ -1,8 +1,9 @@
-import { AbortController, BufferedStream, InspectStream, ReadableStream, ReadableWritablePair, TransformStream, type WritableStreamDefaultWriter } from '@yume-chan/adb';
 import { EventEmitter } from '@yume-chan/event';
+import { AbortController, BufferedStream, InspectStream, ReadableStream, ReadableWritablePair, TransformStream, type WritableStreamDefaultWriter } from '@yume-chan/stream-extra';
 import Struct from '@yume-chan/struct';
+
 import { AndroidMotionEventAction, ScrcpyControlMessageType, ScrcpyInjectKeyCodeControlMessage, ScrcpyInjectTextControlMessage, ScrcpyInjectTouchControlMessage, ScrcpySimpleControlMessage, type AndroidKeyEventAction } from './message.js';
-import type { ScrcpyInjectScrollControlMessage1_22, ScrcpyOptions, VideoStreamPacket } from "./options/index.js";
+import type { ScrcpyInjectScrollControlMessage1_22, ScrcpyOptions, VideoStreamPacket } from './options/index.js';
 
 const ClipboardMessage =
     new Struct()

libraries/scrcpy/src/connection.ts

@@ -1,7 +1,9 @@
-import type { Adb, ReadableStream, ReadableWritablePair } from "@yume-chan/adb";
+import type { Adb } from '@yume-chan/adb';
-import type { Disposable } from "@yume-chan/event";
+import type { Disposable } from '@yume-chan/event';
-import type { ValueOrPromise } from "@yume-chan/struct";
+import type { ReadableStream, ReadableWritablePair } from '@yume-chan/stream-extra';
-import { delay } from "./utils.js";
+import type { ValueOrPromise } from '@yume-chan/struct';
+
+import { delay } from './utils.js';
 
 export interface ScrcpyClientConnectionOptions {
     control: boolean;

libraries/scrcpy/src/decoder/index.ts

@@ -1,3 +1,3 @@
-export * from './types.js';
 export * from './tinyh264/index.js';
+export * from './types.js';
 export * from './web-codecs/index.js';

libraries/scrcpy/src/decoder/tinyh264/index.ts

@@ -1,9 +1,10 @@
-import { WritableStream } from "@yume-chan/adb";
+import { PromiseResolver } from '@yume-chan/async';
-import { PromiseResolver } from "@yume-chan/async";
+import { WritableStream } from '@yume-chan/stream-extra';
-import { AndroidCodecLevel, AndroidCodecProfile } from "../../codec.js";
+
-import type { VideoStreamPacket } from "../../options/index.js";
+import { AndroidCodecLevel, AndroidCodecProfile } from '../../codec.js';
-import type { H264Configuration, H264Decoder } from "../types.js";
+import type { VideoStreamPacket } from '../../options/index.js';
-import { createTinyH264Wrapper, type TinyH264Wrapper } from "./wrapper.js";
+import type { H264Configuration, H264Decoder } from '../types.js';
+import { createTinyH264Wrapper, type TinyH264Wrapper } from './wrapper.js';
 
 let cachedInitializePromise: Promise<{ YuvBuffer: typeof import('yuv-buffer'), YuvCanvas: typeof import('yuv-canvas').default; }> | undefined;
 function initialize() {

libraries/scrcpy/src/decoder/tinyh264/worker.ts

@@ -1,2 +1,3 @@
 import { init } from 'tinyh264';
+
 init();

libraries/scrcpy/src/decoder/types.ts

@@ -1,7 +1,8 @@
-import type { WritableStream } from '@yume-chan/adb';
+import type { Disposable } from '@yume-chan/event';
-import type { Disposable } from "@yume-chan/event";
+import type { WritableStream } from '@yume-chan/stream-extra';
-import type { AndroidCodecLevel, AndroidCodecProfile } from "../codec.js";
+
-import type { VideoStreamPacket } from "../options/index.js";
+import type { AndroidCodecLevel, AndroidCodecProfile } from '../codec.js';
+import type { VideoStreamPacket } from '../options/index.js';
 
 export interface H264Configuration {
     profileIndex: number;

libraries/scrcpy/src/decoder/web-codecs/index.ts

@@ -1,6 +1,7 @@
-import { WritableStream } from '@yume-chan/adb';
+import { WritableStream } from '@yume-chan/stream-extra';
-import type { VideoStreamPacket } from "../../options/index.js";
+
-import type { H264Configuration, H264Decoder } from "../types.js";
+import type { VideoStreamPacket } from '../../options/index.js';
+import type { H264Configuration, H264Decoder } from '../types.js';
 
 function toHex(value: number) {
     return value.toString(16).padStart(2, '0').toUpperCase();

libraries/scrcpy/src/message.ts

@@ -2,6 +2,7 @@ import Struct, { placeholder } from '@yume-chan/struct';
 
 // https://github.com/Genymobile/scrcpy/blob/fa5b2a29e983a46b49531def9cf3d80c40c3de37/app/src/control_msg.h#L23
 // For their message bodies, see https://github.com/Genymobile/scrcpy/blob/5c62f3419d252d10cd8c9cbb7c918b358b81f2d0/app/src/control_msg.c#L92
+// Their IDs change between versions, so always use `options.getControlMessageTypes()`
 export enum ScrcpyControlMessageType {
     InjectKeycode,
     InjectText,

libraries/scrcpy/src/options/1_16/index.ts

@@ -1,12 +1,14 @@
-import { StructDeserializeStream, TransformStream, type Adb } from "@yume-chan/adb";
+import type { Adb } from '@yume-chan/adb';
-import Struct from "@yume-chan/struct";
+import { StructDeserializeStream, TransformStream } from '@yume-chan/stream-extra';
-import type { AndroidCodecLevel, AndroidCodecProfile } from "../../codec.js";
+import Struct from '@yume-chan/struct';
-import { ScrcpyClientConnection, ScrcpyClientForwardConnection, ScrcpyClientReverseConnection, type ScrcpyClientConnectionOptions } from "../../connection.js";
+
-import { AndroidKeyEventAction, ScrcpyControlMessageType, ScrcpySimpleControlMessage } from "../../message.js";
+import type { AndroidCodecLevel, AndroidCodecProfile } from '../../codec.js';
-import type { ScrcpyBackOrScreenOnEvent1_18 } from "../1_18.js";
+import { ScrcpyClientConnection, ScrcpyClientForwardConnection, ScrcpyClientReverseConnection, type ScrcpyClientConnectionOptions } from '../../connection.js';
-import type { ScrcpyInjectScrollControlMessage1_22 } from "../1_22.js";
+import { AndroidKeyEventAction, ScrcpyControlMessageType, ScrcpySimpleControlMessage } from '../../message.js';
-import { toScrcpyOptionValue, type ScrcpyOptions, type ScrcpyOptionValue, type VideoStreamPacket } from "../common.js";
+import type { ScrcpyBackOrScreenOnEvent1_18 } from '../1_18.js';
-import { parse_sequence_parameter_set } from "./sps.js";
+import type { ScrcpyInjectScrollControlMessage1_22 } from '../1_22.js';
+import { toScrcpyOptionValue, type ScrcpyOptions, type ScrcpyOptionValue, type VideoStreamPacket } from '../common.js';
+import { parse_sequence_parameter_set } from './sps.js';
 
 export enum ScrcpyLogLevel {
     Verbose = 'verbose',
@@ -209,7 +211,8 @@ export class ScrcpyOptions1_16<T extends ScrcpyOptionsInit1_16 = ScrcpyOptionsIn
 
     public createConnection(adb: Adb): ScrcpyClientConnection {
         const options: ScrcpyClientConnectionOptions = {
-            // Old scrcpy connection always have control stream no matter what the option is
+            // Old versions always have control stream no matter what the option is
+            // Pass `control: false` to `Connection` will disable the control stream
             control: true,
             sendDummyByte: true,
             sendDeviceMeta: true,

libraries/scrcpy/src/options/1_18.ts

@@ -1,6 +1,7 @@
-import Struct, { placeholder } from "@yume-chan/struct";
+import Struct, { placeholder } from '@yume-chan/struct';
-import { AndroidKeyEventAction, ScrcpyControlMessageType } from "../message.js";
+
-import { ScrcpyBackOrScreenOnEvent1_16, ScrcpyOptions1_16, type ScrcpyOptionsInit1_16 } from "./1_16/index.js";
+import { AndroidKeyEventAction, ScrcpyControlMessageType } from '../message.js';
+import { ScrcpyBackOrScreenOnEvent1_16, ScrcpyOptions1_16, type ScrcpyOptionsInit1_16 } from './1_16/index.js';
 
 export interface ScrcpyOptionsInit1_18 extends ScrcpyOptionsInit1_16 {
     powerOffOnClose: boolean;
@@ -11,7 +12,7 @@ export const ScrcpyBackOrScreenOnEvent1_18 =
         .fields(ScrcpyBackOrScreenOnEvent1_16)
         .uint8('action', placeholder<AndroidKeyEventAction>());
 
-export type ScrcpyBackOrScreenOnEvent1_18 = typeof ScrcpyBackOrScreenOnEvent1_18["TInit"];
+export type ScrcpyBackOrScreenOnEvent1_18 = typeof ScrcpyBackOrScreenOnEvent1_18['TInit'];
 
 export class ScrcpyOptions1_18<T extends ScrcpyOptionsInit1_18 = ScrcpyOptionsInit1_18> extends ScrcpyOptions1_16<T> {
     constructor(value: Partial<ScrcpyOptionsInit1_18>) {

libraries/scrcpy/src/options/1_21.ts

@@ -1,7 +1,7 @@
 // cspell: ignore autosync
 
 import { ScrcpyOptions1_18, type ScrcpyOptionsInit1_18 } from './1_18.js';
-import { toScrcpyOptionValue } from "./common.js";
+import { toScrcpyOptionValue } from './common.js';
 
 export interface ScrcpyOptionsInit1_21 extends ScrcpyOptionsInit1_18 {
     clipboardAutosync?: boolean;

libraries/scrcpy/src/options/1_22.ts

@@ -1,8 +1,9 @@
-import type { Adb } from "@yume-chan/adb";
+import type { Adb } from '@yume-chan/adb';
-import Struct from "@yume-chan/struct";
+import Struct from '@yume-chan/struct';
-import { ScrcpyClientForwardConnection, ScrcpyClientReverseConnection, type ScrcpyClientConnection } from "../connection.js";
+
-import { ScrcpyInjectScrollControlMessage1_16 } from "./1_16/index.js";
+import { ScrcpyClientForwardConnection, ScrcpyClientReverseConnection, type ScrcpyClientConnection } from '../connection.js';
-import { ScrcpyOptions1_21, type ScrcpyOptionsInit1_21 } from "./1_21.js";
+import { ScrcpyInjectScrollControlMessage1_16 } from './1_16/index.js';
+import { ScrcpyOptions1_21, type ScrcpyOptionsInit1_21 } from './1_21.js';
 
 export interface ScrcpyOptionsInit1_22 extends ScrcpyOptionsInit1_21 {
     downsizeOnError: boolean;
@@ -32,9 +33,9 @@ export interface ScrcpyOptionsInit1_22 extends ScrcpyOptionsInit1_21 {
 export const ScrcpyInjectScrollControlMessage1_22 =
     new Struct()
         .fields(ScrcpyInjectScrollControlMessage1_16)
-        .int32("buttons");
+        .int32('buttons');
 
-export type ScrcpyInjectScrollControlMessage1_22 = typeof ScrcpyInjectScrollControlMessage1_22["TInit"];
+export type ScrcpyInjectScrollControlMessage1_22 = typeof ScrcpyInjectScrollControlMessage1_22['TInit'];
 
 export class ScrcpyOptions1_22<T extends ScrcpyOptionsInit1_22 = ScrcpyOptionsInit1_22> extends ScrcpyOptions1_21<T> {
     public constructor(init: Partial<ScrcpyOptionsInit1_22>) {
@@ -58,15 +59,15 @@ export class ScrcpyOptions1_22<T extends ScrcpyOptionsInit1_22 = ScrcpyOptionsIn
         };
     }
 
-    public override createConnection(device: Adb): ScrcpyClientConnection {
+    public override createConnection(adb: Adb): ScrcpyClientConnection {
         const options = {
             ...this.getDefaultValue(),
             ...this.value,
         };
         if (this.value.tunnelForward) {
-            return new ScrcpyClientForwardConnection(device, options);
+            return new ScrcpyClientForwardConnection(adb, options);
         } else {
-            return new ScrcpyClientReverseConnection(device, options);
+            return new ScrcpyClientReverseConnection(adb, options);
         }
     }
 

libraries/scrcpy/src/options/1_23.ts

@@ -1,6 +1,7 @@
-import { TransformStream } from "@yume-chan/adb";
+import { TransformStream } from '@yume-chan/stream-extra';
+
 import { ScrcpyOptions1_22, type ScrcpyOptionsInit1_22 } from './1_22.js';
-import type { VideoStreamPacket } from "./common.js";
+import type { VideoStreamPacket } from './common.js';
 
 export interface ScrcpyOptionsInit1_23 extends ScrcpyOptionsInit1_22 {
     cleanup: boolean;

libraries/scrcpy/src/options/common.ts

@@ -1,9 +1,11 @@
-import type { Adb, TransformStream } from "@yume-chan/adb";
+import type { Adb } from '@yume-chan/adb';
-import type { ScrcpyClientConnection } from "../connection.js";
+import type { TransformStream } from '@yume-chan/stream-extra';
-import type { H264Configuration } from "../decoder/index.js";
+
-import type { ScrcpyControlMessageType } from "../message.js";
+import type { ScrcpyClientConnection } from '../connection.js';
-import type { ScrcpyBackOrScreenOnEvent1_18 } from "./1_18.js";
+import type { H264Configuration } from '../decoder/index.js';
-import type { ScrcpyInjectScrollControlMessage1_22 } from "./1_22.js";
+import type { ScrcpyControlMessageType } from '../message.js';
+import type { ScrcpyBackOrScreenOnEvent1_18 } from './1_18.js';
+import type { ScrcpyInjectScrollControlMessage1_22 } from './1_22.js';
 
 export const DEFAULT_SERVER_PATH = '/data/local/tmp/scrcpy-server.jar';
 

libraries/stream-extra/.npmignore

@@ -0,0 +1,14 @@
+.rush
+
+# Test
+coverage
+**/*.spec.ts
+**/*.spec.js
+**/*.spec.js.map
+**/__helpers__
+jest.config.js
+
+tsconfig.json
+
+# Logs
+*.log

libraries/stream-extra/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2020-2022 Simon Chan
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

libraries/stream-extra/README.md

@@ -0,0 +1,11 @@
+# @yume-chan/stream-extra
+
+Some useful extensions for Web Streams API.
+
+Currently it's using [web-streams-polyfill](https://github.com/MattiasBuelens/web-streams-polyfill) because it's hard to load native implementations from both browsers and Node.js. (An experimental implementation using Top Level Await is available in `native.ts`, but not exported).
+
+## `BufferedStream`
+
+Allowing reading specified amount of data by buffering incoming data.
+
+It's not a Web Stream API `ReadableStream`, because `ReadableStream` doesn't allow hinting the desired read size (except using BYOB readable, but causes extra allocations for small reads).

libraries/stream-extra/jest.config.js

@@ -0,0 +1,13 @@
+/** @type {import('ts-jest').InitialOptionsTsJest} */
+export default {
+    preset: "ts-jest/presets/default-esm",
+    globals: {
+        'ts-jest': {
+            tsconfig: 'tsconfig.test.json',
+            useESM: true,
+        },
+    },
+    moduleNameMapper: {
+        '^(\\.{1,2}/.*)\\.js$': '$1',
+    },
+};

libraries/stream-extra/package.json

@@ -0,0 +1,47 @@
+{
+    "name": "@yume-chan/stream-extra",
+    "version": "0.0.16",
+    "description": "Extensions to Web Streams API",
+    "keywords": [
+        "stream",
+        "web-streams-api"
+    ],
+    "license": "MIT",
+    "author": {
+        "name": "Simon Chan",
+        "email": "cnsimonchan@live.com",
+        "url": "https://chensi.moe/blog"
+    },
+    "homepage": "https://github.com/yume-chan/ya-webadb/tree/master/packages/stream-extra#readme",
+    "repository": {
+        "type": "git",
+        "url": "git+https://github.com/yume-chan/ya-webadb.git",
+        "directory": "packages/stream-extra"
+    },
+    "bugs": {
+        "url": "https://github.com/yume-chan/ya-webadb/issues"
+    },
+    "type": "module",
+    "main": "esm/index.js",
+    "types": "esm/index.d.ts",
+    "scripts": {
+        "build": "tsc -b tsconfig.build.json",
+        "build:watch": "tsc -b tsconfig.build.json",
+        "test": "cross-env NODE_OPTIONS=--experimental-vm-modules jest --coverage",
+        "prepublishOnly": "npm run build"
+    },
+    "dependencies": {
+        "@yume-chan/async": "^2.1.4",
+        "@yume-chan/struct": "^0.0.16",
+        "tslib": "^2.3.1",
+        "web-streams-polyfill": "^4.0.0-beta.3"
+    },
+    "devDependencies": {
+        "@jest/globals": "^28.1.0",
+        "@yume-chan/ts-package-builder": "^1.0.0",
+        "cross-env": "^7.0.3",
+        "jest": "^28.1.0",
+        "ts-jest": "^28.0.2",
+        "typescript": "4.7.2"
+    }
+}

libraries/stream-extra/src/buffered.spec.ts

@@ -1,7 +1,7 @@
 import { describe, expect, it } from '@jest/globals';
 
 import { BufferedStream } from "./buffered.js";
-import { ReadableStream } from "./detect.js";
+import { ReadableStream } from "./stream.js";
 
 function randomUint8Array(length: number) {
     const array = new Uint8Array(length);
@@ -120,6 +120,5 @@ describe('BufferedStream', () => {
         it('input 3 small buffers 2', () => {
             return runTest([3, 3, 3], [7, 2]);
         });
-
     });
 });

libraries/stream-extra/src/buffered.ts

@@ -1,7 +1,5 @@
-import type { StructAsyncDeserializeStream } from '@yume-chan/struct';
+import { PushReadableStream } from "./push-readable.js";
-import type { AdbSocket, AdbSocketInfo } from '../socket/index.js';
+import type { ReadableStream, ReadableStreamDefaultReader } from "./stream.js";
-import type { ReadableStream, ReadableStreamDefaultReader } from './detect.js';
-import { PushReadableStream } from "./transform.js";
 
 export class BufferedStreamEndedError extends Error {
     public constructor() {
@@ -148,21 +146,3 @@ export class BufferedStream {
         await this.reader.cancel();
     }
 }
-
-export class AdbBufferedStream
-    extends BufferedStream
-    implements AdbSocketInfo, StructAsyncDeserializeStream {
-    protected readonly socket: AdbSocket;
-
-    public get localId() { return this.socket.localId; }
-    public get remoteId() { return this.socket.remoteId; }
-    public get localCreated() { return this.socket.localCreated; }
-    public get serviceString() { return this.socket.serviceString; }
-
-    public get writable() { return this.socket.writable; }
-
-    public constructor(socket: AdbSocket) {
-        super(socket.readable);
-        this.socket = socket;
-    }
-}

libraries/stream-extra/src/chunk.ts

@@ -0,0 +1,15 @@
+import { TransformStream } from "./stream.js";
+
+export class ChunkStream extends TransformStream<Uint8Array, Uint8Array>{
+    public constructor(size: number) {
+        super({
+            transform(chunk, controller) {
+                for (let start = 0; start < chunk.byteLength;) {
+                    const end = start + size;
+                    controller.enqueue(chunk.subarray(start, end));
+                    start = end;
+                }
+            }
+        });
+    }
+}

libraries/stream-extra/src/decode-utf8.ts

@@ -0,0 +1,12 @@
+import { decodeUtf8 } from '@yume-chan/struct';
+import { TransformStream } from "./stream.js";
+
+export class DecodeUtf8Stream extends TransformStream<Uint8Array, string>{
+    public constructor() {
+        super({
+            transform(chunk, controller) {
+                controller.enqueue(decodeUtf8(chunk));
+            },
+        });
+    }
+}

libraries/stream-extra/src/duplex.spec.ts

@@ -1,5 +1,6 @@
-import { ReadableStream } from "./detect.js";
+import { describe, it } from '@jest/globals';
-import { DuplexStreamFactory } from './transform.js';
+import { DuplexStreamFactory } from "./duplex.js";
+import { ReadableStream } from "./stream.js";
 
 describe('DuplexStreamFactory', () => {
     it('should close all readable', async () => {

libraries/stream-extra/src/duplex.ts

@@ -0,0 +1,120 @@
+import { PromiseResolver } from "@yume-chan/async";
+import type { ValueOrPromise } from "@yume-chan/struct";
+import { WritableStream, type ReadableStream, type ReadableStreamDefaultController, type WritableStreamDefaultWriter } from "./stream.js";
+import { WrapReadableStream } from "./wrap-readable.js";
+
+export interface DuplexStreamFactoryOptions {
+    /**
+     * Callback when any `ReadableStream` is cancelled (the user doesn't need any more data),
+     * or `WritableStream` is ended (the user won't produce any more data),
+     * or `DuplexStreamFactory#close` is called.
+     *
+     * Usually you want to let the other peer know that the duplex stream should be clsoed.
+     *
+     * `dispose` will automatically be called after `close` completes,
+     * but if you want to wait another peer for a close confirmation and call
+     * `DuplexStreamFactory#dispose` yourself, you can return `false`
+     * (or a `Promise` that resolves to `false`) to disable the automatic call.
+     */
+    close?: (() => ValueOrPromise<boolean | void>) | undefined;
+
+    /**
+     * Callback when any `ReadableStream` is closed (the other peer doesn't produce any more data),
+     * or `WritableStream` is aborted (the other peer can't receive any more data),
+     * or `DuplexStreamFactory#abort` is called.
+     *
+     * Usually indicates the other peer has closed the duplex stream. You can clean up
+     * any resources you have allocated now.
+     */
+    dispose?: (() => void | Promise<void>) | undefined;
+}
+
+/**
+ * A factory for creating a duplex stream.
+ *
+ * It can create multiple `ReadableStream`s and `WritableStream`s,
+ * when any of them is closed, all other streams will be closed as well.
+ */
+export class DuplexStreamFactory<R, W> {
+    private readableControllers: ReadableStreamDefaultController<R>[] = [];
+    private writers: WritableStreamDefaultWriter<W>[] = [];
+
+    private _writableClosed = false;
+    public get writableClosed() { return this._writableClosed; }
+
+    private _closed = new PromiseResolver<void>();
+    public get closed() { return this._closed.promise; }
+
+    private options: DuplexStreamFactoryOptions;
+
+    public constructor(options?: DuplexStreamFactoryOptions) {
+        this.options = options ?? {};
+    }
+
+    public wrapReadable(readable: ReadableStream<R>): WrapReadableStream<R> {
+        return new WrapReadableStream<R>({
+            start: (controller) => {
+                this.readableControllers.push(controller);
+                return readable;
+            },
+            cancel: async () => {
+                // cancel means the local peer closes the connection first.
+                await this.close();
+            },
+            close: async () => {
+                // stream end means the remote peer closed the connection first.
+                await this.dispose();
+            },
+        });
+    }
+
+    public 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.ready;
+                await writer.write(chunk);
+            },
+            abort: async (reason) => {
+                await writer.abort(reason);
+                await this.close();
+            },
+            close: async () => {
+                try { await writer.close(); } catch { }
+                await this.close();
+            },
+        });
+    }
+
+    public async close() {
+        if (this._writableClosed) {
+            return;
+        }
+        this._writableClosed = true;
+
+        // Call `close` first, so it can still write data to `WritableStream`s.
+        if (await this.options.close?.() !== false) {
+            // `close` can return `false` to disable automatic `dispose`.
+            await this.dispose();
+        }
+
+        for (const writer of this.writers) {
+            try { await writer.close(); } catch { }
+        }
+    }
+
+    public async dispose() {
+        this._writableClosed = true;
+        this._closed.resolve();
+
+        for (const controller of this.readableControllers) {
+            try { controller.close(); } catch { }
+        }
+
+        await this.options.dispose?.();
+    }
+}

libraries/stream-extra/src/gather-string.ts

@@ -0,0 +1,15 @@
+import { WritableStream } from "./stream.js";
+
+export class GatherStringStream extends WritableStream<string>{
+    // PERF: rope (concat strings) is faster than `[].join('')`
+    private _result = '';
+    public get result() { return this._result; }
+
+    public constructor() {
+        super({
+            write: (chunk) => {
+                this._result += chunk;
+            },
+        });
+    }
+}

libraries/stream-extra/src/index.ts

@@ -0,0 +1,14 @@
+export * from './buffered.js';
+export * from './chunk.js';
+export * from './decode-utf8.js';
+export * from './duplex.js';
+export * from './gather-string.js';
+export * from './inspect.js';
+export * from './pipe-from.js';
+export * from './push-readable.js';
+export * from './split-string.js';
+export * from './stream.js';
+export * from './struct-deserialize.js';
+export * from './struct-serialize.js';
+export * from './wrap-readable.js';
+export * from './wrap-writable.js';

libraries/stream-extra/src/inspect.ts

@@ -0,0 +1,12 @@
+import { TransformStream } from "./stream.js";
+
+export class InspectStream<T> extends TransformStream<T, T> {
+    constructor(callback: (value: T) => void) {
+        super({
+            transform(chunk, controller) {
+                callback(chunk);
+                controller.enqueue(chunk);
+            }
+        });
+    }
+}

libraries/stream-extra/src/pipe-from.ts

@@ -0,0 +1,27 @@
+import { WritableStream, type ReadableWritablePair } from "./stream.js";
+
+/**
+ * Create a new `WritableStream` that, when written to, will write that chunk to
+ * `pair.writable`, when pipe `pair.readable` to `writable`.
+ *
+ * It's the opposite of `ReadableStream.pipeThrough`.
+ *
+ * @param writable The `WritableStream` to write to.
+ * @param pair A `TransformStream` that converts chunks.
+ * @returns A new `WritableStream`.
+ */
+export function pipeFrom<W, T>(writable: WritableStream<W>, pair: ReadableWritablePair<W, T>) {
+    const writer = pair.writable.getWriter();
+    const pipe = pair.readable
+        .pipeTo(writable);
+    return new WritableStream<T>({
+        async write(chunk) {
+            await writer.ready;
+            await writer.write(chunk);
+        },
+        async close() {
+            await writer.close();
+            await pipe;
+        }
+    });
+}

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

@@ -0,0 +1,62 @@
+import { PromiseResolver } from '@yume-chan/async';
+import { AbortController, AbortSignal, QueuingStrategy, ReadableStream } from "./stream.js";
+
+export interface PushReadableStreamController<T> {
+    abortSignal: AbortSignal;
+
+    enqueue(chunk: T): Promise<void>;
+
+    close(): void;
+
+    error(e?: any): void;
+}
+
+export type PushReadableStreamSource<T> = (controller: PushReadableStreamController<T>) => void;
+
+export class PushReadableStream<T> extends ReadableStream<T> {
+    public constructor(source: PushReadableStreamSource<T>, strategy?: QueuingStrategy<T>) {
+        let waterMarkLow: PromiseResolver<void> | undefined;
+        const canceled: AbortController = new AbortController();
+
+        super({
+            start: (controller) => {
+                source({
+                    abortSignal: canceled.signal,
+                    async enqueue(chunk) {
+                        if (canceled.signal.aborted) {
+                            // If the stream is already cancelled,
+                            // throw immediately.
+                            throw canceled.signal.reason ?? new Error('Aborted');
+                        }
+
+                        // Only when the stream is errored, `desiredSize` will be `null`.
+                        // But since `null <= 0` is `true`
+                        // (`null <= 0` is evaluated as `!(null > 0)` => `!false` => `true`),
+                        // not handling it will cause a deadlock.
+                        if ((controller.desiredSize ?? 1) <= 0) {
+                            waterMarkLow = new PromiseResolver<void>();
+                            await waterMarkLow.promise;
+                        }
+
+                        // `controller.enqueue` will throw error for us
+                        // if the stream is already errored.
+                        controller.enqueue(chunk);
+                    },
+                    close() {
+                        controller.close();
+                    },
+                    error(e) {
+                        controller.error(e);
+                    },
+                });
+            },
+            pull: () => {
+                waterMarkLow?.resolve();
+            },
+            cancel: async (reason) => {
+                canceled.abort(reason);
+                waterMarkLow?.reject(reason);
+            },
+        }, strategy);
+    }
+}

libraries/stream-extra/src/split-string.ts

@@ -0,0 +1,29 @@
+import { TransformStream } from "./stream.js";
+
+function* split(input: string, separator: string): Generator<string, void, void> {
+    let start = 0;
+
+    while (true) {
+        const index = input.indexOf(separator, start);
+        if (index === -1) {
+            return;
+        }
+
+        const part = input.substring(start, index);
+        yield part;
+
+        start = index + 1;
+    }
+}
+
+export class SplitStringStream extends TransformStream<string, string> {
+    public constructor(separator: string) {
+        super({
+            transform(chunk, controller) {
+                for (const part of split(chunk, separator)) {
+                    controller.enqueue(part);
+                }
+            }
+        });
+    }
+}

libraries/stream-extra/src/struct-deserialize.ts

@@ -0,0 +1,52 @@
+import type Struct from "@yume-chan/struct";
+import type { StructValueType } from "@yume-chan/struct";
+import { BufferedStream, BufferedStreamEndedError } from "./buffered.js";
+import { PushReadableStream, PushReadableStreamController } from "./push-readable.js";
+import { ReadableStream, WritableStream, type ReadableWritablePair } from "./stream.js";
+
+// TODO: StructTransformStream: Looking for better implementation
+export class StructDeserializeStream<T extends Struct<any, any, any, any>>
+    implements ReadableWritablePair<Uint8Array, StructValueType<T>>{
+    private _readable: ReadableStream<StructValueType<T>>;
+    public get readable() { return this._readable; }
+
+    private _writable: WritableStream<Uint8Array>;
+    public get writable() { return this._writable; }
+
+    public constructor(struct: T) {
+        // Convert incoming chunks to a `BufferedStream`
+        let incomingStreamController!: PushReadableStreamController<Uint8Array>;
+        const incomingStream = new BufferedStream(
+            new PushReadableStream<Uint8Array>(
+                controller => incomingStreamController = controller,
+            )
+        );
+
+        this._readable = new ReadableStream<StructValueType<T>>({
+            async pull(controller) {
+                try {
+                    const value = await struct.deserialize(incomingStream);
+                    controller.enqueue(value);
+                } catch (e) {
+                    if (e instanceof BufferedStreamEndedError) {
+                        controller.close();
+                        return;
+                    }
+                    throw e;
+                }
+            }
+        });
+
+        this._writable = new WritableStream({
+            async write(chunk) {
+                await incomingStreamController.enqueue(chunk);
+            },
+            abort() {
+                incomingStreamController.close();
+            },
+            close() {
+                incomingStreamController.close();
+            },
+        });
+    }
+}

libraries/stream-extra/src/struct-serialize.ts

@@ -0,0 +1,13 @@
+import type Struct from "@yume-chan/struct";
+import { TransformStream } from "./stream.js";
+
+export class StructSerializeStream<T extends Struct<any, any, any, any>>
+    extends TransformStream<T['TInit'], Uint8Array>{
+    constructor(struct: T) {
+        super({
+            transform(chunk, controller) {
+                controller.enqueue(struct.serialize(chunk));
+            },
+        });
+    }
+}

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

@@ -0,0 +1,70 @@
+import type { ValueOrPromise } from "@yume-chan/struct";
+import { ReadableStream, ReadableStreamDefaultController, ReadableStreamDefaultReader } from "./stream.js";
+
+export type WrapReadableStreamStart<T> = (controller: ReadableStreamDefaultController<T>) => ValueOrPromise<ReadableStream<T>>;
+
+export interface ReadableStreamWrapper<T> {
+    start: WrapReadableStreamStart<T>;
+    cancel?(reason?: any): ValueOrPromise<void>;
+    close?(): ValueOrPromise<void>;
+}
+
+function getWrappedReadableStream<T>(
+    wrapper: ReadableStream<T> | WrapReadableStreamStart<T> | ReadableStreamWrapper<T>,
+    controller: ReadableStreamDefaultController<T>
+) {
+    if ('start' in wrapper) {
+        return wrapper.start(controller);
+    } else if (typeof wrapper === 'function') {
+        return wrapper(controller);
+    } else {
+        // Can't use `wrapper instanceof ReadableStream`
+        // Because we want to be compatible with any ReadableStream-like objects
+        return wrapper;
+    }
+}
+
+/**
+ * This class has multiple usages:
+ *
+ * 1. Get notified when the stream is cancelled or closed.
+ * 2. Synchronously create a `ReadableStream` by asynchronously return another `ReadableStream`.
+ * 3. Convert native `ReadableStream`s to polyfilled ones so they can `pipe` between.
+ */
+export class WrapReadableStream<T> extends ReadableStream<T>{
+    public readable!: ReadableStream<T>;
+
+    private reader!: ReadableStreamDefaultReader<T>;
+
+    public constructor(wrapper: ReadableStream<T> | WrapReadableStreamStart<T> | ReadableStreamWrapper<T>) {
+        super({
+            start: async (controller) => {
+                // `start` is invoked before `ReadableStream`'s constructor finish,
+                // so using `this` synchronously causes
+                // "Must call super constructor in derived class before accessing 'this' or returning from derived constructor".
+                // Queue a microtask to avoid this.
+                await Promise.resolve();
+
+                this.readable = await getWrappedReadableStream(wrapper, controller);
+                this.reader = this.readable.getReader();
+            },
+            cancel: async (reason) => {
+                await this.reader.cancel(reason);
+                if ('cancel' in wrapper) {
+                    await wrapper.cancel?.(reason);
+                }
+            },
+            pull: async (controller) => {
+                const result = await this.reader.read();
+                if (result.done) {
+                    controller.close();
+                    if ('close' in wrapper) {
+                        await wrapper.close?.();
+                    }
+                } else {
+                    controller.enqueue(result.value);
+                }
+            }
+        });
+    }
+}

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

@@ -0,0 +1,65 @@
+import type { ValueOrPromise } from "@yume-chan/struct";
+import { WritableStream, WritableStreamDefaultWriter } from "./stream.js";
+
+export type WrapWritableStreamStart<T> = () => ValueOrPromise<WritableStream<T>>;
+
+export interface WritableStreamWrapper<T> {
+    start: WrapWritableStreamStart<T>;
+    close?(): Promise<void>;
+}
+
+async function getWrappedWritableStream<T>(
+    wrapper: WritableStream<T> | WrapWritableStreamStart<T> | WritableStreamWrapper<T>
+) {
+    if ('start' in wrapper) {
+        return await wrapper.start();
+    } else if (typeof wrapper === 'function') {
+        return await wrapper();
+    } else {
+        // Can't use `wrapper instanceof WritableStream`
+        // Because we want to be compatible with any WritableStream-like objects
+        return wrapper;
+    }
+}
+
+export class WrapWritableStream<T> extends WritableStream<T> {
+    public writable!: WritableStream<T>;
+
+    private writer!: WritableStreamDefaultWriter<T>;
+
+    public constructor(wrapper: WritableStream<T> | WrapWritableStreamStart<T> | WritableStreamWrapper<T>) {
+        super({
+            start: async () => {
+                // `start` is invoked before `ReadableStream`'s constructor finish,
+                // so using `this` synchronously causes
+                // "Must call super constructor in derived class before accessing 'this' or returning from derived constructor".
+                // Queue a microtask to avoid this.
+                await Promise.resolve();
+
+                this.writable = await getWrappedWritableStream(wrapper);
+                this.writer = this.writable.getWriter();
+            },
+            write: async (chunk) => {
+                // Maintain back pressure
+                await this.writer.ready;
+                await this.writer.write(chunk);
+            },
+            abort: async (reason) => {
+                await this.writer.abort(reason);
+                if ('close' in wrapper) {
+                    await wrapper.close?.();
+                }
+            },
+            close: async () => {
+                // Close the inner stream first.
+                // Usually the inner stream is a logical sub-stream over the outer stream,
+                // closing the outer stream first will make the inner stream incapable of
+                // sending data in its `close` handler.
+                await this.writer.close();
+                if ('close' in wrapper) {
+                    await wrapper.close?.();
+                }
+            },
+        });
+    }
+}

libraries/stream-extra/tsconfig.build.json

@@ -0,0 +1,3 @@
+{
+    "extends": "./node_modules/@yume-chan/ts-package-builder/tsconfig.base.json"
+}

libraries/stream-extra/tsconfig.json

@@ -0,0 +1,10 @@
+{
+    "references": [
+        {
+            "path": "./tsconfig.test.json"
+        },
+        {
+            "path": "./tsconfig.build.json"
+        },
+    ]
+}

libraries/stream-extra/tsconfig.test.json

@@ -0,0 +1,8 @@
+{
+    "extends": "./tsconfig.build.json",
+    "compilerOptions": {
+        "types": [
+        ],
+    },
+    "exclude": []
+}

rush.json

@@ -489,6 +489,11 @@
             "projectFolder": "libraries/scrcpy",
             "versionPolicyName": "adb"
         },
+        {
+            "packageName": "@yume-chan/stream-extra",
+            "projectFolder": "libraries/stream-extra",
+            "versionPolicyName": "adb"
+        },
         {
             "packageName": "demo",
             "projectFolder": "apps/demo"