PsQueue: implements PsQueue public operations, finish DoIteration.

qdr.fnd.core.pqueue/Exceptions/PSQueueValidationException.cs

@@ -26,7 +26,15 @@ namespace qdr.fnd.core.pqueue.Exceptions
             InitialAttemptMustBePositive = 7004,
             [ErrorMessage("Queue item validation: Status must be specified when isTransient flag is set to False..")]
             InvalidStatusTransientCombination = 7005,
-
+            [ErrorMessage("ForceStart: Cannot force start due forbidden status '{0}' for this operation.")]
+            InvalidStatusForForceStart = 7006,
+            [ErrorMessage("ForceStop: Cannot force stop due forbidden status '{0}' for this operation.")]
+            InvalidStatusForForceStop = 7007,
+            [ErrorMessage("ForceStop: Cannot pause due forbidden status '{0}' for this operation.")]
+            InvalidStatusForPause = 7008,
+            [ErrorMessage("StatusTransition: Invalid status transition from '{0}' to '{1}'. Allowed transition for status '{0}' are follows: {2}")]
+            InvalidStatusTransition = 7010,
+            
         }
         #endregion
         #region *** Constructors ***

qdr.fnd.core.pqueue/Extensions.cs → qdr.fnd.core.pqueue/Extensions/PsQueueItemDescriptorExt.cs

@@ -1,9 +1,9 @@
-using System.Text;
-using Quadarax.Foundation.Core.Value.Extensions;
+using Quadarax.Foundation.Core.Value.Extensions;
+using System.Text;
 
-namespace qdr.fnd.core.pqueue
+namespace qdr.fnd.core.pqueue.Extensions
 {
-    internal static class Extensions
+    public static class PsQueueItemDescriptorExt
     {
         public static string ToString(this IPsQueueItemDescriptor _)
         {

qdr.fnd.core.pqueue/Extensions/PsQueueItemExt.cs

@@ -0,0 +1,36 @@
+namespace qdr.fnd.core.pqueue.Extensions
+{
+    /// <summary>
+    /// <see cref="IPsQueueItem"/> extension methods."/>
+    /// </summary>
+    public static class PsQueueItemExt
+    {
+        /// <summary>
+        /// Check if <see cref="IPsQueueItem"/> is expired by <see cref="IPsQueueItemDescriptor.ValidTo"/> interval.
+        /// Ignored <see cref="IPsQueueItem.AttemptsLeft"/> and <see cref="IPsQueueItem.Status"/>.
+        /// </summary>
+        /// <param name="item">Current item.</param>
+        /// <returns>If true, item is expired by time span validity.</returns>
+        public static bool IsValiditySpanExpired(this IPsQueueItem item)
+        {
+            if (item.ValidTo == null)
+                return false;
+
+            return item.ValidTo < DateTime.Now;
+        }
+
+        /// <summary>
+        /// Check if <see cref="IPsQueueItem"/> is ready to be processed by <see cref="IPsQueueItemDescriptor.ValidFrom"/> and <see cref="IPsQueueItemDescriptor.ValidTo"/> interval.
+        /// Ignores AttemptsLeft and Status.
+        /// </summary>
+        /// <param name="item">Current item.</param>
+        /// <returns>If true, item can be processed by time span validity.</returns>
+        public static bool IsValiditySpanReady(this IPsQueueItem item)
+        {
+            if (item.ValidFrom <= DateTime.Now && !item.IsValiditySpanExpired())
+                return true;
+
+            return false;
+        }
+    }
+}

qdr.fnd.core.pqueue/IPsQueueItem.cs

@@ -32,7 +32,8 @@ namespace qdr.fnd.core.pqueue
         PsStatusEnum Status { get; }
 
         /// <summary>
-        /// Defines if <see cref="Status"/> is transient state. Changes are logged in <see cref="StatusJournal"/>.
+        /// Defines if <see cref="Status"/> is transient state (going to status change, but status is already changed). Changes are logged in <see cref="StatusJournal"/>.
+        /// NOTE: Transient is flag that set to True directly when status is changed. (means that first change Status, then transient)
         /// </summary>
         bool IsStatusTransient { get; }
         /// <summary>
@@ -46,6 +47,7 @@ namespace qdr.fnd.core.pqueue
         IEnumerable<IError> Log { get; }
         /// <summary>
         /// Collection of status changes of the queue item. When <see cref="Status"/> or <see cref="IsStatusTransient"/> is changed, the status change is logged.
+        /// NOTE: To this collection contains current state also.
         /// </summary>
         IEnumerable<PsStatusHistory> StatusJournal { get;}
         #endregion

qdr.fnd.core.pqueue/IPsQueueItemDescriptor.cs

@@ -13,7 +13,7 @@ namespace qdr.fnd.core.pqueue
         /// </summary>
         public string OwnerRef { get; }
         /// <summary>
-        /// Priority of the queue item. Lower number means higher priority.
+        /// Priority of the queue item. Lower number means higher priority (0 - system, 2 - immediate, 5+ - custom).
         /// </summary>
         int Priority { get; }
         

qdr.fnd.core.pqueue/PsQueue.cs

@@ -2,11 +2,13 @@
 using qdr.fnd.core.pqueue.Configuration;
 using qdr.fnd.core.pqueue.Enums;
 using qdr.fnd.core.pqueue.Exceptions;
+using qdr.fnd.core.pqueue.Extensions;
 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.Extensions;
 using Quadarax.Foundation.Core.Value.Generators;
 
 namespace qdr.fnd.core.pqueue
@@ -15,6 +17,9 @@ namespace qdr.fnd.core.pqueue
     {
         #region *** Constants ***
         private const string CS_DEF_QUEUE_NAME = "PSQUEUE";
+        public const int CN_PRIO_SYSTEM = 0;
+        public const int CN_PRIO_IMMEDIATE = 2;
+        public const int CN_PRIO_CUSTOM = 5;
         #endregion
 
         #region *** Properties ***
@@ -90,53 +95,131 @@ namespace qdr.fnd.core.pqueue
         #endregion
 
         #region *** Public Operations ***
-        
-        public string EnQueue(IPsQueueItemDescriptor item)
+       
+        public IQueryable<IPsQueueItem> Query(string? owner, string? ownerRef, params Tuple<bool,PsStatusEnum>[] status)
+        {
+            throw new NotImplementedException();
+        }
+
+        /// <inheritdoc/>
+        public string EnQueueItem(IPsQueueItemDescriptor item)
         {
-            if (item is null)
-                throw new ArgumentNullException(nameof(item));
             ValidateItem(item);
 
+            Log(LogSeverityEnum.Debug,$"{Name} - Item {item} begins to enqueue.");
             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}");
+                Log(LogSeverityEnum.Debug,$"{Name} - Attribute {argument.Name} was append to item [{itemId}]");
+             }
+            var newItem = GetItemInternal(itemId);
+            SetItemStatusInternal(newItem, PsStatusEnum.Pending, true, "Item was enqueued.");
+            Log(LogSeverityEnum.Info,$"{Name} - Item [{itemId}] was enqueued as {newItem}");
             return itemId;
         }
 
-        public IQueryable<IPsQueueItem> Query(string? owner, string? ownerRef, params Tuple<bool,PsStatusEnum>[] status)
+        /// <inheritdoc/>
+        public IPsQueueItem GetQueueItem(string itemId)
         {
-            throw new NotImplementedException();
+            ValidateItemId(itemId);
+
+            Log(LogSeverityEnum.Debug,$"{Name} - Item [{itemId}] begins to get.");
+            var item = GetItemInternal(itemId);
+            Log(LogSeverityEnum.Info,$"{Name} - Item [{itemId}] was fetched as {item}");
+            return item;
         }
 
-        public IPsQueueItem? Get(string itemId)
+        /// <inheritdoc/>
+        public void ForceStartQueueItem(string itemId, string emitter)
         {
-            throw new NotImplementedException();
+            ValidateItemId(itemId);
+            ValidateEmitter(emitter);
+
+            Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for force start by '{emitter}'.");
+            var item = _storage.GetItem(itemId);
+            EnsureItemStateExpired(item);
+
+            if (item.Status.In(PsStatusEnum.Started, PsStatusEnum.Stopped, PsStatusEnum.Expired))
+                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForForceStart, item.Status);
+
+            _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom));
+            _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE);
+            SetItemStatusInternal(item, PsStatusEnum.Pending, true, $"Item will reset due force start by emitter '{emitter}'.");
+            Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for force start by {emitter}.");
         }
 
-        public bool Postpone(string itemId, TimeSpan postpone)
+        /// <inheritdoc/>
+        public void ForceStopQueueItem(string itemId, string emitter)
         {
-            throw new NotImplementedException();
+            ValidateItemId(itemId);
+            ValidateEmitter(emitter);
+
+            Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for stop start by '{emitter}'.");
+            var item = _storage.GetItem(itemId);
+            EnsureItemStateExpired(item);
+
+            if (item.Status.In(PsStatusEnum.Expired, PsStatusEnum.DoneOk, PsStatusEnum.DoneFailed))
+                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForForceStop, item.Status);
+
+            _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom));
+            _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE);
+            SetItemStatusInternal(item, PsStatusEnum.Stopped, true, $"Item will reset due force stop by emitter '{emitter}'.");
+            Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for force stop by {emitter}.");
+
         }
 
-        public bool Reset(string itemId)
+        /// <inheritdoc/>
+        public void PauseQueueItem(string itemId, string emitter)
         {
-            throw new NotImplementedException();
+            ValidateItemId(itemId);
+            ValidateEmitter(emitter);
+
+            Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for pause by '{emitter}'.");
+            var item = _storage.GetItem(itemId);
+            EnsureItemStateExpired(item);
+
+            if (item.Status.In(PsStatusEnum.Expired, PsStatusEnum.DoneOk, PsStatusEnum.DoneFailed))
+                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForPause, item.Status);
+
+            _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom));
+            _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE);
+            SetItemStatusInternal(item, PsStatusEnum.Paused, true, $"Item will reset due pause by emitter '{emitter}'.");
+            Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for pause by {emitter}.");
+
         }
 
-        public bool SetPriority(string itemId, int priority)
+        public void StartQueueItem(string itemId, string emmiter, bool isImmediate)
         {
             throw new NotImplementedException();
         }
 
-        public IEnumerable<IPsQueueItem> GetWorkingItems()
+        /// <inheritdoc/>
+        public IQueryable<IPsQueueItem> QueryQueueItems(IPaging paging, string? owner, string? ownerRef, bool includeTransient = false, params PsStatusEnum[] statuses)
         {
-            throw new NotImplementedException();
+            if (paging == null)
+                throw new ArgumentNullException(nameof(paging));
+            var statusesForQuery = new List<Tuple<bool, PsStatusEnum>>();
+            foreach (var status in statuses)
+            {
+                statusesForQuery.Add(new Tuple<bool, PsStatusEnum>(false, status));
+                if (includeTransient)
+                    statusesForQuery.Add(new Tuple<bool, PsStatusEnum>(true, status));
+            }
+            return _storage.Query(owner, ownerRef, statusesForQuery.ToArray());
+
+        }
+
+        public void Start()
+        {
+            base.Start();
+        }
+
+        public void Stop()
+        {
+            base.Stop();
         }
         #endregion
 
@@ -148,26 +231,93 @@ namespace qdr.fnd.core.pqueue
         /// <inheritdoc />
         protected override Task DoIteration(IWorkerContext? context)
         {
+            // fetch items to immediate stop - frees workers
+            var itemsToStop = Query(null, null,
+                    new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Stopped),new(true, PsStatusEnum.Paused)  })
+                .OrderBy(x=>x.Priority).ThenBy(x=>x.ValidFrom);
+
+
+            foreach(var item in itemsToStop){
+                
+                var process = _workingItems.TryGetValue(item.Id, out var processProvider) ? processProvider : null;
+                if (process == null)
+                    Log(LogSeverityEnum.Warn, $"{Name} - Cannot find process provider in working cache for item [{item.Id}]. Skiped.");
+                if (process != null)
+                { 
+                    process.Stop();
+                    SetItemStatusInternal(item, item.Status, false, "Item was stopped by schedule.");
+                    _workingItems.TryRemove(item.Id, out _);
+                }
+            }
+
+            // fetch working, stopped, paused items which currently expired - frees workers
+            var itemsToExpire = Query(null, null,
+                    new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Stopped),new(true, PsStatusEnum.Paused), new(true, PsStatusEnum.Started)  })
+                .Where(x=>x.IsValiditySpanExpired());
+            foreach(var item in itemsToExpire)
+            {
+                EnsureItemStateExpired(item);
+            }
+
+            // fetch items to immediate start
             var workersLeft = Configuration.MaxConcurrentProcesses - _workingItems.Count;
             if (workersLeft>0)
             {
+                // fetch 
                 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);
+                        new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Started) })
+                    .Where(x => !x.IsValiditySpanExpired()).Take(workersLeft);
 
+                foreach(var item in itemsToProcess)
+                {
+                    if (item.AttemptsLeft <= 0)
+                    {
+                        SetItemStatusInternal(item, PsStatusEnum.DoneFailed, true, "Item will be marked as failed due no attempts left.");
+                        SetItemStatusInternal(item, PsStatusEnum.DoneFailed, false, "Item will be marked as failed due no attempts left.");
+                        continue;
+                    }
+                    _storage.UpdateItemAttempts(item.Id, item.AttemptsLeft - 1);
+                    var provider = ResolveProvider(item.ProcessProviderClass, item.Arguments);
+                    _storage.UpdateItemStarted(item.Id, DateTime.Now);
+                    _storage.UpdateItemFinished(item.Id, null);
+                    _workingItems.TryAdd(item.Id, provider);
+                    provider.Start();
+                    SetItemStatusInternal(item, PsStatusEnum.Started, false, "Item was started by schedule.");
                 }
             }
 
+            // fetch items to close expired
+            var itemsToClose = Query(null, null,
+                    new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Expired) });
+            foreach (var item in itemsToClose)
+            {
+                SetItemStatusInternal(item, PsStatusEnum.Expired, false, "Item will be expired by schedule.");
+            }
+
+            // fetch pending items to initiate to expired
+            var itemsPendingToExpire = Query(null, null,
+                    new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Pending), new(false, PsStatusEnum.Pending) });
+            foreach(var item in itemsPendingToExpire)
+            {
+                SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by schedule.");
+            }
+
+
             return base.DoIteration(context);
         }
 
         #endregion
 
-
+        private void ValidateItemId(string itemId)
+        {
+            if (string.IsNullOrEmpty(itemId))
+                throw new ArgumentNullException(nameof(itemId));
+        }
+        private void ValidateEmitter(string emitter)
+        {
+            if (string.IsNullOrEmpty(emitter))
+                throw new ArgumentNullException(nameof(emitter));
+        }
         private void ValidateItem(IPsQueueItemDescriptor item)
         {
             if (item is null)
@@ -182,19 +332,30 @@ namespace qdr.fnd.core.pqueue
             var provider =  ResolveProvider(item.ProcessProviderClass, item.Arguments);
             provider.Validate();
 
-            if (item.AtteptsInitial <= 0);
+            if (item.AtteptsInitial <= 0)
                 throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InitialAttemptMustBePositive);
                 
         }
 
-        private void SetItemStatus(IPsQueueItem item, PsStatusEnum? newStatus, bool isTransient, string message)
+        private void SetItemStatusInternal(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);
+           
+            if (newStatus != null)
+            {
+                if (!_statusTransitions.First(x => x.Item1 == oldStatus).Item2.Contains(newStatus.Value))
+                    throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusTransition, oldStatus,newStatus ?? oldStatus, string.Join(",", _statusTransitions.First(x => x.Item1 == oldStatus).Item2.Select(x=>x.ToString())));
+            }
 
+            _storage.AppendStatusJournal(item.Id, newStatus ?? oldStatus, isTransient, message);
+            _storage.UpdateItemStatus(item.Id, newStatus, isTransient);
+            ((PsQueueItem)item).SetStatus(newStatus, isTransient, message);
+            ProcessItemStatusChanged?.Invoke(this, new PsQueueStatusChangedEventArgs(item.Id) { CurrentStatus = newStatus ?? oldStatus, CurrentIsStatusTransient = isTransient, PreviousStatus = oldStatus, PreviousIsStatusTransient = oldIsTransient });
+            Log(LogSeverityEnum.Debug, $"{Name} - Item [{item.Id}] changed status '{oldStatus}[{oldIsTransient}]' to '{(newStatus ?? oldStatus)}[{isTransient}]'.");
         }
 
         private IPsProcessProvider ResolveProvider(string providerClass, IEnumerable<ITypedArgument> arguments)
@@ -213,59 +374,54 @@ namespace qdr.fnd.core.pqueue
             return provider;
         }
 
-        protected override void OnDisposing()
+        private void EnsureItemStateExpired(IPsQueueItem item)
         {
-            base.OnDisposing();
-
-            ProcessItemStatusChanged = null;
-            _storage.Dispose();
-            //TODO: Implement OnDisposing
-        }
-
-        public void Start()
-        {
-            throw new NotImplementedException();
-        }
-
-        public void Stop()
-        {
-            throw new NotImplementedException();
-        }
-
-        public string EnQueueItem(IPsQueueItemDescriptor item)
-        {
-            throw new NotImplementedException();
-        }
+            if (item==null) 
+                throw new ArgumentNullException(nameof(item));
 
-        public IPsQueueItem GetQueueItem(string itemId)
-        {
-            throw new NotImplementedException();
-        }
+            if (item.IsValiditySpanExpired() && item.Status != PsStatusEnum.Expired)
+            {
+                if (item.Status == PsStatusEnum.Started)
+                {
+                    if (!_workingItems.TryGetValue(item.Id, out var process))
+                        throw new InvalidOperationException($"Cannot find process provider in working cache for item [{item.Id}]");
+                    
+                    SetItemStatusInternal(item, PsStatusEnum.Stopped, true, "Item will be stopped because expired.");
+                    SetItemStatusInternal(item, PsStatusEnum.Stopped, false, "Item was stopped by system because expired.");
+                    // NOTE: fast switch to stopped due prevents main loop to handle - stops later
+                    _workingItems.TryRemove(item.Id, out _);
+                    process.Stop();
+                    SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by process.");
+                }
+                else
+                {
+                    SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by process.");
+                }
 
-        public void ForceStartQueueItem(string itemId, string emitter)
-        {
-            throw new NotImplementedException();
+                item = GetItemInternal(item.Id);
+            }
         }
-
-        public void ForceStopQueueItem(string itemId, string emitter)
+        private IPsQueueItem GetItemInternal(string itemId)
         {
-            throw new NotImplementedException();
-        }
+            ValidateItemId(itemId);
 
-        public void PauseQueueItem(string itemId, string emmiter)
-        {
-            throw new NotImplementedException();
+            var item = _storage.GetItem(itemId);
+            return item;
         }
 
-        public void StartQueueItem(string itemId, string emmiter, bool isImmediate)
+        protected override void OnDisposing()
         {
-            throw new NotImplementedException();
-        }
+            base.OnDisposing();
 
-        public IQueryable<IPsQueueItem> QueryQueueItems(IPaging paging, string? owner, string? ownerRef, bool includeTransient = false, params PsStatusEnum[] statuses)
-        {
-            throw new NotImplementedException();
+            ProcessItemStatusChanged = null;
+            _storage.Dispose();
+            //TODO: Implement OnDisposing
+            foreach (var item in _workingItems.Values)
+                item.Stop();
+            _workingItems.Clear();
+            _workingItems = null!;
         }
+        
         #endregion
     }
 }

qdr.fnd.core.pqueue/PsQueueItem.cs

@@ -80,5 +80,16 @@ namespace qdr.fnd.core.pqueue
         }
         #endregion
 
+        #region *** Public operations ***
+        public void SetStatus(PsStatusEnum? status, bool isTransient, string message)
+        {
+            Status = status ?? Status;
+            IsStatusTransient = isTransient;
+            ((List<PsStatusHistory>)StatusJournal).Add(
+                new PsStatusHistory(Status, IsStatusTransient, message)
+            );
+        }
+        #endregion
+
     }
 }