using System; using System.Collections.Generic; using System.IO.Abstractions; using System.Linq; using System.Threading.Tasks; using BO.AppServer.Metadata.Enums; using BO.ProcessServer.Business.Enums; using BO.ProcessServer.Business.StatisticCounters; using BO.ProcessServer.Business.Workers.Contexts; using BO.ProcessServer.Options; using Connector.PS; using NLog; using Quadarax.Foundation.Core.Logging; using Quadarax.Foundation.Core.Thread; using ILogger = Quadarax.Foundation.Core.Logging.ILogger; namespace BO.ProcessServer.Business.Workers { internal class ProcessingDocumentsWorker: LoopWorker { private const string CS_WK_NAME = "ProcessingDocument"; private ILogger _logger; public ProcessingDocumentsWorker(Configuration configuration,IList processes, IConnectionPS connection, IFileSystem fileSystem,Statistics statistics, Pool pool, ILogger logger) : base(CS_WK_NAME, new ProcessingDocumentsCtx(configuration, processes, connection, fileSystem, statistics, pool), logger) { _logger = logger; } public ProcessingDocumentsWorker(string workerName) : base(workerName) { } public ProcessingDocumentsWorker(string workerName, IWorkerContext context, ILogger logger = null) : base(workerName, context, logger) { } protected override async Task DoIteration(IWorkerContext context) { var ctx = (ProcessingDocumentsCtx)context; // ensure connection if (!ctx.Connection.IsOpen) ctx.Connection.Open(ctx.Configuration.ApiUser, ctx.Configuration.ApiUserPwd, ctx.Configuration.ApiMagic); await SaveProcessesState(ctx); while (ctx.Processes.Count(x => x.Status == ServerProcessStateEnum.Pending) < ctx.Configuration.System.ScenarioParallelThreads) { var docFiles = await ctx.Pool.GetSinglePendingDocumentFilesAsync(); if (docFiles.Item1 > -1) { Log(LogSeverityEnum.Info, $"Processing document ({docFiles.Item1}) with files {string.Join(";", docFiles.Item2.Select(x => "'" + ctx.FileSystem.Path.GetFileName(x)) + "'")} ..."); var docinfo = ctx.Connection.GetDocumentInfoAsync(docFiles.Item1); if (docinfo != null) { var process = new SingleDocumentCalculationWorker(ctx.Connection, ctx.FileSystem, docFiles.Item1, docFiles.Item2.ToList(), ctx.Configuration, ctx.Statistics, ctx.Pool, _logger); ctx.Processes.Add(process); process.Start(); await ctx.Connection.SetDocumentStateAsync(docFiles.Item1, DocumentStatusEnum.Processing); } } } } private async Task SaveProcessesState(ProcessingDocumentsCtx ctx) { if (ctx.Configuration.System.Status.Enabled) { if (DateTime.Now - ctx.LastMonitorStatusWrite > ctx.Configuration.System.Status.Interval) { await ctx.Binder.SaveAsync(TranslatePath(ctx.FileSystem, ctx.Configuration.System.Status.File), ctx.Processes); ctx.LastMonitorStatusWrite = DateTime.Now; } } } private string TranslatePath(IFileSystem fs, string path) { if (string.IsNullOrEmpty(path)) return path; return path.Replace(Constants.CS_TAG_CURRENT_DIR, fs.Directory.GetCurrentDirectory()); } } }