Rxjs timeout() 操作符不在管道中工作

问题描述

嗨，我已经使用以下代码达到了一个状态，除了超时之外，其他一切都在工作：

public monitorTask$(id: string): Observable<TaskResponse> {
    return timer(0,4000).pipe(
        switchMap(() => this.fetchTaskStatus(taskId)),timeout(120000),takeWhile(res => this.isInProgress(res),true)
    );
}

private postTask(id: string) {
    this.monitorTask$(id).subscribe(
        state => {
            if (state.status === "SUCCESS") {
                this.onSuccess();
            }
            if (state.status === "FAILURE) {
                this.onFailure();
            }
        },(err) => {
            this.showError();
        }
    );
}

也试过这个：

public monitorTask$(id: string): Observable<TaskResponse> {
    return interval(4000).pipe(
        flatMap(() => this.fetchTaskStatus(id)),takeWhile(res => this.isInProgress(res.status),true),timeout(120000)
    );
}

我期待超时出错并进入 postTask() 中的 (err) 块，但它从未达到超时。我一直在玩不同的变体，但似乎不太正确..这是我拥有的最干净的一个，所以如果有人看到我遗漏的东西，我会非常感激！

解决方法

这里有很多事情需要考虑。

timer 发射间隔 (4s) 小于 timeout (120s)。因此 timeout 永远不会被触发，因为 timer 每 4 秒发出一次。
将 timeout 管道化到 RxJS 内置可观察对象 timer 在逻辑上没有多大意义。我相信您想通过管道将 timeout 传递给 this.fetchTaskStatus() 函数。
在这种情况下，还有一个问题。这里使用的映射运算符是 switchMap，它会在外部 observable (this.fetchTaskStatus()) 发出时取消现有的内部 observable (timer)。您很可能正在寻找 flatMap 运算符。但要注意 timer 的每次发射，this.fetchTaskStatus() 将单独触发。

public monitorTask$(id: string): Observable<TaskResponse> {
    return timer(0,4000).pipe(
        flatMap(() => 
            this.fetchTaskStatus(id).pipe(
                timeout(120000)
            )
        ),takeWhile(res => this.isInProgress(res.status),true),);
}

angular angular rxjs rxjs timeout timeout