Angular项目中RxJS pipe如何将首操作符输出传给两个自定义操作符

解决方案 #

核心思路

你需要先把接口返回的源流转为可组播的热流，保证上游接口只请求一次，结果可以同时输入给两个上传操作符，再通过组合操作符合并两个上传流程的结果。

修改后的代码

addItem(courseCode: string, unitId: string, formValue: AddItemFormValue) {
  // 构造共享源流：接口只请求一次，结果可同时喂给多个下游
  const source$ = this.courseApi.addItem(courseCode, unitId, body).pipe(
    map(data => data.data),
    share()
  );

  // 并行执行两个上传操作，等待全部完成后合并结果返回
  return forkJoin([
    source$.pipe(this.makeFileUploadRequests(formValue)),
    source$.pipe(this.makeVideoUploadRequests(formValue))
  ]).pipe(
    // 可选：按业务需求调整返回结构，这里是合并原始数据和两个上传结果的示例
    withLatestFrom(source$),
    map(([[fileUploadRes, videoUploadRes], originData]) => {
      return {
        ...originData,
        uploadedFiles: fileUploadRes,
        uploadedVideos: videoUploadRes
      };
    })
  );
}

// 你原有文件上传操作符不用做任何修改
private makeFileUploadRequests<T extends { files: { name: string; params: ServerModel.FileResponse }[] }>(formValue: {
  files: UploadingFile[];
}) {
  return mergeMap<T, Observable<ServerModel.FileUploadedResponse[] | T>>(data => {
    return Array.isArray(data.files) && data.files.length > 0 ? this.uploadFiles(data.files, formValue.files) : of(data);
  });
}

// 你编写的视频上传操作符保持和上面一致的结构即可，只需要调整对应的字段和上传逻辑
private makeVideoUploadRequests<T extends { videos: { name: string; params: ServerModel.VideoResponse }[] }>(formValue: {
  videos: UploadingVideo[];
}) {
  return mergeMap<T, Observable<ServerModel.VideoUploadedResponse[] | T>>(data => {
    return Array.isArray(data.videos) && data.videos.length > 0 ? this.uploadVideos(data.videos, formValue.videos) : of(data);
  });
}

注意说明

• share操作符是核心：RxJS默认冷流每被订阅一次就会重新执行上游逻辑，不加share会导致courseApi.addItem被重复调用两次。
• 如果不需要等两个上传都完成再返回结果，可以把forkJoin替换为merge，两个上传的结果会按完成顺序依次推送。
• 可以给每个上传操作符单独加catchError处理单类上传失败的场景，避免其中一个上传失败直接终止整个流程。

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