This is a 2nd follow-up to my previous one about a Scheduler built with observables.
Although the last one was working correctly, it was only possible to see this in LINQPad which I didn't like very much. I prefer to have proper tests so I've redesigned it a little bit to make testing possible.
Core
The Scheduler itself is now pretty simple. It requires an observable that produces ticks which it turns into a hot-observable. On each tick a each job's triggers are evaluated and jobs which match the trigger are executed.
Unscheduling jobs is a blocking operation when a timeout is pecified.
public class Scheduler : IDisposable
{
private readonly IConnectableObservable<DateTime> _scheduler;
private readonly IDisposable _disconnect;
public Scheduler(IObservable<DateTime> ticks)
{
// Not using .RefCount here because it should be ticking regardless of subscriptions.
_scheduler = ticks.Publish();
_disconnect = _scheduler.Connect();
}
public IDisposable Schedule(Job job, CancellationToken cancellationToken = default)
{
var unschedule =
_scheduler
// .ToList the results so that all triggers have the chance to evaluate the tick.
.Where(tick => job.Triggers.Select(t => t.Matches(tick)).ToList().Any(x => x))
.Subscribe(timestamp => job.Execute(cancellationToken));
return Disposable.Create(() =>
{
job.Continuation.Wait(job.UnscheduleTimeout);
unschedule.Dispose();
});
}
public void Dispose()
{
// Stop ticking.
_disconnect.Dispose();
}
}
The Job class is handling job's triggers and tasks and keeping them within the specified max-degree-of-parallelism.
public class Job
{
private readonly List<Task> _tasks = new List<Task>();
public Job(string name, IEnumerable<Trigger> trigger, Func<CancellationToken, Task> action)
{
Name = name;
Triggers = trigger.ToList();
Action = action;
}
public string Name { get; }
public IEnumerable<Trigger> Triggers { get; }
public Func<CancellationToken, Task> Action { get; }
public Action<Job> OnMisfire { get; set; }
public DegreeOfParallelism MaxDegreeOfParallelism { get; set; } = 1;
public TimeSpan UnscheduleTimeout { get; set; }
public Task Continuation => Task.WhenAll(_tasks).ContinueWith(_ => _tasks.Clear());
public int Count => _tasks.Count;
public void Execute(CancellationToken cancellationToken)
{
if (CanExecute())
{
var jobTask = Action(cancellationToken);
_tasks.Add(jobTask);
jobTask.ContinueWith(_ => _tasks.Remove(jobTask), cancellationToken);
}
else
{
OnMisfire?.Invoke(this);
}
}
private bool CanExecute()
{
return
MaxDegreeOfParallelism.Equals(DegreeOfParallelism.Unlimited) ||
Count < MaxDegreeOfParallelism.Value;
}
}
The DegreeOfParallelism is an anti-primitive-obsession wrapper for int.
public class DegreeOfParallelism : Primitive<int>
{
private const int UnlimitedValue = -1;
public DegreeOfParallelism(int value) : base(value) { }
public static readonly DegreeOfParallelism Unlimited = new DegreeOfParallelism(UnlimitedValue);
protected override void Validate(int value)
{
if (value == UnlimitedValue)
{
return;
}
if (value < 1)
{
throw new ArgumentException("Value must be positive.");
}
}
public static implicit operator DegreeOfParallelism(int value) => new DegreeOfParallelism(value);
}
It is supported by the base class Primitive<T> which implements basic operators, equality and comparer.
[PublicAPI]
[CannotApplyEqualityOperator]
public abstract class Primitive<T> : IEquatable<Primitive<T>>, IComparable<Primitive<T>>
{
private static readonly IComparer<Primitive<T>> Comparable = ComparerFactory<Primitive<T>>.Create(p => p.Value);
protected Primitive(T value)
{
// ReSharper disable once VirtualMemberCallInConstructor - it's ok to do this here because Validate is stateless.
Validate(Value = value);
}
protected abstract void Validate(T value);
[AutoEqualityProperty]
public T Value { get; }
#region IEquatable
public bool Equals(Primitive<T> other) => AutoEquality<Primitive<T>>.Comparer.Equals(this, other);
public override bool Equals(object obj) => obj is T other && Equals(other);
public override int GetHashCode() => AutoEquality<Primitive<T>>.Comparer.GetHashCode(this);
#endregion
#region IComparable
public int CompareTo(Primitive<T> other) => Comparable.Compare(this, other);
#endregion
public static implicit operator T(Primitive<T> primitive) => primitive.Value;
}
Triggers that I use are currently very simple too. It's just a base class that provides a single method for checking whether the trigger Matches. I have two of them.
public abstract class Trigger
{
public abstract bool Matches(DateTime tick);
}
public class CronTrigger : Trigger
{
private readonly CronExpression _cronExpression;
public CronTrigger(string cronExpression)
{
_cronExpression = CronExpression.Parse(cronExpression);
}
public string Schedule => _cronExpression.ToString();
public override bool Matches(DateTime tick)
{
return _cronExpression.Contains(tick);
}
}
public class CountTrigger : Trigger
{
public CountTrigger(int count)
{
Counter = new InfiniteCounter(count);
}
public IInfiniteCounter Counter { get; }
public override bool Matches(DateTime tick)
{
Counter.MoveNext();
return Counter.Position == InfiniteCounterPosition.Last;
}
}
Testing
So far I've created two tests for it (with XUnit). One testing the Job and a bigger one testing the Scheduler.
The first test checks whether the max-number of tasks is not exceeded and whether the Continuation tasks works correctly.
public class JobTest
{
[Fact]
public async Task Job_executes_no_more_than_specified_number_of_times()
{
var misfireCount = 0;
var job = new Job("test", Enumerable.Empty<Trigger>(), async token => await Task.Delay(TimeSpan.FromSeconds(3), token))
{
OnMisfire = j => misfireCount++,
MaxDegreeOfParallelism = 2
};
job.Execute(CancellationToken.None);
job.Execute(CancellationToken.None);
job.Execute(CancellationToken.None);
Assert.Equal(2, job.Count);
Assert.Equal(1, misfireCount);
// Wait until all jobs are completed.
await job.Continuation;
Assert.Equal(0, job.Count);
}
}
Testing Scheduler is now possible by using an observable that is ticking as I say:
public class SchedulerTest
{
[Fact]
public void Executes_job_according_to_triggers()
{
var job1ExecuteCount = 0;
var job2ExecuteCount = 0;
var misfireCount = 0;
var subject = new Subject<DateTime>();
var scheduler = new Scheduler(subject);
var unschedule1 = scheduler.Schedule(new Job("test-1", new[] { new CountTrigger(2) }, async token =>
{
Interlocked.Increment(ref job1ExecuteCount);
await Task.Delay(TimeSpan.FromSeconds(3), token);
})
{
MaxDegreeOfParallelism = 2,
OnMisfire = _ => Interlocked.Increment(ref misfireCount),
UnscheduleTimeout = TimeSpan.FromSeconds(4)
});
var unschedule2 = scheduler.Schedule(new Job("test-2", new[] { new CountTrigger(3) }, async token =>
{
Interlocked.Increment(ref job2ExecuteCount);
await Task.Delay(TimeSpan.FromSeconds(3), token);
})
{
MaxDegreeOfParallelism = 2,
OnMisfire = _ => Interlocked.Increment(ref misfireCount),
UnscheduleTimeout = TimeSpan.FromSeconds(4)
});
// Scheduler was just initialized and should not have executed anything yet.
Assert.Equal(0, job1ExecuteCount);
Assert.Equal(0, job2ExecuteCount);
// Tick once.
subject.OnNext(DateTime.Now);
// Still nothing should be executed.
Assert.Equal(0, job1ExecuteCount);
Assert.Equal(0, job2ExecuteCount);
// Now tick twice...
subject.OnNext(DateTime.Now);
subject.OnNext(DateTime.Now);
// Unschedule the job. This blocking call waits until all tasks are completed.
unschedule1.Dispose();
unschedule2.Dispose();
// Tick once again. Nothing should be executed anymore.
subject.OnNext(DateTime.Now);
// ...this should have matched the two triggers.
Assert.Equal(1, job1ExecuteCount);
Assert.Equal(1, job2ExecuteCount);
Assert.Equal(0, misfireCount);
}
}
The extension to fix missing seconds...
public static class ObservableExtensions
{
public static IObservable<DateTime> TruncateMilliseconds(this IObservable<DateTime> ticks)
{
return ticks.Select(DateTimeExtensions.TruncateMilliseconds);
}
public static IObservable<DateTime> FixMissingSeconds(this IObservable<DateTime> ticks)
{
var last = DateTime.MinValue;
return ticks.SelectMany(tick =>
{
if (tick.Millisecond > 0) throw new InvalidOperationException($"{nameof(FixMissingSeconds)} requires ticks without the millisecond part.");
// We have to start somewhere so let it be one second before tick if we are currently nowhere.
last = last == DateTime.MinValue ? tick.AddSeconds(-1) : last;
// Calculates the gap between tick and last. In normal case it's 1.
var gap = tick.DiffInSeconds(last);
// If we missed one second due to time inaccuracy,
// this makes sure to publish the missing second too
// so that all jobs at that second can also be triggered.
return
Enumerable
.Range(0, gap)
.Select(_ => last = last.AddSeconds(1));
});
}
}
can now also be tested.
public class ObservableExtensionsTest
{
[Fact]
public void Returns_ticks_unchanged_when_no_gap()
{
var ticks = new[] { 0, 1, 2 }.Select(s => DateTime.Parse($"2019-01-01 10:00:0{s}")).ToList();
Assert.Equal(ticks, ticks.ToObservable().TruncateMilliseconds().FixMissingSeconds().ToEnumerable().ToList());
}
[Fact]
public void Fixes_tick_gap()
{
var expected = new[] { 0, 1, 2 }.Select(s => DateTime.Parse($"2019-01-01 10:00:0{s}")).ToList();
var missing = new[] { 0, 2 }.Select(s => DateTime.Parse($"2019-01-01 10:00:0{s}")).ToList();
Assert.Equal(expected.ToList(), missing.ToObservable().TruncateMilliseconds().FixMissingSeconds().ToEnumerable().ToList());
}
}
Utilities
There are two more classes that drive the automatic scheduler.
One creates an observable that is ticking every-second:
public static class Tick
{
public static IObservable<DateTime> EverySecond(IDateTime dateTime)
{
return
Observable
.Interval(TimeSpan.FromSeconds(1))
.Select(_ => dateTime.Now());
}
}
the other provides and extension that is fixing missing seconds due to the occasional glitches in the ticking clock:
public static class ObservableExtensions
{
public static IObservable<DateTime> FixMissingSeconds(this IObservable<DateTime> ticks)
{
var last = DateTime.MinValue;
return ticks.SelectMany(tick =>
{
tick = tick.TruncateMilliseconds();
// We have to start somewhere so let it be one second before tick if we are currently nowhere.
last = last == DateTime.MinValue ? tick.AddSeconds(-1) : last;
// Calculates the gap between tick and last. In normal case it's 1.
var gap = (int)((tick - last).Ticks / TimeSpan.TicksPerSecond);
// If we missed one second due to time inaccuracy,
// this makes sure to publish the missing second too
// so that all jobs at that second can also be triggered.
return
Enumerable
.Range(0, gap)
.Select(_ => last = last.AddSeconds(1));
});
}
}
public interface IDateTime
{
DateTime Now();
}
public class DateTimeUtc : IDateTime
{
public DateTime Now() => DateTime.UtcNow;
}
For working with DateTime I also have these:
public static class DateTimeExtensions
{
public static DateTime TruncateMilliseconds(this DateTime dateTime)
{
return new DateTime(dateTime.Ticks - (dateTime.Ticks % TimeSpan.TicksPerSecond), dateTime.Kind);
}
public static int Diff(this DateTime later, DateTime earlier, long ticksPerX)
{
if (later < earlier) throw new ArgumentException($"'{nameof(later)}' must be greater or equal '{nameof(earlier)}'.");
return (int)((later - earlier).Ticks / ticksPerX);
}
public static int DiffInSeconds(this DateTime later, DateTime earlier)
{
return later.Diff(earlier, TimeSpan.TicksPerSecond);
}
}
This time my main focus is on testability and thread-safety. Do you think I need any locking or synchronisation anywhere? I'm not entirely sure I have thought of everythig. How about testing it? Can you see anything that cannot be tested and do you think the two tests are sane?
