RxJS Promise Composition (Weitergabe von Daten)

Question

Ich bin neu bei Rx und finde es schwierig, eine Dokumentation über das Verfassen von Versprechen zu finden, so dass Daten vom ersten Versprechen in das zweite und so weiter übertragen werden. Hier sind drei sehr grundlegende Versprechen, die Berechnungen auf den Daten sind nicht wichtig, nur dass etwas Async mit Daten aus dem vorherigen Versprechen getan werden muss.

const p1 =() => Promise.resolve(1); 
const p2 = x => { const val = x + 1; return Promise.resolve(val); }; 
const p3 = x => { 
     const isEven = x => x % 2 === 0; 
     return Promise.resolve(isEven(x)); 
}; 

Der traditionelle Weg, um die Zusammensetzung ich spreche zu erreichen:

pl().then(p2).then(p3).then(console.log); 

Meine Lieblings Implementierung ist Ramda der composeP und pipeP:

R.pipeP(p1, p2, p3, console.log)() 

Es scheint wahrscheinlich, Rx könnte der Lage sein, mit dieser Art von Situation ziemlich flüssig umzugehen. in der Nähe die ich bisher gefunden habe, ist jedoch von dem RxJS ASYNC (Bibliothek) Vergleich hier https://github.com/Reactive-Extensions/RxJS/blob/master/doc/mapping/async/comparing.md:

var Rx = require('rx'), 
    fs = require('fs'), 
    path = require('path'); 
var file = path.join(__dirname, 'file.txt'), 
    dest = path.join(__dirname, 'file1.txt'), 
    exists = Rx.Observable.fromCallback(fs.exists), 
    rename = Rx.Observable.fromNodeCallback(fs.rename), 
    stat = Rx.Observable.fromNodeCallback(fs.stat); 
exists(file) 
    .concatMap(function (flag) { 
    return flag ? 
     rename(file, dest) : 
     Rx.Observable.throw(new Error('File does not exist.')); 
    }) 
    .concatMap(function() { 
     return stat(dest); 
    }) 
    .forEach(
     function (fsStat) { 
      console.log(JSON.stringify(fsStat)); 
     }, 
     function (err) { 
      console.log(err); 
     } 
    ); 

concatMap scheint vielversprechend, aber der obige Code sieht ziemlich schrecklich. Ich hatte auch Probleme mit meinem Beispiel, weil Rx.Observable.fromPromise (p1) nicht funktionieren wird, da es eine Verheißung selbst erwartet, keine Funktion, und Rx.Observable.defer (p1) scheint keine Parameter wie die zu übergeben Beispiel.

Danke!

ähnliche Frage, aber ohne Datenübergabe: Chaining promises with RxJS

Answer 1

Ich habe nicht alles davon gelesen, aber wenn Sie das gleiche wie pl().then(p2).then(p3).then(console.log);, mit p wobei Funktion Rückkehr Versprechen erreichen wollen, könnten Sie so etwas wie (Beispiel tun here)

Rx.Observable.fromPromise(p1()) 
      .flatMap(function(p1_result){return p2(p1_result);}) 
      .flatMap(function(p2_result){return p3(p2_result);}) 

Oder die mehr symmetrisch:

var chainedPromises$ = 
    Rx.Observable.just() 
      .flatMap(p1) 
      .flatMap(p2) 
      .flatMap(p3); 

function rename (flag){ 
    return flag 
      ? rename(file,dest).flatMap(return Rx.Observable.just(dest)) 
      : Rx.Observable.throw(new Error('File does not exist.')); 
} 

Rx.Observable.just(file) 
      .flatMap(exists) 
      .flatMap(rename) 
      .flatMap(stat) 

Dieser Code ist nicht getestet, also halten Sie mich auf dem Laufenden, ob das funktioniert:Wenn Sie nun nacheinander Rückruf durch fromCallback oder fromNodeCallback gewickelt ausführen möchten, können Sie so etwas wie tun. Letzter Kommentar, dies sollte funktionieren, wenn an jedem Punkt nur ein Wert erzeugt wird (wie ein Versprechen). Wenn Sie mehrere Dateien anstelle von einem haben, können Sie unter flatMap Probleme mit der Bestellung bekommen (wenn Ihnen die Bestellung wichtig ist). In diesem Fall können Sie concatMap als Ersatz verwenden.