Ping API 与 RXJS 连续，超时取消？答案

【问题标题】：Ping API continuous with RXJS, timeout cancels?Ping API 与 RXJS 连续，超时取消？
【发布时间】：2022-01-14 22:29:30
【问题描述】：

我尝试使用 rxjs 连续 ping 一个 URL/API。

我的第一次尝试是：

timer(3000, 2000)
   .pipe(mergeMap(() => this._http.get(environment.pingUrl)))
   .subscribe(res => console.log(res), err => console.log(err))

但是当 URL 无法访问时，我永远不会得到响应。所以我决定设置一个超时：

timer(3000, 2000)
   .pipe(mergeMap(() => this._http.get(environment.pingUrl).pipe(timeout(5000)))
   .subscribe(res => console.log(res), err => console.log(err))

现在，当 api 无法访问时，我得到了第一个超时/响应，但似乎超时触发了对计时器的取消订阅，并且没有进一步的响应。我不知道要防止这种退订。

【问题讨论】：

标签： angular rxjs observable

【解决方案1】：

关于错误

在 RxJS 中，Observables 具有三种类型的发射。

next：steam中的一个值，可以有0到任意多个
error：终端发射。这个 observable 的实例有一个错误并被关闭。它永远不会再次发射。
complete：终端发射。这个 observable 的实例已经完成并关闭。它永远不会再发射。

你会注意到一个 observable 的实例永远不能发射超过一个 error 或 complete 发射，但任何 observable 都可以运行（或重新启动/重试）任意次数。每次您订阅或某个运营商为您订阅时，您都会创建一个新的 observable 实例。

创建一个发出错误但之后仍会继续的流

由于 observables 永远不会在错误后发出，因此在继续时发出错误的唯一方法是捕获 error 发射并将其转换为 next 发射。

一个例子：

在这个例子中，每当源 observable 出现错误时，我都会告诉 catchError 重新订阅源。

我在具有{value: } 属性的对象内定期发射
我在具有{error: } 属性的对象内发出错误。

我只是通过将它们打印到控制台来处理这些捕获的错误，但您可能想要做一些比这更好的事情:)


timer(3000, 2000).pipe(
  mergeMap(_ => this._http.get(environment.pingUrl)),
  map(value => ({value})),
  timeout({each: 5000}),
  catchError((error, src) => {
    // Only catch/handle TimeoutError
    if(error instanceof TimeoutError){
      return src.pipe(startWith({error}));
    } else {
      return throwError(() => error);
    }
  })
).subscribe({
  next: emitted => {
    if("error" in emitted){
      // Found an error! In this case, we only handled
      // TimeoutErrors, so that'll appear here as a next
      // emission. The observable is still running.
      console.log("Caught an error: ", emitted.error);
    }else{
      // got a result!
      console.log(emitted.value);
    }
  },
  // All other errors appear here and are error emissions, 
  // so they're terminal. This observable is closed
  error: err => console.log("Uncaught error: ", err),
  // The complete emission is also terminal. This 
  // observable is now closed.
  complete: () => console.log("Complete")
});

【讨论】：

【解决方案2】：

据我了解，您希望持续 ping API 并在无法访问此 API 时继续运行。

你的问题是，当api失败时，observable已经完成，不会再次触发。

您需要使用retry 或retryWhen

然后做

timer(2000, 3000)
  .pipe(
    mergeMap(() => this._http.get(environment.pingUrl)),
    timeout(5000),
    retryWhen(() => timer(3000))
  )
  .subscribe(
    (res) => console.log(res),
    (err) => console.log(err)
  );

例如：https://stackblitz.com/edit/rxjs-playground-test-mgppcj

请务必在某个时候取消订阅此 observable。

【讨论】：