|
@@ -8,6 +8,7 @@ using Quadarax.Foundation.Core.Business.Processor.Providers;
|
|
|
using Quadarax.Foundation.Core.Business.Processor.Task;
|
|
using Quadarax.Foundation.Core.Business.Processor.Task;
|
|
|
using Quadarax.Foundation.Core.Logging;
|
|
using Quadarax.Foundation.Core.Logging;
|
|
|
using Quadarax.Foundation.Core.Object;
|
|
using Quadarax.Foundation.Core.Object;
|
|
|
|
|
+using Quadarax.Foundation.Core.Value.Extensions;
|
|
|
|
|
|
|
|
namespace Quadarax.Foundation.Core.Business.Processor.Queue
|
|
namespace Quadarax.Foundation.Core.Business.Processor.Queue
|
|
|
{
|
|
{
|
|
@@ -19,19 +20,26 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
|
|
|
public string QueueIdentifier { get; private set; }
|
|
public string QueueIdentifier { get; private set; }
|
|
|
#endregion
|
|
#endregion
|
|
|
|
|
|
|
|
|
|
+ #region *** Private fields ***
|
|
|
|
|
+ private readonly CallbackDelegates _delegates;
|
|
|
|
|
+ private readonly ILog _log;
|
|
|
|
|
+ private readonly ProcessQueueConfiguration _configuration;
|
|
|
|
|
+ private readonly IProcessorStorageProvider _storageProvider;
|
|
|
|
|
+ private bool _isTransient;
|
|
|
|
|
+ #endregion
|
|
|
|
|
+
|
|
|
#region *** Constructors ***
|
|
#region *** Constructors ***
|
|
|
- public ProcessQueue(string processQueueIdentifierToken, IProcessorStorageProvider storage, ProcessQueueConfiguration configuration, CallbackDelegates delegates = null)
|
|
|
|
|
|
|
+ public ProcessQueue(string processQueueIdentifierToken, IProcessorStorageProvider storage, ProcessQueueConfiguration configuration, CallbackDelegates? delegates = null)
|
|
|
{
|
|
{
|
|
|
if (string.IsNullOrEmpty(processQueueIdentifierToken)) throw new ArgumentNullException(nameof(processQueueIdentifierToken));
|
|
if (string.IsNullOrEmpty(processQueueIdentifierToken)) throw new ArgumentNullException(nameof(processQueueIdentifierToken));
|
|
|
- IdentifierToken = processQueueIdentifierToken;
|
|
|
|
|
|
|
+ QueueIdentifier = processQueueIdentifierToken;
|
|
|
_delegates = delegates ?? new CallbackDelegates();
|
|
_delegates = delegates ?? new CallbackDelegates();
|
|
|
|
|
|
|
|
- _log = new QLogger().GetLogger(GetType());
|
|
|
|
|
|
|
+ _log = new ConsoleLogger().GetLogger(GetType());
|
|
|
_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
|
|
_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
|
|
|
_storageProvider = storage ?? throw new ArgumentNullException(nameof(storage));
|
|
_storageProvider = storage ?? throw new ArgumentNullException(nameof(storage));
|
|
|
|
|
|
|
|
- _configuration.Validate();
|
|
|
|
|
- _processingItems = new ProcessQueueItem[_configuration.WorkersCount];
|
|
|
|
|
|
|
+ _configuration.Validate();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@@ -40,25 +48,73 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
|
|
|
#region *** Public Operations ***
|
|
#region *** Public Operations ***
|
|
|
public void Start()
|
|
public void Start()
|
|
|
{
|
|
{
|
|
|
- throw new NotImplementedException();
|
|
|
|
|
|
|
+ if (IsRunning || _isTransient)
|
|
|
|
|
+ {
|
|
|
|
|
+ Log(LogSeverityEnum.Warn, $"ProcessQueue [{QueueIdentifier}] already running! Skip start.");
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ _isTransient = true;
|
|
|
|
|
+ var now = DateTime.Now;
|
|
|
|
|
+ Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] starting...");
|
|
|
|
|
+
|
|
|
|
|
+ //TODO: start mechanism here
|
|
|
|
|
+
|
|
|
|
|
+ Log(LogSeverityEnum.Info, $"ProcessQueue [{QueueIdentifier}] started @{DateTime.Now.Subtract(now).ToReadableString()}");
|
|
|
|
|
+ _isTransient = false;
|
|
|
|
|
+ IsRunning = true;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
public void Stop()
|
|
public void Stop()
|
|
|
{
|
|
{
|
|
|
- throw new NotImplementedException();
|
|
|
|
|
|
|
+ if (!IsRunning || _isTransient)
|
|
|
|
|
+ {
|
|
|
|
|
+ Log(LogSeverityEnum.Warn, $"ProcessQueue [{QueueIdentifier}] already stopped! Skip stop.");
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ _isTransient = true;
|
|
|
|
|
+ var now = DateTime.Now;
|
|
|
|
|
+ Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] stopping...");
|
|
|
|
|
+
|
|
|
|
|
+ //TODO: start mechanism here
|
|
|
|
|
+
|
|
|
|
|
+ Log(LogSeverityEnum.Info, $"ProcessQueue [{QueueIdentifier}] stopped @{DateTime.Now.Subtract(now).ToReadableString()}");
|
|
|
|
|
+ _isTransient = false;
|
|
|
|
|
+ IsRunning = false;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- public ITaskIdentifier CreateTask()
|
|
|
|
|
|
|
+ public ITaskIdentifier CreateTask(string reference, string providerClass, TypedArgument[] initialArguments, byte priority, int initialAttempts, TimeSpan attemptTimeout, DateTime validFrom, DateTime? validTo)
|
|
|
{
|
|
{
|
|
|
-
|
|
|
|
|
|
|
+ try
|
|
|
|
|
+ {
|
|
|
|
|
+ var task = new TaskItem(this,reference, providerClass,initialArguments,priority,initialAttempts,attemptTimeout,validFrom,validTo, null);
|
|
|
|
|
+ var newId = _storageProvider.Create(task);
|
|
|
|
|
+ task.Id = newId.Id;
|
|
|
|
|
+ return task;
|
|
|
|
|
+ }
|
|
|
|
|
+ catch (Exception e)
|
|
|
|
|
+ {
|
|
|
|
|
+ var message = $"Error creating task [Ref:{reference}, Provider:{providerClass}] for queue [{QueueIdentifier}]";
|
|
|
|
|
+ Log(LogSeverityEnum.Error, message, e);
|
|
|
|
|
+ throw new Exception(message, e);
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
public ITaskItem GetTask(ITaskIdentifier id)
|
|
public ITaskItem GetTask(ITaskIdentifier id)
|
|
|
{
|
|
{
|
|
|
-
|
|
|
|
|
|
|
+ if (id == null) throw new ArgumentNullException(nameof(id));
|
|
|
|
|
+ var result = _storageProvider.Get(id).FirstOrDefault();
|
|
|
|
|
+ return result == null ? throw new Exception($"Task [{id.Id}] not found!") : (ITaskItem)result;
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
public ITaskItem[] GetTask(ITaskIdentifier[] ids)
|
|
public ITaskItem[] GetTask(ITaskIdentifier[] ids)
|
|
|
{
|
|
{
|
|
|
|
|
+ var result = _storageProvider.Get(ids);
|
|
|
|
|
+ var diff = result.Select(x => x.Id).CompareDiff(ids.Select(x => x.Id));
|
|
|
|
|
+
|
|
|
|
|
+ if (diff.Any())
|
|
|
|
|
+ throw new Exception($"Tasks [{string.Join(",", diff)}] not found!");
|
|
|
|
|
|
|
|
|
|
+ return (ITaskItem[])result;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
public ITaskIdentifier[] QueryTasks(ProcessStatusEnum[] states, bool onlyValid = true, bool includeDeleted = false)
|
|
public ITaskIdentifier[] QueryTasks(ProcessStatusEnum[] states, bool onlyValid = true, bool includeDeleted = false)
|
|
@@ -69,6 +125,16 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
|
|
|
#endregion
|
|
#endregion
|
|
|
|
|
|
|
|
#region *** Private Operations ***
|
|
#region *** Private Operations ***
|
|
|
|
|
+
|
|
|
|
|
+ protected void Log(LogSeverityEnum type, string message, Exception e = null)
|
|
|
|
|
+ {
|
|
|
|
|
+ if (type == LogSeverityEnum.Error || type == LogSeverityEnum.Fatal)
|
|
|
|
|
+ {
|
|
|
|
|
+ _delegates.OnProcessorError?.Invoke(message, e);
|
|
|
|
|
+ }
|
|
|
|
|
+ _log.Log(type, message,e);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
protected override void OnDisposing()
|
|
protected override void OnDisposing()
|
|
|
{
|
|
{
|
|
|
throw new NotImplementedException();
|
|
throw new NotImplementedException();
|