How to use the 'flatmap' operator conditionally ? (Angular 2/rxjs)

Question

What I'm trying to achieve is to run a series of observables conditionally.

return observable.map(response => response)
       .flatmap(response1 => observable1(response1))
       .flatmap(response2 => observable2(response2))
       .flatmap(response3 => observable3(response3))

I need to check the response1 and invoke the remaining observables if needed, else I need to return response1 and break the execution and so on.

I've gone through the following SO questions but they doesn't seem to answer my question

Conditionally choose observable in RxJS

RXJS if with observable as conditional

Handle Error in RxJs flatMap stream and continue processing

I'm new to rxjs, so forgive me if the question seems too lame.

Any help will be appreciated.

Thanks

Answer 1

This is like calling multiple consecutive HTTP requests where results depend on each other. This is where you want to use concatMap() instead because it'll wait until the Observable returned from the callback function completes.

It's hard to tell what exactly you want to do from your description but if you need to stop propagating the results (and avoid calling unnecessary HTTP requests) have a look takeWhile() operator or simple filter() operators.

With filter():

return observable
    .concatMap(response => response)
    .filter(response1 => response1.whatever === true)
    .concatMap(response1 => observable1(response1))
    .filter(response2 => response2.whatever === true)
    .concatMap(response2 => observable2(response2))
    .filter(response3 => response3.whatever === true)

This simply won't propagate the item further if it fails the filter test. However the source observable can still emit values that'll go through the chain. In other words the filter operator doesn't complete the chain.

With takeWhile():

return observable
    .concatMap(response => response)
    .takeWhile(response1 => response1.whatever === true)
    .concatMap(response1 => observable1(response1))
    .takeWhile(response2 => response2.whatever === true)
    .concatMap(response2 => observable2(response2))
    .takeWhile(response3 => response3.whatever === true)

If any of the takeWhile() result into false it'll complete the chain and no other value can be emitted.

Answer 2

You could do something like:

observable.map(res => res.json())
  .flatMap(json => {
      if (json.something === 'something') {
          return this.http.get('/anotherCall')
              .map(res => 'second response: ' + res);
      } else {
          return Observable.from(['just return something else']);
      }
  });

It's only two calls but you get the point.

Answer 3

I did not find easy way to bypass operators, if you decide to skip n last operators, you still have to pass last valid response through all operators:

 return observable
   .flatmap(r => {
        return r.value > 1 ? observable1(r) : Observable.of(r);
   })
   .flatmap(r => {
        return r.value > 2 ? observable2(r) : Observable.of(r);
   })
   .flatmap(r => {
        return r.value > 3 ? observable3(r) : Observable.of(r);
   })

Answer 4

Alternatively, you can combine all the observables into one single observable and subscribe to that as below,

return response
         .concatMapTo(response1,(response2, response3) => 
             `${response2} ${response3}`
          );

Subscribe to it using

data = response.subscribe(val => console.log(val)); // contains all three responses value

Still you can access each one separately using the response2 and response3 named instance.

Subscribing for a single object using different observables: If you are looking to combine data and properties from different observables you can have a look at this answer