接上一篇
李宇翔：RxJS的另外四种实现方式（五）——使用生成器实现
该实现方式与之前几种不同的，该实现方式仅针对Nodejs环境。在Nodejs环境中，提供了Stream类，包括Readable、Transform、Writeable等子类都是可扩展的。从字面上看，正好对应Rx中的生产者、传递者、消费者。
实现该库的起因是，一次在Nodejs中需要在koa框架里面提供event-stream功能，目前除了IE浏览器外其他浏览器都支持了服务端事件推送，这个功能可以很好的代替轮询。webpack用的热更新就是通过这个功能实现的。
言归正传，首先得实现生产者，我们先来看interval
代码语言：javascript复制class Interval extends Readable {
constructor(period) {
super({ objectMode: true })
this.period = period
this.i = 0
}
_read(size) {
setTimeout(() => this.push(this.i ), this.period)
}
}
exports.interval = period => new Interval(period)说明一下，构造函数传入objectMode:true的对象是让stream处于对象模式，而不是二进制流模式。_read函数必须覆盖父类，否则出错，当有订阅者连接上来后，就会调用_read方法。我们在这个方法里面发送数据，即调用push方法，将数据发送给流的接收者。
当调用过push方法后，后面的接收者如果调用了callback回调，则表示数据消费完毕，会再次调用_read方法，直到push(null)表示生产者已经complete
FromArray也十分简单易读
代码语言：javascript复制class FromArray extends Readable {
constructor(array) {
super({ objectMode: true })
this.array = array
this.pos = 0
this.size = array.length
}
_read(size) {
if (this.pos < this.size) {
this.push(this.array[this.pos ])
} else
this.push(null)
}
}
exports.fromArray = array => new FromArray(array)下面要实现一个转换器（操作符）Filter
代码语言：javascript复制class Filter extends Transform {
constructor(f) {
super({ readableObjectMode: true, writableObjectMode: true })
this.f = f
}
_transform(data, encoding, callback) {
const f = this.f
if (f(data)) {
this.push(data);
}
callback();
}
_flush(callback) {
callback()
}
}
exports.filter = f => new Filter(f)这时候我们需要覆盖_transform、_flush函数，同样的，push方法会让数据流到下面的流中，而callback回调会使得上一个流继续发送数据。
最后我们来实现Subscriber
代码语言：javascript复制class Subscriber extends Writable {
constructor(n, e, c) {
super({ objectMode: true })
this.n = n
this.e = e
this.c = c
}
_write(chunk, encoding, callback) {
this.n(chunk)
callback(null)
}
_final(callback) {
this.c()
callback()
}
}
exports.subscribe = (n, e = noop, c = noop) => new Subscriber(n, e, c)Subscriber是一个可写流，我们必须覆盖_write方法用于消费数据，_final方法用于complete事件处理。这里没有实现error事件。有兴趣的同学可以思考如何实现。
最后我们需要把各种stream串起来，变成一个长长的水管
代码语言：javascript复制exports.pipe = pipeline || ((first, ...cbs) => cbs.reduce((aac, c) => aac.pipe(c), first));高版本的Nodejs已经提供了pipeline方法，可以直接使用，低版本的话，可以用上面的方法进行连接。
至此，我们已经使用Nodejs提供的Stream类实现了Rx的基本逻辑。（完）
