image.png

来源

想法

  1. // 观察者接口
  2. interface Observer<T> {
  3.   next: (value: T) => void;
  4.   error: (value: any) => void;
  5.   complete: () => void;
  6. }
  8. // 单元函数 一元函数 类型别名
  9. type UnaryFunction<T, R> = (source: T) => R;
  11. // 操作符函数 类型别名
  12. type OperatorFunction<T, R> = UnaryFunction<Observable<T>, Observable<R>>;
  14. // 拆分逻辑 类型别名
  15. type TeardownLogic = Subscription | Unsubscribable | void | (() => void);
  17. // 不予订阅 接口
  18. interface Unsubscribable {
  19.   unsubscribe: () => void;
  20. }
  22. // 从数组开始的管道
  23. function pipeFromArray<T, R>(
  24.   /* 数组  元素为单元函数 */
  25.   fns: Array<UnaryFunction<T, R>>
  26. ): UnaryFunction<T, R> {
  27.   // 如果数组元素为空
  28.   if (fns.length === 0) {
  29.     // 返回一个匿名箭头函数，此函数返回参数本身
  30.     return (input) => input as any as R;
  31.   }
  33.   // 如果数组元素只有一个
  34.   if (fns.length === 1) {
  35.     return fns[0];
  36.   }
  38.   // 排除以上情况
  39.   // 返回一个匿名箭头函数，此函数返回以参数位初始值的reduce fns数组
  40.   return (input: T): R => {
  41.     /*
  42.     reduce为数组中的每一个元素依次执行callback函数，不包括数组中被删除或从未被赋值的元素，接受四个参数：
  44.     accumulator 累计器
  45.     currentValue 当前值
  46.     currentIndex 当前索引
  47.     array 数组
  48.     回调函数第一次执行时，accumulator 和currentValue的取值有两种情况：
  49.     如果调用reduce()时提供了initialValue，accumulator取值为initialValue，
  50.     currentValue取数组中的第一个值；如果没有提供 initialValue，
  51.     那么accumulator取数组中的第一个值，currentValue取数组中的第二个值。
  53.     实例：
  54.     const array1 = [1, 2, 3, 4];
  55.     const reducer = (previousValue, currentValue) => previousValue + currentValue;
  56.     // 5 + 1 + 2 + 3 + 4
  57.     console.log(array1.reduce(reducer, 5));
  58.     // expected output: 15
  59.      */
  60.     return fns.reduce((prev: any, fn: any) => fn(prev), input);
  61.   };
  62. }
  64. // 订阅类 实现了不予订阅接口
  65. class Subscription implements Unsubscribable {
  66.   // 拆解
  67.   // private _teardowns: /* 不包含 */Exclude<TeardownLogic, void>[] = []; // 数组第一种写法
  68.   private _teardowns: /* 不包含 */Array<Exclude<TeardownLogic, void>> = []; // 数组第二种写法
  69.   unsubscribe(): void { // 取消订阅
  70.     this._teardowns.forEach((teardown/* 拆卸 */) => {
  71.       if (typeof teardown === 'function') { // 如果拆卸是函数
  72.         teardown(); // 执行此函数
  73.       } else { // 否则
  74.         teardown.unsubscribe(); // 执行unsubscribe方法
  75.       }
  76.     });
  77.   }
  78.   add(teardown: TeardownLogic): void {
  79.     if (teardown /* 拆卸 */) { // 如果teardown非空
  80.       this._teardowns.push(teardown); // 将teardown添加到_teardowns
  81.     }
  82.   }
  83. }
  85. // 订阅者类 继承了订阅类 且实现了观察者接口
  86. class Subscriber<T> extends Subscription implements Observer<T> {
  87.   private isStopped = false; // 已停止
  88.   type: string;
  89.   // 构造函数
  90.   constructor(private observer: Partial<Observer<T>>/* 传入观察者实例 */, type: string) {
  91.     super(); // 执行父类构造函数
  92.     this.type = type;
  93.   }
  95.   // Observer中的next方法
  96.   next(value: T) {
  97.     // 传入的观察者实例
  98.     if (this.observer.next /* 如果传入观察者实例存在next方法 */ && !this.isStopped/* 没有停止 */) {
  99.       this.observer.next(value); // 执行next方法，传入值
  100.     }
  101.   }
  103.   // Observer中的error方法
  104.   error(value: any) {
  105.     // 有错误，则将isStopped设置为true，停止
  106.     this.isStopped = true;
  107.     if (this.observer.error) { // 如果存在error方法
  108.       this.observer.error(value); // 执行error方法并传入值
  109.     }
  110.   }
  112.   // Observer中的complete方法
  113.   complete() {
  114.     // 完成，则将isStopped设置为true，停止
  115.     this.isStopped = true;
  116.     if (this.observer.complete) {// 如果存在complete
  117.       this.observer.complete(); // 执行complete方法
  118.     }
  119.     if (this.unsubscribe) { // 如果存在unsubscribe方法
  120.       this.unsubscribe(); // 执行unsubscribe方法
  121.     }
  122.   }
  123. }
  125. // 可观察类
  126. export class Observable<T> {
  127.   type: string;
  128.   // 构造函数
  129.   constructor(private _subscribe: (observer: Observer<T>) => TeardownLogic, type: string = "user") {
  130.     this._subscribe = _subscribe;
  131.     this.type = type;
  132.   }
  134.   // 订阅函数
  135.   subscribe(observer: Partial<Observer<T>>): Subscription {
  136.     const subscriber = new Subscriber(observer, this.type); // 把观察者传入订阅者
  137.     subscriber.add(this._subscribe(subscriber)); // 父类Subscription中的add方法，将订阅者放入_teardowns中
  138.     return subscriber; // Subscriber继承了Subscription，所以此处可以返回
  139.   }
  141.   // 管道
  142.   pipe(...operations/* 操作 */: OperatorFunction/* 操作符函数 */<any, any>[]): Observable<any> {
  143.     return pipeFromArray(operations)(this); // 执行从数组开始的管道函数
  144.   }
  145. }
  147. // map操作符
  148. export function map<T, R>(project: (value: T, index: number) => R) {
  149.   return (observable: Observable<T>) =>
  150.     new Observable<R>((subscriber) => {
  151.       let i = 0;
  152.       const subcription = observable.subscribe({
  153.         next(value) {
  154.           return subscriber.next(project(value, i++));
  155.         },
  156.         error(err) {
  157.           subscriber.error(err);
  158.         },
  159.         complete() {
  160.           subscriber.complete();
  161.         },
  162.       });
  163.       return subcription;
  164.     }, "map");
  165. }
  167. // 防抖
  168. // debounceTime 延时发送源 Observable 发送的值,但是会丢弃正在排队的发送如果源 Observable 又发出新值。
  169. export function debounceTime<T, R>(delay: number) {
  170.   let time: number = 0;
  171.   return (observable: Observable<T>) =>
  172.     new Observable<R>((subscriber) => {
  173.       const subcription = observable.subscribe({
  174.         next(value) {
  175.           if (time) {
  176.             clearTimeout(time)
  177.           }
  178.           time = setTimeout(() => {
  179.             return subscriber.next(value as any as R);
  180.           }, delay)
  181.         },
  182.         error(err) {
  183.           clearTimeout(time)
  184.           subscriber.error(err);
  185.         },
  186.         complete() {
  187.           clearTimeout(time)
  188.           subscriber.complete();
  189.         },
  190.       });
  191.       return subcription;
  192.     }, "debounceTime");
  193. }

源码仓库：

mini-rxjs

RxJS的冷与热

http://jsbin.com/wabuguy/1/edit?js,console,output
https://github.com/RxJS-CN/rxjs-articles-translation/blob/master/articles/Hot-Vs-Cold-Observables.md
https://juejin.cn/post/6959003628816302087
因为它是一个 “冷 “观察变量，所以每次订阅同一个观察变量都会创建一个新的连接。
每次你订阅同一个可观察对象时，都会创建一个新的连接。
（也就是说，每次调用函数。因为可观察变量只是函数)

TL;DR: 当不想一遍又一遍地创建生产者( producer )时，你需要热的 Observable 。

冷（冷清 PS：我的理解）的是指 Observable 创建了生产者

  1. // 冷的 
  2. var cold = new Observable((observer) => {   var producer = new Producer();   
  3. // observer 会监听 producer 
  4. });

热（热闹 PS：我的理解）的是指 Observable 复用生产者

  1. // 热的 
  2. var producer = new Producer(); 
  3. var hot = new Observable((observer) => {   
  4. // observer 会监听 producer 
  5. });

加上了我的理解，否则抽象出来的冷与热属实很抽象！