WingServerCommon/Utilities/Executors/ExecutorBase.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace WingServerCommon.Utilities.Executors
{
    public abstract class ExecutorBase<T> where T : class
    {
        private readonly TimeSpan _executeDelayTime;
        protected readonly object WorkersLocker = new object();
        private volatile bool _closed;
        private readonly ExecutingStatus _executingStatus = new ExecutingStatus();
        private readonly ManualResetEvent _delayManualResetEvent = new ManualResetEvent(false);

        /// <summary>
        /// Gets the name of this executor.
        /// </summary>
        public string Name { get; }

        /// <summary>
        /// Gets all workers count.
        /// </summary>
        public int WorkerCount => GetWorkerCount();

        /// <summary>
        /// Indicate this executor is closed
        /// </summary>
        public bool IsClosed => _closed;

        protected ExecutorBase(string name)
        {
            _executeDelayTime = TimeSpan.MinValue;
            Name = name;
        }

        protected ExecutorBase(string name, TimeSpan executeDelayTime) : this(name)
        {
            _executeDelayTime = executeDelayTime;
        }

        /// <summary>
        /// Gets current worker count in queue.
        /// </summary>
        /// <returns></returns>
        protected abstract int GetWorkerCount();

        /// <summary>
        /// Add one worker into the queue.
        /// </summary>
        /// <param name="worker"></param>
        protected abstract void AddWorker(ExecuteWorker<T> worker);

        /// <summary>
        /// Take one worker from the queue.
        /// </summary>
        /// <returns></returns>
        protected abstract ExecuteWorker<T> TakeWorker();

        /// <summary>
        /// Gets all workers.
        /// </summary>
        /// <returns></returns>
        protected abstract IEnumerable<ExecuteWorker<T>> GetWorkers();

        /// <summary>
        /// Clear all workers which not stated.
        /// </summary>
        protected abstract void CancelWorkers();

        /// <summary>
        /// Add one execute func into the queue.
        /// </summary>
        /// <param name="executeMethod"></param>
        /// <param name="parameter"></param>
        public void Add(Func<T, bool> executeMethod, T parameter)
        {
            lock (WorkersLocker)
            {
                if (_closed)
                {
                    return;
                }

                var worker = new ExecuteWorker<T>(executeMethod, parameter);
                AddWorker(worker);
                if (!_executingStatus.Executing)
                {
                    _executingStatus.Reset();
                    DoExecute();
                }
            }
        }

        /// <summary>
        /// The final method to run the worker.
        /// </summary>
        private void DoExecute()
        {
            //Logic mush run in another thread.
            Task.Run(() =>
            {
                ExecuteWorker<T> worker = null;
                lock (WorkersLocker)
                {
                    if (WorkerCount > 0)
                    {
                        worker = TakeWorker();
                    }
                }

                var executeResult = worker?.Run() ?? true;

                if (_closed)
                {
                    _executingStatus.Set();
                }

                lock (WorkersLocker)
                {
                    var continueExecute = WorkerCount > 0 && !_closed && executeResult;
                    if (continueExecute)
                    {
                        if (_executeDelayTime != TimeSpan.MinValue)
                        {
                            _delayManualResetEvent.WaitOne(_executeDelayTime);
                        }

                        DoExecute();
                    }
                    else
                    {
                        _executingStatus.Set();
                    }
                }
            });
        }

        /// <summary>
        /// Close the executor and execute all workers.
        /// </summary>
        public void Close()
        {
            StopExecuting();
            lock (WorkersLocker)
            {
                var workers = GetWorkers();
                foreach (var worker in workers)
                {
                    worker?.Run();
                }
            }
        }

        /// <summary>
        /// Dispose current executor.
        /// </summary>
        public void Dispose(int timeoutSeconds = 10)
        {
            _closed = true;
            lock (WorkersLocker)
            {
                CancelWorkers();
                _executingStatus.Wait(timeoutSeconds);
            }
        }

        /// <summary>
        /// Cancel all workers.
        /// </summary>
        public void Cancel()
        {
            lock (WorkersLocker)
            {
                CancelWorkers();
            }
            StopExecuting();
        }

        /// <summary>
        /// Stop executing.
        /// </summary>
        protected void StopExecuting()
        {
            _closed = true;
            lock (WorkersLocker)
            {
                if (WorkerCount == 0)
                {
                    _closed = false;
                    return;
                }
                _executingStatus.Wait();
            }
            _closed = false;
        }

        public bool GetExecutingStatus()
        {
            return _executingStatus.Executing;
        }
    }
}