如何在管道内调用 forkJoin？答案

【问题标题】：How to call forkJoin inside a pipe?如何在管道内调用 forkJoin？
【发布时间】：2019-11-14 13:39:51
【问题描述】：

TL;DR 是否可以在 observable 的管道内使用 forkJoin？

全文：我有一个返回对象数组的 Observable 的服务。对于这些对象中的每一个，我都需要再次调用返回 observable 的服务，并对每个结果应用另一个操作。

op1 ->[] ----> op2 --> op3
          \\-> op2 --> op3
           \-> op2 --> op3

我目前的解决方案是这样的：

    this.tournamentParticipantService.getNotAssigned(this.tournamentId).subscribe(
      (players: Player[]) => {
        let dict = {};
        players.forEach(player => {
          dict[player.id] = this.teamService.add(
            { 
                id: 0,
                members: [],
                tournament: this.tournamentId,
                name: player.name + " " + player.surname
            })
            .pipe(
              map((team: Team) => 
                this.teamMemberService.add({ player: player.id, team: team.id })
              ))
        });
        forkJoin(dict).subscribe(result => console.log(result));
     });

我想摆脱第一个订阅并改用管道。问题是 forkJoin 的教程将其显示为对象或数组被传递到的源，而不是管道的一部分。

从地图内部调用 forkJoin

.pipe(
    map(value=>forkJoin(value))

返回 Observable<Observable<resolved forkJoin arguments>> 我可能需要递归订阅。好像不是什么好办法。

将 fJ 放入不带参数的管道中

this.tournamentParticipantService.getNotAssigned(this.tournamentId).pipe(
      map((players: Player[]) => players.map(
        (player: Player) => this.teamService.add({ id: 0, members: [], tournament: this.tournamentId, name: player.name + " " + player.surname })
          .pipe(
            map((team: Team) => {
              let pipe = new JsonPipe();
              console.log("team: " + pipe.transform(team) + " player: " + pipe.transform(player));
              this.teamMemberService.add({ player: player.id, team: team.id });
            })))),
      forkJoin
    ).subscribe((result: [[Observable<void>]]) => {
      console.log(result)
      result[0].forEach(element => {
        element.subscribe(res => console.log(res));
      });
    });

以奇怪的可观察对象的复杂结构结束。这似乎也不是什么好办法。

甚至可以在管道内使用 forkJoin 吗？

【问题讨论】：

标签： angular rxjs rxjs-pipeable-operators

【解决方案1】：

我认为您缺少的东西是 forkJoin 将返回一个可观察的，因此标准 map 不是您要寻找的：您需要使用其中一张地图来处理以适当的方式观察到的结果（即switchMap、exhaustMap、mergeMap）：

.pipe(
    switchMap(value => forkJoin(getThingOne(value), getThingTwo(value)))
.subscribe(([thing1, thing2]) => 
{

})

【讨论】：

【解决方案2】：

解决方案采取的最终形式是：`

this.tournamentParticipantService.getNotAssigned(this.tournamentId).pipe(
  map((players: Player[]) => players.map(
    (player: Player) => this.teamService.add({ id: 0, members: [], tournament: this.tournamentId, name: player.name + " " + player.surname })
      .pipe(
        map((team: Team) =>
          this.teamMemberService.add({ player: player.id, team: team.id }))
      ))),
  mergeMap((tasks: Observable<Observable<void>>[]) => forkJoin(tasks)),
  mergeMap((tasks: Observable<void>[]) => forkJoin(tasks))
).subscribe(() => {
  this.loadTeams();
});

【讨论】：