RxJS（Reactive Extensions for JavaScript） 是一个非常强大的 JS 库，我们可以使用它轻松编写异步代码。

在本系列文章中，我将带领你学习 RxJS 的最新版本，我们会重点关注如何使用响应式编程范式来解决你在日常工作中碰到的问题。所以这是一个偏实战的系列文章。

在本系列文章中，你将学会 RxJS 中的核心组件是如何使用和运作的。

通过学习这个系列文章，你将亲自使用 RxJS 完成一个完整的项目开发，在这个项目中，你将了解如何处理 DOM 事件、如何构建响应式本地数据库等内容。

RxJS 中测试思路

响应式编程范式与每一个软件开发范式一样，都需要使用测试来维护程序的正确性和可扩展性。

在 RxJS 中，我们最常见的场景就是订阅者订阅一个 Observable 并接收一些值，在这种场景下，我们只需要在对订阅者预期收到的值和实际收到的值进行比对就可以知道程序是否正确。

我们可以按照这种思路开始测试。

传统的测试方法

在正式开始测试之前，你需要对 Mocha.js 有一个基本的了解，因为这是我们接下来要用到的测试库。但是我不会在这里进行详细介绍。

我们首先来看下面这段代码：

const observable$ = getObservableForTest()
it('测试惰性 Observable', () => {
  observable$.subscribe(callbackFn)
})

在这段代码中，我们订阅了需要在测试回调中测试的 Observable。但是测试是同步执行的，这就意味着它会在订阅 Observable 后立即完成。我们希望测试程序在 Observable 发出值，并且验证是否与预期像等候再结束运行。

现在我们要使用 Mocha 中的 done 来手动告知测试程序应该在什么时候完成。

const observable$ = getObservableForTest()
it('测试惰性 Observable', (done) => {
  observable$.subscribe(val => {
    done()
  })
})

这样就满足了我们的需求，但是需要注意一点：这种方式是存在缺点的，我们必须记住要调用 done 方法。

现在我们继续添加一些测试逻辑来完成单元测试，我们测试实际发出的值是否为 1。

const observable$ = getObservableForTest()
it('测试惰性 Observable', (done) => {
  observable$.subscribe(val => {
    expect(val).toBe(1)
    done()
  })
})

到这里，测试单个值的单元测试就完成了。

但是我们该怎么样来进行多个值的测试呢？

这是这种测试方法的第二个缺点：测试发送多个值的 Observable 很困难，我们要编写很复杂的测试逻辑。

一种解决方法是使用 toArray 操作符，这个操作符可以等待 Observable 完成，然后将所有的值以数组的形式一次性发送给订阅者。我们可以使用这种方式编写测试代码：

const observable$ = getObservableForTest()
describe('异步', function () {
    it('转换成 Array', function (done) {
      from([1, 2, 3]).pipe(toArray()).subscribe(val => {
        assert.deepEqual(val, [1, 2, 3])
        done()
      })
    });
});

我们在这个测试中注意到一个问题，在响应式编程中，时间是必不可少的关键因素，因为数据是随着时间推移而产生的，无论是定期的还是不定期的。如果我们使用这种方法进行测试，我们无法知道第一次发出值的时间以及第二次、第三次发出的时间。

还有，我们没有办法使用这种测试方法在数据流中对数据进行转换。

所以，传统的测试方法并不适合在 RxJS 中进行测试。Marble 是一种更好的测试方案，它可以解决我在上面提到的所有缺点。

Marble

marble 可以使用虚拟时间来同步测试异步的 Observable。RxJS 中的虚拟时间是通过 Scheduler 提供的。

Scheduler

Scheduler 是一种数据结构，可以将不同的异步任务排队并根据标准执行它们，标准指的是优先级或者不同的执行算法。它提供了虚拟时钟 now 方法。Scheduler 调度的每个任务都是相对于虚拟时钟执行的。Scheduler 也分为多种类型，例如 AsyncScheduler、QueueScheduler、AsapScheduler 等。

我们可以使用 TestScheduler 来测试 Observable，因为它是内置的，不需要额外安装依赖。

Marble test

我们在学习操作符的时候，展示过一些示例图，这种图就是 Marble 图。这种图可以用来当作编写测试的参考。

下面是一个 Marble 图，它表示一些小写的字符是如何随着时间推移被映射为大写字符的。

下面是根据 Marble 图实现的同步代码。

import { from } from 'rxjs'
import { map }  from 'rxjs/operators'
const lowercaseCharsObservable$ = from(['a', 'b', 'c'])
lowercaseCharsObservable$
    .pipe(map(character => character.toUpperCase()))
    .subscribe(console.log)

对应的，还有异步代码。

import {Subject} from 'rxjs'
import { map } from 'rxjs/operators'
const lowercaseCharsObservable$ = new Subject()
lowercaseCharsObservable$
    .pipe(map(character => character.toUpperCase()))
    .subscribe(console.log)
setTimeout(() => {lowercaseCharsObservable$.next('a')}, 500)
setTimeout(() => {lowercaseCharsObservable$.next('b')}, 1000)
setTimeout(() => {lowercaseCharsObservable$.next('c')}, 2000)

很明显，异步的实现更难以测试，但是 Marble 可以让测试它们变得非常简单。

下面我们学习一下 Marble 语法，然后将这个用例作为代表，使用 Marble 进行测试。

Marble 语法

我们可以通过 Marble 语法创建操作序列，用声明的方式允许 Observable 的预期行为。

下面是一些 Marble 符号。

下面是一个简单的 Marble 语句，描述了第一个 Observable，它在 Observer 订阅后的第二、四、六帧时发出值，并在第六帧后完成。

^--aaa|

现在我们对 Marble 语法有了一个初步的认识，接下来我们思考一下，该如何利用 Marble 语法来测试异步代码呢？

假设一个虚拟时间单位是 1 ms，我们可以像下面这样写：

-a------c---|

为了编写测试，我们需要一个 TestScheduler 实例。

下面的示例通过定义 hot Observable 演示了对上面示例中 Observable 的测试。

import { TestScheduler } from 'rxjs/testing'
import { throttleTime } from 'rxjs/operators'
const expect = require('chai').expect
const testScheduler = new TestScheduler((actual, expected) => {
  expect(actual).deep.equal(expected);
});
it('Test uppercase', () => {
  testScheduler.run(helpers => {
    const { hot, expectObservable } = helpers;
    const values = {
      a: 'A',
      b: 'B',
      c: 'C'
    }
    var characterObservable$ = hot('-a--b--c---|', values);
    var expectedMarbleVisual =      '-a-----c---|';
    expectObservable(characterObservable$.pipe(throttleTime(3, testScheduler))).toBe(expectedMarbleVisual, values);
  });
});

如果我们想要精确地测试不同用例的时间该怎么办？Marble 也支持下面这种语法：

499ms a 499ms b 999ms (c|)

上面这段 Marble 语句表示：在订阅 Observable 500 毫秒后，发出第一个值，再过 500 毫秒后发出第二个值，再间隔 1000 毫秒，发出第三个值。其中我们在每个表示时间的位置都减少了 1 毫秒，原因是发出值本身需要 1 毫秒。