RxJS Promise组合(传递数据)

18

我刚接触Rx,发现很难找到有关如何组合承诺的文档,以便第一个承诺的数据传递到第二个承诺等。以下是三个非常基本的承诺,数据的计算并不重要,只要使用来自前一个承诺的数据执行一些异步操作即可。

 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 到 async(库)比较 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)不会像期望的那样工作,它希望一个承诺本身而不是一个函数,并且Rx.Observable.defer(p1)似乎不像例子传递参数。

谢谢!

类似的问题,但没有数据传递: 使用RxJS链接承诺


你的承诺是否必须包装在一个函数中? - user3743222
只有在定义了一个内联的 Promise,例如 const p1 = new Promise((resolve, reject) => {}),它会立即开始评估,并且无法从先前执行的 Promise 接收数据。或者我对立即评估的理解是错误的吗? - low_ghost
可能有帮助 - 使用rxjs6链接承诺的示例 - https://dev59.com/QlsW5IYBdhLWcg3w2aOK#55991374 - arcseldon
1个回答

26

我没有完全阅读它,但如果你想要实现与pl().then(p2).then(p3).then(console.log);相同的效果,其中p是返回Promise的函数,你可以像这样做(示例在这里

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

现在,如果你想按顺序执行通过 fromCallbackfromNodeCallback 封装的回调函数,你可以这样做:

现在,如果您想依次执行通过 fromCallbackfromNodeCallback 包装的回调函数,可以像这样操作:

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)

后面的代码没有经过测试,如果可以工作,请让我知道更新情况。 最后一条评论,如果每个点只产生一个值(如承诺),则应该可以工作。如果您有多个文件而不是一个文件,则使用flatMap可能会出现排序问题(如果顺序很重要),在这种情况下,您可以使用concatMap来替换。


我有点希望只是稍微更高一些的抽象,比如像flatMapAll(p1, p2, p3)这样的东西。如果通过map生成了一系列的promise,那么这将特别有帮助,例如 const ps = map((x) => promisedFsReadFileCurriedSoThatItDoesSomethingWithPreviousFileData(x), ['1.txt','2.txt','3.txt']); Rx.Observable.just().flatMapAll(...ps);(这只是伪代码)。但这绝对是可行的解决方案,很可能有一种方法可以通过从Promise映射或其他方式来实现。谢谢! - low_ghost
也没有测试第二个代码示例,但第一个代码示例运行得非常好。 - low_ghost
你可以自己实现 flatMapAllflatMapAll :: Rx.Observable -> [a ->a] -> Rx.ObservableflatMapAll = (source, fn_array) -> fn_array.reduce((acc, fn) -> acc.flatMap(fn), source)。在 js 中,Rx.Observable.prototype.flatMapAll = function (fn_array) {source = this; return ...} - user3743222
是的,我也想到了类似的东西。这里是我写的代码:const flatMapAll = (...fns) => fns.reduce((acc, fn) => acc.flatMap(fn), Rx.Observable.just()); flatMapAll(p1, p2, p3).subscribe(console.log)。感谢您的帮助。另外,第一个例子是用什么语言写的?看起来非常接近 Haskell。 - low_ghost
这是伪代码,但语法至少受到了 Haskell 的类型声明的启发。 - user3743222

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接