using System.Text; using System.Threading; using Quadarax.Foundation.Core.Business.Processor.Enums; using Quadarax.Foundation.Core.Business.Processor.Exceptons; using Quadarax.Foundation.Core.Business.Processor.Queue; using Quadarax.Foundation.Core.Data; using Quadarax.Foundation.Core.Logging; using Quadarax.Foundation.Core.Object; using Quadarax.Foundation.Core.Value.Extensions; namespace Quadarax.Foundation.Core.Business.Processor.Task { internal class TaskItem : ITaskItem { #region *** Properties *** /// /// /// public long Id { get; internal set; } /// /// /// public ProcessQueueTypeEnum QueueType { get;private set; } /// /// /// public DateTime Created { get; private set;} /// /// /// public DateTime? Changed { get; private set;} /// /// /// public IAffinityToken? AffinityToken { get; private set;} /// /// /// public string Reference { get; private set;} /// /// /// public ProcessStatusEnum Status { get; private set; } /// /// /// public string ProviderClass { get; private set;} /// /// /// public byte Priority { get; private set;} /// /// /// public DateTime ProcessingValidFrom { get;private set; } /// /// /// public DateTime? ProcessingValidTo { get; private set;} /// /// /// public int AttemptsLeft { get; private set;} /// /// /// public int InitialAttempts { get; private set; } /// /// /// public TimeSpan? AttemptTimeout { get; private set;} /// /// /// public DateTime StatStarted { get; private set;} /// /// /// public DateTime StatFinished { get; private set;} /// /// /// public TimeSpan StatLastAttemptDuration { get; private set;} /// /// /// public ITaskStatusLogItem[] StatusLog => _statusLog.ToArray(); public TypedArgument[] Arguments => _arguments.ToArray(); #endregion #region *** Private Fields *** private readonly List _statusLog = new(); private readonly List _arguments = new(); private ILog _log; private ProcessQueue _parent; private Thread _thread; private CancellationTokenSource _cancellationTokenSource; #endregion #region *** Constructors *** private TaskItem(ProcessQueue parent, long id, ProcessStatusEnum initialStatus, string reference, string source, TypedArgument[] initialArguments, byte priority, int attepts, TimeSpan? attemptTimeout, DateTime validFrom, DateTime? validTo, ITaskStatusLogItem[]? initialStatusLog): this(parent, reference, source, initialArguments, priority, attepts, attemptTimeout, validFrom, validTo, initialStatusLog) { Id = id; SetStatus(initialStatus, "(System) Initialization"); } public TaskItem(ProcessQueue parent, string reference, string providerClass, TypedArgument[] initialArguments, byte priority, int attepts, TimeSpan? attemptTimeout, DateTime validFrom, DateTime? validTo, ITaskStatusLogItem[]? initialStatusLog) { AffinityToken = null; _parent = parent ?? throw new ArgumentNullException(nameof(parent)); _log = new ConsoleLogger().GetLogger(GetType()); Priority = priority; Reference = reference; ProviderClass = providerClass; _arguments.Setup(initialArguments); AttemptTimeout = attemptTimeout; AttemptsLeft = attepts; InitialAttempts = attepts; ProcessingValidFrom = validFrom; ProcessingValidTo = validTo; Created = DateTime.Now; if (initialStatusLog != null) _statusLog.Setup(initialStatusLog); SetStatus(ProcessStatusEnum.Pending, "(System) Initialization"); Id = long.MinValue; } #endregion #region *** Public Operations *** /// /// Starts task and sets to then or /// or (target status is depends on input arguments).. /// /// Allowed state: or /// Set state to: or or /// /// When task state is invalid for operation. /// /// Reason message to stop. /// If true, then will be set to otherwise to or . /// If true, then will be set to otherwise to . public void Start(IAffinityToken token) { if (_cancellationTokenSource != null) throw new InvalidOperationException("Inconsistent task state. Thread must not be initialized before task start."); Reload(); CheckTaskState(ProcessStatusEnum.Pending, ProcessStatusEnum.Paused); if (AttemptsLeft == 0) { SetStatus(ProcessStatusEnum.ExpiredAttemts, "(System) Attempts expired"); _parent.Provider.Set(this); return; } AffinityToken = token; SetStatus(ProcessStatusEnum.Attached, $"Attached to worker {AffinityToken}"); _parent.Provider.Set(this); _cancellationTokenSource = new CancellationTokenSource(); if (AttemptTimeout.HasValue) _cancellationTokenSource.CancelAfter(AttemptTimeout.Value); ThreadPool.QueueUserWorkItem(Execute, _cancellationTokenSource); } public void Postpone(TimeSpan postponeInterval, string reason) { throw new NotImplementedException(); } /// /// /// public void Stop(string reason, bool willPause = true, bool willFail = false) { if (_cancellationTokenSource == null) throw new InvalidOperationException("Inconsistent task state. Thread must be initialized before task stop."); Reload(); CheckTaskState(ProcessStatusEnum.Started); if (AttemptsLeft == 0) { SetStatus(ProcessStatusEnum.ExpiredAttemts, "(System) Attempts expired"); _parent.Provider.Set(this); return; } _cancellationTokenSource.Cancel(); _cancellationTokenSource = null; SetStatus(willPause ? ProcessStatusEnum.Paused : willFail ? ProcessStatusEnum.ProcessedFail : ProcessStatusEnum.ProcessedSucc, reason); StatFinished = DateTime.Now; _parent.Provider.Set(this); } public void Reset(string reason, TimeSpan postponeValid) { throw new NotImplementedException(); } public void Delete(string reason, bool immediate = false) { throw new NotImplementedException(); } public bool Euqals(ITaskItemData data) { 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 *** internal void SetStatus(ProcessStatusEnum status, string reason) { if (status == Status) { var log = _statusLog.Where(x => x.Status == status).MaxBy(x => x.Created); if (log == null) throw new InvalidOperationException($"Inconsistent task state. Status log must not be empty for status '{status}'."); if (log.Reason != reason) { ((TaskStatusLogItem)log).Reason = reason; Changed = DateTime.Now; return; } } _statusLog.Add(new TaskStatusLogItem(status, reason)); switch (status) { case ProcessStatusEnum.Initial: Created = DateTime.Now; break; case ProcessStatusEnum.Started: StatStarted = DateTime.Now; break; case ProcessStatusEnum.ProcessedFail: case ProcessStatusEnum.ProcessedSucc: case ProcessStatusEnum.ExpiredAttemts: case ProcessStatusEnum.ExpiredValidity: case ProcessStatusEnum.Failed: StatFinished = DateTime.Now; break; } Status = status; Changed = DateTime.Now; } internal void Reload() { if (Id == long.MinValue) throw new InvalidOperationException("Task is not saved yet."); var data = _parent.GetTask(this); CopyFrom(data); } internal void CopyFrom(ITaskItemData data) { Id = data.Id; Status = data.Status; Changed = data.Changed; Reference = data.Reference; ProviderClass = data.ProviderClass; Priority = data.Priority; ProcessingValidFrom = data.ProcessingValidFrom; ProcessingValidTo = data.ProcessingValidTo; Created = data.Created; Changed = data.Changed; StatStarted = data.StatStarted; StatFinished = data.StatFinished; StatLastAttemptDuration = data.StatLastAttemptDuration; AttemptsLeft = data.AttemptsLeft; InitialAttempts = data.InitialAttempts; AttemptTimeout = data.AttemptTimeout; _statusLog.Clear(); _statusLog.AddRange(data.StatusLog); _arguments.Clear(); _arguments.AddRange(data.Arguments); } bool IEquatable.Equals(ITaskItemData? other) { if (other == null) return false; var result = Id == other.Id && Status == other.Status && Changed == other.Changed && Reference == other.Reference && ProviderClass == other.ProviderClass && Priority == other.Priority && ProcessingValidFrom == other.ProcessingValidFrom && ProcessingValidTo == other.ProcessingValidTo && Created == other.Created && Changed == other.Changed && StatStarted == other.StatStarted && StatFinished == other.StatFinished && StatLastAttemptDuration == other.StatLastAttemptDuration && InitialAttempts == other.InitialAttempts && AttemptsLeft == other.AttemptsLeft && AttemptTimeout == other.AttemptTimeout && _statusLog.SequenceEqual(other.StatusLog) && _arguments.SequenceEqual(other.Arguments); return result; } #endregion #region *** Private Operations *** private void CheckValidity() { var now = DateTime.Now; if (!(ProcessingValidFrom <= now && now < ProcessingValidTo.GetValueOrDefault(DateTime.MaxValue))) throw new TaskValidityExpiredException(this); } private void CheckTaskState(params ProcessStatusEnum[] allowedStates) { if (allowedStates == null || allowedStates.Length == 0) throw new ArgumentNullException(nameof(allowedStates)); CheckValidity(); if (!allowedStates.Contains(Status)) throw new InvalidTaskStateException(this, allowedStates); } private void Execute(object? state) { try { if (state==null) throw new ArgumentNullException(nameof(state)); var token = (CancellationTokenSource)state; SetStatus(ProcessStatusEnum.Started, $"(System) Task started with attempt #{(InitialAttempts - AttemptsLeft) + 1}"); StatStarted = DateTime.Now; _parent.Provider.Set(this); } catch (Exception e) { } } #endregion } }