RxJS Promise 组合(传递数据)

IT技术 javascript functional-programming rxjs frp ramda.js
2021-02-13 10:49:51

我是 Rx 的新手,我发现很难找到有关撰写Promise的文档,以便将第一个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

1个回答

我没有阅读所有内容,但是如果您想实现与 相同的功能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);

现在,如果您想通过fromCallbackor执行顺序回调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)

后一个代码未经测试,所以如果有效,请让我更新。最后一条评论,如果在每个点您只产生一个值(如Promise),这应该有效。如果您有多个文件而不是一个文件,flatMap那么您可能会遇到排序问题(如果订单对您很重要),那么在这种情况下,您可以将其concatMap用作替代品。

我有点希望有一个稍微高一点的抽象,比如 flatMapAll(p1, p2, p3)。如果一系列 promise 是通过映射生成的,则特别有用,例如 const ps = map((x) => promiseFsReadFileCurriedSoThatItDoesSomethingWithPreviousFileData(x), ['1.txt','2.txt','3.txt']); Rx.Observable.just().flatMapAll(...ps); (只是伪代码)。但这绝对是一个可管理的解决方案,并且可能有一种方法可以通过映射 fromPromise 或其他东西来做到这一点。谢谢!
2021-03-15 10:49:51
是的,正在发生类似的事情。这是我写的const flatMapAll = (...fns) => fns.reduce((acc, fn) => acc.flatMap(fn), Rx.Observable.just()); flatMapAll(p1, p2, p3).subscribe(console.log)谢谢您的帮助。另外,第一个例子是什么语言?看起来很接近haskell
2021-03-18 10:49:51
它是伪代码,但是是的 sintax 是受 Haskell 启发的,至少是类型声明。
2021-03-23 10:49:51
也没有测试第二个代码示例,但第一个代码示例非常有用
2021-04-02 10:49:51
你可以flatMapAll自己flatMapAll :: Rx.Observable -> [a ->a] -> Rx.Observable. flatMapAll = (source, fn_array) -> fn_array.reduce((acc, fn) -> acc.flatMap(fn), source). 在js中,Rx.Observable.prototype.flatMapAll = function (fn_array) {source = this; return ...}
2021-04-13 10:49:51