319 lines
14 KiB
C#
319 lines
14 KiB
C#
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Text;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using LCU.API.Collections.RepoApi.Configuration;
|
|
using LCU.API.Collections.RepoApi.Exceptions;
|
|
using LCU.API.Collections.RepoApi.Services;
|
|
using LCU.API.Collections.RepoApi.Services.Producer;
|
|
using LCU.Api.Internal.MaidBotLib.Models;
|
|
using LCU.Shared.Collections.RepoApiLib.Models;
|
|
using LCU.Shared.Collections.RepoApiLib.Models.Consumer;
|
|
using LCU.Shared.Collections.RepoApiLib.Models.Producer;
|
|
using LCU.Shared.Collections.RepoApiLib.Repositories;
|
|
using LCU.Shared.Realtime.QueueLib.Models;
|
|
using LCU.Shared.Realtime.QueueLib.Queue;
|
|
using Microsoft.CodeAnalysis.Operations;
|
|
using Microsoft.EntityFrameworkCore.Internal;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Newtonsoft.Json;
|
|
using Serilog;
|
|
|
|
namespace LCU.API.Collections.RepoApi.HostedServices
|
|
{
|
|
/**
|
|
* Special Notes.
|
|
* 1. Direct injection of services cannot be done, the main API context is on it's own
|
|
* scope/thread. So instead, the scope factory is passed, so that this can create scopes.
|
|
* 2. This is designed to use 5 threads, from within an IHostedService thread context, to
|
|
* process queues. It's designed to do this in 'batches' to simulate a batch job, which
|
|
* is where the creation of the RunInfo comes into scope.
|
|
* 3. There is a main 'thread' here, which waits for a 'batch' of work to be complete.
|
|
* 4. Under the main thread, there are 5 worker threads, communicating using two queues besides
|
|
* the main Rabbit queue. The queues are used to set something up for processing (_inputQueue)
|
|
* and to mark something as done processing (_outputQueue).
|
|
*/
|
|
public class RepoQueueConsumerService : BackgroundService, IQueueItemProcessor
|
|
{
|
|
private readonly IQueueConnectionConfig _connectionConfig;
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
private RepoRunInfo _runInfo;
|
|
|
|
private const int ThreadCount = 5;
|
|
private BlockingCollection<QueueItem> _inputQueue;
|
|
private BlockingCollection<QueueResult> _outputQueue;
|
|
private readonly ConcurrentDictionary<string, ProcessingItem> _inProgress;
|
|
private readonly Task[] _workers;
|
|
private Task _consumer;
|
|
private QueueItemConsumer _queue;
|
|
private readonly IQueueConsumerConfig _consumerConfig;
|
|
private CancellationTokenSource _individualRunToken;
|
|
|
|
public RepoQueueConsumerService(IQueueConnectionConfig connectionConfig, RepoApiConfiguration repoApiConfiguration, IServiceScopeFactory scopeFactory)
|
|
{
|
|
_connectionConfig = connectionConfig;
|
|
_scopeFactory = scopeFactory;
|
|
_inProgress = new ConcurrentDictionary<string, ProcessingItem>();
|
|
_workers = new Task[ThreadCount];
|
|
_consumerConfig = repoApiConfiguration.QueueConsumerConfigs["RepoAccountQueue"];
|
|
}
|
|
|
|
bool TopLevelThreadController()
|
|
{
|
|
Log.Information($"[Startup] {_consumerConfig.GetQueueName()} Consumer is starting a run.");
|
|
_runInfo = new RepoRunInfo();
|
|
|
|
foreach (var workItem in _outputQueue.GetConsumingEnumerable())
|
|
{
|
|
_queue.AckItem(workItem.DeliveryTag);
|
|
if (workItem.Receipt != null)
|
|
{
|
|
_runInfo.Receipts.Add(workItem.Receipt);
|
|
}
|
|
|
|
var remainingCount = _queue.CheckQueueCount();
|
|
// First the input queue will finish, and we have to let running tasks finish.
|
|
if (remainingCount == 0 && EnumerableExtensions.Any(_inputQueue) == false && !_inputQueue.IsCompleted)
|
|
{
|
|
_inputQueue.CompleteAdding();
|
|
Task.WaitAll(_workers);
|
|
}
|
|
|
|
// That will cause the output queue to have the last THREADCOUNT items added, but
|
|
// we are IN the loop that processes them, so wen can't wait, just continue...
|
|
if (remainingCount == 0 && _inputQueue.IsCompleted && !_outputQueue.IsAddingCompleted)
|
|
{
|
|
_outputQueue.CompleteAdding();
|
|
}
|
|
|
|
// Ok, both queues are complete, which means all done.
|
|
if (_inputQueue.IsCompleted && _outputQueue.IsCompleted)
|
|
{
|
|
_individualRunToken.Cancel();
|
|
_inputQueue.Dispose();
|
|
_outputQueue.Dispose();
|
|
Log.Information($"{_consumerConfig.GetQueueName()} run completed... recording RunInfo: {@_runInfo}", _runInfo);
|
|
_runInfo.CompleteDateTime = DateTime.Now;
|
|
SaveRunInfo(_runInfo);
|
|
SendErrorNotification(_runInfo);
|
|
SendAlertNotification(_runInfo);
|
|
|
|
return true;
|
|
}
|
|
}
|
|
|
|
Log.Error($"[Startup] {_consumerConfig.GetQueueName()} Consumer is finished in an unexpected way. {@_runInfo}", _runInfo);
|
|
return false;
|
|
}
|
|
|
|
private void SaveRunInfo(RepoRunInfo runInfo)
|
|
{
|
|
using (var scope = _scopeFactory.CreateScope())
|
|
{
|
|
try
|
|
{
|
|
var receiptService = scope.ServiceProvider.GetRequiredService<IProcessingReceiptService>();
|
|
receiptService.SaveAllAsync(runInfo.Receipts).GetAwaiter().GetResult();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Log.Error(ex, "Could not save repo processor results.");
|
|
}
|
|
}
|
|
}
|
|
|
|
private void SendErrorNotification(RepoRunInfo runInfo)
|
|
{
|
|
if (runInfo.Receipts.Count > 0)
|
|
{
|
|
using (var scope = _scopeFactory.CreateScope())
|
|
{
|
|
var resp = scope.ServiceProvider.GetService<IEmailer>().SendProccessSummary(_runInfo);
|
|
if (!resp.Result.IsSuccessStatusCode)
|
|
{
|
|
Log.Error(
|
|
$"RepoQueueConsumerService failed to send error email. Emailer Response: {resp.Result.StatusCode} - {resp.Result.ReasonPhrase}");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private void SendAlertNotification(RepoRunInfo runInfo)
|
|
{
|
|
if (runInfo.Receipts.Count > 0)
|
|
{
|
|
using (var scope = _scopeFactory.CreateScope())
|
|
{
|
|
var resp = scope.ServiceProvider.GetService<IEmailer>().SendAlertSummaryEmailWithEndTime(runInfo.CompleteDateTime?? DateTime.Now);
|
|
if (!resp.Result.IsSuccessStatusCode)
|
|
{
|
|
Log.Error(
|
|
$"RepoQueueConsumerService failed to send alerts email. Emailer Response: {resp.Result.StatusCode} - {resp.Result.ReasonPhrase}");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private void SingleThreadMessageWorker(int workerId)
|
|
{
|
|
Log.Debug("[Startup] Worker {0} is starting.", workerId);
|
|
foreach (var workItem in _inputQueue.GetConsumingEnumerable())
|
|
{
|
|
var message = Encoding.UTF8.GetString(workItem.Body.ToArray());
|
|
Log.Information("Worker {0} Processing {1}", workerId, message);
|
|
var result = HandleSingleItem(message).GetAwaiter().GetResult();
|
|
result.DeliveryTag = workItem.DeliveryTag;
|
|
_outputQueue.Add(result);
|
|
}
|
|
|
|
Log.Debug("[Shutdown] Worker {0} is complete.", workerId);
|
|
}
|
|
|
|
/**
|
|
* This function is the main function run by the IHostedService interface, so
|
|
* should be considered the 'main' here.
|
|
*/
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
while (_queue == null)
|
|
{
|
|
try
|
|
{
|
|
_queue = new QueueItemConsumer(_connectionConfig, _consumerConfig, this);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Log.Error(ex, "Failed to create RepoQueueConsumerService, Will try again in 1 minute.");
|
|
await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
|
|
}
|
|
}
|
|
|
|
try
|
|
{
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
Log.Debug($"{_consumerConfig.GetQueueName()} Processor started a run.");
|
|
_inputQueue = new BlockingCollection<QueueItem>();
|
|
_outputQueue = new BlockingCollection<QueueResult>();
|
|
_consumer = Task<bool>.Factory.StartNew(TopLevelThreadController, stoppingToken);
|
|
for (var threadIndex = 0; threadIndex < ThreadCount; ++threadIndex)
|
|
{
|
|
var workerId = threadIndex;
|
|
var task = new Task(() => SingleThreadMessageWorker(workerId));
|
|
_workers[threadIndex] = task;
|
|
task.Start();
|
|
}
|
|
|
|
_individualRunToken = new CancellationTokenSource();
|
|
await _queue.ProcessItems(_individualRunToken.Token, false, ThreadCount);
|
|
|
|
while (!_consumer.IsCompleted)
|
|
{
|
|
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
|
|
Log.Debug($"{_consumerConfig.GetQueueName()} Finishing Heartbeat.");
|
|
}
|
|
|
|
_consumer?.Dispose();
|
|
_individualRunToken.Dispose();
|
|
Log.Debug($"{_consumerConfig.GetQueueName()} Processor Completed a run.");
|
|
await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
|
|
}
|
|
|
|
// the thread is cancelled, so the loop will exit, be sure to propagate the cancellation to our token
|
|
if (_individualRunToken != null && _individualRunToken.IsCancellationRequested == false)
|
|
{
|
|
_individualRunToken.Cancel();
|
|
}
|
|
}
|
|
catch (TaskCanceledException ex) when (stoppingToken.IsCancellationRequested)
|
|
{
|
|
Log.Information(ex, "Repo Processor was shut down normally.");
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group), Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
|
.Fatal(ex,$"There is an unhandled exception when processing repos; The IIS App Pool may need to be restarted.");
|
|
}
|
|
}
|
|
|
|
private async Task<QueueResult> HandleSingleItem(string itemAsString)
|
|
{
|
|
var result = new QueueResult();
|
|
ProcessingItem item = null;
|
|
try
|
|
{
|
|
using (var scope = _scopeFactory.CreateScope())
|
|
{
|
|
var repoAccountProcessor = scope.ServiceProvider.GetRequiredService<IRepoAccountProcessor>();
|
|
|
|
item = JsonConvert.DeserializeObject<ProcessingItem>(itemAsString);
|
|
Log.Debug($"{_consumerConfig.GetQueueName()} Consumer received: {@item}", item);
|
|
|
|
var newKey = item.DetermineAuditItemKey();
|
|
if (_inProgress.TryAdd(newKey, item) == false)
|
|
{
|
|
Log.Information("Conflicting item found for {0}, skipping.",
|
|
item.DetermineAuditItemKey(),
|
|
_inProgress[newKey].DetermineAuditItemKey());
|
|
return result;
|
|
}
|
|
|
|
var receipt = new ProcessingReceipt(item.AccountNo)
|
|
{
|
|
AccountId = item.AccountId ?? "",
|
|
CaseId = item.CaseId ?? "",
|
|
QueuedBy = item.QueuedBy
|
|
};
|
|
|
|
try
|
|
{
|
|
Log.Information($"Processing {item.AccountNo}");
|
|
receipt.Outcome = await repoAccountProcessor.HandleAccount(item);
|
|
receipt.Message = "Success";
|
|
}
|
|
catch (IneligibleAccountException iaEx)
|
|
{
|
|
receipt.Message = $"The item {item.AccountNo} is ineligible {iaEx.Field} : {iaEx.Reason}";
|
|
receipt.Outcome = RepoProcessingResult.Ineligible;
|
|
}
|
|
catch (MissingRequirementException mrEx)
|
|
{
|
|
receipt.Message = $"The item {item.AccountNo} is missing {mrEx.Field} : {mrEx.Advice}.";
|
|
receipt.Outcome = RepoProcessingResult.MissingRequirement;
|
|
}
|
|
catch (Exception ex) /* Mostly ServerErrorException */
|
|
{
|
|
receipt.Message = $"The item {item.AccountNo} produced an error. Please investigate.";
|
|
Log.Error(ex, receipt.Message);
|
|
}
|
|
|
|
if (!_inProgress.TryRemove(newKey, out item))
|
|
{
|
|
Log.Error("Couldn't remove item from concurrent dictionary. (Success branch) {@item}", item);
|
|
}
|
|
|
|
result.Receipt = receipt;
|
|
return result;
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Log.Fatal(ex, $"[INVESTIGATE] Failed processing {itemAsString}.");
|
|
if (item == null) return result;
|
|
var newKey = item.DetermineAuditItemKey();
|
|
if (!_inProgress.TryRemove(newKey, out item))
|
|
{
|
|
Log.Error("Couldn't remove item from concurrent dictionary. (Exception branch) {@item}", item);
|
|
}
|
|
return result;
|
|
}
|
|
}
|
|
|
|
public void HandleItem(QueueItem item)
|
|
{
|
|
_inputQueue.Add(item);
|
|
}
|
|
}
|
|
} |