How to make one Observable sequence wait for another to complete before emitting?

Question

Say I have an Observable, like so:

var one = someObservable.take(1);

one.subscribe(function(){ /* do something */ });

Then, I have a second Observable:

var two = someOtherObservable.take(1);

Now, I want to subscribe() to two, but I want to make sure that one has completed before the two subscriber is fired.

What kind of buffering method can I use on two to make the second one wait for the first one to be completed?

I suppose I am looking to pause two until one is complete.

Answer 1

A couple ways I can think of

import {take, publish} from 'rxjs/operators'
import {concat} from 'rxjs'

//Method one

var one = someObservable.pipe(take(1));
var two = someOtherObservable.pipe(take(1));
concat(one, two).subscribe(function() {/*do something */});

//Method two, if they need to be separate for some reason
var one = someObservable.pipe(take(1));
var two = someOtherObservable.pipe(take(1), publish());
two.subscribe(function(){/*do something */});
one.subscribe(function(){/*do something */}, null, two.connect.bind(two));

Answer 2

If you want to make sure that the order of execution is retained you can use flatMap as the following example

const first = Rx.Observable.of(1).delay(1000).do(i => console.log(i));
const second = Rx.Observable.of(11).delay(500).do(i => console.log(i));
const third = Rx.Observable.of(111).do(i => console.log(i));

first
  .flatMap(() => second)
  .flatMap(() => third)
  .subscribe(()=> console.log('finished'));

The outcome would be:

"1"
"11"
"111"
"finished"

Answer 3

skipUntil() with last()

skipUntil : ignore emitted items until another observable has emitted

last: emit last value from a sequence (i.e. wait until it completes then emit)

Note that anything emitted from the observable passed to skipUntil will cancel the skipping, which is why we need to add last() - to wait for the stream to complete.

main$.skipUntil(sequence2$.pipe(last()))

Official: https://rxjs-dev.firebaseapp.com/api/operators/skipUntil

---

Possible issue: Note that last() by itself will error if nothing is emitted. The last() operator does have a default parameter but only when used in conjunction with a predicate. I think if this situation is a problem for you (if sequence2$ may complete without emitting) then one of these should work (currently untested):

main$.skipUntil(sequence2$.pipe(defaultIfEmpty(undefined), last()))
main$.skipUntil(sequence2$.pipe(last(), catchError(() => of(undefined))

Note that undefined is a valid item to be emitted, but could actually be any value. Also note that this is the pipe attached to sequence2$ and not the main$ pipe.

Answer 4

Here's a reusable way of doing it (it's typescript but you can adapt it to js):

function waitFor<T>(signal: Observable<any>) {
    return (source: Observable<T>) => signal.pipe(
        first(),
        switchMap(_ => source),
    );
}

and you can use it like any operator:

var two = someOtherObservable.pipe(waitFor(one), take(1));

It's basically an operator that defers the subscribe on the source observable until the signal observable emits the first event.

Answer 5

Here is yet another possibility taking advantage of switchMap's result selector

var one$ = someObservable.take(1);
var two$ = someOtherObservable.take(1);
two$.switchMap(
    /** Wait for first Observable */
    () => one$,
    /** Only return the value we're actually interested in */
    (value2, value1) => value2
  )
  .subscribe((value2) => {
    /* do something */ 
  });

Since the switchMap's result selector has been depreciated, here is an updated version

const one$ = someObservable.pipe(take(1));
const two$ = someOtherObservable.pipe(
  take(1),
  switchMap(value2 => one$.map(_ => value2))
);
two$.subscribe(value2 => {
  /* do something */ 
});

Answer 6

If the second observable is hot, there is another way to do pause/resume:

var pauser = new Rx.Subject();
var source1 = Rx.Observable.interval(1000).take(1);
/* create source and pause */
var source2 = Rx.Observable.interval(1000).pausable(pauser);

source1.doOnCompleted(function () { 
  /* resume paused source2 */ 
  pauser.onNext(true);
}).subscribe(function(){
  // do something
});

source2.subscribe(function(){
  // start to recieve data 
});

Also you can use buffered version pausableBuffered to keep data during pause is on.

Answer 7

Here's a custom operator written with TypeScript that waits for a signal before emitting results:

export function waitFor<T>(
    signal$: Observable<any>
) {
    return (source$: Observable<T>) =>
        new Observable<T>(observer => {
            // combineLatest emits the first value only when
            // both source and signal emitted at least once
            combineLatest([
                source$,
                signal$.pipe(
                    first(),
                ),
            ])
                .subscribe(([v]) => observer.next(v));
        });
}

You can use it like this:

two.pipe(waitFor(one))
   .subscribe(value => ...);

Answer 8

Here's yet another, but I feel more straightforward and intuitive (or at least natural if you're used to Promises), approach. Basically, you create an Observable using Observable.create() to wrap one and two as a single Observable. This is very similar to how Promise.all() may work.

var first = someObservable.take(1);
var second = Observable.create((observer) => {
  return first.subscribe(
    function onNext(value) {
      /* do something with value like: */
      // observer.next(value);
    },
    function onError(error) {
      observer.error(error);
    },
    function onComplete() {
      someOtherObservable.take(1).subscribe(
        function onNext(value) {
          observer.next(value);
        },
        function onError(error) {
          observer.error(error);
        },
        function onComplete() {
          observer.complete();
        }
      );
    }
  );
});

So, what's going on here? First, we create a new Observable. The function passed to Observable.create(), aptly named onSubscription, is passed the observer (built from the parameters you pass to subscribe()), which is similar to resolve and reject combined into a single object when creating a new Promise. This is how we make the magic work.

In onSubscription, we subscribe to the first Observable (in the example above, this was called one). How we handle next and error is up to you, but the default provided in my sample should be appropriate generally speaking. However, when we receive the complete event, which means one is now done, we can subscribe to the next Observable; thereby firing the second Observable after the first one is complete.

The example observer provided for the second Observable is fairly simple. Basically, second now acts like what you would expect two to act like in the OP. More specifically, second will emit the first and only the first value emitted by someOtherObservable (because of take(1)) and then complete, assuming there is no error.

Example

Here is a full, working example you can copy/paste if you want to see my example working in real life:

var someObservable = Observable.from([1, 2, 3, 4, 5]);
var someOtherObservable = Observable.from([6, 7, 8, 9]);

var first = someObservable.take(1);
var second = Observable.create((observer) => {
  return first.subscribe(
    function onNext(value) {
      /* do something with value like: */
      observer.next(value);
    },
    function onError(error) {
      observer.error(error);
    },
    function onComplete() {
      someOtherObservable.take(1).subscribe(
        function onNext(value) {
          observer.next(value);
        },
        function onError(error) {
          observer.error(error);
        },
        function onComplete() {
          observer.complete();
        }
      );
    }
  );
}).subscribe(
  function onNext(value) {
    console.log(value);
  },
  function onError(error) {
    console.error(error);
  },
  function onComplete() {
    console.log("Done!");
  }
);

If you watch the console, the above example will print:

1

6

Done!

Answer 9

well, I know this is pretty old but I think that what you might need is:

var one = someObservable.take(1);

var two = someOtherObservable.pipe(
  concatMap((twoRes) => one.pipe(mapTo(twoRes))),
  take(1)
).subscribe((twoRes) => {
   // one is completed and we get two's subscription.
})

Answer 10

You can use result emitted from previous Observable thanks to mergeMap (or his alias flatMap) operator like this:

 const one = Observable.of('https://api.github.com/users');
 const two = (c) => ajax(c);//ajax from Rxjs/dom library

 one.mergeMap(two).subscribe(c => console.log(c))

Answer 11

Perhaps you can use the delayWhen operator.

We have two observables one$ and two$. First observable emits 1 after a 1s delay then completes. Second observable emits 2 only after one$ has emitted:

const one$ = of(1).pipe(
  delay(1000),
  tap(() => console.log('one$ emitted'))
);

const two$ = of(2).pipe(
  delayWhen(() => one$),
  tap(() => console.log('two$ emitted')),
);

two$.subscribe(n => {
  console.log(`n=${n}`);
});

<script src="https://unpkg.com/[email protected]/dist/bundles/rxjs.umd.min.js"></script>
<script>
const {of} = rxjs;
const {delay, delayWhen, tap} = rxjs.operators;
</script>