fix(stream): re-export global TextDecoder stream instead of implementing our own one which doesn't work correctly in stream mode

common/changes/@yume-chan/stream-extra/main_2024-05-10-00-32.json

@@ -0,0 +1,10 @@
+{
+    "changes": [
+        {
+            "packageName": "@yume-chan/stream-extra",
+            "comment": "Re-export global `TextDecoderStream` instead of implementing our own one which doesn't work correctly in stream mode",
+            "type": "none"
+        }
+    ],
+    "packageName": "@yume-chan/stream-extra"
+}

libraries/adb-scrcpy/src/client.ts

@@ -29,10 +29,10 @@ import type {
 import {
     AbortController,
     BufferedReadableStream,
-    DecodeUtf8Stream,
     InspectStream,
     PushReadableStream,
     SplitStringStream,
+    TextDecoderStream,
     WritableStream,
 } from "@yume-chan/stream-extra";
 
@@ -158,7 +158,7 @@ export class AdbScrcpyClient {
             );
 
             const stdout = process.stdout
-                .pipeThrough(new DecodeUtf8Stream())
+                .pipeThrough(new TextDecoderStream())
                 .pipeThrough(new SplitStringStream("\n"));
 
             // Must read all streams, otherwise the whole connection will be blocked.

libraries/adb/src/adb.ts

@@ -2,7 +2,7 @@ import type {
     MaybeConsumable,
     ReadableWritablePair,
 } from "@yume-chan/stream-extra";
-import { ConcatStringStream, DecodeUtf8Stream } from "@yume-chan/stream-extra";
+import { ConcatStringStream, TextDecoderStream } from "@yume-chan/stream-extra";
 import type { ValueOrPromise } from "@yume-chan/struct";
 
 import type { AdbBanner } from "./banner.js";
@@ -112,7 +112,7 @@ export class Adb implements Closeable {
     async createSocketAndWait(service: string): Promise<string> {
         const socket = await this.createSocket(service);
         return await socket.readable
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new ConcatStringStream());
     }
 

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

@@ -1,4 +1,4 @@
-import { ConcatStringStream, DecodeUtf8Stream } from "@yume-chan/stream-extra";
+import { ConcatStringStream, TextDecoderStream } from "@yume-chan/stream-extra";
 
 import { AdbCommandBase } from "../base.js";
 
@@ -112,10 +112,10 @@ export class AdbSubprocess extends AdbCommandBase {
 
         const [stdout, stderr, exitCode] = await Promise.all([
             process.stdout
-                .pipeThrough(new DecodeUtf8Stream())
+                .pipeThrough(new TextDecoderStream())
                 .pipeThrough(new ConcatStringStream()),
             process.stderr
-                .pipeThrough(new DecodeUtf8Stream())
+                .pipeThrough(new TextDecoderStream())
                 .pipeThrough(new ConcatStringStream()),
             process.exit,
         ]);

libraries/android-bin/src/am.ts

@@ -1,6 +1,6 @@
 import type { Adb } from "@yume-chan/adb";
 import { AdbCommandBase } from "@yume-chan/adb";
-import { ConcatStringStream, DecodeUtf8Stream } from "@yume-chan/stream-extra";
+import { ConcatStringStream, TextDecoderStream } from "@yume-chan/stream-extra";
 
 import { Cmd } from "./cmd.js";
 import type { IntentBuilder } from "./intent.js";
@@ -53,7 +53,7 @@ export class ActivityManager extends AdbCommandBase {
         const process = await this.#cmdOrSubprocess(args);
 
         const output = await process.stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new ConcatStringStream())
             .then((output) => output.trim());
 

libraries/android-bin/src/bu.ts

@@ -1,6 +1,6 @@
 import { AdbCommandBase } from "@yume-chan/adb";
 import type { MaybeConsumable, ReadableStream } from "@yume-chan/stream-extra";
-import { ConcatStringStream, DecodeUtf8Stream } from "@yume-chan/stream-extra";
+import { ConcatStringStream, TextDecoderStream } from "@yume-chan/stream-extra";
 
 export interface AdbBackupOptions {
     user: number;
@@ -71,7 +71,7 @@ export class AdbBackup extends AdbCommandBase {
         const process = await this.adb.subprocess.spawn(args);
         const [output] = await Promise.all([
             process.stdout
-                .pipeThrough(new DecodeUtf8Stream())
+                .pipeThrough(new TextDecoderStream())
                 .pipeThrough(new ConcatStringStream()),
             options.file.pipeTo(process.stdin),
         ]);

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

@@ -6,9 +6,9 @@ import { AdbCommandBase, AdbSubprocessShellProtocol } from "@yume-chan/adb";
 import type { AbortSignal, ReadableStream } from "@yume-chan/stream-extra";
 import {
     AbortController,
-    DecodeUtf8Stream,
     PushReadableStream,
     SplitStringStream,
+    TextDecoderStream,
     WrapReadableStream,
     WritableStream,
 } from "@yume-chan/stream-extra";
@@ -212,7 +212,7 @@ export class BugReport extends AdbCommandBase {
         let error: string | undefined;
 
         await process.stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new SplitStringStream("\n"))
             .pipeTo(
                 new WritableStream<string>({
@@ -278,7 +278,7 @@ export class BugReport extends AdbCommandBase {
                     controller.error(e);
                 });
             process.stderr
-                .pipeThrough(new DecodeUtf8Stream())
+                .pipeThrough(new TextDecoderStream())
                 .pipeTo(
                     new WritableStream({
                         write(chunk) {

libraries/android-bin/src/cmd.ts

@@ -10,7 +10,7 @@ import {
     AdbSubprocessNoneProtocol,
     AdbSubprocessShellProtocol,
 } from "@yume-chan/adb";
-import { ConcatStringStream, DecodeUtf8Stream } from "@yume-chan/stream-extra";
+import { ConcatStringStream, TextDecoderStream } from "@yume-chan/stream-extra";
 
 export class Cmd extends AdbCommandBase {
     #supportsShellV2: boolean;
@@ -84,10 +84,10 @@ export class Cmd extends AdbCommandBase {
 
         const [stdout, stderr, exitCode] = await Promise.all([
             process.stdout
-                .pipeThrough(new DecodeUtf8Stream())
+                .pipeThrough(new TextDecoderStream())
                 .pipeThrough(new ConcatStringStream()),
             process.stderr
-                .pipeThrough(new DecodeUtf8Stream())
+                .pipeThrough(new TextDecoderStream())
                 .pipeThrough(new ConcatStringStream()),
             process.exit,
         ]);

libraries/android-bin/src/logcat.ts

@@ -5,8 +5,8 @@ import { AdbCommandBase, AdbSubprocessNoneProtocol } from "@yume-chan/adb";
 import type { ReadableStream } from "@yume-chan/stream-extra";
 import {
     BufferedTransformStream,
-    DecodeUtf8Stream,
     SplitStringStream,
+    TextDecoderStream,
     WrapReadableStream,
     WritableStream,
 } from "@yume-chan/stream-extra";
@@ -421,7 +421,7 @@ export class Logcat extends AdbCommandBase {
 
         const result: LogSize[] = [];
         await stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new SplitStringStream("\n"))
             .pipeTo(
                 new WritableStream({

libraries/android-bin/src/pm.ts

@@ -8,8 +8,8 @@ import { AdbCommandBase, escapeArg } from "@yume-chan/adb";
 import type { MaybeConsumable, ReadableStream } from "@yume-chan/stream-extra";
 import {
     ConcatStringStream,
-    DecodeUtf8Stream,
     SplitStringStream,
+    TextDecoderStream,
 } from "@yume-chan/stream-extra";
 
 import { Cmd } from "./cmd.js";
@@ -352,7 +352,7 @@ export class PackageManager extends AdbCommandBase {
         const process = await this.#cmd.spawn(false, "package", ...args);
 
         const output = process.stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new ConcatStringStream())
             .then((output) => output.trim());
 
@@ -440,7 +440,7 @@ export class PackageManager extends AdbCommandBase {
 
         const process = await this.#cmdOrSubprocess(args);
         const reader = process.stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new SplitStringStream("\n"))
             .getReader();
         while (true) {
@@ -468,7 +468,7 @@ export class PackageManager extends AdbCommandBase {
 
         const process = await this.#cmdOrSubprocess(args);
         const output = await process.stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new ConcatStringStream())
             .then((output) => output.trim());
         if (output !== "Success") {
@@ -489,7 +489,7 @@ export class PackageManager extends AdbCommandBase {
 
         const process = await this.#cmdOrSubprocess(args);
         const output = await process.stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new ConcatStringStream())
             .then((output) => output.trim());
 
@@ -515,7 +515,7 @@ export class PackageManager extends AdbCommandBase {
 
         const process = await this.#cmdOrSubprocess(args);
         const output = await process.stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new ConcatStringStream())
             .then((output) => output.trim());
 
@@ -562,7 +562,7 @@ export class PackageManager extends AdbCommandBase {
 
         const process = await this.#cmdOrSubprocess(args);
         const output = process.stdout
-            .pipeThrough(new DecodeUtf8Stream())
+            .pipeThrough(new TextDecoderStream())
             .pipeThrough(new ConcatStringStream())
             .then((output) => output.trim());
 

libraries/stream-extra/src/concat.ts

@@ -82,8 +82,8 @@ export interface ConcatBufferReadableStream
  * A `TransformStream` that concatenates `Uint8Array`s.
  *
  * If you want to decode the result as string,
- * prefer `.pipeThrough(new DecodeUtf8Stream()).pipeThrough(new ConcatStringStream())`,
- * than `.pipeThough(new ConcatBufferStream()).pipeThrough(new DecodeUtf8Stream())`,
+ * prefer `.pipeThrough(new TextDecoderStream()).pipeThrough(new ConcatStringStream())`,
+ * than `.pipeThough(new ConcatBufferStream()).pipeThrough(new TextDecoderStream())`,
  * because of JavaScript engine optimizations,
  * concatenating strings is faster than concatenating `Uint8Array`s.
  */

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

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

libraries/stream-extra/src/encoding.ts

@@ -0,0 +1,39 @@
+import type { TransformStream } from "./stream.js";
+
+export interface TextDecoderOptions {
+    fatal?: boolean;
+    ignoreBOM?: boolean;
+}
+
+declare class TextDecoderStreamType extends TransformStream<
+    ArrayBufferView | ArrayBuffer,
+    string
+> {
+    constructor(label?: string, options?: TextDecoderOptions);
+
+    readonly encoding: string;
+    readonly fatal: boolean;
+    readonly ignoreBOM: boolean;
+}
+
+declare class TextEncoderStreamType extends TransformStream<
+    string,
+    Uint8Array
+> {
+    constructor();
+
+    readonly encoding: string;
+}
+
+interface GlobalExtension {
+    TextDecoderStream: typeof TextDecoderStreamType;
+    TextEncoderStream: typeof TextEncoderStreamType;
+}
+
+const Global = globalThis as unknown as GlobalExtension;
+
+export const TextDecoderStream = Global.TextDecoderStream;
+export type TextDecoderStream = TextDecoderStreamType;
+
+export const TextEncoderStream = Global.TextEncoderStream;
+export type TextEncoderStream = TextEncoderStreamType;

libraries/stream-extra/src/index.ts

@@ -2,9 +2,9 @@ export * from "./buffered-transform.js";
 export * from "./buffered.js";
 export * from "./concat.js";
 export * from "./consumable.js";
-export * from "./decode-utf8.js";
 export * from "./distribution.js";
 export * from "./duplex.js";
+export * from "./encoding.js";
 export * from "./inspect.js";
 export * from "./maybe-consumable.js";
 export * from "./pipe-from.js";