rxJs Rückruf mit beobachtbaren asynchron

Question

ich eine Karte Operation habe in meinem rxjs fließen

streaming.map((data) => { 
    //example async call 
    methodCall.then((response) => { 
    return data.test 
    }) 
}) 
.filter((value) => ...); 

Die Sache wird Filter aufgerufen wird, bevor data.test natürlich zurückgegeben wird. Also habe ich versucht, es zu switchMap Umwandlung und Rückkehr eines beobachtbare

streaming.switchMap((data) => { 
    return Observable.create((observer) => { 
    //example async call 
    methodCall.then((response) => { 
     observer.next(data.test); 
     observer.complete(); 
    }); 
    }); 
}) 
.filter((value) => ......); 

ich davon aus, dass in diesem Fall, da wir einen beobachtbaren mit explizitem Timing zurückkehren nächste auf, um die Filter genannt würden nur dann aufgerufen werden, nachdem die observer.complete genannt werden aber Filter wurde vorzeitig mit der Wertvariablen undefiniert aufgerufen.

Wie würde ich erreichen dies in der Regel mit rxJs

Answer 1

Ich denke, man eine return-Anweisung oder so etwas fehlt. Deshalb gibt es unerwartete Ergebnisse. Der Operator switchMap() abonniert das Observable, das von seinem Rückruf zurückgegeben wird, und wiederholt alle seine Elemente, bis ein anderes Observable zurückgegeben wird. Es wartet nicht auf die vollständige Benachrichtigung.

Dies sollte Ihr Beispiel simulieren:

function methodCall() { 
    return new Promise(resolve => setTimeout(() => { 
    resolve(123); 
    }, 1000)); 
} 

Observable.of(42) 
    .do(value => console.log('start: ' + value)) 
    .switchMap((data) => { 
    return Observable.create((observer) => { 
     //example async call 
     methodCall().then((response) => { 
     observer.next(response); 
     observer.complete(); 
     }); 
    }); 
    }) 
    .filter(value => true) 
    .subscribe(value => console.log('next: ' + value)); 

Siehe Live-Demo: https://jsbin.com/focili/2/edit?js,console

Beachten Sie, dass 123 abgegeben wird, wenn das Versprechen löst.

Übrigens brauchen Sie vielleicht switchMap() überhaupt nicht zu verwenden. RxJS 5 behandelt Observables, Promises, Arrays, Array-ähnliche Objekte usw. auf die gleiche Weise. Das bedeutet, dass Sie Observables für eines der oben genannten Elemente austauschen können. Zum Beispiel können Sie nur concatMap() verwenden und das Ergebnis wird das gleiche sein:

function methodCall() { 
    return new Promise(resolve => setTimeout(() => { 
    resolve(123); 
    }, 1000)); 
} 

Observable.of(42) 
    .do(value => console.log('start: ' + value)) 
    .concatMap(value => methodCall()) 
    .filter(value => true) 
    .subscribe(value => console.log('next: ' + value)); 

Siehe Live-Demo: https://jsbin.com/wipofiv/2/edit?js,console

Beachten Sie, dass concatMap() erhält ein Versprechen, aber es funktioniert immer noch als Sie erwartet nicht muss es sogar in ein Observable umwandeln.

Answer 2

Normalerweise Sie Promise-Observable mit fromPromise

streaming.switchMap((data) => Observable.fromPromise(methodCall)) 
     .filter((value) => ...); 

Dies tut alles hässlich Job verwandeln kann und Sie nur RxJS Programmierung genießen.

Wahrscheinlich vor filter würden Sie brauchen, um einige .map anwenden zurückzukehren data.test statt data