RXJS的timeout操作符在轮询场景下未按预期生效如何解决

• 你当前使用的timeout操作符的作用是限制流中两次数据发射的最大间隔，而非限制整个轮询流的总运行时长。你设置了每5秒触发一次轮询请求，流每隔5秒就会产生新值，间隔远小于2分钟，因此timeout永远不会触发。

我们可以用timer创建一个2分钟后触发的信号，加入takeUntil的触发条件里，同时添加错误回调处理超时和请求异常场景：

// 引入需要的操作符
import { interval, timer, takeUntil, startWith, switchMap } from 'rxjs';

// 轮询逻辑
const maxPollingTime = 2 * 60 * 1000;
// 创建2分钟后触发的超时信号
const pollingTimeout$ = timer(maxPollingTime);

interval(5000)
  .pipe(
    startWith(0),
    switchMap(() => this.someService.getQueueJob(jobId)),
    // 同时监听手动停止信号和超时信号
    takeUntil(this.stopPolling),
    takeUntil(pollingTimeout$)
  )
  .subscribe({
    next: (res) => {
      this.jobStatus = res.attributes.status;
      if (this.jobStatus === 'Completed') {
        this.fileUploadStatus = 'success';
        this.stopPolling.next();
      } else if (this.jobStatus === 'Failed') {
        this.fileUploadStatus = 'error';
        this.stopPolling.next();
      }
    },
    complete: () => {
      // 流完成时判断是不是超时触发：此时如果任务状态还不是终态，就是超时
      if (this.jobStatus !== 'Completed' && this.jobStatus !== 'Failed') {
        this.fileUploadStatus = 'error';
        // 这里添加用户提示逻辑，比如toast弹窗提示
        console.error('轮询超时，任务未在规定时间内完成');
      }
    },
    error: (err) => {
      // 处理请求本身的报错
      this.fileUploadStatus = 'error';
      console.error('轮询请求异常', err);
      this.stopPolling.next();
    }
  });

如果需要在超时的时候直接抛出错误而不是静默结束，RxJS 7+ 版本可以直接使用timeout的first参数限制流从订阅到完成的总时长：

// 把上面代码里的 takeUntil(pollingTimeout$) 替换为如下配置即可
timeout({
  first: 2 * 60000,
  meta: '轮询超过2分钟未得到终态结果'
})

超时错误会直接进入subscribe的error回调，你可以直接在error回调中处理用户提示逻辑。

内容的提问来源于stack exchange，提问作者Gini