move TransformStream checksum to flush event

Author: kuhe

packages/util-stream/src/checksum/createChecksumStream.browser.ts

@@ -35,6 +35,17 @@ export interface ChecksumStreamInit {
  */
 export type ReadableStreamType = ReadableStream;
 
+/**
+ * This is a local copy of
+ * https://developer.mozilla.org/en-US/docs/Web/API/TransformStreamDefaultController
+ * in case users do not have this type.
+ */
+interface TransformStreamDefaultController {
+  enqueue(chunk: any): void;
+  error(error: unknown): void;
+  terminate(): void;
+}
+
 /**
  * @internal
  *
@@ -64,33 +75,28 @@ export const createChecksumStream = ({
 
   const transform = new TransformStream({
     start() {},
-    transform: async (chunk: any, controller: TransformStreamDefaultController) => {
-      /**
-       * When the upstream source finishes, perform the checksum comparison.
-       */
-      if (null === chunk) {
-        const digest: Uint8Array = await checksum.digest();
-        const received = encoder(digest);
-
-        if (expectedChecksum !== received) {
-          const error = new Error(
-            `Checksum mismatch: expected "${expectedChecksum}" but received "${received}"` +
-              ` in response header "${checksumSourceLocation}".`
-          );
-          controller.error(error);
-          throw error;
-        }
-        controller.terminate();
-        return;
-      }
+    async transform(chunk: any, controller: TransformStreamDefaultController) {
       /**
        * When the upstream source flows data to this stream,
        * calculate a step update of the checksum.
        */
       checksum.update(chunk);
       controller.enqueue(chunk);
     },
-    flush() {},
+    async flush(controller: TransformStreamDefaultController) {
+      const digest: Uint8Array = await checksum.digest();
+      const received = encoder(digest);
+
+      if (expectedChecksum !== received) {
+        const error = new Error(
+          `Checksum mismatch: expected "${expectedChecksum}" but received "${received}"` +
+            ` in response header "${checksumSourceLocation}".`
+        );
+        controller.error(error);
+      } else {
+        controller.terminate();
+      }
+    },
   });
 
   source.pipeThrough(transform);

packages/util-stream/src/checksum/createChecksumStream.spec.ts

@@ -43,6 +43,7 @@ describe("Checksum streams", () => {
       });
 
       expect(checksumStream.constructor.name).toEqual("ChecksumStream");
+      expect(checksumStream).toBeInstanceOf(Readable);
 
       const collected = toUtf8(await headStream(checksumStream, Infinity));
       expect(collected).toEqual(canonicalUtf8);
@@ -84,7 +85,6 @@ describe("Checksum streams", () => {
           canonicalData.forEach((byte) => {
             controller.enqueue(new Uint8Array([byte]));
           });
-          controller.enqueue(null);
           controller.close();
         },
       });

packages/util-stream/src/checksum/createChecksumStream.ts

@@ -103,32 +103,36 @@ class ChecksumStream extends Duplex {
    * calculate a step update of the checksum.
    */
   public _write(chunk: Buffer, encoding: string, callback: (err?: Error) => void): void {
-    this.checksum.update(chunk);
-    this.push(chunk);
-    callback();
+    try {
+      this.checksum.update(chunk);
+      this.push(chunk);
+    } catch (e: unknown) {
+      return callback(e as Error);
+    }
+    return callback();
   }
 
   /**
    * @internal do not call this directly.
    *
    * When the upstream source finishes, perform the checksum comparison.
    */
-  public async _final(callback: (err?: Error) => void) {
+  public async _final(callback: (err?: Error) => void): Promise<void> {
     try {
       const digest: Uint8Array = await this.checksum.digest();
       const received = this.base64Encoder(digest);
       if (this.expectedChecksum !== received) {
-        callback(
+        return callback(
           new Error(
             `Checksum mismatch: expected "${this.expectedChecksum}" but received "${received}"` +
               ` in response header "${this.checksumSourceLocation}".`
           )
         );
       }
     } catch (e: unknown) {
-      callback(e as Error);
+      return callback(e as Error);
     }
     this.push(null);
-    callback();
+    return callback();
   }
 }