javascript - RxJS Promise Composition(datos de paso)

functional-programming frp

Question

Soy nuevo en Rx y me resulta difícil encontrar documentación sobre promesas de composición, de modo que los datos de la primera promesa se pasen a la segunda y así sucesivamente. Aquí hay tres promesas muy básicas, los cálculos sobre los datos no son importantes, solo que hay que hacer algo asíncrono utilizando los datos de la promesa anterior.

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

La forma tradicional de lograr la composición de la que hablo:

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

Mi implementación favorita es composeP y pipeP de Ramda:

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

Parece probable que Rx pueda manejar este tipo de situación con bastante fluidez. Sin embargo, lo más cercano que he encontrado hasta ahora es de la comparación de RxJS a async (biblioteca) aquí 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 parece prometedor, pero el código anterior parece bastante horrible. También tuve problemas con mi ejemplo porque Rx.Observable.fromPromise (p1) no funcionará porque espera una promesa en sí misma, no una función, y Rx.Observable.defer (p1) no parece pasar parámetros como el ejemplo.

¡Gracias!

Pregunta similar pero sin pasar datos: encadenar promesas con RxJS

Answer 1

No leí todo, pero si desea lograr lo mismo que pl().then(p2).then(p3).then(console.log); , siendo p función devolviendo promesas, podrías hacer algo como (ejemplo here )

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

O lo más simétrico:

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

Ahora, si desea ejecutar la devolución de llamada secuencialmente a través de fromCallback o fromNodeCallback , podría hacer algo como:

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)

El último código no está probado, así que mantenme actualizado si eso funciona. Último comentario, esto debería funcionar si en cada punto solo tiene un valor producido (como una promesa). Si tuviera varios archivos en lugar de uno, con flatMap podría tener problemas con el pedido (si el pedido le importa), por lo que, en ese caso, podría usar concatMap como reemplazo.