Rewriting RxJS Nested Subscriptions to Avoid Memory Leaks

Even if you pay attention to cleaning up your subscriptions when components are destroyed, there is still the potential for memory leaks from nested RxJS subscriptions. In fact, there are other problems with nested subscriptions as well, such as timing issues. For these reasons, nested subscriptions have earned the top spot in the list of RxJS anti-patterns. Many developers still use nested subscriptions simply because they don’t know how to refactor them to avoid nesting. It’s a complex problem, one for which there is no one-size-fits-all solution. Indeed, the road to better RxJS subscription handling is to study numerous examples of “correct” usage. Unfortunately, good examples are in short supply. Therefore, we’ll add one more to the pile in this tutorial. In this case, we’ll take some real-life code (albeit rewritten slightly to be more generic) whose nested subscriptions result from conditions and rewrite it without the nesting.

Code Walkthrough

The code that we’ll be looking at here today originates from a real application that collects news articles, social media posts, and individual blogs published online about stocks and other investment instruments. The application then provides analytics that enables investors to gain a reliable view of the crowd‘s impact on their holdings.

The code in question fetches the data for the dashboard screen, pictured above:

this.userInfoService.userPermissions.subscribe((permissions) => {
  this.isLimited = permissions.preLogin;
  if (!this.isLimited) {
    this.trackerService.recordPageView(PageViewCategory.dashboard);

    this.dashboardService
      .takeUntil(this.ngUnsubscribe)
      .subscribe((e) => {
        const id = e.entityId;
        if (id) {
          this.apiService.getArticleStats(id).subscribe(
            (article) => {
              if (article != null) {
                this.entity = article.entity;
                // Set current entity for dashboard page
                this.dashboardService.setCurrentEntity(article.entity);
                if (this.entity?.instrumentId) {
                  this.loadInstrumentData(this.entity.instrumentId);
                } 
              }
            }
          );
        }
      });

There are three IF statements that determine whether or not to subscribe to another service:

1. If the user’s access is not limited
2. If the entity (e) has an entityId
3. If the article has an associated entity

Eliminating the IF Statements

The control flow of the above code is not unlike this example that I came across on stackoverflow.com, which also subscribed to multiple services based on the result of an IF statement:

source.subscribe(x => {
    doAnyway(x);
    if (x.Id) {             
        doSometing(x);
    } else {
        // id Not set, get default Id
        this.idService.getDefault().subscribe(id => {
            x.Id = id;                  
            doSometing(x);
        });
    }
});

In that instance, a respondent suggested the following structure that employs the RxJS tap() and switchMap() operators to replace the above code structure:

source.pipe(
  tap(val => doAnyway(val)),
  switchMap(val => val.id ? of(val.id) : this.idService.getDefault())
).subscribe(id => {
  this.id = id;
  doSomething(id);
});

As it turns out, these two operators will work just as well for us!

RxJS tap() and switchMap() Explained

Like its name implies, tap() perform a side effect for every emission on the source Observable, and returns an Observable that is identical to the source. Hence, it’s perfect for doing something with an Observable where you don’t want to alter it in any way.

We can use tap() to set the isLimited flag as follows:

this.userInfoService.userPermissions
    .pipe(
      takeUntil(this.ngUnsubscribe),
      tap(permissions => this.isLimited = permissions.preLogin)
    )
    .subscribe((article) => {
      // etc...
    });

Moving on to switchMap(), it’s one of numerous RxJS mapping functions that each work a little differently. The switchMap() function creates a derived observable (called an inner observable) from a source observable and emits those values. Whenever the source emits a new value, switchMap() will create a new inner observable and switch to those values instead. Inner observables that get created on the fly get unsubscribed from, leaving the source observable open to emit more values.

The trick to using mapping functions is that we need to return the next observable in the chain. In our case, we need to pass the dashboardService along for the next step. If the user has limited access, we can return an empty DashboardParams object and create a new Observable from it using the RxJS of() function:

this.userInfoService.userPermissions
    .pipe(
      takeUntil(this.ngUnsubscribe),
      tap(permissions => this.isLimited = permissions.preLogin)
      ),
      switchMap(() => {
        if (this.isLimited) {
          return of(new DashboardParams());
        } else {
          this.trackerService.recordPageView(PageViewCategory.dashboard);
  
          return this.dashboardService
        }
      })
    )
    .subscribe((article) => {
      // etc...
    });

We still need to call the apiService to get the article data, so we now need to process the DashboardParams and return the apiService.getArticleStats’s emitted Observable. Again, switchMap() is the function to do it. If the emitted Entity (e) has an entityId, we can then proceed to call apiService.getArticleStats(). Otherwise, we can again return an Observable that emits an empty ArticleData object, which is apiService.getArticleStats()‘ emitted type. Finally, the article data is processed in the subscribe() handler:

this.userInfoService.userPermissions
  .pipe(
    takeUntil(this.ngUnsubscribe),
    tap((permissions) => {
      this.isLimited = permissions.preLogin;
    }),
    switchMap(() => {
      if (this.isLimited) {
        return of(new DashboardParams());
      } else {
        this.trackerService.recordPageView(PageViewCategory.dashboard);

        return this.dashboardService
      }
    }),
    switchMap((e: DashboardParams) => {
      const id = e.entityId;
      return id 
        ? this.apiService.getArticleStats(id) 
        : of(<ArticleData>{})
    })
  )
  .subscribe((article) => {
    if (article != null) {
      this.entity = article.entity;
      // Set current entity for dashboard page
      this.dashboardService.setCurrentEntity(article.entity);
      if (this.entity.instrumentId) {
        this.loadInstrumentData(this.entity.instrumentId);
      } 
    }
  });

Conclusion

With about a zillion operators, RxJS is undoubtedly the most powerful tool for working with asynchronous or callback-based code. It’s also the most complicated. As mentioned in the intro, refactoring nested subscriptions is not easy, as each case must be assessed according to its individual requisites. Hopefully, the example presented here today will help structure your own code in a more correct and efficient manner.