Цепочка promises с RxJS

Я новичок в RxJS и FRP в целом. У меня была идея превратить существующую цепочку обещаний в приложение ExpressJS, чтобы быть наблюдаемой для практики. Я знаю, что это, вероятно, не лучший пример, но, возможно, кто-то может помочь пролить свет.

Что я пытаюсь сделать:

У меня есть два promises - prom1 и prom2
Я хочу, чтобы prom1 запускался до prom2
Если prom1 отправляет отклонение (err), я хочу отменить 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'); });

Ответ 1

Если бы я понял, что вы пытаетесь сделать - вам нужно создать два отложенных наблюдаемых из функций, возвращающих promises и concat их:

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/

Ответ 2

flatMap превращает Наблюдаемое Наблюдение в Наблюдаемое. Он используется во многих примерах с помощью Promises, потому что часто у вас есть наблюдаемая и в функции карты вы хотите создать обещание для каждого "элемента" наблюдаемого emmits. Потому что каждый вызов fromPromise создает новый Observable, что делает его "наблюдаемым для наблюдаемых". flatMap уменьшает это до "плоского" наблюдаемого.

В вашем примере вы делаете что-то другое, вы превращаете одно обещание в наблюдаемое и хотите связать его с другим наблюдаемым (также созданным формальным обещанием). Concat выполняет то, что вы ищете, он объединяет две наблюдаемые вместе.

Случай с ошибкой будет работать так, как вы ожидали.

Ответ 3

Observable.forkJoin отлично работает здесь, получая массив других Observables.

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