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

Refactor RabbitMQ integration for lazy initialization

Browse Source 
Refactored `OutgoingEmailConsumer` and `OutgoingEmailPublisher` to use `Lazy<Task<IChannel>>` 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.
This commit is contained in:
tekh
2026-07-27 15:24:26 +02:00
parent a2373f242a
commit 2d7af80cd3
4 changed files with 76 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,37 +18,21 @@ namespace DigitalData.MessagingService.Infrastructure.Queue;
public sealed class OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailConsumer> Logger, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory) : IAsyncDisposable public sealed class OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailConsumer> Logger, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory) : IAsyncDisposable
{ {
private readonly RabbitMqConfiguration _config = config.Value; 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<Task<IChannel>> _lazyChannel = new(CnnFactory.CreateChannelAsync);
private readonly AsyncEventingBasicConsumer? consumer;
/// <summary> /// <summary>
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. /// 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. /// Called lazily on first use via EnsureInitializedAsync.
/// </summary> /// </summary>
public async Task InitAsync() public async Task InitAsync()
{ {
var connection = await CnnFactory.GetDefaultConnectionAsync(); var channel = await _lazyChannel.Value;
var stoppingToken = CnnFactory.CancellationToken; var consumer = new AsyncEventingBasicConsumer(channel);
// 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);
}
/// <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) => consumer.ReceivedAsync += async (sender, args) =>
{ {
@@ -68,12 +52,12 @@ public sealed class OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config
isHtml: oMailEvent.IsHtml); isHtml: oMailEvent.IsHtml);
// Acknowledge message after successful processing // Acknowledge message after successful processing
await consumeChannel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); await channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken);
} }
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 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) catch (Exception ex)
@@ -101,30 +85,27 @@ public sealed class OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config
// - Real-time alerting via monitoring worker // - Real-time alerting via monitoring worker
// NO RETRY - All failures move directly to DLQ // 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) // Start consuming messages (event-driven, non-blocking)
await consumeChannel.BasicConsumeAsync( await channel.BasicConsumeAsync(
queue: _config.QueueName, queue: _config.QueueName,
autoAck: false, autoAck: false,
consumer: consumer, consumer: consumer,
cancellationToken: cancellationToken); cancellationToken: CnnFactory.CancellationToken);
Logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName); Logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName);
} }
public async ValueTask DisposeAsync() public async ValueTask DisposeAsync()
{ {
await _consumerCts.CancelAsync(); var channel = await _lazyChannel.Value;
_consumerCts.Dispose(); if (channel is not null)
if (_consumeChannel is not null)
{ {
await _consumeChannel.CloseAsync(); await channel.CloseAsync();
await _consumeChannel.DisposeAsync(); await channel.DisposeAsync();
} }
} }
} }

63
src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs
View File

@@ -4,9 +4,7 @@ using DigitalData.MessagingService.Application.Common.Interfaces;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using RabbitMQ.Client; using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using DigitalData.MessagingService.Application.Common.Events; using DigitalData.MessagingService.Application.Common.Events;
using DevExpress.CodeParser;
using DigitalData.MessagingService.RabbitMQ; using DigitalData.MessagingService.RabbitMQ;
namespace DigitalData.MessagingService.Infrastructure.Queue; namespace DigitalData.MessagingService.Infrastructure.Queue;
@@ -16,37 +14,41 @@ 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 OutgoingEmailPublisher(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailPublisher> Logger, RabbitMqConnectionFactory CnnFactory) : IOutgoingEmailPublisher, IAsyncDisposable public sealed class OutgoingEmailPublisher : IOutgoingEmailPublisher, IAsyncDisposable
{ {
private readonly RabbitMqConfiguration _config = config.Value; private readonly RabbitMqConfiguration _config;
private IChannel? _publishChannel = null; // Dedicated channel for publishing private readonly ILogger<OutgoingEmailPublisher> _logger;
private readonly CancellationTokenSource _consumerCts = new(); // Independent lifetime from InitAsync token private readonly RabbitMqConnectionFactory _cnnFactory;
private readonly Lazy<Task<IChannel>> _lazyChannel;
public OutgoingEmailPublisher(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailPublisher> logger, RabbitMqConnectionFactory cnnFactory)
{
_config = config.Value;
_logger = logger;
_cnnFactory = cnnFactory;
_lazyChannel = new(InitChannelAsync);
}
/// <summary> /// <summary>
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync. /// Called lazily on first use via EnsureInitializedAsync.
/// </summary> /// </summary>
public async Task InitAsync() private async Task<IChannel> InitChannelAsync()
{ {
var connection = await CnnFactory.GetDefaultConnectionAsync(); var channel = await _cnnFactory.CreateChannelAsync();
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 // Topology declaration can use either channel; use publish channel here
// Declare Dead Letter Queue (DLQ) exchange // 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) // 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 // 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) // 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 // Declare main queue (durable for persistence) with DLQ arguments
var queueArgs = new Dictionary<string, object?> var queueArgs = new Dictionary<string, object?>
@@ -55,17 +57,14 @@ public sealed class OutgoingEmailPublisher(IOptions<RabbitMqConfiguration> confi
{ "x-dead-letter-routing-key", _config.DlqRoutingKey } { "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 // 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, _logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName);
// 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); return channel;
} }
public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default) public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default)
@@ -80,7 +79,9 @@ public sealed class OutgoingEmailPublisher(IOptions<RabbitMqConfiguration> confi
Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds())
}; };
await _publishChannel!.BasicPublishAsync( var channel = await _lazyChannel.Value;
await channel.BasicPublishAsync(
exchange: _config.ExchangeName, exchange: _config.ExchangeName,
routingKey: _config.RoutingKey, routingKey: _config.RoutingKey,
mandatory: false, mandatory: false,
@@ -91,19 +92,17 @@ public sealed class OutgoingEmailPublisher(IOptions<RabbitMqConfiguration> confi
public async Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default) public async Task<int> 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; return (int)queueInfo.MessageCount;
} }
public async ValueTask DisposeAsync() public async ValueTask DisposeAsync()
{ {
await _consumerCts.CancelAsync(); if (await _lazyChannel.Value is IChannel channel)
_consumerCts.Dispose();
if (_publishChannel is not null)
{ {
await _publishChannel.CloseAsync(); await channel.CloseAsync();
await _publishChannel.DisposeAsync(); await channel.DisposeAsync();
} }
} }
} }

11
src/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs
View File

@@ -1,8 +1,6 @@
using DigitalData.MessagingService.Application.Common.Interfaces; using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.Infrastructure.Queue; using DigitalData.MessagingService.Infrastructure.Queue;
using DigitalData.MessagingService.RabbitMQ;
using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
namespace DigitalData.MessagingService.Infrastructure.Services.Background; 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. /// 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. /// Email account configuration is resolved exclusively from application settings; no database access is performed.
/// </summary> /// </summary>
public class AsyncInitWorker(OutgoingEmailConsumer EmailConsumer, IOutgoingEmailPublisher EmailPublisher, RabbitMqConnectionFactory CnnFactory) : BackgroundService public class AsyncInitWorker(OutgoingEmailConsumer EmailConsumer) : BackgroundService
{ {
protected override async Task ExecuteAsync(CancellationToken stoppingToken) 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(); await EmailConsumer.InitAsync();
if (EmailPublisher is OutgoingEmailPublisher emailPublisher)
await emailPublisher.InitAsync();
} }
} }

58
src/DigitalData.MessagingService.RabbitMQ/RabbitMqConnectionFactory.cs
View File

@@ -1,6 +1,7 @@
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using RabbitMQ.Client; using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System; using System;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -9,6 +10,8 @@ namespace DigitalData.MessagingService.RabbitMQ
{ {
public sealed class RabbitMqConnectionFactory : IAsyncDisposable public sealed class RabbitMqConnectionFactory : IAsyncDisposable
{ {
private readonly CancellationTokenSource _consumerCts = new CancellationTokenSource();
private readonly RabbitMqConfiguration _config; private readonly RabbitMqConfiguration _config;
private readonly ILogger<RabbitMqConnectionFactory> private readonly ILogger<RabbitMqConnectionFactory>
@@ -19,23 +22,23 @@ namespace DigitalData.MessagingService.RabbitMQ
private readonly Lazy<Task<IConnection>> _lazyConnectionProvider; private readonly Lazy<Task<IConnection>> _lazyConnectionProvider;
private IConnection public CancellationToken CancellationToken => _consumerCts.Token;
#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 Task<IConnection> GetDefaultConnectionAsync() public Task<IConnection> GetDefaultConnectionAsync()
{ {
if (_cancellationToken != null) return _lazyConnectionProvider.Value;
return _lazyConnectionProvider.Value; }
else
throw new InvalidOperationException("RabbitMqConnectionFactory is not initialized. Call InitAsync() before using this method."); public async Task<IChannel> CreateChannelAsync()
{
var cnn = await GetDefaultConnectionAsync();
return await cnn.CreateChannelAsync(cancellationToken: CancellationToken);
}
public async Task<AsyncEventingBasicConsumer> CreateConsumerAsync()
{
var channel = await CreateChannelAsync();
return new AsyncEventingBasicConsumer(channel);
} }
public RabbitMqConnectionFactory(IOptions<RabbitMqConfiguration> config) public RabbitMqConnectionFactory(IOptions<RabbitMqConfiguration> config)
@@ -43,7 +46,6 @@ namespace DigitalData.MessagingService.RabbitMQ
_config = config.Value; _config = config.Value;
_lazyConnectionProvider = new Lazy<Task<IConnection>>(async () => _lazyConnectionProvider = new Lazy<Task<IConnection>>(async () =>
{ {
_cancellationToken?.ThrowIfCancellationRequested();
var factory = new ConnectionFactory var factory = new ConnectionFactory
{ {
HostName = _config.HostName, HostName = _config.HostName,
@@ -55,27 +57,23 @@ namespace DigitalData.MessagingService.RabbitMQ
NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds), NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds),
}; };
_connection = await factory.CreateConnectionAsync((CancellationToken)_cancellationToken return await factory.CreateConnectionAsync(CancellationToken);
#if nullable
!
#endif
);
return _connection;
}); });
} }
public async Task InitAsync(CancellationToken cancellationToken = default)
{
_cancellationToken = cancellationToken;
_ = await GetDefaultConnectionAsync();
}
public async ValueTask DisposeAsync() 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.CloseAsync();
await _connection.DisposeAsync(); await connection.DisposeAsync();
} }
} }
} }