Fix text decoding in streams with multiple chunks

parser.ts

@@ -162,11 +162,16 @@ export interface SAXEvent {
 /**
  * SAX-style XML parser.
  */
-export class SAXParser extends ParserBase implements UnderlyingSink<Uint8Array> {
+export class SAXParser extends ParserBase implements UnderlyingSink<Uint8Array>, UnderlyingSink<string> {
     // deno-lint-ignore no-explicit-any
     private _listeners: { [name: string]: ((...arg: any[]) => void)[] } = {};
     private _controller?: WritableStreamDefaultController;
-    private _encoding?: string;
+    private _decoder: TextDecoder;
+
+    constructor(encoding: string = "UTF-8") {
+        super();
+        this._decoder = new TextDecoder(encoding);
+    }
 
     protected fireListeners(event: XMLParseEvent) {
         const [name, ...args] = event;
@@ -204,11 +209,29 @@ export class SAXParser extends ParserBase implements UnderlyingSink<Uint8Array>
      * @param chunk XML data chunk
      * @param controller error reporter, Deno writable stream uses internal.
      */
-    write(chunk: Uint8Array, controller?: WritableStreamDefaultController) {
+    write(chunk: Uint8Array|string, controller?: WritableStreamDefaultController) {
         try {
             this._controller = controller;
-            // TextDecoder can resolve BOM.
-            this.chunk = new TextDecoder(this._encoding).decode(chunk);
+            if (typeof chunk === 'string') {
+                this.chunk = chunk;
+            }
+            else {
+                // TextDecoder can resolve BOM.
+                this.chunk = this._decoder.decode(chunk, {stream: true});
+            }
+            this.run();
+        } finally {
+            this._controller = undefined;
+        }
+    }
+
+    close(controller?: WritableStreamDefaultController) {
+        try {
+            this._controller = controller;
+            // Process any remaining data still pending in `_decoder`
+            // Even if `_decoder` was unused because this parser processed strings,
+            // processing this empty chunk doesn't hurt.
+            this.chunk = this._decoder.decode(new Uint8Array(), {stream: false});
             this.run();
         } finally {
             this._controller = undefined;
@@ -220,18 +243,21 @@ export class SAXParser extends ParserBase implements UnderlyingSink<Uint8Array>
      * @param source Target XML.
      * @param encoding When the source is Deno.Reader or Uint8Array, specify the encoding.
      */
-    async parse(source: ReadableStream<unknown> | Uint8Array | string, encoding?: string) {
-        this._encoding = encoding;
+    async parse(source: ReadableStream<Uint8Array> | ReadableStream<string> | Uint8Array | string, encoding?: string /** @deprecated Use constructor parameter instead */) {
+        if (encoding !== undefined) {
+            this._decoder = new TextDecoder(encoding);
+        }
         if (typeof source === 'string') {
             this.chunk = source;
             this.run();
         } else if (source instanceof Uint8Array) {
             this.write(source);
+            this.close();
         } else {
-            await source.pipeThrough(
-                new TextDecoderStream(this._encoding)
-            ).pipeTo(
-                new WritableStream<string>({ write: str => this.parse(str, encoding) }),
+            type ReadableStreamContent<T> = T extends ReadableStream<infer Content> ? Content : never;
+            type SourceStreamContent = ReadableStreamContent<typeof source>;
+            await source.pipeTo(
+                new WritableStream<SourceStreamContent>(this),
             );
         }
     }

parser_test.ts

@@ -20,6 +20,52 @@ import {
     PullResult,
 } from './parser.ts';
 
+function byteChunkStream(bytes: Uint8Array): ReadableStream<Uint8Array> {
+    return new ReadableStream<Uint8Array>({
+        start(controller: ReadableStreamDefaultController) {
+            for (let pos = 0; pos < bytes.length; ++pos) {
+                controller.enqueue(bytes.subarray(pos, pos+1));
+            }
+            controller.close();
+        }
+    });
+}
+
+function charChunkStream(str: string): ReadableStream<string> {
+    return new ReadableStream<string>({
+        start(controller: ReadableStreamDefaultController) {
+            for (let pos = 0; pos < str.length; ++pos) {
+                controller.enqueue(str.substring(pos, pos+1));
+            }
+            controller.close();
+        }
+    });
+}
+
+Deno.test('byteChunkStream writes single byte chunks', async () => {
+    const input = (new TextEncoder()).encode("abc");
+    const stream = byteChunkStream(input);
+    const output = new Uint8Array(input.length);
+    let outputpos = 0;
+    for await (const chunk of stream) {
+        assertEquals(chunk.length, 1);
+        output.set(chunk, outputpos);
+        outputpos += chunk.length;
+    }
+    assertEquals(output, input);
+});
+
+Deno.test('charChunkStream writes single char chunks', async () => {
+    const input = "abc";
+    const stream = charChunkStream(input);
+    let output = "";
+    for await (const chunk of stream) {
+        assertEquals(chunk.length, 1);
+        output = output + chunk;
+    }
+    assertEquals(output, input);
+});
+
 Deno.test('ParserBase chunk & hasNext & readNext & position', () => {
     // protected -> public visiblity
     class TestParser extends ParserBase {
@@ -74,6 +120,36 @@ Deno.test('SAXParser on & parse(Deno.Reader)', async () => {
     assertEquals(elementCount, 18);
 });
 
+Deno.test('SAXParser UnderlyingSink chunks', async () => {
+    const parser = new SAXParser();
+
+    const input = (new TextEncoder()).encode("<x>ä</x>");
+    parser.on('text', (text) => {
+        assertEquals(text, "\u00E4");
+    });
+    await byteChunkStream(input).pipeTo(new WritableStream(parser));
+});
+
+Deno.test('SAXParser parse(ReadableStream<Uint8Array>)', async () => {
+    const parser = new SAXParser();
+    parser.on('text', (text) => {
+        assertEquals(text, "\u00E4");
+    });
+
+    const input = (new TextEncoder()).encode("<x>ä</x>");
+    await parser.parse(byteChunkStream(input));
+});
+
+Deno.test('SAXParser parse(ReadableStream<string>)', async () => {
+    const parser = new SAXParser();
+    parser.on('text', (text) => {
+        assertEquals(text, "\u00E4");
+    });
+
+    const input = "<x>ä</x>";
+    await parser.parse(charChunkStream(input));
+});
+
 Deno.test('SAXParser parse(Uint8Array)', () => {
     const parser = new SAXParser();
     let flag = false;