Skip to main content
HashnodeCodeCraft with Barcioch

Command Palette

Search for a command to run...

Angular Observable Unsubscription: Preventing Memory Leaks in Your Application

Updated
Angular Observable Unsubscription: Preventing Memory Leaks in Your Application
B
Bartosz Szłapak

A memory leak is a common problem unless dealt properly with the subscriptions. Whenever we subscribe to the observable a subscription is created. If we don't unsubscribe, then the subscription will stay in the memory and run whenever a new value is received. This can be a serious problem if we no longer need it or can reference it. That's why managing them is crucial for the stability and performance of the application. Since most of this article is related to Angular, all the examples will be also based on it.

Also, please note that I'll be dealing with cold observables only. Cold observables start emitting values only when subscribed to.

Setup

If unsure how to set up Angular with Jest please refer to the article: https://barcioch.pro/angular-with-jest-setup. Also, install the @ngneat/until-destroy library npm install @ngneat/until-destroy

Managing the subscriptions

unsubscribe method

The most basic way to unsubscribe is to keep track of the subscriptions and then unsubscribe when needed. Take a look at the following component.

import { Component, inject, OnDestroy, OnInit } from '@angular/core';
import { interval, Subscription, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-unsubscribe-example',
  template: '',
})
export class UnsubscribeExampleComponent implements OnInit, OnDestroy {
  readonly interval = 1000;
  private subscription: Subscription | undefined;
  private readonly logger = inject(LoggerService);

  ngOnDestroy(): void {
    this.subscription?.unsubscribe();
  }

  ngOnInit(): void {
    this.subscription = interval(this.interval).pipe(tap(value => {
      this.logger.log(`value:${ value }`);
    })).subscribe();
  }
}

  • when the component is initialized (ngOnInit)

    • inject LoggerService - does nothing, just used for testing purposes

    • subscribe to RxJS interval that emits values every 1 second

    • call the tap operator and call LoggerService.log method

    • store the subscription reference in this.subscription property

  • when the component is destroyed (ngOnDestroy)

    • unsubscribe by calling this.subscription?.unsubscribe()

The assumptions are

  • values should be emitted after component initialization by an interval of 1000 ms

  • after the component is destroyed no more values should be emitted

Test it

import { ComponentFixture, fakeAsync, TestBed, tick } from '@angular/core/testing';
import { UnsubscribeExampleComponent } from './unsubscribe-example.component';
import { LoggerService } from '../../services/logger.service';


describe('UnsubscribeExampleComponent', () => {
  let fixture: ComponentFixture<UnsubscribeExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [UnsubscribeExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(UnsubscribeExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    describe('and time passes by 1000ms and 5000ms after component destruction', () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(1);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
      }));
    });

    describe(`and time passes by 3000ms and 5000ms after component destruction`, () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval * 3);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(3);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
        expect(logger.log).toHaveBeenNthCalledWith(2, 'value:1');
        expect(logger.log).toHaveBeenNthCalledWith(3, 'value:2');
      }));
    });
  });
});

Parts of these tests require a bit of explanation.

To test intervals within Angular you have to

  • write all tests within fakeAsync zone (because all the periodic timers have to be started and canceled within it)

  • have some mechanism for canceling timers (or you'll get # periodic timer(s) still in the queue error) - in the current example I'm using ngOnDestroy to unsubscribe

The first calls to fixture.detectChanges() are inside it body. This is due to the interval starting within ngOnInit. You have to run change detection to run the lifecycle hooks.

Also, I'm calling tick(fixture.componentInstance.interval * 5) after fixture.destroy() to pass the time by 5 seconds after the component is destroyed. Only then I run the expectations to check the Logger.log method calls. This way I can validate that no value is emitted after the subscription was canceled.

RxJS operators

These operators are responsible for completing the observables but work differently. One of their advantages is that we no longer have to keep the subscription reference and manually call unsubscribe.

first

This operator without any argument emits the very first value and completes the observable.

import { Component, inject, OnInit } from '@angular/core';
import { first, interval, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-first-operator-example',
  template: '',
})
export class FirstOperatorExampleComponent implements OnInit {
  readonly interval = 1000;
  private readonly logger = inject(LoggerService);

  ngOnInit(): void {
    interval(this.interval).pipe(
      first(),
      tap(value => {
        this.logger.log(`value:${ value }`);
      })).subscribe();
  }
}

Test it

import { ComponentFixture, fakeAsync, TestBed, tick } from '@angular/core/testing';
import { FirstOperatorExampleComponent } from './first-operator-example.component';
import { LoggerService } from '../../services/logger.service';


describe('FirstOperatorExampleComponent', () => {
  let fixture: ComponentFixture<FirstOperatorExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [FirstOperatorExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(FirstOperatorExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    describe('and time passes by 1000ms and 5000ms after component destruction', () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(1);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
      }));
    });

    describe(`and time passes by 3000ms and 5000ms after component destruction`, () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval * 3);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(1);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
      }));
    });
  });
});

The test is almost identical to the previous one. Just changed the expected number of calls.

first(predicate)

The first argument is a predicate. The first time it evaluates to true the operator emits the value and completes the observable.

import { Component, inject, OnInit } from '@angular/core';
import { first, from, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-first-operator-with-predicate-example',
  template: '',
})
export class FirstOperatorWithPredicateExampleComponent implements OnInit {
  private readonly logger = inject(LoggerService);

  ngOnInit(): void {
    const source = [0, 1, 2, 3, 4, 5, 6, 7, 8];
    from(source).pipe(
      first((value: number) => value >= 4),
      tap(value => {
        this.logger.log(`value:${ value }`);
      })).subscribe();
  }
}

The from operator takes the source array of 9 numbers and emits them. The first value that is greater or equal to 4 is logged and the observable completes.

Test it

import { ComponentFixture, TestBed } from '@angular/core/testing';
import { FirstOperatorWithPredicateExampleComponent } from './first-operator-with-predicate-example.component';
import { LoggerService } from '../../services/logger.service';


describe('FirstOperatorWithPredicateExampleComponent', () => {
  let fixture: ComponentFixture<FirstOperatorWithPredicateExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [FirstOperatorWithPredicateExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(FirstOperatorWithPredicateExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should call logger', () => {
      fixture.detectChanges();
      fixture.destroy();
      expect(logger.log).toBeCalledTimes(1);
      expect(logger.log).toHaveBeenNthCalledWith(1, 'value:4');
    });
  });
});

This test creates, detects changes and destroys the component. After that, the expectations verify the logged value.

take

This operator required exactly one argument - the number of values to emit before the observable is closed.

import { Component, inject, OnInit } from '@angular/core';
import { first, from, take, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-take-operator-example',
  template: '',
})
export class TakeOperatorExampleComponent implements OnInit {
  private readonly logger = inject(LoggerService);

  ngOnInit(): void {
    const source = [0, 1, 2, 3, 4, 5, 6, 7, 8];
    from(source).pipe(
      take(3),
      tap(value => {
        this.logger.log(`value:${ value }`);
      })).subscribe();
  }
}

The from operator takes the source array of 9 numbers and emits them. The first 3 values (0, 1, 2) are logged and the observable completes.

Test it

import { ComponentFixture, TestBed } from '@angular/core/testing';
import { TakeOperatorExampleComponent } from './take-operator-example.component';
import { LoggerService } from '../../services/logger.service';


describe('TakeOperatorExampleComponent', () => {
  let fixture: ComponentFixture<TakeOperatorExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [TakeOperatorExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(TakeOperatorExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should call logger', () => {
      fixture.detectChanges();
      fixture.destroy();
      expect(logger.log).toBeCalledTimes(3);
      expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
      expect(logger.log).toHaveBeenNthCalledWith(2, 'value:1');
      expect(logger.log).toHaveBeenNthCalledWith(3, 'value:2');
    });
  });
});

This test creates, detects changes and destroys the component. After that, the expectations verify the logged values.

takeUntil

This operator requires exactly one argument to work - the notifier which is an observable. The operator will be emitting values until the notifier emits truthy value. When it happens, the observable completes.

import { Component, inject, OnDestroy, OnInit } from '@angular/core';
import { interval, Subject, takeUntil, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-take-until-operator-example',
  template: '',
})
export class TakeUntilOperatorExampleComponent implements OnInit, OnDestroy {
  readonly interval = 1000;
  private readonly logger = inject(LoggerService);
  private readonly unsubscribe$ = new Subject<void>();

  ngOnInit(): void {
    interval(this.interval).pipe(
      tap(value => {
        this.logger.log(`value:${ value }`);
      }),
      takeUntil(this.unsubscribe$)
    ).subscribe();
  }

  ngOnDestroy(): void {
    this.unsubscribe$.next();
    this.unsubscribe$.complete();
  }
}

In this example, I created a notifier private readonly unsubscribe$ = new Subject<void>(). The component implements the OnDestroy interface and, within ngOnDestroy method, the notifier emits a value this.unsubscribe$.next() and completes this.unsubscribe$.complete().

The main observable contains takeUntil(this.unsubscribe$) operator that takes the notifier. It's a good practice to place this operator on the last position of the operator list.

Test it

import { ComponentFixture, fakeAsync, TestBed, tick } from '@angular/core/testing';
import { TakeUntilOperatorExampleComponent } from './take-until-operator-example.component';
import { LoggerService } from '../../services/logger.service';


describe('TakeUntilOperatorExampleComponent', () => {
  let fixture: ComponentFixture<TakeUntilOperatorExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [TakeUntilOperatorExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(TakeUntilOperatorExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    describe('and time passes by 1000ms and 5000ms after component destruction', () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(1);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
      }));
    });

    describe(`and time passes by 3000ms and 5000ms after component destruction`, () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval * 3);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(3);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
        expect(logger.log).toHaveBeenNthCalledWith(2, 'value:1');
        expect(logger.log).toHaveBeenNthCalledWith(3, 'value:2');
      }));
    });
  });
});

As in previous examples, I'm verifying the calls to LoggerService. When the component gets destroyed there should be no more calls to the service.

takeWhile

This operator takes a predicate as an argument. As long as the predicate returns truthy value the operator emits the source value. Otherwise, it completes the observable.

import { Component, inject, OnInit } from '@angular/core';
import { interval, takeWhile, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-take-while-operator-example',
  template: '',
})
export class TakeWhileOperatorExampleComponent implements OnInit {
  readonly interval = 1000;
  private readonly logger = inject(LoggerService);

  ngOnInit(): void {
    interval(this.interval).pipe(
      takeWhile(value => value <= 3),
      tap(value => {
        this.logger.log(`value:${ value }`);
      }),
    ).subscribe();
  }
}

In this example, the values are emitted until the value is lesser or equal to 3.

Test it

import { ComponentFixture, fakeAsync, TestBed, tick } from '@angular/core/testing';
import { TakeWhileOperatorExampleComponent } from './take-while-operator-example.component';
import { LoggerService } from '../../services/logger.service';


describe('TakeWhileOperatorExampleComponent', () => {
  let fixture: ComponentFixture<TakeWhileOperatorExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [TakeWhileOperatorExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(TakeWhileOperatorExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    describe(`and time passes by 10000ms and 5000ms after component destruction`, () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval * 10);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(4);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
        expect(logger.log).toHaveBeenNthCalledWith(2, 'value:1');
        expect(logger.log).toHaveBeenNthCalledWith(3, 'value:2');
        expect(logger.log).toHaveBeenNthCalledWith(4, 'value:3');
      }));
    });
  });
});

find

This operator is similar to the native Array find method. It takes a predicate as an argument. When the predicate returns truthy value, the source value is emitted and the observable completes.

import { Component, inject, OnInit } from '@angular/core';
import { find, interval, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-find-operator-example',
  template: '',
})
export class FindOperatorExampleComponent implements OnInit {
  readonly interval = 1000;
  private readonly logger = inject(LoggerService);

  ngOnInit(): void {
    interval(this.interval).pipe(
      find(value => value === 4),
      tap(value => {
        this.logger.log(`value:${ value }`);
      }),
    ).subscribe();
  }
}

Test it

import { ComponentFixture, fakeAsync, TestBed, tick } from '@angular/core/testing';
import { FindOperatorExampleComponent } from './find-operator-example.component';
import { LoggerService } from '../../services/logger.service';


describe('FindOperatorExampleComponent', () => {
  let fixture: ComponentFixture<FindOperatorExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [FindOperatorExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(FindOperatorExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    describe(`and time passes by 10000ms and 5000ms after component destruction`, () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval * 10);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(1);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:4');
      }));
    });
  });
});

The test is expecting a single call to LoggerService with value:4 value.

find index

This operator is similar to the native Array findIndex method. It takes a predicate as an argument. When the predicate returns truthy value, the index is emitted and the observable completes.

import { Component, inject, OnInit } from '@angular/core';
import { findIndex, interval, map, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-find-index-operator-example',
  template: '',
})
export class FindIndexOperatorExampleComponent implements OnInit {
  readonly interval = 1000;
  private readonly logger = inject(LoggerService);

  ngOnInit(): void {
    interval(this.interval).pipe(
      map(value => value * 3),
      findIndex(value => value === 15),
      tap(value => {
        this.logger.log(`value:${ value }`);
      }),
    ).subscribe();
  }
}

In this example, the source value is multiplied by 3 to differentiate it from the value index.

Test it

import { ComponentFixture, fakeAsync, TestBed, tick } from '@angular/core/testing';
import { FindIndexOperatorExampleComponent } from './find-index-operator-example.component';
import { LoggerService } from '../../services/logger.service';


describe('FindIndexOperatorExampleComponent', () => {
  let fixture: ComponentFixture<FindIndexOperatorExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [FindIndexOperatorExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(FindIndexOperatorExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    describe(`and time passes by 10000ms and 5000ms after component destruction`, () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval * 10);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(1);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:5');
      }));
    });
  });
});

async pipe

The built-in Angular async pipe takes care of subscribing and unsubscribing. The unsubscription happens when the element gets destroyed.

import { Component, inject, OnInit } from '@angular/core';
import { first, interval, Observable, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';

@Component({
  selector: 'app-async-pipe-example',
  template: '<span class="value">{{ value$ | async }}</span>',
})
export class AsyncPipeExampleComponent implements OnInit {
  readonly interval = 1000;
  private readonly logger = inject(LoggerService);
  value$: Observable<number> | undefined;

  ngOnInit(): void {
    this.value$ = interval(this.interval).pipe(
      tap(value => {
        this.logger.log(`value:${ value }`);
      }));
  }
}

In this example, I created value$ observable that starts emitting values as soon as the async pipe subscribes. I'm also using the interpolation, so the emitted value will be rendered within the span.

Test it

import { ComponentFixture, fakeAsync, TestBed, tick } from '@angular/core/testing';
import { AsyncPipeExampleComponent } from './async-pipe-example.component';
import { LoggerService } from '../../services/logger.service';
import { By } from '@angular/platform-browser';


describe('AsyncPipeExampleComponent', () => {
  let fixture: ComponentFixture<AsyncPipeExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [AsyncPipeExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(AsyncPipeExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    it('should not display any value', () => {
      expect(getValue()).toBe('');
    });

    describe('and time passes by 1000ms and 5000ms after component destruction', () => {
      it('should call logger and display value', fakeAsync(() => {
        fixture.detectChanges();
        expect(getValue()).toBe('');

        tick(fixture.componentInstance.interval);
        fixture.detectChanges();
        expect(getValue()).toBe('0');

        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(1);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
      }));
    });

    describe(`and time passes by 3000ms and 5000ms after component destruction`, () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        expect(getValue()).toBe('');
        tick(fixture.componentInstance.interval * 3);
        fixture.detectChanges();
        expect(getValue()).toBe('2');

        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(3);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
        expect(logger.log).toHaveBeenNthCalledWith(2, 'value:1');
        expect(logger.log).toHaveBeenNthCalledWith(3, 'value:2');
      }));
    });
  });

  const getValue = (): string | undefined  => {
    return fixture.debugElement.query(By.css('.value'))?.nativeElement.textContent;
  }
});

In addition to previous checks of the LoggerService calls I'm also verifying the span's content (identified by value css class).

@ngneat/until-destroy library

The creators of this library call it a neat way to unsubscribe from observables when the component destroyed. And that's exactly what it does.

import { Component, inject, OnInit } from '@angular/core';
import { interval, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';
import { UntilDestroy, untilDestroyed } from '@ngneat/until-destroy';

@UntilDestroy()
@Component({
  selector: 'app-ngneat-until-destroy-example',
  template: '',
})
export class NgneatUntilDestroyExampleComponent implements OnInit {
  readonly interval = 1000;
  private readonly logger = inject(LoggerService);

  ngOnInit(): void {
    interval(this.interval).pipe(
      tap(value => {
        this.logger.log(`value:${ value }`);
      }),
      untilDestroyed(this)
    ).subscribe();
  }
}

The first thing to do is to decorate the component with @UntilDestroy decorator. It has to be done before the @Component decorator to work properly. Next, when using an observable just use the untilDestroy operator and pass this reference untilDestroyed(this).

The @UntilDestroy decorator accepts arguments and allows for more control over the unsubscribing (i.e. can automatically unsubscribe from properties) but it's out of this article's scope.

Test it

import { ComponentFixture, fakeAsync, TestBed, tick } from '@angular/core/testing';
import { NgneatUntilDestroyExampleComponent } from './ngneat-until-destroy-example.component';
import { LoggerService } from '../../services/logger.service';


describe('NgneatUntilDestroyExampleComponent', () => {
  let fixture: ComponentFixture<NgneatUntilDestroyExampleComponent>;
  let logger: LoggerService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [NgneatUntilDestroyExampleComponent]
    })

    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(NgneatUntilDestroyExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    describe('and time passes by 1000ms and 5000ms after component destruction', () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(1);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
      }));
    });

    describe(`and time passes by 3000ms and 5000ms after component destruction`, () => {
      it('should call logger', fakeAsync(() => {
        fixture.detectChanges();
        tick(fixture.componentInstance.interval * 3);
        fixture.destroy();
        tick(fixture.componentInstance.interval * 5);
        expect(logger.log).toBeCalledTimes(3);
        expect(logger.log).toHaveBeenNthCalledWith(1, 'value:0');
        expect(logger.log).toHaveBeenNthCalledWith(2, 'value:1');
        expect(logger.log).toHaveBeenNthCalledWith(3, 'value:2');
      }));
    });
  });
});

A common pitfall

The RxJS operators like take, first, takeWhile, find, findIndex should be always used with a safety belt like takeUntil or untilDestroyed operator. If the component gets destroyed, and you assume, that the observable will always emit at least one value, but it doesn't happen, the subscription will live in the memory. That's the memory leak.

Create a test service

import { Subject } from 'rxjs';
import { Injectable } from '@angular/core';

@Injectable()
export class MyService {
  private values$$ = new Subject<number>();
  value$ = this.values$$.asObservable();

  sendValue(value: number): void {
    this.values$$.next(value);
  }
}

and a component

import { Component, inject, OnInit } from '@angular/core';
import { first, tap } from 'rxjs';
import { LoggerService } from '../../services/logger.service';
import { MyService } from './my-service';

@Component({
  selector: 'app-pitfall-example',
  template: '',
})
export class PitfallExampleComponent implements OnInit {
  private readonly logger = inject(LoggerService);
  private readonly myService = inject(MyService);

  ngOnInit(): void {
    this.myService.value$.pipe(
      first(value => value < 0),
      tap(value => this.logger.log(`value:${ value }`)),
    ).subscribe();
  }
}

In this example, I'm subscribing to MySerice.value$ observable that emits passed numbers. I used first operator first(value => value < 0) that takes the first value lesser than 0. Now, when the component gets destroyed, this subscription will stay in the memory since there were no circumstances to unsubscribe.

Test it

import { ComponentFixture, TestBed } from '@angular/core/testing';
import { PitfallExampleComponent } from './pitfall-example.component';
import { LoggerService } from '../../services/logger.service';
import { MyService } from './my-service';


describe('PitfallExampleComponent', () => {
  let fixture: ComponentFixture<PitfallExampleComponent>;
  let logger: LoggerService;
  let myService: MyService;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [PitfallExampleComponent],
      providers: [MyService]
    })

    myService = TestBed.inject(MyService);
    logger = TestBed.inject(LoggerService);
    jest.spyOn(logger, 'log');
    fixture = TestBed.createComponent(PitfallExampleComponent);
  });

  describe('when component is initialized', () => {
    it('should not call logger', () => {
      fixture.detectChanges()
      expect(logger.log).toBeCalledTimes(0);
    });

    describe('and the value "1" is passed to service', () => {
      beforeEach(() => {
        myService.sendValue(1);
        fixture.detectChanges();
      });

      it('should not call logger', () => {
        expect(logger.log).toBeCalledTimes(0);
      });

      describe('and component gets destroyed', () => {
        beforeEach(() => {
          fixture.destroy();
          fixture.detectChanges();
        });

        it('should not call logger', () => {
          expect(logger.log).toBeCalledTimes(0);
        });

        describe('and the value "-2" is passed to the service', () => {
          beforeEach(() => {
            myService.sendValue(-2);
            fixture.detectChanges();
          });

          it('should call logger', () => {
            expect(logger.log).toBeCalledTimes(1);
            expect(logger.log).toHaveBeenNthCalledWith(1, 'value:-2');
          });
        });
      });
    });
  });
});

The test verifies the memory leak by the following steps

  • initializes component

  • sends value 1

  • verifies that Logger.log was not called

  • destroys the component

  • sends value -2

  • verifies that Logger.log was called once with value:-2 argument

To fix that problem you could use takeUntil or untilDestroyed. The first one has to be somehow tied to ngOnDestroy hook (like in the example). The second requires just annotating the component. Nevertheless, the memory leak is a common problem and should be always kept in mind.

Source code

https://gitlab.com/barcioch-blog-examples/012-angular-rxjs-unsubscribing

#angular#observables#memory-leak#subscriptions

More from this blog

CodeCraft with Barcioch

24 posts