Angular: Manage RxJS Subscriptions Declaratively

#rxjs #angular #javascript

Introduction
For the sake of discussion, I'll use Angular for my examples because it's more common to see RxJS in Angular application.

Managing Subscriptions is very important for your application performance. When you subscribe to an Observable, you register a callback function in the observable and the observable will maintain its callback lists. This can cause memory leak if you don't unsubscribe when your logic is done.

Let's give you an example. It's common to subscribe to different kind of observables in ngOnInit.

ngOnInit () {
  this.service.users$.subscribe(nextCb, errorCb, completeCb)
 }

but what if you navigate to a different route and went back to this component ? you will subscribe again and again.

Someone would say "Hmmm I'll save the subscription in a variable and unsubscribe in ngOnDestroy ".

users$

ngOnInit () {
  this.users$ = this.service.users$.subscribe(nextCb, errorCb, 
  completeCb)
 }
ngOnDestry(){
  this.users$.unsubscribe()
}

Technically you are right but what if there are multiple subscriptions ? Things will get messy real quick.

ngOnDestry(){
  this.variable1$.unsubscribe()
  this.variable2$.unsubscribe()
  this.variable3$.unsubscribe()
....
}

RxJS oberator takeUntil can be useful to declaratively remove your subscription

| Emits the values emitted by the source Observable until a notifier Observable emits a value.


@Component({ ... })
export class AppComponent implements OnInit, OnDestroy {
  destroy$: Subject<boolean> = new Subject<boolean>();

  constructor(private service: Service) {}

  ngOnInit() {
    this.service.users$
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });

 this.productsService.products$
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });
  }

  ngOnDestroy() {
    this.destroy$.next(true);
    this.destroy$.unsubscribe();
  }
}

is this the best way? Actually it's very good trick if you must subscribe to an observable inside a function. Optimally, you should let Angular handle your subscription for you using async pipe. this is very helpful because you will not need to subscribe in the .ts file anymore

here is an example from Deborah Kurata's github

export class ProductListComponent {
  pageTitle = 'Product List';
  private errorMessageSubject = new Subject<string>();
  errorMessage$ = this.errorMessageSubject.asObservable();

  private categorySelectedSubject = new BehaviorSubject<number>(0);
  categorySelectedAction$ = this.categorySelectedSubject.asObservable();

  products$ = combineLatest([
    this.productService.productsWithAdd$,
    this.categorySelectedAction$
  ])
    .pipe(
      map(([products, selectedCategoryId]) =>
        products.filter(product =>
          selectedCategoryId ? product.categoryId === selectedCategoryId : true
        )),
      catchError(err => {
        this.errorMessageSubject.next(err);
        return EMPTY;
      })
    );

  categories$ = this.productCategoryService.productCategories$
    .pipe(
      catchError(err => {
        this.errorMessageSubject.next(err);
        return EMPTY;
      })
    );

  vm$ = combineLatest([
    this.products$,
    this.categories$
  ])
    .pipe(
      map(([products, categories]) =>
        ({ products, categories }))
    );

  constructor(private productService: ProductService,
              private productCategoryService: ProductCategoryService) { }

  onAdd(): void {
    this.productService.addProduct();
  }

  onSelected(categoryId: string): void {
    this.categorySelectedSubject.next(+categoryId);
  }
}

In the previous example, the user can select a category and see the products in this category. all this logic without a single .subscribe() in the .ts file. All subscription are handled with async pipe in the template which automatically unsubscribe for you when it's unmounted.