using System.Threading.Channels; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; namespace Cleanuparr.Infrastructure.Features.Messaging; /// /// Drains a channel one message at a time, each in its own DI scope. /// /// The queued message type. public sealed class QueueWorker : BackgroundService { private readonly ILogger> _logger; private readonly IServiceScopeFactory _scopeFactory; private readonly Channel _channel; /// /// Initializes a new instance of the class. /// /// The logger. /// Creates a DI scope per message. /// The channel to drain. public QueueWorker( ILogger> logger, IServiceScopeFactory scopeFactory, Channel channel) { _logger = logger; _scopeFactory = scopeFactory; _channel = channel; } /// protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await foreach (TMessage message in _channel.Reader.ReadAllAsync(stoppingToken)) { await HandleAsync(message); } } // A separate async method keeps each message's ContextProvider values out of the next one. private async Task HandleAsync(TMessage message) { try { await using AsyncServiceScope scope = _scopeFactory.CreateAsyncScope(); IMessageHandler handler = scope.ServiceProvider.GetRequiredService>(); await handler.HandleAsync(message); } catch (Exception exception) { _logger.LogError(exception, "failed to handle queued message | {MessageType}", typeof(TMessage).Name); } } }