qliberace/Common/qdr.fnd.core/Thread/AbstractWorker.cs

using System;
using System.Threading;
using System.Threading.Tasks;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.Object;
using Quadarax.Foundation.Core.Thread.Extensions;
using Quadarax.Foundation.Core.Value.Extensions;

namespace Quadarax.Foundation.Core.Thread
{
    public abstract class AbstractWorker : DisposableObject, IWorker
    {

        #region *** Private fields ***
        private Task _thread;
        private IQLog _log;
        private TimeSpan _delay;
        private CancellationTokenSource _cancel;
        protected IQLogger _logger;
        #endregion

        #region *** Properties ***
        public bool IsRunning { get; private set; }
        public bool IsTransient { get; private set; }
        public string Name { get; }
        public int CurrentThreadId => System.Threading.Thread.CurrentThread.ManagedThreadId;
        public TimeSpan Timeout { get; set; }
        protected IWorkerContext Context { get; private set; }
        public long IterationCount { get; protected set; } = 0;
        #endregion

        #region *** Constructors ***
        protected AbstractWorker() : this($"Worker#{Guid.NewGuid().ToBase32String()}", null)
        {

        }
        protected AbstractWorker(IWorkerContext context) : this($"Worker#{Guid.NewGuid().ToBase32String()}", context)
        {

        }
        protected AbstractWorker(string workerName) : this(workerName, null)
        {

        }
        protected AbstractWorker(string workerName, IWorkerContext context, IQLogger logger = null)
        {
            if (string.IsNullOrEmpty(workerName))
                throw new ArgumentNullException(nameof(workerName));
            Name = workerName;
            Context = context;
            _log = logger?.GetLogger(GetType());
            _logger = logger;
            Timeout = TimeSpan.Zero;
        }
        #endregion

        #region *** Public Operations ***

        public void Start(bool silentlyIfNotValidState = false)
        {
            Start(TimeSpan.Zero, silentlyIfNotValidState);
        }
        public void Start(TimeSpan runDelay, bool silentlyIfNotValidState = false)
        {
            if (IsTransient || IsRunning)
                if (!silentlyIfNotValidState)
                    throw new InvalidOperationException(
                        $"Worker '{Name}' is in transient or running state. Cannot start.");
                else
                    return;

            OnBeforeStart();

            _cancel = new CancellationTokenSource();
            _delay = runDelay;
            _thread = Task.Factory.StartNew(() => DoInternal(this), _cancel.Token);
            IsRunning = true;
            
        }


        public void Stop(bool silentlyIfNotValidState = false)
        {
            if (IsTransient || !IsRunning)
                if (!silentlyIfNotValidState)
                    throw new InvalidOperationException(
                        $"Worker '{Name}' is in transient or running state. Cannot start.");
                else
                    return;
            IsTransient = true;
            Abort();
            OnAfterStop();
        }
        #endregion

        #region *** Virtuals and protected ***
        protected virtual void OnBeforeStart()
        {
        }
        protected virtual void OnAfterStop()
        {
        }
        protected virtual async Task Do(IWorkerContext context)
        {
            throw new NotImplementedException("AbstractWorker.Do() must be overriden.");
        }

        protected void Log(LogSeverityEnum severity, string message)
        {
            _log?.Log(severity, message);
        }

        protected void Log(LogSeverityEnum severity, string message, Exception exception)
        {
            _log?.Log(severity, message, exception);
        }

        protected override void OnDisposing()
        {
            Abort();
            Context = null;
            _log = null;
            _logger = null;
        }

        #endregion

        #region *** Private Operations ***
        private async void DoInternal(object context)
        {
            var wk = (AbstractWorker)context;
            if (wk._delay != TimeSpan.Zero)
                wk._thread.Wait(wk._delay);
            wk.IsTransient = false;
            try
            {
                if (Timeout == TimeSpan.Zero)
                    await wk.Do(wk.Context);
                else
                    await wk.Do(wk.Context).TimeoutAfter(Timeout);
                IterationCount++;
            }
            catch (Exception e)
            {
                wk.Log(LogSeverityEnum.Error, $"Worker '{Name}' failed.", e);
                throw;
            }
            finally
            {
                wk.IsRunning = false;
            }
        }

        private void Abort()
        {
            if (_thread == null)
                return;
            _cancel.Cancel();
            _cancel.Dispose();
            _cancel = null;
            _thread = null;
            IsRunning = false;
            IsTransient = false;
            
        }
        #endregion
    }
}