using System; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; namespace Flyinsono.Client.Test.Utilities.Executors { public abstract class ExecutorBase 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); /// /// Gets the name of this executor. /// public string Name { get; } /// /// Gets all workers count. /// public int WorkerCount => GetWorkerCount(); /// /// Indicate this executor is closed /// public bool IsClosed => _closed; protected ExecutorBase(string name) { _executeDelayTime = TimeSpan.MinValue; Name = name; } protected ExecutorBase(string name, TimeSpan executeDelayTime) : this(name) { _executeDelayTime = executeDelayTime; } /// /// Gets current worker count in queue. /// /// protected abstract int GetWorkerCount(); /// /// Add one worker into the queue. /// /// protected abstract void AddWorker(ExecuteWorker worker); /// /// Take one worker from the queue. /// /// protected abstract ExecuteWorker TakeWorker(); /// /// Gets all workers. /// /// protected abstract IEnumerable> GetWorkers(); /// /// Clear all workers which not stated. /// protected abstract void CancelWorkers(); /// /// Add one execute func into the queue. /// /// /// public void Add(Func executeMethod, T parameter) { lock (WorkersLocker) { if (_closed) { return; } var worker = new ExecuteWorker(executeMethod, parameter); AddWorker(worker); if (!_executingStatus.Executing) { _executingStatus.Reset(); DoExecute(); } } } /// /// The final method to run the worker. /// private void DoExecute() { //Logic mush run in another thread. Task.Run(() => { ExecuteWorker 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(); } } }); } /// /// Close the executor and execute all workers. /// public void Close() { StopExecuting(); lock (WorkersLocker) { var workers = GetWorkers(); foreach (var worker in workers) { worker?.Run(); } } } /// /// Dispose current executor. /// public void Dispose(int timeoutSeconds = 10) { _closed = true; lock (WorkersLocker) { CancelWorkers(); _executingStatus.Wait(timeoutSeconds); } } /// /// Cancel all workers. /// public void Cancel() { lock (WorkersLocker) { CancelWorkers(); } StopExecuting(); } /// /// Stop executing. /// protected void StopExecuting() { _closed = true; lock (WorkersLocker) { if (WorkerCount == 0) { _closed = false; return; } _executingStatus.Wait(); } _closed = false; } public bool GetExecutingStatus() { return _executingStatus.Executing; } } }