什麼是「被動」的方式來讀取文件中的行由行

Question

我使用RxJS學習反應式編程和遇到一個情況，當我需要閱讀文件中的行由行。其實我解決了它使用的解決方案喜歡：

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

它的工作原理，但我需要使用一些正常的JS代碼轉換緩衝器的流流線。 （上面的例子中使用「readline的」模塊）

不知是否有其他的方法來改造可觀察到的緩衝液的可觀察到的線，採用RxJS運營商，喜歡下面的例子。

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什麼東西。

喜歡的東西：

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"), 
);