bo/Workbench/Processor/Queue/ProcessQueue.cs

using Quadarax.Foundation.Core.Business.Processor.Enums;
using Quadarax.Foundation.Core.Business.Processor.Providers;
using Quadarax.Foundation.Core.Business.Processor.Task;
using Quadarax.Foundation.Core.Data;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.Object;
using Quadarax.Foundation.Core.Value.Extensions;

namespace Quadarax.Foundation.Core.Business.Processor.Queue
{
    public class ProcessQueue : DisposableObject
    {

        #region *** Properties ***
        public bool IsRunning { get; private set; }
        public string QueueIdentifier { get; private set; }
        internal IProcessorStorageProvider Provider => _storageProvider;
        #endregion

        #region *** Private fields ***
        private readonly CallbackDelegates _delegates;
        private readonly ILog _log;
        private readonly ProcessQueueConfiguration _configuration;
        private readonly IProcessorStorageProvider _storageProvider;
        private bool _isTransient;

        private Timer _timerWatchDog;
        private bool _isFirstCycle;

        private TaskSlots _taskSlots;
        #endregion

        #region *** Constructors ***
        public ProcessQueue(string processQueueIdentifierToken, IProcessorStorageProvider storage, ProcessQueueConfiguration configuration, CallbackDelegates? delegates = null)
        {
            if (string.IsNullOrEmpty(processQueueIdentifierToken)) throw new ArgumentNullException(nameof(processQueueIdentifierToken));
            QueueIdentifier = processQueueIdentifierToken;

            _taskSlots = new TaskSlots(configuration.WorkersCount);
            _delegates = delegates ?? new CallbackDelegates();

            _log = new ConsoleLogger().GetLogger(GetType());
            _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
            _storageProvider = storage ?? throw new ArgumentNullException(nameof(storage));

            _configuration.Validate();
        }


        #endregion

        #region *** Public Operations ***
        public void Start()
        {
            if (IsRunning || _isTransient)
            {
                Log(LogSeverityEnum.Warn, $"ProcessQueue [{QueueIdentifier}] already running! Skip start.");
                return;
            }
            _isTransient = true;
            var now = DateTime.Now;
            Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] starting...");

            //TODO: start mechanism here
            _isFirstCycle = true;
            _timerWatchDog = new 
                Timer(OnWatchdog, this, (int) _configuration.WatchdogInterval.TotalMilliseconds, (int) _configuration.WatchdogInterval.TotalMilliseconds);


            Log(LogSeverityEnum.Info, $"ProcessQueue [{QueueIdentifier}] started @{DateTime.Now.Subtract(now).ToReadableString()}");
            _isTransient = false;
            IsRunning = true;
        }

        
        public void Stop()
        {
            if (!IsRunning || _isTransient)
            {
                Log(LogSeverityEnum.Warn, $"ProcessQueue [{QueueIdentifier}] already stopped! Skip stop.");
                return;
            }
            _isTransient = true;
            var now = DateTime.Now;
            Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] stopping...");

            //TODO: start mechanism here

            Log(LogSeverityEnum.Info, $"ProcessQueue [{QueueIdentifier}] stopped @{DateTime.Now.Subtract(now).ToReadableString()}");
            _isTransient = false;
            IsRunning = false;
        }

        public ITaskIdentifier CreateTask(string reference, string providerClass, ITypedArgument[] initialArguments, byte priority, int initialAttempts, TimeSpan attemptTimeout, DateTime validFrom, DateTime? validTo)
        {
            try
            {
                var task = new TaskItem(this,reference, providerClass,initialArguments,priority,initialAttempts,attemptTimeout,validFrom,validTo, null);
                var newId = _storageProvider.Create(task);
                task.Id = newId.Id;
                return task;
            }
            catch (Exception e)
            {
                var message = $"Error creating task [Ref:{reference}, Provider:{providerClass}] for queue [{QueueIdentifier}]";
                Log(LogSeverityEnum.Error, message, e);
                throw new Exception(message, e);
            }
        }

        public ITaskItem GetTask(ITaskIdentifier id)
        {
            if (id == null) throw new ArgumentNullException(nameof(id));
            var result = _storageProvider.Get(id).FirstOrDefault();
            return result == null ? throw new Exception($"Task [{id.Id}] not found!") : (ITaskItem)result;
        }

        public ITaskItem[] GetTask(ITaskIdentifier[] ids)
        {
            if (ids == null) throw new ArgumentNullException(nameof(ids));

            if (!ids.Any()) return Array.Empty<ITaskItem>();

            var result = _storageProvider.Get(ids);
            var diff = result.Select(x => x.Id).CompareDiff(ids.Select(x => x.Id));

            if (diff.Any())
                throw new Exception($"Tasks [{string.Join(",", diff)}] not found!");

            return result.Select(x=>x as ITaskItem).ToArray()!;
        }

        public ITaskIdentifier[] QueryTasks(IPaging paging,ProcessStatusEnum[] states,int maxItems, bool affinityQueue = true, bool onlyTimeValid = true, bool includeDeleted = false)
        {
            if (states == null || !states.Any()) throw new ArgumentNullException(nameof(states));

            var now = DateTime.Now;
            if (!states.Contains(ProcessStatusEnum.Deleted) && includeDeleted)
                states = states.Concat(new[] { ProcessStatusEnum.Deleted }).ToArray();

            var items = _storageProvider.Query(paging, states);
            if (onlyTimeValid)
                items = items.Where(x=> x.ProcessingValidFrom <= now && x.ProcessingValidTo.GetValueOrDefault(DateTime.MaxValue) >= now);
            
            items = affinityQueue ? items.Where(x => x.AffinityToken != null && x.AffinityToken.QueueIdentifier == QueueIdentifier) 
                : items.Where(x => x.AffinityToken == null);
            
            var result = items.OrderBy(x => x.ProcessingValidFrom)
                .ThenBy(x => x.Priority)
                .Take(maxItems)
                .Select(x=>(ITaskIdentifier)x)
                .ToArray();

            Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] Query tasks {result.Length} of {maxItems}@{DateTime.Now.Subtract(now).ToReadableString()}");

            return result;
        }

        #endregion

        #region *** Private Operations ***

        private void OnWatchdog(object? state)
        {
            try
            {
                if (_isFirstCycle)
                {
                    // first cycle scenario
                    // reset previous left tasks
                    var forceLeftTaskIds = QueryTasks(Paging.NoPaging, new[] { ProcessStatusEnum.Started}, _configuration.WorkersCount, true,true, false);
                    var forceLeftTasks = GetTask(forceLeftTaskIds);
                    Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] Watchdog first cycle. Force reset on {forceLeftTaskIds.Count()} tasks.");
                    foreach (var task in forceLeftTasks)
                    {
                        task.Stop("(System) Force stop of interrupted task.", true, false);
                        task.Reset("(System) Reset after force stop.", TimeSpan.FromSeconds(10));
                    }
                    _isFirstCycle = false;
                }

                // gets processing tasks to know thread assigned slots
                var assigningTaskIds = QueryTasks(Paging.NoPaging, new[] { ProcessStatusEnum.Pending }, _taskSlots.FreeSlotsCount, false, true, false);
                var assigningTasks = GetTask(assigningTaskIds);
                var startedCnt = 0;
                foreach (var assign in assigningTasks)
                {
                    var ordinal = _taskSlots.AssignSlot(assign);
                    if (ordinal >= 0)
                    {
                        ((TaskItem)assign).Start(new AffinityToken(QueueIdentifier, ordinal));
                        startedCnt++;
                    }
                }
                Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] Watchdog cycle. Assigned and started {startedCnt} of {assigningTaskIds.Length} tasks.");
            }
            catch (Exception e)
            {
                Log(LogSeverityEnum.Error, e.Message, e);
            }
        }



        protected void Log(LogSeverityEnum type, string message, Exception? e = null)
        {
            if (type == LogSeverityEnum.Error || type == LogSeverityEnum.Fatal)
            {
                _delegates.OnProcessorError?.Invoke(message, e);
            }
            _log.Log(type, message,e);
        }

        protected override void OnDisposing()
        {
            if (IsRunning)
                Stop();
            _delegates?.Reset();
            Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] disposed.");
        }
        #endregion
    }
}