How to use Observables for inter-component communication

Question

I am using Observables and Subjects from Rxjs to communicate between two components, Here is the service part:

import {
  Injectable,
  EventEmitter,
  Output
} from '@angular/core';
import {
  HttpClientModule,
  HttpClient
} from '@angular/common/http';

import {
  Observable
} from 'rxjs/Rx';
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/catch';
import {
  Subject
} from 'rxjs/Subject';


import {
  AppConstants
} from './../config/constants';

@Injectable()

export class GlobalService {
  private subject = new Subject < any > ();
  @Output() LoggedIn: EventEmitter < any > = new EventEmitter();
  mfaData: any;
  constructor(private http: HttpClient) {

  }

  validateCreds(postData, institutionId, customerId) {
    return this.http.post(AppConstants.baseUrl + AppConstants.serverRoutes.validateCreds + institutionId + '/' + customerId, postData)
      .subscribe(response => {
        console.log(response);
        if (response['status'] == 203) {
          this.mfaData = response['body'].questions;
          if (this.mfaData[0].choices || this.mfaData[0].imageChoices) {
            console.log('hey there');
            this.subject.next({
              mfaData: JSON.stringify(this.mfaData)
            });
           
          }
        }
      })
  }


  

  refreshHeaders(): Observable < any > {
    return this.subject.asObservable();
  }





}

and I am calling the subscriber from the constructor of another component the snippet of that is:

import {
  Subscription
} from 'rxjs/Subscription';
import {
  GlobalService
} from '../../../providers/global.serivce';

export class MfaChallengeComponent implements OnInit {
  subscription: Subscription;
  constructor(private activatedRoute: ActivatedRoute, private bankService: BanksService, private globalService: GlobalService) {
    this.subscription = this.globalService.refreshHeaders().subscribe(message => {
      console.log(message);
    });
  }
}

But when I receive the data from the back end I call the next method of the subject and call it again in the constructor of the other component. It doesn't work, whereas I have seen examples of it working. The service has been injected Globally.

Answer 1

I'm afraid you are misunderstanding some core concepts of RxJS. In RxJS, subjects are hot observables. A hot observable will emit events whether or not no one is listening. (checkout this article for more info https://blog.thoughtram.io/angular/2016/06/16/cold-vs-hot-observables.html).

So let's say in your service, you perform a backend call, where you 'pipe' the result of that call through the subject. Later on in the code, your component is started, and the constructor is executed and you start listening to the subject. The event you 'piped' through the subject has already passed unfortunately.

To fix this, you could change the Subject with a ReplaySubject(1). If I'm not mistaken, that should fix the problem.

There are however some other problems with your code. The fact you pipe the result through a subject is kind of unnecessary here. If I look at your code I think you want to do the backend call once and then cache the result. There is a specific operator for this called shareReplay. Using it your code would look like this:

export class GlobalService {
  cachedObservable$;
  @Output() LoggedIn: EventEmitter < any > = new EventEmitter();
  mfaData: any;
  constructor(private http: HttpClient) {
  }

  validateCreds(postData, institutionId, customerId) {
    this.cachedObservable$ = this.http.post(AppConstants.baseUrl + AppConstants.serverRoutes.validateCreds + institutionId + '/' + customerId, postData)
      .map(response => {
        if (response['status'] == 203) {
          this.mfaData = response['body'].questions;
          if (this.mfaData[0].choices || this.mfaData[0].imageChoices) {
            return {
              mfaData: JSON.stringify(this.mfaData)
            };
          }
        }
      })
      .shareReplay(1);
  }

  refreshHeaders(): Observable < any > {
    return this.cachedObservable$;
  }
}

You are creating an observable that will be executed once and the result afterwards will be cached. You should try to avoid using subscriptions in services and subject as much as possible. To learn more about shareReplay, checkout a blogpost about multicasting operators in RxJS I wrote: https://blog.kwintenp.com/multicasting-operators-in-rxjs