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

Refactor email queue interface and RabbitMQ handling

Browse Source 
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.
This commit is contained in:
tekh
2026-07-23 17:00:59 +02:00
parent 55e5d689ad
commit 4a6af885de
2 changed files with 3 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/DigitalData.EmailProfiler.Application/Common/Interfaces/IOutgoingEmailQueue.cs
View File

@@ -9,7 +9,7 @@ namespace DigitalData.EmailProfiler.Application.Common.Interfaces;
public interface IOutgoingEmailQueue
{
Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default);
Task<OutgoingEmailEvent?> DequeueAsync(CancellationToken cancellationToken = default);
Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default);
/// <summary>

30
src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs
View File

@@ -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<OutgoingEmailEvent?> 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<OutgoingEmailEvent>(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<int> GetQueueDepthAsync(CancellationToken cancellationToken = default)
{
var queueInfo = await _channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);