qliberace/Modules/qdr.app.qlbrc.processor/Workers/ProcessingWorker.cs

using Quadarax.Application.QLiberace.Base.Services;
using Quadarax.Application.QLiberace.Common.Executor;
using Quadarax.Application.QLiberace.Processor.Entities;
using Quadarax.Application.QLiberace.Processor.Enums;
using Quadarax.Application.QLiberace.Processor.Executor;
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<ProcessorSetting>(Base.Constants.Modules.Processor.Code)?.Value?.ProcessingWorkerPwi!, logger)
        {
        }
        #endregion


        #region *** Private Operations ***

        protected override async Task DoIteration(IWorkerContext context)
        {
            var ctx = (RepoWorkerContext)context;
                // Get worker name
                var workerName = GetName(Settings.ProcessingWorkerPwi);

                var repoProcessing = ctx.GetRepository<RepoProcessingQueue>();
                var repoData = ctx.GetRepository<RepoQueueData>();
                var repoDone = ctx.GetRepository<RepoDoneQueue>();

                // 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 processing = FetchFromQueue(ctx.GetRepository<RepoPendingQueue>(), ctx.GetRepository<RepoProcessingQueue>());

                foreach (var processItem in processing)
                {
                    processItem.Started = DateTime.Now;
                    processItem.AffinityWorker = workerName;
                    processItem.AttemptLeft -= 1;
                    processItem.Status = QueueStatusEnum.Working;
                    repoProcessing.Set(processItem);

                    var dataList = repoData.GetQueueData(processItem.Id).ToArray();
                    var paramList = dataList.Select(x => (ParameterExecutorValue)x.ToParametrizedValue()).ToArray();

                    var exe = new ExecutorInternal(processItem, paramList);
                    var result = exe.Execute(Timeout, _logger);
                    if (result.IsSuccess)
                    {
                        //success
                        //update modified output values (from Execute)
                        foreach (var parOutput in paramList.Where(x => x.IsOutput))
                        {
                            repoData.UpdateQueueData(processItem.Id, parOutput.Code, parOutput.Value);
                        }

                        var done = repoDone.NewFromProcessing(processItem);
                        done.Status = QueueStatusEnum.Success;
                        processItem.Status = QueueStatusEnum.Deleted;
                    }
                    else
                    {
                        //failed
                        if (processItem.AttemptLeft == 0)
                        {
                            //move to failed
                        }
                        else
                        {
                            // next tryout
                            processItem.Status = QueueStatusEnum.Assigned;
                        }
                    }
                }
        }


        private IEnumerable<ProcessingQueue> 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

    }
}