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