如何使用 TestScheduler 测试对 RxJS Observable 的重新订阅？答案

【问题标题】：How to test resubscriptions to RxJS Observable with TestScheduler?如何使用 TestScheduler 测试对 RxJS Observable 的重新订阅？
【发布时间】：2020-01-23 09:34:27
【问题描述】：

我有一个 observable 可能会引发错误。当它抛出时，我想重新订阅那个 observable 并重试。例如retry() 运算符。

为了测试这个重试逻辑，我需要创建一个 test-observable，它会在前 2 次订阅时抛出错误，并且只有在第 3 次才会产生一个值。

我尝试了以下方法：

import { Observable } from 'rxjs';
import { TestScheduler } from 'rxjs/testing';
import { retry } from 'rxjs/operators';

// Setup for TestScheduler
function basicTestScheduler() {
  return new TestScheduler((actual, expected) => {
    expect(actual).toEqual(expected);
  });
}

// The function we're going to test
function retryMultipleTimes(observable$) {
  return observable$.pipe(retry(2));
}

describe('retryMultipleTimes()', () => {
  it('retries twice when observable throws an error', () => {
    basicTestScheduler().run(({ hot, cold, expectObservable }) => {
      const observable$ = hot('--#--#--Y');
      const expected = '       --------Y'; // This is what I want to get
      const unexpected = '     --#      '; // This is what I get instead

      expectObservable(retryMultipleTimes(observable$)).toBe(expected);
    });
  });
});

似乎使用hot() observable 它总是重新订阅产生错误的同一帧，导致立即再次抛出。

我也尝试了cold() observable，在这种情况下我得到------# - 也就是说，在每次重试时，observable 从头开始重新开始，导致--#，--#，--# - 从不到达--Y。

似乎没有办法用 RxJS TestScheduler 来做这样的事情。或者也许有？

如果 hot() 和 cold() observable-creators 不能胜任这项任务，也许我可以创建自己的......但是怎么做？

我还尝试在重试之间添加一点延迟，因此我不会通过使用 retryWhen 实现重试逻辑来立即重新订阅当前帧：

function retryMultipleTimes(observable$) {
  return observable$.pipe(
    retryWhen(errors => errors.pipe(
      delayWhen(() => timer(2)),   // wait 2 frames before each retry
      take(2),                     // do maximum of 2 retries
      concat(throwError('error')), // finish with error when no success after 2 retries
    )),
  );
}

但这也不起作用。看起来重新订阅仍然发生在与以前相同的帧上。

我怎样才能通过这个测试？

【问题讨论】：

有两件事。首先，您是否期望 Observable 发出多个永远不会发生的 error 通知。第二件事是 hot 即使没有人订阅也会发出。所以我认为你不能得到你所期望的。也许看看 retry() 在 RxJS 中是如何测试的，这可能会给你一个提示 github.com/ReactiveX/rxjs/blob/master/spec/operators/…
感谢您的提示。从那里并没有真正找到我想要的东西。但是，阅读retry() 的测试给了我其他想法，我能够找到解决方案。

标签： unit-testing rxjs

【解决方案1】：

为此想出了一个解决方案。我可以使用iif() 创建一个可观察对象，在订阅时在两个可观察对象之间进行选择：

describe('retryMultipleTimes()', () => {
  it('retries twice when observable throws an error', () => {
    basicTestScheduler().run(({ cold, expectObservable }) => {
      let count = 0;
      const observable$ = iif(
        () => ++count <= 2,
        cold('--#'),
        cold('--Y'),
      );

      //                                                      --#
      //                                                        --#
      //                                                          --Y
      expectObservable(retryMultipleTimes(observable$)).toBe('------Y');
    });
  });
});

【讨论】：