RxJS测试expectObservable().toBe()仅接收Observable流最后一个值问题

问题背景 #

• 待测试服务内部使用 BehaviorSubject 存储状态，对外暴露的 Observable 管道包含 distinctUntilChanged() 操作符
• 运行测试用例时，用于和预期结果比对的实际输出流仅返回最后一个值，不符合预期，需要明确修复该问题需要掌握的相关知识点

问题复现代码 #

import { BehaviorSubject } from 'rxjs';
import { distinctUntilChanged } from 'rxjs/operators';
import { TestScheduler } from 'rxjs/testing';

describe('My exposed stream', () => {
  let testScheduler;

  beforeEach(() => {
    testScheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });
  });

  it('does not propagate if the current value equals the last one', () => {
    testScheduler.run(({ expectObservable }) => {
      const internalStream$ = new BehaviorSubject<string>(null);
      const exposedStream$ = internalStream$.pipe(distinctUntilChanged());

      expectObservable(exposedStream$).toBe('012', [null, 'foo', 'bar']);

      internalStream$.next('foo');
      internalStream$.next('foo');
      internalStream$.next('bar');
    });
  });
});

运行报错信息 #

Expected $.length = 1 to equal 3.
Expected $[0].notification.value = 'bar' to equal null.
Expected $[1] = undefined to equal Object({ frame: 1, notification: Notification({ kind: 'N', value: 'foo', error: undefined, hasValue: true }) }).
Expected $[2] = undefined to equal Object({ frame: 2, notification: Notification({ kind: 'N', value: 'bar', error: undefined, hasValue: true }) }).

修复所需核心知识点 #

• RxJS TestScheduler 运行机制：testScheduler.run() 会先同步执行完回调内的所有逻辑注册，待回调执行完成后才会启动虚拟时间、触发所有订阅动作；只有被虚拟调度器接管的事件才会按照 marble 语法定义的帧时序触发，同步执行的代码不会被分配虚拟帧时间。
• BehaviorSubject 核心特性：BehaviorSubject 属于热 Observable，仅会在订阅发生时向观察者推送当前存储的最新值，订阅前推送过的历史值不会被回放；订阅建立后，每次调用next()会立刻向所有已订阅的观察者同步推送值。
• Marble 测试写法规范：所有需要参与 marble 时序比对的流事件，必须通过 TestScheduler 提供的辅助方法（如cold、hot、schedule）创建或按虚拟帧调度，不能在回调中同步调用 Subject 的next()直接发送值——这类同步调用要么发生在订阅建立前（仅修改Subject存储的当前值，不会被流捕获为时序事件），要么发生在虚拟时间帧外，无法匹配marble定义的帧预期。
• distinctUntilChanged 执行逻辑：该操作符接收上游推送值时会立刻同步做等值比对，过滤连续重复值，不存在异步延迟逻辑。

问题根因：同步执行的三次next调用全部发生在expectObservable建立订阅之前，等虚拟时间启动、订阅真正生效时，BehaviorSubject存储的最新值已经是bar，因此订阅后仅能接收到这一个值。修复时可将源替换为对应时序的hotObservable模拟事件发送，或通过testScheduler.schedule按marble定义的帧时间安排每次next调用，确保事件触发时序和预期帧对齐。

内容的提问来源于stack exchange，提问作者stofl