qdr.fnd/qdr.fnd.core.pqueue/PsQueue.cs

using System.Collections.Concurrent;
using qdr.fnd.core.pqueue.Configuration;
using qdr.fnd.core.pqueue.Enums;
using qdr.fnd.core.pqueue.Exceptions;
using qdr.fnd.core.pqueue.Process;
using qdr.fnd.core.pqueue.Storages;
using Quadarax.Foundation.Core.Data;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.Thread;
using Quadarax.Foundation.Core.Value.Generators;

namespace qdr.fnd.core.pqueue
{
    public class PsQueue : LoopWorker
    {
        #region *** Constants ***
        private const string CS_DEF_QUEUE_NAME = "PSQUEUE";
        #endregion

        #region *** Properties ***
        public PsQueueConfiguration Configuration { get;}

        public event PsQueueStatusChangedEventHandler? ProcessItemStatusChanged;
        #endregion

        #region *** Private Fields ***
        private TinyHash _hashGenerator;
        protected IPsQueueStorage _storage;
        private List<Tuple<PsStatusEnum,PsStatusEnum[]>> _statusTransitions = new List<Tuple<PsStatusEnum, PsStatusEnum[]>>();

        private ConcurrentDictionary<string, IPsProcessProvider> _workingItems = new ConcurrentDictionary<string, IPsProcessProvider>();
        #endregion

        #region *** Constructors ***

        public PsQueue(string queueName, PsQueueConfiguration configuration,IPsQueueStorage storageProvider, ILogger logger) : base(queueName,null)
        {
            _hashGenerator = new TinyHash(5);
            if (string.IsNullOrEmpty(queueName))
                queueName =CS_DEF_QUEUE_NAME + _hashGenerator.NewId();

            Configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
            _storage = storageProvider ?? throw new ArgumentNullException(nameof(storageProvider));

            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.New,new[]
                {
                    PsStatusEnum.Pending
                }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Pending,new[]
            {
                PsStatusEnum.Started,
                PsStatusEnum.Expired
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Started,new[]
            {
                PsStatusEnum.Pending,
                PsStatusEnum.Paused,
                PsStatusEnum.Stopped,
                PsStatusEnum.DoneOk,
                PsStatusEnum.DoneFailed,
                PsStatusEnum.Expired
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Paused,new[]
            {
                PsStatusEnum.Started,
                PsStatusEnum.Stopped,
                PsStatusEnum.Expired
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Stopped,new[]
            {
                PsStatusEnum.DoneFailed,
                PsStatusEnum.Expired
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.DoneOk,new[]
            {
                PsStatusEnum.Deleted
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.DoneFailed,new[]
            {
                PsStatusEnum.Deleted
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Expired,new[]
            {
                PsStatusEnum.Deleted
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Deleted,Array.Empty<PsStatusEnum>()));

        }

        #endregion

        #region *** Public Operations ***
        
        public string EnQueue(IPsQueueItemDescriptor item)
        {
            if (item is null)
                throw new ArgumentNullException(nameof(item));
            ValidateItem(item);

            var itemId = _storage.CreateItem(item.Owner, item.OwnerRef, item.ProcessProviderClass, item.AtteptsInitial, null);
            foreach (var argument in item.Arguments)
            {
                if (argument.ValueType.FullName == null) throw new InvalidOperationException($"Cannot obtain fullName of type '{argument.ValueType.Name}'");
                
                _storage.CreateItemAttribute(itemId, argument.Name, argument.Value, argument.ValueType.FullName, argument.IsOutput);
            }

            Log(LogSeverityEnum.Info,$"{Name} - Item [{itemId}] was enqueued as {item}");
            return itemId;
        }
        public bool Dequeue(string itemId, bool force)
        {
            throw new NotImplementedException();
        }

        public IQueryable<IPsQueueItem> Query(string? owner, string? ownerRef, params Tuple<bool,PsStatusEnum>[] status)
        {
            throw new NotImplementedException();
        }

        public IPsQueueItem? Get(string itemId)
        {
            throw new NotImplementedException();
        }

        public bool Postpone(string itemId, TimeSpan postpone)
        {
            throw new NotImplementedException();
        }

        public bool Reset(string itemId)
        {
            throw new NotImplementedException();
        }

        public bool SetPriority(string itemId, int priority)
        {
            throw new NotImplementedException();
        }

        public IEnumerable<IPsQueueItem> GetWorkingItems()
        {
            throw new NotImplementedException();
        }
        #endregion


        #region *** Private Operations ***

        #region Overrides of LoopWorker

        /// <inheritdoc />
        protected override Task DoIteration(IWorkerContext? context)
        {
            var workersLeft = Configuration.MaxConcurrentProcesses - _workingItems.Count;
            if (workersLeft>0)
            {
                var itemsToProcess = Query(null, null,
                        new Tuple<bool, PsStatusEnum>[] { new(false, PsStatusEnum.Pending) })
                    .Where(x => x.ValidFrom > DateTime.Now && (x.ValidTo == null || x.ValidTo < DateTime.Now)).Take(workersLeft);

                foreach(var item in itemsToProcess){    
                    // set to Pending Transient to Started
                    _storage.UpdateItemStatus(item.Id, null, true);

                }
            }

            return base.DoIteration(context);
        }

        #endregion


        private void ValidateItem(IPsQueueItemDescriptor item)
        {
            if (item is null)
                throw new ArgumentNullException(nameof(item));

            if (string.IsNullOrEmpty(item.Owner))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.OwnerRequired);

            if (string.IsNullOrEmpty(item.OwnerRef))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.OwnerRefRequired);

            var provider =  ResolveProvider(item.ProcessProviderClass, item.Arguments);
            provider.Validate();

            if (item.AtteptsInitial <= 0);
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InitialAttemptMustBePositive);
                
        }

        private void SetItemStatus(IPsQueueItem item, PsStatusEnum? newStatus, bool isTransient, string message)
        {
            var oldStatus = item.Status;
            var oldIsTransient = item.IsStatusTransient;
            
            if (!isTransient && newStatus == null)
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusTransientCombination);

        }

        private IPsProcessProvider ResolveProvider(string providerClass, IEnumerable<ITypedArgument> arguments)
        {
            if (string.IsNullOrEmpty(providerClass))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderRequired);

            var providerType = Type.GetType(providerClass);
            if (providerType == null)
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderCannotResolve, providerClass);

            var provider = Activator.CreateInstance(providerType, arguments) as IPsProcessProvider;
            if (provider == null)
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderCannotResolve, providerClass);

            return provider;
        }

        protected override void OnDisposing()
        {
            base.OnDisposing();

            ProcessItemStatusChanged = null;
            _storage.Dispose();
            //TODO: Implement OnDisposing
        }
        #endregion
    }
}