RxJS Promise构图（传递数据） [英] RxJS Promise Composition (passing data)

问题描述

我是Rx品牌的新成员，并且发现很难找到有关撰写承诺的文档，以便将第一个承诺中的数据传递到第二个承诺中，等等。这里有三个非常基本的承诺，对数据的计算并不重要，只需要使用前一个承诺中的数据就可以完成异步操作。

  const p1 =（）=> Promise.resolve（1）; 
 const p2 = x => {const val = x + 1;返回Promise.resolve（val）; }; 
 const p3 = x => {
 const isEven = x => x％2 === 0; 
返回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'），$ b $ 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 （
函数（fsStat）{
 console.log（JSON.stringify（fsStat））; 
}，
函数（err）{
 console.log（err ）; 
} 
）; 
  

concatMap看起来很有前景，但上面的代码看起来非常可怕。我也遇到了我的例子的问题，因为Rx.Observable.fromPromise（p1）不会工作，因为它期望一个承诺本身，而不是一个函数，而Rx.Observable.defer（p1）似乎不会像例如。

谢谢！

类似的问题，但没有数据传递：
与RxJS承诺承诺

解决方案

我没有阅读所有的内容，但是如果你想达到和 pl code>，其中 p 是函数返回的promise，你可以做一些事情（例如

  Rx.Observable.fromPromise（p1（））
 .flatMap（function（p1_result）{return p2（p1_result）;}）
 .flatMap（function（p2_result）{return p3（p2_result）;}）
  pre> 
 
 
或者更对称：
 
 pre $ $ $ $ 
 Rx.Observable.just（
 .flatMap（ p1）
 .flatMap（p2）
 .flatMap（p3）; 
  

现在，如果您想要按顺序执行回调，则通过 fromCallback 或 fromNodeCallback ，你可以这样做：

 函数重命名（标志）{
返回标志
？重命名（file，dest）.flatMap（返回Rx.Observable.just（dest））
：Rx.Observable.throw（new Error（'File does not exist。'））; 
 
 
 Rx.Observable.just（文件）
 .flatMap（存在）
 .flatMap（重命名）
 .flatMap（stat）
  

后面的代码没有经过测试，所以如果可以的话，让我更新。
最后的评论，如果在每一点你只有一个产生的价值（如承诺），这应该起作用。如果你有几个文件而不是一个文件，使用 flatMap ，你可能会遇到订购问题（如果订单对你很重要），那么在这种情况下，你可以使用 concatMap 作为替换。

I'm brand new to Rx and am finding it difficult to find documentation on composing promises such that data from the first promise is passed into the second and so on. Here's three very basic promises, the calculations on the data aren't important, just that something async has to be done using data from the previous 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));
 };

The traditional way to achieve the composition I'm talking about:

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

My favorite implementation is Ramda's composeP and pipeP:

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

It seems likely Rx might be able to handle this kind of situation pretty fluently. However, the closest I've found so far is from the RxJS to async (library) comparison here 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 seems promising, but the above code looks pretty horrific. I was also having trouble with my example because Rx.Observable.fromPromise(p1) won't work as it expects a promise itself, not a function, and Rx.Observable.defer(p1) doesn't seem to pass parameters like the example.

Thanks!

Similar question but without data passing: Chaining promises with RxJS

解决方案

I did not read all of it, but if you want to achieve the same as pl().then(p2).then(p3).then(console.log);, with p being function returning promises, you could do something like (example here)

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

Or the more symmetric :

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

Now if you want to execute sequentially callback wrapped through fromCallback or fromNodeCallback, you could do something like :

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)

The latter code is untested, so keep me updated if that works. Last comment, this should work if at each point you only have one value produced (like a promise). If you would have several files instead of one, with flatMap you might get ordering issues (if order matters to you), so in that case, you could use concatMap as a replacement.

这篇关于RxJS Promise构图（传递数据）的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！