RxJS Observable forkJoin Not Executing in Parallel

user_13813314 0
Viewed 796

I have the following code and, for the life of me, can't figure out why the requests don't execute concurrently. I'm still new to RxJS and observables, so any help in improving the code below would be greatly appreciated as well. Basically, I'm making a call to a REST API on the backend to get some data. Then, for every element in that array of data I'm making another request to a different endpoint (hence using the 'forkJoin' operator). All the requests get sent at once, but they seem to execute one after another still instead of concurrently.

this.sites$.subscribe(data => {

    // data.forEach(element => {
    //     this.siteCaptureMap[element.id] = new CaptureData();
            
    //     this.sitesService.getCaptureData(element.nameOrNumber, element.owner.name).subscribe(data => {
    //         this.siteCaptureMap[element.id].count = data.length;
    //     });
    // });

    var obs: Observable<any>[] = [];
    for (var _i = 0; _i < data.length; _i++) {
        this.siteCaptureMap[data[_i].id] = new CaptureData();
        this.siteCaptureMap[data[_i].id].id = _i;
        obs.push(this.sitesService.getCaptureData(data[_i].nameOrNumber, data[_i].owner.name));
    }

    forkJoin(obs).subscribe(results => {
        for (var _i = 0; _i < results.length; _i++) {
            this.siteCaptureMap[data[_i].id].count = results[_i].length;
        }
    });


    this.dataSource.data = data;
    this.dataSource.filteredData = data;
});

Again, any help would be greatly appreciated. If I need to clarify anything or provide any additional code snippets, please let me know! Thanks!

rxjs-observables rxjs angular typescript javascript
2 Answers

Having nested subscribes can lead to memory leaks and can be tough to unsubscribe so whenever you have nested subscribes, think of switchMap, concatMap, and mergeMap.

They all have small variations but they switch to a new observable from the previous observable. This post explains the differences.

For you, I would try doing:

import { switchMap } from 'rxjs/operators';

...
this.sites$.pipe(
  switchMap(data => {
    let obs: Observable<any>[] = [];
    for (let _i = 0; _i < data.length; _i++) {
        this.siteCaptureMap[data[_i].id] = new CaptureData();
        this.siteCaptureMap[data[_i].id].id = _i;
        obs.push(this.sitesService.getCaptureData(data[_i].nameOrNumber, data[_i].owner.name));
    }

    return forkJoin(obs);
  }),
).subscribe(results => {
        for (let_i = 0; _i < results.length; _i++) {
            this.siteCaptureMap[data[_i].id].count = results[_i].length;
        }

        this.dataSource.data = data;
        this.dataSource.filteredData = data;
    });

A side note is to use let and const instead of var.

Also, if you see all the requests going out at the same time, that's all you can hope for. If it returns in series, it can either be the browser or the server causing that.

user_73654610

First I would rearrange the code to be more rxjs idiomatic, removing nested subscriptions and using pipe and making it a bit more functional-style. Inline comments try to explain the changes

this.sites$.pipe(
  // this is the rxjs map operator that transform the data received from
  // the rest api into something different
  map(data => {
    // I create the obs array using the map method of JS arrays - this is 
    // NOT the rxjs map operator
    obs = data.map((element, _i) => {
       this.siteCaptureMap[element.id] = new CaptureData();
       this.siteCaptureMap[element.id].id = _i;
       return this.sitesService.getCaptureData(element.nameOrNumber, element.owner.name)
    });
    // return both the array of observables and the data
    return [data, obs];
  }),
  // concatMap makes sure that you wait for the completion of the rest api
  // call and the emission of the data fetched before you execute another
  // async operation via forkJoin
  concatMap(([data, obs]) => forkJoin(obs).pipe(
    // this is the rxjs map operator used to return not only the results
    // received as result of forkJoin, but also the data received with the
    // first rest call - I use this operator to avoid having to define 
    // a variable where to store 'data' and keep the logic stateless 
    map(results => ([data, results]))
  ))
).subscribe(([data, results]) => {
  // here I use forEach to loop through the results and perform
  // your logic
  results.forEach((res, _i) => this.siteCaptureMap[data[_i].id].count = res.length)
})

I think this form is more in line with rxjs and its functional spirit, and should be equivalent to your original form.

At the same time I would be interested to know why you say that your code executes the calls within forkJoin one at the time.

By the way, if you are interested in looking at common patterns of use of rxjs with http, you can read this article.

user_56999930
Related