RxJs Angular 7 HttpClient使用forkJoin进行多个POST请求，如何去掉第二个subscribe？

Question

我正在创建一个带有链接子任务对象的工作项对象。我的函数（createWorkItemAndChildren）接受两个参数，workItem和Task对象的数组。我希望我的函数返回所有创建的ID数组（工作项和任务）。

在进行第一次http POST调用（workItemService.createWorkItem）时，我必须从父级ID获取回复，然后才能使用相同服务上的另一个http POST方法创建子任务。

现在，在createChildWorkItems中使用forkJoin同时返回所有子ID。

如何重构此代码以便只有一个订阅，并将包含父级和子级ID的数组返回？

  createChildWorkItems(parentId, tasks: Task[]): Observable<any> {
    return <Observable<number>> forkJoin(
      tasks.map(task => <Observable<number>> this.workItemService.createChildWorkItem(parentId, task))
    ).pipe(zip());

  }

  createWorkItemAndChildren(workItem, childTasksToSave: Task[]){
    var resultArray = [];
    this.workItemService.createWorkItem(workItem).subscribe(workItemId => {
      var parentId = workItemId;
      resultArray.push(parentId);
      if (parentId !== null){
        this.createChildWorkItems(parentId, childTasksToSave).subscribe((results: number) => {
          resultArray.push(results);
          this.tfsIdsCreated = resultArray;
        });
      }
    });
  }

- Jennifer S

Answer 1

在您的情况下，子进程不适合使用fork join，而应该使用Async/await。fork join以异步方式发送请求，但是Async/await将等待每个请求的响应，当您获得响应时，将该响应附加到父级中，非常简单。

在Async/await中，请求将按照顺序进行，就像循环一样。当所有请求完成后，返回该对象。这里是Async/await的链接：https://lavrton.com/javascript-loops-how-to-handle-async-await-6252dd3c795/。

“最初的回答”

  createWorkItemAndChildTasks(workitem, childTasksToSave: Task[]): any {
    this.workItemService.createWorkItem(workitem).subscribe(workItemId => {
     var parentId = workItemId;
      if (parentId !== null ){
        this.tfsIdsCreated.push(parentId);

    // now create the children on the workItemForm.
    for (let child of childTasksToSave){
//use here Async await when you get response attach to parent 

     this.workItemService.createChildWorkItem(parentId, child).subscribe(task =>{
        if (task !== null){
          console.log(' createWorkItem received child taskid: ' + task);
          this.tfsIdsCreated.push(task);
        }
      });
    }
  }
  return this.tfsIdsCreated;
});

}

- Shahid Islam

---

async/await 的语法会有哪些变化？ - Jennifer S

在 createWorkItemAndChildTasks 函数中添加 async 关键字，尝试如下代码：async createWorkItemAndChildTasks(workitem, childTasksToSave: Task[]) { for (let child of childTasksToSave) { const task = await fetchChild(parentId, child); if (task !== null) { console.log('createWorkItem received child taskid: ' + task); this.tfsIdsCreated.push(task); } } }async fetchChild(parentId, child) { const response = await this.workItemService.createChildWorkItem(parentId, child).toPromise(); return response; } - Shahid Islam

Answer 2

创建一个可观察对象数组并将引用传递到forkjoin函数中。

let observableBatch= [];

for (let child of childTasksToSave){
   observableBatch.push(this.workItemService.createChildWorkItem(parentId, child));
}

Observable.forkJoin(observableBatch).subscribe...;

reference: https://dev59.com/JFsV5IYBdhLWcg3w4iJA#35676917

- Abel Valdez

Answer 3

这是我最终采取的做法，这也是为什么我接受了kos的答案： 我的tfsIdsCreated数组正在订阅结果，提供id列表。

createParentAndKids(workItem, childTasksToSave){
    this.workItemService.createWorkItem(workItem).pipe(
      mergeMap(parentId => {
        if (parentId === null){
          return of([parentId]);
        }

        const childTaskObservables$ = childTasksToSave.map(
          child => this.workItemService.createChildWorkItem(parentId, child)
        );

        return forkJoin(childTaskObservables$).pipe(
          map(ids => [parentId, ...ids])
        );

      })
    ).subscribe(x => this.tfsIdsCreated.push(x));
  }

- Jennifer S

Answer 4

如果您想并行执行子任务-forkJoin是您的选择。

这里是一个简单的例子：

createParent().pipe(
  // switch from parent stream to forkJoin(...children)
  mergeMap(parent =>
    // wait for all children to be created
    forkJoin(children.map(child => createChild(parent, child))).pipe(
      // combine childResults with parent
      map(childResults => {
        // do operations with parent and all children
        parent.childResults = childResults;
        // switch back to parent
        return parent;
      })
    )
  )
)
.subscribe(parent => {
  // ...
})

请注意我们只订阅了一次Observable - 这是一个好习惯。

这里不需要使用async await。

希望这可以帮助。

- kos

---

我很感谢您的建议，但我仍然有一些问题不理解如何将我的代码适应这个。在您的createParent()示例代码中，它返回一个observable还是只是一个数字？ - Jennifer S

@JenniferS， createParent() 应该返回 Observable，类似于你的 this.workItemService.createWorkItem。我编译了一个更接近你代码的小例子：mergeMap with forkJoin。 - kos

感谢您额外的努力解释。但是 rxObserver 是做什么的？ - Jennifer S

@JenniferS 这是图表 API 的一部分，用于呈现那些大理石图。它实现了 Observer 接口，因此如果您想将其输出到控制台，可以将其替换为 x=>console.log(x)。 - kos

@JenniferS，很高兴听到这个好消息 :) GL - kos