用RxJS链接承诺

Question

我是RxJS和FRP的新手.我想到将ExpressJS应用程序中现有的promise链转换为可观察的实践.我知道这可能不是最好的例子,但也许有人可以帮助解决一些问题.

我正在做的事情:

1. 我有两个承诺 - prom1和prom2
2. 我希望prom1在prom2之前运行
3. 如果prom1发送拒绝(错误),我想在启动之前取消prom2.
4. 我希望错误消息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'); });

Answer 1

如果我理解你要做什么 - 你需要从返回promises的函数创建两个延迟的observable并将它们连接起来:

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

并且fisrt承诺失败的情况:

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

http://jsfiddle.net/cL37tgva/

Answer 2

flatMap将Observable of Observable变为Observable.它在Promises的许多例子中使用,因为通常你有一个observable,在map函数中你想为每个"item"创建一个可观察的emmits的promise.因为每个fromPromise调用都会创建一个新的Observable,这使它成为"可观察的可观察对象".flatMap将其缩小为"平坦"可观察量.

在您的示例中,您执行不同的操作,将单个promise转换为observable,并希望将其与另一个observable链接(也可以从单个promise中创建).Concat做你想要的,它将两个可观察的链子连在一起.

错误情况将按预期工作.