如何合并使用 FROM 创建的 observables 数组的所有映射答案

【问题标题】：How to merge all maps of an observables array created with FROM如何合并使用 FROM 创建的 observables 数组的所有映射
【发布时间】：2019-09-04 20:56:08
【问题描述】：

我正在使用 FROM 运算符创建一个可观察对象数组，每个可观察对象都使用 margeMap 进行转换。首先，使用此 uploadService.addFile 将项目添加到存储库，然后如果未将 File 标记为已上传，则使用 uploadService.uploadFile 进行上传，此服务会发出上传进度，因此我添加了一个过滤器以仅在文件时发出上传或未上传

我想要做的只是发出一个信号，表明所有文件都已上传

我尝试使用 forkJoin，但使用该运算符我没有任何发射，我认为是因为在某些时候我需要指示操作在最后一个管道内完成，但我不知道如何执行此操作。

/*this.therapyFiles this is an array of elements*/
from(this.therapyFiles).pipe(
      mergeMap(therapyFile => this.fileParser.parseTherapyEndTime(therapyFile.file).pipe(
        map(date => {
          therapyFile.endTime = date;
          return therapyFile;
        })
      )),
      mergeMap(therapyFile =>
        this.uploadService.addFile(
          this.facilityId,
          therapyFile.idDevice,
          therapyFile.file.name,
          therapyFile.endTime
        ).pipe(
          switchMap(file => file.isUploaded ?
            of(this.AlreadyUploaded) :
            this.uploadService.uploadFile(therapyFile.file)),
          map(progress => {
            therapyFile.uploadProgress = progress;
            return therapyFile;
          }),
          filter(uploadedFile => uploadedFile.uploadProgress > 99 || uploadedFile.uploadProgress < 0)
        )
      )
    ).subscribe(z => console.log(z)); // here I get all emition for every item in therapyFiles

/*this.therapyFiles this is an array of elements*/
forkJoin(from(this.therapyFiles).pipe(
      mergeMap(therapyFile => this.fileParser.parseTherapyEndTime(therapyFile.file).pipe(
        map(date => {
          therapyFile.endTime = date;
          return therapyFile;
        })
      )),
      mergeMap(therapyFile =>
        this.uploadService.addFile(
          this.facilityId,
          therapyFile.idDevice,
          therapyFile.file.name,
          therapyFile.endTime
        ).pipe(
          switchMap(file => file.isUploaded ?
            of(this.AlreadyUploaded) :
            this.uploadService.uploadFile(therapyFile.file)),
          map(progress => {
            therapyFile.uploadProgress = progress;
            return therapyFile;
          }),
          filter(uploadedFile => uploadedFile.uploadProgress > 99 || uploadedFile.uploadProgress < 0)
        )
      )
    )).subscribe(z => console.log(z)); // I tried in this way but never get an emition, but all inside code of forkjoin works as expected

我希望只有一个发射表明所有文件都已上传

【问题讨论】：

标签： angular rxjs

【解决方案1】：

使用 concatAll

   of(this.therapyFiles).pipe(
      mergeMap(therapyFile => this.fileParser.parseTherapyEndTime(therapyFile.file).pipe(
        map(date => {
          therapyFile.endTime = date;
          return therapyFile;
        })
      )),
      mergeMap(therapyFile =>
        this.uploadService.addFile(
          this.facilityId,
          therapyFile.idDevice,
          therapyFile.file.name,
          therapyFile.endTime
        ).pipe(
          switchMap(file => file.isUploaded ?
            of(this.AlreadyUploaded) :
            this.uploadService.uploadFile(therapyFile.file)),
          map(progress => {
            therapyFile.uploadProgress = progress;
            return therapyFile;
          }),
          filter(uploadedFile => uploadedFile.uploadProgress > 99 || uploadedFile.uploadProgress < 0)
        )
      ),
      concatAll()
    ).subscribe(z => console.log(z)); // I tried in

【讨论】：

如果我尝试使用 concatAll 而不将初始 FROM 更改为 OF，它会出错，如果我没记错的话，如果我使用 OF 而不是 FROM，我将发射所有数组的元素，但我需要的是对数组的每个元素都有一个发射，

【解决方案2】：

经过一番研究，我发现在第一个管道中，fileParser.parseTherapyEndTime 正在使用一个主题来发出他的结果，但是该主题没有使用 complete() 方法完成，该服务的所有调用都不会完成，并且所有组合运算符都需要完成所有内部可观察对象。所以我在 fileParser.parseTherapyEndTime 中添加了 complete() 方法而且我使用 toArray() rxjs 运算符将所有发射组合成一个数组，这是所需的行为。

我以这种方式更新我的代码：

    from(this.therapyFiles).pipe(
      mergeMap(therapyFile => this.fileParser.parseTherapyEndTime(therapyFile.file)
      .pipe(
        map(date => {
          therapyFile.endTime = date;
          return therapyFile;
        })
      )),
      mergeMap(therapyFile =>
        this.uploadService.addFile(
          this.facilityId,
          therapyFile.idDevice,
          therapyFile.file.name,
          therapyFile.endTime
        )
        .pipe(
          switchMap(file => file.isUploaded ?
            of(this.AlreadyUploaded) :
            this.uploadService.uploadFile(therapyFile.file)),
          map(progress => {
            therapyFile.uploadProgress = progress;
            return therapyFile;
          }),
          filter(uploadedFile => uploadedFile.uploadProgress > 99 || uploadedFile.uploadProgress === this.AlreadyUploaded)
        ),
      ),
      toArray()
    ).subscribe(() => {
      this.uploadedFiles = true;
    });

对不起，我的英语不好，但我还在练习

【讨论】：