为使用 retryWhen 运算符的 RxJS 编写测试（了解与 retry 运算符的区别）

Question

我正在尝试为以下使用 retryWhen 运算符的函数编写测试：

// some API I'm using and mocking out in test
import { geoApi } from "api/observable";

export default function retryEpic(actions$) {
  return actions$.pipe(
    filter(action => action === 'A'),
    switchMap(action => {
      return of(action).pipe(
        mergeMap(() => geoApi.ipLocation$()),
        map(data => ({ data })),
        retryWhen(errors => {
          return errors.pipe(take(2));
        }),
      );
    }),
  );
}

该代码应该执行对某个远程 API geoApi.ipLocation$() 的请求。如果出现错误，它会在放弃之前重试 2 次。

我编写了以下使用 Jest 和 RxJS TestScheduler 的测试代码：

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

const mockApi = jest.fn();
jest.mock('api/observable', () => {
  return {
    geoApi: {
      ipLocation$: (...args) => mockApi(...args),
    },
  };
});

describe('retryEpic()', () => {
  it('retries fetching 2 times before succeeding', () => {
    basicTestScheduler().run(({ hot, cold, expectObservable, expectSubscriptions }) => {
      const actions$ = hot('-A');

      // The first two requests fail, third one succeeds
      const stream1 = cold('-#', {}, new Error('Network fail'));
      const stream2 = cold('-#', {}, new Error('Network fail'));
      const stream3 = cold('-r', { r: 123 });

      mockApi.mockImplementationOnce(() => stream1);
      mockApi.mockImplementationOnce(() => stream2);
      mockApi.mockImplementationOnce(() => stream3);

      expectObservable(retryEpic(actions$)).toBe('----S', {
        S: { data: 123 },
      });

      expectSubscriptions(stream1.subscriptions).toBe('-^!');
      expectSubscriptions(stream2.subscriptions).toBe('--^!');
      expectSubscriptions(stream3.subscriptions).toBe('---^');
    });
  });
});

此测试失败。

但是，当我将retryWhen(...) 替换为简单的retry(2) 时，测试成功。

看起来我不太明白如何用retryWhen 实现retry。我怀疑这个take(2) 正在关闭流并阻止一切继续进行。但是我不太明白。

我其实想在retryWhen() 里面写一些额外的逻辑，但首先我需要了解如何正确实现retry() 和retryWhen()。或者这实际上是不可能的？

其他资源

我对 retryWhen + take 的实现是基于这个 SO 答案：

• How to create an RXjs RetryWhen with delay and limit on tries

官方文档：

• retryWhen

Answer 1

您可以将retryWhen 用于这两个目的，一个是包含您的逻辑，第二个是您想给它的重试次数（无需使用retry 运算符）：

// some API I'm using and mocking out in test
import { geoApi } from "api/observable";

export default function retryEpic(actions$) {
  return actions$.pipe(
    filter(action => action === 'A'),
    switchMap(action => {
      return of(action).pipe(
        mergeMap(() => geoApi.ipLocation$()),
        map(data => ({ data })),
        retryWhen(errors =>
          errors.pipe(
            mergeMap((error, i) => {
              if (i === 2) {
                throw Error();
              }
              // return your condition code
            })
          )
        )
      )
    }),
  );
}

这是一个简单的DEMO。

至于理解这个逻辑：

retryWhen 和 retry 运算符，根据您引用的官方文档：

重新订阅源 Observable（如果没有错误或完整执行）

这就是为什么您不能将retry 和retryWhen 连接在一起的原因。你可以说这些运营商是断链者……