Promise.all() with RxJS

Question

I'm writing an app in Angular 2 and I want to execute several http requests and run a function on the responses.

In Angular 1, I would write something like $q.all([$http.get(...), $http.get(...), ...]).then(doSomethingWithResponses);

But Angular 2 returns RxJS Observables and after a bunch of reading I still can't figure out how to get the responses of several http requests. How can this can be done?

Hawkbill

Use forkJoin, that's the equivalent for Promise.all – Diplopod

I have a live forkJoin sample here syntaxsuccess.com/viewarticle/angular-2.0-and-http – Brahmana

Answer 1

As @Eric Martinez pointed out, there is forkJoin. forkJoin runs all observable sequences in parallel and collect their last elements.

Rx.Observable.forkJoin([a,b]).subscribe(t=> {
        var firstResult = t[0];
        var secondResult = t[1];
});

Pressroom

@EricMartinez is there a way for an index in the resulting array to "dissolve" if Observable a or Observable b fail (or are canceled or have an error or don't meet a condition)? So for example, if the Observable b fails for data1, then the length of the resulting array would be 1 instead accounting for the 2 calls that were made inside the Observable.forkJoin( ... ) argument block? – Capriccio

Written a RxJs v6 solution here using zip - https://mcmap.net/q/180891/-promise-all-behavior-with-rxjs-observables – Rural

Answer 2

I'm not sure you'd want to use forkJoin/zip, especially considering combineLatest is easier to understand and will emit on every sub-stream event, whereas forkJoin basically samples on every sub-stream having emitted.

This might come to bite you later when you want to combine multi-item Observables down the road.

Folse

The difference is that you don't want to be notified of every emit on every stream; you only care about when all streams have emitted data. Think of it as the observable equivalent to Promise.all(promises) or $q.all(promises). – Turley

Answer 3

Wouldn't a merge work? You can subscribe and attach a handler to the onComplete callback.

I first build an array of my observables and then use static merge:

let obs_ary: any = [obs1, obs2, obs3];
Observable.merge(...obs_ary);

Gourde

Answer 4

I'm learning RxJS and I was trying to do the same thing with RxJS v5

It seems like we don't have forkJoin on v5 anymore, so here's how I got it working (works with flatMap or mergeMap, which are aliases):

const callOne = value =>
    new window.Promise(resolve =>
        setTimeout(() => resolve(value + 10), 3000)
    );

const callTwo = value =>
    new window.Promise(resolve =>
        setTimeout(() => resolve(value + 20), 1000)
    );

Rx.Observable
    .of(2)
    .do(() => console.log('querying...'))
    .mergeMap(number =>
        Rx.Observable.zip(
            Rx.Observable.fromPromise(callOne(number)),
            Rx.Observable.fromPromise(callTwo(number))
        )
    ).concatAll()
    .subscribe(createSubscriber('promises in parallel'));

Ledaledah

I'm looking through the changelog of rxjs and I don't see anything about forkJoin being deprecated github.com/ReactiveX/rxjs/blob/master/CHANGELOG.md – Hawkbill

@Hawkbill did you get it to run on v5? I get an error when trying to use forkJoin and I don't see it under src/operator. – Ledaledah

I didn't try running it, but it looks like v5 is still in beta. Perhaps that could have caused the issue. Happy to set this answer as correct if forkJoin really is gone – Hawkbill

Yes, it's in Beta but at least for now the forkJoin operator is gone. I really just posted this answer here cause I thought it would be helpful for those trying this on the latest version of RxJS. – Ledaledah

Cool, yea! Maybe it was moved? I'm idly curious – Hawkbill

Strange cause I can see it in the docs and searching the repo, I can still find references to it, so now I'm confused =/ – Ledaledah

forkJoin is static so it lives under src/observable not src/operator. – Handle