refactor(infrastructure): Update RabbitMqCommandConsumer with improved error handling
This commit is contained in:
@@ -45,6 +45,34 @@ public class RabbitMqCommandConsumer(
|
|||||||
Connection = await factory.CreateConnectionAsync(stoppingToken);
|
Connection = await factory.CreateConnectionAsync(stoppingToken);
|
||||||
_channel = await Connection.CreateChannelAsync(cancellationToken: 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)
|
// Set prefetch count (process one message at a time)
|
||||||
await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, cancellationToken: stoppingToken);
|
await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false, cancellationToken: stoppingToken);
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user