Angular 7 - Am I creating too many subscriptions?

Question

I would like to know if my code could create memory leaks?

Context

I have a component class that should display 'Applications' objects. It has filtering and pagination capacities.

I created a method loadAppsData() in which I subscribe to an Observable returned after a request to a web service.

This method is called at initialization time, ngOnInit(), or after the user interact with the filtering input field or the paginator (see method onUserInteractionsWithTree() )

My question

To avoid memory leaks I already use

.pipe(takeUntil(this.ngUnsubscribe))

and

ngOnDestroy(): void {
  this.ngUnsubscribe.next(); //  Unsubscribe from observables.
  this.ngUnsubscribe.complete(); // Unsubscribe from ngUnsubscribe.
}

But it seems to me that I create new Subscription object each time I send a request to the server, when I call the subscribe() method. Can that create memory leaks? Should I try to reuse subscriptions object?

Thanks in advance for your help,

Below the Typescript code of my component

import {Component, OnInit, ViewChild, ElementRef, OnDestroy} from '@angular/core';
import {FlatTreeControl} from '@angular/cdk/tree';

import {MatPaginator} from '@angular/material';

import {DynamicFlatNode} from './dynamic-flat-node';
import {ApplicationService} from '../shared/service/application-service';
import {DataRequestOptions} from '../../shared/data/data-request-options';
import {MetaDescriptor} from '../../shared/data/meta/meta-descriptor';
import {TableDataRequestParamsService} from '../../shared/data/table-data-request-params.service';


import {ApplicationTreeDatabase} from './application-tree-database';
import {ApplicationTreeDatasource} from './application-tree-datasource';

// Observable classes and extensions.
import {BehaviorSubject, Subject, fromEvent, of, merge} from 'rxjs';
// Observable operators.
import {debounceTime, distinctUntilChanged, switchMap, takeUntil} from 'rxjs/operators';


@Component({
  selector: 'app-application-tree',
  templateUrl: './application-tree.component.html',
  styleUrls: ['./application-tree.component.css'],
  providers: [ApplicationTreeDatabase]
})
export class ApplicationTreeComponent implements OnInit, OnDestroy {

  @ViewChild('appfilter') inputfilter: ElementRef;
  @ViewChild(MatPaginator) paginator: MatPaginator;

  readonly defaultPaginatorPageIndex = 0;
  readonly defaultPaginatorPageSize = 2;
  readonly defaultPaginatorPageRange = this.defaultPaginatorPageIndex + '-' + (this.defaultPaginatorPageSize - 1);


  private ngUnsubscribe: Subject<void> = new Subject<void>();

  // Application name filter. START
  _inputFilterChange = new BehaviorSubject('');
  get inputFilterValue(): string {
    return this._inputFilterChange.value;
  }
  set inputFilterValue(inputFilterValue: string) {
    this._inputFilterChange.next(inputFilterValue);
  }
  // Application name filter. END

  treeControl: FlatTreeControl<DynamicFlatNode>;

  dataSource: ApplicationTreeDatasource;

  getLevel = (node: DynamicFlatNode) => node.level;

  isExpandable = (node: DynamicFlatNode) => node.expandable;

  hasChild = (_: number, _nodeData: DynamicFlatNode) => _nodeData.expandable;


  constructor(
    private applicationService: ApplicationService,
    private dataRequestHelper: TableDataRequestParamsService,
    private database: ApplicationTreeDatabase) {
    this.treeControl = new FlatTreeControl<DynamicFlatNode>(this.getLevel, this.isExpandable);
    this.dataSource = new ApplicationTreeDatasource(this.treeControl, this.paginator, database);
  }


  ngOnInit(): void {
    fromEvent(this.inputfilter.nativeElement, 'keyup').pipe(
      debounceTime(150)
      , distinctUntilChanged()
      , switchMap(term => of(term))
      , takeUntil(this.ngUnsubscribe)
    )
      .subscribe(() => {
        if (!this.dataSource) {
          return;
        }
        // this.resetPaginator();
        this.inputFilterValue = this.inputfilter.nativeElement.value;
      });
    this.loadAppsData();
    this.onUserInteractionsWithTree();
  }
  ngOnDestroy(): void {
    this.ngUnsubscribe.next(); //  Unsubscribe from observables.
    this.ngUnsubscribe.complete(); // Unsubscribe from ngUnsubscribe.
  }

  resetFilterAndTriggerChange() {
    // Clear HTML filter content.
    this.inputfilter.nativeElement.value = '';
    // Clear filter data stream. => This will trigger database.load()
    // because of Event emmited by inputFilterValueChange.
    this.inputFilterValue = '';
  }

  buildAppDataRequestParams(): DataRequestOptions {
    let range = this.dataRequestHelper.buildRequestRangeValue(this.paginator);
    if (!range) { // paginator not initialized.
      range = this.defaultPaginatorPageRange;
    }
    return new DataRequestOptions(this.inputFilterValue, 'name', range);
  }

  private loadAppsData() {
    this.applicationService.getDataObjects(this.buildAppDataRequestParams())
      .pipe(takeUntil(this.ngUnsubscribe))
      .subscribe(dataAndMeta => {
        // Update local Apps database.
        this.database.updateApplicationData(dataAndMeta.data);
        this.updatePaginator(dataAndMeta.meta);
        // Inform datasource that data has changed.
        this.dataSource.data = this.database.getAppsAsRootLevelNodes();
      },
      error => {
        const errMsg = 'Echec d\'acces aux données';
        throw new Error(errMsg);
      }
      );
  }

  private onUserInteractionsWithTree() {
    const treeUserActionsListener = [
      this._inputFilterChange,
      this.paginator.page
    ];
    // Merge the array of Observable inputs of treeUserActionsListener
    // and put into the source property of a newly created Observable.
    const mergeOfObservables = merge(...treeUserActionsListener);
    // Create new Observable<RoleMemberClient[]> by calling the function defined below.
    mergeOfObservables
      .pipe(takeUntil(this.ngUnsubscribe))
      .subscribe((data: any) => {
        this.loadAppsData();
      });
  }

  private updatePaginator(meta: MetaDescriptor) {
    if ((meta) && (meta.isPaginatedData)) {
      const contentRange = meta.contentRange;
      const rangeStart = contentRange.rangeStart;
      this.paginator.pageIndex = Math.floor(rangeStart / this.paginator.pageSize);
      this.paginator.length = contentRange.size;
    } else if (meta) {
      // All data can be contained within the first table page.

      this.paginator.length = meta.count;
      if (this.paginator.pageIndex * this.paginator.pageSize < meta.count) {
        // If last requested page do not contain data, do not reset table page index.
        // The user will do it by itself.

        // Otherwise reset the table page index to zero.
        this.paginator.pageIndex = 0;
      }

    }
  }

}

Answer 1

Memory Leak :

Observable pattern is prone to memory leak because a subscription that persists after the component (in this context) death, will persist during application's lifetime.

For example : Let's say you have a component that subscribes to a formControl when the component is created but never close the subscription, each time you create the component you create a new subscription. You have a leak, yourmight overload the memory.

Closing subscription :

A subscription ends when the observable completes or you manually unsubscribe from it.
You have chosen to create a Subject (you called it ngUnsubscribe - which is a very bad name). You complete() the subject when the component is destroyed.
It means that every subscription to that subject will be closed when the component is destroyed.
When you subscribe, you use takeUntil(ngUnsubscribe), in fact, you create a mirror of your original observable and subscribe on this mirror.
As a consequence, when your component is destroyed, every subscriptions made on your mirror (ngUnsubscribe), is destroyed. So no, you don't have memory leaks.

Notes :
Because a subscription is closed when observable completes, you don't need to create observable mirror / unsubscribe from methods that complete your obervable such as angular HttpClient(get, post, ...).

You can find all these informations on the link provided in comment by Alex Beugnet.
For a better understanding, you can check :

rxmarbles
learn-rxjs

Answer 2

What you are doing is ok but it is, but ideally you should subscribe using the | async pipe inside your template as angular will then handle all of the subscribe and unsubscribe for you. Orphan subscriptions will cause memory leaks in your application.

So in your component do this

ngOnInit() {
           this.dataObjects = this.applicationService.getDataObjects(this.buildAppDataRequestParams());
        }

and in your template

<ng-container *ngFor="let dataObject of dataObjects | async">
    // Your html markup for each dataObject here 
    {{dataObject | json}}
 </ng-container>

If you want to do complex mapping or events then you should consider using BehaviourSubjects then something along the lines of

getDataObjects.toPromise().then(r => 
{
  dataObjectSubject.next(r.map(i => // mapping here))
})

And subscribe to the BehaviourSubject in the template the same way with the |async pipe.

Hope that helps

Answer 3

Big code snippet, few things I noticed :), but it's hard to process whole peace.

1.

this.applicationService.getDataObjects(this.buildAppDataRequestParams())
      .pipe(takeUntil(this.ngUnsubscribe))

It looks like you don't need pipe(takeUntil(this.ngUnsubscribe)) here, because getDataObjects seems like xmlhttprequest. So it's completed immediately.

2. switchMap(term => of(term)) -- looks bad for me. You create stream and flatten it immediately, I guess main purpose is to kill stream when new value comes, but doesn't look good.

3. .subscribe((data: any) => { this.loadAppsData();}); -- In a subscribe you call another function that as well subscribes to another stream, makes it hard to follow.

I would make your component more events driven.

Split you logic to when you need to trigger data, what need to happen when new data is arrived. Subscribe once.

idea:

this.data$ = merge(componentInit$, inputChanged$, nextPage$).pipe(
   switchMap(() => loadData()/* should return stream, not subscription */)
)
//...

 this.data$.pipe(takeUntil(/*bablba*/), map(/* transformations */)).subscribe(() => {
    //actions with data
 })

Answer 4

First of all,

this.ngUnsubscribe.next(); //  Unsubscribe from observables.

next() NOT unsuscribe from observables. It is use to send data through the observable.

To receive this data you need to have been subscribed to your "ngUnsuscribe" (Bad name because it would be generates missunderstoods) subject before you call function next.

So the correct flow of a subject is

1. Create the subject (private subjectName: Subject = new Subject();)
2. Subscribe to this subject. (this.subjectName.subscribe())
3. Send data through this subject. (this.subjectName.next("Test"))
4. Close the subject after using it. (this.subjectName.complete())

If you close the subject in the next line you sent data, it could generates problems cause of the asynchronous nature of subject communication.