Promise.all() с RxJS

Я пишу приложение в Angular 2, и я хочу выполнить несколько HTTP-запросов и запустить функцию для ответов.

В Angular 1 я бы написал что-то вроде $q.all([$http.get(...), $http.get(...), ...]).then(doSomethingWithResponses);

Но Angular 2 возвращает RxJS Observables и после кучи чтения я все еще не могу понять, как получить ответы нескольких HTTP-запросов. Как это можно сделать?

Ответ 1

Как отметил @Eric Martinez, есть forkJoin. forkJoin запускает все наблюдаемые последовательности параллельно и собирает их последние элементы.

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

Ответ 2

Я не уверен, что вы захотите использовать forkJoin/zip, особенно учитывая combineLatest легче понять и будет выделяться на каждом событии субпотока, тогда как forkJoin в основном сэмплы на каждом подпотоке, испущенном.

Это может привести к укусу вас позже, когда вы захотите объединить многозначные наблюдения в будущем.

Ответ 3

Не работает merge? Вы можете подписаться и прикрепить обработчик к обратному сообщению onComplete.

Сначала я создаю массив моих наблюдаемых, а затем статический merge:

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

Ответ 4

Я изучаю RxJS, и я пытался сделать то же самое с RxJS v5

Похоже, у нас больше нет forkJoin на v5, так вот как я его работал (работает с flatMap или mergeMap, которые являются псевдонимами):

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