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> _statusTransitions = new List>(); private ConcurrentDictionary _workingItems = new ConcurrentDictionary(); #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.New,new[] { PsStatusEnum.Pending })); _statusTransitions.Add(new Tuple(PsStatusEnum.Pending,new[] { PsStatusEnum.Started, PsStatusEnum.Expired })); _statusTransitions.Add(new Tuple(PsStatusEnum.Started,new[] { PsStatusEnum.Pending, PsStatusEnum.Paused, PsStatusEnum.Stopped, PsStatusEnum.DoneOk, PsStatusEnum.DoneFailed, PsStatusEnum.Expired })); _statusTransitions.Add(new Tuple(PsStatusEnum.Paused,new[] { PsStatusEnum.Started, PsStatusEnum.Stopped, PsStatusEnum.Expired })); _statusTransitions.Add(new Tuple(PsStatusEnum.Stopped,new[] { PsStatusEnum.DoneFailed, PsStatusEnum.Expired })); _statusTransitions.Add(new Tuple(PsStatusEnum.DoneOk,new[] { PsStatusEnum.Deleted })); _statusTransitions.Add(new Tuple(PsStatusEnum.DoneFailed,new[] { PsStatusEnum.Deleted })); _statusTransitions.Add(new Tuple(PsStatusEnum.Expired,new[] { PsStatusEnum.Deleted })); _statusTransitions.Add(new Tuple(PsStatusEnum.Deleted,Array.Empty())); } #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 Query(string? owner, string? ownerRef, params Tuple[] 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 GetWorkingItems() { throw new NotImplementedException(); } #endregion #region *** Private Operations *** #region Overrides of LoopWorker /// protected override Task DoIteration(IWorkerContext? context) { var workersLeft = Configuration.MaxConcurrentProcesses - _workingItems.Count; if (workersLeft>0) { var itemsToProcess = Query(null, null, new Tuple[] { 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 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 } }