RxJs 轮询、错误处理、retryWhen & retryCount 重置

Question

我正在尝试实施如下所示的轮询策略：

• 每 X 秒从服务器请求数据
• 如果请求失败，显示错误消息：“错误接收数据，重试尝试N°”... 每 X 秒重试一次，最多重试 Y 次
• 如果重试次数 > Y 则显示错误消息“接收数据时出错，请稍后重试”

目前的代码如下：

  polling$ = combineLatest([timer(0, 1000), this._filters]).pipe(
exhaustMap((_) =>
  this.noteDataService.getAll().pipe(
    retryWhen((error$) =>
      error$.pipe(
        map((err, i) => ({ err, i })),
        tap(({ i }) =>
          console.log(`Error receiving data, retry attempt ${i + 1}`)
        ),
        // retry every X seconds for a maximum of Y times
        take(5),
        delay(1000),
        tap({
          complete: () =>
            console.log("Error receiving data, please try again later"),
        })
      )
    )
  )
)

);

问题是在 X 次尝试后它不会停止并再次尝试 X 次......等等。我想我与 combineLatest 运算符中的第一个计时器有关，但我没有看到解决方案，我尝试插入“第一个”运算符，但在第一次失败后轮询不会重新尝试 X 次。

Answer 1

使用局部变量来保持重试次数是完全可以的。此外，您想在此处使用concat，以确保发射不会并行发生：

  startPolling(interval = 1000, retry = 10): Observable<Response> {
    let retryCount = 0;

    // pull the data every "interval" seconds
    return timer(0, interval).pipe(
      switchMap((_) => this.requestData()),
      tap((res) => {
        if (!res.ok) {
          // if response is failed, throw error, so "retryWhen" will be triggered
          throw res;
        } else {
          // if response is success, reset retryCount
          retryCount = 0;
        }
      }),
      retryWhen((res) => {
        return res.pipe(
          // use "concatMap" to ensure that data emitting will be happening in order
          concatMap((data) => {
            // if we exceed retry count, then just interrupt "retryWhen" loop
            if (retryCount >= retry) {
              console.log('Error receiving data, please try again later');
              return throwError(EMPTY);
            }

            // increment retry count and run next delay next retry
            retryCount += 1;
            console.log(`Error receiving data. Retry number ${retryCount}`);
            return of(data).pipe(delay(interval));
          }),
        );
      }),
    );
  }

Answer 2

这是我将如何执行此操作的示例。大多数情况下，您可以使用 RxJS 运算符为您管理状态，无需创建自己的局部变量。

const requestData = () => timer(2000).pipe(
  tap(_ => {throw "This is an error";})
);

// Request data from server every X seconds
const requestInterval = 1000;
interval(requestInterval).pipe(
  exhaustMap(_ => requestData().pipe(
    retryWhen(error$ => error$.pipe(
      map((err,i) => ({err,i})),
      tap(({i}) => console.log(`Error receiving data, retry attempt ${i + 1}`)),
      // retry every X seconds for a maximum of Y times
      take(5),
      delay(requestInterval),
      tap({
        complete: () => console.log("Error receiving data, please try again later")
      })
    ))
  ))
).subscribe();

---

更新：

如果您希望在收到太多连续错误后轮询间隔出错，而不是完成您的 retryWhen observable，您可以发出错误。

const requestData = () => timer(2000).pipe(
  tap(_ => {throw "This is an error";})
);

// Request data from server every X seconds
const requestInterval = 1000;
interval(requestInterval).pipe(
  exhaustMap(_ => requestData().pipe(
    retryWhen(error$ => error$.pipe(
      map((err,i) => ({err,i})),
      tap(({i}) => console.log(`Error receiving data, retry attempt ${i + 1}`)),
      // retry every X seconds for a maximum of Y times
      take(5),
      delay(requestInterval),
      concatWith(throwError(() => "Error receiving data, please try again later"))
    ))
  ))
).subscribe({error: e => console.log("Printing the error: ", e)});