ノードストリーム：n行を変換して変換して続行する

Question

NodeJSでは、大きなファイルを読み込み、フラグメント（n行）で処理し、データを消費してから次のn行を処理したいと考えています。

私はいくつかのモジュール（fs、es-stream、main-onesのためのノード-ele）を使ってみたいと思っています。

新しい行を処理する前に私がやったベストは（コードは以下です）、変換の完了を待たずにいました。

ここに私のスニペット：

const fs = require('fs'); 
const es = require('event-stream'); 
const parse = require('csv-parse'); 
const stringify = require('csv-stringify'); 
const etl = require('etl'); 

exports.toDatabase = (file, done) => { 

    // File contains six lines wit htwo values (example : aa:bb for the first line) 
    let input = fs.createReadStream(todayTeamsFile); 

    input 
    .pipe(es.split()) 
    .pipe(etl.collect(2)) 
    .pipe(es.map((data, nextMap) => { 
     // I'd like to process all this code before continuing to read my stream 
     let date = Date.now(); 
     console.log('map data ' + date); 
     console.log(data); 

     parse(data[0], { 
     delimiter: ';' 
     }, (err, output) => { 
     console.log('Row done ' + date); 
     // Treatment to do would be to insert in database the output 
     console.log(output); 
     console.log('------ ' + date); 
     return nextMap(); 
     }); 

    })); 

}; 

しかし、次のマップが最初の呼び出し

 
TESTING !! 
map data 1481824486765 
[ 'aa;zz', 'bb;cc' ] 
map data 1481824486771 
[ 'dd;ee', 'ff;gg' ] 
Row done 1481824486765 
[ [ 'aa', 'zz' ] ] 
------ 1481824486765 
Row done 1481824486771 
[ [ 'dd', 'ee' ] ] 
------ 1481824486771 
map data 1481824486785 
[ 'hh;ii', '' ] 
Row done 1481824486785 
[ [ 'hh', 'ii' ] ] 
------ 1481824486785 

Answer 1

なぜわからないの完了前に起動されたことを出力ショーチャンクされたストリームをチャンクしたいのですが、ここでは、

var through = require('through2'); 
var split = require('split') 
var fs = require('fs') 
var handler = (function(len) { 
    var buff = []; 
    var p = 0; 
    return through.obj(function(chunk, enc, cb) { 
    buff.push(chunk); 
    p++; 
    if(buff.length===len) { 
     console.log('send--------------'); 
     this.push(buff); 
     buff = []; 
    } 
    if (p>25) { 
     this.emit('error', 'kill the stream') 
    } 
    cb(); 
    }, function (cb) { 
    this.push(buff); // may be much larger than 4, it may need a logic to re split. 
    buff = []; 
    cb(); 
    }); 
})(4); 

var origin = fs.createReadStream('tomate.csv'); 

origin 
.pipe(split()) 
.pipe(handler) 
.pipe(through.obj(function(chunk, enc, cb){ 
    console.log('process: %v', chunk); 
    cb() 
})); 

handler.on('error', function() { 
    origin.close(); 
    // still need o unpipe everything, use mississipi 
});