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

Introduce RabbitMQ consumer pool for parallel processing

Browse Source 
Enhanced RabbitMQ email processing by introducing a `SendingEmailConsumerPool` to enable the competing consumers pattern. Each consumer operates on its own channel, improving scalability and thread safety.

- Added `SendingEmailConsumerPool` to manage multiple consumers.
- Updated `DependencyInjection` to register the consumer pool.
- Refactored `SendingEmailConsumer` for better logging and error handling.
- Updated `AsyncInitWorker` to initialize the consumer pool.
- Added `ConsumerConcurrency` to RabbitMQ configuration.
- Improved error handling in `LimilabsEmailService` with detailed SMTP error messages.
This commit is contained in:
tekh
2026-08-05 15:34:33 +02:00
parent bd31bfe528
commit fa9b4973b9
6 changed files with 114 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs
View File

@@ -33,7 +33,7 @@ public static class DependencyInjection
services.AddSingleton<IEncryptionService, DataProtectionEncryptionService>();
// --- Email Queue (RabbitMQ) ---
services.AddSingleton<SendingEmailConsumer>();
services.AddSingleton<SendingEmailConsumerPool>();
services.AddMessagingServicePublisher();
// --- RabbitMQ Configuration ---

46
src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumer.cs
View File

@@ -4,20 +4,19 @@ using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.Abstraction;
using DigitalData.MessagingService.RabbitMQ;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.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.
/// A single RabbitMQ consumer that processes one email message at a time on its own dedicated channel.
/// Multiple instances run in parallel via <see cref="SendingEmailConsumerPool"/> (competing consumers pattern).
/// Each instance owns exactly one channel — channels are not thread-safe and must not be shared.
/// </summary>
public sealed class SendingEmailConsumer : IAsyncDisposable
{
private readonly RabbitMqConfiguration _config;
private readonly string _queueName;
private readonly Lazy<Task<IChannel>> _lazyChannel;
@@ -25,15 +24,27 @@ public sealed class SendingEmailConsumer : IAsyncDisposable
private readonly ILogger<SendingEmailConsumer>? _logger;
public SendingEmailConsumer(IOptions<RabbitMqConfiguration> config, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory, ILogger<SendingEmailConsumer>? logger = null)
/// <summary>
/// Transient identifier assigned to this consumer instance at runtime.
/// A new value is generated each time the application starts or a new consumer is created.
/// Use this to correlate log entries belonging to the same consumer session across competing instances.
/// </summary>
public Guid RuntimeId { get; } = Guid.NewGuid();
public SendingEmailConsumer(string queueName, IEmailService emailService, RabbitMqConnectionFactory cnnFactory, ILogger<SendingEmailConsumer>? logger = null)
{
_logger = logger;
_config = config.Value;
_queueName = queueName;
_lazyChannel = new(CnnFactory.CreateChannelAsync);
_lazyInit = new(async () => {
_lazyChannel = new(cnnFactory.CreateChannelAsync);
_lazyInit = new(async () =>
{
var channel = await _lazyChannel.Value;
// prefetchCount=1 ensures this consumer processes one message at a time before acking.
// Parallelism comes from running multiple consumer instances, not from within a single channel.
await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false);
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (sender, args) =>
@@ -47,10 +58,14 @@ public sealed class SendingEmailConsumer : IAsyncDisposable
if (oMailEvent is not null)
{
// Send email via SMTP (SMTP config is injected in IEmailService via IOptions)
await EmailService.SendEmailAsync(oMailEvent.Mail);
await emailService.SendEmailAsync(oMailEvent.Mail, args.CancellationToken);
// Acknowledge message after successful processing
await channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken);
logger?.LogDebug(
"Email successfully sent and acknowledged. RuntimeId={RuntimeId}, Queue={QueueName}, DeliveryTag={DeliveryTag}, To={Recipients}, Subject={Subject}, EventId={EventId}",
RuntimeId, _queueName, args.DeliveryTag, oMailEvent.Mail.Recipients, oMailEvent.Mail.Subject, oMailEvent.Id);
}
else
{
@@ -89,19 +104,18 @@ public sealed class SendingEmailConsumer : IAsyncDisposable
// Start consuming messages (event-driven, non-blocking)
await channel.BasicConsumeAsync(
queue: _config.QueueName,
queue: _queueName,
autoAck: false,
consumer: consumer,
cancellationToken: CnnFactory.CancellationToken);
cancellationToken: cnnFactory.CancellationToken);
logger?.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName);
logger?.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _queueName);
});
}
/// <summary>
/// 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.
/// Starts the consumer: opens a channel, sets QoS, and registers the event handler.
/// Called by <see cref="SendingEmailConsumerPool.InitAsync"/>.
/// </summary>
public async Task InitAsync()
{

51
src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumerPool.cs Normal file
View File

@@ -0,0 +1,51 @@
using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.RabbitMQ;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace DigitalData.MessagingService.Infrastructure.Queue;
/// <summary>
/// Manages a pool of <see cref="SendingEmailConsumer"/> instances that compete for messages
/// on the same RabbitMQ queue (competing consumers pattern).
/// Each consumer owns a dedicated channel, so they process messages fully in parallel
/// without any shared locking or synchronization primitives.
/// </summary>
public sealed class SendingEmailConsumerPool : IAsyncDisposable
{
private readonly List<SendingEmailConsumer> _consumers;
private readonly ILogger<SendingEmailConsumerPool>? _logger;
private readonly int _concurrency;
public SendingEmailConsumerPool(
IOptions<RabbitMqConfiguration> config,
IEmailService emailService,
RabbitMqConnectionFactory cnnFactory,
ILogger<SendingEmailConsumerPool>? logger = null,
ILogger<SendingEmailConsumer>? consumerLogger = null)
{
_logger = logger;
_concurrency = config.Value.ConsumerConcurrency;
_consumers = [.. Enumerable
.Range(0, _concurrency)
.Select(_ => new SendingEmailConsumer(config.Value.QueueName, emailService, cnnFactory, consumerLogger))];
}
/// <summary>
/// Starts all consumers in parallel. Each consumer opens its own channel and begins listening.
/// </summary>
public async Task InitAsync()
{
_logger?.LogInformation("Starting {Count} competing email consumers.", _concurrency);
await Task.WhenAll(_consumers.Select(c => c.InitAsync()));
_logger?.LogInformation("All {Count} email consumers started.", _concurrency);
}
public async ValueTask DisposeAsync()
{
await Task.WhenAll(_consumers.Select(async c => await c.DisposeAsync().AsTask()));
}
}

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

@@ -1,19 +1,18 @@
using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.Infrastructure.Queue;
using Microsoft.Extensions.Hosting;
namespace DigitalData.MessagingService.Infrastructure.Services.Background;
/// <summary>
/// A hosted background service responsible for initializing the outgoing email queue consumer.
/// A hosted background service responsible for initializing the competing email consumer pool.
/// 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(SendingEmailConsumer EmailConsumer) : BackgroundService
public class AsyncInitWorker(SendingEmailConsumerPool ConsumerPool) : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await EmailConsumer.InitAsync();
await ConsumerPool.InitAsync();
await Task.Delay(Timeout.Infinite, stoppingToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
}

30
src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/LimilabsEmailService.cs
View File

@@ -28,7 +28,7 @@ public class LimilabsEmailService(
public async Task SendEmailAsync(EmailContext context, CancellationToken cancellationToken = default)
{
using var smtp = new Smtp();
ISendMessageResult? result = null;
try
{
await ConnectAndAuthenticateSmtpAsync(smtp, context.Sender);
@@ -48,25 +48,24 @@ public class LimilabsEmailService(
var mail = builder.Create();
var result = await smtp.SendMessageAsync(mail, cancellationToken);
result = await smtp.SendMessageAsync(mail, cancellationToken);
if (result.Status != SendMessageStatus.Success)
{
throw new InvalidOperationException($"Failed to send email. Status: {result.Status}");
throw new InvalidOperationException($"Failed to send email. Status: {result.Status}. {ErrorMessageBuilder(result)}");
}
await smtp.CloseAsync(cancellationToken);
await Task.CompletedTask; // For async consistency
}
catch (Limilabs.Client.ServerException ex)
{
await smtp.CloseSafelyAsync();
throw new AuthenticationFailedException("SMTP authentication failed. Check credentials or OAuth2 configuration.", ex);
throw new AuthenticationFailedException($"SMTP authentication failed. Check credentials or OAuth2 configuration. {ErrorMessageBuilder(result)}", ex);
}
catch (Exception ex)
{
await smtp.CloseSafelyAsync();
throw new InvalidOperationException("Failed to send email via SMTP server.", ex);
throw new InvalidOperationException($"Failed to send email via SMTP server. {ErrorMessageBuilder(result)}", ex);
}
}
@@ -92,4 +91,21 @@ public class LimilabsEmailService(
await smtp.LoginAsync(smtpAccount.Username, password);
}
}
}
private static string ErrorMessageBuilder(ISendMessageResult? result = null)
{
if(result is null || result.GeneralErrors.Count == 0)
return string.Empty;
else if(result.GeneralErrors.Count == 1)
return $"Error: {result.GeneralErrors.FirstOrDefault()}";
var message = new StringBuilder("Errors:\n");
foreach (var error in result.GeneralErrors)
{
message.AppendLine($" • {error}");
}
return message.ToString();
}
}