Need help to execute sequential api calls in for-loop

Issue

I am working with Angular and NodeJs (with Axios) using RxJs, and currently find it very challenging to achieve this task. I will explain my problem based on a scenario.

I have an array of objects like this, with possibly even more than 100 objects:

let objArr = [{name: 'john', id: '123', country: 'usa'}, {name: 'doe', id: '456', country: 'china'}....]

Then I have another 4-5 validation APIs that can be called for different params e.g. id, name and country based on each object:

api_1 = "validate_step_1/:id"
api_2 = "validate_step_2/:id/:name"
api_3 = "validate_step_3/:id/:country"
api_4 = "validate_step_4/:id:/:name/:country" 

These API calls should strictly happen one after another in a sequential pattern, e.g. api_2 should only be called if api_1 returns true and so on.

What I want:

I would like to execute for-loop on the array that should run in parallel, and each object should then sequentially validate itself based on these 4 API calls. Something like sequential API calls based on each item in for-loop in parallel for all 100 objects.

Is this even possible? Also, any solutions to achieve this on the Node side are also welcomed.

What I tried

Right now I am using this method, but it's very slow, even sometimes resulting in timeout errors in Axios:

  of(...this.objArr).pipe(
    concatMap((obj: any) => this.service.api_1(id)),
    concatMap((obj: any) => this.service.api_2(id, name)),
    concatMap((obj: any) => this.service.api_3(id, country)),
    concatMap((obj: any) => this.service.api_4(id, name, country)),
    catchError( error => this.handleError(error))
  ).subscribe(
    success => {
      console.log("validation succeed", success);
    },
    errorData => {
      console.log("validation failure: ", errorData);
    }
  )

Solution

I would adopt this approach.

First of all create a function that build an Observable that runs all the validations in parallel and handle any error we may encounter (more on errors later). Something like this

function validate(id, name, country) {
  return concat(
    this.service.api1(id),
    this.service.api2(id, name),
    this.service.api3(id, country),
    this.service.api4(id, name, country)
  ).pipe(
    catchError(err => of(err.message))
  );
}

Then I would use the from function from the rxjs library to turn the array of objects into a stream and then apply the mergeMap operator to launch all the validations in parallel, like this

from(objArr)
  .pipe(mergeMap((obj) => validate(obj.id, obj.name, obj.country)))
  .subscribe((v) => console.log(v));

Why use mergeMap over forkJoin in this case. The main reason is that with mergeMap you can control the level of concurrency you want to have. If you do not specify anything, all the validations run in parallel. But, for instance, if you want to limit the number of parallel validations, you can use the optional concurrent parameter of mergeMap like this

const concurrent = 3 // whatever limit 
from(objArr)
   .pipe(mergeMap((obj) => validate(obj.id, obj.name, obj.country), concurrent))
   .subscribe((v) => console.log(v));

If you want to proceed in the validations only if a certain criteria is met, you can simply use the tap operator and throw an error if the condition is not met. The error will be caught by the catchError operator which we have added at the end of the pipe in the validate function. So, a validation API would look like this

api3(id, country) {
  return this.invokeApi3(id, country).pipe(
    tap(() => {
      if (// check the criteria for which you do not want to continue) {
        throw new Error('Error in API 3');
      }
    })
  );
}

You can look at this stackblitz for an examp.le

Answered By - Picci