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?
withLatestFrom
in your situation? See staltz.com/rx-glitches-arent-actually-a-problem.html – Redvers