feat: update write methods to include totalLength parameter for improved chunk processing

Author: ido-pluto

src/download/download-engine/download-file/download-engine-file.ts

@@ -359,12 +359,12 @@ export default class DownloadEngineFile extends EventEmitter<DownloadEngineFileE
         const allWrites = new Set<Promise<any>>();
 
         let lastChunkSize = 0, lastInProgressIndex = startChunk;
-        await fetchState.fetchChunks((chunks, writePosition, index) => {
+        await fetchState.fetchChunks((chunks, writePosition, index, totalLength) => {
             if (this._closed || this._progress.chunks[index] != ChunkStatus.IN_PROGRESS) {
                 return;
             }
 
-            const writePromise = this.options.writeStream.write(downloadedPartsSize + writePosition, chunks);
+            const writePromise = this.options.writeStream.write(downloadedPartsSize + writePosition, chunks, totalLength);
             if (writePromise) {
                 allWrites.add(writePromise);
                 writePromise.then(() => {

src/download/download-engine/streams/download-engine-fetch-stream/base-download-engine-fetch-stream.ts

@@ -96,7 +96,7 @@ export type BaseDownloadEngineFetchStreamEvents = {
     streamNotRespondingOff: () => void;
 };
 
-export type WriteCallback = (data: Uint8Array[], position: number, index: number) => void;
+export type WriteCallback = (data: Uint8Array[], position: number, index: number, totalLength: number) => void;
 
 const DEFAULT_OPTIONS: BaseDownloadEngineFetchStreamOptions = {
     retryOnServerError: true,

src/download/download-engine/streams/download-engine-fetch-stream/download-engine-fetch-stream-xhr.ts

@@ -192,7 +192,7 @@ export default class DownloadEngineFetchStreamXhr extends BaseDownloadEngineFetc
             if (this.aborted) return;
 
             const chunk = await this.fetchBytes(this.state.activePart.downloadURL, this._startSize, this._endSize, this.state.onProgress);
-            callback([chunk], this._startSize, this.state.startChunk++);
+            callback([chunk], this._startSize, this.state.startChunk++, chunk.length);
         }
     }
 
@@ -213,7 +213,7 @@ export default class DownloadEngineFetchStreamXhr extends BaseDownloadEngineFetc
 
             const chunk = relevantContent.slice(start, end);
             totalReceivedLength += chunk.byteLength;
-            callback([chunk], index * this.state.chunkSize, index++);
+            callback([chunk], index * this.state.chunkSize, index++, chunk.length);
         }
     }
 

src/download/download-engine/streams/download-engine-fetch-stream/utils/smart-chunk-split.ts

@@ -43,7 +43,7 @@ export default class SmartChunkSplit {
 
     closeAndSendLeftoversIfLengthIsUnknown() {
         if (this._chunks.length > 0 && this._options.endChunk === Infinity) {
-            this._callback(this._chunks, this._bytesWriteLocation, this._options.startChunk++);
+            this._callback(this._chunks, this._bytesWriteLocation, this._options.startChunk++, this._savedLength);
             this._chunks = [];
             this._savedLength = 0;
         }
@@ -78,7 +78,7 @@ export default class SmartChunkSplit {
                     }
 
                     this._savedLength -= calcChunkThreshold;
-                    this._callback(sendChunks, this._bytesWriteLocation, this._options.startChunk++);
+                    this._callback(sendChunks, this._bytesWriteLocation, this._options.startChunk++, calcChunkThreshold);
                     this._bytesWriteLocation += calcChunkThreshold;
                     break;
                 }

src/download/download-engine/streams/download-engine-write-stream/base-download-engine-write-stream.ts

@@ -1,5 +1,5 @@
 export default abstract class BaseDownloadEngineWriteStream {
-    abstract write(cursor: number, buffers: Uint8Array[]): Promise<void> | void;
+    abstract write(cursor: number, buffers: Uint8Array[], totalLength: number): Promise<void> | void;
 
     close(): void | Promise<void> {
     }

src/download/download-engine/streams/download-engine-write-stream/download-engine-write-stream-browser.ts

@@ -9,7 +9,7 @@ export type DownloadEngineWriteStreamOptionsBrowser = {
     file?: DownloadFile
 };
 
-export type DownloadEngineWriteStreamBrowserWriter = (cursor: number, buffers: Uint8Array[], options: DownloadEngineWriteStreamOptionsBrowser) => Promise<void> | void;
+export type DownloadEngineWriteStreamBrowserWriter = (cursor: number, buffers: Uint8Array[], options: DownloadEngineWriteStreamOptionsBrowser, totalLength: number) => Promise<void> | void;
 
 export default class DownloadEngineWriteStreamBrowser extends BaseDownloadEngineWriteStream {
     protected readonly _writer?: DownloadEngineWriteStreamBrowserWriter;
@@ -44,13 +44,12 @@ export default class DownloadEngineWriteStreamBrowser extends BaseDownloadEngine
         return this._memory = newMemory;
     }
 
-    public write(cursor: number, buffers: Uint8Array[]) {
+    public write(cursor: number, buffers: Uint8Array[], totalLength: number) {
         if (this.writerClosed) {
             throw new WriterIsClosedError();
         }
 
         if (!this._writer) {
-            const totalLength = buffers.reduce((sum, buffer) => sum + buffer.length, 0);
             const bigBuffer = this._ensureBuffer(cursor + totalLength);
             let writeLocation = cursor;
 
@@ -63,7 +62,7 @@ export default class DownloadEngineWriteStreamBrowser extends BaseDownloadEngine
             return;
         }
 
-        return this._writer(cursor, buffers, this.options);
+        return this._writer(cursor, buffers, this.options, totalLength);
     }
 
     public get result() {

src/download/download-engine/streams/download-engine-write-stream/download-engine-write-stream-nodejs.ts

@@ -101,8 +101,8 @@ export default class DownloadEngineWriteStreamNodejs extends BaseDownloadEngineW
      * Fragments are concatenated into a single Buffer, contiguous regions merged,
      * and auto-flushed when the adaptive threshold is exceeded.
      */
-    write(cursor: number, buffers: Uint8Array[]) {
-        return this._writeQueue.addWrite(cursor, buffers);
+    write(cursor: number, buffers: Uint8Array[], totalLength: number) {
+        return this._writeQueue.addWrite(cursor, buffers, totalLength);
     }
 
     ensureBytesSynced() {

src/download/download-engine/streams/download-engine-write-stream/utils/WriteQueue.ts

@@ -57,12 +57,12 @@ export default class WriteQueue {
      * Buffer a write. Concatenates fragments into a single Buffer,
      * merges with adjacent regions, and flushes when threshold is exceeded.
      */
-    addWrite(cursor: number, buffers: Uint8Array[]): void | Promise<void> {
+    addWrite(cursor: number, buffers: Uint8Array[], totalLength: number): void | Promise<void> {
         if (this._closed) {
             throw new WriterIsClosedError("Cannot add write to closed WriteQueue");
         }
 
-        const length = buffers.reduce((sum, buf) => sum + buf.length, 0);
+        const length = totalLength;
 
         const merged = this._tryMerge(cursor, buffers, length);
         if (!merged) {