Enhance consumer management with extended options and statistics

[mmalkhatib]

Summary by CodeRabbit

Release Notes

• New Features

  • Added consumer statistics and monitoring capabilities to query real-time metrics including message counts, processing rates, and consumer status
  • Added RabbitMQ management API configuration options for enhanced queue and consumer management

• Refactor

  • Improved internal consumer management with thread-safe operations and robust retry logic for enhanced stability and reliability

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

Caution

Review failed

The pull request is closed.

📝 Walkthrough

Walkthrough

The changes introduce consumer statistics reading via a new IConsumerReader interface and ConsumerCount record for querying RabbitMQ consumer metrics. Extended consumer interfaces (IConsumeExtended, IConsumeExtended<T>) enable per-consumer configuration options. Thread-safety improvements with topology synchronization locks are added to ConsumersService. RabbitMQ management API integration credentials are exposed in configuration.

Changes

Cohort / File(s)	Summary
Consumer Statistics Reading SW.Bus.RabbitMqExtensions/IConsumerReader.cs, SW.Bus/ConsumerReader.cs	New public IConsumerReader interface with methods to retrieve ConsumerCount statistics per-consumer, per-message, and globally. ConsumerReader implementation queries RabbitMQ management API with in-memory caching, aggregates queue/retry/failed counts, and handles graceful fallbacks when definitions unavailable.
Consumer Configuration Extensions SW.Bus.RabbitMqExtensions/IConsumeExtended.cs, SW.Bus.RabbitMqExtensions/ConsumerOptions.cs, SW.Bus.RabbitMqExtensions/QueueOptions.cs	New IConsumeExtended and IConsumeExtended<TMessage> interfaces enabling consumers to return custom ConsumerOptions. Added XML docs to ConsumerOptions. New QueueOptions class extends ConsumerOptions with RetryCount, RetryAfterSeconds, and computed ConsumerArgs property.
Consumer Discovery & Definition SW.Bus/ConsumerDiscovery.cs, SW.Bus/ConsumerDefinition.cs	ConsumerDiscovery now detects IConsumeExtended<T> and asynchronously retrieves per-consumer options via reflection. ConsumerDefinition.queueOptions made mutable; new UpdateConsumerProps() method added for option updates.
Bus Configuration & Registration SW.Bus/BusOptions.cs, SW.Bus/IServiceCollectionExtensions.cs, SW.Bus/SW.Bus.csproj	Four new properties added to BusOptions for RabbitMQ management credentials (ManagementUrl, ManagementUsername, ManagementPassword, VirtualHost). IServiceCollectionExtensions wires defaults from connection factory and registers IConsumerReader, ConsumerDiscovery, and memory cache. NuGet dependency EasyNetQ.Management.Client (3.0.1) added.
Consumer Service Improvements SW.Bus/ConsumersService.cs	Replaces Dictionary with thread-safe ConcurrentDictionary. Introduces SemaphoreSlim for topology synchronization. Refactors StartBusAsync with retry loop and backoff; adds idempotent binding guards. Expands RefreshConsumers to detect and apply consumer argument/priority changes, cancel/restart affected consumers. Improves error handling and logging granularity.
Configuration Files SW.Bus.SampleWeb/appsettings.json, SW.Bus.UnitTests/appsettings.json	RabbitMQ connection string replaced from placeholder #{rabbitMQConnection}# to concrete local URL amqp://guest:guest@localhost:5672.

Sequence Diagram(s)

sequenceDiagram
    actor Client
    participant Reader as ConsumerReader
    participant Discovery as ConsumerDiscovery
    participant Cache as Memory Cache
    participant RabbitMQ as RabbitMQ Mgmt API
    participant Definitions as Consumer Definitions

    Client->>Reader: GetConsumerCount(TConsumer, messageName)
    Reader->>Discovery: Load definitions for TConsumer
    Discovery->>Definitions: Resolve queue/message metadata
    Definitions-->>Discovery: Return ConsumerDefinition
    Discovery-->>Reader: Return definition or null
    
    alt Definition Found
        Reader->>Reader: Derive queue names (primary, retry, failed)
    else Definition Not Found
        Reader->>Reader: Derive queue names from convention
    end
    
    Reader->>Cache: Check cached queue info
    alt Cache Hit
        Cache-->>Reader: Return cached QueueInfo
    else Cache Miss
        Reader->>RabbitMQ: Query queue statistics
        RabbitMQ-->>Reader: Return queue data
        Reader->>Cache: Store with expiration
    end
    
    Reader->>Reader: Aggregate counts (ProcessingCount, QueueCount, RetryCount, etc.)
    Reader-->>Client: Return ConsumerCount with metrics

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Possibly related PRs

• Muhannad/dynamic consumer props #21 — Adds/modifies the same RabbitMQ extension types and consumer definition APIs (ConsumerOptions, IConsumeExtended, QueueOptions, ConsumerDefinition).

Poem

🐰 Through queues we hop with stats so neat,
Consumer counts in rhythm beat,
With locks held tight and cache so spry,
Extended options dance on high! ✨

✨ Finishing touches

• 📝 Generate docstrings

---

📜 Recent review details

Configuration used: defaults

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 2205a10 and e4ae583.

📒 Files selected for processing (13)

• SW.Bus.RabbitMqExtensions/ConsumerOptions.cs
• SW.Bus.RabbitMqExtensions/IConsumeExtended.cs
• SW.Bus.RabbitMqExtensions/IConsumerReader.cs
• SW.Bus.RabbitMqExtensions/QueueOptions.cs
• SW.Bus.SampleWeb/appsettings.json
• SW.Bus.UnitTests/appsettings.json
• SW.Bus/BusOptions.cs
• SW.Bus/ConsumerDefinition.cs
• SW.Bus/ConsumerDiscovery.cs
• SW.Bus/ConsumerReader.cs
• SW.Bus/ConsumersService.cs
• SW.Bus/IServiceCollectionExtensions.cs
• SW.Bus/SW.Bus.csproj

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}