Unification of TPL TaskScheduler and RX IScheduler
- by JoshReuben
using System; using System.Collections.Generic; using System.Reactive.Concurrency; using System.Security; using System.Threading; using System.Threading.Tasks; using System.Windows.Threading; namespace TPLRXSchedulerIntegration { public class MyScheduler :TaskScheduler, IScheduler { private readonly Dispatcher _dispatcher; private readonly DispatcherScheduler _rxDispatcherScheduler; //private readonly TaskScheduler _tplDispatcherScheduler; private readonly SynchronizationContext _synchronizationContext; public MyScheduler(Dispatcher dispatcher) { _dispatcher = dispatcher; _rxDispatcherScheduler = new DispatcherScheduler(dispatcher); //_tplDispatcherScheduler = FromCurrentSynchronizationContext(); _synchronizationContext = SynchronizationContext.Current; } #region RX public DateTimeOffset Now { get { return _rxDispatcherScheduler.Now; } } public IDisposable Schedule<TState>(TState state, DateTimeOffset dueTime, Func<IScheduler, TState, IDisposable> action) { return _rxDispatcherScheduler.Schedule(state, dueTime, action); } public IDisposable Schedule<TState>(TState state, TimeSpan dueTime, Func<IScheduler, TState, IDisposable> action) { return _rxDispatcherScheduler.Schedule(state, dueTime, action); } public IDisposable Schedule<TState>(TState state, Func<IScheduler, TState, IDisposable> action) { return _rxDispatcherScheduler.Schedule(state, action); } #endregion #region TPL /// Simply posts the tasks to be executed on the associated SynchronizationContext [SecurityCritical] protected override void QueueTask(Task task) { _dispatcher.BeginInvoke((Action)(() => TryExecuteTask(task))); //TryExecuteTaskInline(task,false); //task.Start(_tplDispatcherScheduler); //m_synchronizationContext.Post(s_postCallback, (object)task); } /// The task will be executed inline only if the call happens within the associated SynchronizationContext [SecurityCritical] protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) { if (SynchronizationContext.Current != _synchronizationContext) { SynchronizationContext.SetSynchronizationContext(_synchronizationContext); } return TryExecuteTask(task); } // not implemented [SecurityCritical] protected override IEnumerable<Task> GetScheduledTasks() { return null; } /// Implementes the MaximumConcurrencyLevel property for this scheduler class. /// By default it returns 1, because a <see cref="T:System.Threading.SynchronizationContext"/> based /// scheduler only supports execution on a single thread. public override Int32 MaximumConcurrencyLevel { get { return 1; } } //// preallocated SendOrPostCallback delegate //private static SendOrPostCallback s_postCallback = new SendOrPostCallback(PostCallback); //// this is where the actual task invocation occures //private static void PostCallback(object obj) //{ // Task task = (Task) obj; // // calling ExecuteEntry with double execute check enabled because a user implemented SynchronizationContext could be buggy // task.ExecuteEntry(true); //} #endregion } } What Design Pattern did I use here?
