RxJs - Dynamically add synchronous calls that depend on eachother to a fork join without nesting-CodePudding

I'm trying to avoid nesting some synchronous calls that depend on each other. I also want to do a subscribe at the end result when all are finished, like a forkjoin.

I have found how to do this on stackoverflow (Wait for nested subscribe in angular before continuing with main subscribe (rxJs)), but not how to do it Dynamically.

Currently I'm using a callback, but this doesn't work with a subscribe when all finished / forkjoin behavior:

public keys = ['stateId', 'countyId'];


function populate(index, params = null) {


let route = this.baseRoute   this.keys[index];

this.apiCall(route, params).subscribe(response => {

if (index   1 < keys.length) {
this.populate(index   1, this.getParamsFromResponse(response));

}

Here's the answer from the above referenced stackoverflow link:

 getData() {
  isLoading = true;
  this.someSevice.getSomeData().pipe(
    mergeMap(getUserData),
    mergeMap(getMetadata),
  )
  .subscribe(
    response => {
      isLoading = false;
    },
    error => {
      isLoading = false;
      this.errors = error;
    },
  );
}

Using the answer in the above referenced stackoverflow link, I don't know how to

A - plug in the mergeMaps dynamically via loop instead of hardcoding them in

B - pass the appropriate arguments to the mergeMaps (route and params)

I'd appreciate any help. Thanks.

CodePudding user response：

Is it what you mean? If you can have the observable in an array you could use reduce operator to go through them in sequential order.

 ngOnInit() {
    const listOfString = ['item1', 'item2', 'item3', 'item4'];
    listOfString
      .slice(1, listOfString.length) // has to exclude the first item as we already defined it below
      .reduce(
        (a, b) => a.pipe(mergeMap((aResult) => this.getData(b, aResult))),
        this.getData(listOfString[0], null) //first call
      )
      .subscribe();
  }

  getData(input: string, params: string): Observable<string> {
    const result = `${input} ${params ? params : ''}`;
    console.log(result);
    return of(result);
  }

CodePudding user response：

Solution without reduce:

const {pipe, of} = rxjs;
const {mergeMap, tap} = rxjs.operators;

of({
  initValue: 1,
  sequence: [
    value => of(value   2),
    value => of(value * 2),
    value => of(`result is: ${value}`),
  ],
})
  .pipe(
    mergeMap(({ initValue, sequence }) =>
      of(initValue).pipe(
        ...sequence.map(stepFn => mergeMap(stepFn))
      )
    )
  )
  .subscribe(console.log);

<script src="https://unpkg.com/rxjs@^7/dist/bundles/rxjs.umd.min.js"></script>

CodePudding user response：

I'm not sure what exactly are the function you have in your example doing but if getParamsFromResponse() returns an Observable you could use something along these lines:

this.apiCall(route, params)
  .pipe(
    mergeMap(response => 
      forkJoin(keys.map(key => this.populate(index   1, this.getParamsFromResponse(response))))
    )
  )
  .subscribe(allResponses => /* one large array after all Observables completed */)