I'm learning RxJS and trying to replace some code I did with promises before, but I'm struggling to find the right way to combine observables.

I have a code that calls multiple APIs and for each API call executes the same code. Also, after API calls are done I have some other code that is supposed to be executed only then and only once. I did this with Promises, I made multiple Promises and for each Promise I execute the same code, and on Promise.all() I just do the other code.

So I tried to do this with RxJS:

var apiCallIterator = ["id1", "id2", id3];
var apiCallObservable = from(apiCallIterator)
  .pipe(
    mergeMap(apiCallIterator => {
      return makeAnApiCall();
    })
   );

apiCallObservable.subscribe({
next: value => {
 // do some code for each API call
},
error: () => {
 // api call failed
}
})

This works fine for the first part of my task, it will execute the same code after each API call is done. But I can't find a way to get all results in one spot after all API calls are done.

Maybe I'm doing this wrong from the beginning, but this part seems to work fine for me.

I'm learning RxJS and trying to replace some code I did with promises before, but I'm struggling to find the right way to combine observables.

I have a code that calls multiple APIs and for each API call executes the same code. Also, after API calls are done I have some other code that is supposed to be executed only then and only once. I did this with Promises, I made multiple Promises and for each Promise I execute the same code, and on Promise.all() I just do the other code.

So I tried to do this with RxJS:

var apiCallIterator = ["id1", "id2", id3];
var apiCallObservable = from(apiCallIterator)
  .pipe(
    mergeMap(apiCallIterator => {
      return makeAnApiCall();
    })
   );

apiCallObservable.subscribe({
next: value => {
 // do some code for each API call
},
error: () => {
 // api call failed
}
})

This works fine for the first part of my task, it will execute the same code after each API call is done. But I can't find a way to get all results in one spot after all API calls are done.

Maybe I'm doing this wrong from the beginning, but this part seems to work fine for me.

• javascript
• rxjs

4 Answers 4

Most easily just add toArray():

from(apiCallIterator)
  .pipe(
    mergeMap(val => makeAnApiCall()),
    toArray(), // collect all results
  )
  .subscribe(allResults => ...); // allResults.forEach(...) to iterate all results

You could also use forkJoin but then you'd have to prepare an array of all Observables source beforehand. However, this is only a matter of your preference.

https://stackblitz.com/edit/rxjs6-demo-qbvhjh?file=index.ts

Using forkJoin you can achieve this

forkJoin is an operator that takes any number of Observables which can be passed either as an array or directly as arguments. If no input Observables are provided, resulting stream will complete immediately.

forkJoin will wait for all passed Observables to complete and then it will emit an array with last values from corresponding Observables. So if you pass n Observables to the operator, resulting array will have n values, where first value is the last thing emitted by the first Observable, second value is the last thing emitted by the second Observable and so on.

I think you need forkJoin operator for this.

This operator is best used when you have a group of observables and only care about the final emitted value of each. One common use case for this is if you wish to issue multiple requests on page load (or some other event) and only want to take action when a response has been receieved for all. In this way it is similar to how you might use Promise.all. https://www.learnrxjs.io/operators/combination/forkjoin.html

For your code :

var apiCallIterator = ["id1", "id2", id3];
var apiCallObservable = from(apiCallIterator)
  .pipe(
    mergeMap(apiCallIterator => {
      return makeAnApiCall();
    })
   );

forkJoin(apiCallObservable)
  .subscribe( 
     (combinedCalls) => {
        console.log(combinedCalls) //receive all the values at once
     },
     (error) => {
     // api call failed
     }
   })

You can zip() them all which will only emit once all the observables complete.

var apiCallIterator = ['id1', 'id2', 'id3'];

var apiCallObservable = zip(apiCallIterator.map(id => makeAnApiCall(id)))
  .subscribe({
    next: value => {
      // do some code for each API call
    },    
    error: () => {    
      // api call failed    
    }    
  })

makeAnApiCall(id) {
    return from(id).pipe(tap(() => {
     // do some code for each API call
    }));
}

also a great site to learn rxjs - https://www.learnrxjs.io/