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

Refactor email processing and RabbitMQ initialization

Browse Source 
Centralized email processing logic in `RabbitMqEmailQueue` by moving it from `EmailSenderWorker`. Updated `IEmailQueue` to replace `StartConsumerAsync` with `InitAsync`, shifting to an initialization-based model for RabbitMQ.

Refactored `RabbitMqEmailQueue` to handle email processing inline, including deserialization, logging, and sending emails via `IEmailService`. Enhanced error handling with detailed logging for failures. Removed lazy initialization (`Lazy<Task>`) in favor of explicit initialization via `InitAsync`.

Simplified `EmailSenderWorker` by removing `ProcessEmailAsync` and its dependency on `IEmailService`. Updated it to call `EmailQueue.InitAsync` for initialization.

Improved logging and error handling for better visibility into email processing and failure scenarios. Updated RabbitMQ acknowledgment and rejection logic to use `args.CancellationToken`.
This commit is contained in:
tekh
2026-07-23 16:47:14 +02:00
parent 55feaed361
commit fbd6c0c521
3 changed files with 36 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/DigitalData.EmailProfiler.API/Workers/EmailSenderWorker.cs
View File

@@ -1,7 +1,5 @@
using DigitalData.EmailProfiler.API.Configurations;
using DigitalData.EmailProfiler.Application.Common.Events;
using DigitalData.EmailProfiler.Application.Common.Interfaces;
using DigitalData.EmailProfiler.Application.EmailSending.Commands;
using Microsoft.Extensions.Options;
namespace DigitalData.EmailProfiler.API.Workers;
@@ -13,7 +11,6 @@ namespace DigitalData.EmailProfiler.API.Workers;
/// </summary>
public class EmailSenderWorker(
IEmailQueue EmailQueue,
IEmailService EmailService,
ILogger<EmailSenderWorker> Logger,
IOptions<EmailSenderWorkerConfiguration> configuration) : BackgroundService
{
@@ -32,7 +29,7 @@ public class EmailSenderWorker(
try
{
// Start RabbitMQ consumer (event-driven, non-blocking)
await EmailQueue.StartConsumerAsync(ProcessEmailAsync, stoppingToken);
await EmailQueue.InitAsync(stoppingToken);
// Keep worker alive until cancellation
await Task.Delay(Timeout.Infinite, stoppingToken);
@@ -49,35 +46,4 @@ public class EmailSenderWorker(
Logger.LogInformation("EmailSenderWorker stopped");
}
/// <summary>
/// Process single email message (callback from RabbitMQ consumer)
/// NO database operations - only send email and log result
/// </summary>
private async Task ProcessEmailAsync(OutgoingEmailEvent outgoingEmailEvent)
{
try
{
Logger.LogInformation("Processing outgoing email: To={To}, Subject={Subject}",
outgoingEmailEvent.Recipient, outgoingEmailEvent.Subject);
// Send email via SMTP (SMTP config is injected in IEmailService via IOptions)
await EmailService.SendEmailAsync(
outgoingEmailEvent.Recipient,
outgoingEmailEvent.Subject,
outgoingEmailEvent.Body,
isHtml: outgoingEmailEvent.IsHtml);
Logger.LogInformation("Email sent successfully: To={To}, Subject={Subject}",
outgoingEmailEvent.Recipient, outgoingEmailEvent.Subject);
}
catch (Exception ex)
{
Logger.LogError(ex, "Failed to send email: To={To}, Subject={Subject}. Moving to DLQ.",
outgoingEmailEvent.Recipient, outgoingEmailEvent.Subject);
// Re-throw to trigger NACK in RabbitMQ consumer (requeue=false → DLQ)
throw;
}
}
}

7
src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs
View File

@@ -13,9 +13,8 @@ public interface IEmailQueue
Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Start event-driven consumer that calls callback when mail received
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync.
/// </summary>
/// <param name="onMailReceived">Callback to process received mail</param>
/// <param name="cancellationToken">Cancellation token</param>
Task StartConsumerAsync(Func<OutgoingEmailEvent, Task> onMailReceived, CancellationToken cancellationToken = default);
Task InitAsync(CancellationToken cancellationToken = default);
}

73
src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs
View File

@@ -22,26 +22,22 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
private readonly RabbitMqConfiguration _config;
private IConnection? _connection;
private IChannel _channel;
// Lazy<T> ensures InitAsync is called only ONCE (thread-safe)
private readonly Lazy<Task> _initializationTask;
private readonly IEmailService _emailService;
#pragma warning disable CS8618 // channel and connection are initialized in InitAsync, not in constructor
public RabbitMqEmailQueue(IOptions<RabbitMqConfiguration> config, ILogger<RabbitMqEmailQueue> logger)
public RabbitMqEmailQueue(IOptions<RabbitMqConfiguration> config, ILogger<RabbitMqEmailQueue> logger, IEmailService emailService)
#pragma warning restore CS8618
{
_logger = logger;
_config = config.Value;
// Lazy initialization - NOT executed until first access (non-blocking constructor)
_initializationTask = new Lazy<Task>(InitAsync);
_emailService = emailService;
}
/// <summary>
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync.
/// </summary>
private async Task InitAsync()
public async Task InitAsync(CancellationToken cancellationToken = default)
{
_logger.LogInformation("Initializing RabbitMQ connection and queues...");
@@ -106,23 +102,15 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
queue: _config.QueueName,
exchange: _config.ExchangeName,
routingKey: _config.RoutingKey);
await StartConsumerAsync(cancellationToken);
_logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName);
}
/// <summary>
/// Ensure RabbitMQ is initialized before any operation.
/// Thread-safe and guarantees single initialization via Lazy<T>.
/// </summary>
private async Task EnsureInitializedAsync()
{
await _initializationTask.Value;
}
public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default)
{
await EnsureInitializedAsync(); // Initialize on first call
var json = JsonSerializer.Serialize(outgoingEmailEvent);
var body = Encoding.UTF8.GetBytes(json);
@@ -133,7 +121,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds())
};
await _channel!.BasicPublishAsync(
await _channel.BasicPublishAsync(
exchange: _config.ExchangeName,
routingKey: _config.RoutingKey,
mandatory: false,
@@ -144,9 +132,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
public async Task<OutgoingEmailEvent?> DequeueAsync(CancellationToken cancellationToken = default)
{
await EnsureInitializedAsync(); // Initialize on first call
var result = await _channel!.BasicGetAsync(_config.QueueName, false, cancellationToken);
var result = await _channel.BasicGetAsync(_config.QueueName, false, cancellationToken);
if (result == null)
return null;
@@ -171,50 +157,55 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
public async Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default)
{
await EnsureInitializedAsync(); // Initialize on first call
var queueInfo = await _channel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);
var queueInfo = await _channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);
return (int)queueInfo.MessageCount;
}
/// <summary>
/// Start event-driven consumer that processes messages as they arrive
/// </summary>
public async Task StartConsumerAsync(
Func<OutgoingEmailEvent, Task> onMailReceived,
CancellationToken cancellationToken = default)
private async Task StartConsumerAsync(CancellationToken cancellationToken)
{
await EnsureInitializedAsync(); // Initialize on first call
var consumer = new AsyncEventingBasicConsumer(_channel!);
var consumer = new AsyncEventingBasicConsumer(_channel);
consumer.ReceivedAsync += async (sender, args) =>
{
OutgoingEmailEvent? oMailEvent = null;
try
{
var json = Encoding.UTF8.GetString(args.Body.ToArray());
var email = JsonSerializer.Deserialize<OutgoingEmailEvent>(json);
oMailEvent = JsonSerializer.Deserialize<OutgoingEmailEvent>(json);
if (email != null)
if (oMailEvent is not null)
{
_logger.LogDebug("Received email message: To={To}, Subject={Subject}", email.Recipient, email.Subject);
_logger.LogDebug("Received email message: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject);
// Process message via callback
await onMailReceived(email);
_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(
oMailEvent.Recipient,
oMailEvent.Subject,
oMailEvent.Body,
isHtml: oMailEvent.IsHtml);
_logger.LogInformation("Email sent successfully: To={To}, Subject={Subject}",
oMailEvent.Recipient, oMailEvent.Subject);
// Acknowledge message after successful processing
await _channel.BasicAckAsync(args.DeliveryTag, false, cancellationToken);
await _channel.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, cancellationToken); // Don't requeue invalid messages
await _channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // Don't requeue invalid messages
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to process email message: DeliveryTag={DeliveryTag}. Moving to DLQ (NO retry).", 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)
@@ -237,7 +228,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable
// - Real-time alerting via monitoring worker
// NO RETRY - All failures move directly to DLQ
await _channel.BasicNackAsync(args.DeliveryTag, false, false, cancellationToken); // requeue=false → DLQ
await _channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ
}
};