RxJS的另外四种实现方式（五）——使用生成器实现

接上一篇RxJS的另外四种实现方式（四）——性能最高的库（续）

js的生成器一般情况下使用场景很少，开发者接触的不是很多。不了解的可以先行查看js语法了解。

这里把其中的执行顺序图解一下

调用方                                                                 数据源
next(value)--------------------------------------->              开始执行生成器函数体
            <-------------------------------------------------yield value2
next(value3)--------------------------------------->
            <-------------------------------------------------yield value4
next(value5)--------------------------------------->
            <-------------------------------------------------return value6

以上是正常返回最后值的过程，也可以永远不return，变成一个无限生成数据的过程。另一种情况是提前终止

调用方                                                                 数据源
next(value)--------------------------------------->              开始执行生成器函数体
            <-------------------------------------------------yield value2
next(value3)--------------------------------------->
            <-------------------------------------------------yield value4
return()--------------------------------------->

这种情况下相当于主动关闭生成器。可以向数据源的函数发出错误：

调用方                                                                 数据源
next(value)--------------------------------------->              开始执行生成器函数体
            <-------------------------------------------------yield value2
next(value3)--------------------------------------->
            <-------------------------------------------------try catch
throw(err)

以上各种行为都可以对应Rx，那么生成器和Rx的最大区别是什么呢？

就是谁是主动方，谁是被动方。在生成器中，调用方是主动方，相当于主动pull数据，而Rx中，数据源是主动方，相当于主动push数据。（这里和Rx中的推拉模式有区别）

那么如何使用生成器实现Rx呢？其实你估计已经想到了，就是反过来即可：

Observable                                                    Observer
next(value)--------------------------------------->              开始执行生成器函数体
            <-------------------------------------------------yield value2(得到返回值是value3)
next(value3)--------------------------------------->
            <-------------------------------------------------yield value4(返回值是value5)
next(value5)--------------------------------------->
            <-------------------------------------------------return value6()
done==true

于是我们就得到了由Observable主动推送过来的数据了。

我们还是以interval举例

exports.interval = period => sink => {
    if (sink.next().done) return noop
    let i = 0;
    const id = setInterval(() => sink.next(i++).done && clearInterval(id), period)
    return () => clearInterval(id)
}

这里传入的sink就是迭代器实例，我们主动调用next发送数据这里我们判断了next函数的返回值里面的done属性，如果Observer主动取消订阅了（在生成器函数里面执行了return语句）那么done就为true

下面是filter操作符：

function* _filter(sink, f) {
    for (let done = sink.next().done; !done;) {
        let x = yield 0
        if (x === _done) break
        if (f(x)) done = sink.next(x).done
    }
    sink.next(_done)
    sink.return()
}
exports.filter = f => source => sink => source(_filter(sink, f))

_done是一个Symbol，用来表示Observable的complete事件 _filter是一个生成器，调用它时传入下一级的迭代器（Observer） yeild 0 不断获取上一级的Observable的数据，一旦收到_done，立即跳出循环，并将_done传入sink中。

最后是实现Subscriber

function* subscribe(n, e, c) {
    while (true) {
        try {
            let result = yield 0
            while (result !== _done) {
                if (n(result) === _done) return
                result = yield 0
            }
            c && c()
        } catch (err) {
            e && e(err)
        }
    }
}
exports.subscribe = subscribe

是一个死循环，直到收到_done，或者抛出异常。至此，我们的Rx的基本功能已经实现，由于生成器的性能较差，所以本人没有花很多时间去完善各种操作符，只作为一种可以实现的方式展示出来。

下一篇我们介绍最后一种实现方法。