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&lt;DateTime&gt; Create(TimeSpan interval, IDateTime dateTime)
    {
        return
            Observable
                .Interval(TimeSpan.FromSeconds(1))
                .Select(_ =&gt; dateTime.Now());
    }

    public static IDisposable Schedule(this IObservable&lt;DateTime&gt; schedules, string cronExpressionString, Action&lt;DateTime&gt; 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() =&gt; DateTime.Now;
}

or
public class UtcDateTime : IDateTime
{
    public DateTime Now() =&gt; 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(&quot;0/1 * * * * * *&quot;, schedule =&gt;
{
    Console.WriteLine($&quot;DEBUG: {schedule} [{Thread.CurrentThread.ManagedThreadId}]&quot;);
});

scheduler.Schedule(&quot;0/5 * * * * * *&quot;, schedule =&gt;
{
    Console.WriteLine($&quot;ACTION: {schedule}&quot;);
});

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?
public static IObservable&lt;DateTime&gt; Create(TimeSpan interval, IDateTime dateTime)
{
    return
        Observable
            .Interval(TimeSpan.FromSeconds(1))
            .Select(_ =&gt; 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 =&gt;
{
    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&lt;DateTime&gt; 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 =&gt; 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&lt;DateTime&gt; schedules, string cronExpressionString, Action&lt;DateTime&gt; action)
{
    var cronExpression = Regex.Match(cronExpressionString, @"^0/(?&lt;s&gt;\d+)");
    return
        schedules
            .Scan(default(TimeRange), (previous, x) =&gt; 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;
    }
}