角度递归HTTP请求，组合响应，最终返回单个可观察对象

Question

我想对分页的API进行递归HTTP调用，直到遍历完所有页面为止。每个页面包含一个资源数组，我将把它们连接到一个临时数组中。一旦我读完了所有的页面，我想返回一个包含所有资产的数组作为一个可观察对象。

我一直在使用带有扩展操作符的管道。我注意到我可以在订阅完成时打印我的最终数组，但我不知道如何优雅地将此数组作为可观察对象返回。是否有某种方法可以等待一个观察值的完成，然后返回另一个观察值？

getAllAssets(sort: string, filter: AssetFilter): Observable<Asset[]> {
  let allAssets: Asset[] = [];
  console.log('getAllAssets()', sort, filter);
  this._getAllAssets(null, null).subscribe((moreAssets) => {
    allAssets = allAssets.concat(moreAssets);
  }, null, () => console.log(allAssets));
  return of(allAssets);
}

_getAllAssets(sort: string, filter: AssetFilter) {
  let currentPage = 0;
  return this.getAssets(null, null, sort, filter).pipe(
    expand(assetsPage => {
      if (assetsPage && assetsPage.page.number + 1 < assetsPage.page.totalPages) {
        return this.getAssets(currentPage += 1, null, sort, filter);
      }
      return empty();
    }),
    map((value) => value._embedded.assets)
  );
}

getAssets(page: number, size: number, sort: string, filter: AssetFilter): Observable<Assets> {
  let url = 'https://gateway.' + environment.region + '.mindsphere.io/api/assetmanagement/v3/assets?';
  if (page) {
    url += 'page=' + page;
  }
  if (size) {
    url += 'size=' + size;
  }
  if (sort) {
    url += 'sort=' + sort;
  }
  if (filter) {
    url += 'filter=' + JSON.stringify(filter);
  }
  return this.http.get<Assets>(url, {
    headers: new HttpHeaders().set('Content-Type', 'application/json').set('Authorization', 'Bearer ' + this.accessToken)
  });
}

我正在将正确的数组打印到屏幕上，但我希望将此数组作为可观察数返回，而不是创建可观察数，返回可观察数，并在完成时发送给它。

编辑:我的解决方案是使用toArray()，如下所述。然后我做了一张地图，把Observable<item[][]>变成了Observable<item[]>

getAllAssets(sort: string, filter: AssetFilter): Observable<Asset[]> {
  let currentPage = 0;
  return this.getAssets(null, null, sort, filter).pipe(
    expand(assetsPage => {
      if (assetsPage && assetsPage.page.number + 1 < assetsPage.page.totalPages) {
        return this.getAssets((currentPage += 1), null, sort, filter);
      }
      return empty();
    }),
    map(value => value._embedded.assets),
    toArray()
  ).pipe(map(assets => [].concat.apply([], assets)));
}

getAssets(page: number, size: number, sort: string, filter: AssetFilter): Observable<Assets> {
  let url = 'https://gateway.' + environment.region + '.mindsphere.io/api/assetmanagement/v3/assets?';
  if (page) {
    url += 'page=' + page;
  }
  if (size) {
    url += 'size=' + size;
  }
  if (sort) {
    url += 'sort=' + sort;
  }
  if (filter) {
    url += 'filter=' + JSON.stringify(filter);
  }
  return this.http.get<Assets>(url, {
    headers: new HttpHeaders().set('Content-Type', 'application/json').set('Authorization', 'Bearer ' + this.accessToken)
  });
}

Answer 1

您需要一个toArray运算符

下面是一个包含timer的示例

timer(0, 5).pipe(
    take(3),
    toArray()
  )

​

​

Answer 2

这是一些需要筛选的代码，因此我将对其进行简化和泛化，因为这是一个常见的模式，它将采用以下形式：

   _getAllPages(page, constantArgs, lastSet) {
     lastSet = lastSet || []; // deal with entry
     return this.getPage(page, constantArgs).pipe( // fetch the page
        switchMap(res => 
          (res.nextPage) // if nextpage, recurse and concat to last set
             ? this._getAllPages(res.nextPage, lastSet.concat(res.items))
             : of(lastSet.concat(res.items)))); // else break
   }

出于风格的目的，我通常会让它成为私密的，并且我会有一个公共的，比如：

getAllPages(constantArgs) {
  return this._getAllPages(0, constantArgs);
}

使API更好/更容易理解。