[译] RxJS: 操作符状态管理

2,145 阅读4分钟

原文链接:RxJS: Managing Operator State
原文作者:Nicholas Jamieson;发表于2019年2月12日
译者:yk;如需转载,请注明出处,谢谢合作!

摄影:Victoire Joncheray,来自 Unsplash

在 RxJS 5.5 引入了管道操作符(pipeable operators)之后,编写用户级(userland)操作符变得更为简单了。

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

这听起来很简单。

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

举个例子

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

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

import { MonoTypeOperatorFunction } from "rxjs";
import { tap } from "rxjs/operators";

export function debug<T>(): MonoTypeOperatorFunction<T> {
  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] 2
second 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] 2
second 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<T>(): MonoTypeOperatorFunction<T> {
  return source => new Observable<T>(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<T>(): MonoTypeOperatorFunction<T> {
  return source => defer(() => {
    let index = -1;
    return source.pipe(
      tap(t => console.log(`[${++index}]: ${t}`))
    );
  });
}

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

译者注:accumulatorseedscan() 的两个参数,详见文档

import { MonoTypeOperatorFunction } from "rxjs";
import { map, scan } from "rxjs/operators";

export function debug<T>(): MonoTypeOperatorFunction<T> {
  return source => source.pipe(
    scan<T, [T, number]>(([, index], t) => [t, index + 1], [undefined!, -1]),
    map(([t, index]) => (console.log(`[${index}]: ${t}`), t))
  );
}

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

first use:
[0] 1
[1] 2
second use:
[0] 1
[1] 2

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