Angular 递归 HTTP 请求，组合响应，最终返回单个 observable

Question

我想对分页 API 进行递归 HTTP 调用，直到浏览完所有页面。每个页面都包含一个资产数组，我将它们连接到一个临时数组。阅读完所有页面后，我想将所有资产的单个数组作为可观察对象返回。

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

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

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

   _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。

Answer 2

你想要一个toArray 运算符

这是一个带有timer的示例

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