bo/ProcessServer/Business/Workers/ProcessingDocumentsWorker.cs

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<SingleDocumentCalculationWorker> 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());
        }

    }
}