具有多个订户的Angular 2 Observable

Question

我有一个角度2服务,从API获取数据,这个服务有3个订阅者(在组件中定义),每个订阅者使用数据执行其他操作(不同的图表)

我注意到我正在向API发出三个GET请求,而我想要实现的是一个请求,并且订阅者将共享我看到HOT和COLD可观察的数据并尝试使用.share()可观察但我仍在进行3次个人通话

更新,添加代码

服务

import { Injectable } from '@angular/core';
import { Http, Response } from '@angular/http';

import {Observable} from 'rxjs/Rx';

// Import RxJs required methods
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/catch';

import { StationCompliance } from './model/StationCompliance';


@Injectable()
export class StationComplianceService {

  private url = '/api/read/stations';

  constructor(private http : Http) {
    console.log('Started Station compliance service');
   }

   getStationCompliance() : Observable<StationCompliance []> {
     return this.http.get(this.url)
      .map((res:Response) => res.json())
      .catch((error:any) => Observable.throw(error.json().error || 'Server Error'));
   }
}

组件1

import { Component, OnInit } from '@angular/core';
import { CHART_DIRECTIVES } from 'angular2-highcharts';

import { StationComplianceService } from '../station-compliance.service';


@Component({
  selector: 'app-up-down-graph',
  templateUrl: './up-down-graph.component.html',
  styleUrls: ['./up-down-graph.component.css']
})
export class UpDownGraphComponent implements OnInit {

  graphData;

  errorMessage: string;

  options;

  constructor(private stationService : StationComplianceService) { }

  ngOnInit() {
    this.getStationTypes();
  }

  getStationTypes(){
    this.stationService.getStationCompliance()
      .subscribe(
        graphData => {
          this.graphData = graphData;
          this.options = {
            chart : {type: 'pie',
                    plotShadow: true
            },
            plotOptions : {
              showInLegend: true
            },
            title : {text: 'Up and Down devices'},
            series: [{
              data: this.processStationType(this.graphData)
            }]
          }
        },
        error => this.errorMessage = <any>error
      );
  }

其他两个组件几乎相同,它们只显示其他图形

Answer 1

我遇到了类似的问题并使用Aran的建议来解决它,以引用Cory Rylan的Angular 2 Observable Data Services博客文章.我的关键是使用BehaviorSubject.这是最终适合我的代码片段.

数据服务:

BehaviorSubject初始化服务时,数据服务会创建一个内部缓存数据.消费者使用该subscribeToDataService()方法访问数据.

    import { Injectable } from '@angular/core';
    import { Http, Response } from '@angular/http';

    import { BehaviorSubject } from 'rxjs/BehaviorSubject';
    import { Observable } from 'rxjs/Observable';

    import { Data } from './data';
    import { properties } from '../../properties';

    @Injectable()
    export class DataService {
      allData: Data[] = new Array<Data>();
      allData$: BehaviorSubject<Data[]>;

      constructor(private http: Http) {
        this.initializeDataService();
      }

      initializeDataService() {
        if (!this.allData$) {
          this.allData$ = <BehaviorSubject<Data[]>> new BehaviorSubject(new Array<Data>());

          this.http.get(properties.DATA_API)
            .map(this.extractData)
            .catch(this.handleError)
            .subscribe(
              allData => {
                this.allData = allData;
                this.allData$.next(allData);
              },
              error => console.log("Error subscribing to DataService: " + error)
            );
        }
      }

      subscribeToDataService(): Observable<Data[]> {
        return this.allData$.asObservable();
      }

      // other methods have been omitted

    }

零件:

组件可以在初始化时订阅数据服务.

    export class TestComponent implements OnInit {
      allData$: Observable<Data[]>;

      constructor(private dataService: DataService) {
      }

      ngOnInit() {
        this.allData$ = this.dataService.subscribeToDataService();
      }

    }

组件模板:

然后,模板可以根据需要使用异步管道迭代observable.

    *ngFor="let data of allData$ | async" 

每次在数据服务中next()调用方法时,都会更新订阅者BehaviorSubject.

Answer 2

您的代码中存在的问题是，每次调用函数时，您都将返回一个新的observable。这是因为http.get每次调用时都会创建一个新的Observable。解决此问题的方法可能是将可观察对象（通过关闭）存储在服务中，这将确保所有主题都订阅相同的可观察对象。这不是完美的代码，但是我遇到了类似的问题，这暂时解决了我的问题。

import { Injectable } from '@angular/core';
import { Http, Response } from '@angular/http';

import {Observable} from 'rxjs/Rx';

// Import RxJs required methods
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/catch';

import { StationCompliance } from './model/StationCompliance';


@Injectable()
export class StationComplianceService {

  private url = '/api/read/stations';

  constructor(private http : Http) {
    console.log('Started Station compliance service');
   }

   private stationComplianceObservable: Rx.Observable<StationCompliance[]>;


   getStationCompliance() : Observable<StationCompliance []> {

    if(this.stationComplianceObservable){
        return this.stationComplianceObservable;
    }        

      this.stationComplianceObservable = this.http.get(this.url)
      .debounce(1000)
      .share()
      .map((res:Response) => res.json())
      .finally(function () { this.stationComplianceObservable = null}) 
      .catch((error:any) => Observable.throw(error.json().error || 'Server Error'));

    return this.stationComplianceObservable;
   }
}