NgNotes - Don't ever subscribe!
Tags: angular, rxjs, ngrx, javascript
Published: 2019-12-04 01:21:36

---
title: NgNotes - Don't ever subscribe!
published: true
description: Learn to avoid explicit subscriptions by leveraging Angular 2+ and RxJs.
tags: angular, rxjs, ngrx, javascript
---

### Update

Shout out to Andrei Gatej for an additional improvement that I've added as a final iteration below.

---

Let me start off by saying that there's always going to be exceptions and you should not refactor your current Angular source code. Rather take into consideration my notes and continue improving your style.

Observables and Subscriptions are a big part of the Angular 2+ web framework. The theory behind it can be a bit confusing at first and many Angular developers avoid Observables unless absolutely necessary. In my own experience, I've encountered production code bases where Observables were avoided like the plague and badly coerced into behaving in incorrect ways. This lead to memory-intensive applications that ran like a turtle after eating at a buffet.

Learning more about the RxJs API and Observables can lead to leaner applications that express and achieve a lot more in less code. Also, the Angular team recommends the *smart* use of Observables where possible.

Rather than go into an abstract discussion about Observables and Streams and Functional Programming and all that jazz, I'm going to illustrate a common mistake made by many developers when starting out with RxJs and Angular 2+. The first example implements a naive implementation of a numeric counter.

Below is a naive implementation which demonstrates quite a few initial misconceptions of RxJs and Angular. If you've been working with Angular for a while, you've probably encountered code similar to this, maybe even written it yourself. Don't worry, this is a no shame zone.

`naive-counter.service.ts`

```typescript
import { Injectable } from "@angular/core";
import { ReplaySubject } from "rxjs";
import { first } from "rxjs/operators";

@Injectable()
export class NaiveCounterService {
  counter$ = new ReplaySubject<number>(1);

  constructor() {
    this.counter$.next(0);
  }

  increment(): void {
    this.counter$
      .pipe(first())
      .subscribe((counter: number) => this.counter$.next(counter + 1));
  }

  decrement(): void {
    this.counter$
      .pipe(first())
      .subscribe((counter: number) => this.counter$.next(counter - 1));
  }
}


```

`naive-counter.component.ts`

```typescript
import { Component, OnInit } from "@angular/core";
import { NaiveCounterService } from "./naive-counter.service";

@Component({
  selector: "app-naive-counter",
  template: `
    Counter: <strong>{{ counter }}</strong>
    <button (click)="naiveCounterService.increment()">Increment</button>
    <button (click)="naiveCounterService.decrement()">Decrement</button>
  `
})
export class NaiveCounterComponent implements OnInit {
  counter: number;

  constructor(readonly naiveCounterService: NaiveCounterService) {}

  ngOnInit(): void {
    this.naiveCounterService.counter$.subscribe(
      (counter: number) => (this.counter = counter)
    );
  }
}


```

Let's identify every area of improvement. Biggest red flag, in my opinion, is the fact that we have a subscription which is not being disposed of when the component unmounts. This leads to memory leaks as these subscriptions could potentially continue running until the web application navigates away. Even worse, if the component is mounted and unmounted repeatedly, that's an additional, long running subscription added per life-cycle.

The simples way to solve this issue, and perhaps the most pragmatic, is to bind the observable itself to the template using the **async pipe**. Let's refactor.

`naive-counter.component.ts`

```typescript
import { Component } from "@angular/core";
import { NaiveCounterService } from "./naive-counter.service";

@Component({
  selector: "app-naive-counter",
  template: `
    Counter: <strong>{{ naiveCounterService.counter$ | async }}</strong>
    <button (click)="naiveCounterService.increment()">Increment</button>
    <button (click)="naiveCounterService.decrement()">Decrement</button>
  `
})
export class NaiveCounterComponent {
  constructor(readonly naiveCounterService: NaiveCounterService) {}
}
```

With this improvement, the responsibility over managing subscriptions is delegated from the developers to "Angular", or the Angular developers. The *async pipe*, as demonstrated above, subscribes to the observable and updates the UI with each notification received. Once the component unmounts, the *async pipe* unsubscribes from the observable automatically. The risk of long living subscriptions has been removed.

The above design illustrates a common pattern in Angular to provide long-living state for a component, or series of components, that will mount and unmount repeatedly throughout the application's usage. You should definitely *not* be creating long-living state where you don't really need it as this is the main cause of memory consumption over time.

The next step to improving this design is in the service. Here it is once again.

`naive-counter.service.ts`

```typescript
import { Injectable } from "@angular/core";
import { ReplaySubject } from "rxjs";
import { first } from "rxjs/operators";

@Injectable()
export class NaiveCounterService {
  counter$ = new ReplaySubject<number>(1);

  constructor() {
    this.counter$.next(0);
  }

  increment(): void {
    this.counter$
      .pipe(first())
      .subscribe((counter: number) => this.counter$.next(counter + 1));
  }

  decrement(): void {
    this.counter$
      .pipe(first())
      .subscribe((counter: number) => this.counter$.next(counter - 1));
  }
}

```

The first issue to address is the `increment()` and `decrement()` methods. The original implementation attempts to read the last emitted notification by subscribing with a `first()` pipe which completes the observable once a single notification is received. Further, this implementation is relying on a `ReplaySubject` in order to receive a cached value immediately after subscribing.

The simplest solution is to replace the `ReplaySubject` with a `BehaviorSubject` which includes an initial value and a `getValue` method that returns the cached value synchronously. This solution, however, misses the point. 

The current design doesn't consider the calling of the `increment()` and `decrement()` methods as notifications that affect the stream. For this reason, the calculation of the next notification falls outside of the stream. The stream is itself only being used to push a value to its observers. The best approach is to replace the `ReplaySubject` with a multi-casting observable.

The `publishBehavior(initialNotification)` operator will convert the observable into a `ConnectableObservable`. This `ConnectableObservable` will begin emitting notifications when its `connect()` method is called. To avoid this additional step, the `refCount()` operator can be added right after which will begin emitting notifications as soon as a single observer subscribes and will unsubscribe from the source if no observers remain subscribed. Further, it will re-subscribe to the source if a new observer subscribes and the source is has not finalized.

With a bit of refactoring,

`naive-counter.service.ts`

```typescript
import { Injectable } from "@angular/core";
import { Subject } from 'rxjs';
import { scan, publishBehavior, refCount } from "rxjs/operators";

@Injectable()
export class NaiveCounterService {
  private change$ = new Subject<number>();

  readonly counter$ = this.change$.pipe(
    scan((counter: number, change: number) => counter + change),
    publishBehavior(0),
    refCount()
  );

  increment(): void {
    this.change$.next(1);
  }

  decrement(): void {
    this.change$.next(-1);
  }
}


```

The above design has replaced the `ReplaySubject` with a `ConnectableObservable` that behaves similar to a `BehaviorSubject` in that it has an initial notification, specifically `0`, and the last notification is immediately emitted to each new subscriber. Subscriptions are managed automatically by the `refCount()` pipe. Yet, it receives all consecutive values from the observable stream.

The `increment()` and `decrement()` methods now act as proxy methods for dispatching notifications on the `change$` stream which is an `Subject`. Finally, the state of the `counter` is calculated with each notification from `change$` and the previously emitted notification.

One problem that might arise from this design is caused by the `refCount()` operator. If all observers unsubscribe, the subject will unsubscribe from the source. This is fine in most cases but one caveat is that there wont be a cached value for a new observer. The next observer to subscribe will have to wait until the source emits a new notification to receive its first notification. This could be a problem if the UI is trying to mount again with the last notification as its current state.

A `BehaviorSubject` and the `asObservable()` method provide an even better solution. Here is the latest iteration:

`naive-counter.service.ts`

```typescript
import { Injectable } from "@angular/core";
import { scan } from "rxjs/operators";
import { BehaviorSubject } from "rxjs";

@Injectable()
export class NaiveCounterService {
  private readonly change$ = new BehaviorSubject<number>(0);

  readonly counter$ = this.change$
    .asObservable()
    .pipe(scan((counter: number, change: number) => counter + change));

  increment(): void {
    this.change$.next(1);
  }

  decrement(): void {
    this.change$.next(-1);
  }
}

```

The above improvement relies on the `BehaviorSubject` to receive and cache new notifications. The `asObservable()` method returns an observable that conceals the `BehaviorSubject` so that external actors cannot push notifications into the stream. The stream logic is attached to the returned observable.