Rxjs distinctUntilChanged with timer

Question

Is there a way to add a timer to distinctUntilChanged?

In rxjs 6

of(1, 2, 3, 3, 5, 6)
  .pipe(distinctUntilChanged)
  .subscribe(x => console.log(x)

My question is, is there a way to add a timer so that repeated values will emit if they fall outside the time range but will be ignored if they fall within the time range.

Like if the say the timer is set to 1000 (1s), in this case interval(1000) and a repeated value is emitted within the 1s it will be ignore but if after the interval (1s) it should be allowed through. I hope am making sense here

Also if there's a better way to do it without distinctUntilChanged i will gladly welcome that.

Thanks in advance

Answer 1

This is one of the very rare occasions where windowTime is useful. windowTime creates a new Subject after a period of time that is passed further (so called higher-order Observable). Then you can use concatMap to chain it with distinctUntilChanged. This all means you'll create a new distinctUntilChanged every 1s:

const subject = new Subject();

subject
  .pipe(
    windowTime(1000),
    concatMap(obs => obs.pipe(distinctUntilChanged())),
  )
  .subscribe(console.log);

subject.next(1);
subject.next(2);

setTimeout(() => subject.next(2), 1200);

See live demo: https://stackblitz.com/edit/rxjs6-demo-pcmrzz?file=index.ts

Answer 2

Just append a date property to the previous map output, and compare it in the distinctUntilChanged, and pluck out the date property after :)

fromEvent(document, 'keydown')
    .pipe(
        map(e => ({key: e.key, time: Date.now()})),
        distinctUntilChanged((a, b) => (a.key == b.key) && (a.time > b.time - 500)),
        pluck('key')
    )

Answer 3

skipUntil should do what you are looking for, if I understood right.

This can be an example

const obs = interval(100).pipe(
    tap(d => console.log('data emitted', d)),
    take(10),
);


obs
.pipe(
    skipUntil(interval(500).pipe(take(1)))
)
.subscribe(
    data => console.log('data read in the subscription', data)
)