refactor: Rename EmailPorifler to MessagingService
This commit is contained in:
@@ -0,0 +1,217 @@
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using DigitalData.MessagingService.Application.Common.Interfaces;
|
||||
using DigitalData.MessagingService.Infrastructure.Messaging;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
using DigitalData.MessagingService.Application.Common.Events;
|
||||
|
||||
namespace DigitalData.MessagingService.Infrastructure.Queue;
|
||||
|
||||
/// <summary>
|
||||
/// RabbitMQ-based email queue implementation for outgoing emails.
|
||||
/// Provides message persistence, scalability, and reliability.
|
||||
/// Uses Lazy<T> initialization pattern to avoid blocking constructor.
|
||||
/// </summary>
|
||||
public sealed class OutgoingEmailQueue(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailQueue> Logger, IEmailService EmailService) : IOutgoingEmailQueue, IAsyncDisposable
|
||||
{
|
||||
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 stoppingToken = default)
|
||||
{
|
||||
Logger.LogInformation("Initializing RabbitMQ connection and queues...");
|
||||
|
||||
var factory = new ConnectionFactory
|
||||
{
|
||||
HostName = _config.HostName,
|
||||
Port = _config.Port,
|
||||
UserName = _config.UserName,
|
||||
Password = _config.Password,
|
||||
VirtualHost = _config.VirtualHost,
|
||||
AutomaticRecoveryEnabled = _config.AutomaticRecoveryEnabled,
|
||||
NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds)
|
||||
};
|
||||
|
||||
_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 _publishChannel.ExchangeDeclareAsync(exchange: _config.DlqExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: stoppingToken);
|
||||
|
||||
// Declare Dead Letter Queue (DLQ)
|
||||
await _publishChannel.QueueDeclareAsync(queue: _config.DlqQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: stoppingToken);
|
||||
|
||||
// Bind DLQ to DLQ exchange
|
||||
await _publishChannel.QueueBindAsync(queue: _config.DlqQueueName, exchange: _config.DlqExchangeName, routingKey: _config.DlqRoutingKey, cancellationToken: stoppingToken);
|
||||
|
||||
// Declare main exchange (Direct type for routing)
|
||||
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?>
|
||||
{
|
||||
{ "x-dead-letter-exchange", _config.DlqExchangeName },
|
||||
{ "x-dead-letter-routing-key", _config.DlqRoutingKey }
|
||||
};
|
||||
|
||||
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 _publishChannel.QueueBindAsync(queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey, cancellationToken: stoppingToken);
|
||||
|
||||
// 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);
|
||||
}
|
||||
|
||||
public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var json = JsonSerializer.Serialize(outgoingEmailEvent);
|
||||
var body = Encoding.UTF8.GetBytes(json);
|
||||
|
||||
var properties = new BasicProperties
|
||||
{
|
||||
Persistent = true, // Message persistence
|
||||
ContentType = "application/json",
|
||||
Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds())
|
||||
};
|
||||
|
||||
await _publishChannel!.BasicPublishAsync(
|
||||
exchange: _config.ExchangeName,
|
||||
routingKey: _config.RoutingKey,
|
||||
mandatory: false,
|
||||
basicProperties: properties,
|
||||
body: body,
|
||||
cancellationToken: cancellationToken);
|
||||
}
|
||||
|
||||
public async Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
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(IChannel consumeChannel, CancellationToken cancellationToken)
|
||||
{
|
||||
var consumer = new AsyncEventingBasicConsumer(consumeChannel);
|
||||
|
||||
consumer.ReceivedAsync += async (sender, args) =>
|
||||
{
|
||||
OutgoingEmailEvent? oMailEvent = null;
|
||||
try
|
||||
{
|
||||
var json = Encoding.UTF8.GetString(args.Body.ToArray());
|
||||
oMailEvent = JsonSerializer.Deserialize<OutgoingEmailEvent>(json);
|
||||
|
||||
if (oMailEvent is not null)
|
||||
{
|
||||
Logger.LogDebug("Received email message: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.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(
|
||||
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 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 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);
|
||||
|
||||
// TODO: Error Reporting Strategy
|
||||
// Option 1: Separate RabbitMQ Queue (emailprofiler.errors)
|
||||
// - Create EmailErrorReport entity { OutgoingEmailEventId, Exception, StackTrace, Timestamp, RetryAttempt }
|
||||
// - Publish to error queue: await _errorQueue.EnqueueAsync(errorReport)
|
||||
// - Separate worker processes error queue → Log to DB/File/External monitoring
|
||||
//
|
||||
// Option 2: Database Table (TBEMLP_ERROR_LOG)
|
||||
// - Columns: ERROR_ID, OUTBOX_ID, ERROR_MESSAGE, STACK_TRACE, ERROR_DATE
|
||||
// - Insert via IErrorLogRepository.CreateAsync(errorLog)
|
||||
//
|
||||
// Option 3: External Monitoring Service
|
||||
// - Sentry: SentrySdk.CaptureException(ex)
|
||||
// - Application Insights: _telemetryClient.TrackException(ex)
|
||||
// - Elasticsearch: _elasticClient.IndexDocument(errorLog)
|
||||
//
|
||||
// Recommended: Option 1 (RabbitMQ Error Queue) + Option 2 (DB persistence)
|
||||
// - Fast async error logging (non-blocking)
|
||||
// - Persistent storage for audit
|
||||
// - Real-time alerting via monitoring worker
|
||||
|
||||
// NO RETRY - All failures move directly to DLQ
|
||||
await consumeChannel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ
|
||||
}
|
||||
};
|
||||
|
||||
// Start consuming messages (event-driven, non-blocking)
|
||||
await consumeChannel.BasicConsumeAsync(
|
||||
queue: _config.QueueName,
|
||||
autoAck: false,
|
||||
consumer: consumer,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
Logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user