using System.Collections.Generic; using System.Reactive.Disposables; namespace System.Reactive.Concurrency { public abstract class VirtualTimeScheduler<TAbsolute, TRelative> : VirtualTimeSchedulerBase<TAbsolute, TRelative> where TAbsolute : IComparable<TAbsolute> { private readonly SchedulerQueue<TAbsolute> queue = new SchedulerQueue<TAbsolute>(); protected VirtualTimeScheduler() { } protected VirtualTimeScheduler(TAbsolute initialClock, IComparer<TAbsolute> comparer) : base(initialClock, comparer) { } protected override IScheduledItem<TAbsolute> GetNext() { lock (queue) { while (queue.Count > 0) { ScheduledItem<TAbsolute> scheduledItem = queue.Peek(); if (!scheduledItem.IsCanceled) return scheduledItem; queue.Dequeue(); } } return null; } public override IDisposable ScheduleAbsolute<TState>(TState state, TAbsolute dueTime, Func<IScheduler, TState, IDisposable> action) { if (action == null) throw new ArgumentNullException("action"); ScheduledItem<TAbsolute, TState> si = null; Func<IScheduler, TState, IDisposable> action2 = delegate(IScheduler scheduler, TState state1) { lock (this.queue) { this.queue.Remove((ScheduledItem<TAbsolute>)si); } return action(scheduler, state1); }; si = (ScheduledItem<TAbsolute, TState>)new ScheduledItem<TAbsolute, TState>((IScheduler)this, state, action2, dueTime, base.Comparer); lock (this.queue) { this.queue.Enqueue((ScheduledItem<TAbsolute>)si); } return Disposable.Create(((ScheduledItem<TAbsolute>)si).Cancel); } } }