add watchdog

Workbench/Processor/Dependencies/Reflection/AttributeQuery.cs

@@ -1,6 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Reflection;
+using System.Reflection;
 
 namespace Quadarax.Foundation.Core.Reflection
 {

Workbench/Processor/Program.cs

@@ -1,2 +1,32 @@
 // See https://aka.ms/new-console-template for more information
-Console.WriteLine("Hello, World!");
+
+using Quadarax.Foundation.Core.Business.Processor.Providers;
+using Quadarax.Foundation.Core.Business.Processor.Queue;
+using Quadarax.Foundation.Core.Object;
+
+Console.WriteLine("Processor test bench ...");
+
+using (var processor = new ProcessQueue("TEST", new InMemoryStorageProvider(), new ProcessQueueConfiguration(3, TimeSpan.FromSeconds(5))))
+{
+
+    for (int i = 0; i < 10; i++)
+    {
+        var item = processor.CreateTask($"task{i:D2}",
+            "test-class", Array.Empty<TypedArgument>(),
+            5,
+            5,
+            TimeSpan.FromSeconds(5),
+            DateTime.Now.AddSeconds(5),
+            DateTime.Now.AddHours(1));
+    }
+
+    processor.Start();
+    Console.WriteLine("Processor started ...");
+    Console.WriteLine("Press any key to stop ...");
+    Console.ReadKey();
+    processor.Stop();
+    Console.WriteLine("Processor stopped ...");
+}
+
+
+

Workbench/Processor/Queue/ProcessQueue.cs

@@ -26,6 +26,9 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
         private readonly ProcessQueueConfiguration _configuration;
         private readonly IProcessorStorageProvider _storageProvider;
         private bool _isTransient;
+
+        private Timer _timerWatchDog;
+        private bool _isFirstCycle;
         #endregion
 
         #region *** Constructors ***
@@ -58,12 +61,17 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
             Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] starting...");
 
             //TODO: start mechanism here
+            _isFirstCycle = true;
+            _timerWatchDog = new 
+                Timer(OnWatchdog, this, (int) _configuration.WatchdogInterval.TotalMilliseconds, (int) _configuration.WatchdogInterval.TotalMilliseconds);
+
 
             Log(LogSeverityEnum.Info, $"ProcessQueue [{QueueIdentifier}] started @{DateTime.Now.Subtract(now).ToReadableString()}");
             _isTransient = false;
             IsRunning = true;
         }
 
+        
         public void Stop()
         {
             if (!IsRunning || _isTransient)
@@ -117,7 +125,7 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
             return (ITaskItem[])result;
         }
 
-        public ITaskIdentifier[] QueryTasks(ProcessStatusEnum[] states,int maxItems, bool onlyTimeValid = true, bool includeDeleted = false)
+        public ITaskIdentifier[] QueryTasks(ProcessStatusEnum[] states,int maxItems, bool affinityQueue = true, bool onlyTimeValid = true, bool includeDeleted = false)
         {
             if (states == null || !states.Any()) throw new ArgumentNullException(nameof(states));
 
@@ -128,6 +136,10 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
             var items = _storageProvider.Query(states);
             if (onlyTimeValid)
                 items = items.Where(x=> x.ProcessingValidFrom <= now && x.ProcessingValidTo.GetValueOrDefault(DateTime.MaxValue) >= now);
+            
+            items = affinityQueue ? items.Where(x => x.AffinityToken != null && x.AffinityToken.QueueIdentifier == QueueIdentifier) 
+                : items.Where(x => x.AffinityToken == null);
+            
             var result = items.OrderBy(x => x.ProcessingValidFrom)
                 .ThenBy(x => x.Priority)
                 .Take(maxItems)
@@ -143,6 +155,40 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
 
         #region *** Private Operations ***
 
+        private void OnWatchdog(object? state)
+        {
+            try
+            {
+                if (_isFirstCycle)
+                {
+                    // 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.");
+                    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));
+                    }
+                    _isFirstCycle = false;
+                }
+
+                // gets processing tasks to know thread assigned slots
+
+                // calculate free slots and assigned slots
+                
+                // start assigned slots
+                
+            }
+            catch (Exception e)
+            {
+                Log(LogSeverityEnum.Error, e.Message, e);
+            }
+        }
+
+
+
         protected void Log(LogSeverityEnum type, string message, Exception? e = null)
         {
             if (type == LogSeverityEnum.Error || type == LogSeverityEnum.Fatal)

Workbench/Processor/Task/TaskItem.cs

@@ -1,4 +1,5 @@
-using Quadarax.Foundation.Core.Business.Processor.Enums;
+using System.Text;
+using Quadarax.Foundation.Core.Business.Processor.Enums;
 using Quadarax.Foundation.Core.Business.Processor.Queue;
 using Quadarax.Foundation.Core.Logging;
 using Quadarax.Foundation.Core.Object;
@@ -104,7 +105,7 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
             Priority = priority;
             Reference = reference;
             ProviderClass = providerClass;
-            Arguments.Setup(initialArguments);
+            _arguments.Setup(initialArguments);
             AttemptTimeout = attemptTimeout;
             AttemptsLeft = attepts;
             ProcessingValidFrom = validFrom;
@@ -143,6 +144,16 @@ namespace Quadarax.Foundation.Core.Business.Processor.Task
             return ((ITaskItemData)this).Equals(data);
         }
 
+        public override string ToString()
+        {
+            var sb = new StringBuilder();
+            sb.Append("Id=");
+            sb.Append(Id == long.MinValue ? "NEW" : Id.ToString()).Append(";");
+            sb.Append("Ref=").Append(Reference).Append(";");
+            sb.Append("Sta=").Append(Status).Append(";");
+            sb.Append("Cre=").Append(Created.ToString("G")).Append(";");
+            return sb.ToString();
+        }
         #endregion
 
         #region *** Internal Operations ***