rxjs中的

如果没有了解源码的确很难回答我这个问题。
我也只是rxjs入门水平，勉强实现 Observable (包括pipe与退订机制)、 Subject、map、filter、timer、asyncScheduler 调度 (保持与官方API使用与运行一致)

我简单描述一下，

pipe 内部的逻辑：调用了 pipeFromArray(operations)(this)

operations就是 map(fn) 的返回值，pipeFromArray(operations)返回值只可能有3种情况

1. 返回一个identity函数，就是把传递的this原样返回，this是原可观察对象

2. 返回一个 fns[0] 函数这个 fns[0] 是 map(fn) 的调用的返回值，也是一个函数

3. 通过 reduce 依次迭代调用每个 map(fn)返回值函数，其中初始化值是原可观察对象，然后返回值函数会当作下一个返回值函数调用运行
map接收的 project 映射函数，map返回一个函数，返回的这个函数接收上一个可观察对象，订阅、然后返回一个新的可观察对象。
这个东西主要是依赖闭包、递归。

ok，非常Nice啊

这种对源码的探索精神可以保留~~，后面可以总结成笔记，发到Q群与大家分享~