Ändern des beobachtbaren Streams beim Beibehalten des Abonnements

Question

In RxJS möchte ich, dass das Abonnement im Stream beibehalten wird, auch wenn der Stream geändert wird. Im Folgenden werde ich ein Intervall Strom das Verhalten

//Works because foo$ is unchanged 
let foo$ = Rx.Observable.interval(1000); 
foo$.subscribe(x => console.log(`foo$: ${x}`)); 

//Doesn't work because bar$ is changed 
let bar$ = Rx.Observable.never(); 
bar$.subscribe(x => console.log(`bar$: ${x}`)) 
bar$ = Rx.Observable.interval(1000); 

jsbin Live Demo

Wie bestehen ich das Abonnement zu testen, während bar$ Strom zu ändern? Muss ich das Abonnement ablegen und ein anderes Abonnement einrichten, nachdem ich bar $ geändert habe?

Answer 1

Das allgemeine Muster in Rx besteht darin, die Zustandsänderung durch einen Strom von Werten zu ersetzen (d. H. Beobachtbar). Anstatt $ bar neu zuzuweisen, sollte bar $ als Observable<Observable<T>> modelliert werden (dh ein Strom von Datenströmen vom Typ T). Es kann dann in einen Strom von Werten "verflacht" werden (in diesem Fall mit switch).

Zum Beispiel:

const bar$ = new Rx.Subject(); 
bar$.switch().subscribe(x => console.log(`bar$: ${x}`)); 
bar$.onNext(Rx.Observable.fromArray([1,2,3])); 
bar$.onNext(Rx.Observable.interval(1000).take(3)); 

https://jsbin.com/firoso/edit?js,console,output