A File Change Monitor Class in C#

Wrote this File changes monitor service to monitor 1 or more files in a particular directory for file write accesses.

public class FileChangedEvent
{
    public FileChangedEvent(string fullPath, string fileName, 
      WatcherChangeTypes changeType)
    {
        FullPath = fullPath;
        FileName = fileName;
        ChangeType = changeType;
    }

    // Full file path and name
    public string FullPath { get; private set; }
    
    // File name only
    public string FileName { get; private set; }
    
    // The type of change
    public WatcherChangeTypes ChangeType { get; private set; }
}

public interface IFileChangeMonitor
{
    void Dispose();
    string DirectoryPath { get; }
    void Start();
    public bool IsActive { get; }
    void Stop();
    IDisposable SubscribeOnChanged(string fileName, 
      Action<FileChangedEvent> onChanged, double throttle_ms = 100);
    IDisposable SubscribeOnError(Action<ErrorEventArgs> onError);
}

public class FileChangeMonitor : IDisposable, IFileChangeMonitor
{
    // https://weblogs.asp.net/ashben/31773 
    private readonly FileSystemWatcher _fileSystemWatcher; 
    private readonly DirectoryInfo _directory; // directory being monitored
    private const string DefaultFileFilter = @"*.json";

    // TODO: add NotifyFilter parameter
    public FileChangeMonitor(DirectoryInfo dir, 
      string defaultFileFilter = DefaultFileFilter)
    {
        //const int MAX_NUM_FILES_TO_MONITOR = 10;
        //const int AVG_FILE_NAME_LENGTH = 30;
        //const int FOUR_KB = 1024 * 4;
        Debug.Assert(dir != null);
        Debug.Assert(dir.Exists == true);

        _directory = new DirectoryInfo(dir.FullName);
        //int bufferSize = ((MAX_NUM_FILES_TO_MONITOR * 
        // (16 + AVG_FILE_NAME_LENGTH * 2) % FOUR_KB) + 1) * FOUR_KB;
        _fileSystemWatcher = new FileSystemWatcher(_directory.FullName)
        {
            //InternalBufferSize = bufferSize, default is enough
            NotifyFilter = NotifyFilters.CreationTime | NotifyFilters.LastWrite,
            Filter = defaultFileFilter,
            IncludeSubdirectories = false
        };
    }

    public string DirectoryPath => _directory?.FullName;

    public IDisposable SubscribeOnChanged(string fileName, 
      Action<FileChangedEvent> onChanged, 
      double throttle_ms = 200)
    {
        Debug.Assert(_directory?.Exists == true);
        Trace.WriteLine($"*** Subscribing for changes to 
  {Path.Combine(DirectoryPath, fileName)}");
        return GetFileChangedPublisher(fileName, throttle_ms)
          .Subscribe(onChanged);
    }

    public void Start()
    {
        Debug.Assert(_directory?.Exists == true);
        _fileSystemWatcher.EnableRaisingEvents = true;
    }

    public bool IsActive => _fileSystemWatcher.EnableRaisingEvents;

    public void Stop()
    {
        Debug.Assert(_directory?.Exists == true);
        _fileSystemWatcher.EnableRaisingEvents = false;
    }

    public void Dispose()
    {
        if (IsActive)
            Stop();
        GC.SuppressFinalize(this);
        _fileSystemWatcher.Dispose();
    }

    // Returns an IObservable, which is effectively a publisher which
    // can be subscribed to and disposed of when finished with
    private IObservable<FileChangedEvent> GetFileChangedPublisher(
      string targetFileName, 
      double throttle_ms = 200)
    {
        Debug.Assert(_directory?.Exists == true);

        IObservable<EventPattern<FileSystemEventArgs>> fswCreated = 
          Observable.FromEventPattern<FileSystemEventArgs>(_fileSystemWatcher, "Changed");

        return fswCreated.
               Where((EventPattern<FileSystemEventArgs> ev) => ev.EventArgs.Name == targetFileName).
               // Rx Extension Sample() will only allow the last event within a time
               // interval to pass through. This effectively throttles events
               Sample(TimeSpan.FromMilliseconds(throttle_ms)).
               Select(ev => new FileChangedEvent(ev.EventArgs.FullPath, ev.EventArgs.Name, ev.EventArgs.ChangeType)); 
    }

    private IObservable<ErrorEventArgs> GetErrorPublisher()
    {
        Debug.Assert(_directory?.Exists == true);

        return Observable
          .FromEventPattern<ErrorEventHandler, ErrorEventArgs>(
                   ev => _fileSystemWatcher.Error += ev,
                   ev => _fileSystemWatcher.Error -= ev)
          .Select(x => x.EventArgs);
    }

    public IDisposable SubscribeOnError(Action<ErrorEventArgs> onError)
    {
        Debug.Assert(_directory?.Exists == true);
        return GetErrorPublisher().Subscribe(onError);
    }
}

The thing is it is an example of how to integrate React with a file system watcher or an existing class that publishes events using standard C# events. Here is an example of how to use the service

DirectoryInfo dir = new DirectoryInfo("\\Some\Directory");
FileChangeMonitor datafileMonitor = new FileChangeMonitor(dir);
datafileMonitor.Start();
...
// Subscribed to just 1 file but we could subscribe to multiple in the same directory
IDisposable subscription = datafileMonitor.SubscribeOnChanged(
    "FileToMonitor.json",
    fce => DoSomethingWith(fce));   
...
subscription.Dispose(); // Finish with our subscription to a specific file's changes
...
datafileMonitor.Stop(); // Optionally we could just call just Dispose()
datafileMonitor.Dispose();

We could change the class so that the type of changes monitored could be passed in through the class constructor (the NotifyFilter enumeration). Also, an option to change the internal buffer size could be required.

Exposing C# Events using Reactive Extensions

There is a way to convert existing C# events to an IObservable<>. There is a good article here on this

public class SimpleEventSource
{
  public event EventHandler<MyEventArgs> SimpleEvent;
     
  // Normally this would be private (event is triggered internally) 
  // but for test purposes we make it public
  public void FireSimpleEvent(MyEvent myevent)
  {
    EventHandler<MyEventArgs> tmp = SimpleEvent;
    if (tmp != null)
    {
      tmp(this, new MyEventArgs(myEvent));
    }
  }   
 
  // This returns an IObservable (effectively an event pulisher)
  public IObservable<EventPattern<MyEventArgs>> MyEventPublisher
  {
    get
    {       // To consume SimpleEvent as an IObservable:
        IObservable<EventPattern<MyEventArgs>> eventAsObservable = Observable.
            FromEventPattern<MyEventArgs>(
              ev => SimpleEvent += ev,
              ev => SimpleEvent -= ev);
        return eventAsObservable;
    }
  }

}

Note that MyEventArgs is something derived from EventArgs that allows access to the "MyEvent" instance, see below. Here is a tester:

public void TestSimpleEventSource()
{
    var ses = new SimpleEventSource();
    ses.FireSimpleEvent(new MyEvent("1st event")); // This one is missed

    using (var subscription = ses.MyEventPublisher.Subscribe(
        ev => Console.WriteLine(ev.EventArgs.MyEvent.Text + " fired")))
    {
        ses.FireSimpleEvent(new MyEvent("2nd event"));
        ses.FireSimpleEvent(new MyEvent("3rd event"));               
    }

    Console.WriteLine("Subscription cancelled");
}

and the output

2nd event fired
3rd event fired
Subscription cancelled

Replacing C# Events using Reactive Extensions

Lets take this one step further, and completely replace the C# event with a reactive style event publisher

internal class SimpleEventSource
{
    private Subject<MyEvent> _myEventSubject = new Subject<MyEvent>();

    public IObservable<MyEvent> MyEventPublisher => this._myEventSubject.AsObservable();

    public void FireSimpleEvent(MyEvent myEvent) => this._myEventSubject.OnNext(myEvent);
}

And the same test but using this class:

public void TestSimpleEventSource2()
{
    var ses = new SimpleEventSource2();
    ses.FireSimpleEvent(new MyEvent("1st event"));

    using (var subscription = ses.MyEventPublisher.Subscribe(
        ev => Console.WriteLine(ev.Text + " fired")))
    {
        ses.FireSimpleEvent(new MyEvent("2nd event"));
        ses.FireSimpleEvent(new MyEvent("3rd event"));
    }

    Console.WriteLine("Subscription cancelled");
}

Here is the code for the simple event classes if it makes things clearer

public class MyEvent
{
    public MyEvent(string text)
    {
        Text = text;
    }
    public string Text { get; private set; }
}

public class MyEventArgs : EventArgs
{
    public MyEventArgs(MyEvent myEvent)
        : base()
    {
        MyEvent = myEvent;
    }
    public MyEvent MyEvent { get; private set; }
}

C# Event Aggregators

Did some research into event aggregators. They are a great way to allow events to be fired betwen classes without the classes being tightly coupled, in other words without either class having to know about the other. Standard C# events make the classes tighly coupled, the listener needs to know about the firer. Sometimes with standard c# events, an event will need to be transmitted through an intermediate sequence of objects just to reach the target recipient, this is very cumbersome and laborious. Event aggregators completely decouple the event publishers from the event subscribers as long as both subscriber and publisher objects can access the event aggregator their is no need for intermediate objects to relay the events

Would make an excellent start for a simulator which has both hardware and software events occuring, an event aggregator could be used to input events to the system. As it is completely decoupled from the system, it would be easy to emulate firing these software events say from a GUI for emulation purposes, and also easy to connect to real hardware to populate those same software events.

I found several links that use a Reactive Extensions to implement an Event Aggregator. Using the Reactive Extensions the code to implement an aggregator is very small and compact.

1. https://mikebridge.github.io/post/csharp-domain-event-aggregator

It is also a great example in the usage of Reactive Extensions. Currently I have combined them into this form:

using System.Reactive.Linq; // You'll need the Reactive nuget package
using System.Reactive.Subjects;
...

public interface IRxEventAggregator : IEventAggregator, IDisposable
{
    /// <summary>
    ///  Use this method to apply RX (LINQ style) expressions on
    ///  the events before subscribing
    /// </summary>
    /// <typeparam name="TEvent"></typeparam>
    /// <returns></returns>
    IObservable<TEvent> GetEventsObservable<TEvent>();
}

public interface IEventAggregator
{
    IDisposable Subscribe<T>(Action<T> action);
    void Publish<T>(T @event);
}


/// <summary>
/// An EventAggregator that is based on reactive extensions and supports
/// </summary>
public class RxEventAggregator
    : IRxEventAggregator
{
    readonly Subject<object> _subject = new Subject<object>();

    // Use this method to use RX (LINQ style) expressions
    // on the events before subscribing
    public IObservable<TEvent> GetEventsObservable<TEvent>()
    {
        return _subject.OfType<TEvent>().
            AsObservable();
    }

    public IDisposable Subscribe<T>(Action<T> action)
    {
        return GetEventsObservable<T>().
            Subscribe(action);
    }

    public void Publish<TEvent>(TEvent sampleEvent)
    {
        _subject.OnNext(sampleEvent);
    }

    #region IDisposable

... // See below

    #endregion
}

Generally an Event Aggregator lives throughout the life of your software so it really does not need disposing of. You could remove the Dispose code completely. For that reason I list it here just for reference:

    #region IDisposable

    bool _disposed;

    ~RxEventAggregator()
    {
        Dispose(false);
         // Useful for detecting memory leaks
        Debug.Assert(false, nameof(RxEventAggregator) + 
            " was not disposed of");
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this); 
    }

    protected virtual void Dispose(bool disposing)
    {
        if (_disposed) return;

        _subject.Dispose();
        _disposed = true;
    }

    #endregion

Here's another event aggregator:

// When there are a lot of events with lots of subscribers
public class RxEventAggregator
: IRxEventAggregator
{
    private volatile ConcurrentDictionary<Type, object> _subjects
        = new ConcurrentDictionary<Type, object>();

    // Use this method to use RX (LINQ style) expressions
    // on the events before subscribing
    public IObservable<TEvent> GetEventsObservable<TEvent>()
    {
        Trace.Assert(!_disposed, "Trying to access a disposed EventAggregator");
        var subject =
            (ISubject<TEvent>)_subjects.GetOrAdd(typeof(TEvent),
                        _ => new Subject<TEvent>());
        return subject.AsObservable();
    }

    public IDisposable Subscribe<T>(Action<T> action)
    {
        Trace.Assert(!_disposed, "Trying to access a disposed EventAggregator");
        return GetEventsObservable<T>().
            Subscribe(action);
    }

    public void Publish<TEvent>(TEvent sampleEvent)
    {
        Trace.Assert(!_disposed, "Trying to access a disposed EventAggregator");
        if (_subjects.TryGetValue(typeof(TEvent), out var subject))
        {
            ((ISubject<TEvent>)subject)
                .OnNext(sampleEvent);
        }
    }

    // If your Aggregator is a life (of the application) long one you can simply not bother with disposing of it
    #region IDisposable 

    bool _disposed;

    ~RxEventAggregator2()
    {
        Debug.Assert(false, nameof(RxEventAggregator2) + " was not disposed of"); // Useful for detecting memory leaks
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);  // no need to call finalizer now
    }

    protected virtual void Dispose(bool disposing)
    {
        if (_disposed) return;

        _disposed = true;
        // Make them unaccessible
        var hiddenSubjects = Interlocked.Exchange(ref _subjects, new ConcurrentDictionary<Type, object>());
        // Then dispose of them
        foreach (var subject in hiddenSubjects.Values.Cast<IDisposable>())
        {
            subject.Dispose();
        }

    }

    #endregion
}

Reactive Extensions

Here is a good introduction to the main interfaces: http://introtorx.com/Content/v1.0.10621.0/02_KeyTypes.html
IObserver<T> - Reader/Consumer/Observer
IObservable<T> - Writer/Publisher/Observable sequence. [I find this interface name confusing.]
ISubject<in TSource, out TResult> - Represents an object that is both an observable sequence as well as an observer. Used in (non-reactive) samples but not in real situations.
In comparison to IEnumerable<T> where items are pulled out, in an IObservable<T> items are pushed in
A simple sample: Consider an image processing system in a factory that scans parts for results. The outcome of the scan could be one of three states

public enum ImageProcessingSystemResult
{
    Failed, // failed image processing check
    Passed, // succeeded  image processing check
    Indeterminate
}

Here is the publisher of the image processing system results:

// Publishes a sequence of ImageProcessingSystemResult to the observer
public class ImageProcessingSystemPublisher : IObservable<ImageProcessingSystemResult>
{
    public IDisposable Subscribe(IObserver<ImageProcessingSystemResult> observer)
    {
        // These results in this case are just published one after another
        observer.OnNext(ImageProcessingSystemResult.Passed); 
        observer.OnNext(ImageProcessingSystemResult.Passed);
        observer.OnNext(ImageProcessingSystemResult.Failed);
        observer.OnNext(ImageProcessingSystemResult.Passed);
        observer.OnCompleted();
        return Disposable.Empty;
    }
}

Here is a generic results observer:

// Reads/Observes a sequence of T and decides what to do with it
// Basically, what you want to do when you receive some data
public class MyObserver<T> : IObserver<T>
{
    public void OnNext(T value)
    {
        string output = value != null ? value.ToString() : "null";
        Console.WriteLine("Received value: {0}", output);
    }
    public void OnError(Exception error)
    {
        Console.WriteLine("Sequence faulted with exception: {0}", error.ToString());
    }
    public void OnCompleted()
    {
        Console.WriteLine("Sequence terminated");
    }
}

Here is how it all goes together:

public void UnreactiveSimpleTest()
{
    var imageProcessingSystemPublisher = new ImageProcessingSystemPublisher();
    var observer = new MyObserver<ImageProcessingSystemResult>();
    // Here is how the observer subscribes to the publisher
    using (imageProcessingSystemPublisher.Subscribe(observer))
    { }
}

Here is the output:

Received value: Passed
Received value: Passed
Received value: Failed
Received value: Passed
Sequence terminated

Press any key to continue ...

Hot and Cold Observables (i.e. publishers)

Cold - Sequences that start producing notifications on request (when subscribed to), and each subscriber gets a copy of the same notification stream from the beginning.
Hot - Sequences that are active and produce notifications regardless of subscriptions (just like c# events).

A 'Cold' Observable can be converted to a 'Hot' Observable by using the Publish() extension method. With a hot Observable it is necessary to use "Connect()" to connect to the published sequence. This returns an IDisposable which must be disposed of when the Observable sequence is finished with.