接上篇RxJS的另外四种实现方式（序）

起因

想到这个库的原因，是看了callbag库想到的，callbag库的原理大家可以自己找资料了解，我就不多赘述，我只谈谈我的理解。callbag的设计思路是把消费者和生产者合并成一个，通过互相传递一个回调函数实现通讯。看过部分操作符实现原理的同学肯定觉得逻辑十分难解，因为过多的回调使得你的脑回路不够用了。我用了一些库函数后，我意识到，其实不需要如此复杂的设计，为什么呢？请看下文

大同小异的callbag

callbag里面有很多代码是重复书写的，原因很简单，功能是确定的，如订阅功能，这是必不可少的操作，下面我来比较一下我的库的实现和callbag的实现。

对比实现生产者interval

先上callbag的源码

const interval = period => (start, sink) => {
  if (start !== 0) return;
  let i = 0;
  const id = setInterval(() => {
    sink(1, i++);
  }, period);
  sink(0, t => {
    if (t === 2) clearInterval(id);
  });
};

export default interval;

说明一下

if(start!=0)return

这句话在callbag实现库里面随处可见，我就是因为这句话引起的思考，为什么每次都要重复写呢？
当然是因为这是一个生产者，只发送数据，不会去接受数据。

  sink(0, t => {
    if (t === 2) clearInterval(id);
  });

上面这段代码其实是实现了一个取消订阅功能，实现方法是向传来的回调函数再传回一个回调函数，估计读者脑子要烧糊了。

上面这个interval可观察对象的原型可以代表大多数的callbag的案例，那么有没有办法用更为简洁的方式实现呢？

ShowTime

exports.interval = period => n => {
    let i = 0;
    const id = setInterval(() => n(i++), period)
    return () => clearInterval(id)
}

什么，只有这么几行代码吗？，没错，这就是我认为实现代码最小的库了，不服来战。此代码不仅小，性能好，还通俗易懂。当然我还是得稍微解释一下要使得interval(1000)成为一个地道的生产者，必须要实现可以订阅，可以取消订阅，以及可以得到生产者发出的数据（有些还需要得到complete和error事件，interval不会complete也不会error）

• interval(1000)将得到一个函数n=>……，这个函数接受一个next函数用于发送数据
• 调用interval(1000)这个高阶函数等同于“订阅”，此处是重点（代替了callbag中发送type为0的行为）
• 返回的是一个dispose函数，即用于“取消订阅”的功能（代替了callbag中传回一个回调并在里面接受type为2的行为）
• 函数中调用了传入的next函数n，即发送出去了数据

当然interval不会独立工作，我们需要更多的操作符和观察者使得库来运作。

对比操作符filter

下面是callbag的实现

const filter = condition => source => (start, sink) => {
  if (start !== 0) return;
  let talkback;
  source(0, (t, d) => {
    if (t === 0) {
      talkback = d;
      sink(t, d);
    } else if (t === 1) {
      if (condition(d)) sink(t, d);
      else talkback(1);
    }
    else sink(t, d);
  });
};

module.exports = filter;

依然出现了

if(start!=0)return

没错，因为filter只用于被订阅，本身作为数据响应者，有人说不对，filter需要对上一级的源做响应，没错，所以需要订阅上一级的源，但传入的不是自身，而是另一个回调函数来响应，否则就会有问题。核心代码就一句，却需要一大堆代码来维持正常运行，我看不下去了。

ShowTime

exports.filter = f => source => (n, c) => source(d => f(d) && n(d), c)

What?就一行代码？你没看错，你没看错，你没看错！
我来解释一下，这一行代码。filter是一个操作符，filter(d=>d>1)代表我只接受大于1的数据，这个将返回一个source=>……的函数，这个函数接受一个source作为上一级数据源，可以是上文的interval(1000)这样的生产者，也可以是其他操作符。所以

const obserable = filter(d => d > 1)(interval(1000))

你将得到一个(n,c)=>……的函数，这个就是可观察者，你可以传入next函数n，和complete函数c来进行“订阅”了

const disposable = obserable(d => console.log('得到',d),err => console.log('完成'))//err代表有错误，这里先不处理

你订阅过后会得到一个函数disposable，用于“取消订阅”

disposable()//取消订阅

这个filter代表了最小库的精髓：disposable可以从箭头函数一路返回，在filter中是隐含的，无需显示实现而代表complete的c函数也是直接透传，无需更改。唯独需要操作的就是next函数，需要向source传一个新的next函数。当满足条件时就向下一级的next函数发送数据，否则啥也不干。

（未完待续）