ポーリングを観測可能にする方法不完全な約束を検出する方法

Question

この観察可能なポーリングは、毎秒getPromise()の機能をポーリングします。 getPromise()関数が3つの約束を返すと、それらの解決を停止します。 getPromise()機能が過去の約束を解決/拒否していないことを検出するにはどうすればいいですか？たとえば2秒とし、onErrorハンドラに電話してください。私はそれをtimeoutオペレータと無駄に動作させるようにしました。何か案は？あなたが唯一の発する最初の観察可能にサブスクライブrace演算子を使用することができます

Rx.Observable.interval(1000) 
 
    .switchMap(() => Rx.Observable.fromPromise(getPromise())) 
 
    .subscribe(onValue, onError); 
 

 
function onValue(value){ 
 
    console.log('value: ', value); 
 
} 
 
function onError(error){ 
 
    console.log('error: ', error); 
 
} 
 
var getPromise = (function(){ 
 
    var counter = 3; 
 
    return function(){ 
 
    return new Promise(function(resolve, reject){ 
 
     if(counter > 0) resolve(1); 
 
     counter--; 
 
    }) 
 
    } 
 
})();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.3.0/Rx.js"></script>

Answer 1

。

2つの非アクティブの後にonErrorハンドラに電話したいとしました。これは、switchMapを使用することと矛盾し、新しいObservableがコールバックから返されたときに自動的に登録を解除します。だから、代わりにexhaustMapを使いたいかもしれません。また、エラー通知を出すと、チェーンは購読を中止し、決して他の値を受け取ることはありません。つまり、タイムアウトをerrorとして発行したり、retry演算子を使用して自動的に再登録したりしないでください（ただし、これは実際に達成しようとしているものによって異なります）。

これは、race()演算子を使用した例です。

Rx.Observable.interval(1000) 
 
    .switchMap(() => 
 
    Rx.Observable.race(
 
     Rx.Observable.fromPromise(getPromise()), 
 
     Rx.Observable.timer(0, 1000).mapTo(42) 
 
    ) 
 
) 
 
    .subscribe(onValue, onError); 
 

 
function onValue(value){ 
 
    console.log('value: ', value); 
 
} 
 
function onError(error){ 
 
    console.log('error: ', error); 
 
} 
 
var getPromise = (function(){ 
 
    var counter = 3; 
 
    return function(){ 
 
    return new Promise(function(resolve, reject){ 
 
     if(counter > 0) resolve(1); 
 
     counter--; 
 
    }) 
 
    } 
 
})();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.3.0/Rx.js"></script>

編集：非アクティブの2秒後に単一のエラー通知を送信します。

Rx.Observable.interval(1000) 
 
    .switchMap(() => Rx.Observable.fromPromise(getPromise())) 
 
    .timeout(2000) 
 
    .subscribe(onValue, onError); 
 

 
function onValue(value){ 
 
    console.log('value: ', value); 
 
} 
 
function onError(error){ 
 
    console.log('error: ', error); 
 
} 
 
var getPromise = (function(){ 
 
    var counter = 3; 
 
    return function(){ 
 
    return new Promise(function(resolve, reject){ 
 
     if(counter > 0) resolve(1); 
 
     counter--; 
 
    }) 
 
    } 
 
})();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.2.0/Rx.js"></script>

バグがtimeout()オペレータではなく、スケジューリング非同期アクションに直接5.3.0には本当にありません。 timeout()オペレータなしhttps://github.com/ReactiveX/rxjs/pull/2580

：

Rx.Observable.interval(1000) 
    .switchMap(() => 
    Rx.Observable.race(
     Rx.Observable.fromPromise(getPromise()), 
     Rx.Observable.timer(0, 2000).map(function(_) { 
     throw new Error('timeout'); 
     }) 
    ) 
) 
    .subscribe(onValue, onError);