From fa9b4973b989ee7554d966896bb452804765d8f5 Mon Sep 17 00:00:00 2001 From: TekH Date: Wed, 5 Aug 2026 15:34:33 +0200 Subject: [PATCH] Introduce RabbitMQ consumer pool for parallel processing 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. --- .../DependencyInjection.cs | 2 +- .../Queue/SendingEmailConsumer.cs | 46 +++++++++++------ .../Queue/SendingEmailConsumerPool.cs | 51 +++++++++++++++++++ .../Services/Background/AsyncInitWorker.cs | 7 ++- .../Services/LimilabsEmailService.cs | 30 ++++++++--- .../RabbitMqConfiguration.cs | 6 +++ 6 files changed, 114 insertions(+), 28 deletions(-) create mode 100644 src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumerPool.cs diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs index acb1122..f0a9d56 100644 --- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs +++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs @@ -33,7 +33,7 @@ public static class DependencyInjection services.AddSingleton(); // --- Email Queue (RabbitMQ) --- - services.AddSingleton(); + services.AddSingleton(); services.AddMessagingServicePublisher(); // --- RabbitMQ Configuration --- diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumer.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumer.cs index 4101c4f..0763124 100644 --- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumer.cs +++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumer.cs @@ -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; /// -/// RabbitMQ-based email queue implementation for outgoing emails. -/// Provides message persistence, scalability, and reliability. -/// Uses Lazy 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 (competing consumers pattern). +/// Each instance owns exactly one channel — channels are not thread-safe and must not be shared. /// public sealed class SendingEmailConsumer : IAsyncDisposable { - private readonly RabbitMqConfiguration _config; + private readonly string _queueName; private readonly Lazy> _lazyChannel; @@ -25,15 +24,27 @@ public sealed class SendingEmailConsumer : IAsyncDisposable private readonly ILogger? _logger; - public SendingEmailConsumer(IOptions config, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory, ILogger? logger = null) + /// + /// 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. + /// + public Guid RuntimeId { get; } = Guid.NewGuid(); + + public SendingEmailConsumer(string queueName, IEmailService emailService, RabbitMqConnectionFactory cnnFactory, ILogger? 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); }); } /// - /// 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 . /// public async Task InitAsync() { diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumerPool.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumerPool.cs new file mode 100644 index 0000000..691b62c --- /dev/null +++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumerPool.cs @@ -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; + +/// +/// Manages a pool of 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. +/// +public sealed class SendingEmailConsumerPool : IAsyncDisposable +{ + private readonly List _consumers; + private readonly ILogger? _logger; + private readonly int _concurrency; + + public SendingEmailConsumerPool( + IOptions config, + IEmailService emailService, + RabbitMqConnectionFactory cnnFactory, + ILogger? logger = null, + ILogger? consumerLogger = null) + { + _logger = logger; + _concurrency = config.Value.ConsumerConcurrency; + + _consumers = [.. Enumerable + .Range(0, _concurrency) + .Select(_ => new SendingEmailConsumer(config.Value.QueueName, emailService, cnnFactory, consumerLogger))]; + } + + /// + /// Starts all consumers in parallel. Each consumer opens its own channel and begins listening. + /// + 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())); + } +} diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs index 05f2bd0..628638d 100644 --- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs +++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/AsyncInitWorker.cs @@ -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; /// -/// 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. /// -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); } diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/LimilabsEmailService.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/LimilabsEmailService.cs index dae58e1..17e1afb 100644 --- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/LimilabsEmailService.cs +++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/LimilabsEmailService.cs @@ -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(); + } +} \ No newline at end of file diff --git a/src/infrastructure/DigitalData.MessagingService.RabbitMQ/RabbitMqConfiguration.cs b/src/infrastructure/DigitalData.MessagingService.RabbitMQ/RabbitMqConfiguration.cs index 4b46f5b..3f21893 100644 --- a/src/infrastructure/DigitalData.MessagingService.RabbitMQ/RabbitMqConfiguration.cs +++ b/src/infrastructure/DigitalData.MessagingService.RabbitMQ/RabbitMqConfiguration.cs @@ -75,5 +75,11 @@ namespace DigitalData.MessagingService.RabbitMQ /// Routing key used to bind to . /// public string DlqRoutingKey { get; set; } = "email.outbox.dlq"; + + /// + /// Maximum number of email messages processed concurrently by the consumer. + /// Maps directly to RabbitMQ prefetchCount. Recommended: 3–5. + /// + public ushort ConsumerConcurrency { get; set; } = 5; } } \ No newline at end of file