bo/Workbench/Processor/Task/TaskItem.cs

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.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 TypedArgument[] Arguments => _arguments.ToArray(); 
        #endregion

        #region *** Private Fields ***
        private readonly List<ITaskStatusLogItem> _statusLog = new();
        private readonly List<TypedArgument> _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;
            Status = initialStatus;
        }
        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);
            Status = ProcessStatusEnum.Pending; 
            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) return;

            _statusLog.Add(new TaskStatusLogItem(Status, reason));
            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)
        {
            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);

            
        }

        #endregion
    }
}