API轮询和超时

Question

我有一个轮询的用例，其中：

1. 我想调用一个API，根据业务逻辑，立即返回数字（1-10）或错误（网络问题/ API异常等）（1.5-2秒）。
2. 如果API返回错误（网络问题/ API异常等），则我希望取消订阅轮询并显示错误。
3. 如果API返回成功，则我想检查返回值并取消订阅（如果返回值为5）或继续轮询。
4. 我想每5秒调用一次API。
5. 我想将轮询的最大时间（超时/阈值）保持为3分钟。如果在这3分钟内没有得到所需的响应（数字5），则轮询应该报错。

这是我目前的实现方式：

this.trackSpoke$ = interval(5000)
                .pipe(
                    timeout(250000),
                    startWith(0),
                    switchMap(() =>
                        this.sharedService.pollForAPropertyValue(
                            "newuser"
                        )
                    )
                )
                .subscribe(
                    (data: SpokeProperty) => {
                        this.CheckSpokeStatus(data);
                    },
                    error => {
                        this.trackSpoke$.unsubscribe();
                        this.createSpokeState.CdhResponseTimedOut();
                    }
                );


private CheckSpokeStatus(data) {
        if (data.PropertyValue === "5") {
            this.trackSpoke$.unsubscribe();
            //display success   
        } else {
              //keep the polling going
        }
    }

然而，上述实现没有超时。

需要做什么才能使其超时，以便我能够实现所有提到的用例？

- chirag_lad

---

interval(5000) 每隔5秒发出一次信号，因此紧接着的超时永远不会抛出错误，因为它每5秒接收一个值。 - martin

明白了。现在我该如何设置超时？ - chirag_lad

每当我看到轮询和API时...我不得不问一下，你考虑过使用消息队列吗？ - Kyle Burkett

Answer 1

首先，使用interval进行API轮询是一种反模式，因为interval不会“等待”您的http请求完成 - 如果请求需要超过5秒才能完成，则可能触发多个请求（）。

我更喜欢使用defer与repeatWhen和delay（见下面的代码）。

由于interval每5秒钟就会滴答一次，因此timeout不会触发，从而防止了超时的发生。 defer / repeatWhen组合也应该解决这个问题。

不要手动取消订阅，可以考虑使用takeWhile来自动取消订阅Observable。

另外，在错误处理程序中使用this.trackSpoke$.unsubscribe();不必要，因为在出现错误的情况下，Observable会自动取消订阅。

this.trackSpoke$ = defer(() => this.sharedService.pollForAPropertyValue("newuser"))
    .pipe(
        timeout(250000),
        repeatWhen(notifications => notifications.delay(5000)),
        takeWhile(data => this.CheckSpokeStatus(data)),
    )
    .subscribe(
        error => {
            this.createSpokeState.CdhResponseTimedOut();
        }
    );


private CheckSpokeStatus(data) {
    return data.PropertyValue !== "5";
}

- m1ch4ls

Answer 2

我有一个非常相似的用例。

使用 rxjs 7.8.1 编写。

from(fakeDelayedRequest()).pipe(
  map((response) => {
    if (isValidResponse(response)) {
      return response;
    } else {
      throw `Not a valid response`;
    }
  }),
  retry({
    // getRetryDelayTimeOrThrow decides if we retry or not, depending on the error
    //  - returns an observable that triggers when the next request shall be sent
    //  - or throws the error which leads to exiting the retry loop
    delay: getRetryDelayTimeOrThrow,
  }),
  timeout({
    each: maxTimeout,
    with: () => throwError(() => 'TIMEOUT'),
  })
);

function getRetryDelayTimeOrThrow(e: any, retryCount: number) {
  console.error(e);

  if (repeatAfterError(e)) {
    return timer(getPollInterval(retryCount));
  } else {
    throw e;
  }
}

你可以在这个 StackBlitz 上找到完整的工作代码。

- andymel