Reactive extension Timer/Interval reset

Question

I have a project where I need to send a status message every 10 seconds unless there's been an update in the meantime. Meaning, every time there would be an update, the timer would reset.

var res = Observable
  .Interval(TimeSpan.FromSeconds(10))
  .Where(_ => condition);

res.Subscribe(_ => Console.WriteLine("Status sent."));

Now I know that the "Where" will only be applied when the timer ends, so it doesn't help. But, I'm wondering if there's a way to reset the Interval; or to use a Timer() with a repeat.

Answer 1

This is pretty easy to implement using standard Rx operators.

What isn't clear from your question is exactly what an "update" is. I'm going to assume that you have some sort of observable that fires for every update or that you can create a subject that you'll call .OnNext(...) when there is an update. Without observable updates it is hard to know when to reset the timer.

So here's the code:

var update = new Subject<bool>();

var res =
    update
        .Select(x => Observable.Interval(TimeSpan.FromSeconds(10.0)))
        .Switch();

res
    .Subscribe(_ => Console.WriteLine("Status sent."));

update.OnNext(true);

The res query now waits until it gets a value from update and then it selects a new Observable.Interval. This means that after the Select the type is an IObservable<IObservable<long>>, so the .Switch() is required to turn it in to a IObservable<long>. .Switch() does this by only passing out values from the latest observed observable and disposing of any previous observables. In other words, for each update a new timer is started and the previous timer is cancelled. This means that if you have updates occurring more frequently than 10 seconds then the timer will never fire.

Now, if the res observable is an update in its own right, then you can do this:

res
    .Subscribe(_ =>
    {
        update.OnNext(true);
        Console.WriteLine("Status sent.");
    });

That's fine - it still works, but for each timer firing res will create a new timer. It will mean that anything relying on your update observable/subject will still function correctly.

Answer 2

I keep this little helper method with me:

public static IObservable<long> CreateAutoResetInterval<TSource>(IObservable<TSource> resetter, TimeSpan timeSpan, bool immediate = false)
{
    return resetter.Select(_ => immediate ? Observable.Interval(timeSpan).StartWith(0) : Observable.Interval(timeSpan)).Switch();
}

It's basically the same mechanism as Enigmativity's answer