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

Refactor RabbitMQ email queue into publisher/consumer

Browse Source 
Refactored the RabbitMQ-based email queue system by splitting
`OutgoingEmailQueue` into `OutgoingEmailPublisher` and
`OutgoingEmailConsumer` to separate publishing and consuming
responsibilities.

- Introduced `IOutgoingEmailConsumer` and renamed
  `IOutgoingEmailQueue` to `IOutgoingEmailPublisher` for clarity.
- Updated `SendEmailCommandHandler` to use the new publisher
  abstraction.
- Added `RabbitMqConnectionFactory` to centralize RabbitMQ
  connection management.
- Updated dependency injection to register new services.
- Simplified RabbitMQ initialization logic by delegating it to
  `RabbitMqConnectionFactory`.
- Enhanced logging for better observability.
- Improved modularity and maintainability by separating concerns
  between message publishing and consuming.
This commit is contained in:
tekh
2026-07-24 19:01:42 +02:00
parent 5e587da957
commit 1617d4ab43
8 changed files with 205 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailConsumer.cs Normal file
View File

@@ -0,0 +1,8 @@
namespace DigitalData.MessagingService.Application.Common.Interfaces;
/// <summary>
/// Email queue interface for outgoing emails.
/// </summary>
public interface IOutgoingEmailConsumer
{
}

2
src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailQueue.cs → src/DigitalData.MessagingService.Application/Common/Interfaces/IOutgoingEmailPublisher.cs
View File

@@ -6,7 +6,7 @@ namespace DigitalData.MessagingService.Application.Common.Interfaces;
/// <summary>
/// Email queue interface for outgoing emails.
/// </summary>
public interface IOutgoingEmailQueue
public interface IOutgoingEmailPublisher
{
Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default);

4
src/DigitalData.MessagingService.Application/EmailSending/Commands/SendEmailCommand.cs
View File

@@ -35,14 +35,14 @@ public record SendEmailCommand : IRequest<OutgoingEmailEvent>
/// Handler for SendEmailCommand
/// Creates EmailOutbox entity via AutoMapper and enqueues to RabbitMQ
/// </summary>
public class SendEmailCommandHandler(IOutgoingEmailQueue EmailQueue, IMapper Mapper) : IRequestHandler<SendEmailCommand, OutgoingEmailEvent>
public class SendEmailCommandHandler(IOutgoingEmailPublisher Publisher, IMapper Mapper) : IRequestHandler<SendEmailCommand, OutgoingEmailEvent>
{
public async Task<OutgoingEmailEvent> Handle(SendEmailCommand request, CancellationToken cancellationToken)
{
var outgoingEmailEvent = Mapper.Map<OutgoingEmailEvent>(request);
// Enqueue to RabbitMQ
await EmailQueue.EnqueueAsync(outgoingEmailEvent, cancellationToken);
await Publisher.EnqueueAsync(outgoingEmailEvent, cancellationToken);
return outgoingEmailEvent;
}
}

4
src/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs
View File

@@ -32,7 +32,9 @@ public static class DependencyInjection
services.AddSingleton<IEncryptionService, DataProtectionEncryptionService>();
// --- Email Queue (RabbitMQ) ---
services.AddSingleton<IOutgoingEmailQueue, OutgoingEmailQueue>();
services.AddSingleton<IOutgoingEmailConsumer, OutgoingEmailConsumer>();
services.AddSingleton<IOutgoingEmailPublisher, OutgoingEmailPublisher>();
services.AddSingleton<RabbitMqConnectionFactory>();
// --- RabbitMQ Configuration ---
services.Configure<RabbitMqConfiguration>(

99
src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailQueue.cs → src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs
View File

@@ -15,11 +15,9 @@ namespace DigitalData.MessagingService.Infrastructure.Queue;
/// 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
public sealed class OutgoingEmailConsumer(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailConsumer> 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,88 +25,24 @@ public sealed class OutgoingEmailQueue(IOptions<RabbitMqConfiguration> config, I
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync.
/// </summary>
public async Task InitAsync(CancellationToken stoppingToken = default)
public async Task InitAsync()
{
Logger.LogInformation("Initializing RabbitMQ connection and queues...");
var connection = await CnnFactory.GetConnectionAsync();
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);
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<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);
_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<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>
@@ -126,11 +60,6 @@ public sealed class OutgoingEmailQueue(IOptions<RabbitMqConfiguration> 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<RabbitMqConfiguration> 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<RabbitMqConfiguration> 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();
}
}
}

109
src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailPublisher.cs Normal file
View File

@@ -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;
/// <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 OutgoingEmailPublisher(IOptions<RabbitMqConfiguration> config, ILogger<OutgoingEmailPublisher> 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
/// <summary>
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync.
/// </summary>
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<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());
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;
}
public async ValueTask DisposeAsync()
{
await _consumerCts.CancelAsync();
_consumerCts.Dispose();
if (_publishChannel is not null)
{
await _publishChannel.CloseAsync();
await _publishChannel.DisposeAsync();
}
}
}

67
src/DigitalData.MessagingService.Infrastructure/Queue/RabbitMqConnectionFactory.cs Normal file
View File

@@ -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<RabbitMqConnectionFactory>? _logger;
private readonly Lazy<Task<IConnection>> _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<IConnection> 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<RabbitMqConfiguration> 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();
}
}
}

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

@@ -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.
/// </summary>
public class AsyncInitWorker(IOutgoingEmailQueue EmailQueue, ILogger<AsyncInitWorker> 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();
}
}