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

Refactor OutgoingEmailConsumer for better initialization

Browse Source 
Refactored the `OutgoingEmailConsumer` class to improve maintainability, readability, and robustness. Changed the class to explicitly inherit from `IAsyncDisposable` and introduced lazy initialization for RabbitMQ connections and consumers via `_lazyInit`.

Enhanced error handling in `consumer.ReceivedAsync` by adding detailed logging, `BasicNack` for invalid messages, and placeholders for error reporting strategies. Improved logging for consumer startup and added safeguards against multiple initializations.

Removed outdated comments, updated documentation, and ensured proper resource cleanup in `DisposeAsync`.
This commit is contained in:
tekh
2026-07-27 16:32:34 +02:00
parent bf06b31656
commit b2d478c335
1 changed files with 87 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs
View File

@@ -15,21 +15,25 @@ namespace DigitalData.MessagingService.Infrastructure.Queue;
/// Provides message persistence, scalability, and reliability. /// Provides message persistence, scalability, and reliability.
/// Uses Lazy<T> initialization pattern to avoid blocking constructor. /// Uses Lazy<T> initialization pattern to avoid blocking constructor.
/// </summary> /// </summary>
public sealed class OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailConsumer> Logger, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory) : IAsyncDisposable public sealed class OutgoingEmailConsumer : IAsyncDisposable
{ {
private readonly RabbitMqConfiguration _config = config.Value; private readonly RabbitMqConfiguration _config;
private readonly Lazy<Task<IChannel>> _lazyChannel = new(CnnFactory.CreateChannelAsync); private readonly Lazy<Task<IChannel>> _lazyChannel;
private readonly AsyncEventingBasicConsumer? consumer; private readonly AsyncEventingBasicConsumer? consumer;
/// <summary> private readonly Lazy<Task> _lazyInit;
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Start event-driven consumer that processes messages as they arrive private readonly ILogger<OutgoingEmailConsumer>? _logger;
/// Called lazily on first use via EnsureInitializedAsync.
/// </summary> public OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory, ILogger<OutgoingEmailConsumer>? logger = null)
public async Task InitAsync()
{ {
_logger = logger;
_config = config.Value;
_lazyChannel = new(CnnFactory.CreateChannelAsync);
_lazyInit = new(async () => {
var channel = await _lazyChannel.Value; var channel = await _lazyChannel.Value;
var consumer = new AsyncEventingBasicConsumer(channel); var consumer = new AsyncEventingBasicConsumer(channel);
@@ -56,13 +60,13 @@ public sealed class OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config
} }
else else
{ {
Logger.LogWarning("Failed to deserialize email message: DeliveryTag={DeliveryTag}", args.DeliveryTag); 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 await channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // Don't requeue invalid messages
} }
} }
catch (Exception ex) 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 // TODO: Error Reporting Strategy
// Option 1: Separate RabbitMQ Queue (emailprofiler.errors) // Option 1: Separate RabbitMQ Queue (emailprofiler.errors)
@@ -96,7 +100,21 @@ public sealed class OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config
consumer: consumer, consumer: consumer,
cancellationToken: CnnFactory.CancellationToken); cancellationToken: CnnFactory.CancellationToken);
Logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName); logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName);
});
}
/// <summary>
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Start event-driven consumer that processes messages as they arrive
/// Called lazily on first use via EnsureInitializedAsync.
/// </summary>
public async Task InitAsync()
{
if (_lazyInit.IsValueCreated)
_logger?.LogWarning("OutgoingEmailConsumer already initialized. InitAsync() called multiple times.");
await _lazyInit.Value;
} }
public async ValueTask DisposeAsync() public async ValueTask DisposeAsync()