[译] RxJS: 操作符状态管理-白红宇

原文链接：
原文作者：；发表于2019年2月12日
译者：；如需转载，请注明，谢谢合作！

摄影：，来自

在 RxJS 5.5 引入了之后，编写用户级（userland）操作符变得更为简单了。

管道操作符属于高阶函数（higher-order function）：即返回值为函数的函数。所返回的函数接受一个 observable（可观察对象）作为参数，并返回一个 observable。所以，要创建一个操作符，你不必划分 Operator 和 Subscriber，只要写一个函数就行了。

这听起来很简单。

然而在某些情况下你需要格外小心，尤其是当你的操作符在存储内部状态时更应如此。

举个例子

让我们来看这样一个例子：一个将接收到的数据及其索引显示到终端的 debug 操作符。

我们的操作符需要维护一些内部状态：索引——每收到一次 next 通知时就会递增。有个很戆的办法是直接将状态存储在操作符的内部，就像这样：

import { MonoTypeOperatorFunction } from "rxjs";import { tap } from "rxjs/operators";export function debug
    
     (): MonoTypeOperatorFunction
     
       {  let index = -1;  // 让我们假设不存在 map 操作符，于是我们只能用 tap 来维护内部存储中的索引  // 该操作符的目的是为了表明：运行结果取决于状态存储的位置  return tap(t => console.log(`[${++index}]: ${t}`));}复制代码

该办法会存在许多问题，并导致一些意料之外的行为和难以定位的 bug。

存在的问题

第一个问题是：我们的操作符不具有引用透明（referentially transparent）性。当一个函数的返回值可以替代该函数而不影响程序运行，那么我们称这个函数是引用透明的。

让我们来看看当这个操作符的返回值与多个 observables 进行组合时会发生什么：

import { range } from "rxjs";import { debug } from "./debug";const op = debug();console.log("first use:");range(1, 2).pipe(op).subscribe();console.log("second use:");range(1, 2).pipe(op).subscribe();复制代码

运行结果为：

first use:[0] 1[1] 2second use:[2] 1[3] 2复制代码

好吧，我知道这很令人意外。在第二个 observable 中，索引并没有从 0 开始计数。

第二个问题是：只有在首次订阅该操作符返回的 observable 时，其行为才会是合理的。

现在，让我们多次订阅由 debug 操作符组成的 observable，看看会发生什么：

import { range } from "rxjs";import { debug } from "./debug";const source = range(1, 2).pipe(debug());console.log("first use:");source.subscribe();console.log("second use:");source.subscribe();复制代码

运行结果为：

first use:[0] 1[1] 2second use:[2] 1[3] 2复制代码

还是同样令人意外的结果：在第二次订阅中，索引依旧没有从 0 开始计数。

所以该如何解决这些问题呢？

解决方案

这两个问题都可以通过基于每个订阅的状态存储（storing the state on a per-subscription basis）来解决。以下是几种实现方法：

第一种方法是使用 Observable 的构造函数来创建操作符返回值（observable）。如果将 index 变量放入传给 constructor 的函数中，那么每次订阅的状态都会被独立存储。写法如下：

import { MonoTypeOperatorFunction, Observable } from "rxjs";import { tap } from "rxjs/operators";export function debug
    
     (): MonoTypeOperatorFunction
     
       {  return source => new Observable
      
       (subscriber => {    let index = -1;    return source.pipe(      tap(t => console.log(`[${++index}]: ${t}`))    ).subscribe(subscriber);  });}复制代码

第二种方法，也是我比较喜欢的，就是使用 defer 来实现基于每个订阅的状态存储。如果将 index 变量放入传给 defer 的工厂函数中，它就可以按每个订阅独立存储状态。写法如下：

译者注：defer() 的参数为一个返回值为 observable 的工厂函数 observableFactory，详见。

import { defer, MonoTypeOperatorFunction } from "rxjs";import { tap } from "rxjs/operators";export function debug
    
     (): MonoTypeOperatorFunction
     
       {  return source => defer(() => {    let index = -1;    return source.pipe(      tap(t => console.log(`[${++index}]: ${t}`))    );  });}复制代码

还有个较为复杂的方法，就是使用 scan 操作符。scan 会维护每个订阅的状态，该状态由 seed 参数进行初始化，然后通过 accumulator（累加器）函数计算并返回结果。在本例中，index 可以像这样存储在 scan 中：

译者注：accumulator 和 seed 为 scan() 的两个参数，详见。

import { MonoTypeOperatorFunction } from "rxjs";import { map, scan } from "rxjs/operators";export function debug
    
     (): MonoTypeOperatorFunction
     
       {  return source => source.pipe(    scan
      
       (([, index], t) => [t, index + 1], [undefined!, -1]),    map(([t, index]) => (console.log(`[${index}]: ${t}`), t))  );}复制代码

如果用以上任意一种方法来代替一开始那个很戆的办法，输出将会是下面这样：

first use:[0] 1[1] 2second use:[0] 1[1] 2复制代码

如你所愿：一切都在意料之中。