How to convert IEnumerable to IObservable using HistoricalScheduler

Question

I have an IEnumerable<T> where T allows derivation of a relevant timestamp.

I would like to convert that to an IObservable<T> but I wish to use the HistoricalScheduler so that notifications occur as per the derived timestamp. Doing this allows use of built-in RX methods for windowing, sliding windows, etc., which are ultimately what I am trying to utilise.

Many of the suggestions of how to go about this suggest using Generate(). However, this method causes StackOverflowExceptions. E.g.:

    static void Main(string[] args)
    {

        var enumerable = Enumerable.Range(0, 2000000);
        var now = DateTimeOffset.Now;
        var scheduler = new HistoricalScheduler(now);
        var observable = Observable.Generate(
            enumerable.GetEnumerator(), 
            e => e.MoveNext(), 
            e => e, 
            e => Timestamped.Create(e, now.AddTicks(e.Current)), 
            e => now.AddTicks(e.Current), 
            scheduler);
        var s2 = observable.Count().Subscribe(eventCount => Console.WriteLine("Found {0} events @ {1}", eventCount, scheduler.Now));
        scheduler.Start();
        s2.Dispose();
        Console.ReadLine();
    }

This will result in a stack overflow.

The standard ToObservable() method cannot be used because, although it allows a custom scheduler to be specified, it does not provide any mechanism to control how the resulting notifications are scheduled on that scheduler.

How do I convert an IEnumerable to an IObservable with explicity scheduled notification?

UPDATE

Attempted to use Asti's code in the following test:

    static void Main(string[] args)
    {
        var enumerable = Enumerable.Range(0, 2000000);
        var now = DateTimeOffset.Now;
        var series = enumerable.Select(i => Timestamped.Create(i, now.AddSeconds(i)));
        var ticks = Observable.Interval(TimeSpan.FromSeconds(1)).Select(i => now.AddSeconds(i));
        var scheduler = new HistoricalScheduler(now);
        Playback(series,ticks,scheduler).Subscribe(Console.WriteLine);
        scheduler.Start();
    }

However it throws an ArgumentOutOfRangeException:

Specified argument was out of the range of valid values.
Parameter name: time

   at System.Reactive.Concurrency.VirtualTimeSchedulerBase`2.AdvanceTo(TAbsolute time)
   at System.Reactive.AnonymousSafeObserver`1.OnNext(T value)
   at System.Reactive.Linq.ObservableImpl.Select`2._.OnNext(TSource value)
   at System.Reactive.Linq.ObservableImpl.Timer.TimerImpl.Tick(Int64 count)
   at System.Reactive.Concurrency.DefaultScheduler.<>c__DisplayClass7_0`1.<SchedulePeriodic>b__1()
   at System.Reactive.Concurrency.AsyncLock.Wait(Action action)
   at System.Reactive.Concurrency.DefaultScheduler.<>c__DisplayClass7_0`1.<SchedulePeriodic>b__0()
   at System.Reactive.Concurrency.ConcurrencyAbstractionLayerImpl.PeriodicTimer.Tick(Object state)
   at System.Threading.TimerQueueTimer.CallCallbackInContext(Object state)
   at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
   at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
   at System.Threading.TimerQueueTimer.CallCallback()
   at System.Threading.TimerQueueTimer.Fire()
   at System.Threading.TimerQueue.FireNextTimers()
   at System.Threading.TimerQueue.AppDomainTimerCallback()

Answer 1

We make an operator which replays an ordered sequence of events on a historical scheduler with the time moving according to a specified observable.

    public static IObservable<T> Playback<T>(
        this IEnumerable<Timestamped<T>> enumerable,
        IObservable<DateTimeOffset> ticks,
        HistoricalScheduler scheduler = default(HistoricalScheduler)
    )
    {
        return Observable.Create<T>(observer =>
        {
            scheduler = scheduler ?? new HistoricalScheduler();

            //create enumerator of sequence - we're going to iterate through it manually
            var enumerator = enumerable.GetEnumerator();

            //set scheduler time for every incoming value of ticks
            var timeD = ticks.Subscribe(scheduler.AdvanceTo);

            //declare an iterator 
            Action scheduleNext = default(Action);
            scheduleNext = () =>
            {
                //move
                if (!enumerator.MoveNext())
                {
                    //no more items
                    //sequence has completed
                    observer.OnCompleted();
                    return;
                }

                //current item of enumerable sequence
                var current = enumerator.Current;

                //schedule the item to run at the timestamp specified
                scheduler.ScheduleAbsolute(current.Timestamp, () =>
                {
                    //push the value forward
                    observer.OnNext(current.Value);

                    //schedule the next item
                    scheduleNext();                        
                });
            };

            //start the process by scheduling the first item
            scheduleNext();

            //dispose the enumerator and subscription to ticks
            return new CompositeDisposable(timeD, enumerator);
        });
    }

Porting your earlier example,

   var enumerable = Enumerable.Range(0, 20000000);
            var now = DateTimeOffset.Now;
            var series = enumerable.Select(i => Timestamped.Create(i, now.AddSeconds(i)));
            var ticks = Observable.Interval(TimeSpan.FromSeconds(1)).Select(i => now.AddSeconds(i));

            series.Playback(ticks).Subscribe(Console.WriteLine);

We read through the enumerable to keep it lazy, and set the clock with a simple Interval observable. Reducing the interval causes it to playback faster.

Answer 2

Phil, I think you will find problems regardless if you try to issue a notification every tick. Rx just wont be able to keep up.

I see what @asti is doing here, But I think it can made a touch simpler using what Paul already had (IEnumerable<Timestamped<T>>)

public static IObservable<T> Playback<T>(
   this IEnumerable<Timestamped<T>> enumerable,
   IScheduler scheduler)
{
    return Observable.Create<T>(observer =>
    {
        var enumerator = enumerable.GetEnumerator();

        //declare a recursive function 
        Action<Action> scheduleNext = (self) =>
        {
            //move
            if (!enumerator.MoveNext())
            {
                //no more items (or we have been disposed)
                //sequence has completed
                scheduler.Schedule(()=>observer.OnCompleted());
                return;
            }

            //current item of enumerable sequence
            var current = enumerator.Current;

            //schedule the item to run at the timestamp specified
            scheduler.Schedule(current.Timestamp, () =>
            {
                //push the value forward
                observer.OnNext(current.Value);

                //Recursively call self (via the scheduler API)
                self();
            });
        };

        //start the process by scheduling the recursive calls.
        // return the scheduled handle to allow disposal.
        var scheduledTask = scheduler.Schedule(scheduleNext);
        return StableCompositeDisposable.Create(scheduledTask, enumerator);
    });
}

This is also scheduler agnostic so will work with any scheduler.