Fileinterface_Cron_Job/FileinterfaceCronScheduler/Scheduler/FileSystemWatcherManager.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading.Tasks;

namespace FileInterface.Scheduler
{
    public sealed class FileSystemWatcherManager : IDisposable
    {
        private sealed class ActiveWatcher : IDisposable
        {
            public WatcherJobConfig Job;
            public FileSystemWatcher Watcher;
            public void Dispose() { if (Watcher != null) { Watcher.EnableRaisingEvents = false; Watcher.Dispose(); } }
        }

        private readonly object _sync = new object();
        private readonly WatcherJobRepository _repository;
        private readonly WatcherJobExecutor _executor;
        private readonly JobLog _log;
        private readonly Dictionary<string, ActiveWatcher> _watchers = new Dictionary<string, ActiveWatcher>(StringComparer.OrdinalIgnoreCase);
        private readonly ConcurrentDictionary<string, DateTime> _lastEvents = new ConcurrentDictionary<string, DateTime>(StringComparer.OrdinalIgnoreCase);
        private readonly ConcurrentDictionary<string, byte> _runningFiles = new ConcurrentDictionary<string, byte>(StringComparer.OrdinalIgnoreCase);
        private bool _disposed;

        public FileSystemWatcherManager(WatcherJobRepository repository, WatcherJobExecutor executor, JobLog log)
        {
            _repository = repository;
            _executor = executor;
            _log = log;
        }

        public void Start() { Reload(); }

        public void Reload()
        {
            lock (_sync)
            {
                ThrowIfDisposed();
                foreach (ActiveWatcher old in _watchers.Values) old.Dispose();
                _watchers.Clear();

                foreach (WatcherJobConfig job in _repository.GetAll())
                {
                    if (!job.Enabled) continue;
                    try
                    {
                        if (!Directory.Exists(job.SourceDirectory))
                        {
                            _log.Error(job, "Watcher source directory does not exist: " + job.SourceDirectory);
                            continue;
                        }

                        var fsw = new FileSystemWatcher(job.SourceDirectory, String.IsNullOrWhiteSpace(job.SearchPattern) ? "*.*" : job.SearchPattern)
                        {
                            IncludeSubdirectories = job.IncludeSubdirectories,
                            NotifyFilter = NotifyFilters.FileName | NotifyFilters.LastWrite | NotifyFilters.Size | NotifyFilters.CreationTime,
                            EnableRaisingEvents = false
                        };

                        WatcherJobConfig captured = job;
                        if (job.WatchCreated) fsw.Created += (s, e) => OnFileEvent(captured, e.FullPath, "Created");
                        if (job.WatchChanged) fsw.Changed += (s, e) => OnFileEvent(captured, e.FullPath, "Changed");
                        if (job.WatchRenamed) fsw.Renamed += (s, e) => OnFileEvent(captured, e.FullPath, "Renamed");
                        fsw.Error += (s, e) => _log.Error(captured, "FileSystemWatcher error: " + (e.GetException() == null ? "unknown" : e.GetException().Message));

                        var active = new ActiveWatcher { Job = job, Watcher = fsw };
                        _watchers[job.Id] = active;
                        fsw.EnableRaisingEvents = true;
                        _log.Info(job, "FileSystemWatcher active: " + job.SourceDirectory + "\\" + job.SearchPattern);
                    }
                    catch (Exception ex)
                    {
                        _log.Error(job, "Cannot start FileSystemWatcher: " + ex.Message);
                    }
                }
            }
        }

        public IList<string> GetActiveWatcherIds()
        {
            lock (_sync) return _watchers.Keys.ToList();
        }

        public bool ProcessExistingNow(string watcherId)
        {
            WatcherJobConfig job = _repository.GetById(watcherId);
            if (job == null) return false;
            Task.Run(() => ProcessExisting(job));
            return true;
        }

        private void ProcessExisting(WatcherJobConfig job)
        {
            if (!Directory.Exists(job.SourceDirectory))
            {
                _log.Error(job, "Source directory does not exist: " + job.SourceDirectory);
                return;
            }
            try
            {
                SearchOption option = job.IncludeSubdirectories ? SearchOption.AllDirectories : SearchOption.TopDirectoryOnly;
                string[] files = Directory.GetFiles(job.SourceDirectory, String.IsNullOrWhiteSpace(job.SearchPattern) ? "*.*" : job.SearchPattern, option)
                    .OrderBy(f => f, StringComparer.OrdinalIgnoreCase).ToArray();
                _log.Info(job, "Manual watcher scan found " + files.Length + " file(s).");
                foreach (string file in files) QueueFile(job, file, "ManualScan", false);
            }
            catch (Exception ex) { _log.Error(job, "Manual watcher scan failed: " + ex.Message); }
        }

        private void OnFileEvent(WatcherJobConfig job, string fullPath, string reason)
        {
            if (!job.Enabled || String.IsNullOrWhiteSpace(fullPath)) return;
            if (IsMoveTarget(job, fullPath)) return;
            QueueFile(job, fullPath, reason, true);
        }

        private void QueueFile(WatcherJobConfig job, string fullPath, string reason, bool debounce)
        {
            string key = job.Id + "|" + fullPath;
            DateTime now = DateTime.UtcNow;
            if (debounce)
            {
                DateTime previous;
                if (_lastEvents.TryGetValue(key, out previous) && (now - previous).TotalMilliseconds < Math.Max(0, job.DebounceMilliseconds))
                    return;
                _lastEvents[key] = now;
            }

            byte ignored;
            if (!_runningFiles.TryAdd(key, 0)) return;
            Task.Run(() =>
            {
                try
                {
                    if (!File.Exists(fullPath)) return;
                    _log.Info(job, "File event queued: " + reason + " - " + fullPath);
                    _executor.ExecuteFile(job, fullPath, reason);
                }
                catch (Exception ex) { _log.Error(job, "Watcher processing failed for " + fullPath + ": " + ex); }
                finally { _runningFiles.TryRemove(key, out ignored); }
            });
        }

        private static bool IsMoveTarget(WatcherJobConfig job, string fullPath)
        {
            if (!job.MoveFileAfterSuccess || String.IsNullOrWhiteSpace(job.MoveTargetDirectory)) return false;
            try
            {
                string file = Path.GetFullPath(fullPath).TrimEnd(Path.DirectorySeparatorChar, Path.AltDirectorySeparatorChar) + Path.DirectorySeparatorChar;
                string target = Path.GetFullPath(job.MoveTargetDirectory).TrimEnd(Path.DirectorySeparatorChar, Path.AltDirectorySeparatorChar) + Path.DirectorySeparatorChar;
                return file.StartsWith(target, StringComparison.OrdinalIgnoreCase);
            }
            catch { return false; }
        }

        private void ThrowIfDisposed() { if (_disposed) throw new ObjectDisposedException(GetType().FullName); }

        public void Dispose()
        {
            lock (_sync)
            {
                if (_disposed) return;
                _disposed = true;
                foreach (ActiveWatcher w in _watchers.Values) w.Dispose();
                _watchers.Clear();
            }
        }
    }
}