RxJs map operator 工作原理分析

使用一个例子来研究 map 操作符的工作原理。

推荐阅读本文之前，先浏览这篇文章RxJs fromEvent 工作原理分析以了解相关知识。

源代码：

import { Component, OnInit, Inject } from '@angular/core';
import { fromEvent, combineLatest } from 'rxjs';
import { mapTo, startWith, scan, tap, map } from 'rxjs/operators';
import { DOCUMENT } from '@angular/common';

@Component({
  selector: 'app-combine-latest',
  templateUrl: './combine-latest.component.html'
})
export class CombineLatestComponent implements OnInit {
  readonly document: Document;

  constructor(
    // https://github.com/angular/angular/issues/20351
    @Inject(DOCUMENT) document: any) {
      this.document = document as Document;
    }

  redTotal:HTMLElement;
  blackTotal: HTMLElement;
  total:HTMLElement;  
  test:HTMLElement;

  ngOnInit(): void {
    this.redTotal = this.document.getElementById('red-total'); 
    this.blackTotal = this.document.getElementById('black-total');
    this.total = this.document.getElementById('total');
    this.test = this.document.getElementById('test');

    combineLatest(this.addOneClick$('red'), 
    
    this.addOneClick$('black')).subscribe(([red, black]: any) => {
      this.redTotal.innerHTML = red;
      this.blackTotal.innerHTML = black;
      this.total.innerHTML = red + black;
    });

    fromEvent(this.test, 'click').pipe(map( event => event.timeStamp)).subscribe((event) => console.log(event));
  }

  addOneClick$ = id =>
  fromEvent(this.document.getElementById(id), 'click').pipe(
    // map every click to 1
    mapTo(1),
    // keep a running total
    scan((acc, curr) => acc + curr, 0),
    startWith(0)
  );
}

打开页面，点击 Test 按钮，能在 Chrome 控制台里看到每次点击发生时的 timestamp 时间戳：

下面介绍 map 操作符是如何起作用的。

先缕一缕顺序：

1. 首先执行fromEvent，返回一个 Observable 对象。

2. 执行 map 操作符，其结果作为输入，传入 pipe

2.执行 pipe：

3. 执行 subscribe 操作。

我们可以把 pipe 形象地想象成管道，通过 fromEvent 返回的 Observable 对象，流过一根根管道，最后触发其订阅者，执行订阅者的逻辑。那么 RxJs 提供的各种 operator，就是安装在管道里的处理器。

map 操作的输入是我们定义的映射函数，在 RxJs 上下文里，称为 project:

map 返回一个新的函数，名为 mapOperation. 新函数体里，基于传入的 project，创建一个新的 MapOperator. 这个 MapOperator，作为新函数输入参数 source 的 lift 方法调用的输入参数。到现在为止，我们尚且不知道 source 参数的类型。

接下来执行 Observable 的 pipe 方法。

operations 参数是 map operator 返回的新函数，mapOperation：

pipeFromArray 的实现，如果 pipe 输入只有一个 operator，这种情况比较简单，进入第 9 行的 IF 分支，直接将 map 返回的 mapOperation 函数作为 pipeFromArray 调用的返回结果。

注意到 Observable.js 实现里，在 pipeFromArray(operations) 返回之后，紧跟了另一个括号，说明这是另一个函数调用，输入参数为 this，即 Observable 对象本身。

现在进入到 map 操作返回的 新函数 mapOperation 的函数体内部了：

因为此时 button 尚未点击，因此 Observable 对象并没有 emit 值，只是完成相关的 setup 工作。

这行语句：

return source.lift(new MapOperator(project, thisArg));

只是返回一个新的 Observable 对象，其 source 属性指向调用 lift 操作的原始 Observable 对象，而 operator 属性指向 new MapOperator 返回的结果，后者是 project 的 wrapper.

如此一来，调用 subscribe 方法注册应用程序监听函数的 Observable 对象，再也不是 fromEvent 返回的原始 Observable 对象，而是前者调用了 pipe，接收了 map 指定的 project 之后，由 source.lift( new MapOperator) 返回的新 Observable 对象。

这个新的 Observable 对象，调用 subscribe 方法，执行逻辑和这篇文章RxJs fromEvent 工作原理分析介绍的相比有所差异，复杂度稍稍增加了。

把 Observable 对象 operator 属性值提取出来：

接下来的 21行代码执行，和之前没有 operator 时相比，没有差异，略过。

前一篇文章进入 ELSE 分支，而本文因为 operator 的存在，进入 22 行的 IF 分支：

首先执行 operator.call 方法：

MapSubscriber 也是 Subscriber 的子类之一，和其父类相比，多了 project 属性。

再次执行 subscribe：

因为这次传入的 Observable 是最原始的即 fromEvent 返回的 Observable，因此不存在 operator，所以进入 ELSE 分支执行：

重点分析 this 和 sink：

this 是 fromEvent 返回的原始 Observable，而 sink 是包含了 map operator 以及应用程序定义的订阅逻辑的 Subscriber：

_trySubscribe 调用 _subscribe:

最终仍旧进入了 fromEvent 的核心逻辑：

这段代码，定义了 fromEvent，以什么样的方式，emit 何种类型的数据。

• 什么样的方式？addEventListener，每次 eventTarget 定义的 HTMLElement 发生 click 事件时，emit 数据
• emit 的数据格式为 MouseEvent.
  

至此 Observable 相关的 setup 执行完毕。

点击按钮，触发之前通过 addListener 注册的 handler 函数。fromEvent.js 此处 subscriber 不是原始的 subscriber，而是 MapSubscriber，其 destination 属性的 _next, 指向了应用程序指定的订阅处理逻辑。Emit 的数据是 MouseEvent.

MapSubscriber 的特色：在将原始值 MouseEvent 交给应用程序之前，先要执行 project 对其进行处理：

这个 project 的逻辑是，将 MouseEvent 对象映射成 timestamp 时间戳：

将 project 处理结果返回给destination 继续进行传递：

this._next 指向的是应用程序定义的 console.log(event), 在这里得到执行：