使用 RxJS 链接承诺答案

【问题标题】：Chaining promises with RxJS使用 RxJS 链接承诺
【发布时间】：2015-07-08 21:40:19
【问题描述】：

总的来说，我是 RxJS 和 FRP 的新手。我的想法是将我的 ExpressJS 应用程序中的现有承诺链转换为可观察的练习。我知道这可能不是最好的例子，但也许有人可以帮助阐明一下。

我想做的事：

我有两个承诺 - prom1 和 prom2
我希望 prom1 在 prom2 之前运行
如果 prom1 发送拒绝（错误），我想在 prom2 开始之前取消它。
我希望 prom1 返回的错误消息可用于观察者的 onError 方法。

var prom1 = new Promise(function(resolve, reject) {
    if (true) {
       reject('reason');
    }
    resolve(true);
});

var prom2 = new Promise(function(resolve, reject) {
    resolve(true);
});

// What do I do here? This is what I've tried so far...
var source1 = Rx.Observable.fromPromise(prom1);
var source2 = source1.flatMap(Rx.Observable.fromPromise(prom2));

var subscription = source2.subscribe(
    function (result) { console.log('Next: ' + result); },

    // I want my error 'reason' to be made available here
    function (err) { console.log('Error: ' + err); },

    function () { console.log('Completed'); });

【问题讨论】：

Promise 不是“运行”的，它们不能被“启动”。什么意思？
你使用的是什么承诺库，原生承诺？它如何支持取消？
@Bergi - 我正在使用github.com/then/promise。当我说“运行”或“开始”时，我想这是我在创建可观察对象并向订阅者广播时的说法。
flatMap 肯定会（就像then）接受一个回调，而不是一个可观察的（或分别的承诺）。
@Bergi - 看起来 flatMap() 确实接受了一个 observable。 xgrommx.github.io/rx-book/content/core_objects/observable/…。我错过了什么吗？我不是说 flatMap() 是答案，它只是我知道要尝试的唯一方法。这就是我需要帮助的原因。

标签： javascript observable frp rxjs

【解决方案1】：

Observable.forkJoin 在接收其他 Observable 数组时效果很好。

Rx.Observable.forkJoin([this.http.get('http://jsonplaceholder.typicode.com/posts'), this.http.get('http://jsonplaceholder.typicode.com/albums')]).subscribe((data) => {
      console.log(data);
    });

【讨论】：

forkJoin 在这里可能不是正确的选择，因为 forkJoin 允许它的所有 observables 并行运行。 OP 希望 observables 顺序运行。

【解决方案2】：

如果我理解你想要做什么 - 你需要从返回 promise 的函数创建两个延迟的 observables 并连接它们：

var shouldFail = false;

function action1() {
    return new Promise(function (resolve, reject) {    
        console.log('start action1');
        if (shouldFail) {
            reject('reason');
        }
        resolve(true);
    });
}

function action2() {
    return new Promise(function (resolve, reject) {    
        console.log('start action2');
        resolve(true);
    });
}

var source1 = Rx.Observable.defer(action1);
var source2 = Rx.Observable.defer(action2);

var combination = Rx.Observable.concat(source1, source2);

var logObserver = Rx.Observer.create(

function (result) {
    console.log('Next: ' + result);
},

function (err) {
    console.log('Error: ' + err);
},

function () {
    console.log('Completed');
});

那么对于正常情况：

combination.subscribe(logObserver);
// start action1
// Next: true
// start action2
// Next: true
// Completed

第一个承诺失败的情况：

shouldFail = true;
combination.subscribe(logObserver);
// start action1
// Error: reason

http://jsfiddle.net/cL37tgva/

【讨论】：

谢谢！我发现我不需要使用 defer() 位，但可以将两个 promise 函数传递给 concat() 并获得我需要的结果。我这样做有什么遗漏吗？
在这种情况下，两个 Promise 都将立即创建（启动）（在您调用 concat 时，但不是在您实际订阅时），并且第二个 Promise 不会在开始前等待第一个 Promise 解决。
使用 defer()s，如果第二个 promise 先于第一个返回怎么办？这是 concat() 保持它们有序的地方吗？
看起来 concat(a,b,c) 足够聪明，可以同步运行。即使b先完成，它仍然按照a、b、c的顺序运行。

【解决方案3】：

flatMap 将 Observable 的 Observable 转换为 Observable。它在许多带有 Promises 的示例中使用，因为通常您有一个 observable，并且在 map 函数中，您希望为 observable emmits 的每个“项目”创建一个 Promise。因为每个 fromPromise 调用都会创建一个新的 Observable，这使它成为“可观察对象的可观察对象”。 flatMap 将其简化为“平面”可观察对象。

在您的示例中，您做了一些不同的事情，您将单个 promise 转换为 observable，并希望将它与另一个 observable 链接起来（也是从单个 promise 创建的）。 Concat 可以满足您的需求，它将两个可观察对象链接在一起。

错误案例将按您的预期工作。

【讨论】：