Parallel Execution Issue with RxJS Observable forkJoin

Question

Parallel Execution Issue with RxJS Observable forkJoin

Struggling to understand why my requests aren't executing concurrently with the following code. As a newcomer to RxJS and observables, I would greatly appreciate any guidance on improving this snippet below. Essentially, I am fetching data from a REST API backend. Subsequently, for every element in the array of data, I'm calling a different endpoint using the 'forkJoin' operator. Despite sending all requests at once, they appear to execute sequentially rather than concurrently.

this.sites$.subscribe(data => {

    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;
});

Your assistance is highly valued. If further clarification or additional code snippets are needed, please do not hesitate to reach out. Thank you for your help!

javascript angular typescript rxjs rxjs-observables

Answer 1

Answer №1

When dealing with nested subscribe calls, it's important to be cautious of potential memory leaks and difficulty in unsubscribing. In such cases, consider using switchMap, concatMap, and mergeMap.

Although they have slight differences, all three operators switch to a new observable from the previous one. This article provides a detailed explanation of their distinctions.

For your situation, you could try implementing:

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;
    });

Remember to use let and const instead of var. Also, if you notice all requests going out simultaneously, that indicates optimal performance. However, if they are processed sequentially, it could be due to either browser or server limitations.

Answer 2

When dealing with nested subscribe calls, it's important to be cautious of potential memory leaks and difficulty in unsubscribing. In such cases, consider using switchMap, concatMap, and mergeMap.

Although they have slight differences, all three operators switch to a new observable from the previous one. This article provides a detailed explanation of their distinctions.

For your situation, you could try implementing:

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;
    });

Remember to use let and const instead of var. Also, if you notice all requests going out simultaneously, that indicates optimal performance. However, if they are processed sequentially, it could be due to either browser or server limitations.

Answer 3

Answer №2

To enhance the code's readability and conformity with rxjs standards, my approach would involve restructuring it to reflect a more idiomatic style. This would entail eliminating nested subscriptions, utilizing the pipe function, and embracing a functional programming approach. As indicated in inline comments, these modifications aim to clarify the code.

this.sites$.pipe(
  // Utilizing the rxjs map operator to transform data fetched from the rest api
  map(data => {
    // Generating an array of observables using JavaScript array's map method (note: 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)
    });
    // Returning both the array of observables and the original data
    return [data, obs];
  }),
  // Using concatMap to ensure sequential execution of async operations via forkJoin
  concatMap(([data, obs]) => forkJoin(obs).pipe(
    // Leveraging the rxjs map operator to include both results from forkJoin and initial data
    map(results => ([data, results]))
  ))
).subscribe(([data, results]) => {
  // Employing forEach loop to iterate through results and implement necessary logic
  results.forEach((res, _i) => this.siteCaptureMap[data[_i].id].count = res.length)
})

This revised structure aligns better with rxjs principles, emphasizing its functional essence while maintaining equivalence with the original script.

If you have mentioned that your code executes the calls within forkJoin sequentially, I would appreciate further insight into this claim.

Additionally, for an exploration of common rxjs usage patterns in conjunction with HTTP requests, feel free to peruse this informative article.

Answer 4

To enhance the code's readability and conformity with rxjs standards, my approach would involve restructuring it to reflect a more idiomatic style. This would entail eliminating nested subscriptions, utilizing the pipe function, and embracing a functional programming approach. As indicated in inline comments, these modifications aim to clarify the code.

this.sites$.pipe(
  // Utilizing the rxjs map operator to transform data fetched from the rest api
  map(data => {
    // Generating an array of observables using JavaScript array's map method (note: 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)
    });
    // Returning both the array of observables and the original data
    return [data, obs];
  }),
  // Using concatMap to ensure sequential execution of async operations via forkJoin
  concatMap(([data, obs]) => forkJoin(obs).pipe(
    // Leveraging the rxjs map operator to include both results from forkJoin and initial data
    map(results => ([data, results]))
  ))
).subscribe(([data, results]) => {
  // Employing forEach loop to iterate through results and implement necessary logic
  results.forEach((res, _i) => this.siteCaptureMap[data[_i].id].count = res.length)
})

This revised structure aligns better with rxjs principles, emphasizing its functional essence while maintaining equivalence with the original script.

If you have mentioned that your code executes the calls within forkJoin sequentially, I would appreciate further insight into this claim.

Additionally, for an exploration of common rxjs usage patterns in conjunction with HTTP requests, feel free to peruse this informative article.

Parallel Execution Issue with RxJS Observable forkJoin

Answer №1

Answer №2

Similar questions

Issue with the useState hook not correctly updating a value

Find unique numbers within a specified range using NodeJS

The Art of Checkbox Design

Error: The operation 'join' cannot be performed on an undefined value within Fast2sms

The Outer Div Can't Contain Google Maps

The error message "Unexpected token < in JSON at position 0" is indicating a SyntaxError in the

Issue with Dynamic Image Path in Require Function: Unable to locate the relative module

Generate a high-resolution image using PhaserJS

select specific region within image

Guide to hosting AngularJS Material documentation on a local server

Angular directive problem

How to process JSON data that includes a function

Guide to hosting a Razor partial view within Angular, without using IIS

Configuring vue-jest: Does anyone know how to set up aliases for template/script src attributes in Vue?

Tips for implementing self-managed state in Vue.js data object

We were unable to locate a declaration file for the module known as 'firebase-tools'

Transferring information between an ASP web form page and an Angular2 application

Dynamically loading JQuery causes highcharts to encounter errors

[Simple TypeScript]: Assign the parameter value as the key of the object returned by the function

Get the socket identification from ngx-socket-io