RxJS

Question

を再試行するか、リセットするには、のは、私はこのようなsequenceてきたとしましょう。

本当のシナリオはsequenceたちは一種のretryWhen(errors)作品のように、unsubscribeとresubscribeする必要がありますが、エラーで...理想的なもののような... retryWhen(bool:Subject)なりません特定の条件に、ソケットストリームであるということです。

Answer 1

古いObservableから自動的に退会し、新しいObservableにサブスクライブするので、私はswitchMap()を使用しています。この場合、我々は.switchMap(() => source)を使用します：

const subject = new Subject(); 

const source = Observable.create(obs => { 
    console.log('Observable.create'); 
    obs.next(42); 
}); 

subject.switchMap(() => source) 
    .subscribe(v => console.log('next:', v)); 


setTimeout(() => subject.next(), 1000); 
setTimeout(() => subject.next(), 5000); 

これは、次のように出力されます

Observable.create 
next: 42 
Observable.create 
next: 42 

だけではなく、あなたのWebSocketソース（またはものは何でも、あなたが持っている）を持っていますsource。