Angular 7 wait for all subscriptions

Question

I have a component that is currently set up like this:

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]);
      });
    });
  }
}

As you can see from the listProducts method, I get a list of products based on the gtins that have been passed as a parameter. What I would like to do is wait for all the subscriptions to finish in listProducts and after they have done, to run some code.

How can I do this?

I think you're looking for something like this to combine the observables and wait for all, that's RxJs [How to 'wait' for two observables in RxJS](https://stackoverflow.com/questions/44004144/how-to-wait-for-two-observables-in-rxjs) You might also want to check out [Observable](http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html) — rmjoia, Apr 15 '19 at 12:57

Daniel B · Answer 1 · 2019-04-15T13:15:43.090

4

If you want to wait for all observables, use the forkJoin operator and then flatten the result to push it to your products array.

I'd also suggest to use the takeUntil operator as a cleanup helper for your subscriptions. See how it can be used in my example.

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

And in your onDestroy hook

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

And in your component, define destroyed$ as

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

edited Apr 15 '19 at 13:15

answered Apr 15 '19 at 13:00

Daniel B

8,770
5
43
76

@ritaj beat me to it, but I'll leave this as a reference for a chained answer. – Daniel B Apr 15 '19 at 13:01
1

Wouldn't it be more idiomatic to do the pushing in `subscribe` rather than `map`? (AFAIK `map` is supposed to be side-effect free in functional reactive programming) – meriton Apr 15 '19 at 13:13
@meriton, very true, since we don't mutate the data there's no need for the map! Thanks! :) – Daniel B Apr 15 '19 at 13:18
@meriton Yes, it should be done inside `tap` or `subscribe`. – Roberto Zvjerković Apr 15 '19 at 13:19
@DanielB Not only that, but you will map the result to the length of `this.products` array (since push returns array length), which will lose you the information for later stream chaining. – Roberto Zvjerković Apr 15 '19 at 13:21
I get this error when using your code: Argument of type 'void' is not assignable to parameter of type 'ObservableInput<{}>'. – r3plica Apr 15 '19 at 14:46
on `gtins.forEach` – r3plica Apr 15 '19 at 14:46

score 3 · Answer 2 · answered Apr 15 '19 at 12:58

I believe you will need to use forkJoin.

It waits for all the Observables from the array to finish and then emits.

Do the individual observable operations inside tap of each of them.

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');
    });
}

Angular 7 wait for all subscriptions

2 Answers2