RxJS Skład obietnicy (przekazywanie danych)

Question

Jestem nowy dla Rx i mam trudności ze znalezieniem dokumentacji dotyczącej komponowania obietnic w taki sposób, że dane z pierwszej obietnicy są przekazywane do drugiego i tak dalej. Oto trzy bardzo podstawowe obietnice, obliczenia na danych nie są ważne, tylko że coś asynchronicznego musi zostać wykonane przy użyciu danych z poprzedniej obietnicy.

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)); 
}; 

Tradycyjnym sposobem osiągnięcia składu mówię:

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

Moja ulubiona wdrożenia jest Ramda za composeP i pipeP:

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

Wydaje się prawdopodobne, Rx może być w stanie radzić sobie z taką sytuacją dość płynnie. Jednak najbliżej znalazłem tak daleko jest od RxJS do asynchroniczny (biblioteka) porównanie tutaj 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 Wygląda obiecująco, ale powyższy kod wygląda całkiem przerażające. Miałem też problem z moim przykładem, ponieważ Rx.Observable.fromPromise (p1) nie będzie działał, ponieważ oczekuje obietnicy, a nie funkcji, a Rx.Observable.defer (p1) nie wydaje się przekazywać parametrów takich jak przykład.

Dzięki!

podobne pytanie, ale bez przechodzenia danych: Chaining promises with RxJS

Answer 1

Nie czytałem to wszystko, ale jeśli chcesz, aby osiągnąć taki sam jak pl().then(p2).then(p3).then(console.log); z p jest funkcja zwracająca obietnic, można zrobić coś takiego (przykład here)

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

lub bardziej symetryczny:

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

Teraz jeśli chcesz wykonać kolejno zwrotnego owinięty przez fromCallback lub fromNodeCallback, można zrobić coś takiego:

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) 

Drugi kod jest niesprawdzone, więc informuj mnie czy zadziała. Ostatnia uwaga, powinno to zadziałać, jeśli w każdym punkcie masz tylko jedną wyprodukowaną wartość (jak obietnica). Gdybyś miał kilka plików zamiast jednego, z flatMap możesz uzyskać problemy z zamawianiem (jeśli zamówienie jest dla ciebie ważne), więc w takim przypadku możesz użyć zamiennika jako concatMap.