Typescript 使用管道和清理转换 forkJoin 结果

Question

forkJoin() 转换结果的最佳方法是什么，所以在 subscribe() 方法中我可以这样写：

this.refreshGraph(dependencies,microservices);

这是它现在的工作方式：

  loadAll(): void {
    const dependencies = this.dependencyService.query();
    const microservices = this.microserviceService.query();


    forkJoin({dependencies, microservices})
      .subscribe(results => {
        this.refreshGraph(results.dependencies.body || [], results.microservices.body || []);
      });
  }

  refreshGraph(dependencies: IDependency[], microservices: IMicroservice[]): void {
  ...
  }

我是否需要将 forkJoin().subscribe() 的结果存储在 Subscription 中并在组件被销毁时手动取消订阅？ dependencyService 和 microserviceService 在我的例子中只是包装 http:

  query(req?: any): Observable<EntityArrayResponseType> {
    const options = createRequestOption(req);
    return this.http.get<IDependency[]>(this.resourceUrl, { params: options, observe: 'response' });
  }

完整的源代码可用here

Answer 1

这只是一点点重构。您可以在 query() 方法内转换数据，也可以在 loadAll() 方法内转换数据。

以下是您可以重写 loadAll() 方法的一种方式。

 loadAll(): void {
    const dependencies = this.dependencyService.query().pipe(map(result => result.body || []));
    const microservices = this.microserviceService.query().pipe(map(result => result.body || []));;


    forkJoin({dependencies, microservices}).subscribe(
       ({dependencies, microservices}) => this.refreshGraph(dependencies,microservices)
    );
  }

请记住，如果提供给 forkJoin 的任何内部 observables 出现错误，您将失去任何其他将或已经完成的 observables 的值。 所以最好正确处理错误。您可以像这样捕获错误并返回一个空数组。

 const dependencies = this.dependencyService.query().pipe(
  map(result => result.body || []),
  catchError(() => of ([]))
);
const microservices = this.microserviceService.query().pipe(
  map(result => result.body || []),
  catchError(() => of ([]))
);

而且你也不需要在这里手动取消订阅，因为 observable 是由 Angular 的 httpClient 服务创建的。看到这个post。一旦所有内部可观察对象完成，forkJoin 将自动完成。我希望这会有所帮助。

Answer 2

感谢article，有两种可能性

使用 zip()：

  loadAll(): void {
    zip(this.dependencyService.query(), this.microserviceService.query())
      .pipe(
        map(([dependencyResponse, microserviceResponse]) => {
          return {
            dependencies: dependencyResponse.body || [],
            microservices: microserviceResponse.body || []
          }
        })
      ).subscribe(results => {
      this.refreshGraph(results.dependencies, results.microservices);
    });
  }

或使用 forkJoin()

  loadAll(): void {
    const dependencies$ = this.dependencyService.query();
    const microservices$ = this.microserviceService.query();

    forkJoin({dependencies$, microservices$})
      .pipe(
        map(result => {
            return {
              dependencies: result.dependencies$.body || [],
              microservices: result.microservices$.body || []
            }
          }
        )
      )
      .subscribe(results => {
        this.refreshGraph(results.dependencies, results.microservices);
      });
  }

在这种特殊情况下，两者的工作方式相同，因为源 observable 只会发射一次（请参阅来自 Michael D 的 cmets）

您不需要取消订阅，因为 http 是 finite observable 另请参阅 this answer