Fix progress reporting causing stream data to be consumed prior to fetch

Author: ljwagerfield

src/private/UploadManagerBase.ts

@@ -92,7 +92,7 @@ export abstract class UploadManagerBase<TSource, TInit> implements UploadManager
     part: UploadPart,
     contentLength: number,
     source: TSource,
-    onProgress: (bytesSentDelta: number) => void,
+    onProgress: (totalBytesTransferred: number) => void,
     addCancellationHandler: AddCancellationHandler
   ): Promise<PutUploadPartResult>;
 

src/public/node/ProgressStream.ts

@@ -0,0 +1,32 @@
+import { Transform, TransformCallback, TransformOptions } from "stream";
+
+interface ProgressStreamOptions extends TransformOptions {
+  /**
+   * Optional callback to handle progress updates.
+   * @param totalBytesTransferred - The total number of bytes transferred so far.
+   */
+  onProgress?: (totalBytesTransferred: number) => void;
+}
+
+export class ProgressStream extends Transform {
+  private readonly onProgress: ((bytesTransferred: number) => void) | undefined;
+  private totalBytes: number = 0;
+
+  constructor(options?: ProgressStreamOptions) {
+    super(options);
+    this.onProgress = options?.onProgress;
+  }
+
+  _transform(chunk: Buffer, _encoding: BufferEncoding, callback: TransformCallback): void {
+    this.totalBytes += chunk.length;
+
+    // Invoke the progress callback if provided
+    if (this.onProgress !== undefined) {
+      this.onProgress(this.totalBytes);
+    }
+
+    // Pass the chunk along
+    this.push(chunk);
+    callback();
+  }
+}

src/public/node/UploadManagerNode.ts

@@ -9,6 +9,7 @@ import { AddCancellationHandler } from "../../private/model/AddCancellationHandl
 import { StreamUtils } from "../../private/StreamUtils";
 import { BlobLike, UploadManagerParams, UploadSource } from "../shared/CommonTypes";
 import { UploadManagerFetchUtils } from "../../private/UploadManagerFetchUtils";
+import { ProgressStream } from "./ProgressStream";
 
 type UploadManagerNodeInit =
   | undefined
@@ -96,27 +97,17 @@ export class UploadManager extends UploadManagerBase<UploadSourceProcessedNode,
     part: UploadPart,
     contentLength: number,
     source: UploadSourceProcessedNode,
-    onProgress: (bytesSentDelta: number) => void,
+    onProgress: (totalBytesTransferred: number) => void,
     addCancellationHandler: AddCancellationHandler
   ): Promise<{ etag: string | undefined; status: number }> {
-    // Report progress:
-    let hasWarned = false;
-    let bytesSent = 0;
-    const stream = await this.sliceDataForRequest(source, part);
-    const streamWithProgress = stream.on("data", (data: Partial<Buffer>) => {
-      if (data.byteLength !== undefined) {
-        bytesSent += data.byteLength;
-        onProgress(bytesSent);
-      } else if (!hasWarned) {
-        hasWarned = true;
-        console.warn("Expected stream to contain buffers, but it did not, so upload progress won't be reported.");
-      }
-    });
+    const inputStream = await this.sliceDataForRequest(source, part);
+    const progressStream = new ProgressStream({ onProgress });
+    inputStream.pipe(progressStream);
 
     return await UploadManagerFetchUtils.doPutUploadPart(
       this.config,
       part,
-      this.coerceRequestBody(streamWithProgress),
+      this.coerceRequestBody(progressStream),
       contentLength,
       addCancellationHandler
     );