Notify when Sequence of Observables Complete

Question

I'm trying to use RxJS Observables to filter a list of objects based on a users selection. To get the list of filtered objects, I need to call three separate services in sequence and pass in the values from the previous service to the subsequent service.

NOTE 1: Each of these three services return Observable

NOTE 2: Each service can return an arbitrary number of results, which each need to get passed to the next service.

Only after the third service has completed emitting all of its values can my code continue normally.

Here is what I have right now:

this.firstService.getFirstResults(filterOption)
            .subscribe(firstResult => this.secondService.getSecondResults(firstResult)
                .subscribe(secondResult => this.thirdService.getThirdResults(secondResult)
                    .subscribe(thirdResult => this.processResult(thirdResult, firstResult),
                    null,
                    () => console.log("COMPLETE")))
            );

The above code works almost perfectly for me. In the end, the processResult() function correctly builds an Array with all the filtered objects.

However, I don't know how to get notified when the sequence of Observables has truly completed. I had hoped that the complete section of the third service would do the job, but it is printing to the console several times rather than just once.

NOTE 3: In the final call to processResult() I need to pass both the thirdResult value as well as the corresponding firstResult value returned from the firstService.getFirstResults().

Use the switchMap() or mergeMap() or concatMap() operator (depending on what you really want). How could they run in parallel since you need the value emitted by the first one in order to call the second one? — JB Nizet
Sorry, parallel was a bad word choice. I think I meant to say async. Just edited post to correct. — JDuffy
Could you provide a code snippet of how I would convert my code above to use either a switchMap or mergeMap? I have tried this myself thinking it would fix my issue but the final filter Array was never built and the COMPLETE console log was never printed. When looking for example code online, most seem to demonstrate function sequences that take no parameters or static ones. — JDuffy
gist.github.com/jnizet/aeb37e62a059c99f2bb43f19d0b82e4a. But your observable will only complete if all the observables complete (in case of mergeMap) — JB Nizet
Thanks for the snippet as I seem to be making progress now. My only remaining issue it seems is that I mistakenly put the filterOption variable as the second parameter to the processResult() function, when in fact it should be the value returned by the firstService e.g ``processResult(thirdResult, firstResult). Though with the new code setup, the firstResult` variable is not visible once that function is called. Any ideas? — JDuffy

t7yang t7yang · Accepted Answer · 2018-12-13T05:37:57

switchMap provided second parameter to restructure the return value, below is the demo show how to do.

function getResult(filterOption) {
  return this.firstService.getFirstResult(filterOption).pipe(
    switchMap(
      firstResult => this.secondeService.getSecondResult(firstResult),
      ([firstResult, secondResult]) => ({ firstResult, secondResult }),
    ),
    switchMap(
      ({ secondResult }) => this.thirdService.getThirdResult(secondResult),
      ([{ firstResult }, thirdResult]) => ({ firstResult, thirdResult }),
    ),
  );
}

getResult(filterOption).subscribe({
  next: ({ firstResult, thirdResult }) => this.processResult(thirdResult, firstResult),
  completed: () => console.log('completed'),
});

Notify when Sequence of Observables Complete

1 Answers