该实现方式与之前几种不同的,该实现方式仅针对Nodejs环境。在Nodejs环境中,提供了Stream类,包括Readable、Transform、Writeable等子类都是可扩展的。从字面上看,正好对应Rx中的生产者、传递者、消费者。
实现该库的起因是,一次在Nodejs中需要在koa框架里面提供event-stream功能,目前除了IE浏览器外其他浏览器都支持了服务端事件推送,这个功能可以很好的代替轮询。webpack用的热更新就是通过这个功能实现的。
言归正传,首先得实现生产者,我们先来看interval
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也十分简单易读
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
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
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串起来,变成一个长长的水管
exports.pipe = pipeline || ((first, ...cbs) => cbs.reduce((aac, c) => aac.pipe(c), first));
高版本的Nodejs已经提供了pipeline方法,可以直接使用,低版本的话,可以用上面的方法进行连接。
至此,我们已经使用Nodejs提供的Stream类实现了Rx的基本逻辑。(完)