some pqueue changes

qdr.fnd.core.pqueue.test/Storages/MemoryStorageProviderTest.cs

@@ -2,8 +2,8 @@
 using qdr.fnd.core.pqueue.Exceptions;
 using qdr.fnd.core.pqueue.Storages;
 using qdr.fnd.core.pqueue.test.Mocks;
-using qdr.fnd.core.pqueue.test.Storages.Mocks;
 using Quadarax.Foundation.Core.Data;
+using Quadarax.Foundation.Core.Thread;
 using Quadarax.Foundation.Core.Value.Extensions;
 using Quadarax.Foundation.Core.Value.Generators;
 

qdr.fnd.core.pqueue/Enums/PsStatusEnum.cs

@@ -2,13 +2,51 @@
 {
     public enum PsStatusEnum : int
     {
+        /// <summary>
+        /// Initial status of the queue item.
+        /// This state can be moved to states: <see cref="Pending"/>
+        /// </summary>
         New = 0,
+        /// <summary>
+        /// Enqueued item waiting for processing.
+        /// This state can be moved to states: <see cref="Started"/>, <see cref="Expired"/>
+        /// </summary>
         Pending = 1,
+        /// <summary>
+        /// Item started to processing.
+        /// This state can be moved to states: <see cref="Pending"/>, <see cref="Paused"/>, <see cref="Stopped"/>
+        /// <see cref="DoneOk"/>, <see cref="DoneFailed"/>, <see cref="Expired"/>
+        /// </summary>
         Started = 2,
+        /// <summary>
+        /// Item is paused in processing and can be resumed with postponed time.
+        /// This state can be moved to states: <see cref="Started"/>, <see cref="Stopped"/>, <see cref="Expired"/>
+        /// </summary>
         Paused = 3,
+        /// <summary>
+        /// Item is stopped in processing and cannot be resumed.
+        /// This state can be moved to states: <see cref="DoneFailed"/>, <see cref="Expired"/>
+        /// </summary>
         Stopped = 4,
+        /// <summary>
+        /// Item is successfully processed. Final logical state.
+        /// This state can be moved to states: <see cref="Deleted"/>
+        /// </summary>
         DoneOk = 5,
+        /// <summary>
+        /// Item is failed in processing. Final logical state.
+        /// This state can be moved to states: <see cref="Deleted"/>
+        /// </summary>
         DoneFailed = 6,
-        Deleted = 7
+        /// <summary>
+        /// Item is expired and ready to delete.
+        /// This state can be moved to states: <see cref="Deleted"/>
+        /// </summary>
+        Expired = 7,
+        /// <summary>
+        /// Item is ready to physical delete.
+        /// This state technical final state.
+        /// </summary>
+        Deleted = 8
     }
 }

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

@@ -0,0 +1,54 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Quadarax.Foundation.Core.Attributes;
+using Quadarax.Foundation.Core.Exceptions;
+
+namespace qdr.fnd.core.pqueue.Exceptions
+{
+    public class PSQueueValidationException: CodeException<PSQueueValidationException.ErrorCodes>
+    {
+        #region *** Error Codes ***
+
+        public enum ErrorCodes : int
+        {
+            [ErrorMessage("Queue item validation: Owner value is required.")]
+            OwnerRequired = 7000,
+            [ErrorMessage("Queue item validation: OwnerRef value is required.")]
+            OwnerRefRequired = 7001,
+            [ErrorMessage("Queue item validation: ProcessProvider value is required.")]
+            ProcessProviderRequired = 7002,
+            [ErrorMessage("Queue item validation: ProcessProvider '{0}' type cannot be resolved.")]
+            ProcessProviderCannotResolve = 7003,
+            [ErrorMessage("Queue item validation: InitialAttempt must be bigger than zero.")]
+            InitialAttemptMustBePositive = 7004,
+            [ErrorMessage("Queue item validation: Status must be specified when isTransient flag is set to False..")]
+            InvalidStatusTransientCombination = 7005,
+
+        }
+        #endregion
+        #region *** Constructors ***
+        /// <inheritdoc />
+        public PSQueueValidationException(ErrorCodes code, Exception innerException) : base(code, innerException)
+        {
+        }
+
+        /// <inheritdoc />
+        public PSQueueValidationException(ErrorCodes code) : base(code)
+        {
+        }
+
+        /// <inheritdoc />
+        public PSQueueValidationException(ErrorCodes code, params object[] messageArguments) : base(code, messageArguments)
+        {
+        }
+
+        /// <inheritdoc />
+        public PSQueueValidationException(ErrorCodes code, Exception innerException, params object[] messageArguments) : base(code, innerException, messageArguments)
+        {
+        }
+        #endregion
+    }
+}

qdr.fnd.core.pqueue/Extensions.cs

@@ -0,0 +1,20 @@
+using System.Text;
+using Quadarax.Foundation.Core.Value.Extensions;
+
+namespace qdr.fnd.core.pqueue
+{
+    internal static class Extensions
+    {
+        public static string ToString(this IPsQueueItemDescriptor _)
+        {
+            var sb = new StringBuilder();
+            sb.Append("Own: ").Append(_.Owner).Append("; ");
+            sb.Append("OwnRef: ").Append(_.OwnerRef).Append("; ");
+            sb.Append("Prv: ").Append(_.ProcessProviderClass).Append("; ");
+            sb.Append("PrvArgs: ").Append(string.Join(",",_.Arguments.Select(x=>x.Name))).Append("; ");
+            sb.Append("From: ").Append(_.ValidFrom.ToFileUtcString()).Append("; ");
+            sb.Append("To: ").Append((_.ValidTo == null ? "NULL" : _.ValidTo.Value.ToFileUtcString())).Append("; ");
+            return sb.ToString();
+        }
+    }
+}

qdr.fnd.core.pqueue/MoshPit.cs

@@ -1,65 +0,0 @@
-namespace qdr.fnd.core.pqueue
-{
-
-    public class MoshPit<TContext> where TContext : class
-    {
-        private readonly List<Action<TContext>> _actions;
-        private Timer? _timer;
-        private readonly Random _random;
-
-
-        public TContext Context { get; set; }
-
-        public long Iterations { get; private set; }
-        public TimeSpan AvgIterationDuration { get; private set; }
-        public TimeSpan TotalDuration { get; private set; }
-        public decimal AvgIterationPerSecond { get; private set; }
-        public MoshPit(TContext context)
-        {
-            Context = context;
-            _actions = new List<Action<TContext>>();
-            _random = new Random();
-        }
-
-        public void AddAction(Action<TContext> action)
-        {
-            _actions.Add(action);
-        }
-
-        public void Start(int iterationInterval, int totalDuration)
-        {
-            _timer = new Timer(ExecuteRandomAction, Context, 0, iterationInterval);
-
-            // Stop the _timer after the specified duration
-            _ = new Timer((e) => _timer.Dispose(), null, totalDuration, Timeout.Infinite);
-
-            Thread.Sleep(totalDuration);
-        }
-
-        private void ExecuteRandomAction(object? state)
-        {
-            if (state == null)
-                throw new ArgumentNullException(nameof(state));
-
-            var context = (TContext)state;
-
-            if (_actions.Count > 0)
-            {
-                Iterations++;
-                var index = _random.Next(_actions.Count);
-                var startIteration = DateTime.Now;
-
-                _actions[index](context);
-                
-                var durationIteration = DateTime.Now - startIteration;
-                TotalDuration = TotalDuration.Add(durationIteration);
-                AvgIterationDuration = TimeSpan.FromTicks(TotalDuration.Ticks / Iterations);
-                if (AvgIterationDuration==TimeSpan.Zero)
-                    AvgIterationPerSecond = 0;
-                else
-                    AvgIterationPerSecond = TimeSpan.FromSeconds(1).Ticks / (decimal)AvgIterationDuration.Ticks;
-            }
-        }
-    }
-
-}

qdr.fnd.core.pqueue/Process/PsProcessProviderBase.cs

@@ -0,0 +1,77 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Quadarax.Foundation.Core.Data;
+using Quadarax.Foundation.Core.Logging;
+using Quadarax.Foundation.Core.Object;
+
+namespace qdr.fnd.core.pqueue.Process
+{
+    public class PsProcessProviderBase : DisposableObject
+    {
+        protected ILog Log { get; private set; }
+
+        public IDictionary<string, TypedArgument> Arguments { get; private set; }
+        public bool IsRunning { get; private set; }
+        public string ItemId { get; private set; }
+
+        public PsProcessProviderBase(string itemId, TypedArgument[] arguments, ILogger logger)
+        {
+            if (string.IsNullOrEmpty(itemId)) throw new ArgumentNullException(nameof(itemId));
+            ItemId = itemId;
+            Log = logger.GetLogger(GetType());
+            Arguments = arguments.ToDictionary(x => x.Name, x => x);
+        }
+        #region Overrides of DisposableObject
+
+        /// <inheritdoc />
+        protected override void OnDisposing()
+        {
+            throw new NotImplementedException();
+        }
+
+        #endregion
+
+        #region Implementation of IPsProcessProvider
+
+        /// <inheritdoc />
+        public void Validate()
+        {
+        }
+
+        /// <inheritdoc />
+        public async Task Start()
+        {
+            Validate();
+            try
+            {
+                await OnProcess();
+            }
+            catch (Exception e)
+            {
+                await OnProcessError(e);
+                Log.Log(LogSeverityEnum.Error, "Provider running failed.", e);
+            }
+        }
+
+        private async Task OnProcessError(Exception exception)
+        {
+            throw new NotImplementedException();
+        }
+
+        private async Task OnProcess()
+        {
+            throw new NotImplementedException();
+        }
+
+        /// <inheritdoc />
+        public async void Stop()
+        {
+            throw new NotImplementedException();
+        }
+
+        #endregion
+    }
+}

qdr.fnd.core.pqueue/PsQueue.cs

@@ -1,63 +1,112 @@
-using qdr.fnd.core.pqueue.Configuration;
+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.Object;
+using Quadarax.Foundation.Core.Thread;
 using Quadarax.Foundation.Core.Value.Generators;
 
 namespace qdr.fnd.core.pqueue
 {
-    public class PsQueue : DisposableObject
+    public class PsQueue : LoopWorker
     {
+        #region *** Constants ***
+        private const string CS_DEF_QUEUE_NAME = "PSQUEUE";
+        #endregion
+
         #region *** Properties ***
-        public string Name { get; }
-        public bool IsRunning { get; private set; }
         public PsQueueConfiguration Configuration { get;}
 
         public event PsQueueStatusChangedEventHandler? ProcessItemStatusChanged;
         #endregion
 
         #region *** Private Fields ***
-        private ILog _log;
         private TinyHash _hashGenerator;
-        protected IPsQueueStorage? _storage;
+        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)
+        public PsQueue(string queueName, PsQueueConfiguration configuration,IPsQueueStorage storageProvider, ILogger logger) : base(queueName,null)
         {
             _hashGenerator = new TinyHash(5);
             if (string.IsNullOrEmpty(queueName))
-                queueName = "PSQUEUE" + _hashGenerator.NewId();
+                queueName =CS_DEF_QUEUE_NAME + _hashGenerator.NewId();
 
-            Name = queueName;
-            IsRunning = false;
             Configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
             _storage = storageProvider ?? throw new ArgumentNullException(nameof(storageProvider));
-            _log = logger.GetLogger(queueName);
+
+            _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 void Start()
-        {
-            if (IsRunning) throw new PSQueueException(PSQueueException.ErrorCodes.PsQueueIsRunning);
-            _log.Log(LogSeverityEnum.Info, $"{Name}:queue started.");
-        }
-        public void Stop()
-        {
-            if (!IsRunning) throw new PSQueueException(PSQueueException.ErrorCodes.PsQueueIsRunning);
-            _log.Log(LogSeverityEnum.Info, $"{Name}:queue stopped.");
-        }
-
+        
         public string EnQueue(IPsQueueItemDescriptor item)
         {
-            throw new NotImplementedException();
+            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)
         {
@@ -97,13 +146,83 @@ namespace qdr.fnd.core.pqueue
 
 
         #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()
         {
-            if(IsRunning)
-                Stop();
+            base.OnDisposing();
 
             ProcessItemStatusChanged = null;
-            _storage = null;
+            _storage.Dispose();
             //TODO: Implement OnDisposing
         }
         #endregion

qdr.fnd.core.pqueue/qdr.fnd.core.pqueue.csproj

@@ -7,6 +7,10 @@
     <Nullable>enable</Nullable>
   </PropertyGroup>
 
+  <ItemGroup>
+    <Content Remove="C:\Users\Dalibor Votruba\.nuget\packages\qdr.fnd.core\0.0.4-alpha\contentFiles\any\net7.0\releasenotes.md" />
+  </ItemGroup>
+
   <ItemGroup>
     <PackageReference Include="qdr.fnd.core" Version="0.0.4-alpha" />
   </ItemGroup>