ノードストリームを反復可能として扱う方法はありますか？

Question

私はstdinに渡された一連のテキストを処理するツールを作成していますが、各行は「エントリ」です。コードをより機能的にしたいので、一連の行を "シーケンス"または "iterable"として扱い、reduceを使って繰り返します。

私は現在の線の集合としてstdinを処理するNodeモジュールLineStreamを使用していますが、それは行ごとにdataイベントを送出することによって動作します - それはReadable Streamインタフェースを実装しています、大丈夫です。

だから私は現在、私の関数にするたびに、暫定値を渡すことによってdataイベントが発生減らす非常に「取扱説明書」のようなものやってる：

を

var windows = []; 

linestream.on('data', function(line) { 
    return windows = rollup(windows, extractDate(line), argv.w); 
}); 

linestream.on('end', function() { 
    return process.stdout.write(toCsv(windows)); 
}); 

process.stdin.resume(); 

をしかし、それを行うのは、より機能的になるだろう何かのように：私は上の私のツールを実行しているとき、もちろん

linestream.lines.reduce(rollup, []); 

function rollup(windows, line) { 
    // would return a new interim or final value 
} 

、私は通常の配列にすべての行を「収集」し、それを減らすことが、私はそれを試してみましたが、それはあまりにも多くのメモリを使用して可能性があり大規模なデータセット - それでは、繰り返しのようなものeストリームは本当に必要なものです。

私は、これを行うNode関数/モジュールを書くことができるかどうか、または既に存在するかどうかを聞いています。

ありがとうございます！

Answer 1

非同期性を扱っているため、これをより機能的にする方法はないと思います。存在するlinestream.linesについては

、私は2つのうちの1つが起こる必要があると思います：

• インメモリすでに述べてきたすべての行のバッファが新しい
• あまりにも多くのメモリを占有非同期の制御フローを同期的なもののように見せることができるようにすることである。

私は（jqueryのか、他のいくつかの約束のAPIを使用して仮定して）あなたがこれを行うことができたとします

var op = (function(){ 

    var windows = [] 
     ,done = $.deferred(); 

    linestream.on('data', function(line) { 
     return windows = rollup(windows, extractDate(line), argv.w); 
    }); 

    linestream.on('end', function() { 
     process.stdout.write(toCsv(windows)); 
     return done.resolve(windows); 
    }); 

    process.stdin.resume(); 

    return done.promise(); 

})(); 

しかし、本当にそれだけの事を隠しています。

Rxのようなものを使用するか、generatorsが周りに来るまで待つことができます。

Answer 2

あなたはすでにそれを機能的な方法でやっています。あなたはイベントを聴いていて、そのイベントがトリガされたときに関数を実行すると、それ以上の機能はできません。

これは関数を変更しない例ですが、それは最初の例と同じように機能します。しかし、それが変わるのは、関数が実行されるソースです。 Reduceは大量のデータに依存していますが、すべて同じ時間にメモリに格納されます。これは、言うまでもなく、非常に大きなメモリ使用量につながります。

私があなただったらデフォルトノードのままにしておきます。

Answer 3

rollupは何を想定しているのかよく分かりませんが、他の人も指摘しているように、一度にすべてのデータを一度に持っていなくてもすぐにすべてのデータを保持することを期待するreduce機能メモリ。

ただし、できることは、データイベントコールバックの削減ロジックだけです。最後の値や合計値などの状態がさらに必要な場合は、そのデータをコールバックのクロージャに保持できます。

たとえば、数値の非同期ストリームでのローリング平均があります。この例では

var total = 0; 
var items = 0; 
var average; 

stream.on('data', function (line) { 
    var num = parseInt(line, 10); 
    total += line; 
    items++; 
    average = total/items; 
}); 

stream.on('end', function() { 
    console.log("The average is %s", average); 
}); 

、私はそれが入って来て、各ラインのうち、関連するデータを取得し、常に私の状況を知るために十分な余分なデータを周りに保つことです。この場合、私は平均を計算しているので、そこにいくつの合計項目があるかを知る必要があります。