implements TaskItem (not comleted)

Workbench/Processor/Exceptons/InvalidTaskStateException.cs

@@ -26,6 +26,15 @@ namespace Quadarax.Foundation.Core.Business.Processor.Exceptons
             
         }
 
+        public InvalidTaskStateException(ITaskItem task, params ProcessStatusEnum[] expectedStatuses) : base(
+            GetMessage(task, expectedStatuses))
+        {
+            Data.Add(CS_EXCC_ENT_TYPE, task.GetType());
+            Data.Add(CS_EXCC_ENT_ID, task.Id);
+            Data.Add(CS_EXCC_ENT_CURRENT_STA, task.Status);
+            Data.Add(CS_EXCC_ENT_NEW_STA, string.Empty);
+            Data.Add(CS_EXCC_ENT_EXCEPTED_STA, expectedStatuses);
+        }
         #endregion
 
         #region *** Private Operations ***
@@ -34,6 +43,10 @@ namespace Quadarax.Foundation.Core.Business.Processor.Exceptons
         {
             return $"Invalid task state exception. Task: {task.GetType()}[{task.Id}] is in state {task.Status} and cannot be changed to {newStatus}. Expected states: {string.Join(", ", expectedStatuses)}";
         }
+        private static string GetMessage(ITaskItem task, params ProcessStatusEnum[] expectedStatuses)
+        {
+            return $"Invalid task state exception. Task: {task.GetType()}[{task.Id}] is in state {task.Status}. Expected states: {string.Join(", ", expectedStatuses)}";
+        }
         #endregion
     }
 }

Workbench/Processor/Exceptons/TaskValidityExpiredException.cs

@@ -0,0 +1,37 @@
+using Quadarax.Foundation.Core.Business.Processor.Task;
+
+namespace Quadarax.Foundation.Core.Business.Processor.Exceptons
+{
+    public class TaskValidityExpiredException : Exception
+    {
+        
+        #region *** Constants ***
+        private const string CS_EXCC_ENT_TYPE = "EntityType";
+        private const string CS_EXCC_ENT_ID = "EntityIdentifier";
+        private const string CS_EXCC_ENT_CURRENT_STA = "CurrentStatus";
+        private const string CS_EXCC_ENT_VALFROM = "ValidFrom";
+        private const string CS_EXCC_ENT_VALTO = "ValidTo";
+        #endregion
+
+        #region *** Constructors ***
+        public TaskValidityExpiredException(ITaskItem task) : base(GetMessage(task))
+        {
+            Data.Add(CS_EXCC_ENT_TYPE, task.GetType());
+            Data.Add(CS_EXCC_ENT_ID, task.Id);
+            Data.Add(CS_EXCC_ENT_CURRENT_STA, task.Status);
+            Data.Add(CS_EXCC_ENT_VALFROM, task.ProcessingValidFrom);
+            Data.Add(CS_EXCC_ENT_VALTO, task.ProcessingValidTo.GetValueOrDefault(DateTime.MaxValue));
+            
+        }
+
+        #endregion
+
+        #region *** Private Operations ***
+        private static string GetMessage(ITaskItem task)
+        {
+            return $"Task validity expired. Task: {task.GetType()}[{task.Id}] is out of validity ({task.ProcessingValidFrom} - {task.ProcessingValidTo.GetValueOrDefault(DateTime.MaxValue)})";
+        }
+        
+        #endregion
+        
+}

Workbench/Processor/Queue/ProcessQueue.cs

@@ -18,6 +18,7 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
         #region *** Properties ***
         public bool IsRunning { get; private set; }
         public string QueueIdentifier { get; private set; }
+        internal IProcessorStorageProvider Provider => _storageProvider;
         #endregion
 
         #region *** Private fields ***
@@ -29,6 +30,8 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
 
         private Timer _timerWatchDog;
         private bool _isFirstCycle;
+
+        private TaskSlots _taskSlots;
         #endregion
 
         #region *** Constructors ***
@@ -36,6 +39,8 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
         {
             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());
@@ -163,23 +168,30 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
                 {
                     // first cycle scenario
                     // reset previous left tasks
-                    var forceLeftTasks = QueryTasks(new[] { ProcessStatusEnum.Started}, _configuration.WorkersCount, true,true, false);
-
-                    Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] Watchdog first cycle. Force reset on {forceLeftTasks.Count()} tasks.");
+                    var forceLeftTaskIds = QueryTasks(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)
                     {
-                        ((ITaskItem)task).Stop("(System) Force stop of interrupted task.", true, false);
-                        ((ITaskItem)task).Reset("(System) Reset after force stop.", TimeSpan.FromSeconds(10));
+                        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
-
-                // calculate free slots and assigned slots
-                
-                // start assigned slots
-                
+                var assigningTaskIds = QueryTasks(new[] { ProcessStatusEnum.Pending }, _taskSlots.FreeSlotsCount, false, true, false);
+                var assigningTasks = GetTask(assigningTaskIds);
+                var startedCnt = 0;
+                foreach (var assign in assigningTasks)
+                {
+                    if (_taskSlots.AssignSlot(assign) >= 0)
+                    {
+                        ((TaskItem)assign).Start();
+                        startedCnt++;
+                    }
+                }
+                Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] Watchdog cycle. Assigned and started {startedCnt} of {assigningTaskIds.Length} tasks.");
             }
             catch (Exception e)
             {

Workbench/Processor/Queue/TaskSlots.cs

@@ -0,0 +1,90 @@
+using Quadarax.Foundation.Core.Business.Processor.Enums;
+using Quadarax.Foundation.Core.Business.Processor.Task;
+using Quadarax.Foundation.Core.Logging;
+
+namespace Quadarax.Foundation.Core.Business.Processor.Queue
+{
+    internal class TaskSlots
+    {
+        #region *** Properties ***
+        public int FreeSlotsCount => _taskSlots.Count(x => x == null);
+        #endregion
+
+        #region *** Fields ***
+        private ITaskItem?[] _taskSlots;
+        private readonly ILog _log;
+        #endregion
+
+        #region *** Indexer ***
+        public ITaskItem? this[int ordinal] => GetSlot(ordinal);
+        #endregion
+
+        #region *** Constructor ***
+        public TaskSlots(int initialMaxSlots)
+        {
+            _taskSlots = new ITaskItem?[initialMaxSlots];
+            _log = new ConsoleLogger().GetLogger(nameof(TaskSlots));
+        }
+        #endregion
+
+        #region *** Public Operations ***
+        public int AssignSlot(ITaskItem task)
+        {
+            if(task == null) throw new ArgumentNullException(nameof(task));
+
+            var ordinal = GetNextFreeSlotOrdinal();
+            if (ordinal  == -1) return -1;
+            _taskSlots[ordinal] = task;
+        
+            _log.Log(LogSeverityEnum.Debug, $"Task ({task}) assigned to slot # {ordinal}");
+
+            return ordinal;
+        }
+
+        public ITaskItem? GetSlot(int ordinal)
+        {
+            if (ordinal < 0 || ordinal >= _taskSlots.Length) throw new ArgumentOutOfRangeException(nameof(ordinal));
+            return _taskSlots[ordinal];
+        }
+
+        public void ReleaseSlot(int ordinal)
+        {
+
+            var task = GetSlot(ordinal);
+            if (task == null) return;
+
+            if (task.Status == ProcessStatusEnum.Started)
+                task.Stop("(System) Paused due change configuration.", true, false);
+
+            _taskSlots[ordinal] = null;
+            _log.Log(LogSeverityEnum.Debug, $"Task ({task}) was released from slot # {ordinal}");
+        }
+
+        public void ReleaseAll()
+        {
+            _log.Log(LogSeverityEnum.Debug, "Releasing all slots ...");
+            for (var a = 0; a < _taskSlots.Length; a++)
+                ReleaseSlot(a);
+        }
+
+        public void SetMaxSlots(int maxSlots)
+        {
+            _log.Log(LogSeverityEnum.Debug, $"Set max slots from {_taskSlots.Length} to {maxSlots}");
+            if (maxSlots < 1) throw new ArgumentOutOfRangeException(nameof(maxSlots));
+            ReleaseAll();
+            _taskSlots = new ITaskItem[maxSlots];
+        }
+        #endregion
+
+        #region *** Private Operations ***
+        private int GetNextFreeSlotOrdinal()
+        {
+            for (var i = 0; i < _taskSlots.Length; i++)
+            {
+                if (_taskSlots[i] == null) return i;
+            }
+            return -1;
+        }
+        #endregion
+    }
+}

Workbench/Processor/Task/ITaskItemData.cs

@@ -43,6 +43,10 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
         /// </summary>
         public int AttemptsLeft { get; }
         /// <summary>
+        /// Initial attempt count value
+        /// </summary>
+        public int InitialAttempts { get; }
+        /// <summary>
         /// Statistical timestamp when task begins to process.
         /// </summary>
         public DateTime StatStarted { get; }

Workbench/Processor/Task/TaskItem.cs

@@ -1,5 +1,7 @@
 using System.Text;
+using System.Threading;
 using Quadarax.Foundation.Core.Business.Processor.Enums;
+using Quadarax.Foundation.Core.Business.Processor.Exceptons;
 using Quadarax.Foundation.Core.Business.Processor.Queue;
 using Quadarax.Foundation.Core.Logging;
 using Quadarax.Foundation.Core.Object;
@@ -59,6 +61,11 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
         /// <inheritdoc cref="ITaskItem.AttemptsLeft"/>
         /// </summary>
         public int AttemptsLeft { get; private set;}
+        /// <summary>
+        /// <inheritdoc cref="ITaskItem.InitialAttepts"/>
+        /// </summary>
+        public int InitialAttempts { get; private set; }
+
         /// <summary>
         /// <inheritdoc cref="ITaskItem.AttemptTimeout"/>
         /// </summary>
@@ -87,7 +94,9 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
         private readonly List<ITaskStatusLogItem> _statusLog = new();
         private readonly List<TypedArgument> _arguments = new();
         private ILog _log;
-        private ProcessQueue _parent; 
+        private ProcessQueue _parent;
+        private Thread _thread;
+        private CancellationTokenSource _cancellationTokenSource;
         #endregion
 
         #region *** Constructors ***
@@ -108,6 +117,7 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
             _arguments.Setup(initialArguments);
             AttemptTimeout = attemptTimeout;
             AttemptsLeft = attepts;
+            InitialAttempts = attepts;
             ProcessingValidFrom = validFrom;
             ProcessingValidTo = validTo;
             Created = DateTime.Now;
@@ -120,14 +130,70 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
 
         
         #region *** Public Operations ***
+        /// <summary>
+        /// Starts task and sets <see cref="Status"/> to <see cref="ProcessStatusEnum.Attached"/> then <see cref="ProcessStatusEnum.Started"/> or
+        /// <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/> (target status is depends on input arguments)..
+        /// <list type="bullet">
+        ///	<item><description>Allowed state: <see cref="ProcessStatusEnum.Pending"/> or <see cref="ProcessStatusEnum.Paused"/></description></item>
+        ///	<item><description>Set state to: <see cref="ProcessStatusEnum.Fa"/> or <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/></description></item>
+        /// </list>
+        /// <exception cref="InvalidTaskStateException">When task state is invalid for operation.</exception>
+        /// </summary>
+        /// <param name="reason">Reason message to stop.</param>
+        /// <param name="willPause">If true, then <see cref="Status"/> will be set to <see cref="ProcessStatusEnum.Paused"/> otherwise to <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/>.</param>
+        /// <param name="willFail">If true, then <see cref="Status"/> will be set to <see cref="ProcessStatusEnum.ProcessedFail"/> otherwise to <see cref="ProcessStatusEnum.ProcessedSucc"/>.</param>
+        public void Start(IAffinityToken token)
+        {
+            if (_cancellationTokenSource != null) throw new InvalidOperationException("Inconsistent task state. Thread must not be initialized before task start.");
+
+            Reload();
+            CheckTaskState(ProcessStatusEnum.Pending, ProcessStatusEnum.Paused);
+            if (AttemptsLeft == 0)
+            {
+                SetStatus(ProcessStatusEnum.ExpiredAttemts, "(System) Attempts expired");
+                _parent.Provider.Set(this);
+                return;
+            }
+
+            AffinityToken = token;
+            SetStatus(ProcessStatusEnum.Attached, $"Attached to worker {AffinityToken}");
+            _parent.Provider.Set(this);
+
+            _cancellationTokenSource = new CancellationTokenSource();
+            if (AttemptTimeout.HasValue)
+                _cancellationTokenSource.CancelAfter(AttemptTimeout.Value);
+            
+            ThreadPool.QueueUserWorkItem(Execute, _cancellationTokenSource);
+        }
+
+
         public void Postpone(TimeSpan postponeInterval, string reason)
         {
             throw new NotImplementedException();
         }
 
+        /// <summary>
+        /// <inheritdoc cref="ITaskItem.Stop"/>
+        /// </summary>
         public void Stop(string reason, bool willPause = true, bool willFail = false)
         {
-            throw new NotImplementedException();
+            if (_cancellationTokenSource == null) throw new InvalidOperationException("Inconsistent task state. Thread must be initialized before task stop.");
+
+            Reload();
+            CheckTaskState(ProcessStatusEnum.Started);
+            if (AttemptsLeft == 0)
+            {
+                SetStatus(ProcessStatusEnum.ExpiredAttemts, "(System) Attempts expired");
+                _parent.Provider.Set(this);
+                return;
+            }
+
+            _cancellationTokenSource.Cancel();
+            _cancellationTokenSource = null;
+            SetStatus(willPause ? ProcessStatusEnum.Paused : willFail ? ProcessStatusEnum.ProcessedFail : ProcessStatusEnum.ProcessedSucc, reason);
+            StatFinished = DateTime.Now;
+            _parent.Provider.Set(this);
+            
         }
 
         public void Reset(string reason, TimeSpan postponeValid)
@@ -158,6 +224,22 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
 
         #region *** Internal Operations ***
 
+        internal void SetStatus(ProcessStatusEnum status, string reason)
+        {
+            if (status == Status) return;
+
+            _statusLog.Add(new TaskStatusLogItem(Status, reason));
+            Status = status; 
+            Changed = DateTime.Now;
+        }
+
+        internal void Reload()
+        {
+            if (Id == long.MinValue) throw new InvalidOperationException("Task is not saved yet.");
+            var data = _parent.GetTask(this);
+            CopyFrom(data);
+        }
+
         internal void CopyFrom(ITaskItemData data)
         {
             Id = data.Id;
@@ -174,6 +256,7 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
             StatFinished = data.StatFinished;
             StatLastAttemptDuration = data.StatLastAttemptDuration;
             AttemptsLeft = data.AttemptsLeft;
+            InitialAttempts = data.InitialAttempts;
             AttemptTimeout = data.AttemptTimeout;
             _statusLog.Clear();
             _statusLog.AddRange(data.StatusLog);
@@ -199,6 +282,7 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
                 StatStarted == other.StatStarted &&
                 StatFinished == other.StatFinished &&
                 StatLastAttemptDuration == other.StatLastAttemptDuration &&
+                InitialAttempts == other.InitialAttempts &&
                 AttemptsLeft == other.AttemptsLeft &&
                 AttemptTimeout == other.AttemptTimeout &&
                 _statusLog.SequenceEqual(other.StatusLog) &&
@@ -206,5 +290,38 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
             return result;
         }
         #endregion
+
+        #region *** Private Operations ***
+
+        private void CheckValidity()
+        {
+            var now = DateTime.Now;
+            if (!(ProcessingValidFrom <= now && now < ProcessingValidTo.GetValueOrDefault(DateTime.MaxValue)))
+                throw new TaskValidityExpiredException(this);
+        }
+
+        private void CheckTaskState(params ProcessStatusEnum[] allowedStates)
+        {
+            if (allowedStates == null || allowedStates.Length == 0) throw new ArgumentNullException(nameof(allowedStates));
+
+            CheckValidity();    
+
+            if (!allowedStates.Contains(Status))
+                throw new InvalidTaskStateException(this, allowedStates);
+        }
+
+        private void Execute(object? state)
+        {
+            if (state==null) throw new ArgumentNullException(nameof(state));
+            var token = (CancellationTokenSource)state;
+
+            SetStatus(ProcessStatusEnum.Started, $"(System) Task started with attempt #{(InitialAttempts - AttemptsLeft) + 1}");
+            StatStarted = DateTime.Now;
+            _parent.Provider.Set(this);
+
+            
+        }
+
+        #endregion
     }
 }

Workbench/Processor/Task/TaskStatusLogItem.cs

@@ -0,0 +1,22 @@
+using Quadarax.Foundation.Core.Business.Processor.Enums;
+
+namespace Quadarax.Foundation.Core.Business.Processor.Task
+{
+    internal class TaskStatusLogItem : ITaskStatusLogItem
+    {
+        #region *** Public Properties ***
+        public ProcessStatusEnum Status { get; }
+        public string Reason { get; }
+        public DateTime Created { get; }
+        #endregion
+
+        #region *** Constructors ***
+        public TaskStatusLogItem(ProcessStatusEnum status, string reason)
+        {
+            Status = status;
+            Reason = reason;
+            Created = DateTime.Now;
+        }
+        #endregion
+    }
+}