From 0adc74e19fdc9aa7a5a8e39b5afa14c5941db582 Mon Sep 17 00:00:00 2001 From: TekH Date: Thu, 23 Jul 2026 11:20:04 +0200 Subject: [PATCH] refactor(infrastructure): Update RabbitMqCommandConsumer with improved error handling --- .../Messaging/RabbitMqCommandConsumer.cs | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs index d541710..96e3d13 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/Messaging/RabbitMqCommandConsumer.cs @@ -45,6 +45,34 @@ public class RabbitMqCommandConsumer( Connection = await factory.CreateConnectionAsync(stoppingToken); _channel = await Connection.CreateChannelAsync(cancellationToken: stoppingToken); + // Declare exchange (Direct type for routing) + await _channel.ExchangeDeclareAsync( + exchange: Config.ExchangeName, + type: ExchangeType.Direct, + durable: true, + autoDelete: false, + cancellationToken: stoppingToken); + + // Declare queue (durable for persistence) + await _channel.QueueDeclareAsync( + queue: Config.QueueName, + durable: true, + exclusive: false, + autoDelete: false, + arguments: null, + cancellationToken: stoppingToken); + + // Bind queue to exchange with routing key + await _channel.QueueBindAsync( + queue: Config.QueueName, + exchange: Config.ExchangeName, + routingKey: Config.RoutingKey, + cancellationToken: stoppingToken); + + Logger?.LogInformation( + "RabbitMQ initialized: Exchange={Exchange}, Queue={Queue}, RoutingKey={RoutingKey}", + Config.ExchangeName, Config.QueueName, Config.RoutingKey); + // Set prefetch count (process one message at a time) await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, cancellationToken: stoppingToken);