为每个循环获取可观察数据

Question

我对 Observable 相当陌生，所以我相信必须有一种方法可以绕过我的 for 每个循环。

this.pages.forEach(page => {
  page.tasks = [];
  page.image = this.getImageData(this.proof.proofGuid, page.pageGuid);
  page.tasks = this.getPageTasks(this.order.orderNum);
});

这两个函数都从我的服务订阅了一个 observable，但是由于它们是异步的，所以 for 每个循环在数据返回并放在对象上之前就完成了。

getImageData(proofGuid, pageGuid) {
  this.proofService.getProofImage(proofGuid, pageGuid)
    .subscribe(image => {
      return image;
    });
}

getPageTasks(orderNum) {
  this.taskService.getTasksByOrderIdForProof(orderNum)
    .subscribe(data => {
      if (data) {
        return data;
      }
    });
}

现在我别无选择，只能循环获取每个页面的数据。我相信必须有一种方法可以基本上强制完成订阅，然后才能进入 for each 的下一页。

Answer 1

您对 observables 的使用/对异步执行的理解需要一些清理。

首先这个：

getImageData(proofGuid, pageGuid) {
this.proofService.getProofImage(proofGuid, pageGuid)
  .subscribe(image => {
    return image;
  });
}

getPageTasks(orderNum) {
this.taskService.getTasksByOrderIdForProof(orderNum)
  .subscribe(data => {
    if (data) {
      return data;
    }
  });
 }

没有意义。你实际上并没有返回任何东西。这些函数将在调用时执行，然后您“返回”的值将被抛出。因为 subscribe 函数没有返回。它的接口将其定义为返回 void，因此在其内部调用 return 是无用的。相反，您应该返回流本身并在需要值的地方调用 subscribe ：

getImageData(proofGuid, pageGuid) {
  return this.proofService.getProofImage(proofGuid, pageGuid);
}

getPageTasks(orderNum) {
  return this.taskService.getTasksByOrderIdForProof(orderNum);
}

好多了。现在为您执行。

this.pages.forEach(page => {
  page.tasks = [];
  page.image = this.getImageData(this.proof.proofGuid, page.pageGuid);
  page.tasks = this.getPageTasks(this.order.orderNum);
});

这已经没有意义了。更准确地说，应该是：

this.pages.forEach(page => {
  page.tasks = [];
  this.getImageData(this.proof.proofGuid, page.pageGuid).subscribe(image => page.image = image);
  this.getPageTasks(this.order.orderNum).subscribe(tasks => page.tasks = tasks);
});

在这一点上，它实际上可能工作得很好。但是你是对的，你可以将 observable 和所有这些结合起来，你可以这样做：

this.pages.forEach(page => {
  page.tasks = [];
  Observable.forkJoin(this.getImageData(this.proof.proofGuid, page.pageGuid), 
      this.getPageTasks(this.order.orderNum))
      .subscribe(([image, tasks]) => {
          page.image = image;
          page.tasks = tasks;
      });
});

forkJoin 运算符将并行执行异步调用，并在两者都完成后在匹配数组中为您提供结果。

您可以获得更多创意/效率，并且只执行一次任务调用，因为看起来输入参数永远不会改变，并且您可以将所有图像调用合并到一个并行流中。

重要的是要记住，observables 只是你如何处理数据流的定义，在你订阅之前实际上不会做任何事情。所以你可以做这样的事情：

getFullPages(orderNum, pages) {
  // define simple stream to retrieve tasks
  let tasks$ = this.getPageTasks(orderNum); 

  // map pages array into an array of streams to fetch image and add image to page and return the page w the image and join them all
  let pageImages$ = Observable.forkJoin(
                       pages.map(page => // be careful, this is the array map operator
                            this.getImageData(this.proof.proofGuid, page.pageGuid)
                                .map(image => { // this is the rx map operator
                                    page.image = image;
                                    return page; 
                                })));

  // define stream to join tasks and pages into one stream and merge
  let combined$ = Observable.forkJoin(tasks$, pageImages$, 
                           (tasks, pageImages) => pageImages.map(page => { //array map operator
                              page.tasks = tasks;
                              return page;
                           }));
  return combined$; // return to caller
}

// execute and log full pages array
this.getFullPages(this.order.orderNum, this.pages)
    .subscribe(fullPages => console.log(fullPages));

请注意，在您调用 subscribe 之前，什么都不会执行，因为您只是定义了处理和组合数据流的方法，然后一旦您订阅了，所有内容都会立即执行，并且您会在所有内容都完成后获得最终结果.

您可能想知道，该死的，代码有点多，方式也更复杂，为什么要这样做？原因是 1. 更高效的执行和 2. 这个流定义是完全可重用的。如果我需要再次调用它，我可以调用该函数并订阅该流定义，并且完全定义的 pages 数组将仅通过该调用返回给您。在这种情况下，这对您来说可能并不重要，但通常情况下，它非常有用。