ForkJoin - ForLoop - Handle some logic after each subscription and some after all obseravables resolved

Question

I have a http call inside a forloop. The APICall will return a HTML response which i am using it render on the page.

Now I also have a requirement where after all API calls's are completed i have to perform some logic to update some data on the backend.

I understand we can use a forkjoin operator to capture array of observables and then update the BE data. But i am unable to understand how to handle the requirement that has to be completed on each subscription.

for(let item of Items){
    this.myService
            .getMyItemData(item.key)
            .pipe(
              takeUntil(this.destroyed),
              distinctUntilChanged(),
              catchError((e: Error) => {
                this.logger.logError('Error loading', e);
                return of('An Error Occurred');
              })
            ).subscribe((resp) => {

//How can i handle this subscription when using forkjoin ??

             this.elementRef.nativeElement.html = resp;
    }) 
}

Now after getting all the itemData i want to perform an update for the Backend. For this i am thinking of using a forkJoin to capture all observables data. But at the same time i want to use the subscription code to render the HTML. Can someone help how can i achieve that.

Reference for my forkJoin Code*

let arrayOfObservables  = Items.map((item) => this.myService
                .getMyItemData(item.key))

let dataSource =  Rx.Observable.forkJoin(arrayOfObservables);

dataSource.subscribe((resp) => {
  // update my BE data
})

Does each response from `getMyItemData()` need to update a different `elementRef`? — BizzyBob, Mar 08 '21 at 18:06
@BizzyBob yes, i believe yes each response is different elementref. So basically i am using this itemdatacomponent as a child component inside a loop in parent component html conditionally. So suppose i am displaying 20 result rows on the page then around 5 will be the HTML from the child component (itemDataComponent.) — Deadpool_er, Mar 08 '21 at 18:16

ruth · Accepted Answer · 2021-03-08T17:37:23.260

forkJoin will only emit when all the source observable complete. Given the usage of distinctUntilChanged() operator I assume each observable is a stream of notifications.

In that case combineLatest is better fit than forkJoin. It'll emit when any of the source observable emits. But note: each observable should've emitted at least once for the subscription to be triggered. If you wish to trigger the subscription before some observables haven't emitted yet, you could pipe startWith(null) to each source observable. Also look into RxJS zip function. Here is a quick run down b/n different functions.
To perform something after each emission, you could either tap operator (if performing side-effects) or map operator (to transform the data).

import { of, combineLatest, Subject } from 'rxjs';
import { tap, takeUntil, startWith, distinctUntilChanged, catchError } from 'rxjs/operators';

combineLatest(
  Items.map((item) => 
    this.myService.getMyItemData(item.key).pipe(
      startWith(null),           // <-- use conditionally (see above)
      distinctUntilChanged(),
      tap((resp) => {
        if (!!resp) {            // <-- avoid `null` from `startWith`
          this.elementRef.nativeElement.html = resp;
        }
      }),
      catchError((e: Error) => {
        this.logger.logError('Error loading', e);
        return of('An Error Occurred');
      })
    )
  )
).pipe(
  takeUntil(this.destroyed)
).subscribe((resp) => {
  // update my BE data
});

Combine Latest worked for me in this scenario. Thank you. – Deadpool_er Dec 14 '21 at 05:53 — Deadpool_er, Dec 14 '21 at 05:53

score 0 · Answer 2 · answered Mar 08 '21 at 19:36

To answer your main question:

How to handle some logic after each subscription and some after all observables resolved

If I understand correctly, you want to create a single dataSource that represents data from multiple different calls. You want to do something when each individual call returns (Work #1), and do something else when all calls have completed (Work #2).

To perform your Work #1, you can use the tap operator in the definition of your "individual" call. tap allows you to execute code whenever a value passes through the observable stream. If multiple values will be passing through and you only want to run logic when the observable completes, you can use finalize instead.

forkJoin creates the dataSource composed of many individual calls and it will emit an array of all the results once they have all completed. So, you can easily do your Work #2 in the subscribe. However, you can also do your work in tap or finalize as well:

individual$ = key => this.myService.getMyItemData(key).pipe(
    tap(result => console.log(`call #${key} DONE`)) // <-- Do Work #1
);

dataSource$ = forkJoin(Items.map(i => individual$(i.key)).pipe(
    tap(allResults => console.log('all calls DONE')) // <-- Do Work #2
);

dataSource$.subscribe();

A single subscription to dataSource$ will execute the logic for the individual calls (Work #1) and the logic for "all calls" (Work #2).

It seems like it would get messy to manage lots of elementRefs in the controller and manipulate the DOM using this.elementRef.nativeElement.html. Depending on your scenario, it may be simpler to bind to the innerHTML property using *ngFor:

<ul *ngIf="dataSource$ | async as items">
  <li *ngFor="let item of items" [innerHTML]="item"></li>
</ul>

Notice if you use the async pipe, you don't even need to subscribe in your controller.

Here is a StackBlitz sample.

ForkJoin - ForLoop - Handle some logic after each subscription and some after all obseravables resolved

2 Answers2