javascript - RxJS Promise Composition(传递数据)

Question

我是 Rx 的新手，我发现很难找到关于组合 promise 的文档，以便将来自第一个 promise 的数据传递到第二个等等。这是三个非常基本的 promise ，对数据的计算并不重要，只是必须使用先前 promise 中的数据完成一些异步操作。

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

实现我所说的构图的传统方式:

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

我最喜欢的实现是 Ramda 的 composeP 和 pipeP:

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

看起来 Rx 可能能够非常流畅地处理这种情况。然而，到目前为止我发现的最接近的是从 RxJS 到异步(库)的比较这里 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 看起来很有前途，但上面的代码看起来很可怕。我的示例也遇到了问题，因为 Rx.Observable.fromPromise(p1) 不会工作，因为它期望一个 promise 本身，而不是一个函数，并且 Rx.Observable.defer(p1) 似乎没有像示例。

谢谢!

类似的问题，但没有数据传递: Chaining promises with RxJS

Answer 1

我没有读完所有内容，但如果你想实现与 pl().then(p2).then(p3).then(console.log); 相同的效果，请使用p 是返回 promise 的函数，你可以做类似的事情(例如 here )

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

或者更对称的:

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

现在，如果你想执行通过 fromCallback 或 fromNodeCallback 包装的顺序回调，你可以这样做:

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)

后面的代码未经测试，所以如果可行的话请让我更新。最后评论，如果在每一点你只产生一个值(value)(比如 promise )，这应该有效。如果你有多个文件而不是一个文件，使用 flatMap 你可能会遇到排序问题(如果顺序对你很重要)，那么在这种情况下，你可以使用 concatMap 作为替换。