Catch errors in 'data' event; Add readable and writable properties;

lib/csv.js

@@ -55,6 +55,12 @@ module.exports = function(){
             bufferSize: null,
             end: true // Call `end()` on close
         };
+        // A boolean that is true by default, but turns false after an 'error' occurred, 
+        // the stream came to an 'end', or destroy() was called. 
+        this.readable = true;
+        // A boolean that is true by default, but turns false after an 'error' occurred 
+        // or end() / destroy() was called. 
+        this.writable = true;
     }
     CSV.prototype.__proto__ = EventEmitter.prototype;
 
@@ -92,6 +98,7 @@ module.exports = function(){
                 parse(data);
             }catch(e){
                 self.emit('error', e);
+                // Destroy the input stream
                 this.destroy();
             }
         });
@@ -150,9 +157,11 @@ module.exports = function(){
                     csv.writeStream.end();
                 }else{
                     csv.emit('end', state.count);
+                    csv.readable = false;
                 }
             }else{
                 csv.emit('end', state.count);
+                csv.readable = false;
             }
         }
     }
@@ -178,7 +187,9 @@ module.exports = function(){
                 break;
         }
         writeStream.on('close', function(){
-            self.emit('end',state.count);
+            self.emit('end', state.count);
+            self.readable = false;
+            self.writable = false;
         })
         this.writeStream = writeStream;
         state.buffer = new Buffer(this.writeOptions.bufferSize||this.readOptions.bufferSize);
@@ -219,7 +230,13 @@ module.exports = function(){
             return;
         }
         if(!preserve){
-            csv.emit('data', line, state.count);
+            try {
+                csv.emit('data', line, state.count);
+            }catch(e){
+                csv.emit('error', e);
+                csv.readable = false;
+                csv.writable = false;
+            }
         }
         if(typeof line === 'object'){
             if(!(line instanceof Array)){
@@ -246,7 +263,7 @@ module.exports = function(){
                         // Cast number to string
                         field = '' + field;
                     }else if(typeof field === 'boolean'){
-                        // Cast number to string
+                        // Cast boolean to string
                         field = field ? '1' : '';
                     }else if(field instanceof Date){
                         // Cast date to timestamp string
@@ -285,6 +302,7 @@ module.exports = function(){
         if(!preserve){
             state.countWriten++;
         }
+        return true;
     }
 
     /**
@@ -383,6 +401,7 @@ module.exports = function(){
         }
     }
 
+    // Called by the `parse` function on each line. It will then call `write`
     function flush(){
         if(csv.readOptions.columns){
             if(state.count === 0 && csv.readOptions.columns === true){

test/reader.coffee

@@ -0,0 +1,19 @@
+
+fs = require 'fs'
+should = require 'should'
+csv = require '..'
+
+describe 'reader', ->
+    it 'Test exception in on(data)"', (next) ->
+        count = 0
+        errors = 0
+        test = csv()
+        .toPath( "#{__dirname}/write/write_array.tmp" )
+        .on 'data', (data, index) ->
+            throw new Error 'Error in data' if index % 10 is 0
+        .on 'error', ->
+            next()
+        .on 'end', ->
+            false.should.be.ok
+        for i in [0...1000]
+            test.write ['Test '+i, i, '"'] if test.writable