How to prevent combine_latest firing multiple times when multiple inputs tick simultaneously?

Question

I'm using Reactive Extensions' combine_latest to perform an action whenever any inputs tick. The problem is if multiple inputs tick at the same time then combine_latest fires multiple times after each individual input ticks. This causes a headache because combine_latest is effectively ticking spuriously with stale values.

Minimum working example where a fast observable ticks every 10ms and a slow observable ticks every 30ms:

from rx.concurrency import HistoricalScheduler
from rx import Observable
from __future__ import print_function

scheduler = HistoricalScheduler(initial_clock=1000)
fast = Observable.generate_with_relative_time(1, lambda x: True, lambda x: x + 1, lambda x: x, lambda x: 10, scheduler=scheduler)
slow = Observable.generate_with_relative_time(3, lambda x: True, lambda x: x + 3, lambda x: x, lambda x: 30, scheduler=scheduler)
Observable.combine_latest(fast, slow, lambda x, y: dict(time=scheduler.now(), fast=x, slow=y)).subscribe(print)
scheduler.advance_to(1100)

Every 30ms, when fast and slow tick simultaneously, combine_latest fires twice undesirably -- the output is below:

{'slow': 3, 'fast': 2, 'time': 1030}  # <- This shouldn't be here
{'slow': 3, 'fast': 3, 'time': 1030}
{'slow': 3, 'fast': 4, 'time': 1040}
{'slow': 3, 'fast': 5, 'time': 1050}
{'slow': 6, 'fast': 5, 'time': 1060}  # <- This shouldn't be here
{'slow': 6, 'fast': 6, 'time': 1060}
{'slow': 6, 'fast': 7, 'time': 1070}
{'slow': 6, 'fast': 8, 'time': 1080}
{'slow': 9, 'fast': 8, 'time': 1090}  # <- This shouldn't be here
{'slow': 9, 'fast': 9, 'time': 1090}
{'slow': 9, 'fast': 10, 'time': 1100}

How can I prevent combine_latest from ticking spuriously?

Answer 1

I think debounce does exactly what you want:

---A-B-C-----D------E-F----|->
  [debounce(some_interval)]
--------C-----D--------F---|->

After getting a value A, debounce will wait for some_interval. If another value B appears, it will emit B instead. So you could debounce your input stream, which will 'catch' those extra clicks and only emit the last one.

More in the official docs

(the question was asked a long time ago, but I'm answering it for future googlers.)