ファイルを1行ずつ読み込むための「反応的な」方法

Question

RxJSを使用してリアクティブプログラミングを学習していて、ファイルを1行ずつ読み込む必要がある場合があります。

https://gist.github.com/yvele/447555b1c5060952a279

それは動作しますが、私はラインの流れにバッファのストリームを変換するために、いくつかの通常のJSコードを使用する必要があります：実際に私は、ソリューションを使用して、それを解決したのが好き。 （上記の例では "readline"モジュールを使用してください）

Observable of BufferをRxJS演算子を使用してObservable of Lineに変換する方法が他にもありますか？

var Rx = require('rx'); 
var fs = require('fs'); 
var lines = Rx.Observable 
    .fromEvent(rl, 'data') // emits buffers overtime 
    // some transforms ... 
    .subscribe(
    (line) => console.log(line), // emit string line by line 
    err => console.log("Error: %s", err), 
    () => console.log("Completed") 
); 

Answer 1

あなたはおそらく、あなたがscanとconcatMapで欲しいものにかなり近いものを達成することができます。あなたは、通常のjsのコードを考え何をすべきか

bufferSource 
    .concat(Rx.Observable.of("\n")) // parens was missing // to make sure we don't miss the last line! 
    .scan(({ buffer }, b) => { 
    const splitted = buffer.concat(b).split("\n"); 
    const rest = splitted.pop(); 
    return { buffer: rest, items: splitted }; 
    }, { buffer: "", items: [] }) 
    // Each item here is a pair { buffer: string, items: string[] } 
    // such that buffer contains the remaining input text that has no newline 
    // and items contains the lines that have been produced by the last buffer 
    .concatMap(({ items }) => items) 
    // we flatten this into a sequence of items (strings) 
    .subscribe(
    item => console.log(item), 
    err => console.log(err), 
    () => console.log("Done with this buffer source"), 
);