CODEWITHSUNDEEP
reactive-programmingsystem.reactivereactiveuirx.net
resp78
resp78

Reputation: 1534

Reactive extensions - how to do variable rate polling?

I am wondering how can I poll/invoke a method at a set interval. I would like to be able to vary this regular interval also. So something like this.

[Reactive]
public TimeSpan Rate { get; set; }

IObservable<TimeSpan> rate = this.WhenAnyValue(vm => vm.Rate);

// poll should emit at the current rate set in `rate observable`
rate.Poll().InvokeCommand(cmd);

In marble diagram

rate         ---4----------2--------∞----------4----------------
Poll         ---X---X---X--X-X-X-X-XX----------X---X---X---X---X

X - the time when method is executed

Point to note, a new value in rate should force Poll to emit immediately and cancel any previous value that was supposed to be emitted.

I attempted to write a Poll Operator. However, it complains when I try to pass a large TimeSpan. 

    public static IObservable<TimeSpan> Poll(this IObservable<TimeSpan> sourceObservable, IScheduler scheduler = null)
    {
        scheduler = scheduler ?? NewThreadScheduler.Default;

        return Observable.Create<TimeSpan>(observer =>
        {
            return scheduler.ScheduleAsync(async (s, ct) =>
            {
                var timerTokenSource = new CancellationTokenSource();
                var compositeTokenSource = CancellationTokenSource.CreateLinkedTokenSource(timerTokenSource.Token, ct);

                TimeSpan interval = TimeSpan.MaxValue;

                sourceObservable.Subscribe(t =>
                    {
                        interval = t;
                        timerTokenSource.Cancel();
                        timerTokenSource = new CancellationTokenSource();
                        compositeTokenSource = CancellationTokenSource.CreateLinkedTokenSource(timerTokenSource.Token, ct);
                    },
                    ct);

                while (!ct.IsCancellationRequested)
                {
                    await s.Sleep(interval, compositeTokenSource.Token);
                    observer.OnNext(interval);
                }
            });
        });
    }

Upvotes: 0

Views: 183

Answers (1)

Enigmativity
Enigmativity

Reputation: 117027

I think this is what you want:

public static IObservable<TimeSpan> Poll(this IObservable<TimeSpan> sourceObservable, IScheduler scheduler = null)
{
    scheduler = scheduler ?? NewThreadScheduler.Default;

    return 
        sourceObservable
            .Select(ts => Observable.Timer(TimeSpan.Zero, ts).Select(x => ts))
            .ObserveOn(scheduler)
            .Switch();
}

I don't know why you're returning a TimeSpan. Why is that?

Upvotes: 4

Related Questions