From 2d7af80cd371a875f624cb5e58c930345ef0ddf3 Mon Sep 17 00:00:00 2001 From: TekH Date: Mon, 27 Jul 2026 15:24:26 +0200 Subject: [PATCH] Refactor RabbitMQ integration for lazy initialization Refactored `OutgoingEmailConsumer` and `OutgoingEmailPublisher` to use `Lazy>` for channel initialization, ensuring channels are created only when needed. Simplified initialization and disposal logic by centralizing channel management. Replaced `CancellationTokenSource` in both classes with the `CancellationToken` provided by `RabbitMqConnectionFactory`, centralizing token management. Updated methods to use the new lazy initialization pattern. Removed `RabbitMqConnectionFactory.InitAsync` and introduced `CreateChannelAsync` and `CreateConsumerAsync` methods for simplified channel and consumer creation. Managed cancellation tokens internally with a `CancellationTokenSource`. Simplified `AsyncInitWorker` by removing dependencies on `RabbitMqConnectionFactory` and `OutgoingEmailPublisher`. Removed redundant initialization logic. Cleaned up unused imports, improved logging consistency, and enhanced code readability and maintainability. --- .../Queue/OutgoingEmailConsumer.cs | 51 +++++---------- .../Queue/OutgoingEmailPublisher.cs | 63 +++++++++---------- .../Services/Background/AsyncInitWorker.cs | 11 +--- .../RabbitMqConnectionFactory.cs | 58 +++++++++-------- 4 files changed, 76 insertions(+), 107 deletions(-) diff --git a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs index b4daafe..89026cb 100644 --- a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs +++ b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs @@ -18,37 +18,21 @@ namespace DigitalData.MessagingService.Infrastructure.Queue; public sealed class OutgoingEmailConsumer(IOptions config, ILogger Logger, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory) : IAsyncDisposable { private readonly RabbitMqConfiguration _config = config.Value; - private IChannel? _consumeChannel = null; // Dedicated channel for consuming - private readonly CancellationTokenSource _consumerCts = new(); // Independent lifetime from InitAsync token + + private readonly Lazy> _lazyChannel = new(CnnFactory.CreateChannelAsync); + + private readonly AsyncEventingBasicConsumer? consumer; /// /// 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. /// public async Task InitAsync() { - var connection = await CnnFactory.GetDefaultConnectionAsync(); + var channel = await _lazyChannel.Value; - var stoppingToken = CnnFactory.CancellationToken; - - // Separate channels: IChannel is not thread-safe; one per role is best practice - _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); - await StartConsumerAsync(_consumeChannel, _consumerCts.Token); - - Logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName); - } - - /// - /// Start event-driven consumer that processes messages as they arrive - /// - private async Task StartConsumerAsync(IChannel consumeChannel, CancellationToken cancellationToken) - { - var consumer = new AsyncEventingBasicConsumer(consumeChannel); + var consumer = new AsyncEventingBasicConsumer(channel); consumer.ReceivedAsync += async (sender, args) => { @@ -68,12 +52,12 @@ public sealed class OutgoingEmailConsumer(IOptions config isHtml: oMailEvent.IsHtml); // Acknowledge message after successful processing - await consumeChannel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); + await channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); } 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 + await channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // Don't requeue invalid messages } } catch (Exception ex) @@ -101,30 +85,27 @@ public sealed class OutgoingEmailConsumer(IOptions config // - 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 + await channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ } }; // Start consuming messages (event-driven, non-blocking) - await consumeChannel.BasicConsumeAsync( + await channel.BasicConsumeAsync( queue: _config.QueueName, autoAck: false, consumer: consumer, - cancellationToken: cancellationToken); + cancellationToken: CnnFactory.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) + var channel = await _lazyChannel.Value; + if (channel is not null) { - await _consumeChannel.CloseAsync(); - await _consumeChannel.DisposeAsync(); + await channel.CloseAsync(); + await channel.DisposeAsync(); } } - } diff --git a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs index b31219a..d11545a 100644 --- a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs +++ b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs @@ -4,9 +4,7 @@ using DigitalData.MessagingService.Application.Common.Interfaces; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using RabbitMQ.Client; -using RabbitMQ.Client.Events; using DigitalData.MessagingService.Application.Common.Events; -using DevExpress.CodeParser; using DigitalData.MessagingService.RabbitMQ; namespace DigitalData.MessagingService.Infrastructure.Queue; @@ -16,37 +14,41 @@ namespace DigitalData.MessagingService.Infrastructure.Queue; /// 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 +public sealed class OutgoingEmailPublisher : 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 + private readonly RabbitMqConfiguration _config; + private readonly ILogger _logger; + private readonly RabbitMqConnectionFactory _cnnFactory; + private readonly Lazy> _lazyChannel; + + public OutgoingEmailPublisher(IOptions config, ILogger logger, RabbitMqConnectionFactory cnnFactory) + { + _config = config.Value; + _logger = logger; + _cnnFactory = cnnFactory; + _lazyChannel = new(InitChannelAsync); + } /// /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. /// Called lazily on first use via EnsureInitializedAsync. /// - public async Task InitAsync() + private async Task InitChannelAsync() { - var connection = await CnnFactory.GetDefaultConnectionAsync(); - - var stoppingToken = CnnFactory.CancellationToken; - - // Separate channels: IChannel is not thread-safe; one per role is best practice - _publishChannel = await connection.CreateChannelAsync(cancellationToken: stoppingToken); + var channel = await _cnnFactory.CreateChannelAsync(); // 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); + await channel.ExchangeDeclareAsync(exchange: _config.DlqExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: _cnnFactory.CancellationToken); // Declare Dead Letter Queue (DLQ) - await _publishChannel.QueueDeclareAsync(queue: _config.DlqQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: stoppingToken); + await channel.QueueDeclareAsync(queue: _config.DlqQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: _cnnFactory.CancellationToken); // Bind DLQ to DLQ exchange - await _publishChannel.QueueBindAsync(queue: _config.DlqQueueName, exchange: _config.DlqExchangeName, routingKey: _config.DlqRoutingKey, cancellationToken: stoppingToken); + await channel.QueueBindAsync(queue: _config.DlqQueueName, exchange: _config.DlqExchangeName, routingKey: _config.DlqRoutingKey, cancellationToken: _cnnFactory.CancellationToken); // Declare main exchange (Direct type for routing) - await _publishChannel.ExchangeDeclareAsync(exchange: _config.ExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: stoppingToken); + await channel.ExchangeDeclareAsync(exchange: _config.ExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: _cnnFactory.CancellationToken); // Declare main queue (durable for persistence) with DLQ arguments var queueArgs = new Dictionary @@ -55,17 +57,14 @@ public sealed class OutgoingEmailPublisher(IOptions confi { "x-dead-letter-routing-key", _config.DlqRoutingKey } }; - await _publishChannel.QueueDeclareAsync(queue: _config.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueArgs, cancellationToken: stoppingToken); + await channel.QueueDeclareAsync(queue: _config.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueArgs, cancellationToken: _cnnFactory.CancellationToken); // Bind main queue to exchange with routing key - await _publishChannel.QueueBindAsync(queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey, cancellationToken: stoppingToken); + await channel.QueueBindAsync(queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey, cancellationToken: _cnnFactory.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()); + _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); + return channel; } public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default) @@ -80,7 +79,9 @@ public sealed class OutgoingEmailPublisher(IOptions confi Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) }; - await _publishChannel!.BasicPublishAsync( + var channel = await _lazyChannel.Value; + + await channel.BasicPublishAsync( exchange: _config.ExchangeName, routingKey: _config.RoutingKey, mandatory: false, @@ -91,19 +92,17 @@ public sealed class OutgoingEmailPublisher(IOptions confi public async Task GetQueueDepthAsync(CancellationToken cancellationToken = default) { - var queueInfo = await _publishChannel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); + var channel = await _lazyChannel.Value; + var queueInfo = await channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); return (int)queueInfo.MessageCount; } public async ValueTask DisposeAsync() { - await _consumerCts.CancelAsync(); - _consumerCts.Dispose(); - - if (_publishChannel is not null) + if (await _lazyChannel.Value is IChannel channel) { - await _publishChannel.CloseAsync(); - await _publishChannel.DisposeAsync(); + await channel.CloseAsync(); + await channel.DisposeAsync(); } } } diff --git a/src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs b/src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs index 1b1cf00..175f1d7 100644 --- a/src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs +++ b/src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs @@ -1,8 +1,6 @@ using DigitalData.MessagingService.Application.Common.Interfaces; using DigitalData.MessagingService.Infrastructure.Queue; -using DigitalData.MessagingService.RabbitMQ; using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; namespace DigitalData.MessagingService.Infrastructure.Services.Background; @@ -11,17 +9,10 @@ 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(OutgoingEmailConsumer EmailConsumer, IOutgoingEmailPublisher EmailPublisher, RabbitMqConnectionFactory CnnFactory) : BackgroundService +public class AsyncInitWorker(OutgoingEmailConsumer EmailConsumer) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - // Initialize the RabbitMQ push-based consumer. This call is non-blocking; - // message processing is handled asynchronously via registered event callbacks. - await CnnFactory.InitAsync(stoppingToken); - await EmailConsumer.InitAsync(); - - if (EmailPublisher is OutgoingEmailPublisher emailPublisher) - await emailPublisher.InitAsync(); } } \ No newline at end of file diff --git a/src/DigitalData.MessagingService.RabbitMQ/RabbitMqConnectionFactory.cs b/src/DigitalData.MessagingService.RabbitMQ/RabbitMqConnectionFactory.cs index a5bbca6..ff5df33 100644 --- a/src/DigitalData.MessagingService.RabbitMQ/RabbitMqConnectionFactory.cs +++ b/src/DigitalData.MessagingService.RabbitMQ/RabbitMqConnectionFactory.cs @@ -1,6 +1,7 @@ using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using RabbitMQ.Client; +using RabbitMQ.Client.Events; using System; using System.Threading; using System.Threading.Tasks; @@ -9,6 +10,8 @@ namespace DigitalData.MessagingService.RabbitMQ { public sealed class RabbitMqConnectionFactory : IAsyncDisposable { + private readonly CancellationTokenSource _consumerCts = new CancellationTokenSource(); + private readonly RabbitMqConfiguration _config; private readonly ILogger @@ -19,23 +22,23 @@ namespace DigitalData.MessagingService.RabbitMQ private readonly Lazy> _lazyConnectionProvider; - private IConnection -#if nullable -? -#endif - _connection = null; - - private CancellationToken? _cancellationToken; - - public CancellationToken CancellationToken => _cancellationToken - ?? throw new InvalidOperationException("RabbitMqConnectionFactory is not initialized. Call InitAsync() before using this method."); + public CancellationToken CancellationToken => _consumerCts.Token; public Task GetDefaultConnectionAsync() { - if (_cancellationToken != null) - return _lazyConnectionProvider.Value; - else - throw new InvalidOperationException("RabbitMqConnectionFactory is not initialized. Call InitAsync() before using this method."); + return _lazyConnectionProvider.Value; + } + + public async Task CreateChannelAsync() + { + var cnn = await GetDefaultConnectionAsync(); + return await cnn.CreateChannelAsync(cancellationToken: CancellationToken); + } + + public async Task CreateConsumerAsync() + { + var channel = await CreateChannelAsync(); + return new AsyncEventingBasicConsumer(channel); } public RabbitMqConnectionFactory(IOptions config) @@ -43,7 +46,6 @@ namespace DigitalData.MessagingService.RabbitMQ _config = config.Value; _lazyConnectionProvider = new Lazy>(async () => { - _cancellationToken?.ThrowIfCancellationRequested(); var factory = new ConnectionFactory { HostName = _config.HostName, @@ -55,27 +57,23 @@ namespace DigitalData.MessagingService.RabbitMQ NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds), }; - _connection = await factory.CreateConnectionAsync((CancellationToken)_cancellationToken -#if nullable -! -#endif - ); - return _connection; + return await factory.CreateConnectionAsync(CancellationToken); }); } - public async Task InitAsync(CancellationToken cancellationToken = default) - { - _cancellationToken = cancellationToken; - _ = await GetDefaultConnectionAsync(); - } - public async ValueTask DisposeAsync() { - if (_connection != null) +#if NET + await _consumerCts.CancelAsync(); +#else + _consumerCts.Cancel(); +#endif + + var connection = await _lazyConnectionProvider.Value; + if (connection != null) { - await _connection.CloseAsync(); - await _connection.DisposeAsync(); + await connection.CloseAsync(); + await connection.DisposeAsync(); } } }