How to combine observables in a non-blocking way?

user_7374703 0
Viewed 709

I have a collection of Observables that retrieve different data types each. I'm chaining those Observable to get all the data I want for my page. Fact is all those informations are independent, so loading one should not block or interfere in loading the others. which I am not able to achieve.

Here is an example of what I've done for the moment:

  getAll(): Observable<any[]> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      mergeMap((home: Home) =>
        combineLatest([
          this.getAssets(home.images).pipe(defaultIfEmpty([])),
          this.getStyling(home.styling).pipe(defaultIfEmpty({})),
          this.getCategories(home.categories).pipe(defaultIfEmpty([])),
        ])
      )
    );
  }

Here, the Observable are successfully chained, so I get all the data I want, but combineLatest() does the next request once the previous is completed, and parent emits collected data once everything is finished which creates delay. I've been trying to use merge() instead of combineLatest() so that each data received emits it instantly to the parent Observable, but apparently I just can't do that

So, how can I manage to chain those Observables so that each submit in child Observable does emit directly to parent Observable ?

rxjs angular javascript
3 Answers

You need forkJoin, it returns an Array with the results of every Observable (in the same order than forkJoin parameters.

  getAll(): Observable<any[]> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      mergeMap((home: Home) =>
        forkJoin(
          this.getAssets(home.images).pipe(defaultIfEmpty([])),
          this.getStyling(home.styling).pipe(defaultIfEmpty({})),
          this.getCategories(home.categories).pipe(defaultIfEmpty([]))
        )
      )
    );
  }
user_37381710

If they are independent, you can handle them independently and emit them on a subject.

  getAll(): Observable<any> {
    const subject = new Subject<any>();
    const handler = (res: any) => subject.next(res);

    this.getHome().pipe(
      tap((home: Home) => this.application = home),
      tap(handler),
      tap((home: Home) => {
        forkJoin(
          this.getAssets(home.images).pipe(defaultIfEmpty([]),tap(handler))
          this.getStyling(home.styling).pipe(defaultIfEmpty({}), tap(handler))                  
          this.getCategories(home.categories).pipe(defaultIfEmpty([]), tap(handler))
        ).subscribe(res => subject.complete())
      })
    ).subscribe();
    return subject.asObservable();
  }

With this method, type checking has to be done by the consumer.

Subjects are documented here.

user_96805340

I've been trying to use merge() instead of combineLatest() so that each data received emits it instantly to the parent Observable, but apparently I just can't do that

It sounds like you're looking for partial data as soon as it is ready. Both combineLatest() and forkJoin() will only emit data when they can populate an array of equal length to the number of observables. So you either wait for at least one value in each observable or wait for them all to complete, but they emit an array and they need to know what to put in it.

I think you want to emit each piece of data as soon as it's ready.

Start by using merge() to emit values from all 3 observables, but for each observable use map() to create an object key/value pair for that given type (images, styling, categories).

Finally, use a scan() operator to aggregate the values into a single object that will emit each property as it is updated.

Consumers of the stream can check if a property is null to see if it is ready or not. The stream will emit 3 values before it is complete.

  getAll(): Observable<{home: Home, images: any[], styling: any[], categories: any[]}> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      switchMap((home: Home) =>
        merge([
          this.getAssets(home.images).pipe(
             defaultIfEmpty([]),
             map(images => ({images}))
          ),
          this.getStyling(home.styling).pipe(
             defaultIfEmpty({}),
             map(styling => ({styling}))
          ),
          this.getCategories(home.categories).pipe(
             defaultIfEmpty([]),
             map(categories => ({categories})
          ),
        ]),
        scan((acc, next) => ({...acc, ...next}), {home, images: null, styling: null, categories: null})
      )
    );
  }

You could add the startWith({}) operator to the merge() operator if you wanted to emit a first value where all properties are null. As an early value for consumers so they know reading for the other values has started.

user_10315690
Related