如何在rxjs中测试串联流

如何解决如何在rxjs中测试串联流

给定的类具有一个方法，该方法返回缓存的流，但是该流可以由另一个私有热流触发，这会使缓存的流发出新值。

班级

export class SomeClass {
  private cache: Observable<number>;
  private trigger$ = new Subject();
  private multiply = 1;

  constructor(private num: number) {}

  getNumber(): Observable<number> {
    return (
      this.cache ||
      (this.cache = concat(of(void 0),this.trigger$).pipe(
        switchMap(() => of(this.num * this.multiply++)),shareReplay(1)
      ))
    );
  }

  trigger(): void {
    this.trigger$.next();
  }
}

示例：https://stackblitz.com/edit/rxjs-gpyc46?file=index.ts

测试方式是什么？ 尝试失败

it("trigger updates",() => {
  testScheduler.run(({ expectObservable }) => {
    const num$ = someClass.getNumber();
    expectObservable(num$).toBe("a",{ a: 3 });
    someClass.trigger();
    expectObservable(num$).toBe("a",{ a: 6 });
    someClass.trigger();
    expectObservable(num$).toBe("a",{ a: 9 });
  });
});

示例：https://stackblitz.com/edit/rxjs-test-tricky-flow?file=src%2Fsome-class.spec.ts

UPD：所以这里的问题似乎无法模拟trigger$属性。

看起来像这样

it("trigger updates",() => {
  testScheduler.run(({ hot,expectObservable }) => {
    spyOnProperty(someClass,'trigger$','get').and.returnValue(hot('^--b--c'));
    const num$ = someClass.getNumber();
    expectObservable(num$).toBe("a--b--c",{ a: 3,b: 6,c: 9 });
  });
});

但是必须以这种方式更改trigger$属性

  get trigger$() {
    return new Subject();
  }

示例：https://stackblitz.com/edit/rxjs-test-tricky-flow-x2arxf?file=src%2Fsome-class.ts