逐行读取文件的“反应式”方式是什么

Question

我正在使用 RxJS 学习反应式编程，并遇到需要逐行读取文件的情况。实际上我使用类似的解决方案解决了它：

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

它可以工作，但我需要使用一些普通的 JS 代码将缓冲区流转换为行流。（在上面的示例中使用“readline”模块）

我想知道是否还有其他方法可以使用 RxJS 运算符将 Buffer 的 Observable 转换为 Line 的 Observable ，如下例所示。

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