Fix bug where batch is cleared before transform function can use it

diff --git a/src/core.js b/src/core.js
index ef97463a3a2f92c9eec327ba30a184d219dc4202..4e4912523194db81040db3471224695fc87c5106 100644 (file)
--- a/src/core.js
+++ b/src/core.js
@@ -377,7 +377,6 @@ export class CSVTransformer extends TransformStream {
 
   async #enqueueConcurrent(controller) {
     const results = await Promise.allSettled(this.#concurrent);
-    this.#concurrent.length = 0;
     for (const r of results) {
       if (r.status === 'rejected') {
         if (r.reason instanceof Error) {
@@ -391,6 +390,10 @@ export class CSVTransformer extends TransformStream {
         this.#enqueueRow(r.value, controller);
       }
     }
+    // Only clear these arrays AFTER all transform processing is complete. Otherwise downstream functions
+    // may not be able to access the row data.
+    this.#concurrent.length = 0;
+    this.#batch.length = 0;
   }
 
   /** @type {TransformerTransformCallback<Array<string>,Array<string>>} */
@@ -413,7 +416,6 @@ export class CSVTransformer extends TransformStream {
       this.#batch.push(chunk);
       if (this.#batch.length === this.#maxBatchSize) {
         this.#concurrent.push(this.#wrappedFn(this.#batch));
-        this.#batch.length = 0;
       }
     }
     else {

diff --git a/test/core.test.js b/test/core.test.js
index fb7a4af37d26683265155041b02d7a1be8534e09..cead01f036091de2c15942946cb5b7f5b98ef2e6 100644 (file)
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -482,6 +482,19 @@ suite('CSVTransformer', { concurrency: true }, () => {
       { name: 'Error', message: 'Not an error type' }
     );
   });
+  test('can batch row data', { concurrency: true }, async (t) => {
+    await createCSVMockStream([
+      ['columnA'],
+      ['1'],
+      ['2']
+    ])
+      .pipeThrough(new CSVTransformer(r => r, { maxBatchSize: 2 }))
+      .pipeTo(csvStreamEqualWritable([
+        ['columnA'],
+        ['1'],
+        ['2']
+      ]));
+  });
   test('creates batch of 1 when maxBatchSize is equal to 1', { concurrency: true }, async (t) => {
     const fn = t.mock.fn(b => b.map(r => r.map(f => Number(f) * 2)));
     await createCSVMockStream([