RxJS Promises Chaining

Question

I'm pretty new to RxJS. So I have a stream that creates the payload for different ajax calls, then I use a flatMap to retrieve the data I need and it works fine. Easy.

const streamA = Rx.Observable.from(array);

const streamB = streamA
  .map( val => /* build payload */ )
  .flatMap( payload => Rx.Observable.fromPromise($.ajax(payload))

streamB.subscribe( result => /* got it */)

Now I would create an array of payloads for every item, but the problem is that now when I subscribe the stream I get every single request back, but I would have only the initial elements to be returned when completed.

const streamC = streamA
  .flatMap( payloads => {
    return Rx.Observable.from(payloads)
      .flatMap( payload => Rx.Observable.fromPromise($.ajax(payload))

streamC.subscribe( result => /* executed for every payload */)

I tried with the groupBy that returns the proper grouped array and showed me that I can chain Observable, but I still can't figure out how properly subscribe the observers to get the elements fulfilled.

const streamWLF = streamA
  .flatMap( payloads => {
    return Rx.Observable.from(payloads)
      .flatMap( payload => Rx.Observable.fromPromise($.ajax(payload))
      .groupBy((obs) => obs.key, (obs) => obs)

streamWLF.subscribe( result => {
  result.subscribe(/* did my magic here*/);
})

So my question is, which is the best way to do that?

Does the main Stream subscribed always triggered when a subStream item is received?

And, if is possible, how can I subscribe subStream to trigged the mainStream subscribed only when subStream is completed?

I'm not clear on your issue. What do you mean you "would have only the initial elements to be returned when completed"? Promises only return a single item, Are you trying to associate the results of each set of payloads together somehow? — paulpdaniels
@paulpdaniels yes, I have items containing an array of payloads, and i want to have back, not all the payloads in a stream, but the elements with the returned payloads inside. — user2131283

paulpdaniels paulpdaniels · Accepted Answer · 2016-02-10T17:54:29

I think you are looking for the forkJoin operator.

const streamC = streamA
  .flatMap(payloads => Rx.Observable.forkJoin(payloads));

streamC.subscribe( results => /*An array containing the results from all payloads*/);

RxJS Promises Chaining

1 Answers