From 4a6af885de3170aef3c1f23199bde4c37309fcfa Mon Sep 17 00:00:00 2001 From: TekH Date: Thu, 23 Jul 2026 17:00:59 +0200 Subject: [PATCH] Refactor email queue interface and RabbitMQ handling Removed `DequeueAsync` from `IOutgoingEmailQueue` and added `GetQueueDepthAsync` to query the queue's message count. Updated RabbitMQ connection and channel creation methods to support `CancellationToken`. Removed `DequeueAsync` implementation from `OutgoingEmailQueue`, signaling a shift away from direct message consumption. These changes improve cancellation handling and simplify the queue's responsibilities. --- .../Common/Interfaces/IOutgoingEmailQueue.cs | 2 +- .../Queue/OutgoingEmailQueue.cs | 30 ++----------------- 2 files changed, 3 insertions(+), 29 deletions(-) diff --git a/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IOutgoingEmailQueue.cs b/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IOutgoingEmailQueue.cs index e1fec6c..fe9a1a2 100644 --- a/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IOutgoingEmailQueue.cs +++ b/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IOutgoingEmailQueue.cs @@ -9,7 +9,7 @@ namespace DigitalData.EmailProfiler.Application.Common.Interfaces; public interface IOutgoingEmailQueue { Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default); - Task DequeueAsync(CancellationToken cancellationToken = default); + Task GetQueueDepthAsync(CancellationToken cancellationToken = default); /// diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs b/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs index fcf374f..f4f7f50 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs @@ -52,8 +52,8 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds) }; - _connection = await factory.CreateConnectionAsync(); - _channel = await _connection.CreateChannelAsync(); + _connection = await factory.CreateConnectionAsync(cancellationToken); + _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken); // Declare Dead Letter Queue (DLQ) exchange await _channel.ExchangeDeclareAsync( @@ -108,7 +108,6 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable _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); @@ -130,31 +129,6 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable cancellationToken: cancellationToken); } - public async Task DequeueAsync(CancellationToken cancellationToken = default) - { - var result = await _channel.BasicGetAsync(_config.QueueName, false, cancellationToken); - - if (result == null) - return null; - - try - { - var json = Encoding.UTF8.GetString(result.Body.ToArray()); - var email = JsonSerializer.Deserialize(json); - - // Acknowledge message after successful deserialization - await _channel.BasicAckAsync(result.DeliveryTag, false, cancellationToken); - - return email; - } - catch - { - // Reject and requeue message on error - await _channel.BasicNackAsync(result.DeliveryTag, false, true, cancellationToken); - throw; - } - } - public async Task GetQueueDepthAsync(CancellationToken cancellationToken = default) { var queueInfo = await _channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);