// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT License.
// See the LICENSE file in the project root for more information.
using System.Reactive.Disposables;
namespace System.Reactive.Concurrency
{
///
/// Represents an object that schedules units of work on the platform's default scheduler.
///
/// Singleton instance of this type exposed through this static property.
public sealed class DefaultScheduler : LocalScheduler, ISchedulerPeriodic
{
private static readonly Lazy DefaultInstance = new(() => new DefaultScheduler());
private static readonly IConcurrencyAbstractionLayer Cal = ConcurrencyAbstractionLayer.Current;
///
/// Gets the singleton instance of the default scheduler.
///
public static DefaultScheduler Instance => DefaultInstance.Value;
private DefaultScheduler()
{
}
///
/// Schedules an action to be executed.
///
/// The type of the state passed to the scheduled action.
/// State passed to the action to be executed.
/// Action to be executed.
/// The disposable object used to cancel the scheduled action (best effort).
/// is null.
public override IDisposable Schedule(TState state, Func action)
{
if (action == null)
{
throw new ArgumentNullException(nameof(action));
}
var workItem = new UserWorkItem(this, state, action);
workItem.CancelQueueDisposable = Cal.QueueUserWorkItem(
static closureWorkItem => ((UserWorkItem)closureWorkItem!).Run(),
workItem);
return workItem;
}
///
/// Schedules an action to be executed after dueTime, using a System.Threading.Timer object.
///
/// The type of the state passed to the scheduled action.
/// State passed to the action to be executed.
/// Action to be executed.
/// Relative time after which to execute the action.
/// The disposable object used to cancel the scheduled action (best effort).
/// is null.
public override IDisposable Schedule(TState state, TimeSpan dueTime, Func action)
{
if (action == null)
{
throw new ArgumentNullException(nameof(action));
}
var dt = Scheduler.Normalize(dueTime);
if (dt.Ticks == 0)
{
return Schedule(state, action);
}
var workItem = new UserWorkItem(this, state, action);
workItem.CancelQueueDisposable = Cal.StartTimer(
static closureWorkItem => ((UserWorkItem)closureWorkItem!).Run(),
workItem,
dt);
return workItem;
}
///
/// Schedules a periodic piece of work, using a System.Threading.Timer object.
///
/// The type of the state passed to the scheduled action.
/// Initial state passed to the action upon the first iteration.
/// Period for running the work periodically.
/// Action to be executed, potentially updating the state.
/// The disposable object used to cancel the scheduled recurring action (best effort).
/// is less than .
/// is null.
public IDisposable SchedulePeriodic(TState state, TimeSpan period, Func action)
{
if (period < TimeSpan.Zero)
{
throw new ArgumentOutOfRangeException(nameof(period));
}
if (action == null)
{
throw new ArgumentNullException(nameof(action));
}
return new PeriodicallyScheduledWorkItem(state, period, action);
}
private sealed class PeriodicallyScheduledWorkItem : IDisposable
{
private TState _state;
private Func _action;
private readonly IDisposable _cancel;
private readonly AsyncLock _gate = new();
public PeriodicallyScheduledWorkItem(TState state, TimeSpan period, Func action)
{
_state = state;
_action = action;
_cancel = Cal.StartPeriodicTimer(Tick, period);
}
private void Tick()
{
_gate.Wait(
this,
static closureWorkItem => closureWorkItem._state = closureWorkItem._action(closureWorkItem._state));
}
public void Dispose()
{
_cancel.Dispose();
_gate.Dispose();
_action = Stubs.I;
}
}
///
/// Discovers scheduler services by interface type.
///
/// Scheduler service interface type to discover.
/// Object implementing the requested service, if available; null otherwise.
protected override object? GetService(Type serviceType)
{
if (serviceType == typeof(ISchedulerLongRunning))
{
if (Cal.SupportsLongRunning)
{
return LongRunning.Instance;
}
}
return base.GetService(serviceType);
}
private sealed class LongRunning : ISchedulerLongRunning
{
private sealed class LongScheduledWorkItem : ICancelable
{
private readonly TState _state;
private readonly Action _action;
private SingleAssignmentDisposableValue _cancel;
public LongScheduledWorkItem(TState state, Action action)
{
_state = state;
_action = action;
Cal.StartThread(
thisObject =>
{
var @this = (LongScheduledWorkItem)thisObject!;
//
// Notice we don't check d.IsDisposed. The contract for ISchedulerLongRunning
// requires us to ensure the scheduled work gets an opportunity to observe
// the cancellation request.
//
@this._action(@this._state, @this);
},
this
);
}
public void Dispose()
{
_cancel.Dispose();
}
public bool IsDisposed => _cancel.IsDisposed;
}
public static readonly ISchedulerLongRunning Instance = new LongRunning();
public IDisposable ScheduleLongRunning(TState state, Action action)
{
if (action == null)
{
throw new ArgumentNullException(nameof(action));
}
return new LongScheduledWorkItem(state, action);
}
}
}
}