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
}
}