FileSystemWatcher Using Rx

This is a simple program which checks and reports for error files produced by ETL jobs.

namespace ErrorFileWatcher
{
public class ErrFileWatcher
{
private readonly FileSystemWatcher _fileSystemWatcher;
public IObservable ChangedFiles { get; set; }
public IObservable CreatedFiles { get; set; }
public IObservable DeletedFiles { get; set; }
public IObservable RenamedFiles { get; set; }
public IObservable Errors { get; set; }


public ErrFileWatcher(string directory, string filter, bool includeSubdirectories)
{
_fileSystemWatcher = new FileSystemWatcher(directory, filter)
{
EnableRaisingEvents = true,
IncludeSubdirectories = includeSubdirectories
};

ChangedFiles = Observable.FromEvent(
(EventHandler e) => new FileSystemEventHandler(e),
e => _fileSystemWatcher.Changed += e,
e => _fileSystemWatcher.Changed -= e).Select(x => x.EventArgs);

CreatedFiles = Observable.FromEvent(
(EventHandler e) => new FileSystemEventHandler(e),
e => _fileSystemWatcher.Created += e,
e => _fileSystemWatcher.Created -= e).Select(x => x.EventArgs);


DeletedFiles = Observable.FromEvent(
(EventHandler e) => new FileSystemEventHandler(e),
e => _fileSystemWatcher.Deleted += e,
e => _fileSystemWatcher.Deleted -= e).Select(x => x.EventArgs);


RenamedFiles = Observable.FromEvent(
(EventHandler e) => new RenamedEventHandler(e),
e => _fileSystemWatcher.Renamed += e,
e => _fileSystemWatcher.Renamed -= e).Select(x => x.EventArgs);



Errors = Observable.FromEvent(
(EventHandler e) => new ErrorEventHandler(e),
e => _fileSystemWatcher.Error += e,
e => _fileSystemWatcher.Error -= e).Select(x => x.EventArgs);


}

}
}

Windows Service code which calls the above class and also calls the sendemail method

try
{

while (true)
{
if (!RunToday)
{
if (IsItTimeToRun())
{
EventLog.WriteEntry("ErrorFileWactcher", "Watcher Started For " + DateTime.Now.Date, EventLogEntryType.Information);
SendEmails emails = new SendEmails();
Stopwatch sW = new Stopwatch();
sW.Start();
TextWriter writer = new StreamWriter(ConfigurationManager.AppSettings["FilePath"].ToString());
ErrFileWatcher fileWatch = new ErrFileWatcher(ConfigurationManager.AppSettings["FILEPATHTOBEMONITORED"], "*.txt", false);
fileWatch.CreatedFiles.Where(p => p.FullPath.Length > 0).Select(p => p.Name).Subscribe(p => writer.WriteLine(p.ToString() + " Created"));
fileWatch.ChangedFiles.Where(p => p.FullPath.Length > 0).Select(p => p.Name).Subscribe(p => writer.WriteLine(p.ToString() + " Changed"));
fileWatch.DeletedFiles.Where(p => p.FullPath.Length > 0).Select(p => p.Name).Subscribe(p => writer.WriteLine(p.ToString() + " Deleted"));
fileWatch.RenamedFiles.Where(p => p.FullPath.Length > 0).Select(p => p.Name).Subscribe(p => writer.WriteLine(p.ToString() + " Renamed"));
while (true)
{
if (sW.Elapsed.Minutes > 10)
{
sW.Stop();
writer.Flush();
writer.Dispose();
emails.SendEmail(true, ConfigurationManager.AppSettings["FilePath"].ToString(), "ERRORREPORT");
break;
}
}
}
else
{
Thread.Sleep(int.Parse(ConfigurationManager.AppSettings["SleepTime"]));
}
}
else
{
Thread.Sleep(int.Parse(ConfigurationManager.AppSettings["SleepTime"]));
}
}
}
catch (Exception ex)
{
EventLog.WriteEntry("ErrorFileWactcher", ex.Message, EventLogEntryType.Error);
}

Rx sample

In this post i have implemented a simple program which writes all the information of a customer class to a file.I took this example as it can be functional in a very short period of time.The code in this blog is inspired from Rx wiki.

using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Linq;
using System.Text;
using System.Collections.Specialized;
using System.Configuration;
using System.IO;
using System.Threading;
using System.Concurrency;
using System.Diagnostics;

namespace ObservablePrototype
{
public class Prototype
{
string errorfile = ConfigurationManager.AppSettings["CustomerFile"];

public void RunProto()
{
Stopwatch sw = new Stopwatch();
sw.Start();

//Making Customer as observableCollection 
var customers = new ObservableCollection();

// adding event handler 
var customerChanges = Observable.FromEvent(
(EventHandler ev)
=> new NotifyCollectionChangedEventHandler(ev),
ev => customers.CollectionChanged += ev,
ev => customers.CollectionChanged -= ev);

//ling query for notify collection changed

var line = (from c in customerChanges
where c.EventArgs.Action == NotifyCollectionChangedAction.Add
from cu in c.EventArgs.NewItems.Cast().ToObservable()
select cu).BufferWithCount(10);

//using scheduler or u can just subscribe directly
Scheduler.ThreadPool.Schedule(() =>
{
line.Subscribe(cus =>
{
WriteCustomers(cus.ToList());
}
);
});

//start adding customers  
AddCustomers(customers);
WriteTimings(sw.Elapsed.Seconds + "Observable");
}

private void WriteTimings(string p)
{
TextWriter writer = new StreamWriter(errorfile, true);
writer.WriteLine(p);
writer.Close();
}



private void AddCustomers(ObservableCollection customers)
{
for (int i = 0; i < 50; i++)
            {
                Thread.Sleep(500);
                customers.Add(new Customer() { firstName = "K", lastName = "P" + i });
            }
            
        }



        public void StraightWrite()
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            List customers = new List();
for (int i = 0; i < 50; i++)
            {
                Thread.Sleep(500);
                customers.Add(new Customer() { firstName = "K", lastName = "P" + i });
            }
            WriteCustomers(customers);
            WriteTimings(sw.Elapsed.Seconds + "Normal");
        }



        private void WriteCustomers(List customers)
{
foreach (var c in customers)
{
TextWriter writer = new StreamWriter(errorfile, true);
writer.WriteLine("FirstName : " + c.firstName + "  LastName : " + c.lastName);
writer.Close();
}
}

}
}