bo/ProcessServer/Business/Workers/PullDocumentsWorker.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.StatisticCounters;
using BO.ProcessServer.Business.Workers.Contexts;
using BO.ProcessServer.Options;
using Connector.PS;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.Thread;

namespace BO.ProcessServer.Business.Workers
{
    internal class PullDocumentsWorker : LoopWorker
    {
        #region *** Constants ***
        private const string CS_WK_NAME = "PullDocument";
        #endregion
        
        #region *** Constructors ***
        public PullDocumentsWorker(IConnectionPS connection, TimeSpan loopEveryTimeSpan, IFileSystem fileSystem,Configuration configuration, Statistics statistics,Pool pool, ILogger logger) : base(CS_WK_NAME, new PullCtx(connection, fileSystem,configuration, statistics, pool), logger)
        {
            IterationsDelayBetween = loopEveryTimeSpan;
        }

        public PullDocumentsWorker(string workerName) : base(workerName)
        {
        }

        public PullDocumentsWorker(string workerName, IWorkerContext context, ILogger logger = null) : base(workerName, context, logger)
        {
        }
        #endregion

        #region *** Iteration ***

        protected override async Task DoIteration(IWorkerContext context)
        {
            var ctx = (PullCtx)context;

            var scenarioMan = new Scenarios.Scenarios(ctx.Configuration, GetLogger());

            // get all pending document and set lock status (pending->working)
            var documents = await ctx.Connection.GetDocumentsAsync(DocumentsScopeEnums.Pending);
            foreach (var document in documents)
            {
                if (scenarioMan.ExistsScenario(document))
                {
                    // scenario for document exists
                    var downloadedFiles = new List<string>();
                    foreach (var art in document.Artifacts.Where(x => x.Type == ArtifactTypeEnum.Input))
                    {
                        await using (var content = await ctx.Connection.GetDocumentContentAsync(document.Id, art.Id))
                        {
                            var fileName = ctx.FileSystem.Path.ChangeExtension(art.Name, art.Extension);
                            fileName = await ctx.Pool.PutDocumentAsync(document.Id, fileName, content);
                            downloadedFiles.Add(fileName);
                            Log(LogSeverityEnum.Info,
                                $"Document '{document.Name}' [{document.Id}] file '{fileName}' downloaded. (size {content.Length} bytes)");
                        }
                    }

                    if (!downloadedFiles.All(x => ctx.FileSystem.File.Exists(x)))
                    {
                        // not all  downloaded -> rollback
                        ctx.Pool.DeleteDocumentFiles(Pool.PoolTypeEnum.Pending, document.Id);
                        await ctx.Connection.SetDocumentStateAsync(document.Id, DocumentStatusEnum.New);
                        Log(LogSeverityEnum.Warn, $"Document '{document.Name}' [{document.Id}] roll-backed.");
                    }
                }
                else
                {
                    // scenario for document doesn't exists => set document for pending (working->pending)
                    await ctx.Connection.SetDocumentStateAsync(document.Id, DocumentStatusEnum.Pending);
                }
            }
        }

        #endregion

        #region *** Private Operations ***
        #endregion
    }
}