fix(csv-parse): destroy on end and call close event (fix #333)

packages/csv-parse/dist/cjs/index.cjs

@@ -1322,6 +1322,7 @@ class Parser extends stream.Transform {
       this.push(record);
     }, () => {
       this.push(null);
+      this.on('end', this.destroy);
     });
     if(err !== undefined){
       this.state.stop = true;
@@ -1337,6 +1338,7 @@ class Parser extends stream.Transform {
       this.push(record);
     }, () => {
       this.push(null);
+      this.on('end', this.destroy);
     });
     callback(err);
   }

packages/csv-parse/dist/esm/index.js

@@ -6380,6 +6380,7 @@ class Parser extends Transform {
       this.push(record);
     }, () => {
       this.push(null);
+      this.on('end', this.destroy);
     });
     if(err !== undefined){
       this.state.stop = true;
@@ -6395,6 +6396,7 @@ class Parser extends Transform {
       this.push(record);
     }, () => {
       this.push(null);
+      this.on('end', this.destroy);
     });
     callback(err);
   }

packages/csv-parse/dist/iife/index.js

@@ -6383,6 +6383,7 @@ var csv_parse = (function (exports) {
                   this.push(record);
                 }, () => {
                   this.push(null);
+                  this.on('end', this.destroy);
                 });
                 if(err !== undefined){
                   this.state.stop = true;
@@ -6398,6 +6399,7 @@ var csv_parse = (function (exports) {
                   this.push(record);
                 }, () => {
                   this.push(null);
+                  this.on('end', this.destroy);
                 });
                 callback(err);
               }

packages/csv-parse/dist/umd/index.js

@@ -6386,6 +6386,7 @@
                   this.push(record);
                 }, () => {
                   this.push(null);
+                  this.on('end', this.destroy);
                 });
                 if(err !== undefined){
                   this.state.stop = true;
@@ -6401,6 +6402,7 @@
                   this.push(record);
                 }, () => {
                   this.push(null);
+                  this.on('end', this.destroy);
                 });
                 callback(err);
               }

packages/csv-parse/lib/index.js

@@ -32,6 +32,7 @@ class Parser extends Transform {
       this.push(record);
     }, () => {
       this.push(null);
+      this.on('end', this.destroy);
     });
     if(err !== undefined){
       this.state.stop = true;
@@ -47,6 +48,7 @@ class Parser extends Transform {
       this.push(record);
     }, () => {
       this.push(null);
+      this.on('end', this.destroy);
     });
     callback(err);
   }

packages/csv-parse/test/api.stream.events.coffee

@@ -79,3 +79,29 @@ describe 'API events', ->
     parser.on 'error', (err) ->
       err.message.should.eql 'Invalid Record Length: expect 3, got 2 on line 2'
       next()
+
+  it 'emit `destroy` event', (next) ->
+    parser = parse """
+    a,a,a
+    b,b,b
+    c,c,c
+    """
+    parser.on 'readable', (data) ->
+      while this.read() isnt null then true
+    parser.on 'close', next
+    parser.on 'error', ->
+      next Error 'Event `error` should not be fired'
+
+  it 'emit `destroy` event with `to_line` option', (next) ->
+    # See https://github.com/adaltas/node-csv/issues/333
+    parser = parse """
+    a,a,a
+    b,b,b
+    c,c,c
+    """, to_line: 2
+    parser.on 'readable', (data) ->
+      while this.read() isnt null then true
+    parser.on 'close', next
+    parser.on 'error', ->
+      next Error 'Event `error` should not be fired'
+

packages/csv-parse/test/api.stream.finished.coffee

@@ -0,0 +1,41 @@
+
+import * as stream from 'node:stream/promises'
+import { Readable } from 'stream'
+import { generate } from 'csv-generate'
+import { parse } from '../lib/index.js'
+
+describe 'API stream.finished', ->
+
+  it 'resolved at the end', ->
+    # See https://github.com/adaltas/node-csv/issues/333
+    records = []
+    parser = generate(length: 10).pipe parse()
+    parser.on 'readable', () =>
+      while (record = parser.read()) isnt null
+        records.push record
+    await stream.finished parser
+    records.length.should.eql 10
+
+  it 'resolved with `to_line`', ->
+    # See https://github.com/adaltas/node-csv/issues/333
+    records = []
+    parser = generate(length: 10).pipe parse to_line: 3
+    parser.on 'readable', () =>
+      while (record = parser.read()) isnt null
+        records.push record
+    await stream.finished parser
+    records.length.should.eql 3
+
+  it 'rejected on error', ->
+    parser = parse to_line: 3
+    parser.write 'a,b,c\n'
+    parser.write 'd,e,f\n'
+    parser.write 'h,i,j,ohno\n'
+    parser.write 'k,l,m\n'
+    parser.end()
+    parser.on 'readable', () =>
+      while (record = parser.read()) isnt null then true
+    stream
+    .finished parser
+    .should.be.rejectedWith
+      code: 'CSV_RECORD_INCONSISTENT_FIELDS_LENGTH'