关于javascript:RxJS-源码解析五-Operator-III

DevUI 是一支兼具设计视角和工程视角的团队，服务于华为云 DevCloud 平台和华为外部数个中后盾零碎，服务于设计师和前端工程师。
官方网站：devui.design
Ng 组件库：ng-devui（欢送 Star）
官网交换群：增加 DevUI 小助手（微信号：devui-official）
DevUIHelper 插件：DevUIHelper-LSP（欢送 Star）

引言

在本文开始之前，先定义一些自定义术语，不便浏览。

• 顶流：调用了操作符的流。
• 上游流：操作符的外部订阅器所订阅的流。
• 上游流：由操作符的外部订阅器治理的流。
• 上游订阅：订阅了操作符生成的流的订阅者。

在上一篇中，我形容了 OuterSubscriber 和 InnerSubscriber 的作用，并将几个 Join Creation Operator 的源码解析了一遍。上面，咱们将进入的是 Transformation Operator 的源码剖析。

在晓得了 OuterSubscriber 和 InnerSubscriber 是一种通过委托模式实现治理上游流订阅的办法后，我发现其实这种实现技巧用于很多的 operators。那么本篇及下一篇将会介绍更多这种相似的设计。

（PS：为了不便形容，我拿了官网的相干的图片，时序图的浏览程序是从左到右。）

根底映射

map

map 是最为根底的映射，他将上游流的值通过运算，转发给上游订阅。这部分源码不是很简单，实际上就是做了一层转发。

protected _next(value: T) {
  let result: R;
  try {result = this.project.call(this.thisArg, value, this.count++);
  } catch (err) {this.destination.error(err);
    return;
  }
  this.destination.next(result);
}

scan

Scan 和 的作用跟 reduce 一样，传入一个函数把一组数据打平，然而跟 reduce 不一样的点在于，每一次联合结束都会立即返回后果。

const clicks1 = fromEvent(document, 'click');
const ones1 = clicks.pipe(mapTo(1));
const seed1 = 0;
const count1 = ones.pipe(
  // 输出的是返回任意值的函数
  scan((acc, one) => acc + one, seed)
);
count.subscribe(x => console.log(x));

这部分的实现同样也不是很简单，在拿到上游流数据后，应用 accumulator 对数据进行累加操作。

protected _next(value: T): void {
  // 须要判断是否带有初始值。if (!this.hasSeed) {
    this.seed = value;
    this.hasSeed = true;
    this.destination.next(value);
  } else {return this._tryNext(value);
    }
}

private _tryNext(value: T): void {
  const index = this.index++;
  let result: any;
  try {
      // 计算结果
    result = this.accumulator(<R>this.seed, value, index);
  } catch (err) {this.destination.error(err);
  }
    // 保留，以备下次应用
  this.seed = result;
  this.destination.next(result);
}

五种根底复合映射

所谓复合映射，意思就是这些操作符接管的参数是一个带有上游流数据作为参数并返回 Observable 的函数，同时把其中的订阅数据转发给上游订阅。

mergeMap，switchMap，exhaustMap，concatMap，mergeScan 是五种复合映射操作符，它使得上游流的数据能够传递给上游流，并交由其解决。concatMap 和 mergeScan 是 mergeMap 的一种非凡状况，所以咱们只须要关注残余的三种。

mergeMap，switchMap，exhaustMap，这三种操作符的源码构造分为这三个局部：

• 通过 lift 操作，将原有的流映射成新的流。
• 实现 Operator 接口，通过 call 返回一个 Subscriber。
• 通过继承 OuterSubscriber 实现这个 Subscriber。

其中，前两个局部都领有十分相似的构造，都是通过这种样板代码来进行编写。

export function someMap<T, R, O extends ObservableInput<any>>(project: (value: T, index: number) => O,
): OperatorFunction<T, ObservedValueOf<O> | R> {return (source: Observable<T>) => source.lift(new SomeMapOperator(project));
}

class SomeMapOperator<T, R> implements Operator<T, R> {constructor(private project: (value: T, index: number) => ObservableInput<R>) { }

  call(Subscriber: Subscriber<R>, source: any): any {return source.subscribe(new SomeMapSubscriber(Subscriber, this.project));
  }
}

通过 _innerSub 提供的外部注册办法，在外面创立 InnerSubscriber，并传入以后的 OuterSubscriber。

private _innerSub(input: ObservableInput<R>, value: T, index: number): void {const innerSubscriber = new InnerSubscriber(this, value, index);
  const destination = this.destination as Subscription;
  destination.add(innerSubscriber);
  const innerSubscription = subscribeToResult<T, R>(this, input, undefined, undefined, innerSubscriber);
  
  // 因为这里的 input 可能不是 observable，那么返回的
  // 订阅后果也可能跟 innserSubscriber 相等，所以这里要
  // 解决一下。if (innerSubscription !== innerSubscriber) {destination.add(innerSubscription);
  }
}

最终，交由 subscribeToResult 创立一个外部订阅来治理上游流。

mergeMap

mergeMap 提供的是一种合并操作，通过在外部保护了多个上游流的订阅，使得上游流能够将数据下发给多个上游流。它提供了一个并发数限度的参数，次要用于管制上游流并发的数量。

export function mergeMap<T, R, O extends ObservableInput<any>>(project: (value: T, index: number) => O,
  concurrent: number = Number.POSITIVE_INFINITY
): OperatorFunction<T, ObservedValueOf<O> | R> {return (source: Observable<T>) => source.lift(new MergeMapOperator(project, concurrent));
}

上面，咱们关注的点将转移到 MergeMapSubscriber。首先看看它的数据结构。

export class MergeMapSubscriber<T, R> extends OuterSubscriber<T, R> {
  // 标记是否曾经实现
  private hasCompleted: boolean = false;
  // 上流 observable 的数据缓存 
  private buffer: T[] = [];
  // 以后正在开启的流的数量
  private active: number = 0;
  // 数据的索引
  protected index: number = 0;

  constructor(
    // 内部传入的订阅者
    destination: Subscriber<R>,
    // 须要合并的 Observable 的工厂
    private project: (value: T, index: number) => ObservableInput<R>,
    // 并发数量
    private concurrent: number = Number.POSITIVE_INFINITY,
  ) {super(destination);
  }
  ...
}

Subscriber

MergeMapSubscriber 的 _next 调用的时候，会比拟 active（上游流的数量）与 concurrent（最大并发数）的大小，active 小于 concurrent 则调用 _tryNext，否则将曾经到来的数据放入缓冲区中，然而你晓得的，JavaScript 并没有真正的并发，这就是一个异步队列。而每一次进行 _tryNext，都会通过 project 来创立一个上游流，同时让更新 active，将上游流传入并触发 _innerSub。

protected _next(value: T): void {if (this.active < this.concurrent) {this._tryNext(value);
  } else {this.buffer.push(value);
  }
}

protected _tryNext(value: T) {
  let result: ObservableInput<R>;
  const index = this.index++;
  try {result = this.project(value, index);
  } catch (err) {this.destination.error(err);
    return;
  }
  this.active++;
  // 
  this._innerSub(result, value, index);
}

在上游流实现时，会触发 _complete。

protected _complete(): void {
  this.hasCompleted = true;
  if (this.active === 0 && this.buffer.length === 0) {this.destination.complete();
  }
  this.unsubscribe();}

如果所有的上游流都曾经实现，且缓冲区中没有数据，则告诉上游订阅数据曾经输入结束。

notify

notifyNext 就是单纯的将后果传递给上游订阅，而 notifyComplete 则有意思多了。

通过 notifyComplete，能够得悉哪些流曾经实现工作并且敞开。如果 buffer 中存在数据，那么将数据交由 _next 发送进来并创立新的上游流。过这种递归操作，能够将所有 buffer 中的数据都发送进来。最初判断上游流和上游流是不是都曾经完结了，如果曾经完结了，则告诉上游订阅数据曾经输入结束。

notifyNext(
  outerValue: T, innerValue: R,
  outerIndex: number, innerIndex: number,
  innerSub: InnerSubscriber<T, R>
): void {this.destination.next(innerValue);
}

notifyComplete(innerSub: Subscription): void {
  const buffer = this.buffer;
  this.remove(innerSub);
  this.active--;
  if (buffer.length > 0) {this._next(buffer.shift());
  } else if (this.active === 0 && this.hasCompleted) {this.destination.complete();
  }
}

switchMap

switchMap 提供的是一个上游流为主的映射操作，当上游流的订阅数据到来的时候，旧的上游流会被勾销订阅，而后从新订阅一组新的上游流。

export function switchMap<T, R, O extends ObservableInput<any>>(project: (value: T, index: number) => O
): OperatorFunction<T, ObservedValueOf<O>|R> {return (source: Observable<T>) => source.lift(new SwitchMapOperator(project));
}

Subscriber

innerSubscription 保留了以后上游流的订阅，所以这个操作符只会保护一个上游流的订阅。

private index: number = 0;
private innerSubscription: Subscription;

当进行 next 操作的时候，会先创立新的上游流，如果旧的上游流存在，那么会被勾销订阅。

protected _next(value: T) {
  let result: ObservableInput<R>;
  const index = this.index++;
  try {
    // 上游流的数据到来了，创立新的上游流。result = this.project(value, index);
  } catch (error) {this.destination.error(error);
    return;
  }

  // 旧的上游流勾销订阅
  const innerSubscription = this.innerSubscription;
  if (innerSubscription) {innerSubscription.unsubscribe();
  }

  this._innerSub(result, value, index);
}

该 Subscriber 重写了 _complete。这里意味着上游流曾经输入结束，那么如果上游订阅

protected _complete(): void {const {innerSubscription} = this;
  if (!innerSubscription || innerSubscription.closed) {super._complete();
    return;
  }
  this.unsubscribe();}

notify

跟之前一样，notifyNext 仍旧是将上游流中的数据转发进来。次要关注点还是在于 notifyComplete。因为 innerSubscription 被置为空了，所以调用 this._complete 无意义，不会触发到其父类函数。

notifyComplete(innerSub: Subscription): void {
  const destination = this.destination as Subscription;
  destination.remove(innerSub);
  this.innerSubscription = null;
  if (this.isStopped) {super._complete();
  }
}

如果以后的上游流曾经实现了，那么就要将它从上游订阅（destination）中移除，如果上游流曾经进行（error 或者 complete 被调用，或者被勾销订阅），那么还得调用 super._complete 示意曾经实现。

exhaustMap

跟 switchMap 相同，exhaustMap 提供了一种以上游流为主的映射操作。如果上游流曾经开启，那么上游流之后到来的订阅数据都将会被摈弃，直到该上游流实现订阅。上游流实现订阅后，上游流的数据才会持续跟新的上游流联合，并造成新的订阅。

export function exhaustMap<T, R, O extends ObservableInput<any>>(project: (value: T, index: number) => O,
): OperatorFunction<T, ObservedValueOf<O>|R> {return (source: Observable<T>) => source.lift(new ExhaustMapOperator(project));
}

Subscriber

exhaustMap 的实现很简略，通过保护 hasSubscription 这样一个外部状态，标记上游流是否被订阅了。hasCompleted 则是上游流实现状况的标记。

private hasSubscription = false;
private hasCompleted = false;

订阅会调用 _next，标记上游流是否曾经开启（订阅是否曾经存在），如果未开启，则构建新的上游流，并标记 hasSubscription 为 true。

protected _next(value: T): void {if (!this.hasSubscription) {
      let result: ObservableInput<R>;
      const index = this.index++;
      try {result = this.project(value, index);
      } catch (err) {this.destination.error(err);
        return;
      }
      // 标记为 true
      this.hasSubscription = true;
      this._innerSub(result, value, index);
  }
}

上游流和上游流的数据都曾经输入结束了，那么把实现信号传递给上游订阅。

protected _complete(): void {
  this.hasCompleted = true;
  if (!this.hasSubscription) {this.destination.complete();
  }
  this.unsubscribe();}

notify

如果上游流的数据输入结束，那么就应该要将 hasSubscription 标记为 false。

notifyComplete(innerSub: Subscription): void {
  const destination = this.destination as Subscription;
  destination.remove(innerSub);

  // 标记为 false
  this.hasSubscription = false;

  // 此处判断上游流是否曾经实现
  if (this.hasCompleted) {this.destination.complete();
  }
}

concatMap

concatMap 是 mergeMap 的一种非凡模式。

export function concatMap<T, R, O extends ObservableInput<any>>(project: (value: T, index: number) => O,
): OperatorFunction<T, ObservedValueOf<O>|R> {return mergeMap(project, 1);
}

mergeScan

mergeScan 的源码跟 mergeMap 相似。只不过就是把传入的函数替换了一下，并且在外部缓存了上一个联合后的值。

const clicks2 = fromEvent(document, 'click');
const ones2 = click$.pipe(mapTo(1));
const seed2 = 0;
const count2 = one$.pipe(
  // 输出一个 Observable 工厂
  mergeScan((acc, one) => of(acc + one), seed),
);

concat & merge

上一篇中，对于 concat 和 merge 两个相干的 operators 并没有讲到，因为这它们其实最终都是调用 mergeMap。

小结

通过这三个不同的映射操作符，使得上游流能够通过肯定的形式跟上游流联合。那么，联合一张图，能够看看相干操作符的关系。

对这些操作符分一下类。

• 属于 Transformation Operators 的有：concatMap，concatMapTo，mergeMap，mergeMapTo，switchMap，switchMapTo，exhaustMap，exhaustMapTo。
• 属于 Join Creation Operators 的有：merge, concat。
• 属于 Join Operators 的有：mergeAll，concatAll，switchAll，startWith，endWith。

零散的高阶操作符

expand

expand 将传入的 Observable 工厂进行递归操作。与下面的复合映射相似，expand 也是一种复合映射，只不过，他会一直的去复合上游流的数据，也就是相似上图的模式。

Subscriber

为了实现绝对应的性能，expand 定义了以下数据结构。

export class ExpandSubscriber<T, R> extends OuterSubscriber<T, R> {
  // 以后索引
  private index: number = 0;
  // 已启动的上游流的数量
  private active: number = 0;
  // 上游流是否曾经实现
  private hasCompleted: boolean = false;
  // 对于索引的缓存数据
  private buffer: any[];
  // 上游流工厂
  private project: (value: T, index: number) => ObservableInput<R>,
    // 并发数  
     private concurrent: number;
}

上游流数据到来的时候，跟 mergeMap 比拟相似，也会比拟 active 和 concurrent，如果 active 大于 concurrent，那么便会用 buffer 缓存上游流的数据，如果 active 小于 concurrent，那么间接发送数据给到上游订阅，并订阅一个新的上游流。须要留神的一点，为了避免爆栈，expand 在这里加了一个判断条件，在 notify 中，将利用这一条件，来完结递归。

protected _next(value: any): void {
  const destination = this.destination;

  if (destination.closed) {this._complete();
    return;
  }

  const index = this.index++;
  if (this.active < this.concurrent) {destination.next(value);
    try {const { project} = this;
      const result = project(value, index);
      this.subscribeToProjection(result, value, index);
    } catch (e) {destination.error(e);
    }
  } else {this.buffer.push(value);
  }
}
// 订阅新的上游流
private subscribeToProjection(result: any, value: T, index: number): void {
  this.active++;
  const destination = this.destination as Subscription;
  destination.add(subscribeToResult<T, R>(this, result, value, index));
}

当上游流实现时，须要标记 hasComplete 为 true。这一步是完结递归的重要标记。

protected _complete(): void {
  this.hasCompleted = true;
  if (this.hasCompleted && this.active === 0) {this.destination.complete();
  }
    this.unsubscribe();}

notify

那么 expand 是怎么形成递归的呢，当上游流有数据到来的时候，他会间接调用 _next。最终造成了 _next -> subscribeToProjection -> next -> notifyNext -> _next 这样的一条递归链。

notifyNext(
  outerValue: T, 
  innerValue: R,
    outerIndex: number, 
  innerIndex: number,
    innerSub: InnerSubscriber<T, R>
): void {this._next(innerValue);
}

上游流实现时，须要依据 hasCompleted 和 buffer 的状态来决定是否完结递归。在这里，也造成了一条这样的递归链： _next -> subscribeToProjection -> next -> notifyComplete -> _next。

notifyComplete(innerSub: Subscription): void {
  const buffer = this.buffer;
  const destination = this.destination as Subscription;
  destination.remove(innerSub);
  this.active--;
  if (buffer && buffer.length > 0) {this._next(buffer.shift());
  }
  if (this.hasCompleted && this.active === 0) {this.destination.complete();
  }
}

exhaust

exhaust 是一种打平操作，它的源码并没有调用 exhaustMap。它的实现思路很简略，通过判断以后是否存在前一个上游流订阅（hasSubscription），来决定以后到来的上游流是否开启。

private hasCompleted: boolean = false;
private hasSubscription: boolean = false;


protected _next(value: T): void {
  // 如果存在订阅，那么摈弃这个值
  if (!this.hasSubscription) {
    this.hasSubscription = true;
    this.add(subscribeToResult(this, value));
  }
}

protected _complete(): void {
  this.hasCompleted = true;
  if (!this.hasSubscription) {this.destination.complete();
  }
}

notifyComplete(innerSub: Subscription): void {this.remove(innerSub);
  this.hasSubscription = false;
  if (this.hasCompleted) {this.destination.complete();
  }
}

总结

本篇次要的内容集中在剖析操作符是如何进行数据的映射，那么下一篇将解说的是 buffer 和 window 相干的缓存操作符是如何运行和实现的。

退出咱们

咱们是 DevUI 团队，欢送来这里和咱们一起打造优雅高效的人机设计 / 研发体系。招聘邮箱：muyang2@huawei.com。

作者：zcx（公众号：Coder 写字的中央）

原文链接：https://mp.weixin.qq.com/s/lrawMOuHNj6GyQJMqK1Now

往期文章举荐

《好用到飞起！VSCode 插件 DevUIHelper 设计开发全攻略（三）》

《Web 界面深色模式和主题化开发》

《手把手教你搭建一个灰度公布环境》