Rxjs timeout（）运算符无法在管道中工作

Question

您好，我已经达到了以下代码的状态，其中除超时外一切正常：

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) 块，但它永远不会达到超时。我一直在尝试不同的变体，但似乎没有把它弄好。这是我拥有的最干净的变体，所以如果有人看到我遗漏的东西，我将非常感激！

Answer 1

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

timer发射间隔（4s）小于timeout（120s）。所以 timeout 永远不会被触发，因为 timer 每 4 秒发出一次。
将 timeout 管道化到 RxJS 内置可观察对象 timer 在逻辑上毫无意义。我相信您想将 timeout 传递给 this.fetchTaskStatus() 函数。
那样的话还有一个问题。这里使用的映射运算符是 switchMap，当外部可观察对象（timer）发出时，它将取消现有的内部可观察对象（this.fetchTaskStatus()）。您很可能正在寻找 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),
    );
}

Rxjs timeout（）运算符无法在管道中工作

Rxjs timeout() operator not working in pipe

timeout

rxjs

angular