Understanding the Challenge of Nested Observables in Arrays

RxJS is a widely adopted library designed to manage asynchronous event streams. While it offers considerable power, its learning curve is steep, and it can present tricky situations, especially for newcomers.

One particularly tricky scenario involves having a stream that emits an array of objects, where a property of each object requires an additional asynchronous operation. When attempting to fetch these sub-values, you often end up with a structure like Observable<Array<Observable<T>>>. Subscribing to this correctly can be awkward. We will look at various strategies to handle this situation, ultimately transforming our data into an Observable<Observable<Array<T>>> stream.

This guide is also the solution for challenge #11 from Angular Challenges. While it's aimed at developers with a solid grasp of observables, anyone can follow along to learn the concepts. If you want to test your skills first, attempt the challenge on your own before comparing your approach with the one detailed here. (You can also submit a PR for review))

Let's illustrate the core issue with a simple use case.

readonly persons$ = this.service.get(selectPersons);
         // ^? Observable<Persons[]> where Person = {id: number, name: string}
Enter fullscreen mode Exit fullscreen mode

We begin with a persons$ observable that emits a list of persons, likely from a store. This could be NgRx, RxAngular, or a simple service backed by a Subject. In this example, the specific implementation is abstracted by this.service.

Our goal is to enrich each person in the array by fetching their address from the backend, then return the complete list with addresses attached.

Imperative approach:

Frequently, I see codebases handling this with an imperative style, using nested subscribe callbacks. The code often looks like this:

let personWithAddress: PersonWithAddress[];

this.persons$.subscribe((persons) => {
  persons.forEach((person) => {
    this.http.getAddress(person.id).subscribe((address) => {
      personWithAddress.push({...person, ...address});
    });
  });
});
Enter fullscreen mode Exit fullscreen mode

While this logic will execute correctly, it is generally discouraged. The resulting nested subscribes make the code hard to follow and maintain, even in this simple example.

Naive approach with Rxjs:

Let's attempt a more reactive solution by using the standard rxjs operators.

When I introduce this type of problem, I frequently see students attempting to use mergeMap or switchMap to manage the inner observables. This code is a common attempt:

 personWithAddress$ = this.persons$.pipe(
  // ?^ Observable<Observable<PersonWithAddress>>
    mergeMap((persons) => 
        persons.map((p) =>
                //^ map is not a rxjs operator but a Array function
            this.http.getAddress(p.id).pipe(map(address => ({...p, ...address})))))
                        // ^return an observable we never subscribe to  
);
Enter fullscreen mode Exit fullscreen mode

The problem is that persons.map creates an array of observables. The higher-order mapping operators like mergeMap or switchMap are designed to flatten an Observable<Observable<T>>, not an Array<Observable<T>>. Since you're providing an array, these operators won't work as intended.

Forkjoin:

Another popular operator for handling this is forkJoin. It is specifically designed to handle a collection of observables and emit their last values, but only after all of them have completed. This is a crucial distinction, as forkJoin will stay silent until every single source observable has finished.

Here is a working solution that uses forkJoin:

  personWithAddress$ = this.persons$.pipe(
    mergeMap((persons) =>
      forkJoin(
        persons.map((p) =>
          this.http.getAddress(p.id).pipe(map((address) => ({ ...p, ...address })))
        )
      )
    )
  );
Enter fullscreen mode Exit fullscreen mode

The solution does work. However, the significant level of nesting needed to handle each address and spread the results can make the code quite verbose and difficult to trace during debugging.

The final trick:

One of the most valuable tips for managing streams of object arrays is to flatten the structure first so you are working with a stream of individual objects. Operating on arrays often adds unnecessary complexity. By contrast, we are much more comfortable working with single objects in a stream.

The key operators to accomplish this are mergeAll and toArray. The mergeAll operator will flatten an outer observable that emits inner observables (like an array) into a single stream of emissions. Once that inner stream is complete, the toArray operator collects all the emitted items and emits them as a single array.

Here is how these operators solve our problem in practice:

  personWithAddress$ = this.persons$.pipe(
    mergeAll(), // flatten to Observable<Person>
    mergeMap((p) =>
      this.getAddress(p.id).pipe(map((address) => ({ ...p, ...address })))
    ),
    toArray() // back to Observable<PersonWithAddress[]>
  );
Enter fullscreen mode Exit fullscreen mode

Key observations:

  • The nesting is kept to a single, manageable level.
  • It is cleaner and simpler to reason about individual object streams than arrays of objects.

I hope you find this technique helpful for your rxjs projects and that it improves your reactive programming skills. 🚀

You can connect with me on Twitter or Github. Feel free to ask any questions.