| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364 |
- 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 ***
- /// <summary>
- /// <inheritdoc cref="ITaskItem.Id"/>
- /// </summary>
- public long Id { get; internal set; }
- /// <summary>
- /// <inheritdoc cref="ITaskItem.QueueType"/>
- /// </summary>
- public ProcessQueueTypeEnum QueueType { get;private set; }
- /// <summary>
- /// <inheritdoc cref="ITaskItem.Created"/>
- /// </summary>
- public DateTime Created { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.Changed"/>
- /// </summary>
- public DateTime? Changed { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.AffinityToken"/>
- /// </summary>
- public IAffinityToken? AffinityToken { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.Reference"/>
- /// </summary>
- public string Reference { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.Status"/>
- /// </summary>
- public ProcessStatusEnum Status { get; private set; }
- /// <summary>
- /// <inheritdoc cref="ITaskItem.ProviderClass"/>
- /// </summary>
- public string ProviderClass { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.Priority"/>
- /// </summary>
- public byte Priority { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.ProcessingValidFrom"/>
- /// </summary>
- public DateTime ProcessingValidFrom { get;private set; }
- /// <summary>
- /// <inheritdoc cref="ITaskItem.ProcessingValidTo"/>
- /// </summary>
- public DateTime? ProcessingValidTo { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.AttemptsLeft"/>
- /// </summary>
- public int AttemptsLeft { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.InitialAttepts"/>
- /// </summary>
- public int InitialAttempts { get; private set; }
- /// <summary>
- /// <inheritdoc cref="ITaskItem.AttemptTimeout"/>
- /// </summary>
- public TimeSpan? AttemptTimeout { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.StatStarted"/>
- /// </summary>
- public DateTime StatStarted { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.StatFinished"/>
- /// </summary>
- public DateTime StatFinished { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.StatLastAttemptDuration"/>
- /// </summary>
- public TimeSpan StatLastAttemptDuration { get; private set;}
- /// <summary>
- /// <inheritdoc cref="ITaskItem.StatusLog"/>
- /// </summary>
- public ITaskStatusLogItem[] StatusLog => _statusLog.ToArray();
- public ITypedArgument[] Arguments => _arguments.ToArray();
- #endregion
- #region *** Private Fields ***
- private readonly List<ITaskStatusLogItem> _statusLog = new();
- private readonly List<ITypedArgument> _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, ITypedArgument[] 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 ***
- /// <summary>
- /// Starts task and sets <see cref="Status"/> to <see cref="ProcessStatusEnum.Attached"/> then <see cref="ProcessStatusEnum.Started"/> or
- /// <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/> (target status is depends on input arguments)..
- /// <list type="bullet">
- /// <item><description>Allowed state: <see cref="ProcessStatusEnum.Pending"/> or <see cref="ProcessStatusEnum.Paused"/></description></item>
- /// <item><description>Set state to: <see cref="ProcessStatusEnum.Fa"/> or <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/></description></item>
- /// </list>
- /// <exception cref="InvalidTaskStateException">When task state is invalid for operation.</exception>
- /// </summary>
- /// <param name="reason">Reason message to stop.</param>
- /// <param name="willPause">If true, then <see cref="Status"/> will be set to <see cref="ProcessStatusEnum.Paused"/> otherwise to <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/>.</param>
- /// <param name="willFail">If true, then <see cref="Status"/> will be set to <see cref="ProcessStatusEnum.ProcessedFail"/> otherwise to <see cref="ProcessStatusEnum.ProcessedSucc"/>.</param>
- 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();
- }
- /// <summary>
- /// <inheritdoc cref="ITaskItem.Stop"/>
- /// </summary>
- 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<ITaskItemData>.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
- }
- }
|