From 77d3b52d16f2e58679c9d20470986f6b693c2354 Mon Sep 17 00:00:00 2001 From: TekH Date: Fri, 24 Jul 2026 13:42:53 +0200 Subject: [PATCH] Refactor OutgoingEmailQueue for RabbitMQ best practices Refactored the `OutgoingEmailQueue` class to use separate RabbitMQ channels for publishing and consuming, improving thread safety and aligning with RabbitMQ best practices. Replaced the synchronous `Dispose` method with an asynchronous `DisposeAsync` for proper cleanup of resources. Introduced a `CancellationTokenSource` to manage the consumer's lifetime independently, ensuring graceful shutdown. Updated the `InitAsync` method to initialize dedicated channels and adjusted RabbitMQ topology declarations to use the publish channel. Replaced `_logger` with the injected `Logger` instance for consistency and updated RabbitMQ operations to use the appropriate channels with cancellation token support. Improved error handling and logging in the consumer, and ensured acknowledgments operate on the consume channel. Simplified the class structure by removing redundant fields and adopting C# 12 primary constructor syntax. Adjusted method parameters for clarity and added comments to explain design decisions. These changes enhance maintainability, scalability, and reliability. --- .../Queue/OutgoingEmailQueue.cs | 139 ++++++++---------- 1 file changed, 65 insertions(+), 74 deletions(-) diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs b/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs index f4f7f50..ba33c3e 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs @@ -1,7 +1,6 @@ using System.Text; using System.Text.Json; using DigitalData.EmailProfiler.Application.Common.Interfaces; -using DigitalData.EmailProfiler.Application.EmailSending.Commands; using DigitalData.EmailProfiler.Infrastructure.Messaging; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -16,30 +15,21 @@ namespace DigitalData.EmailProfiler.Infrastructure.Queue; /// Provides message persistence, scalability, and reliability. /// Uses Lazy initialization pattern to avoid blocking constructor. /// -public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable +public sealed class OutgoingEmailQueue(IOptions config, ILogger Logger, IEmailService EmailService) : IOutgoingEmailQueue, IAsyncDisposable { - private readonly ILogger _logger; - private readonly RabbitMqConfiguration _config; - private IConnection? _connection; - private IChannel _channel; - private readonly IEmailService _emailService; - -#pragma warning disable CS8618 // channel and connection are initialized in InitAsync, not in constructor - public OutgoingEmailQueue(IOptions config, ILogger logger, IEmailService emailService) -#pragma warning restore CS8618 - { - _logger = logger; - _config = config.Value; - _emailService = emailService; - } - + private readonly RabbitMqConfiguration _config = config.Value; + private IConnection? _connection = null; + private IChannel? _publishChannel = null; // Dedicated channel for publishing + private IChannel? _consumeChannel = null; // Dedicated channel for consuming + private readonly CancellationTokenSource _consumerCts = new(); // Independent lifetime from InitAsync token + /// /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. /// Called lazily on first use via EnsureInitializedAsync. /// - public async Task InitAsync(CancellationToken cancellationToken = default) + public async Task InitAsync(CancellationToken stoppingToken = default) { - _logger.LogInformation("Initializing RabbitMQ connection and queues..."); + Logger.LogInformation("Initializing RabbitMQ connection and queues..."); var factory = new ConnectionFactory { @@ -52,36 +42,24 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds) }; - _connection = await factory.CreateConnectionAsync(cancellationToken); - _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken); + _connection = await factory.CreateConnectionAsync(stoppingToken); + // Separate channels: IChannel is not thread-safe; one per role is best practice + _publishChannel = await _connection.CreateChannelAsync(cancellationToken: stoppingToken); + _consumeChannel = await _connection.CreateChannelAsync(cancellationToken: stoppingToken); + + // Topology declaration can use either channel; use publish channel here // Declare Dead Letter Queue (DLQ) exchange - await _channel.ExchangeDeclareAsync( - exchange: _config.DlqExchangeName, - type: ExchangeType.Direct, - durable: true, - autoDelete: false); + await _publishChannel.ExchangeDeclareAsync(exchange: _config.DlqExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: stoppingToken); // Declare Dead Letter Queue (DLQ) - await _channel.QueueDeclareAsync( - queue: _config.DlqQueueName, - durable: true, - exclusive: false, - autoDelete: false, - arguments: null); + await _publishChannel.QueueDeclareAsync(queue: _config.DlqQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: stoppingToken); // Bind DLQ to DLQ exchange - await _channel.QueueBindAsync( - queue: _config.DlqQueueName, - exchange: _config.DlqExchangeName, - routingKey: _config.DlqRoutingKey); + await _publishChannel.QueueBindAsync(queue: _config.DlqQueueName, exchange: _config.DlqExchangeName, routingKey: _config.DlqRoutingKey, cancellationToken: stoppingToken); // Declare main exchange (Direct type for routing) - await _channel.ExchangeDeclareAsync( - exchange: _config.ExchangeName, - type: ExchangeType.Direct, - durable: true, - autoDelete: false); + await _publishChannel.ExchangeDeclareAsync(exchange: _config.ExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: stoppingToken); // Declare main queue (durable for persistence) with DLQ arguments var queueArgs = new Dictionary @@ -90,22 +68,18 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable { "x-dead-letter-routing-key", _config.DlqRoutingKey } }; - await _channel.QueueDeclareAsync( - queue: _config.QueueName, - durable: true, - exclusive: false, - autoDelete: false, - arguments: queueArgs); + await _publishChannel.QueueDeclareAsync(queue: _config.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueArgs, cancellationToken: stoppingToken); // Bind main queue to exchange with routing key - await _channel.QueueBindAsync( - queue: _config.QueueName, - exchange: _config.ExchangeName, - routingKey: _config.RoutingKey); + await _publishChannel.QueueBindAsync(queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey, cancellationToken: stoppingToken); - await StartConsumerAsync(cancellationToken); + // Consumer uses its own CancellationToken independent of the startup token, + // so it keeps running after InitAsync completes or its token is cancelled. + // Link stoppingToken so the consumer stops when the host stops. + stoppingToken.Register(() => _consumerCts.Cancel()); + await StartConsumerAsync(_consumeChannel, _consumerCts.Token); - _logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName); + Logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName); } public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default) @@ -120,7 +94,7 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) }; - await _channel.BasicPublishAsync( + await _publishChannel!.BasicPublishAsync( exchange: _config.ExchangeName, routingKey: _config.RoutingKey, mandatory: false, @@ -131,16 +105,16 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable public async Task GetQueueDepthAsync(CancellationToken cancellationToken = default) { - var queueInfo = await _channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); + var queueInfo = await _publishChannel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); return (int)queueInfo.MessageCount; } /// /// Start event-driven consumer that processes messages as they arrive /// - private async Task StartConsumerAsync(CancellationToken cancellationToken) + private async Task StartConsumerAsync(IChannel consumeChannel, CancellationToken cancellationToken) { - var consumer = new AsyncEventingBasicConsumer(_channel); + var consumer = new AsyncEventingBasicConsumer(consumeChannel); consumer.ReceivedAsync += async (sender, args) => { @@ -152,34 +126,34 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable if (oMailEvent is not null) { - _logger.LogDebug("Received email message: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject); + Logger.LogDebug("Received email message: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject); - _logger.LogInformation("Processing outgoing email: To={To}, Subject={Subject}", + Logger.LogInformation("Processing outgoing email: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject); // Send email via SMTP (SMTP config is injected in IEmailService via IOptions) - await _emailService.SendEmailAsync( + await EmailService.SendEmailAsync( oMailEvent.Recipient, oMailEvent.Subject, oMailEvent.Body, isHtml: oMailEvent.IsHtml); - _logger.LogInformation("Email sent successfully: To={To}, Subject={Subject}", + Logger.LogInformation("Email sent successfully: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject); // Acknowledge message after successful processing - await _channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); - _logger.LogDebug("Message acknowledged: DeliveryTag={DeliveryTag}", args.DeliveryTag); + await consumeChannel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); + Logger.LogDebug("Message acknowledged: DeliveryTag={DeliveryTag}", args.DeliveryTag); } else { - _logger.LogWarning("Failed to deserialize email message: DeliveryTag={DeliveryTag}", args.DeliveryTag); - await _channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // Don't requeue invalid messages + Logger.LogWarning("Failed to deserialize email message: DeliveryTag={DeliveryTag}", args.DeliveryTag); + await consumeChannel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // Don't requeue invalid messages } } catch (Exception ex) { - _logger.LogError(ex, "Failed to process email [To={To}, Subject={Subject}] message: DeliveryTag={DeliveryTag}. Moving to DLQ (NO retry).", oMailEvent?.Recipient, oMailEvent?.Subject, args.DeliveryTag); + Logger.LogError(ex, "Failed to process email [To={To}, Subject={Subject}] message: DeliveryTag={DeliveryTag}. Moving to DLQ (NO retry).", oMailEvent?.Recipient, oMailEvent?.Subject, args.DeliveryTag); // TODO: Error Reporting Strategy // Option 1: Separate RabbitMQ Queue (emailprofiler.errors) @@ -202,25 +176,42 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable // - Real-time alerting via monitoring worker // NO RETRY - All failures move directly to DLQ - await _channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ + await consumeChannel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ } }; // Start consuming messages (event-driven, non-blocking) - await _channel.BasicConsumeAsync( + await consumeChannel.BasicConsumeAsync( queue: _config.QueueName, autoAck: false, consumer: consumer, cancellationToken: cancellationToken); - _logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName); + Logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName); } - public void Dispose() + public async ValueTask DisposeAsync() { - _channel?.CloseAsync().GetAwaiter().GetResult(); - _channel?.Dispose(); - _connection?.CloseAsync().GetAwaiter().GetResult(); - _connection?.Dispose(); + await _consumerCts.CancelAsync(); + _consumerCts.Dispose(); + + if (_consumeChannel is not null) + { + await _consumeChannel.CloseAsync(); + await _consumeChannel.DisposeAsync(); + } + + if (_publishChannel is not null) + { + await _publishChannel.CloseAsync(); + await _publishChannel.DisposeAsync(); + } + + if (_connection is not null) + { + await _connection.CloseAsync(); + await _connection.DisposeAsync(); + } } + }