Observable.Interval restarts for each subscription

Question

My first attempt to define a measurement schedule was:

    var schedule = Observable.Concat(
                Observable.Interval(TimeSpan.FromSeconds(1)).Take(3),
                Observable.Interval(TimeSpan.FromSeconds(3)).Take(3),
                Observable.Interval(TimeSpan.FromSeconds(5)));

Unfortunately, it restarts when unsubscribed and resubscribed, which is not a desired behavior in my case. Therefore, I came with something like this:

    class Schedule : IObservable<DateTime>, IDisposable
    {
        readonly ISubject<DateTime> _subject;
        readonly IDisposable _subscrption;

        public Schedule()
        {
            _subject = new BehaviorSubject<DateTime>(DateTime.UtcNow);
            _subscrption = Observable.Concat(
                Observable.Interval(TimeSpan.FromSeconds(1)).Take(3),
                Observable.Interval(TimeSpan.FromSeconds(3)).Take(3),
                Observable.Interval(TimeSpan.FromSeconds(5)))
                .Select(i => DateTime.UtcNow)
                .Subscribe(_subject);
        }

        public IDisposable Subscribe(IObserver<DateTime> observer)
        {
            return _subject.Subscribe(observer);
        }

        public void Dispose()
        {
            _subscrption.Dispose(); 
        }
    }

It works but requires disposing after use. Are there any simple way to define Schedule without exposing IDisposable?

Answer 1

If you will be unsubscribing all of your observers and then resubscribing, there really is no way around keeping a Disposable around for tracking the connection. Since your Observable would not have anyway of directly knowing if the latest unsubscribe is really the last one.

I would recommend the use of Generate instead of Concatenating Observables

IConnectableObservable<DateTime> source = Observable.Generate(
  0,
  _ => true,
  x => x + 1,
  x => DateTime.UtcNow,
  x => {
    if (x < 3) return TimeSpan.FromSeconds(1);
    else if (x < 5) return TimeSpan.FromSeconds(3);
    else return TimeSpan.FromSeconds(5);
}).Publish();

Publish will turn your Observable into a ConnectableObservable, and gives you two options for managing when the source actually produces events.

//1) Explicit connect
IDisposable connection = source.Connect();

//2) RefCounted connection
IObservable<DateTime> rcSource = source.RefCount();

In the first version the source will become "hot" once you connect, and when you want to disconnect you should dispose of connection.

In the second the source will automatically disconnect when it has no more observers.

see also hot vs cold observables

Answer 2

Here is the solution I defined which does not require disposing. It is pretty complex though, so I would highly appreciate if something can be simplified using Rx instrumentation.

The usage:

  IObservable<DateTime> schedule = new Schedule()
            .Setup(TimeSpan.FromSeconds(1), 3)
            .Setup(TimeSpan.FromSeconds(3), 3)
            .Setup(TimeSpan.FromSeconds(5));

Where Schedule is:

    class Schedule : IObservable<DateTime>
    {
        readonly TimeSpan TimerPrecision = TimeSpan.FromMilliseconds(1);
        readonly IEnumerable<TimeSpan> Intervals;
        readonly IEnumerator<DateTime> Event;

        public Schedule()
            : this(new TimeSpan[0])
        {
        }

        Schedule(IEnumerable<TimeSpan> intervals)
        {
            Intervals = intervals;
            Event = Start();
            Event.MoveNext();
        }

        public Schedule Setup(TimeSpan interval)
        {
            return Setup(interval, Int32.MaxValue);
        }

        public Schedule Setup(TimeSpan interval, int repeat)
        {
            return new Schedule(
                Intervals.Concat(
                    Enumerable.Repeat(interval, repeat)));
        }

        public IDisposable Subscribe(IObserver<DateTime> observer)
        {
            var timer = new System.Timers.Timer() { AutoReset = true };
            timer.Elapsed += (s, a) => 
            {                                                
                observer.OnNext(DateTime.UtcNow);
                if (!TryArm(timer))
                    observer.OnCompleted();
            };

            if (!TryArm(timer))
                observer.OnCompleted();

            return timer;
        }

        IEnumerator<DateTime> Start()
        {
            var time = DateTime.UtcNow;
            yield return time;

            foreach (var interval in Intervals)
            {
                time += interval;
                yield return time;
            }
        }

        TimeSpan Delay()
        {
            var now = DateTime.UtcNow;
            lock (Event)
                while (Event.Current - DateTime.UtcNow < TimerPrecision)
                    Event.MoveNext();

            return Event.Current - now;
        }

        bool TryArm(System.Timers.Timer timer)
        {
            try
            {
                timer.Interval = Delay().TotalMilliseconds;
                timer.Start();
                return true;
            }            
            catch(ObjectDisposedException)   
            {
                return false;
            }
            catch(InvalidOperationException)   
            {
                return false;
            }
        }
    }