forkJoin's success callback is not getting called

Question

I have to make two parallel API calls in my angular application, and the data from each is to be used in the component. Below is my code for the services and component file. I have used forkJoin to make parallel calls and process. Because as per my understanding forkJoin makes sure that the success callback will be called when both API calls are successful.

While running, I could see that the success callback is getting called in each service but not in the component file.

api.services.ts

$getRequestById(id: string): Subject<any> {
    const data_subject = new Subject<any>();
    this.$http.get(API_URL + '/' + id).subscribe(
      (dto) => data_subject.next(new Request(dto)), // THIS IS GETTING EXECUTED
      (error) => data_subject.error(error)
    );
    return data_subject;
  }

  $getAllRequestsForId(id: string): Subject<any> {
    const data_subject = new Subject<any>();

    this.$http.get(API_URL + '?request_id=' + id).subscribe(
      (dto) => data_subject.next(dto.map(i => new Request(i))),
      (error) => data_subject.error(error)
    );

    return data_subject;
  }

And in my component file,

home.component.ts

const request = this.service.$getRequestById(this.id);
const allRequests = this.service.$getAllRequestsForId(this.id);

forkJoin([request, allRequests])
 .subscribe(
    (response) => this.setResponse(response), // THIS LINE IS NOT GETTING CALLED.
    (error) => this.notification.error('Error occured while fetching request details! Please try again.')
 );

But, if I ran this API call independently in any component, like below, it works without any issues.

this.service.$getRequestById(this.id).subscribe(
(response) => this.setResponse(response), // THIS GETS EXECUTED 
(error) => this.notification.error("An error occured")
)

I am using those services in other parts of the application independently without any problems, but when combining with forkJoin, the success callback is not getting called. Any idea how to fix this problem which works in both forkJoin as well as independent calls?

Answer 1

I'd say the calls are unnecessarily complicated by the usage of RxJS Subject. You could return the observable from the HttpClient and combine them using forkJoin.

From forkJoin docs:

Wait for Observables to complete and then combine last values they emitted.

So if one of the Subject doesn't complete, the forkJoin won't emit a response. If you wish to modify the response from an observable before the subscription, you could use RxJS map and catchError operators. You might not need the catchError here since you aren't modifying the error from the HTTP request.

Try the following

Service

import { pipe } from 'rxjs';
import { map } from 'rxjs/operators';

$getRequestById(id: string): Observable<any> {
  return this.$http.get(API_URL + '/' + id).pipe(
    map(dto => new Request(dto))
  );
}

$getAllRequestsForId(id: string): Observable<any> {
  return this.$http.get(API_URL + '?request_id=' + id).pipe(
    map(response => response.map(item => new Request(i)))     // <-- inner `map` is array method, outer `map` is RxJS operator
  );
}

Controller

const request = this.service.$getRequestById(this.id);
const allRequests = this.service.$getAllRequestsForId(this.id);

forkJoin([request, allRequests]).subscribe(
  (response) => this.setResponse(response),
  (error) => this.notification.error('Error occured while fetching request details! Please try again.')
);

Update: catchError

Most important thing to remember using catchError is to return an observable. You could use RxJS throwError which emits an error and completes. Or you could use RxJS of to return an observable.

import { pipe, throwError, of } from 'rxjs';
import { map, catchError } from 'rxjs/operators';

// throwing error using `throwError`
$getRequestById(id: string): Observable<any> {
  return this.$http.get(API_URL + '/' + id).pipe(
    map(dto => new Request(dto)),
    catchError(error => throwError('Ran into an error: ', error))       
  );
}

// throwing error using `of`
$getRequestById(id: string): Observable<any> {
  return this.$http.get(API_URL + '/' + id).pipe(
    map(dto => new Request(dto)),
    catchError(error => of('Ran into an error: ', error))               
  );
}

Answer 2

forkjoin will only call the subscribe callback once every source observable completes. In your service functions, you return a subject which never completes, so the forkjoin will wait forever.

Also, you do not need create a new subject in your service functions. Return the result of http.get directly. If you need to transform the result within the service, you can pipe the result through some rxjs operators

api.services.ts

$getRequestById(id: string): Subject<any> {
    return this.$http.get(API_URL + '/' + id)
  }

$getAllRequestsForId(id: string): Subject<any> {
    return this.$http.get(API_URL + '?request_id=' + id).pipe(
      concatAll(),
      map(i => new Request(i)),
      toArray()
    );
  }

Answer 3

API is not getting called because you already consumed the API by using a subscribe so remove the subscribe and try again like this

api.services.ts

$getRequestById(id: string) {
    return API_URL + '/' + id;
  }

  $getAllRequestsForId(id: string) {
     return API_URL + '?request_id=' + id;
  }

home.component.ts

const request = this.service.$getRequestById(this.id);
const allRequests = this.service.$getAllRequestsForId(this.id);

forkJoin([request, allRequests])
 .subscribe(
    (response) => {
     console.log(response);
     this.setResponse(response)
    },
    (error) => this.notification.error('Error occured while fetching request details! Please try again.')
 );

Answer 4

Your code should be

$getRequestById(id: string): Observable<any> {
    return this.$http.get(API_URL + '/' + id).pipe(map(
      (dto) => new Request(dto)), // THIS IS GETTING EXECUTED
      (catchError) => throwError(err)
    );
  }

  $getAllRequestsForId(id: string): Observable<any> {
    return this.$http.get(API_URL + '?request_id=' + id).pipe(map(
      (dto) => dto.map(i => new Request(i))),
      (catchError) => throwError(error)
    );
  }

home.component.ts

const request = this.service.$getRequestById(this.id);
const allRequests = this.service.$getAllRequestsForId(this.id);

forkJoin([request, allRequests])
 .subscribe(
    (response) => this.setResponse(response), // THIS LINE IS NOT GETTING CALLED.
    (error) => this.notification.error('Error occured while fetching request details! Please try again.')
 );