如何使用 angular 6 和 RxJs 6 从订阅 forkJoin 中返回值？答案

【问题标题】：How to return value from subscribe of forkJoin with angular 6 and RxJs 6?如何使用 angular 6 和 RxJs 6 从订阅 forkJoin 中返回值？
【发布时间】：2019-02-28 23:10:46
【问题描述】：

private save(payload) {

  const newMedias: Observable<Media>[] = [];
  const newMediaInStepIndex: number[] = [];

  (payload.formData.steps).forEach((step: Step, stepIndex: number) => {
    const media: Observable<Media> = this.createOneMedia(step);
    if (media !== undefined) {
      newMedias.push(media);
      newMediaInStepIndex.push(stepIndex);
    }
  });

  forkJoin(newMedias).subscribe(medias => {
    medias.forEach((media, i) => {
      console.log('Media created with id: ' + media.id + ' in step ' + newMediaInStepIndex[i]);
      payload.formData.steps[newMediaInStepIndex[i]].media[0].id  = media.id;
    });
  });

  return this.createRecipe(payload);
}

我的问题是以下行是异步执行的，因此不是在结束返回之前：

payload.formData.steps[newMediaInStepIndex[i]].media[0].id  = media.id;

所以我需要在订阅结束后返回结束。我想我们必须把这个功能减半。

重要提示：保存时返回（有效负载）

【问题讨论】：

类似问题https://stackoverflow.com/q/48879798/1876572
@Eldho，在此示例中不会返回 save(isNew)。那是我的问题。

标签： angular rxjs6

【解决方案1】：

需要一个函数来返回从 forkjoin 组合的 observables，然后从其他地方订阅它

private saveToServer(payload) {

  (payload.formData.steps).forEach((step: Step, stepIndex: number) => {
    const media: Observable<Media> = this.createOneMedia(step);
    if (media !== undefined) {
      newMedias.push(media);
      newMediaInStepIndex.push(stepIndex);
    }
  });

  return forkJoin(newMedias) 
}


private save(payload) { 

  this.saveToServer(payload).subscribe(medias => {
    medias.forEach((media, i) => {
      console.log('Media created with id: ' + media.id + ' in step ' + newMediaInStepIndex[i]);
      payload.formData.steps[newMediaInStepIndex[i]].media[0].id  = media.id;

      //call it here
      this.createRecipe(payload);
    });
  }); 
}

【讨论】：

您的代码最终会为每个 observable 调用 createRecipe。 OP 希望在收到所有有效负载后运行 createRecipe。
ForkJoin 只有在所有请求都得到解决后才能订阅。
试图为你寻找一个例子。希望这个答案表明stackoverflow.com/a/50944184/1504265
stackoverflow.com/questions/47450639/… 这个也是
仔细阅读答案。 forkJoin waits for all observables to complete.

【解决方案2】：

subscribe方法有3个参数函数。

1) next()，你用过
2) error()，你还没有
3) complete()，在流完成后运行。

所以你可以像这样在完整的回调中添加你的 createRecipe：

forkJoin(newMedias).subscribe(medias => {
    medias.forEach((media, i) => {
      console.log('Media created with id: ' + media.id + ' in step ' + newMediaInStepIndex[i]);
      payload.formData.steps[newMediaInStepIndex[i]].media[0].id  = media.id;
    });
  },
err => {},
() => {
    this.createRecipe(payload);
});

这将在加载所有有效负载时调用 createRecipe 方法（一旦解析了 newMedias 数组中的所有可观察对象）。

希望这会有所帮助。

【讨论】：

如果newMedias 为空this.createRecipe(payload); 从不调用
那是因为订阅本身永远不会被调用。只有当 newMedias 不为空时，您才应该执行 forkJoin。否则直接调用createRecipe
谢谢，您的解决方案很好。我对media.id setter 有最后一个问题。错误类型错误：无法添加属性 id，对象不可扩展
您需要将有效负载对象克隆到一个新对象中并在那里分配它。在内部，payload 对象上调用了 preventExtensions。在这里阅读stackoverflow.com/a/49022130/1504265
我尝试const p = Object.assign({ }, payload); 但不起作用

【解决方案3】：

我的解决方案结束了吗：

private save(p): Observable<Action> {

    const payload = JSON.parse(JSON.stringify(p));

    const newMedias: Observable<Media>[] = [];
    const newMediaInStepIndex: number[] = [];

    (payload.formData.steps).forEach((step: Step, stepIndex: number) => {
        const media: Observable<Media> = this.createOneMedia(step);
        if (media !== undefined) {
          newMedias.push(media);
          newMediaInStepIndex.push(stepIndex);
        }
    });

    if (newMedias.length > 0) {
        return forkJoin(newMedias).pipe(mergeMap( (medias) => {
          medias.forEach((media, i) => {
            payload.formData.steps[newMediaInStepIndex[i]].media[0].id = media.id;
          });
          return this.createRecipe(payload);
        }));
    } else {
        return this.createRecipe(payload);
    }

}

我在pipe 中将subscribe 更改为mergeMap。我在forkJoin 的返回中返回Observable<Action>，并在save(p) 的末尾返回。

我添加这个：const payload = JSON.parse(JSON.stringify(p));，因为 p 来自 @ngrx。

【讨论】：

很好的解决方案！就我而言，“mergeMap”方法不起作用，而“map”有帮助。