using AsbCloudApp.Data; using AsbCloudApp.Services; using Microsoft.Extensions.Configuration; using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Tasks; namespace AsbCloudInfrastructure.Services { /// /// Сервис выстраивает очередь из фоновых задач. Ограничивает количество одновременно выполняющихся задач. /// public class BackgroundWorkerService : IDisposable, IBackgroundWorkerService { private readonly Worker[] workers; private readonly Dictionary works = new Dictionary(); private bool isRunning = false; private CancellationTokenSource cts; private Task task; public BackgroundWorkerService(IConfiguration configuration) { var workersCount = configuration.GetValue("BackgroundWorkersCount", 4); workers = new Worker[workersCount]; for (int i = 0; i < workers.Length; i++) workers[i] = new Worker(); } ~BackgroundWorkerService() { Dispose(); } public string Enqueue(Func func) { var work = new Work { ActionAsync = func }; return Enqueue(work); } public string Enqueue(string id, Func func) { var work = new Work(id, func); return Enqueue(work); } public string Enqueue(string id, Func func, Func onError) { var work = new Work(id, func) { OnErrorAsync = onError }; return Enqueue(work); } string Enqueue(Work work) { works[work.Id] = work; if (!isRunning) { isRunning = true; cts = new CancellationTokenSource(); task = Task.Run(() => ExecuteAsync(cts.Token), cts.Token); } return work.Id; } private Work Dequeue() { var item = works.First(); works.Remove(item.Key); return item.Value; } public bool TryRemove(string id) => works.Remove(id); public bool Contains(string id) => works.ContainsKey(id); protected async Task ExecuteAsync(CancellationToken token) { while (works.Any() && !token.IsCancellationRequested) { var freeworker = workers.FirstOrDefault(w => !w.IsBusy); if (freeworker is not null) { var work = Dequeue(); freeworker.Start(work); } else await Task.Delay(10, token).ConfigureAwait(false); } isRunning = false; } public void Dispose() { cts?.Cancel(); task?.Wait(1); task?.Dispose(); cts?.Dispose(); task = null; cts = null; GC.SuppressFinalize(this); } } class Worker : IDisposable { private CancellationTokenSource cts; private Task task; public bool IsBusy { get; private set; } ~Worker() { Dispose(); } public void Dispose() { Stop(); GC.SuppressFinalize(this); } public void Start(Work work) { IsBusy = true; cts = new CancellationTokenSource(); task = Task.Run(async () => { try { var actionTask = work.ActionAsync(work.Id, cts.Token); await actionTask.WaitAsync(TimeSpan.FromMinutes(2), cts.Token); } catch (Exception ex) { Trace.TraceError(ex.Message); if (work.OnErrorAsync is not null) { try { await work.OnErrorAsync(work.Id, ex, cts.Token).ConfigureAwait(false); } catch (Exception exOnErrorHandler) { Trace.TraceError(exOnErrorHandler.Message); } } } finally { cts?.Dispose(); cts = null; IsBusy = false; } }, cts.Token); } public void Stop() { cts?.Cancel(); task?.Wait(1); task = null; cts?.Dispose(); cts = null; IsBusy = false; } } class Work { public string Id { get; private set; } public Func ActionAsync { get; set; } public Func OnErrorAsync { get; set; } public Work() { Id = Guid.NewGuid().ToString(); } public Work(string id, Func actionAsync) { Id = id; ActionAsync = actionAsync; } } }