using System; using System.Threading; using System.Threading.Tasks; using LCU.API.Collections.RepoApi.Configuration; using LCU.Shared.Collections.RepoApiLib.Hosted; using LCU.Shared.Realtime.QueueLib.Models; using LCU.Shared.Realtime.QueueLib.Queue; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Serilog; namespace LCU.API.Collections.RepoApi.HostedServices { public class HostedRepoTaskConsumer : BackgroundService, IQueueItemProcessor { private readonly IServiceScopeFactory _scopeFactory; private readonly IQueueConnectionConfig _configuration; private readonly IQueueConsumerConfig _consumerConfig; public HostedRepoTaskConsumer(IQueueConnectionConfig configuration, RepoApiConfiguration repoApiConfiguration, IServiceScopeFactory scopeFactory) { _configuration = configuration; _scopeFactory = scopeFactory; _consumerConfig = repoApiConfiguration.QueueConsumerConfigs["RepoProcessorTask"]; } protected override async Task ExecuteAsync(CancellationToken cancellationToken) { Log.Information("Hosted Queue Producer is starting."); try { var proxyQueue = new QueueItemConsumer(_configuration, _consumerConfig, this); await proxyQueue.ProcessItems(cancellationToken, true, 0); } catch (TaskCanceledException ex) when (cancellationToken.IsCancellationRequested) { Log.Information(ex, "Repo Producer was shut down normally."); } catch (Exception ex) { Log.Error(ex, "Error in main thread for HostedRepoTaskConsumer"); } Log.Information("Hosted Repo Producer is stopping."); } public void HandleItem(QueueItem item) { var settings = new TaskSettings(); using (var scope = _scopeFactory.CreateScope()) { var queueProducerService = scope.ServiceProvider.GetRequiredService(); queueProducerService.ImportData(settings).GetAwaiter().GetResult(); } } } }