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