rxjs5/Angular - Clear ReplaySubject Puffer

Question

Ich habe einen Angular-Dienst, der einen Datenstrom (http-Aufruf) an mehrere Komponenten asynchron teilt. Ich muss den http-Dienst auf Benutzeraktion von Zeit zu Zeit erinnern.

Ich verwende ein ReplaySubject, um geladene Werte zu speichern und an Abonnenten zu senden, die nach dem http-Aufruf abonnieren.

Ich würde gerne wissen, ob es eine Möglichkeit gibt, den ReplaySubject Puffer zu löschen, bevor Sie nachfolgende HTTP-Aufrufe machen? Während ich dabei bin, vermute ich, dass ich mich auch abmelden muss, um kein Leck zu erstellen?

Service:

@Injectable() 
export class GreatDataService { 

    public data$: ReplaySubject<any>; 
    private subs: Subscription; 

    constructor(private http: Http) { 
     this.data$ = new ReplaySubject(1); 
    } 

    public refresh() { 
     if (this.subs) { 
      this.subs.unsubscribe(); 
      this.subs = null; 
     } 
     this.subs = this.http.get('/api').subscribe(this.data$) 
    } 
} 

Top-Level-Abschnitt Komponente:

... 
    constructor(private greatDataService: GreatDataService) {} 
    ngOnInit() { 
     this.greatDataService.refresh(); 
    } 
... 

Komponente 1:

... 
    constructor(private greatDataService: GreatDataService) {} 
    ngOnInit() { 
    this.greatDataService.data$.subscribe(
     x => console.log('subscriber 1: ' + x), 
     err => console.log('subscriber 1: ' + err), 
     () => console.log('subscriber 1: Completed') 
    ); 
... 

Komponente 2:

... 
    constructor(private greatDataService: GreatDataService) {} 
    ngOnInit() { 
    this.greatDataService.data$.subscribe(
     x => console.log('subscriber 2: ' + x), 
     err => console.log('subscriber 2: ' + err), 
     () => console.log('subscriber 2: Completed') 
    ); 
... 

Answer 1

Sie können ein Rx.Subject verwenden, um neue 'get-fresh-data'-Ereignisse zu senden, damit Sie neue Daten abrufen können, wenn mit einem .switchMap() aufgerufen wird. Sehen Sie dieses Beispiel, wie dies zu tun:

function getData() { 
 
    return Rx.Observable.of('retrieving new data') 
 
    .timestamp() 
 
    .delay(500); 
 
} 
 

 
// in this example i use an eventStream of clicks 
 
// you can use Rx.Subject() and manually .next() a new value 
 
// when somebody invokes .refresh() 
 
const refreshDataClickStream = Rx.Observable.fromEvent(document.getElementById('refresh_stream'), 'click'); 
 

 
const dataStream = refreshDataClickStream 
 
    .startWith('PAGE_LOAD') /* let the stream always first time fetching data */ 
 
    .switchMap(() => getData()) /* getData() is not cached so we switchMap to a new instance, abandoning the previous result*/ 
 
    .publishReplay().refCount(); /* refCounter so everybody gets the same results */ 
 

 
dataStream.subscribe(data => console.log('sub1 data: ' + JSON.stringify(data))); 
 

 
setTimeout(() => { 
 
    console.log('late arriving subscription (gets same stream)'); 
 
    dataStream.subscribe(data => console.log('sub2 data: ' + JSON.stringify(data))); 
 
}, 2000);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.3/Rx.js"></script> 
 
<input type='button' id='refresh_stream' value="refresh_stream" />