|
|
@@ -117,16 +117,33 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
|
|
|
return (ITaskItem[])result;
|
|
|
}
|
|
|
|
|
|
- public ITaskIdentifier[] QueryTasks(ProcessStatusEnum[] states, bool onlyValid = true, bool includeDeleted = false)
|
|
|
+ public ITaskIdentifier[] QueryTasks(ProcessStatusEnum[] states,int maxItems, bool onlyTimeValid = true, bool includeDeleted = false)
|
|
|
{
|
|
|
+ if (states == null || !states.Any()) throw new ArgumentNullException(nameof(states));
|
|
|
|
|
|
+ var now = DateTime.Now;
|
|
|
+ if (!states.Contains(ProcessStatusEnum.Deleted) && includeDeleted)
|
|
|
+ states = states.Concat(new[] { ProcessStatusEnum.Deleted }).ToArray();
|
|
|
+
|
|
|
+ var items = _storageProvider.Query(states);
|
|
|
+ if (onlyTimeValid)
|
|
|
+ items = items.Where(x=> x.ProcessingValidFrom <= now && x.ProcessingValidTo.GetValueOrDefault(DateTime.MaxValue) >= now);
|
|
|
+ var result = items.OrderBy(x => x.ProcessingValidFrom)
|
|
|
+ .ThenBy(x => x.Priority)
|
|
|
+ .Take(maxItems)
|
|
|
+ .Select(x=>(ITaskIdentifier)x)
|
|
|
+ .ToArray();
|
|
|
+
|
|
|
+ Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] Query tasks {result.Length} of {maxItems}@{DateTime.Now.Subtract(now).ToReadableString()}");
|
|
|
+
|
|
|
+ return result;
|
|
|
}
|
|
|
|
|
|
#endregion
|
|
|
|
|
|
#region *** Private Operations ***
|
|
|
|
|
|
- protected void Log(LogSeverityEnum type, string message, Exception e = null)
|
|
|
+ protected void Log(LogSeverityEnum type, string message, Exception? e = null)
|
|
|
{
|
|
|
if (type == LogSeverityEnum.Error || type == LogSeverityEnum.Fatal)
|
|
|
{
|
|
|
@@ -137,7 +154,10 @@ namespace Quadarax.Foundation.Core.Business.Processor.Queue
|
|
|
|
|
|
protected override void OnDisposing()
|
|
|
{
|
|
|
- throw new NotImplementedException();
|
|
|
+ if (IsRunning)
|
|
|
+ Stop();
|
|
|
+ _delegates?.Reset();
|
|
|
+ Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] disposed.");
|
|
|
}
|
|
|
#endregion
|
|
|
}
|