Rx.Observable Validieren Sie Elemente im Stream (Pass oder Throw)

Question

Gegeben ein Stream innerhalb eines Observable, ich möchte jedes Element validieren/überprüfen. Wenn einer defekt ist, möchte ich einen Fehler über Observable.throw werfen, damit bricht die weitere Verarbeitung ab.

würde meine klobig Lösung

import * as Rx from 'rxjs' 

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

Dies scheint hässlich sein, da es eine Reihe von unnötigen Observable für die positive Strömung konstruiert.

Gibt es eine bessere Möglichkeit, Elemente in einem Observable zu überprüfen?

Answer 1

Sie können ganz normale map verwenden und eine Ausnahme im Inneren werfen:

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

Answer 2

Wenn ich Ihre Frage gut verstehen, Sie takeWhile Betreiber, dies zu tun verwenden können. Beispielsweise;

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

Es dauert nur Werte, während Ausdruck wahr ist. Wenn es falsch wird, hört es auf. Sie können mehr von here erfahren. Sie können auch this page überprüfen. Ich hoffe, es hilft!