Emit next value from the source Observable when another Observable, the notifier, emits

Question

I would expect that my case is common but can't really find anything appropriate. What I want to achieve, in Angular2 / RxJS 5 is this:

source:   ---1--2--3--4---------5--------6-|-->
notifier: -o------------o-----o---o--o-o------>
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
output:   ---1----------2-----3---4--5---6-|-->

So, I have a source Observable that emits values, and I want each of them to get into the output only when a second Observable (call it notifier) emits. It's like one event from the notifier means "allow next to pass through".

I tried delayWhen, but my main problem with this is that all the source values are waiting for the same one event from the notifier, so for example if 3 source values are "queued" and notifier emits once, all 3 values pass through, which is not what I want.

Answer 1

The answer is zip:

const valueStream = 
    Rx.Observable.from([0, 1, 2, 3, 4, 5, 6]);

const notificationStream = 
    Rx.Observable.interval(1000).take(7);


Rx.Observable
    .zip(valueStream, notificationStream, (val, notification) => val)
    .subscribe(val => console.log(val));

Working example here.

This produces a value when a pair is produced from both streams. So the example will print a value from valueStream when notificationStream produces a value.

Answer 2

I think that the zip operator is what you're looking for:

sourceSubject:Subject = new Subject();
notifierSubject:Subject = new Subject();

index = 1;

constructor() {
  Observable.zip(
    this.sourceSubject, this.notifierSubject
  )
  .map(data => data[0])
  .subscribe(data => {
    console.log('>> output = '+data.id);
  });
}

emit() {
  this.sourceSubject.next({id: this.index});
  this.index++;
}

notify() {
  this.notifierSubject.next();
}

See this plunkr: https://plnkr.co/edit/MK30JR2qK8aJIGwNqMZ5?p=preview.

See also this question:

• How to use frokJoin of Observable with own event