node.js - 如何在 rxjs 中使用 Node 的转换流？

Question

我使用 rxjs 已经有一段时间了，我喜欢如何使用它的运算符进行逻辑而不是命令式编程。但是，我也喜欢 Node 的流，它也是高度可组合的，所以我的明显 react 是同时使用它们，但除了在 r​​xjs 的 book 中绑定(bind)它之外，我还没有看到它被提及很多(实际上，我根本没有) .

所以，我的真正问题是，如何在 RxJS 上使用 npm 中的所有转换流？或者，有可能吗？
示例:-

var fs = require('fs');
var csv = require('csv-parse')({delimiter:';'});
var src = fs.createReadStream('./myFile.csv');
src.pipe(csv).pipe(process.stdout);

本质上，我想这样做:-

var fs = require('fs');
var csv = require('csv-parse')({delimiter:';'});
var rx= require('rx-node');
var src = fs.createReadStream('./myFile.csv');

var obj = rx.fromReadableStream(src);
obj.pipe(csb).map(x=>console.log(x));

过去我被告知要使用 highland 但我在这里严格寻找 rxjs 解决方案。

Answer 1

你不一定要使用 rx-node 但你可以!请记住:所有流都是事件发射器!。

准备:输入.txt

Hello World!
Hello World!
Hello World!
Hello World!
Hello World!

运行:

npm install through2 split2 rx rx-node

在 index.js 中:

var Rx = require('rx');
Rx.Node = require('rx-node');
var fs = require('fs');
var th2 = require('through2');
var split2 = require('split2');

var file = fs.createReadStream('./input.txt').on('error', console.log.bind(console, 'fs err'));

var transform = th2(function(ch, en, cb) {
  cb(null, ch.toString());
}).on('error', function(err) {
  console.log(err, err.toString());
});

// All streams are event emitters ! (one way without using rx-node)
// var subs = Rx.Observable.fromEvent(transform, 'data').share();
// subs
//   .map(value => 'Begin line: ' + value)
//   .subscribe(value => console.log(value));

// rx-node has convenience functions (another way)
var subs = Rx.Node.fromTransformStream(transform).share()
  .map(value => 'Begin line: ' + value)
  .subscribe(value => console.log(value));

file.pipe(split2()).pipe(transform);

输出:

Begin line: Hello World!
Begin line: Hello World!
Begin line: Hello World!
Begin line: Hello World!
Begin line: Hello World!