Phase 3.1: Implement RabbitMqConsumer

[artcava]

📝 Description

Implements RabbitMQ message consumer for receiving and processing messages from queues as defined in issue #50.

Closes #50

🎯 Implementation

This PR implements:

1. RabbitMqConsumer (src/StarGate.Infrastructure/Messaging/RabbitMQ/RabbitMqConsumer.cs)

• Implements IMessageConsumer using AsyncEventingBasicConsumer for async message processing
• Manual acknowledgment via MessageContext.AcknowledgeAsync and MessageContext.RejectAsync delegates
• QoS configuration with configurable PrefetchCount (default: 10)
• Graceful shutdown with configurable grace period (default: 1000ms)
• Dedicated channel per queue for isolation
• Consumer event handlers (Registered, Unregistered, Shutdown) with structured logging
• Comprehensive error handling:
  • Deserialization/validation errors → reject to DLQ (no requeue)
  • OperationCanceledException → requeue for another worker
  • Unexpected errors → requeue for retry
  • AlreadyClosedException handling for closed channels
• Thread-safe with SemaphoreSlim for consumer lifecycle
• Implements IAsyncDisposable for proper resource cleanup

2. NullMessageConsumer (src/StarGate.Infrastructure/Messaging/NullMessageConsumer.cs)

• Null Object pattern implementation for when RabbitMQ is disabled
• Logs warnings and ignores consumption requests
• Maintains consistent API with real consumer

3. RabbitMqOptions extension

• Added PrefetchCount property (default: 10) for QoS configuration
• Added ShutdownGracePeriodMs property (default: 1000) for graceful shutdown

4. Dependency Injection update

• Registers IMessageConsumer as singleton
• Uses RabbitMqConsumer when RabbitMQ is enabled
• Uses NullMessageConsumer when RabbitMQ is disabled

✅ Acceptance Criteria

• RabbitMqConsumer implements IMessageConsumer interface
• Async message consumption with AsyncEventingBasicConsumer
• Manual acknowledgment based on MessageContext delegates (existing design)
• Message requeue on processing failure via RejectAsync(requeue: true)
• QoS configuration with PrefetchCount
• Graceful shutdown with configurable grace period
• Dedicated channel per queue
• Consumer event handlers (registered, unregistered, shutdown)
• Comprehensive error handling for deserialization, cancellation and unexpected errors
• NullMessageConsumer fallback when RabbitMQ disabled
• Proper resource disposal (IAsyncDisposable)
• Thread-safe consumer lifecycle management
• Dependency injection wiring

📝 Notes

Design Alignment

This implementation follows the existing project design where acknowledgment is handled via MessageContext delegates (AcknowledgeAsync/RejectAsync) rather than the MessageHandlingResult enum proposed in the original issue description. This maintains consistency with the current IMessageConsumer interface and MessageContext abstraction.

Key Differences from Issue Proposal

• Acknowledgment mechanism: Uses MessageContext delegates instead of returning MessageHandlingResult enum
• Queue naming: Derives queue name from message type using convention stargate.{typename}
• Single consumer: Current implementation supports one queue at a time (can be extended to multiple queues in future iterations)

Testing

Unit tests will be added in a follow-up PR. Integration tests are tracked in issue #35.

🔗 Related Issues

• Part of Phase 3: Message Broker - Sprint 3.1
• Depends on feat(phase-2.1): MongoDB Documents and Mapping Logic (#21) #29 (Message serialization)
• Depends on feat: Implement MongoProcessRepository (Issue #19) #30 (RabbitMqBroker)
• Integration tests in fix: Correct PolicyMapper BackoffStrategy parsing (Hotfix #2) #35