RxJS v4 测试工具指南：深入理解 `Rx.ReactiveTest` 与虚拟时间断言体系

RxJS v4 测试工具指南：深入理解Rx.ReactiveTest与虚拟时间断言体系

【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址: https://gitcode.com/gh_mirrors/rxj/RxJS

导读

Rx.ReactiveTest是 RxJS v4 测试体系（rx.testing.js）中的核心工具类，为单元测试提供「在指定虚拟时间点构造通知记录」的工厂方法，以及三个标准的虚拟时间常量。本文以 ReactiveTest 官方文档 为骨架，结合 源码实现、TestScheduler 及其配套的 Recorded / Subscription / MockObserver 等测试组件，系统讲解如何用onNext/onError/onCompleted构造时间戳化通知、如何用created/subscribed/disposed控制测试时间轴，并给出可直接运行的完整测试用例。读完本文，你将掌握 RxJS v4 虚拟时间测试的完整方法论，能够独立编写可断言的响应式流测试。

一、Rx.ReactiveTest是什么

Rx.ReactiveTest是一个静态工具类（本质是一个挂载在Rx.ReactiveTest上的对象字面量），用于创建用于测试目的的通知记录（notification records）。它位于rx.testing.js中，是Rx.TestScheduler虚拟时间测试体系的基础设施。

从源码看，Rx.ReactiveTest对象定义了三个字段与四个工厂方法（reactivetest.js）：

1.1 与 Recorded / Subscription 的关系

ReactiveTest工厂方法返回的对象并非普通通知，而是携带虚拟时间戳的记录：

• onNext/onError/onCompleted返回 Rx.Recorded 实例，内部封装一个 Rx.Notification；
• subscribe返回 Rx.Subscription 实例，记录订阅与退订时间。

这一点在 reactivetest.js 中体现得非常直接：

onNext: function (ticks, value) { return typeof value === 'function' ? new Recorded(ticks, new OnNextPredicate(value)) : new Recorded(ticks, Notification.createOnNext(value)); }, onError: function (ticks, error) { return typeof error === 'function' ? new Recorded(ticks, new OnErrorPredicate(error)) : new Recorded(ticks, Notification.createOnError(error)); }, onCompleted: function (ticks) { return new Recorded(ticks, Notification.createOnCompleted()); }, subscribe: function (start, end) { return new Subscription(start, end); }

关键点：onNext与onError的第二个参数既可以是具体的值/错误对象，也可以是一个谓词函数（predicate）。传入函数时，源码会分别包装成OnNextPredicate/OnErrorPredicate（reactivetest.js），其equals方法会在断言时用该函数对实际值求值，从而实现「按条件比较」而非「严格相等比较」。

二、类方法详解：三个通知工厂

2.1Rx.ReactiveTest.onCompleted(ticks)

在给定时间创建 OnCompleted 通知记录的工厂方法。

• 参数：ticks(Number)—— OnCompleted 通知发生的记录虚拟时间。
• 返回值：(Recorded)—— OnCompleted 通知记录。
• 位置：rx.testing.js（源码见 reactivetest.js）

示例：构造一个在虚拟时间260完成的热可观察序列，并断言结果：

var onCompleted = Rx.ReactiveTest.onCompleted; var scheduler = new Rx.TestScheduler(); var xs = scheduler.createHotObservable( onCompleted(260) ); var res = scheduler.startScheduler(function () { return xs.map(function (x) { return x; }); }); // 编写自定义断言 collectionAssert.assertEqual(res.messages, [ onCompleted(260) ]);

2.2Rx.ReactiveTest.onError(ticks, exception)

在给定时间、携带给定错误创建 OnError 通知记录的工厂方法。

• 参数：
  1. ticks(Number)—— OnError 通知发生的记录虚拟时间；
  2. exception(Error | Function)—— 存入 OnError 通知的错误对象，或一个谓词函数。
• 返回值：(Recorded)—— OnError 通知记录。
• 位置：rx.testing.js（源码见 reactivetest.js）

示例：用new Error('woops')构造错误，断言时既可以直接比较错误对象，也可以改用谓词按e.message匹配：

var ex = new Error('woops'); var onError = Rx.ReactiveTest.onError; var scheduler = new Rx.TestScheduler(); var xs = scheduler.createHotObservable( onError(201, ex) ); var res = scheduler.startScheduler(function () { return xs.map(function (x) { return x; }); }); // 编写自定义断言 collectionAssert.assertEqual(res.messages, [ // 使用谓词 onError(201, function (e) { return e.message === 'woops'; }) ]);

2.3Rx.ReactiveTest.onNext(ticks, value)

在给定时间、携带给定值创建 OnNext 通知记录的工厂方法。

• 参数：
  1. ticks(Number)—— OnNext 通知发生的记录虚拟时间；
  2. value(Any | Function)—— 存入 OnNext 通知的值，或一个谓词函数。
• 返回值：(Recorded)—— OnNext 通知记录。
• 位置：rx.testing.js（源码见 reactivetest.js）

示例：断言收到值42，既可直接比较，也可用谓词x === 42：

var onNext = Rx.ReactiveTest.onNext; var scheduler = new Rx.TestScheduler(); var xs = scheduler.createHotObservable( onNext(201, 42) ); var res = scheduler.startScheduler(function () { return xs.map(function (x) { return x; }); }); // 编写自定义断言 collectionAssert.assertEqual(res.messages, [ // 使用谓词 onNext(201, function (x) { return x === 42; }) ]);

设计意图：谓词形式的断言在比较引用类型、时间戳或异步错误时尤其有用——断言只关心「这条记录是否满足某个条件」，而不必与预期对象严格同引用。这正是OnNextPredicate/OnErrorPredicate的equals实现（reactivetest.js）所保证的。

三、类字段详解：三个虚拟时间常量

ReactiveTest还提供了三个默认虚拟时间常量，它们是 TestScheduler.startScheduler 在未显式指定设置时的默认值：

源码中直接定义了这三个常量（reactivetest.js）：

created: 100, subscribed: 200, disposed: 1000,

而startScheduler会在设置缺失时回退到这些值（testscheduler.js）：

settings || (settings = {}); settings.created == null && (settings.created = ReactiveTest.created); settings.subscribed == null && (settings.subscribed = ReactiveTest.subscribed); settings.disposed == null && (settings.disposed = ReactiveTest.disposed);

3.1created—— 创建时间（默认 100）

默认用于在单元测试中创建可观察序列的虚拟时间。显式传入时通常以位置参数形式出现：

var scheduler = new Rx.TestScheduler(); var xs = scheduler.createHotObservable( Rx.ReactiveTest.onNext(201, 42), Rx.ReactiveTest.onNext(202, 56), Rx.ReactiveTest.onCompleted(203) ); var res = scheduler.startScheduler( function () { return xs.map(function (x) { return x; })}, Rx.ReactiveTest.created, // 创建时间 Rx.ReactiveTest.subscribed, // 订阅时间 Rx.ReactiveTest.disposed // 释放时间 );

3.2subscribed—— 订阅时间（默认 200）

默认用于在单元测试中订阅可观察序列的虚拟时间。它是「热序列 vs 冷序列」行为差异的关键分界点（下文 4.1 会详述）。

3.3disposed—— 释放时间（默认 1000）

默认用于在单元测试中释放订阅的虚拟时间。释放后，后续到达的通知不再被记录到res.messages。

这三个字段除位置参数形式外，还可以封装进一个settings对象传给startScheduler：

var res = scheduler.startScheduler( function () { return xs.map(function (x) { return x; })}, { created: Rx.ReactiveTest.created, subscribed: Rx.ReactiveTest.subscribed, disposed: Rx.ReactiveTest.disposed } );

两种传参方式等价——startScheduler(create, settings)的第二个参数既可传扁平位置参数（created, subscribed, disposed），也可传{ created, subscribed, disposed }对象，见 testscheduler.js。

四、实战：与 TestScheduler 组合的完整测试流程

ReactiveTest的工厂方法本身不执行任何调度，它们只负责「造记录」。真正驱动虚拟时间前进的是Rx.TestScheduler——一个继承自Rx.VirtualTimeScheduler的虚拟时间调度器，专用于测试基于 Reactive Extensions 构建的应用与库（testscheduler.js）。

4.1 经典完整用例（热序列）

下面的例子是官方文档给出的标准工作流：先创建热可观察序列（在150发1、210发2、220发3、230完成），再以默认时间轴（created=100、subscribed=200、disposed=1000）启动，对结果做map(x => x * x)后断言：

function createMessage(expected, actual) { return 'Expected: [' + expected.toString() + ']\r\nActual: [' + actual.toString() + ']'; } // 使用 QUnit 进行断言 var collectionAssert = { assertEqual: function (actual, expected) { var comparer = Rx.internals.isEqual, isOk = true; if (expected.length !== actual.length) { ok(false, 'Not equal length. Expected: ' + expected.length + ' Actual: ' + actual.length); return; } for(var i = 0, len = expected.length; i < len; i++) { isOk = comparer(expected[i], actual[i]); if (!isOk) { break; } } ok(isOk, createMessage(expected, actual)); } }; var onNext = Rx.ReactiveTest.onNext, onCompleted = Rx.ReactiveTest.onCompleted, subscribe = Rx.ReactiveTest.subscribe; var scheduler = new Rx.TestScheduler(); // 创建热可观察序列，订阅后立即开始发射 var xs = scheduler.createHotObservable( onNext(150, 1), onNext(210, 2), onNext(220, 3), onCompleted(230) ); // 注意：默认在 200 订阅，因此会错过 150 时刻的通知 var res = scheduler.startScheduler(function () { return xs.map(function (x) { return x * x }); }); // 实现集合断言 collectionAssert.assertEqual(res.messages, [ onNext(210, 4), onNext(220, 9), onCompleted(230) ]); // 检查订阅/退订时间 collectionAssert.assertEqual(xs.subscriptions, [ subscribe(200, 230) ]);

这个用例同时展示了三件事：

1. 通知断言：res.messages记录了订阅窗口内收到的全部带时间戳通知；
2. 订阅断言：xs.subscriptions记录了热序列被订阅（200）与退订（230，因流在 230 完成）的时间；
3. 时间过滤：150时刻的onNext(150, 1)因为发生在订阅时刻200之前而被正确忽略——这是热可观察序列（hotobservable.js）的固有行为：通知按绝对虚拟时间发射，订阅者只收到订阅之后的通知。

4.2 冷序列：偏移语义

与热序列相反，冷可观察序列（createColdObservable）的通知时刻是相对订阅时刻的偏移量。其实现（coldobservable.js）在订阅时才将相对时间换算为绝对时间，因此下面例子里onNext(150, 1)实际出现在绝对时间200 + 150 = 350之后……不过要注意的是，startScheduler默认在200订阅、1000释放，冷序列onNext(150,1)、onNext(200,2)、onNext(250,3)、onCompleted(300)会分别落在绝对时间350、400、450、500：

var onNext = Rx.ReactiveTest.onNext, onCompleted = Rx.ReactiveTest.onCompleted, subscribe = Rx.ReactiveTest.subscribe; var scheduler = new Rx.TestScheduler(); // 创建冷可观察序列，时刻相对订阅时间偏移 var xs = scheduler.createColdObservable( onNext(150, 1), onNext(200, 2), onNext(250, 3), onCompleted(300) ); // 注意：默认在 200 订阅 var res = scheduler.startScheduler(function () { return xs.filter(function (x) { return x % 2 === 0; }); }); // 实现集合断言 collectionAssert.assertEqual(res.messages, [ onNext(400, 2), onCompleted(500) ]); // 检查订阅/退订时间 collectionAssert.assertEqual(xs.subscriptions, [ subscribe(200, 500) ]);

由于filter只放行偶数，1与3被过滤，最终在绝对时间400（= 200 + 200）收到2，在500（= 200 + 300）完成，订阅记录为subscribe(200, 500)。冷序列因此特别适合测试「相对时序」与运算符的重排行为。

4.3 用createObserver手工控制时间轴

当需要更精细地控制「何时订阅、何时退订」时，可以用scheduler.createObserver()创建记录型观察者，再配合scheduleAbsolute手工编排。MockObserver（mockobserver.js）内部维护一个messages数组，每次收到通知都会用当前scheduler.clock打上时间戳：

var onNext = Rx.ReactiveTest.onNext; var scheduler = new Rx.TestScheduler(); var d = new Rx.SerialDisposable(); var xs = Rx.Observable.return(42, scheduler); var res = scheduler.createObserver(); scheduler.scheduleAbsolute(null, 100, function () { return d.setDisposable(xs.subscribe( function (x) { d.dispose(); res.onNext(x); }, res.onError.bind(res), res.onCompleted.bind(res) )); }); scheduler.start(); collectionAssert.assertEqual(res.messages, [ onNext(101, 42) ]);

Rx.Observable.return(42, scheduler)在虚拟时间100 + 1 = 101时刻发射42，测试断言精确到单个 tick。

4.4 测试 Promise：createResolvedPromise/createRejectedPromise

TestScheduler 还提供了两个 Promise 工厂（testscheduler.js），内部用MockPromise（mockpromise.js）实现，它把「resolve/reject」也翻译成带时间戳的虚拟通知：

• createResolvedPromise(ticks, value)→ 在ticks时刻依次发射onNext(ticks, value)与onCompleted(ticks)；
• createRejectedPromise(ticks, reason)→ 在ticks时刻发射onError(ticks, reason)。

解析成功示例：

var onNext = Rx.ReactiveTest.onNext, onError = Rx.ReactiveTest.onError, onCompleted = Rx.ReactiveTest.onCompleted; var scheduler = new Rx.TestScheduler(); // 创建已解析的 Promise var xs = scheduler.createResolvedPromise(201, 1); // 注意：默认在 200 订阅 var res = scheduler.startScheduler(function () { // 由于 issue #976，需要显式传入测试调度器 return Rx.Observable.fromPromise(xs, scheduler); }); // 实现集合断言 collectionAssert.assertEqual(res.messages, [ onNext(201, 1), onCompleted(201) ]);

拒绝示例：

var scheduler = new Rx.TestScheduler(); // 创建被拒绝的 Promise var error = new Error(); var xs = scheduler.createRejectedPromise(201, error); var res = scheduler.startScheduler(function () { // 由于 issue #976，需要显式传入测试调度器 return Rx.Observable.fromPromise(xs, scheduler); }); // 实现集合断言 collectionAssert.assertEqual(res.messages, [ onError(201, error) ]);

注意官方文档标注的坑：调用Rx.Observable.fromPromise时必须把scheduler作为第二参数显式传入（参见 issue #976），否则 Promise 的调度时机无法被虚拟时间控制。

4.5startScheduler的三个默认时刻

startScheduler(create, settings)的完整签名与默认值（testscheduler.js）：

其内部流程如下（源码级）：

1. createObserver()创建记录型观察者；
2. 在created时刻调度createFn()，得到source；
3. 在subscribed时刻调度source.subscribe(observer)；
4. 在disposed时刻调度subscription.dispose()；
5. 调用this.start()推进虚拟时钟直到队列清空；
6. 返回携带时间戳通知的 observer。

返回的 observer 的messages属性（Recorded[]）即最终断言对象。

五、配套断言对象：Recorded 与 Subscription 深入

理解ReactiveTest工厂返回的对象类型，才能写出准确的断言。

5.1Rx.Recorded—— 值 + 虚拟时间

Recorded记录「值在哪个虚拟时间产生」（recorded.js）：

var Recorded = Rx.Recorded = function (time, value, comparer) { this.time = time; this.value = value; this.comparer = comparer || defaultComparer; };

• 实例属性：time（产生值的虚拟时间）、value（产生的值）、comparer（比较器，默认使用Rx.internals.isEqual）；
• equals(other)：要求time严格相等，且comparer(this.value, other.value)为真（recorded.js）；
• toString()：返回value@time形式，如foo@201（recorded.js）。

var r1 = new Rx.Recorded(201, 'foo'); var r2 = new Rx.Recorded(201, 'bar'); var r3 = new Rx.Recorded(201, 'foo'); console.log(r1.equals(r2)); // => false console.log(r1.equals(r3)); // => true console.log(r1.toString()); // => foo@201

当ReactiveTest.onNext收到谓词函数时，value位置保存的其实是一个OnNextPredicate包装对象，其equals通过调用谓词完成比较（reactivetest.js）。

5.2Rx.Subscription—— 订阅/退订时间对

Subscription记录对可观察序列的订阅与退订（subscription.js）：

var Subscription = Rx.Subscription = function (start, end) { this.subscribe = start; this.unsubscribe = end || Number.MAX_VALUE; };

• 实例属性：subscribe（订阅虚拟时间）、unsubscribe（退订虚拟时间，缺省为Number.MAX_VALUE，即「无限/未退订」）；
• equals(other)：订阅与退订时间均严格相等才返回true（subscription.js）；
• toString()：未退订输出(201, Infinite)，已退订输出(201, 1000)（subscription.js）。

var s1 = new Rx.Subscription(201, 500); var s2 = new Rx.Subscription(201); var s3 = new Rx.Subscription(201, 500); console.log(s1.equals(s2)); // => false console.log(s1.equals(s3)); // => true console.log(s2.toString()); // => (201, Infinite)

热/冷可观察序列的subscriptions数组正是由Subscription实例构成，订阅时记录subscribe时间，退订时补全unsubscribe时间（见 hotobservable.js）。

六、在仓库中的实际印证

ReactiveTest并非「文档专属」的纸面 API，仓库自身的测试套件就大量使用它。以 tests/observable/where.js 为例，其头部直接从Rx.ReactiveTest解构出全部工厂方法与常量：

onNext = Rx.ReactiveTest.onNext, onError = Rx.ReactiveTest.onError, onCompleted = Rx.ReactiveTest.onCompleted, subscribe = Rx.ReactiveTest.subscribe, created = Rx.ReactiveTest.created, subscribed = Rx.ReactiveTest.subscribed, disposed = Rx.ReactiveTest.disposed;

随后几乎所有测试都遵循同一模式：用Rx.TestScheduler+createHotObservable/createColdObservable构造输入，用startScheduler运行被测运算符，最后用collectionAssert.assertEqual(res.messages, [...])断言通知序列、用xs.subscriptions断言订阅生命周期。你可以参考 tests/observable 目录下的 130 余个测试文件，为每个运算符找到对应的ReactiveTest使用范式。

加载与依赖说明

• ReactiveTest位于rx.testing.js（构建产物见 modules/rx-lite-testing/rx.lite.testing.js 与 modules/rx-core-testing/rx.core.testing.js）；
• 使用前提：需先加载rx.js/rx.compat.js/rx.lite.js等任一完整核心库；
• 包分发：npm 包rx、NuGet 包RxJS-All与RxJS-Testing（对应 nuget/RxJS-Testing）均包含该模块；
• 相关类型定义位于 ts/rx.testing.d.ts，可配合 TypeScript 使用。

七、小结

Rx.ReactiveTest是 RxJS v4 虚拟时间测试的「语法核心」：

• 三个工厂方法onNext/onError/onCompleted把「值/错误/完成 + 虚拟时间」封装成Recorded记录，且支持谓词形式的柔性断言；
• 三个时间常量created(100)/subscribed(200)/disposed(1000)定义了startScheduler的默认测试时间轴；
• subscribe(start, end)生成Subscription记录，用于断言订阅/退订生命周期；
• 与Rx.TestScheduler（热/冷序列工厂、createObserver、Promise 工厂、startScheduler）组合后，即可对任意运算符的时序行为做确定性、可重复的单元测试。

这套「记录-调度-断言」的测试方法论不仅支撑了本仓库 tests 目录下全部运算符测试，也构成了后来 RxJS 各版本虚拟时间测试（marble testing）的直接前身。掌握ReactiveTest，就等于掌握了 RxJS 时序测试的根基。

【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址: https://gitcode.com/gh_mirrors/rxj/RxJS