diff --git a/src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailConsumer.cs b/src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailConsumer.cs new file mode 100644 index 0000000..227df77 --- /dev/null +++ b/src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailConsumer.cs @@ -0,0 +1,8 @@ +namespace DigitalData.MessagingService.Application.Common.Interfaces; + +/// +/// Email queue interface for outgoing emails. +/// +public interface IOutgoingEmailConsumer +{ +} diff --git a/src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailQueue.cs b/src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailPublisher.cs similarity index 92% rename from src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailQueue.cs rename to src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailPublisher.cs index 575cbc8..a964979 100644 --- a/src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailQueue.cs +++ b/src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailPublisher.cs @@ -6,7 +6,7 @@ namespace DigitalData.MessagingService.Application.Common.Interfaces; /// /// Email queue interface for outgoing emails. /// -public interface IOutgoingEmailQueue +public interface IOutgoingEmailPublisher { Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default); diff --git a/src/DigitalData.MessagingService.Application/EmailSending/Commands/SendEmailCommand.cs b/src/DigitalData.MessagingService.Application/EmailSending/Commands/SendEmailCommand.cs index 6f4571b..89e98bb 100644 --- a/src/DigitalData.MessagingService.Application/EmailSending/Commands/SendEmailCommand.cs +++ b/src/DigitalData.MessagingService.Application/EmailSending/Commands/SendEmailCommand.cs @@ -35,14 +35,14 @@ public record SendEmailCommand : IRequest /// Handler for SendEmailCommand /// Creates EmailOutbox entity via AutoMapper and enqueues to RabbitMQ /// -public class SendEmailCommandHandler(IOutgoingEmailQueue EmailQueue, IMapper Mapper) : IRequestHandler +public class SendEmailCommandHandler(IOutgoingEmailPublisher Publisher, IMapper Mapper) : IRequestHandler { public async Task Handle(SendEmailCommand request, CancellationToken cancellationToken) { var outgoingEmailEvent = Mapper.Map(request); // Enqueue to RabbitMQ - await EmailQueue.EnqueueAsync(outgoingEmailEvent, cancellationToken); + await Publisher.EnqueueAsync(outgoingEmailEvent, cancellationToken); return outgoingEmailEvent; } } diff --git a/src/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs b/src/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs index 0f29092..e421110 100644 --- a/src/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs +++ b/src/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs @@ -32,7 +32,9 @@ public static class DependencyInjection services.AddSingleton(); // --- Email Queue (RabbitMQ) --- - services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); // --- RabbitMQ Configuration --- services.Configure( diff --git a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailQueue.cs b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs similarity index 53% rename from src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailQueue.cs rename to src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs index d8a8c17..918f74e 100644 --- a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailQueue.cs +++ b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs @@ -15,11 +15,9 @@ namespace DigitalData.MessagingService.Infrastructure.Queue; /// Provides message persistence, scalability, and reliability. /// Uses Lazy initialization pattern to avoid blocking constructor. /// -public sealed class OutgoingEmailQueue(IOptions config, ILogger Logger, IEmailService EmailService) : IOutgoingEmailQueue, IAsyncDisposable +public sealed class OutgoingEmailConsumer(IOptions config, ILogger Logger, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory) : IOutgoingEmailConsumer, 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 @@ -27,87 +25,23 @@ public sealed class OutgoingEmailQueue(IOptions config, I /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. /// Called lazily on first use via EnsureInitializedAsync. /// - public async Task InitAsync(CancellationToken stoppingToken = default) + public async Task InitAsync() { - 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) - }; + var connection = await CnnFactory.GetConnectionAsync(); - _connection = await factory.CreateConnectionAsync(stoppingToken); + var stoppingToken = CnnFactory.CancellationToken; // 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 - { - { "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); + _consumeChannel = await connection.CreateChannelAsync(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()); + 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 GetQueueDepthAsync(CancellationToken cancellationToken = default) - { - var queueInfo = await _publishChannel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); - return (int)queueInfo.MessageCount; - } /// /// Start event-driven consumer that processes messages as they arrive @@ -126,11 +60,6 @@ public sealed class OutgoingEmailQueue(IOptions config, I 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, @@ -138,12 +67,8 @@ public sealed class OutgoingEmailQueue(IOptions config, I 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 { @@ -200,18 +125,6 @@ public sealed class OutgoingEmailQueue(IOptions config, I 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(); - } } } diff --git a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs new file mode 100644 index 0000000..9578388 --- /dev/null +++ b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs @@ -0,0 +1,109 @@ +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; +using DevExpress.CodeParser; + +namespace DigitalData.MessagingService.Infrastructure.Queue; + +/// +/// RabbitMQ-based email queue implementation for outgoing emails. +/// Provides message persistence, scalability, and reliability. +/// Uses Lazy initialization pattern to avoid blocking constructor. +/// +public sealed class OutgoingEmailPublisher(IOptions config, ILogger Logger, RabbitMqConnectionFactory CnnFactory) : IOutgoingEmailPublisher, IAsyncDisposable +{ + private readonly RabbitMqConfiguration _config = config.Value; + private IChannel? _publishChannel = null; // Dedicated channel for publishing + 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() + { + var connection = await CnnFactory.GetConnectionAsync(); + + var stoppingToken = CnnFactory.CancellationToken; + + // Separate channels: IChannel is not thread-safe; one per role is best practice + _publishChannel = 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 + { + { "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()); + + 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 GetQueueDepthAsync(CancellationToken cancellationToken = default) + { + var queueInfo = await _publishChannel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); + return (int)queueInfo.MessageCount; + } + + public async ValueTask DisposeAsync() + { + await _consumerCts.CancelAsync(); + _consumerCts.Dispose(); + + if (_publishChannel is not null) + { + await _publishChannel.CloseAsync(); + await _publishChannel.DisposeAsync(); + } + } +} diff --git a/src/DigitalData.MessagingService.Infrastructure/Queue/RabbitMqConnectionFactory.cs b/src/DigitalData.MessagingService.Infrastructure/Queue/RabbitMqConnectionFactory.cs new file mode 100644 index 0000000..1a3cae2 --- /dev/null +++ b/src/DigitalData.MessagingService.Infrastructure/Queue/RabbitMqConnectionFactory.cs @@ -0,0 +1,67 @@ +using DigitalData.MessagingService.Infrastructure.Messaging; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; + +namespace DigitalData.MessagingService.Infrastructure.Queue; + +public sealed class RabbitMqConnectionFactory : IAsyncDisposable +{ + private readonly RabbitMqConfiguration _config; + + private readonly ILogger? _logger; + + private readonly Lazy> _lazyConnectionProvider; + + private IConnection? _connection = null; + + private CancellationToken? _cancellationToken; + + public CancellationToken CancellationToken => _cancellationToken + ?? throw new InvalidOperationException("RabbitMqConnectionFactory is not initialized. Call InitAsync() before using this method."); + + public Task GetConnectionAsync() + { + if (_cancellationToken is not null) + return _lazyConnectionProvider.Value; + else + throw new InvalidOperationException("RabbitMqConnectionFactory is not initialized. Call InitAsync() before using this method."); + } + + public RabbitMqConnectionFactory(IOptions config) + { + _config = config.Value; + _lazyConnectionProvider = new (async () => + { + _cancellationToken?.ThrowIfCancellationRequested(); + 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((CancellationToken)_cancellationToken!); + return _connection; + }); + } + + public async Task InitAsync(CancellationToken cancellationToken = default) + { + _cancellationToken = cancellationToken; + _ = await GetConnectionAsync(); + } + + public async ValueTask DisposeAsync() + { + if (_connection is not null) + { + await _connection.CloseAsync(); + await _connection.DisposeAsync(); + } + } +} diff --git a/src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs b/src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs index 14084b5..276e7fd 100644 --- a/src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs +++ b/src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs @@ -10,23 +10,18 @@ namespace DigitalData.MessagingService.Infrastructure.Services.Background; /// Leverages a push-based, event-driven RabbitMQ consumer to eliminate polling overhead. /// Email account configuration is resolved exclusively from application settings; no database access is performed. /// -public class AsyncInitWorker(IOutgoingEmailQueue EmailQueue, ILogger Logger) : BackgroundService +public class AsyncInitWorker(IOutgoingEmailConsumer EmailConsumer, IOutgoingEmailPublisher EmailPublisher, RabbitMqConnectionFactory CnnFactory) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - Logger.LogInformation("Outgoing email queue worker is starting. Initializing event-driven RabbitMQ consumer."); + // Initialize the RabbitMQ push-based consumer. This call is non-blocking; + // message processing is handled asynchronously via registered event callbacks. + await CnnFactory.InitAsync(stoppingToken); - try - { - // Initialize the RabbitMQ push-based consumer. This call is non-blocking; - // message processing is handled asynchronously via registered event callbacks. - if (EmailQueue is OutgoingEmailQueue outgoingEmailQueue) - await outgoingEmailQueue.InitAsync(stoppingToken); - } - catch (Exception ex) - { - Logger.LogError(ex, "A critical error occurred while initializing the outgoing email queue consumer. The worker cannot proceed."); - throw; - } + if (EmailConsumer is OutgoingEmailConsumer outgoingEmailConsumer) + await outgoingEmailConsumer.InitAsync(); + + if (EmailPublisher is OutgoingEmailPublisher outgoingEmailPublisher) + await outgoingEmailPublisher.InitAsync(); } } \ No newline at end of file