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 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, Statistics statistics,Pool pool, ILogger logger) : base(CS_WK_NAME, new PullCtx(connection, fileSystem, 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 documents = await ctx.Connection.GetDocumentsAsync(DocumentsScopeEnums.Pending);
            foreach (var document in documents)
            {
                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.");
                }
            }
        }

        #endregion

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