RxJS forkJoin单个请求失败时如何获取已成功Observable响应

动态生成Observables数组存储文件上传请求（POST类型），必须等数组内所有请求执行完成后，才能调用finalize()接口——该接口调用后会自动锁定数据库内的目标对象，禁止任何修改操作。
使用原生forkJoin实现时存在明确问题：

• 所有上传请求无报错时流程可正常运行
• 只要任意一个请求失败，forkJoin就会直接进入error回调，此前已经成功的请求响应会全部丢失，无法定位哪些请求已经执行成功
• 数组内所有请求都调用同一个API服务，仅传入的data和requestParams存在差异，需要确认适配该场景的RxJS实现方案

原有业务代码如下：

....

const observables = [];
// 循环动态填充observables数组
for(let file of this.obj.files){
  ...
    // 满足条件时生成上传请求Observable
    const apiUpload = this.service.upload(file);
    observables.push(apiUpload)
}

forkJoin<any[]>(observables).subscribe({
   next: (res) => {
     console.log('responses ', res);
     // 所有请求成功后准备调用finalize
     // this.finalize();
   },
   error: (error) => {
     console.log(error);
    // 错误场景下无法定位哪些上传请求已经成功
   },
 });

....

目前已知forkJoin更适配GET类拉取数据的接口场景，不适合这类提交存储数据的POST请求场景。曾考虑过通过全局Interceptor捕获所有接口响应的hack方案，但不确定合理性。

不需要使用全局拦截器这类高耦合的hack方案，也不需要更换RxJS组合操作符，核心思路是在单个请求流层面捕获错误，避免错误冒泡到外层forkJoin，把成功、失败的结果统一包装成结构化返回值，让forkJoin始终能拿到所有请求的执行结果。
具体实现逻辑：

• 给每个上传请求的Observable挂载pipe，用catchError捕获单请求错误，返回结构化的错误结果，不向外抛出错误
• 用map把请求成功的返回值也包装成统一结构，和错误返回格式对齐
• forkJoin执行完成后，遍历结果数组拆分成功、失败的请求列表，按照业务规则做对应处理：失败请求可提示用户、支持重试，确认所有请求都成功后再调用finalize()接口

改造后的代码示例：

import { catchError, map, of } from 'rxjs';

// ...

const observables = [];
for(let file of this.obj.files){
  // 原有前置判断逻辑保持不变
  const apiUpload = this.service.upload(file).pipe(
    // 捕获单个请求错误，包装为统一格式返回，不向外冒泡
    catchError(err => of({
      success: false,
      error: err,
      currentFile: file,
      data: null
    })),
    // 包装成功返回结果，和错误格式对齐
    map(res => ({
      success: true,
      error: null,
      currentFile: file,
      data: res
    }))
  );
  observables.push(apiUpload)
}

forkJoin(observables).subscribe(resList => {
  // 此处始终能拿到所有请求的执行结果，不会因为单个请求失败进入error回调
  const successList = resList.filter(item => item.success);
  const failList = resList.filter(item => !item.success);

  if (failList.length) {
    console.log('上传失败的文件列表：', failList);
    // 失败场景业务处理：提示用户、重试失败请求等
    return;
  }

  console.log('所有上传请求成功，返回结果：', successList.map(item => item.data));
  // 全量成功后再调用finalize接口
  this.finalize();
});

方案说明 #

• zip、combineLatest等其他组合操作符和原生forkJoin一样存在错误冒泡问题，只要单个流报错就会终止整个订阅，无法满足需求，不需要更换
• 全局Interceptor方案不推荐：拦截器全局生效会污染其他无关接口的处理逻辑，耦合度高，后续维护成本远高于单流捕获错误的方案
• 该实现可以精确拿到每个请求的执行状态、返回结果/错误信息，完全适配文件上传这类写操作场景，不需要修改现有服务的上传逻辑

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