RxJS Promise 组合(传递数据)
Posted
技术标签:
【中文标题】RxJS Promise 组合(传递数据)【英文标题】:RxJS Promise Composition (passing data) 【发布时间】:2016-04-14 13:37:38 【问题描述】:我是 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) 无法正常工作,因为它需要一个承诺本身,而不是一个函数,并且 Rx.Observable.defer(p1) 似乎没有传递像例子。
谢谢!
类似的问题,但没有数据传递: Chaining promises with RxJS
【问题讨论】:
你的promise是否必须包装在一个函数中? 只有这样,如果你在 Promise 链之外定义了一个内联的 Promise,或者使用 const p1 = new Promise((resolve, reject) => ) 之类的东西可以观察到它,它会立即开始评估并且不能'不接收来自先前执行的承诺的数据。还是我对立即评估有误? 可能有帮助 - 使用 rxjs6 链接承诺的示例 - ***.com/a/55991374/1882064 【参考方案1】:我没有阅读全部内容,但如果你想实现与 pl().then(p2).then(p3).then(console.log);
相同的效果,p
是函数返回承诺,你可以做类似的事情(例如 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)
后面的代码未经测试,如果可行,请随时更新。
最后评论,如果在每一点你只产生一个值(如承诺),这应该有效。如果您有多个文件而不是一个文件,使用 flatMap
您可能会遇到订购问题(如果订购对您很重要),那么在这种情况下,您可以使用 concatMap
作为替代品。
【讨论】:
我有点希望有一个稍微更高的抽象,比如 flatMapAll(p1, p2, p3)。如果通过映射生成一系列承诺特别有用,例如const ps = map((x) => promiseFsReadFileCurriedSoThatItDoesSomethingWithPreviousFileData(x), ['1.txt','2.txt','3.txt']); Rx.Observable.just().flatMapAll(...ps); (只是伪代码)。但这绝对是一个可管理的解决方案,并且可能有一种方法可以通过映射 fromPromise 或其他东西来做到这一点。谢谢! 也没有测试第二个代码示例,但第一个代码就像一个魅力 你可以自己做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 ...
是的,正在发生类似的事情。这是我写的const flatMapAll = (...fns) => fns.reduce((acc, fn) => acc.flatMap(fn), Rx.Observable.just()); flatMapAll(p1, p2, p3).subscribe(console.log)
。谢谢您的帮助。另外,第一个例子是什么语言?看起来非常接近 haskell
它是伪代码,但是是的,sintax 的灵感来自 Haskell,至少是类型声明。以上是关于RxJS Promise 组合(传递数据)的主要内容,如果未能解决你的问题,请参考以下文章
如何将额外数据传递到 Parse Promise 链 [重复]