using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using Quadarax.Foundation.Core.Business.Processor.Enums; using Quadarax.Foundation.Core.Business.Processor.Providers; using Quadarax.Foundation.Core.Business.Processor.Task; using Quadarax.Foundation.Core.Logging; using Quadarax.Foundation.Core.Object; using Quadarax.Foundation.Core.Value.Extensions; namespace Quadarax.Foundation.Core.Business.Processor.Queue { public class ProcessQueue : DisposableObject { #region *** Properties *** public bool IsRunning { get; private set; } public string QueueIdentifier { get; private set; } #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 *** public ProcessQueue(string processQueueIdentifierToken, IProcessorStorageProvider storage, ProcessQueueConfiguration configuration, CallbackDelegates? delegates = null) { if (string.IsNullOrEmpty(processQueueIdentifierToken)) throw new ArgumentNullException(nameof(processQueueIdentifierToken)); QueueIdentifier = processQueueIdentifierToken; _delegates = delegates ?? new CallbackDelegates(); _log = new ConsoleLogger().GetLogger(GetType()); _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration)); _storageProvider = storage ?? throw new ArgumentNullException(nameof(storage)); _configuration.Validate(); } #endregion #region *** Public Operations *** public void Start() { 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() { 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(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) { 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) { 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,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) { if (type == LogSeverityEnum.Error || type == LogSeverityEnum.Fatal) { _delegates.OnProcessorError?.Invoke(message, e); } _log.Log(type, message,e); } protected override void OnDisposing() { if (IsRunning) Stop(); _delegates?.Reset(); Log(LogSeverityEnum.Debug, $"ProcessQueue [{QueueIdentifier}] disposed."); } #endregion } }