fix(csv-parse): premature close error

wdavidw · wdavidw · commit c6473a9cb6c5 · 2023-10-05T09:16:53.000+02:00
diff --git a/packages/csv-parse/dist/cjs/index.cjs b/packages/csv-parse/dist/cjs/index.cjs
@@ -1324,15 +1324,18 @@ class Parser extends stream.Transform {
     this.info = this.api.info;
   }
   // Implementation of `Transform._transform`
-  _transform(buf, encoding, callback){
+  _transform(buf, _, callback){
     if(this.state.stop === true){
       return;
     }
     const err = this.api.parse(buf, false, (record) => {
       this.push(record);
     }, () => {
       this.push(null);
-      this.on('end', this.destroy);
+      this.end();
+      this.destroy();
+      // Note 231005, end wasnt used and destroy was called as:
+      // this.on('end', this.destroy);
     });
     if(err !== undefined){
       this.state.stop = true;
diff --git a/packages/csv-parse/dist/esm/index.js b/packages/csv-parse/dist/esm/index.js
@@ -6446,15 +6446,18 @@ class Parser extends Transform {
     this.info = this.api.info;
   }
   // Implementation of `Transform._transform`
-  _transform(buf, encoding, callback){
+  _transform(buf, _, callback){
     if(this.state.stop === true){
       return;
     }
     const err = this.api.parse(buf, false, (record) => {
       this.push(record);
     }, () => {
       this.push(null);
-      this.on('end', this.destroy);
+      this.end();
+      this.destroy();
+      // Note 231005, end wasnt used and destroy was called as:
+      // this.on('end', this.destroy);
     });
     if(err !== undefined){
       this.state.stop = true;
diff --git a/packages/csv-parse/dist/iife/index.js b/packages/csv-parse/dist/iife/index.js
@@ -6449,15 +6449,18 @@ var csv_parse = (function (exports) {
                 this.info = this.api.info;
               }
               // Implementation of `Transform._transform`
-              _transform(buf, encoding, callback){
+              _transform(buf, _, callback){
                 if(this.state.stop === true){
                   return;
                 }
                 const err = this.api.parse(buf, false, (record) => {
                   this.push(record);
                 }, () => {
                   this.push(null);
-                  this.on('end', this.destroy);
+                  this.end();
+                  this.destroy();
+                  // Note 231005, end wasnt used and destroy was called as:
+                  // this.on('end', this.destroy);
                 });
                 if(err !== undefined){
                   this.state.stop = true;
diff --git a/packages/csv-parse/dist/umd/index.js b/packages/csv-parse/dist/umd/index.js
@@ -6452,15 +6452,18 @@
                 this.info = this.api.info;
               }
               // Implementation of `Transform._transform`
-              _transform(buf, encoding, callback){
+              _transform(buf, _, callback){
                 if(this.state.stop === true){
                   return;
                 }
                 const err = this.api.parse(buf, false, (record) => {
                   this.push(record);
                 }, () => {
                   this.push(null);
-                  this.on('end', this.destroy);
+                  this.end();
+                  this.destroy();
+                  // Note 231005, end wasnt used and destroy was called as:
+                  // this.on('end', this.destroy);
                 });
                 if(err !== undefined){
                   this.state.stop = true;
diff --git a/packages/csv-parse/lib/index.js b/packages/csv-parse/lib/index.js
@@ -24,15 +24,18 @@ class Parser extends Transform {
     this.info = this.api.info;
   }
   // Implementation of `Transform._transform`
-  _transform(buf, encoding, callback){
+  _transform(buf, _, callback){
     if(this.state.stop === true){
       return;
     }
     const err = this.api.parse(buf, false, (record) => {
       this.push(record);
     }, () => {
       this.push(null);
-      this.on('end', this.destroy);
+      this.end();
+      this.destroy();
+      // Note 231005, end wasnt used and destroy was called as:
+      // this.on('end', this.destroy);
     });
     if(err !== undefined){
       this.state.stop = true;
diff --git a/packages/csv-parse/test/api.stream.finished.coffee b/packages/csv-parse/test/api.stream.finished.coffee
@@ -1,6 +1,5 @@
 
 import * as stream from 'node:stream/promises'
-import { Readable } from 'stream'
 import { generate } from 'csv-generate'
 import { parse } from '../lib/index.js'
 
diff --git a/packages/csv-parse/test/option.to_line.coffee b/packages/csv-parse/test/option.to_line.coffee
@@ -1,5 +1,8 @@
 
+import { Readable } from 'stream'
+import { finished } from 'node:stream/promises'
 import { parse } from '../lib/index.js'
+import { generate } from 'csv-generate'
 
 describe 'Option `to_line`', ->
   
@@ -97,3 +100,16 @@ describe 'Option `to_line`', ->
         [ 'd','e','f' ]
       ] unless err
       next err
+
+  it 'resolved with `to_line`', ->
+    # Prevent `Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close`
+    reader = new Readable
+      highWaterMark: 100
+      read: (size) ->
+        setImmediate =>
+          for i in [0...size]
+            this.push "#{size},#{i}\n"
+    parser = reader.pipe parse to_line: 3
+    parser.on 'readable', () =>
+      while parser.read() isnt null then true
+    await finished parser
