Supporting startRow option

Author: Igor Drobiazko

lib/actions/write.js

@@ -45,18 +45,25 @@ function init(cfg) {
 function processAction(msg, cfg) {
     const self = this;
 
+    const startRow = cfg.reader.startRow || 0;
+
+    if (startRow !== 0 && rowCount === 0) {
+        console.log('Skipping the first %s rows', startRow + 1);
+    }
+
     if (timeout) {
         clearTimeout(timeout);
     }
 
     timeout = setTimeout(()=> {
         console.log('Closing the stream due to inactivity');
         co(function* gen() {
-            console.log('The resulting CSV file contains %s rows', rowCount);
+            const finalRowCount = rowCount - startRow;
+            console.log('The resulting CSV file contains %s rows', finalRowCount);
             stringifier.end();
 
             const messageToEmit = messages.newMessageWithBody({
-                rowCount
+                rowCount: finalRowCount
             });
             const fileName = messageToEmit.id + '.csv';
             messageToEmit.attachments[fileName] = {
@@ -73,12 +80,14 @@ function processAction(msg, cfg) {
         });
     }, 10000);
 
-    let row = msg.body;
-    if( cfg.reader.columns) {
-        const columns = Object.keys(_.keyBy(cfg.reader.columns, 'property'));
-        row = _.pick(msg.body, columns);
+    if (startRow <= rowCount) {
+        let row = msg.body;
+        if( cfg.reader.columns) {
+            const columns = Object.keys(_.keyBy(cfg.reader.columns, 'property'));
+            row = _.pick(msg.body, columns);
+        }
+        stringifier.write(row);
     }
-    stringifier.write(row);
     rowCount++;
     this.emit('end');
 }