AppStd/DigitalData.MessagingService
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor OutgoingEmailQueue for RabbitMQ best practices

Browse Source 
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.
This commit is contained in:
tekh
2026-07-24 13:42:53 +02:00
parent 3cba69ec42
commit 77d3b52d16
1 changed files with 65 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs
View File

@@ -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<T> initialization pattern to avoid blocking constructor.
/// </summary>
public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable
public sealed class OutgoingEmailQueue(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailQueue> Logger, IEmailService EmailService) : IOutgoingEmailQueue, IAsyncDisposable
{
private readonly ILogger<OutgoingEmailQueue> _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<RabbitMqConfiguration> config, ILogger<OutgoingEmailQueue> 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
/// <summary>
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync.
/// </summary>
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<string, object?>
@@ -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<int> GetQueueDepthAsync(CancellationToken cancellationToken = default)
{
var queueInfo = await _channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);
var queueInfo = await _publishChannel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);
return (int)queueInfo.MessageCount;
}
/// <summary>
/// Start event-driven consumer that processes messages as they arrive
/// </summary>
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();
}
}
}