RxJS 和 Angular HttpClient：使用 forkJoin 处理多个顺序请求

Question

在 RxJS (ES6) 中，我试图在单个 Observable 中通过一系列顺序操作获得结果。

我不确定我是否必须使用 forkJoin（但我希望操作按顺序执行）或 concat 运算符（但我希望在所有操作都执行时在最后收到通知） .

我试过了：

forkJoin

  sync(): Observable<any> {
    return from(this.db.getProducts()).pipe(
        map(products => {
            if ( !products ) {
                return of(true);
            }

            const batch: Observable<any>[] = [];
            for ( const product of products ) {
              if ( product.toBeSync ) {
                  batch.push(this.api.updateProduct(product));
              }
            }

            return forkJoin(batch);
        })
    );
  }

连接

  sync(): Observable<any> {
    return from(this.db.getProducts()).pipe(
        map(products => {
            if ( !products ) {
                return of(true);
            }

            const batch: Observable<any>[] = [];
            for ( const product of products ) {
              if ( product.toBeSync ) {
                  batch.push(this.api.updateProduct(product));
              }
            }

            return concat(batch);
        })
    );
  }

在这两种情况下，我都看不到来自 HttpClient 的可观察对象正在执行他们的工作（没有发送 http 请求），但我可以看到日志记录。

批处理中被调用的方法如下：

  updateProduct(product: Product) {
      console.log('calling...');
      const options = this.getDefaultRequestOptions();
      return this.http.post('update_product', product, options);
  }

我像下面这样调用sync()函数：

this.productService.sync().subscribe(() => {
  console.log('sync done');
}, error => this.utils.handleError(error));

输出：

(8) calling...
sync done

但没有启动 HTTP 请求。

如果我在管道图中的 forkJoin/concat 之外执行相同操作，我可以看到发送的 HTTP 请求。

sync(): Observable<any> {

    const product = new Product(null, 'Title', 'Desc.', 'CODE');
    return this.api.updateProduct(product);

}

我错过了什么？

--- 更新 - 解决方案 ---

sync(): Observable<any> {
    return from(this.db.getProducts()).pipe(
        flatMap(products => {

            if ( !products ) {
                return of(true);
            }

            const batch: Observable<any>[] = [];
            for ( const product of products ) {
              if ( product.toBeSync ) {
                  batch.push(this.api.updateProduct(product));
              }
            }

            console.log(batch);

            return concat(...batch);
            // return forkJoin(batch);
        })
    );

Answer 1

试试

sync(): Observable<any> {
  return from(this.db.getProducts()).pipe(
    map(products => {
      const batch: Observable<any>[] = [];
      for (const product of products) {
        batch.push(this.api.updateProduct(product));
      }
     return batch;
   }),
   switchMap(batch => forkJoin(batch)),
  )
}

这是假设this.db.getProducts() 是同步静态数据，而不是可观察/异步数据。

那你可以试试做

  this.productService.sync().subscribe(() => { console.log('sync done'); });

并查看是否正在进行任何 API 调用。

Answer 2

你的问题不是forkJoin/concat这里，它是使用map返回另一个observable。

在您的代码中，map 将返回 Observable<Observable<any>>。您的 IDE 应该突出显示这些东西（不要使用 any，可能这里有问题）。

您的解决方案是将map 更改为concatMap（至少用于测试）并且下一个链应该可以工作。

另外，调试subscribe 及其返回值，看看你有什么。