javascript - Chaining promises with RxJS - Stack Overflow

admin管理员组

文章数量:1193364

I'm new to RxJS and FRP in general. I had the idea of converting an existing promise chain in my ExpressJS application to be an observable for practice. I am aware that this probably isn't the best example but maybe someone can help shed some light.

What I'm trying to do:

1. I have two promises - prom1 and prom2
2. I want prom1 to run before prom2
3. If prom1 sends a reject(err), I want to cancel prom2 before it starts.
4. I want the error message prom1 returns to be available to the onError method on the observer.

---

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'); });

I'm new to RxJS and FRP in general. I had the idea of converting an existing promise chain in my ExpressJS application to be an observable for practice. I am aware that this probably isn't the best example but maybe someone can help shed some light.

What I'm trying to do:

1. I have two promises - prom1 and prom2
2. I want prom1 to run before prom2
3. If prom1 sends a reject(err), I want to cancel prom2 before it starts.
4. I want the error message prom1 returns to be available to the onError method on the observer.

---

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'); });

• javascript
• observable
• frp
• rxjs

Share Improve this question Follow edited Jun 5, 2015 at 3:02 M J 4,31733 gold badges3636 silver badges4646 bronze badges asked Apr 29, 2015 at 14:21 PathsofdesignPathsofdesign 5,03855 gold badges2323 silver badges2626 bronze badges 6

• Promises are not "run", they cannot be "started". What do you mean? – Bergi Commented Apr 29, 2015 at 14:32
• What promise library are you using, native promises? How does it support cancellation? – Bergi Commented Apr 29, 2015 at 14:34
• @Bergi - I'm using github.com/then/promise. When I say "run" or "started" I guess it's my way of saying when the observable is created and broadcasts to the subscriber. – Pathsofdesign Commented Apr 29, 2015 at 14:35
• flatMap for sure does (just as then) take a callback, not an observable (or promise respectively). – Bergi Commented Apr 29, 2015 at 14:35
• @Bergi - It looks like flatMap() does accept an observable. xgrommx.github.io/rx-book/content/core_objects/observable/…. Am I missing something? I'm not saying flatMap() is the answer it's just the only thing I know to try. That's why I need help. – Pathsofdesign Commented Apr 29, 2015 at 15:06

 |  Show 1 more comment

3 Answers 3

Sorted by: Reset to default 19

If I understood what you are trying to do - you need to create two deferred observables from functions that return promises and concat them:

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');
});

then for normal case:

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

And case where fisrt promise fails:

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

http://jsfiddle.net/cL37tgva/

flatMap turns an Observable of Observables into an Observable. It's used in many examples with Promises because often you have an observable and in the map function you want to create a promise for each "item" the observable emmits. Because every fromPromise call creates a new Observable, that makes it an "observable of observables". flatMap reduces that to a "flat" observable.

In your example you do something different, you turn a single promise into an observable and want to chain it with another observable (also created form a single promise). Concat does what you are looking for, it chains two observables together.

The error case will work as you would expect.

Observable.forkJoin works great here receiving array of other 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);
    });

本文标签： javascriptChaining promises with RxJSStack Overflow