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

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<ProcessorSetting>(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<RepoProcessingQueue>();

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

            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<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

    }
}