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 _watchers = new Dictionary(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _lastEvents = new ConcurrentDictionary(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _runningFiles = new ConcurrentDictionary(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 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(); } } } }