如何使用 Observables 进行组件间通信

Question

我正在使用 Rxjs 中的 Observables 和 Subjects 来在两个组件之间进行通信，这里是服务部分：

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





}

我从另一个组件的构造函数调用订阅者，它的 sn-p 是：

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

但是当我从后端收到数据时，我会调用主题的下一个方法，并在另一个组件的构造函数中再次调用它。它不起作用，而我已经看到它起作用的例子。该服务已在全球范围内注入。

Answer 1

恐怕你对 RxJS 的一些核心概念有误解。在 RxJS 中，主题是热门的可观察对象。无论是否有人在监听，一个热的 observable 都会发出事件。 （查看这篇文章了解更多信息https://blog.thoughtram.io/angular/2016/06/16/cold-vs-hot-observables.html）。

假设在您的服务中，您执行了一个后端调用，通过主题“管道”调用该调用的结果。 稍后在代码中，你的组件被启动，构造函数被执行，你开始监听主题。不幸的是，您通过主题“传递”的事件已经过去了。

要解决此问题，您可以使用 ReplaySubject(1) 更改主题。如果我没记错的话，应该可以解决问题。

但是，您的代码还存在一些其他问题。通过主题管道传输结果的事实在这里是不必要的。如果我查看您的代码，我认为您想进行一次后端调用，然后缓存结果。有一个特定的运算符，称为shareReplay。使用它，您的代码将如下所示：

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

您正在创建一个将执行一次的可观察对象，之后的结果将被缓存。您应该尽量避免在服务和主题中使用订阅。 要了解有关 shareReplay 的更多信息，请查看我写的关于 RxJS 中的多播运算符的博文：https://blog.kwintenp.com/multicasting-operators-in-rxjs