Winkel 2 Kann mit mehreren Teilnehmern beobachtet werden

Das Problem, das Sie in Ihrem Code haben, besteht darin, dass Sie bei jedem Aufruf Ihrer Funktion ein neues Observable zurückgeben. Dies liegt daran, dass http.get bei jedem Aufruf ein neues Observable erstellt. Die Lösung dieses Problems könnte darin bestehen, das Beobachtbare (durch Schließung) im Dienst zu speichern, wodurch sichergestellt wird, dass alle Probanden dasselbe Beobachtbare abonnieren. Dies ist kein perfekter Code, aber ich hatte ein ähnliches Problem und dies löste mein Problem vorerst.

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

sie können einen reaktiven Datendienst erstellen und eine lokale beobachtbare Variable definieren, die intern aktualisiert wird und die Abonnenten selbst aktualisieren können. Dieser Artikel erklärt es richtig Datendienste

Ich weiß, dass dieser Thread alt ist, aber die akzeptierte Antwort hat mir sehr geholfen und ich möchte einige mögliche Verbesserungen hinzufügen, indem ich debounce, switchmap und ein hacky globales Benachrichtigungssystem verwende (dazu sollte das richtige ngrx verwendet werden: https: //ngrx.io/ )

Die Voraussetzung des Konzepts ist, dass ein Benachrichtigungsdienst verwendet werden kann, um alle anderen Dienste zu pushen, die sie auffordern, ihre Daten abzurufen:

export class NotifyerService {

  constructor() { }

  notifyer: Subject<any> = new Subject

  notifyAll() {
    console.log("ALL NOTIFIED")
    this.notifyer.next("GO")
  }

}

Ein Betreff wird verwendet, weil der Aufruf von .next (val) für einen Betreff die Daten an alle Listener weiterleitet

Im Service für eine bestimmte Komponente (in Ihrem Fall "DataService") können Sie die Datenerfassung und die Caching-Aktivität verwalten:

export class GetDataService {

  // cache the incoming data
  cache: Subject<any> = new Subject

  constructor(private http: HttpClient,
    private notifyerService: NotifyerService) {

    // subscribe to any notification from central message broker
    this.notifyerService.notifyer.pipe(

      // filtering can be used to perform different actions based on different notifications
      filter(val => val == "GO"),

      // prevent notification spam by debouncing
      debounceTime(300),

      // SUBSCRIBE to the output of getData, cancelling previous subscription
      switchMap(() => this.getData())

    ).subscribe(res => {

      console.log("expensive server request")

      // save data in cache, notify listeners
      this.cache.next(res)
    })

  }

  // expensive function which gets data
  getData(): Observable<any> {
    return this.http.get<any>(BASE_URL);
  }

} 

Das Schlüsselkonzept im obigen Code ist, dass Sie ein Cache-Objekt einrichten und aktualisieren, wenn eine Benachrichtigung vorliegt. Im Konstruktor möchten wir alle zukünftigen Benachrichtigungen über eine Reihe von Operatoren weiterleiten:

• Die Filterung kann verwendet werden, wenn Sie den Cache nur aktualisieren möchten, wenn der Benachrichtigungsdienst bestimmte Dinge ausgibt (in diesem Fall "GO"). Wenn Sie damit beginnen, wird mit ziemlicher Sicherheit stattdessen ngrx verwendet.
• debounceTime verhindert, dass Ihr Server mit vielen Anfragen überflutet wird (z. B. basieren die Benachrichtigungen auf Benutzereingaben)
• switchMap ist ein sehr wichtiger Operator für Caching und Statusänderungen. switchMap führt die Funktion inside aus (in diesem Fall die Funktion getData) und abonniert deren Ausgabe . Dies bedeutet, dass die Ausgabe von switchMap das Abonnement der Serverdaten ist.
• Ein weiteres wichtiges Merkmal von switchMap ist, dass frühere Abonnements abbricht . Dies bedeutet, dass, wenn Ihre Serveranforderung nicht zurückgegeben wurde, bevor eine andere Benachrichtigung eingeht, die alte abgebrochen und der Server erneut per Ping benachrichtigt wird.

Danach werden die Serverdaten mit der .next (res) -Methode in den Cache gestellt.

Jetzt muss die letzte Komponente nur noch den Cache auf Aktualisierungen abhören und entsprechend vorgehen:

export class ButtonclickerComponent implements OnInit {

  value: any;

  constructor(private getDataService: GetDataService,
    private notifyerService: NotifyerService) { }

  ngOnInit() {

    // listen to cache for updates
    this.getDataService.cache.pipe(

      // can do something specific to this component if have multiple subscriptions off same cache
      map(x => x)

      // subsc
    ).subscribe(x => { console.log(x); this.value = x.value })

  }

  onClick() {

    // notify all components of state changes
    this.notifyerService.notifyAll()
  }

}

Das Konzept in Aktion:

Angular App auf Knopfdruck

Serverantwort

Die Lösung wird gespeichert, sobald sie beobachtbar erstellt wurde, und sie kann gemeinsam genutzt werden (standardmäßig nicht). So wird Ihr Service aussehen:

@Injectable()
export class StationComplianceService {

  private stationCompliance: StationCompliance;
  private stream: Observable<StationCompliance []>;
  private url = '/api/read/stations';

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

   getStationCompliance() : Observable<StationCompliance []> {
     /** is remote value is already fetched, just return it as Observable */
     if (this.stationComliance) {
       return Observable.of(this.stationComliance);
     }
     /** otherwise if stream already created, prevent another stream creation (exactly your question */
     if (this.stream) {
       return this.stream;
     }
     /** otherwise run remote data fetching */
     this.stream = this.http.get(this.url)
      .map((res:Response) => res.json())
      .catch((error:any) => Observable.throw(error.json().error || 'Server Error'))
      .share(); /** and make the stream shareable (by default it is not) */
     return this.stream;
   }
}

shareReplay sollte es jetzt tun - "(...) wertvoll in Situationen, in denen Sie wissen, dass Sie spät Abonnenten eines Streams haben werden, der Zugriff auf zuvor ausgegebene Werte benötigt."