Rxjs重试延迟功能

Question

我想使用retry与delay功能,我预计1000毫秒的延迟后功能会调用,但它不列入,什么都可以在这里的错误？看看控制台输出,它同时是16:22:48.

我期待那里16:22:48,16:22:59 ......

canCreate: boolean;
getSomeFunction(): Observable<boolean> {
        return new Observable<boolean>(
            observer => {
                const canCreate = null; // this is just null for now, will some value later
                if (canCreate == null) {
                    observer.error('error');
                } else {
                    observer.next(true);
                }
                observer.complete();
            }
        )
    }


this.getSomeFunction()
      .do((value) => {
        this.cCreate = value;
      }, (error) => {
         console.log(error + new Date().toTimeString());
      })
      .delay(1000)
      .retry(10)
      .subscribe(
        value => this.cCreate = value,
        error => {
          this.cCreate = false;
        },
        () => {}
      );
  }

和控制台的结果是:

Answer 1

delay()用于在observable发出的事件之间引入延迟.但是观察者永远不会发出任何事件.它只是立即出错.

您正在寻找的是retryWhen(),它允许在重试多长时间后决定:

RxJS 5:

  .retryWhen(errors => errors.delay(1000).take(10))

RxJS 6:

import { retryWhen, delay, take } from 'rxjs/operators'
someFunction().pipe(
  // ...
  retryWhen(errors => errors.pipe(delay(1000), take(10)))
)

这将在10次尝试后完成整个观察.如果你想在10次尝试后输出整个observable,则retryWhen回调返回的observable必须抛出:

RxJS 5:

  .retryWhen(errors => errors.delay(1000).take(10).concat(Observable.throw()))

RxJS 6:

import { retryWhen, delay, take, concatMap, throwError } from 'rxjs/operators'
someFunction().pipe(
  // ...
  retryWhen(errors => errors.pipe(delay(1000), take(10), concatMap(throwError)))
)

Answer 2

由于当时接受的答案建议使用retryWhen()是可以的，现在，正如我们在提供的retryWhen()链接上看到的那样，该运算符已被弃用，并将从 RxJS 版本 9 或 10 中删除。使用 retryWhen() 的解决方案也已被弃用。

现在，RxJS 开发团队和文档建议的解决方案是使用retry()运算符，它也是一个错误处理运算符，并带有 retryConfig 参数。

现在，要使用更少的代码实现相同的逻辑，您可以使用以下代码：

// A custom method to check should retry a request or not
shouldRetry(error: HttpErrorResponse) {
    // Example for catching specific error code as well
    if (error.status === 503) {
      return timer(1000); // Adding a timer from RxJS to return observable to delay param.
    }

    throw error;
  }

// Retry operator somewhere in your code
retry({ count: 2, delay: this.shouldRetry })

正如您从上面的示例中看到的，我们使用了更少的代码和更少的 RxJS 运算符来实现一些逻辑，而我们需要使用 retryWhen() 运算符来实现更多逻辑。这也是团队弃用此运算符的原因，因为使用此配置的 retry() 可以实现相同的逻辑。

注意：您必须根据您的需要采用我提供的代码，但这是如何使用 retry() 运算符而不是 retryWhen() 的示例。

Answer 3

添加到@JB Nizet的答案.如果您在rxjs 5+中使用lettable运算符编写它,请将其结构化为

retryWhen(errors => errors.pipe(delay(1000), take(5)))

Answer 4

重试5次，延迟500ms：

\n

RxJS 6.x

\n

\n
• 使用索引来计数
\n
• retryWhen将的误差 Observable转换为timerorthrowError
\n

\n

retryWhen(concatMap((err, index) => index < 5 ? timer(500) : throwError(err)))\n

\n

RxJS 6.3 \xe2\xac\x86

\n

\n
• 使用索引来计数
\n
• 在retryWhenObservable 的错误中，使用delayWhen等待timer或throwError
\n

\n

retryWhen(delayWhen((err, index) => index < 5 ? timer(500) : throwError(err)))\n

\n

或者你可以这样做：

\n

const delayFn = (delay = 0, count = Infinity) => {\n    return (err, index) => index < count ? timer(delay) : throwError(err);\n};\n

\n

retryWhen(concatMap(delayFn(500, 5))) // RxJS 6.x\n

\n

retryWhen(delayWhen(delayFn(500, 5))) // RxJS 6.3 \xe2\xac\x86\n

\n\n

RxJS 7.3 \xe2\xac\x86

\n

\n
• 只是retry
\n

\n

retry({ count: 5, delay: 500 })\n

\n

Answer 5

所有这一切都是 RxJS 6+

\n\n

长话短说

\n

您可以使用此包中经过全面测试的运算符，或向下滚动以查看源代码:)

\n

npm i rxjs-boost\n

\n

import { retryWithDelay } from \'rxjs-boost/operators\';\n\nobs$.pipe(\n  // will retry 4 times with a 1s delay before each try:\n  retryWithDelay(1000, 4)\n);\n

\n\n

标准

\n

由于大多数（或可能没有）其他答案不符合我的所有标准，我将在下面列出我的解决方案。目标：

\n

\n
• 如果没有抛出错误，则定期发出并完成。\xe2\x9c\x85
\n
• x如果抛出错误，则重试次数。\xe2\x9c\x85
\n
• 每次重试之前都有延迟y。\xe2\x9c\x85
\n
• 返回最后发出的错误。（很多其他答案都在努力解决这个问题。） \xe2\x9c\x85
\n
• 使用 \xe2\x80\x93 正确输入strict: true，但这很难搞乱。\xe2\x9c\x85
\n

\n\n

解决方案

\n

与其他所有答案一样，我们将使用retryWhen运算符来捕获错误。要跟踪重复次数可以使用扫描运算符。为了限制重复次数，我们只需在地图内抛出一个错误运算符内抛出一个错误。

\n

\n

原始来源使用throwIf，但在这种情况下，我们可以简单地使用rxjs-boost中的retryWithDelay。

\n

\n

最后我们将使用延迟运算符来添加不同执行之间的延迟：

\n

import { MonoTypeOperatorFunction } from \'rxjs\';\nimport { delay as delayOperator, map, retryWhen, scan } from \'rxjs/operators\';\n\nexport function retryWithDelay<T>(\n  delay: number,\n  count = 1\n): MonoTypeOperatorFunction<T> {\n  return (input) =>\n    input.pipe(\n      retryWhen((errors) =>\n        errors.pipe(\n          scan((acc, error) => ({ count: acc.count + 1, error }), {\n            count: 0,\n            error: undefined as any,\n          }),\n          map((current) => {\n            if (current.count > count) {\n              throw current.error;\n            }\n            return current;\n          }),\n          delayOperator(delay)\n        )\n      )\n    );\n}\n

\n\n

来源

\n

\n
• rxjs 升压
\n
• 原始来源
\n
• 带有测试\xe2\x80\x93 的规范文件可能会派上用场
\n

\n