Rx.Observableストリーム内の項目を検証する（パスまたはスロー）

Question

Observable内でストリームを指定すると、各項目を検証/チェックしたいと思います。 1つが壊れている場合、私はObservable.throwを介してエラーを投げたいので、それ以上の処理はすべて中断します。

私の不格好なソリューションは、

import * as Rx from 'rxjs' 

inputStream.mergeMap(item => (isValid(item)) 
    ? Rx.Observable.of(item) 
    : Rx.Observable.throw(new Error("not valid")) 
) 

だろう、それは正の流れのために不必要な観測の束を構築し、これは、醜いようです。

Observableでアイテムをチェックする良い方法はありますか？

Answer 1

あなただけの通常のmapを使用し、その中に例外をスローすることができます

inputStream.map(item => { 
    if (isValid(item)) { 
    return item; 
    } 
    throw new Error("not valid"); 
}) 

Answer 2

私はよくあなたの質問を理解していれば、あなたがこれを行うにはtakeWhile演算子を使用することができます。例えば;

yourObservable.takeWhile(item => { 
    //your condition 
}).subscribe(i => console.log(i)); 

あなたの式が真である間だけ値を取得します。偽になると停止します。

詳しくは、hereからご覧ください。またthis pageをチェックすることもできます。私はそれが助けて欲しい！