@smithy/core - npm Package Compare versions

Comparing version 3.29.0 to 3.29.1

dist-cjs/submodules/serde/index.js

@@ -5,3 +5,3 @@ const { createHmac, createHash, getRandomValues } = require("node:crypto");
 const { toEndpointV1 } = require("@smithy/core/endpoints");
-const { Duplex, Readable, Writable, PassThrough } = require("node:stream");
+const { Readable, Writable, PassThrough } = require("node:stream");
 
@@ -991,3 +991,3 @@ const isArrayBuffer = (arg) => (typeof ArrayBuffer === "function" && arg instanceof ArrayBuffer) ||
 
-let ChecksumStream$1 = class ChecksumStream extends Duplex {
+let ChecksumStream$1 = class ChecksumStream extends Readable {
     expectedChecksum;
@@ -998,11 +998,8 @@ checksumSourceLocation;
     base64Encoder;
-    pendingCallback = null;
     constructor({ expectedChecksum, checksum, source, checksumSourceLocation, base64Encoder, }) {
         super();
-        if (typeof source.pipe === "function") {
-            this.source = source;
-        }
-        else {
+        if (typeof source.pipe !== "function") {
             throw new Error(`@smithy/util-stream: unsupported source type ${source?.constructor?.name ?? source} in ChecksumStream.`);
         }
+        this.source = source;
         this.base64Encoder = base64Encoder ?? toBase64$1;
@@ -1012,26 +1009,26 @@ this.expectedChecksum = expectedChecksum;
         this.checksumSourceLocation = checksumSourceLocation;
-        this.source.pipe(this);
+        this.source.on("data", this.onSourceData);
+        this.source.on("end", this.onSourceEnd);
+        this.source.on("error", this.onSourceError);
+        this.source.pause();
     }
-    _read(size) {
-        if (this.pendingCallback) {
-            const callback = this.pendingCallback;
-            this.pendingCallback = null;
-            callback();
+    onSourceData = (chunk) => {
+        if (this.destroyed) {
+            return;
         }
-    }
-    _write(chunk, encoding, callback) {
         try {
             this.checksum.update(chunk);
-            const canPushMore = this.push(chunk);
-            if (!canPushMore) {
-                this.pendingCallback = callback;
-                return;
-            }
         }
         catch (e) {
-            return callback(e);
+            this.destroy(e);
+            return;
         }
-        return callback();
-    }
-    async _final(callback) {
+        if (!this.push(chunk)) {
+            this.source.pause();
+        }
+    };
+    onSourceEnd = async () => {
+        if (this.destroyed) {
+            return;
+        }
         try {
@@ -1041,11 +1038,18 @@ const digest = await this.checksum.digest();
             if (this.expectedChecksum !== received) {
-                return callback(new Error(`Checksum mismatch: expected "${this.expectedChecksum}" but received "${received}"` +
+                this.destroy(new Error(`Checksum mismatch: expected "${this.expectedChecksum}" but received "${received}"` +
                     ` in response header "${this.checksumSourceLocation}".`));
+                return;
             }
         }
         catch (e) {
-            return callback(e);
+            this.destroy(e);
+            return;
         }
         this.push(null);
-        return callback();
+    };
+    onSourceError = (error) => {
+        this.destroy(error);
+    };
+    _read(size) {
+        this.source.resume();
     }
@@ -1052,0 +1056,0 @@ _destroy(error, callback) {

dist-es/submodules.../util-stream/checksum/ChecksumStream.js

@@ -1,4 +0,4 @@
-import { Duplex } from "node:stream";
+import { Readable } from "node:stream";
 import { toBase64 } from "../../util-base64/toBase64";
-export class ChecksumStream extends Duplex {
+export class ChecksumStream extends Readable {
     expectedChecksum;
@@ -9,11 +9,8 @@ checksumSourceLocation;
     base64Encoder;
-    pendingCallback = null;
     constructor({ expectedChecksum, checksum, source, checksumSourceLocation, base64Encoder, }) {
         super();
-        if (typeof source.pipe === "function") {
-            this.source = source;
-        }
-        else {
+        if (typeof source.pipe !== "function") {
             throw new Error(`@smithy/util-stream: unsupported source type ${source?.constructor?.name ?? source} in ChecksumStream.`);
         }
+        this.source = source;
         this.base64Encoder = base64Encoder ?? toBase64;
@@ -23,26 +20,26 @@ this.expectedChecksum = expectedChecksum;
         this.checksumSourceLocation = checksumSourceLocation;
-        this.source.pipe(this);
+        this.source.on("data", this.onSourceData);
+        this.source.on("end", this.onSourceEnd);
+        this.source.on("error", this.onSourceError);
+        this.source.pause();
     }
-    _read(size) {
-        if (this.pendingCallback) {
-            const callback = this.pendingCallback;
-            this.pendingCallback = null;
-            callback();
+    onSourceData = (chunk) => {
+        if (this.destroyed) {
+            return;
         }
-    }
-    _write(chunk, encoding, callback) {
         try {
             this.checksum.update(chunk);
-            const canPushMore = this.push(chunk);
-            if (!canPushMore) {
-                this.pendingCallback = callback;
-                return;
-            }
         }
         catch (e) {
-            return callback(e);
+            this.destroy(e);
+            return;
         }
-        return callback();
-    }
-    async _final(callback) {
+        if (!this.push(chunk)) {
+            this.source.pause();
+        }
+    };
+    onSourceEnd = async () => {
+        if (this.destroyed) {
+            return;
+        }
         try {
@@ -52,11 +49,18 @@ const digest = await this.checksum.digest();
             if (this.expectedChecksum !== received) {
-                return callback(new Error(`Checksum mismatch: expected "${this.expectedChecksum}" but received "${received}"` +
+                this.destroy(new Error(`Checksum mismatch: expected "${this.expectedChecksum}" but received "${received}"` +
                     ` in response header "${this.checksumSourceLocation}".`));
+                return;
             }
         }
         catch (e) {
-            return callback(e);
+            this.destroy(e);
+            return;
         }
         this.push(null);
-        return callback();
+    };
+    onSourceError = (error) => {
+        this.destroy(error);
+    };
+    _read(size) {
+        this.source.resume();
     }
@@ -63,0 +67,0 @@ _destroy(error, callback) {

dist-types/submodu...til-stream/checksum/ChecksumStream.d.ts

@@ -1,2 +0,2 @@
-import { Duplex, type Readable } from "node:stream";
+import { Readable } from "node:stream";
 import type { Checksum, Encoder } from "@smithy/types";
@@ -32,30 +32,38 @@ /**
  *
+ * Note: this effectively behaves as a duplex, reading from the source on one
+ * side and forwarding chunks to its own readable side on the other. It should
+ * not be rewritten back into a Duplex (or Transform). The source is observed
+ * and driven manually (pause/resume in onSourceData/_read) so data is pulled
+ * at the rate it is consumed and never buffered twice; this manual control is
+ * used deliberately for performance and would be lost with the built-in duplex
+ * machinery.
+ *
  * @internal
  */
-export declare class ChecksumStream extends Duplex {
+export declare class ChecksumStream extends Readable {
     private expectedChecksum;
     private checksumSourceLocation;
     private checksum;
-    private source?;
+    private source;
     private base64Encoder;
-    private pendingCallback;
     constructor({ expectedChecksum, checksum, source, checksumSourceLocation, base64Encoder, }: ChecksumStreamInit<Readable>);
     /**
-     * Do not call this directly.
-     * @internal
+     * Update the checksum and forward each source chunk to the readable side,
+     * pausing the source when the readable side signals backpressure.
      */
-    _read(size: number): void;
+    private onSourceData;
     /**
-     * When the upstream source flows data to this stream,
-     * calculate a step update of the checksum.
-     * Do not call this directly.
-     * @internal
+     * When the source finishes, perform the checksum comparison and end this stream.
      */
-    _write(chunk: Buffer, encoding: string, callback: (err?: Error) => void): void;
+    private onSourceEnd;
     /**
-     * When the upstream source finishes, perform the checksum comparison.
+     * Surface source errors on this stream.
+     */
+    private onSourceError;
+    /**
+     * Resume the source so it flows at the rate this stream is consumed.
      * Do not call this directly.
      * @internal
      */
-    _final(callback: (err?: Error) => void): Promise<void>;
+    _read(size: number): void;
     /**
@@ -62,0 +70,0 @@ * Destroy the upstream source for cleanup so it is not left dangling, then

package.json

 {
   "name": "@smithy/core",
-  "version": "3.29.0",
+  "version": "3.29.1",
   "scripts": {
@@ -5,0 +5,0 @@ "benchmark:cbor": "node ./scripts/cbor-perf.mjs",

New alerts

AI-detected potential code anomaly

Supply chain risk

AI has identified unusual behaviors that may pose a security risk.

URL strings

Supply chain risk

Package contains fragments of external URLs or IP addresses, which the package may be accessing at runtime.

Fixed alerts

AI-detected potential code anomaly

Supply chain risk

AI has identified unusual behaviors that may pose a security risk.

URL strings

Supply chain risk

Package contains fragments of external URLs or IP addresses, which the package may be accessing at runtime.

Improved metrics

Total package byte prevSize

1359352

0.04%

Lines of code

34265

0.05%

No dependency changes