some changes

Common/qdr.app.qlbrc.common/Entities/Base/QlbrcEntityTracked.cs

@@ -15,5 +15,22 @@ namespace Quadarax.Application.QLiberace.Common.Entities.Base
         public string? Modifier { get; protected set;}
 
 
+        protected QlbrcEntityTracked()
+        {
+            SetCreated();
+        }
+
+        public void SetCreated(DateTime? created = null)
+        {
+            Created = created ?? DateTime.Now;
+        }
+
+        public void SetModify(string? modifier, DateTime? modified = null)
+        {
+            Modifier = modifier;
+            Modified = modified ?? DateTime.Now;
+        }
+
+
     }
 }

Common/qdr.app.qlbrc.common/Executor/BaseParametricExecutor.cs

@@ -18,9 +18,6 @@ namespace Quadarax.Application.QLiberace.Common.Executor
         {
             Log = logger?.GetLogger(GetType());
             Timeout = defaultTimeout;
-            
-
-            ends here
         }
 
 
@@ -39,11 +36,7 @@ namespace Quadarax.Application.QLiberace.Common.Executor
 
                 try
                 {
-                    var task = Task.Run(() => OnExecute(iid));
-                    if (task.Wait(Timeout))
-                        return task.Result;
-
-
+                    result = OnExecute(iid);
                 }
                 catch (Exception e)
                 {
@@ -53,7 +46,7 @@ namespace Quadarax.Application.QLiberace.Common.Executor
                 }
                 
                 Finished = DateTime.Now;
-                Log?.Log(LogSeverityEnum.Debug, $"({iid.ToBase32String()}) Executor type '{GetType().Name}' finished in duration {(Finished - Started).Value.ToReadableString()}. ");
+                Log?.Log(LogSeverityEnum.Debug, $"({iid.ToBase32String()}) Executor type '{GetType().Name}' finished (result={result}) in duration {(Finished - Started).Value.ToReadableString()}. ");
                 return result;
         }
 

Common/qdr.app.qlbrc.common/Executor/IParametricExecutorProvider.cs

@@ -16,5 +16,7 @@ namespace Quadarax.Application.QLiberace.Common.Executor
         string Code { get;}
         string TypeQualifiedName { get; }
         string Value { get; }
+
+        bool IsOutput { get; }
     }
 }

Common/qdr.app.qlbrc.common/Executor/ParameterExecutorValue.cs

@@ -4,15 +4,19 @@
     {
         public string Code { get; }
         public string TypeQualifiedName { get;}
-        public string Value { get; }
+        public string? Value { get; }
+        public bool IsOutput { get; }
 
-        public ParameterExecutorValue(string code, string typeQualifiedName, string value)
+
+        public ParameterExecutorValue(string code, string typeQualifiedName, string? value, bool isOutput)
         {
             if (string.IsNullOrEmpty(code)) throw new ArgumentNullException(nameof(code));
             Code = code;
             if (string.IsNullOrEmpty(typeQualifiedName)) throw new ArgumentNullException(nameof(typeQualifiedName));
+            IsOutput = isOutput;
             TypeQualifiedName = typeQualifiedName;
             Value = value;
+            IsOutput = isOutput;
         }
     }
 }

Common/qdr.app.qlbrc.common/Repositories/QlbrcTrackedRepository.cs

@@ -14,15 +14,13 @@ namespace Quadarax.Application.QLiberace.Common.Repositories
         public override TEntityTracked New()
         {
             var dao =  base.New();
-            dao.Created = DateTime.Now;
             return dao;
         }
 
         public override TEntityTracked Set(TEntityTracked entity)
         {
             var dao = base.Set(entity);
-            dao.Modified = DateTime.Now;
-            dao.Modifier = OperationContext.GetContext<QlbrcContext>().CurrentUser.Identity?.Name;
+            dao.SetModify(OperationContext.GetContext<QlbrcContext>().CurrentUser.Identity?.Name);
             return dao;
         }
 

Common/qdr.fnd.core.business/Exceptions/InvalidStateException.cs

@@ -0,0 +1,9 @@
+using System;
+
+namespace Quadarax.Foundation.Core.Business.Exceptions
+{
+    public class InvalidStateException : Exception
+    {
+        public InvalidStateException(string? message) : base(message) { }
+    }
+}

Common/qdr.fnd.core.business/ProcessQueue/IQueueItem.cs

@@ -0,0 +1,27 @@
+using System;
+
+namespace Quadarax.Foundation.Core.Business.ProcessQueue
+{
+    public interface IQueueItem : IDisposable
+    {
+        QueueItemStateEnum State { get; }
+
+        DateTime Created { get; }
+        DateTime? Modified { get; }
+        DateTime? Started { get; }
+        DateTime? Finished { get; }
+
+        DateTime StartFrom { get; }
+        DateTime? StartTo { get; }
+        string? Affinity { get; }
+        int Priority { get; }
+
+        void Assign(string? assignTo);
+        void Start();
+        void Stop();
+        void Pause();
+
+        string GetIdentifier();
+
+    }
+}

Common/qdr.fnd.core.business/ProcessQueue/IQueueProvider.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace Quadarax.Foundation.Core.Business.ProcessQueue
+{
+    public interface IQueueProvider : IDisposable
+    {
+        bool IsOpen { get; }
+        IQueryable<IQueueItem> Query();
+        void UpdateAndCommit(IEnumerable<IQueueItem> items);
+        void Open();
+        void Close();
+    }
+}

Common/qdr.fnd.core.business/ProcessQueue/Queue.cs

@@ -0,0 +1,69 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Quadarax.Foundation.Core.Data.Extensions;
+using Quadarax.Foundation.Core.Logging;
+using Quadarax.Foundation.Core.Object;
+using Quadarax.Foundation.Core.Value.Extensions;
+
+namespace Quadarax.Foundation.Core.Business.ProcessQueue
+{
+    public class Queue<TProvider> : DisposableObject where TProvider : IQueueProvider
+    {
+        private TProvider _provider;
+        private IQLogger _logger;
+        private IQLog _log;
+        private string _queueName;
+        private TimeSpan? _timeout;
+        
+        protected TProvider Provider => _provider;
+        protected IQLogger Logger => _logger;
+        public string Name => _queueName;
+        protected IQLog Log => _log;
+
+        
+
+        public Queue(TProvider provider,string queueName, TimeSpan? initialTimeout, IQLogger logger)
+        {
+            _provider = provider ?? throw new ArgumentNullException(nameof(provider));
+            if (string.IsNullOrEmpty(queueName)) throw new ArgumentNullException(nameof(queueName));
+            _queueName = queueName;
+            _timeout = initialTimeout;
+
+            if (logger != null)
+            {
+                _logger = logger;
+                _log = logger.GetLogger(GetType());
+            }
+        }
+
+        public IEnumerable<IQueueItem> Fetch(int itemsCount)
+        {
+            if (!Provider.IsOpen)
+                Provider.Open();
+            
+            var items = BuildFetchOrder(BuildFetchQuery(Provider.Query())).FirstPage(itemsCount).ToArray();
+
+            foreach (var item in items) item.Assign(Name);
+            Provider.UpdateAndCommit(items);
+            return items;
+        }
+
+
+        protected virtual IQueryable<IQueueItem> BuildFetchQuery(IQueryable<IQueueItem> query)
+        {
+            var now = DateTime.Now;
+            return query.Where(x => x.State.In(QueueItemStateEnum.New, QueueItemStateEnum.Pending) &&
+                                    x.StartFrom > now &&
+                                    x.StartTo.GetValueOrDefault(now.AddHours(1)) < now && x.Affinity == null);
+        }
+        protected virtual IQueryable<IQueueItem> BuildFetchOrder(IQueryable<IQueueItem> query)
+        {
+            return query.OrderBy(x => x.Priority).ThenBy(x => x.StartFrom);
+        }
+
+        protected override void OnDisposing()
+        {
+        }
+    }
+}

Common/qdr.fnd.core.business/ProcessQueue/QueueItem.cs

@@ -0,0 +1,85 @@
+using System;
+using System.IO;
+using System.Linq;
+using System.Text;
+using Quadarax.Foundation.Core.Business.Exceptions;
+using Quadarax.Foundation.Core.Object;
+using Quadarax.Foundation.Core.Value.Extensions;
+
+namespace Quadarax.Foundation.Core.Business.ProcessQueue
+{
+    public class QueueItem : DisposableObject, IQueueItem
+    {
+
+        public QueueItemStateEnum State { get; protected set;}
+        public DateTime Created { get; }
+        public DateTime? Modified { get; }
+        public DateTime? Started { get; }
+        public DateTime? Finished { get; }
+        public DateTime StartFrom { get; }
+        public DateTime? StartTo { get; }
+        public string Affinity { get; protected set; }
+        public int Priority { get; }
+
+        public void Assign(string assignTo)
+        {
+            if (!string.IsNullOrEmpty(Affinity) && !string.IsNullOrEmpty(assignTo))
+                throw new InvalidStateException(
+                    $"Query item '{GetIdentifier()}' has already set Affinity to '{Affinity}'");
+
+            ValidateState("set Affinity", QueueItemStateEnum.New, QueueItemStateEnum.Paused);
+
+            ends here
+
+
+            Affinity = assignTo;
+            
+        }
+
+        public void Start()
+        {
+            throw new NotImplementedException();
+        }
+
+        public void Stop()
+        {
+            throw new NotImplementedException();
+        }
+
+        public void Pause()
+        {
+            throw new NotImplementedException();
+        }
+
+        public string GetIdentifier()
+        {
+            return OnGetIdentifer();
+        }
+
+
+        protected void ValidateState(string operationName, params QueueItemStateEnum[] allowedStates)
+        {
+            if (string.IsNullOrEmpty(operationName)) throw new ArgumentNullException(nameof(operationName));
+
+            if (allowedStates.Length==0)
+                return;
+
+            if (!State.In(allowedStates))
+                throw new InvalidStateException(
+                    $"Query item '{GetIdentifier()}' has invalid state to {operationName} (expecting. {string.Join(",", allowedStates.Select(x=>x.ToString()))}");
+
+        }
+
+        protected virtual string OnGetIdentifer()
+        {
+            var sb = new StringBuilder();
+            sb.Append("QI").Append(Created.ToFileTimeUtc()).Append("_").Append(StartFrom.ToFileTimeUtc()).Append("_")
+                .Append(string.IsNullOrEmpty(Affinity) ? 0 : Affinity.GetHashCode());
+            return sb.ToString();
+        }
+
+        protected override void OnDisposing()
+        {
+        }
+    }
+}

Common/qdr.fnd.core.business/ProcessQueue/QueueItemStateEnum.cs

@@ -0,0 +1,15 @@
+using System;
+
+namespace Quadarax.Foundation.Core.Business.ProcessQueue
+{
+    public enum QueueItemStateEnum : int
+    {
+        New = 0,
+        Pending = 1,
+        Working = 2,
+        Paused = 10,
+        Finished = 3,
+        Failed = 4,
+        Deleted = 20,
+    }
+}

Common/qdr.fnd.core.data/Extensions/PagingExt.cs

@@ -23,5 +23,21 @@ namespace Quadarax.Foundation.Core.Data.Extensions
 
             return queryable.Skip(page.Value * pageSize.Value).Take(pageSize.Value + overlap);
         }
+
+        public static IEnumerable<TEntity> FirstPage<TEntity>(this IEnumerable<TEntity> queryable, int? itemsTake, int overlap = 0)
+        {
+            if (!(itemsTake.HasValue))
+                return queryable;
+            
+            return queryable.Page(1, itemsTake, overlap);
+        }
+
+        public static IQueryable<TEntity> FirstPage<TEntity>(this IQueryable<TEntity> queryable, int? itemsTake, int overlap = 0)
+        {
+            if (!(itemsTake.HasValue))
+                return queryable;
+            
+            return queryable.Page(1, itemsTake, overlap);
+        }
     }
 }

Common/qdr.fnd.core/Thread/AbstractWorker.cs

@@ -16,6 +16,7 @@ namespace Quadarax.Foundation.Core.Thread
         private IQLog _log;
         private TimeSpan _delay;
         private CancellationTokenSource _cancel;
+        protected IQLogger _logger;
         #endregion
 
         #region *** Properties ***
@@ -48,6 +49,7 @@ namespace Quadarax.Foundation.Core.Thread
             Name = workerName;
             Context = context;
             _log = logger?.GetLogger(GetType());
+            _logger = logger;
             Timeout = TimeSpan.Zero;
         }
         #endregion
@@ -118,6 +120,7 @@ namespace Quadarax.Foundation.Core.Thread
             Abort();
             Context = null;
             _log = null;
+            _logger = null;
         }
 
         #endregion

Modules/qdr.app.qlbrc.processor/Entities/QueueData.cs

@@ -1,4 +1,5 @@
 using Quadarax.Application.QLiberace.Common.Entities.Base;
+using Quadarax.Application.QLiberace.Common.Executor;
 
 namespace Quadarax.Application.QLiberace.Processor.Entities
 {
@@ -12,5 +13,13 @@ namespace Quadarax.Application.QLiberace.Processor.Entities
         public string ValueTypeQualified { get; set; } = null!;
 
         public string? Value { get; set; }
+
+        public bool IsInput { get; set; }
+
+
+        public IParametricExecutorValue ToParametrizedValue()
+        {
+            return new ParameterExecutorValue(Code, ValueTypeQualified, Value, !IsInput);
+        }
     }
 }

Modules/qdr.app.qlbrc.processor/Executor/DummyExecutor.cs

@@ -0,0 +1,22 @@
+using Quadarax.Application.QLiberace.Common.Executor;
+using Quadarax.Foundation.Core.Logging;
+
+namespace Quadarax.Application.QLiberace.Processor.Executor
+{
+    public class DummyExecutor : BaseParametricExecutor
+    {
+        private static Random _rnd = new Random();
+        
+        public DummyExecutor(TimeSpan defaultTimeout, IQLogger? logger = null) : base(defaultTimeout, logger)
+        {
+        }
+
+        protected override bool OnExecute(Guid iid)
+        {
+            var result = DummyExecutor._rnd.Next(100) >= 50;
+            var throwError = DummyExecutor._rnd.Next(100) >= 50;
+            if (throwError) throw new InvalidOperationException("Dummy random exception throw.");
+            return result;
+        }
+    }
+}

Modules/qdr.app.qlbrc.processor/Executor/ExecutorInternal.cs

@@ -0,0 +1,36 @@
+using Quadarax.Application.QLiberace.Processor.Entities;
+using Quadarax.Application.QLiberace.Common.Executor;
+using Quadarax.Foundation.Core.Data.Interface.Entity.Dto;
+using Quadarax.Foundation.Core.Logging;
+using Quadarax.Foundation.Core.Reflection.Extensions;
+
+namespace Quadarax.Application.QLiberace.Processor.Executor
+{
+    internal class ExecutorInternal
+    {
+        private ProcessingQueue _item;
+        private IList<ParameterExecutorValue> _data;
+
+        public ExecutorInternal(ProcessingQueue item, IEnumerable<ParameterExecutorValue> data)
+        {
+            _item = item ?? throw new ArgumentNullException(nameof(item));
+            _data = new List<ParameterExecutorValue>();
+        }
+
+        public ResultBoolDto Execute(TimeSpan timeout, IQLogger? logger = null)
+        {
+            try
+            {
+                var provider = (IParametricExecutorProvider) ActivatorExt.CreateInstance(_item.Provider,true, timeout, logger);
+                if (provider == null)
+                    throw new InvalidOperationException($"Cannot create executor provider from '{_item.Provider}'.");
+                var result = provider.Execute(_data);
+                return new ResultBoolDto(result);
+            }
+            catch (Exception e)
+            {
+                return new ResultBoolDto(e);
+            }
+        }
+    }
+}

Modules/qdr.app.qlbrc.processor/Executor/ItemExecutor.cs

@@ -1,12 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace Quadarax.Application.QLiberace.Processor.Executor
-{
-    internal class ItemExecutor
-    {
-    }
-}

Modules/qdr.app.qlbrc.processor/Repositories/RepoDoneQueue.cs

@@ -10,5 +10,14 @@ namespace Quadarax.Application.QLiberace.Processor.Repositories
         public RepoDoneQueue(IUnitOfWork<IDataDomain> unitOfWork, QlbrcContext operationContext) : base(unitOfWork, operationContext)
         {
         }
+
+        public DoneQueue NewFromProcessing(ProcessingQueue processing)
+        {
+            var done = New();
+            done.CloneFrom(processing);
+            Set(done);
+            //Context.Commit();
+            return done;
+        }
     }
 }

Modules/qdr.app.qlbrc.processor/Repositories/RepoQueueData.cs

@@ -19,9 +19,21 @@ namespace Quadarax.Application.QLiberace.Processor.Repositories
             data.ParentId = parentId;
         }
 
-        public IQueryable<QueueData> GetQueueData(int parentId)
+        public IQueryable<QueueData> GetQueueData(long headDataId)
         {
-            return Query(Paging.NoPaging()).Where(x => x.ParentId == parentId).OrderBy(x => x.SequenceNo);
+            return Query(Paging.NoPaging()).Where(x => x.ParentId == headDataId || x.Id == headDataId).OrderBy(x => x.SequenceNo);
+        }
+
+        public void UpdateQueueData(long headDataId, string code, string? value)
+        {
+            var item = Query(Paging.NoPaging()).FirstOrDefault(x => (x.ParentId == headDataId || x.Id == headDataId) && x.Code == code);
+            if (item == null)
+                throw new InvalidOperationException(
+                    $"No data found in QueueData for headDataId = {headDataId} and code = '{code}'!");
+            if (item.IsInput)
+                throw new InvalidOperationException(
+                    $"QueueData for headDataId = {headDataId} and code = '{code}' is only for input, cannot modify value!");
+            item.Value = value;
         }
 
 

Modules/qdr.app.qlbrc.processor/Workers/ProcessingWorker.cs

@@ -1,6 +1,8 @@
 using Quadarax.Application.QLiberace.Base.Services;
+using Quadarax.Application.QLiberace.Common.Executor;
 using Quadarax.Application.QLiberace.Processor.Entities;
 using Quadarax.Application.QLiberace.Processor.Enums;
+using Quadarax.Application.QLiberace.Processor.Executor;
 using Quadarax.Application.QLiberace.Processor.Repositories;
 using Quadarax.Application.QLiberace.Processor.Settings;
 using Quadarax.Foundation.Core.Business;
@@ -30,34 +32,61 @@ namespace Quadarax.Application.QLiberace.Processor.Workers
 
         #region *** Private Operations ***
 
-        protected override Task DoIteration(IWorkerContext context)
+        protected override async Task DoIteration(IWorkerContext context)
         {
             var ctx = (RepoWorkerContext)context;
-            // Get worker name
-            var workerName = GetName(Settings.ProcessingWorkerPwi);
-
-            var repoProcessing = ctx.GetRepository<RepoProcessingQueue>();
-
-            // 1. Continue to work on already processing entries in ProcessQ (paused, tryout failed)
-            // 2. If still available packet, lock pending entries in PendingQ
-            // 3. Move locked entries to ProcessQ
-            var processings = FetchFromQueue(ctx.GetRepository<RepoPendingQueue>(), ctx.GetRepository<RepoProcessingQueue>());
-
-            foreach (var processing in processings)
-            {
-                processing.Started = DateTime.Now;
-                processing.AffinityWorker = workerName;
-                processing.Status = QueueStatusEnum.Working;
-                repoProcessing.Set(processing);
-
-
-
-
-            }
-            // 4. Start to process
-            // 5. Decrement attempt
-            // 6. Move processed entries to DoneQ or FailedQ
-
+                // Get worker name
+                var workerName = GetName(Settings.ProcessingWorkerPwi);
+
+                var repoProcessing = ctx.GetRepository<RepoProcessingQueue>();
+                var repoData = ctx.GetRepository<RepoQueueData>();
+                var repoDone = ctx.GetRepository<RepoDoneQueue>();
+
+                // 1. Continue to work on already processing entries in ProcessQ (paused, tryout failed)
+                // 2. If still available packet, lock pending entries in PendingQ
+                // 3. Move locked entries to ProcessQ
+                var processing = FetchFromQueue(ctx.GetRepository<RepoPendingQueue>(), ctx.GetRepository<RepoProcessingQueue>());
+
+                foreach (var processItem in processing)
+                {
+                    processItem.Started = DateTime.Now;
+                    processItem.AffinityWorker = workerName;
+                    processItem.AttemptLeft -= 1;
+                    processItem.Status = QueueStatusEnum.Working;
+                    repoProcessing.Set(processItem);
+
+                    var dataList = repoData.GetQueueData(processItem.Id).ToArray();
+                    var paramList = dataList.Select(x => (ParameterExecutorValue)x.ToParametrizedValue()).ToArray();
+
+                    var exe = new ExecutorInternal(processItem, paramList);
+                    var result = exe.Execute(Timeout, _logger);
+                    if (result.IsSuccess)
+                    {
+                        //success
+                        //update modified output values (from Execute)
+                        foreach (var parOutput in paramList.Where(x => x.IsOutput))
+                        {
+                            repoData.UpdateQueueData(processItem.Id, parOutput.Code, parOutput.Value);
+                        }
+
+                        var done = repoDone.NewFromProcessing(processItem);
+                        done.Status = QueueStatusEnum.Success;
+                        processItem.Status = QueueStatusEnum.Deleted;
+                    }
+                    else
+                    {
+                        //failed
+                        if (processItem.AttemptLeft == 0)
+                        {
+                            //move to failed
+                        }
+                        else
+                        {
+                            // next tryout
+                            processItem.Status = QueueStatusEnum.Assigned;
+                        }
+                    }
+                }
         }
 
 

Tests/qdr.app.qlbrc.base.Tests/Mapper/BaseMapperDtoTest.cs

@@ -131,9 +131,8 @@ namespace Quadarax.Application.QLiberace.Base.Tests.Mapper
         private void SetTrackedAttributes(QlbrcEntityTracked entity)
         {
             entity.Id = EntityId;
-            entity.Modified = DateTime.Now;
-            entity.Created = DateTime.Now.AddHours(1);
-            entity.Modifier = EntityModifier;
+            entity.SetCreated(DateTime.Now.AddHours(1));
+            entity.SetModify(EntityModifier);
         }