RxJS中如何在catchError操作符内获取触发Observable执行的源值

结论 #

需求可以实现。

实现思路 #

核心逻辑是在每一步操作时将原始源值和后续操作的流绑定，通过自定义错误对象或者携带源值的返回结果，将原始值一路传递到catchError捕获阶段。

代码实现示例 #

import { Subject, mergeMap, catchError, EMPTY } from 'rxjs';

// 自定义错误类型，包含原始源任务和错误详情
interface TaskProcessError {
  error: unknown;
  sourceTask: Task;
}

// 模拟Task类型，可根据实际业务调整
interface Task {
  id: string;
  // 其他业务字段
}

this.queue = new Subject<Task>();
this.queue.pipe(
  mergeMap(source => 
    // 第一个业务操作流，内部捕获错误时携带原始源
    firstTaskOperate(source).pipe(
      catchError(err => {
        throw {
          error: err,
          sourceTask: source
        } as TaskProcessError
      })
    )
  ),
  mergeMap(prevResult => 
    // 第二个业务操作流，如果需要保留原始源，上一步可以返回 { result: 操作结果, sourceTask: 原始源 } 的结构
    secondTaskOperate(prevResult.result).pipe(
      catchError(err => {
        throw {
          error: err,
          sourceTask: prevResult.sourceTask
        } as TaskProcessError
      })
    )
  ),
  catchError((err: TaskProcessError) => {
    // 此处即可获取到触发错误的原始源值
    console.log('触发错误的原始任务：', err.sourceTask);
    console.log('错误详情：', err.error);
    // 返回EMPTY避免队列整体中断，后续新增任务可正常执行
    return EMPTY;
  })
);

function addToQueue(source: Task): void {
  this.queue.next(source);
}

可选实现方案 #

• 若正常业务流程也需要用到原始源值，可省略内部catchError，每一步mergeMap都返回携带原始源的结构：mergeMap(source => operate(source).pipe(map(res => ({ res, sourceTask: source }))))，后续所有操作都可以直接读取原始源，出错时也能在顶层catchError中获取到对应值。
• 若仅需要错误场景获取原始源，直接使用示例中的自定义错误对象方式即可，不需要额外修改正常业务返回结构。

注意事项 #

如果需要保证队列在单个任务出错后仍能正常处理后续新增任务，catchError中必须返回无错误的Observable（比如EMPTY、默认值包装的of流），否则整个队列流会终止，后续调用addToQueue新增的任务不会被执行。

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