using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Text; using System.Threading; namespace FileInterface.Scheduler { public sealed class WatcherJobExecutor { private readonly ICustomFunctionInvoker _invoker; private readonly JobLog _log; public WatcherJobExecutor(ICustomFunctionInvoker invoker, JobLog log) { _invoker = invoker; _log = log; } public void ExecuteFile(WatcherJobConfig job, string file, string reason) { if (job == null) throw new ArgumentNullException("job"); if (String.IsNullOrWhiteSpace(file)) return; if (!WaitUntilReadable(file, job.FileReadyRetries, job.FileReadyDelayMilliseconds)) { _log.Error(job, "File is not readable after retries: " + file); return; } string rawContent; using (var stream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)) using (var reader = new StreamReader(stream, Encoding.UTF8, true)) rawContent = reader.ReadToEnd(); string processedContent = ProcessContent(rawContent, job.RemoveEmptyLines, job.JoinSeparator); string[] inArgs = (job.InArgs ?? new List()) .Select(arg => Expand(arg, job, file, rawContent, processedContent, reason)) .ToArray(); _log.Info(job, "Calling CustomFunction '" + job.CustomFunction + "' with " + inArgs.Length + " inArg(s) for " + Path.GetFileName(file) + " (" + reason + ")."); CustomFunctionCallResult result = _invoker.Invoke(job.CustomFunction, inArgs); if (result == null) result = CustomFunctionCallResult.Fail("Invoker returned null."); if (!result.Success) { _log.Error(job, "CustomFunction failed: " + result.Message); return; } _log.Info(job, "CustomFunction succeeded: " + result.Message); if (job.MoveFileAfterSuccess && File.Exists(file)) { string movedTo = MoveProcessedFile(file, job.MoveTargetDirectory, job.OverwriteTargetFile); _log.Info(job, "Moved file to: " + movedTo); } } private static bool WaitUntilReadable(string file, int retries, int delayMs) { retries = Math.Max(2, retries); delayMs = Math.Max(10, delayMs); long previousLength = -1; DateTime previousWrite = DateTime.MinValue; int stableChecks = 0; for (int i = 0; i < retries; i++) { if (!File.Exists(file)) return false; try { var info = new FileInfo(file); long length = info.Length; DateTime lastWrite = info.LastWriteTimeUtc; using (var stream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)) { if (stream.Length != length) length = stream.Length; } if (length == previousLength && lastWrite == previousWrite) { stableChecks++; if (stableChecks >= 1) return true; } else { stableChecks = 0; previousLength = length; previousWrite = lastWrite; } } catch (IOException) { stableChecks = 0; } catch (UnauthorizedAccessException) { stableChecks = 0; } Thread.Sleep(delayMs); } return false; } private static string ProcessContent(string text, bool removeEmptyLines, string separator) { if (!removeEmptyLines) return text ?? ""; string[] lines = (text ?? "").Replace("\r\n", "\n").Replace('\r', '\n').Split('\n'); return String.Join(separator ?? ";", lines.Where(l => !String.IsNullOrWhiteSpace(l)).Select(l => l.Trim())); } private static string Expand(string template, WatcherJobConfig job, string file, string rawContent, string processedContent, string reason) { string value = template ?? ""; string fileName = Path.GetFileName(file) ?? ""; string extension = Path.GetExtension(file) ?? ""; string directory = Path.GetDirectoryName(file) ?? ""; return value .Replace("{{CONTENT}}", processedContent ?? "") .Replace("{{RAW_CONTENT}}", rawContent ?? "") .Replace("{{FILE_NAME}}", fileName) .Replace("{{FILE_PATH}}", file ?? "") .Replace("{{FILE_DIR}}", directory) .Replace("{{FILE_EXT}}", extension) .Replace("{{JOB_NAME}}", job.Name ?? "") .Replace("{{EVENT}}", reason ?? "") .Replace("{{NOW}}", DateTime.Now.ToString("yyyy-MM-ddTHH:mm:ss.fffK")); } private static string MoveProcessedFile(string source, string targetDirectory, bool overwrite) { Directory.CreateDirectory(targetDirectory); string target = Path.Combine(targetDirectory, Path.GetFileName(source)); if (File.Exists(target)) { if (overwrite) File.Delete(target); else { string name = Path.GetFileNameWithoutExtension(source); string ext = Path.GetExtension(source); target = Path.Combine(targetDirectory, name + "_" + DateTime.Now.ToString("yyyyMMdd_HHmmss_fff") + ext); int i = 1; while (File.Exists(target)) { target = Path.Combine(targetDirectory, name + "_" + DateTime.Now.ToString("yyyyMMdd_HHmmss_fff") + "_" + i + ext); i++; } } } File.Move(source, target); return target; } } }