Angular 7等待所有订阅

Question

我有一个组件，目前是这样设置的：

export class CompareComponent implements OnInit, OnDestroy {
  criteria: Criteria[]
  products: Product[] = [];

  private organisationId: string

  private criteriaSubscription: Subscription
  private routeSubscription: Subscription
  private productSubscription: Subscription

  constructor(
    @Inject(PLATFORM_ID) private platformId: Object,
    private route: ActivatedRoute,

    private attributeMatchService: AttributeMatchService,
    private criteriaService: CriteriaService,
    private optionsService: OptionsService,
    private productService: ProductService,
  ) { }

  ngOnInit() {
    this.organisationId = this.optionsService.get().organisation;
    this.routeSubscription = this.route.paramMap.subscribe(paramMap => {
      let category = paramMap.get('category');
      let gtins = paramMap.get('gtins').split('.');
      this.listCriteria(category);
      this.listProducts(category, gtins);
    });
  }

  ngOnDestroy() {
    if (this.criteriaSubscription) this.criteriaSubscription.unsubscribe();
    if (this.productSubscription) this.productSubscription.unsubscribe();
    if (this.routeSubscription) this.routeSubscription.unsubscribe();
  }

  close() {
    if (isPlatformServer(this.platformId)) return;
    window.history.back();
  }

  private listCriteria(category: string): void {
    this.criteriaSubscription = this.criteriaService.list(category, 'Attributes').subscribe(criteria => this.criteria = criteria);
  }

  private listProducts(category: string, gtins: string[]) {
    gtins.forEach(gtin => {
      this.productSubscription = this.productService.get(category, this.organisationId, parseInt(gtin)).subscribe(products => {
        this.products.push(products[0]);
      });
    });
  }
}

从listProducts 方法中可以看出，我得到了一个基于作为参数传递的gtins 的产品列表。 我想做的是等待所有订阅在listProducts 中完成，并在完成后运行一些代码。

我该怎么做？

Answer 1

我相信你需要使用forkJoin。

它等待数组中的所有 Observables 完成，然后发出。

在tap 中对它们每个进行单独的可观察操作。

private listProducts(category: string, gtins: string[]) {
    const products$: Observable<any>[] = [];

    gtins.forEach(gtin => {
        products$.push(this.productService.get(category, 1, parseInt(gtin))
                        .pipe(tap(product => {
                              this.products.push(products[0]);
        }));
    });

    forkJoin(products$).subscribe(() => {
        console.log('All subscriptions done');
    });
}

Answer 2

如果您想等待所有 observables，请使用 forkJoin 运算符，然后将结果展平以将其推送到您的 products 数组。

我还建议使用 takeUntil 运算符作为订阅的清理助手。看看如何在我的示例中使用它。

forkJoin(
      gtins.forEach(gtin => this.productService.get(category, this.organisationId, parseInt(gtin)))
    ).pipe(
      takeUntil(this.destroyed$)
    ).subscribe(products => {
      this.products.push(...products.flat())
    });

在你的onDestroy钩子里

  onDestroy() {
    this.destroyed$.next();
    this.destroyed$.complete();
  }

在您的组件中，将 destroyed$ 定义为

destroyed$: Subject<any> = new Subject<any>();