Handling parallel nested observables in RxJS

Question

I'm working on cleaning up nested parallel observables in my project and I am having some issues wrapping my head around handling the observable events. Here's one situation in my music application where I load the Playlist page and need to load the playlist data, information about the user, and if the user is following the playlist:

Route Observable (returns parameters from URL, I need to extract the playlistId)
- Playlist Observable (needs playlistId and returns PlaylistObject and resolves)
- User Observable (returns UserObject)
  - IsUserFollowing Observable (needs playlistId and UserObject.id and returns T/F if user is following the playlist)

In summary, I receive the route parameter from Angular's Route observable, I need to fire two other observables in parallel, the Playlist Observable and User Observable. The Playlist Observable can resolve once the playlist data is returned, but the User Observable needs to continue on to another observable, IsFollowingUser, until it can resolve.

Here is my current (bad) implementation with nested observables:

this.route.params.pipe(
  first()
).subscribe((params: Params) => {

  this.playlistId = parseInt(params["playlistId"]);

  if (this.playlistId) {

    this.playlistGQL.watch({ 
      id: this.playlistId,
    }).valueChanges.pipe(
      takeUntil(this.ngUnsubscribe)
    ).subscribe((response) => { 
      this.playlist = response; // use this to display playlist information on HTML page
      this.playlistTracks = response.tracks; //loads tracks into child component
    });

    this.auth.user$.pipe(
      filter(stream => stream != null)
    ).subscribe((user) => {
      this.user = user;
      this.userService.isFollowingPlaylist(this.user.id, this.playlistId).subscribe((response) => {
        this.following = response; //returns a T/F value that I use to display a follow button
      })
    })

  }

This works but obviously does not follow RxJS best practices. What is the best way to clean up nested parallel observables like this?

AliF50 AliF50 · Accepted Answer · 2020-01-04T01:58:47

This is something I would do although this may not be the most optimal but could be better than before.

I will assume the previous code block is in an ngOnInit.

async ngOnInit() {
  // get the params as if it is a promise
  const params: Params = await this.route.params.pipe(
                                      first()
                                    ).toPromise();

  this.playlistId = parseInt(params["playlistId"]);
  if (this.playlistId) {
    // can name these functions to whatever you like (better names)
    this.reactToPlaylistGQL();
    this.reactToUser();
  }
}

private reactToPlaylistGQL() {
  this.playListGQL.watch({
    id: this.playListId,
  }).valueChanges.pipe(
   takeUntil(this.ngUnsubscribe)
 ).subscribe(response => {
   this.playlist = response;
   this.playlistTracks = response.tracks;
 });
}

private reactToUser() {
  this.auth.user$.pipe(
    filter(stream => stream != null),
    // switch over to the isFollowingPlaylist observable
    switchMap(user => {
     this.user = user;
     return this.userService.isFollowingPlaylist(this.user.id, this.playListId);
   }),
    // consider putting a takeUntil or something of that like to unsubscribe from this stream
  ).subscribe(response => this.following = response);
}

I find naming the functions in the if (this.playlistId) block to human friendly names makes it unnecessary to write comments (the descriptive function name says what it should do) and looks cleaner than putting a blob of an RxJS stream.

I wrote this all freehand so there might be some typos.

Handling parallel nested observables in RxJS

1 Answers