# Scheduler built with observables

I sometimes have to schedule some tasks and wanted to have my own reusable scheduler. Since I like the Rx, I build it on top of it.

### Follow-up

There is a newer version of the Scheduler.

The Scheduler class is really simple. It's backed by a newer implementation of my old CronExpression.

There are currently only these two methods. The factory method Create creates a new scheduler that ticks at the specified intervals and provides schedules as DateTimes to the observers.

Jobs are scheduled with the Schedule extension. This one requires a cron-expression and the action to execute.

public static class Scheduler
{
public static IObservable<DateTime> Create(TimeSpan interval, IDateTime dateTime)
{
return
Observable
.Interval(TimeSpan.FromSeconds(1))
.Select(_ => dateTime.Now());
}

public static IDisposable Schedule(this IObservable<DateTime> schedules, string cronExpressionString, Action<DateTime> action)
{
var cronExpression = CronExpression.Parse(cronExpressionString);
return
schedules
.Where(cronExpression.Contains)
.Subscribe(action);
}
}


The DateTime abstraction is supported by the IDateTime interface:

public interface IDateTime
{
DateTime Now();
}


which is implemented as

public class LocalDateTime : IDateTime
{
public DateTime Now() => DateTime.Now;
}


or

public class UtcDateTime : IDateTime
{
public DateTime Now() => DateTime.UtcNow;
}


### Example

In order to use it I just create a scheduler, specify what kind of timestamp it should generate and schedule some actions:

var scheduler = Scheduler.Create(TimeSpan.FromSeconds(1), new LocalDateTime());

scheduler.Schedule("0/1 * * * * * *", schedule =>
{
Console.WriteLine($"DEBUG: {schedule} [{Thread.CurrentThread.ManagedThreadId}]"); }); scheduler.Schedule("0/5 * * * * * *", schedule => { Console.WriteLine($"ACTION: {schedule}");
});


The output is:

DEBUG: 13/04/2018 22:32:09 [10]
DEBUG: 13/04/2018 22:32:10 [12]
ACTION: 13/04/2018 22:32:10
DEBUG: 13/04/2018 22:32:11 [10]
DEBUG: 13/04/2018 22:32:12 [14]
DEBUG: 13/04/2018 22:32:13 [8]
DEBUG: 13/04/2018 22:32:14 [12]
ACTION: 13/04/2018 22:32:15
DEBUG: 13/04/2018 22:32:15 [8]
DEBUG: 13/04/2018 22:32:16 [12]
DEBUG: 13/04/2018 22:32:17 [8]


There seems to be no rocket science here but this might elusive. Can/should this scheduler by improved in any way?

• .Interval(TimeSpan.FromSeconds(1)) you should be using the parameter interval – Xiaoy312 Apr 13 '18 at 21:53

public static IObservable<DateTime> Create(TimeSpan interval, IDateTime dateTime)
{
return
Observable
.Interval(TimeSpan.FromSeconds(1))
.Select(_ => dateTime.Now());
}


You forgot to plug interval parameter. And, I would consider renaming it resolution.

I wouldn't trust the values from selecting DatetTime::Now or DateTime::UtcNow:

// try running this for a while
var scheduler = Scheduler.Create(TimeSpan.FromSeconds(1), new LocalDateTime());
scheduler.Schedule("0/1 * * * * * *", schedule =>
{
Console.WriteLine(\$"ACTION: {schedule}: {schedule:ss.fff}");
});

ACTION: 2018-04-13 6:28:29 PM: 29.987
ACTION: 2018-04-13 6:28:30 PM: 30.987
ACTION: 2018-04-13 6:28:32 PM: 32.001
ACTION: 2018-04-13 6:28:33 PM: 33.001


Depending on CronExpression::Contains implementation, the scheduler could be skipping task when DateTime.Now.Millisecond gets near 0 or 999.

I've attempted to solve that by adding index * interval to a snapshot of time, however this solution suffered a problem: The delay is cumulative, so the schedule's time will slowly deviate(fall behind) from the real time:

// dont use this
public static IObservable<DateTime> Create2(TimeSpan interval, IDateTime dateTime)
{
var snapshot = dateTime.Now();
var offset = interval.Ticks - snapshot.Ticks % interval.Ticks;

return Observable.Interval(interval)
.Delay(TimeSpan.FromTicks(offset))
.Select(x => snapshot.AddTicks(interval.Ticks * (x + 1) + offset));
}


An better solution would be to provide a range of time, and check if expression fall between the range:

public static IDisposable Schedule2(this IObservable<DateTime> schedules, string cronExpressionString, Action<DateTime> action)
{
var cronExpression = Regex.Match(cronExpressionString, @"^0/(?<s>\d+)");
return
schedules
.Scan(default(TimeRange), (previous, x) => new TimeRange(previous?.End, x))
.Where(cronExpression.Contains)
.Subscribe(action);
}

public class TimeRange
{
public DateTime? Start { get; }
public DateTime End { get; }

public TimeRange(DateTime? start, DateTime end)
{
this.Start = start;
this.End = end;
}
}

• You forgot to plug interval parameter. oops, refactoring mistake :-] Since the smallest value of a cron-expression is seconds, I think the milliseconds can be ignored... but I see that I have to take care of the missing ones :-o Could you explain the range approach in more detail? I cannot figure this out :-( but you have my vote! – t3chb0t Apr 14 '18 at 8:00

Xiaoy312 has a point about the risk of missing seconds when using Interval. It seems that each "tick" of Observable.Interval() waits until the previous returns. So if the OnNext halts the thread a while, the next "tick" is fired too late and you may miss a second or two.

Trying to start each scheduled task on a new thread - for instance via a System.Timers.Timer seems to be a bad idea, since the same Cron job probably can not run concurrently(?).

One way not to lose any seconds using Observable.Interval is as follows:

  Random rand = new Random(5);

IDisposable subscription = source.Subscribe(
x =>
{
Console.WriteLine("OnNext: {0}", x);
Console.WriteLine("After Sleep OnNext: {0}", x);
},
ex => Console.WriteLine("OnError: {0}", ex.Message),
() => Console.WriteLine("OnCompleted"));

Console.WriteLine("Press ENTER to unsubscribe...");

• This is an interesting approach but unfortunatelly if you constrain the thread, all worker tasks will execute on the same one. If you get the x.Timestamp.DateTime property and format it with milliseconds you'll notice that it suffers from the same inaccuracy as other techniques :-( I've been experimenting with the missing second for some time and I think I might have found how to fix it. I'll post it later but it checks if now is >1s from last. If so, I return two timestamps via SelectMany. It doesn't matter that I return two at the same time, most important is to not loose it. – t3chb0t Apr 14 '18 at 22:13
• @t3chb0t: OK, it seems to be two different problems then. As I understand/understood ObserveOn(NewThreadScheduler.Default), each interval should be executed on its own thread, but that is not the case. And it seems not to be the case no matter how you configure Observable.Intervals (?) – Henrik Hansen Apr 15 '18 at 4:49