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;
}
}
}