using Quadarax.Application.QLiberace.Base.Services; using Quadarax.Application.QLiberace.Processor.Entities; using Quadarax.Application.QLiberace.Processor.Enums; using Quadarax.Application.QLiberace.Processor.Repositories; using Quadarax.Application.QLiberace.Processor.Settings; using Quadarax.Foundation.Core.Business; using Quadarax.Foundation.Core.Logging; using Quadarax.Foundation.Core.Thread; namespace Quadarax.Application.QLiberace.Processor.Workers { internal class ProcessingWorker : QlbrcProcessorWorker { #region *** Private fields *** // TODO: may be shift to ProcessorSetting private const int CN_ITEMS_PER_CYCLE = 1; #endregion #region *** Properties *** #endregion #region *** Constructor *** public ProcessingWorker(IWorkerContext context, SettingsService? srvSettings, IQLogger logger) : base(context, srvSettings, srvSettings?.GetSettings(Base.Constants.Modules.Processor.Code)?.Value?.ProcessingWorkerPwi!, logger) { } #endregion #region *** Private Operations *** protected override Task DoIteration(IWorkerContext context) { var ctx = (RepoWorkerContext)context; // Get worker name var workerName = GetName(Settings.ProcessingWorkerPwi); var repoProcessing = ctx.GetRepository(); // 1. Continue to work on already processing entries in ProcessQ (paused, tryout failed) // 2. If still available packet, lock pending entries in PendingQ // 3. Move locked entries to ProcessQ var processings = FetchFromQueue(ctx.GetRepository(), ctx.GetRepository()); foreach (var processing in processings) { processing.Started = DateTime.Now; processing.AffinityWorker = workerName; processing.Status = QueueStatusEnum.Working; repoProcessing.Set(processing); } // 4. Start to process // 5. Decrement attempt // 6. Move processed entries to DoneQ or FailedQ } private IEnumerable FetchFromQueue(RepoPendingQueue repoPending, RepoProcessingQueue repoProcessing) { if (repoPending == null) throw new ArgumentNullException(nameof(repoPending)); if (repoProcessing == null) throw new ArgumentNullException(nameof(repoProcessing)); // Get worker name var workerName = GetName(Settings.ProcessingWorkerPwi); // Get from pending queue var pendings = repoPending.GetPending(CN_ITEMS_PER_CYCLE).ToArray(); // Assign pending foreach (var pending in pendings) { // Set Assigned pending.Status = QueueStatusEnum.Assigned; pending.AffinityWorker = workerName; // Create processing entry from pending entry var processing = repoProcessing.NewFromPending(pending); // Delete pending (mark it) pending.Status = QueueStatusEnum.Deleted; repoPending.Set(pending); } // Try to load forgot items from previous start var processings = repoProcessing.GetAssignedOrWorking(workerName, CN_ITEMS_PER_CYCLE).ToArray(); return processings; } #endregion } }