契约：十几行手写一个 Observable

RxJS 的地基不是一大堆 operator，而是一个标准契约。它小到可以脱离库、用十几行手写出来：Observable 持有一段 producer 逻辑，只有当你 subscribe 时才执行；producer 拿到一个 Observer（一个带 next / error / complete 的对象），通过它把值推给你；subscribe 返回一个 Subscription（一个能 unsubscribe 的句柄），用来随时取消。

这正是 TC39 Observable proposal 与如今 WHATWG DOM Observable(el.when('click'))的形状。下面这十行，就是整个 RxJS 的种子：

class Observable {
  constructor(onSubscribe) {
    this.onSubscribe = onSubscribe;
  }
  subscribe(observer) {
    const teardown = this.onSubscribe(observer);
    return { unsubscribe: () => teardown && teardown() };
  }
}

1 · 推值：producer 通过 observer 把值送出去

点「下一步」逐行执行下面这个 producer。注意：整段 producer 在 subscribe 的那一刻才开始跑——这是 Observable 与「立即执行」的普通函数最不同的地方。每调用一次 observer.next(v)，右侧 observer 就收到一个值、时间轴上多一颗 marble；最后 complete() 用竖线封口。

2 · 语法契约：next* (error | complete)?

一条流的事件序列受一条简单语法约束：可以推任意多个 next，之后至多用一个 error 或一个 complete 收尾——二者互斥，且一旦收尾就不再有任何事件。这条契约让下游可以放心地在 complete / error 里做清理。

1  2  3  |      # 合法:三个 next 后 complete
1  2  ✕         # 合法:两个 next 后 error 收尾
|               # 合法:一个值都不推,直接 complete(空流)
1  |  2         # 非法:complete 之后不能再有 next
|  ✕            # 非法:complete 与 error 不能同时出现

3 · 可取消：unsubscribe 触发 teardown

producer 可以返回一个 teardown 函数；subscribe 把它包进 Subscription。调用 unsubscribe() 就会运行 teardown——停掉定时器、断开事件监听、放弃在途请求。这是 Promise 做不到的：Promise 一旦发起就无法取消。下面这条 interval$ 每 600ms 推一个值，随时可以取消：