Observables: Chaining promises & firing events on each promise

Question

I just read this SO question RxJS Promise Composition (passing data) and had a question.

The setting is the same as the aforementioned question: I have 3 promises.

 const p1 = () => Promise.resolve(1);
 const p2 = x => { const val = x + 1; return Promise.resolve(val); };
 const p3 = x => {
      const isEven = x => x % 2 === 0;
      return Promise.resolve(isEven(x));
 };

And I'm chaining them, as per the answer, like this:

 var chainedPromises$ = 
     Rx.Observable.just()
             .flatMap(p1)
             .flatMap(p2)
             .flatMap(p3);

However, chainedPromises$ only fires up once, at the end of all the promise chain, with the resolved value true. Namely:

chainedPromises$.subscribe(function(x){console.log(x)}); // Logs `true`

How can I change chainedPromises$ so that it fires once at the end of each promise? Ie:

chainedPromises$.subscribe(function(x){console.log(x)});
// Logs `1`
// Logs `2`
// Logs `true`

Observable & Promises are not the same thing, what your trying to do here doesn't make sense. If your going to make an Oserable return multiple results, then you fire them off using `next` — Keith, Apr 11 '18 at 15:28

score 2 · Accepted Answer · answered Apr 11 '18 at 18:19

Looks like you need to emit the generated value and use it in the next step of the chain. I don't know if there's sugar for this, but I got it working by making this distinction explicit.

Observable
  .fromPromise(p1())
  .flatMap(val => Observable.merge(
    Observable.fromPromise(p2(val))
      .flatMap(val => Observable.merge(
        Observable.fromPromise(p3(val)),
        Observable.of(val)
      )),
    Observable.of(val)
  ))
  .subscribe(console.log);

You can see it working.

martin · Answer 2 · 2018-04-11T16:12:41.793

1

Just don't use mergeMap that is made to "map" one value to another and use concat that subscribes to an Observable only if the previous Observable completed. This way you'll achieve the same effect where you process Observables one at a time but each emission will be propagated:

Observable.concat(p1, p2, p3)
  .subscribe(console.log);

Edit:

Observable.of(null)
  .expand((value, index) => {
    switch (index) {
      case 0:
        return p1();
      case 1:
        return p2(value);
      case 2:
        return p3(value);
      default:
        return Observable.empty();
    }
  })
  .filter((value, index) => index > 0) // Ignore the `null` value that was necessary to start the chain
  .subscribe(console.log);

This prints:

1
2
true

See live demo (open console): https://stackblitz.com/edit/rxjs5-fucqhq?file=index.ts

edited Apr 11 '18 at 16:12

answered Apr 11 '18 at 15:31

martin

93,354
25
191
226

I assume you mean `Rx.Observable.concat(p1(),p2(),p3())`? – Keith Apr 11 '18 at 15:55
When chaining promises the value of the previous promise is sent to the next promise. Concat doesn't achieve that. – jeanpaul62 Apr 11 '18 at 15:57
@amaurymartiny and that's exactly what you want, right? – martin Apr 11 '18 at 16:02
@martin Your example with concat logs `1`, `Nan`, `false`. I want an Observable that logs `1`, `2`, `true`. – jeanpaul62 Apr 11 '18 at 16:05
@amaurymartiny I've heard one main rule of Observable, don't try using them for everything. `async / await` is a much better fit for what your asking. – Keith Apr 11 '18 at 16:08
@amaurymartiny Then you want `expand`, see my update. – martin Apr 11 '18 at 16:14

Keith · Answer 3 · 2018-04-11T16:28:48.187

For me you wasting the whole point of an observable here.

If you have a set of Promises you want a subscriber to receive as a stream you simply pass on to next, you now have a stream you can subscribe too.

Below is an example. ->

const p1 = () => Promise.resolve(1);
const p2 = x => { const val = x + 1; return  Promise.resolve(val); };
const p3 = x => {
  const isEven = x => x % 2 === 0;
  return Promise.resolve(isEven(x));
};

const chainedProm = (observer) => {
  let params;
  return async (p) => 
    observer.next(params = await p(params));
};

var observable = Rx.Observable.create(async function (observer) {
  const doProm = chainedProm(observer);
  await doProm(p1);
  await doProm(p2);
  await doProm(p3);
  observer.complete();
});


observable.subscribe(function(x){console.log(x)});

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.9/Rx.min.js"></script>

I've also updated now to include a utility function called, `chainedProm` your can use for any observable were you expecting a promise chain. — Keith, Apr 11 '18 at 16:31

Observables: Chaining promises & firing events on each promise

3 Answers3