rust-rabbit 0.3.1

# RustRabbit 🐰


[![Rust](https://github.com/nghiaphamln/rust-rabbit/workflows/CI/badge.svg)](https://github.com/nghiaphamln/rust-rabbit/actions)
[![Crates.io](https://img.shields.io/crates/v/rust-rabbit.svg)](https://crates.io/crates/rust-rabbit)
[![Documentation](https://docs.rs/rust-rabbit/badge.svg)](https://docs.rs/rust-rabbit)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

A **high-performance, production-ready** RabbitMQ client library for Rust with **zero-configuration** simplicity and enterprise-grade features. Built for reliability, observability, and developer happiness.

## 🚀 **Quick Start - One Line Magic!**


```rust
use rust_rabbit::{
    config::RabbitConfig,
    connection::ConnectionManager,
    consumer::{Consumer, ConsumerOptions},
    retry::RetryPolicy,
};

#[tokio::main]

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. Connection
    let config = RabbitConfig::builder()
        .connection_string("amqp://user:pass@localhost:5672/vhost")
        .build();
    let connection = ConnectionManager::new(config).await?;

    // 2. Consumer with retry (1 line!)
    let options = ConsumerOptions {
        auto_ack: false,
        retry_policy: Some(RetryPolicy::fast()),
        ..Default::default()
    };

    // 3. Start consuming
    let consumer = Consumer::new(connection, options).await?;
    
    consumer.consume("queue_name", |delivery| async move {
        match your_processing_logic(&delivery.data).await {
            Ok(_) => delivery.ack(Default::default()).await?,
            Err(e) if is_retryable(&e) => delivery.nack(Default::default()).await?,
            Err(_) => delivery.reject(Default::default()).await?,
        }
        Ok(())
    }).await?;

    Ok(())
}
```

**What `RetryPolicy::fast()` creates automatically:**
- ✅ **5 retries**: 200ms → 300ms → 450ms → 675ms → 1s (capped at 10s)
- ✅ **Dead Letter Queue**: Automatic DLX/DLQ setup for failed messages
- ✅ **Backoff + Jitter**: Intelligent delay with randomization
- ✅ **Production Ready**: Optimal settings for most use cases

## 📦 **Installation**


```toml
[dependencies]
rust-rabbit = "0.3.0"
tokio = { version = "1.0", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }
```

## 💡 **Core Features**


### 🎯 **Smart Automation**

- **One-Line Setup**: `RetryPolicy::fast()` configures everything
- **Auto Infrastructure**: Queues, exchanges, and bindings created automatically  
- **Intelligent Defaults**: Production-ready settings without configuration
- **Dead Letter Automation**: Automatic failure recovery and monitoring

### 🔄 **Advanced Retry System**

- **Multiple Presets**: Fast, slow, linear, and custom patterns
- **Exponential Backoff**: Smart delay calculations with jitter
- **Delayed Message Exchange**: RabbitMQ x-delayed-message integration
- **Dead Letter Integration**: Seamless failure handling

### 🏗️ **Enterprise Messaging Patterns**

- **Request-Response**: RPC-style messaging with correlation IDs and timeouts
- **Saga Pattern**: Distributed transaction coordination with compensation actions
- **Event Sourcing**: CQRS implementation with event store and aggregate management
- **Message Deduplication**: Multiple strategies for duplicate message detection
- **Priority Queues**: Configurable priority-based message processing

### 🔍 **Production Observability**

- **Prometheus Metrics**: Comprehensive metrics for throughput, latency, errors
- **Health Monitoring**: Real-time connection health with auto-recovery
- **Circuit Breaker**: Automatic failure detection and graceful degradation
- **Structured Logging**: Distributed tracing with correlation IDs

### 🛡️ **Reliability & Performance**

- **Connection Pooling**: Automatic connection management with health monitoring
- **Graceful Shutdown**: Multi-phase shutdown with signal handling
- **Error Recovery**: Comprehensive error handling with recovery strategies
- **Type Safety**: Strongly typed message handling with serde integration
- **Minutes Retry Preset**: One-line setup for business-critical operations
- **Intelligent Defaults**: Production-ready settings out of the box
- **Dead Letter Handling**: Automatic failure recovery and monitoring

### 🔄 **Intelligent Retry Patterns**

```rust
// Quick presets for common scenarios
RetryPolicy::fast()               // 1s, 2s, 4s, 8s, 16s (transient failures)
RetryPolicy::slow()               // 10s, 20s, 40s, 80s, 160s (resource-heavy)
RetryPolicy::aggressive()         // 15 retries with exponential backoff
RetryPolicy::minutes_exponential() // 1min, 2min, 4min, 8min, 16min (business-critical)

// Custom builder
RetryPolicy::builder()
    .max_retries(5)
    .initial_delay(Duration::from_secs(30))
    .backoff_multiplier(1.5)
    .jitter(0.2)  // 20% randomization prevents thundering herd
    .dead_letter_exchange("failed.orders")
    .build()
```

### 🏗️ **Enterprise Messaging Patterns** *(Phase 2 - NEW)*

- **Request-Response**: RPC-style messaging with correlation IDs and timeouts
- **Saga Pattern**: Distributed transaction coordination with compensation
- **Event Sourcing**: CQRS implementation with event store and snapshots
- **Message Deduplication**: Multiple strategies for duplicate detection
- **Priority Queues**: Configurable priority-based message processing

### 🔍 **Production Observability**

- **Prometheus Metrics**: Throughput, latency, error rates, queue depths
- **Health Monitoring**: Real-time connection health with auto-recovery
- **Circuit Breaker**: Automatic failure detection and graceful degradation
- **Structured Logging**: Distributed tracing with correlation IDs

## 📋 **Usage Examples**


## ⚡ **Retry Patterns & Setup**


### **Quick Retry Presets**


| Preset | Retries | Initial Delay | Max Delay | Backoff | Use Case |
|--------|---------|---------------|-----------|---------|----------|
| `RetryPolicy::fast()` | 5 | 200ms | 10s | 1.5x | API calls, DB queries |
| `RetryPolicy::slow()` | 3 | 1s | 1min | 2.0x | Heavy processing |
| `RetryPolicy::linear(500ms, 3)` | 3 | 500ms | 500ms | 1.0x | Fixed intervals |

### **Consumer Setup Patterns**


```rust
// 🔥 SUPER FAST - Copy/paste setup:
let config = RabbitConfig::builder()
    .connection_string("amqp://user:pass@localhost:5672/vhost")
    .build();
let connection = ConnectionManager::new(config).await?;
let options = ConsumerOptions {
    auto_ack: false,
    retry_policy: Some(RetryPolicy::fast()),
    prefetch_count: Some(10),
    ..Default::default()
};
let consumer = Consumer::new(connection, options).await?;

// ⚡ Custom with builder
let retry = RetryPolicy::builder()
    .fast_preset()          // Use preset as base
    .max_retries(3)         // Override retry count
    .build();

// 🛠 Ultra custom
let retry = RetryPolicy::builder()
    .max_retries(5)
    .initial_delay(Duration::from_millis(100))
    .backoff_multiplier(2.0)
    .jitter(0.1)
    .dead_letter_exchange("my.dlx")
    .build();
```

### **Message Handling Pattern**


```rust
consumer.consume("queue_name", |delivery| async move {
    match process_message(&delivery.data).await {
        Ok(_) => {
            // ✅ Success -> ACK
            delivery.ack(Default::default()).await?;
        }
        Err(e) if should_retry(&e) => {
            // ⚠️ Retryable error -> NACK (will retry)
            delivery.nack(Default::default()).await?;
        }
        Err(_) => {
            // ❌ Fatal error -> REJECT (send to DLQ)
            delivery.reject(Default::default()).await?;
        }
    }
    Ok(())
}).await?;

fn should_retry(error: &MyError) -> bool {
    match error {
        MyError::NetworkTimeout => true,     // Retry
        MyError::ApiRateLimit => true,       // Retry
        MyError::TempUnavailable => true,    // Retry
        MyError::InvalidData => false,       // Don't retry
        MyError::Unauthorized => false,      // Don't retry
    }
}
```

## � **Consumer Retry Configuration & Message Handling**


### **📋 Consumer Setup with Retry Policy**


```rust
use rust_rabbit::{
    consumer::{Consumer, ConsumerOptions, MessageHandler, MessageResult, MessageContext},
    retry::RetryPolicy,
    connection::ConnectionManager,
    config::RabbitConfig,
};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};

// 1. Define your message type
#[derive(Debug, Deserialize, Serialize)]

struct OrderMessage {
    order_id: String,
    customer_id: String,
    total: f64,
}

// 2. Create your message handler
struct OrderHandler;

#[async_trait]

impl MessageHandler<OrderMessage> for OrderHandler {
    async fn handle(&self, message: OrderMessage, context: MessageContext) -> MessageResult {
        info!("Processing order: {}", message.order_id);
        
        match process_order(&message).await {
            Ok(_) => {
                info!("✅ Order {} processed successfully", message.order_id);
                MessageResult::Ack  // Success -> Acknowledge message
            }
            Err(e) => {
                match classify_error(&e) {
                    ErrorType::Retryable => {
                        warn!("⚠️ Retryable error for order {}: {}", message.order_id, e);
                        MessageResult::Retry  // Will trigger retry mechanism
                    }
                    ErrorType::Fatal => {
                        error!("❌ Fatal error for order {}: {}", message.order_id, e);
                        MessageResult::Reject  // Send to dead letter queue
                    }
                }
            }
        }
    }
}

// 3. Setup consumer with retry policy
#[tokio::main]

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = RabbitConfig::builder()
        .connection_string("amqp://admin:password@localhost:5672/")
        .build();
    
    let connection_manager = ConnectionManager::new(config).await?;
    
    // Configure retry policy
    let retry_policy = RetryPolicy::builder()
        .max_retries(3)                              // Max 3 retry attempts
        .initial_delay(Duration::from_millis(500))   // Start with 500ms delay
        .max_delay(Duration::from_secs(30))          // Cap at 30 seconds
        .backoff_multiplier(2.0)                     // Double delay each retry
        .jitter(0.1)                                 // Add 10% randomization
        .dead_letter_exchange("orders.dlx".to_string())  // Failed messages go here
        .dead_letter_queue("orders.dlq".to_string())     // Dead letter queue
        .build();
    
    // Setup consumer with retry configuration
    let consumer_options = ConsumerOptions::builder("orders.processing")
        .auto_declare_queue()        // Create queue if not exists
        .auto_declare_exchange()     // Create exchange if not exists  
        .retry_policy(retry_policy)  // Enable retry mechanism
        .prefetch_count(10)          // Process up to 10 messages concurrently
        .build();
    
    let consumer = Consumer::new(connection_manager, consumer_options).await?;
    let handler = std::sync::Arc::new(OrderHandler);
    
    // Start consuming with retry support
    consumer.consume(handler).await?;
    
    Ok(())
}
```

### **🎯 MessageResult Types - Action Guide**


| MessageResult | Action | Description | Use Case |
|---------------|--------|-------------|----------|
| `MessageResult::Ack` | ✅ **Acknowledge** | Message processed successfully | Successful processing |
| `MessageResult::Retry` | 🔄 **Retry** | Retry message with delay | Temporary failures (network, rate limits) |
| `MessageResult::Reject` | ❌ **Dead Letter** | Send to DLQ, don't retry | Permanent failures (invalid data) |
| `MessageResult::Requeue` | 🔃 **Requeue** | Put back in queue immediately | Temporary resource unavailable |

### **⚠️ Error Classification Strategy**


```rust
#[derive(Debug)]

enum ErrorType {
    Retryable,  // Can retry
    Fatal,      // Don't retry
}

fn classify_error(error: &ProcessingError) -> ErrorType {
    match error {
        // 🔄 RETRYABLE - Temporary issues
        ProcessingError::NetworkTimeout => ErrorType::Retryable,
        ProcessingError::ApiRateLimit => ErrorType::Retryable,
        ProcessingError::DatabaseConnectionLost => ErrorType::Retryable,
        ProcessingError::TemporaryUnavailable => ErrorType::Retryable,
        ProcessingError::ConcurrencyLimit => ErrorType::Retryable,
        
        // ❌ FATAL - Permanent issues
        ProcessingError::InvalidData => ErrorType::Fatal,
        ProcessingError::AuthenticationFailed => ErrorType::Fatal,
        ProcessingError::PermissionDenied => ErrorType::Fatal,
        ProcessingError::RecordNotFound => ErrorType::Fatal,
        ProcessingError::ValidationFailed(_) => ErrorType::Fatal,
    }
}

async fn process_order(message: &OrderMessage) -> Result<(), ProcessingError> {
    // Your business logic here
    validate_order(message)?;      // Can throw ValidationFailed (Fatal)
    call_payment_api(message).await?;  // Can throw NetworkTimeout (Retryable)  
    update_inventory(message).await?;  // Can throw DatabaseConnectionLost (Retryable)
    send_confirmation(message).await?; // Can throw ApiRateLimit (Retryable)
    
    Ok(())
}
```

### **🛠️ Advanced Handler Patterns**


#### **Pattern 1: Conditional Retry with Context**

```rust
#[async_trait]

impl MessageHandler<OrderMessage> for OrderHandler {
    async fn handle(&self, message: OrderMessage, context: MessageContext) -> MessageResult {
        // Check retry count from context
        let retry_count = context.headers
            .get("x-retry-count")
            .and_then(|v| v.as_long_int())
            .unwrap_or(0);
            
        if retry_count >= 2 {
            warn!("Order {} failed after {} retries, sending to manual review", 
                  message.order_id, retry_count);
            return MessageResult::Reject;  // Send to DLQ for manual review
        }
        
        match process_order(&message).await {
            Ok(_) => MessageResult::Ack,
            Err(e) if is_worth_retrying(&e, retry_count) => MessageResult::Retry,
            Err(_) => MessageResult::Reject,
        }
    }
}
```

#### **Pattern 2: Circuit Breaker with Handler**

```rust
use std::sync::atomic::{AtomicU32, Ordering};

struct OrderHandlerWithCircuitBreaker {
    failure_count: AtomicU32,
    circuit_open: AtomicBool,
}

#[async_trait]

impl MessageHandler<OrderMessage> for OrderHandlerWithCircuitBreaker {
    async fn handle(&self, message: OrderMessage, _context: MessageContext) -> MessageResult {
        // Check circuit breaker
        if self.circuit_open.load(Ordering::Relaxed) {
            warn!("Circuit breaker open, rejecting message");
            return MessageResult::Requeue;  // Try again later
        }
        
        match process_order(&message).await {
            Ok(_) => {
                self.failure_count.store(0, Ordering::Relaxed);  // Reset on success
                MessageResult::Ack
            }
            Err(e) => {
                let failures = self.failure_count.fetch_add(1, Ordering::Relaxed);
                if failures >= 5 {
                    self.circuit_open.store(true, Ordering::Relaxed);
                    warn!("Circuit breaker triggered after {} failures", failures);
                }
                
                if is_retryable(&e) {
                    MessageResult::Retry
                } else {
                    MessageResult::Reject
                }
            }
        }
    }
}
```

### **📊 Retry Policy Examples by Use Case**


#### **Fast API Calls** (Network hiccups)

```rust
let api_retry = RetryPolicy::builder()
    .max_retries(5)
    .initial_delay(Duration::from_millis(100))
    .max_delay(Duration::from_secs(5))
    .backoff_multiplier(1.5)
    .jitter(0.2)
    .build();
```

#### **Database Operations** (Connection issues)

```rust
let db_retry = RetryPolicy::builder()
    .max_retries(3)
    .initial_delay(Duration::from_millis(500))
    .max_delay(Duration::from_secs(30))
    .backoff_multiplier(2.0)
    .jitter(0.1)
    .build();
```

#### **Business Critical** (Financial transactions)

```rust
let critical_retry = RetryPolicy::builder()
    .max_retries(3)
    .initial_delay(Duration::from_secs(30))   // Wait longer initially
    .max_delay(Duration::from_secs(300))      // Max 5 minutes
    .backoff_multiplier(1.2)                  // Gentle backoff
    .jitter(0.0)                              // No randomization for predictability
    .dead_letter_exchange("financial.dlx".to_string())
    .dead_letter_queue("financial.manual.review".to_string())
    .build();
```

### **🔍 Debugging Tips**


#### **Log Retry Information**

```rust
#[async_trait]

impl MessageHandler<OrderMessage> for OrderHandler {
    async fn handle(&self, message: OrderMessage, context: MessageContext) -> MessageResult {
        let retry_count = context.get_retry_count();
        let original_timestamp = context.get_original_timestamp();
        
        info!("Processing order {} (retry: {}, age: {:?})", 
              message.order_id, retry_count, 
              Utc::now() - original_timestamp);
        
        // Your processing logic...
        MessageResult::Ack
    }
}
```

#### **Monitor Retry Queue**

```bash
# Check retry exchange bindings

sudo rabbitmqctl list_bindings | grep retry

# Monitor message flow

sudo rabbitmqctl list_queues name messages

# Check dead letter queue

sudo rabbitmqctl list_queues | grep dlq
```

## 🐛 **Prefetch Count Debugging**


**Common Issue**: `prefetch_count` not working?

### ❌ **WRONG** (prefetch_count ignored):

```rust
ConsumerOptions {
    auto_ack: true,         // ← Wrong! Messages are ACKed immediately
    prefetch_count: Some(5), // → No effect
    ..Default::default()
}
```

## �🐛 **Prefetch Count Debugging**


**Common Issue**: `prefetch_count` not working?

### ❌ **WRONG** (prefetch_count ignored):

```rust
ConsumerOptions {
    auto_ack: true,         // ← Wrong! Messages are ACKed immediately
    prefetch_count: Some(5), // → No effect
    ..Default::default()
}
```

### ✅ **CORRECT** (prefetch_count works):

```rust
ConsumerOptions {
    auto_ack: false,        // ← Correct! Messages must be ACKed manually
    prefetch_count: Some(5), // → Limit 5 unACKed messages
    ..Default::default()
}
```

**Why?** 
- `prefetch_count` only works when there are "unACKed" messages
- `auto_ack: true` → Messages ACKed immediately → No backpressure
- `auto_ack: false` → Messages wait for manual ACK → QoS limits work

### **Debug Commands**

```bash
# Compile check

cargo check

# Run with RabbitMQ

cargo run --example fast_consumer_template

# Test prefetch behavior

cargo run --example prefetch_verification
```

### **Smart Publisher with Auto-Declare**


```rust
use rust_rabbit::publisher::{Publisher, PublishOptions};

let publisher = rabbit.publisher();

// Auto-declare exchange when publishing
let options = PublishOptions::builder()
    .auto_declare_exchange()
    .durable()
    .build();

publisher.publish_to_exchange(
    "orders.processing",  // exchange (auto-created)
    "orders.processing",  // routing key  
    &order_message,
    Some(options)
).await?;
```

### **Advanced Retry Configuration**


```rust
// Business-critical with custom settings
let custom_retry = RetryPolicy::builder()
    .max_retries(3)
    .initial_delay(Duration::from_secs(30))
    .backoff_multiplier(1.5)
    .jitter(0.2)  // 20% randomization
    .dead_letter_exchange("failed.orders.dlx")
    .dead_letter_queue("failed.orders.dlq")
    .build();

// Use with consumer
let options = ConsumerOptions::builder("orders.processing")
    .auto_declare_queue()
    .auto_declare_exchange()
    .retry_policy(custom_retry)
    .prefetch_count(1)    // Reliable processing
    .manual_ack()         // Explicit acknowledgment
    .build();
```

## 🏗️ **Advanced Patterns** *(Phase 2 - NEW)*


### **Request-Response (RPC)**


```rust
use rust_rabbit::patterns::request_response::*;

// Server side
let server = RequestResponseServer::new(rabbit.clone(), "calc_queue".to_string());
server.handle_requests(|req: CalculateRequest| async move {
    Ok(CalculateResponse { result: req.x + req.y })
}).await?;

// Client side
let client = RequestResponseClient::new(rabbit, "calc_queue".to_string());
let response: CalculateResponse = client
    .send_request(&CalculateRequest { x: 5, y: 3 })
    .with_timeout(Duration::from_secs(30))
    .await?;
```

### **Saga Pattern (Distributed Transactions)**


```rust
use rust_rabbit::patterns::saga::*;

// Define compensation logic
async fn reserve_inventory(order_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    println!("✅ Reserved inventory for order {}", order_id);
    Ok(())
}

async fn compensate_inventory(order_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    println!("🔄 Released inventory for order {}", order_id);
    Ok(())
}

// Execute saga
let mut coordinator = SagaCoordinator::new(rabbit);
let mut saga = SagaInstance::new("order-saga-123".to_string());

saga.add_step(
    "reserve_inventory",
    |data| Box::pin(reserve_inventory(&data)),
    |data| Box::pin(compensate_inventory(&data))
);

match coordinator.execute_saga(saga, "order-456".to_string()).await {
    Ok(_) => println!("✅ Saga completed successfully"),
    Err(e) => println!("❌ Saga failed, compensation completed: {}", e),
}
```

### **Event Sourcing (CQRS)**


```rust
use rust_rabbit::patterns::event_sourcing::*;

#[derive(Debug, Clone, Serialize, Deserialize)]

struct BankAccount {
    id: AggregateId,
    sequence: EventSequence,
    balance: f64,
}

impl AggregateRoot for BankAccount {
    fn apply_event(&mut self, event: &DomainEvent) -> Result<()> {
        match event.event_type.as_str() {
            "MoneyDeposited" => {
                let amount: f64 = serde_json::from_slice(&event.event_data)?;
                self.balance += amount;
            }
            "MoneyWithdrawn" => {
                let amount: f64 = serde_json::from_slice(&event.event_data)?;
                self.balance -= amount;
            }
            _ => {}
        }
        Ok(())
    }
}

// Usage
let event_store = Arc::new(InMemoryEventStore::new());
let repository = EventSourcingRepository::<BankAccount>::new(event_store);

let mut account = BankAccount::new(AggregateId::new());
account.deposit(100.0)?; // Generates MoneyDeposited event
repository.save(&mut account).await?;
```

## 🔍 **Production Features**


### **Health Monitoring**


```rust
use rust_rabbit::health::HealthChecker;

let health_checker = HealthChecker::new(connection_manager.clone());

match health_checker.check_health().await {
    Ok(status) => println!("Connection healthy: {:?}", status),
    Err(e) => println!("Connection issues: {}", e),
}
```

use rust_rabbit::metrics::PrometheusMetrics;

let metrics = PrometheusMetrics::new();

// Metrics are automatically collected:
// - rust_rabbit_messages_published_total
// - rust_rabbit_messages_consumed_total  
// - rust_rabbit_message_processing_duration_seconds
// - rust_rabbit_connection_health
// - rust_rabbit_queue_depth

// Expose metrics endpoint
warp::serve(metrics.metrics_handler())
    .run(([0, 0, 0, 0], 9090))
    .await;
```

## 📁 **Examples**


The library includes comprehensive examples for all features:

### **Core Examples**

- **`consumer_example.rs`** - Basic consumer setup and message handling
- **`publisher_example.rs`** - Publishing messages with different options
- **`fast_consumer_template.rs`** - Production-ready consumer template

### **Retry & Error Handling**

- **`quick_retry_consumer.rs`** - All retry preset demonstrations
- **`retry_example.rs`** - Custom retry logic implementation
- **`retry_policy_presets.rs`** - Retry policy patterns

### **Advanced Patterns**

- **`event_sourcing_example.rs`** - CQRS and event sourcing implementation
- **`saga_example.rs`** - Distributed transaction coordination
- **`phase2_patterns_example.rs`** - Enterprise messaging patterns

### **Production Features**

- **`health_monitoring_example.rs`** - Health checks and monitoring
- **`metrics_example.rs`** - Prometheus metrics integration
- **`comprehensive_demo.rs`** - Full-featured production example

### **Integration**

- **`actix_web_api_example.rs`** - Web API integration
- **`builder_pattern_example.rs`** - Configuration patterns

Run any example:
```bash
cargo run --example consumer_example
cargo run --example fast_consumer_template
```

## 🧪 **Testing**


### **Integration Testing with Docker**


RustRabbit supports real RabbitMQ integration testing:

```bash
# Start RabbitMQ with required plugins

docker-compose -f docker-compose.test.yml up -d

# Run integration tests  

cargo test --test integration_example -- --test-threads=1

# Or use make

make test-integration
```

### **Unit Tests**

```bash
# Run unit tests

cargo test

# Run with coverage

cargo test --all-features

# Test specific modules

cargo test consumer::tests
cargo test retry::tests
```

## 🚨 **Common Mistakes & Best Practices**


### ❌ **Common Mistakes**


1. **Wrong prefetch_count setup**:
   ```rust
   auto_ack: true,  // ← Wrong! prefetch_count won't work
   ```

2. **Not handling retry errors**:
   ```rust
   // ❌ Not distinguishing retryable vs non-retryable errors
   Err(_) => delivery.nack(Default::default()).await?,
   ```

3. **Missing manual ACK**:
   ```rust
   // ❌ Forgot to ACK message
   // Message will be stuck in unACK'd state
   ```

### ✅ **Best Practices**


1. **Always use auto_ack: false for production**:
   ```rust
   ConsumerOptions {
       auto_ack: false,  // Required for retry and backpressure
       retry_policy: Some(RetryPolicy::fast()),
       ..Default::default()
   }
   ```

2. **Implement smart retry logic**:
   ```rust
   match error {
       ApiError::Timeout => delivery.nack(Default::default()).await?, // Retry
       ApiError::Unauthorized => delivery.reject(Default::default()).await?, // DLQ
   }
   ```

3. **Set appropriate prefetch_count**:
   ```rust
   prefetch_count: Some(cpu_cores * 2), // For I/O bound
   prefetch_count: Some(cpu_cores),     // For CPU bound
   prefetch_count: Some(1),             // For memory-heavy tasks
   ```

4. **Always handle graceful shutdown**:
   ```rust
   tokio::select! {
       _ = consumer.consume("queue", handler) => {}
       _ = tokio::signal::ctrl_c() => {
           println!("Shutting down gracefully...");
       }
   }
   ```

## 🔧 **Configuration**


### **Connection Configuration**


```rust
let config = RabbitConfig::builder()
    .connection_string("amqp://user:pass@localhost:5672/vhost")
    .connection_timeout(Duration::from_secs(30))
    .heartbeat(Duration::from_secs(10))
    .retry_config(RetryConfig::default())
    .health_check_interval(Duration::from_secs(30))
    .pool_config(PoolConfig::new(5, 10)) // min 5, max 10 connections
    .build();
```

### **Consumer Configuration**


```rust
ConsumerOptions {
    auto_ack: false,                        // Manual ACK for reliability
    prefetch_count: Some(10),               // QoS prefetch limit
    retry_policy: Some(RetryPolicy::fast()), // Retry configuration
    concurrent_limit: Some(50),             // Max concurrent messages
    ..Default::default()
}
```

## 🚀 **Performance Tips**


- **prefetch_count**: Set to `CPU cores × 2-5` for I/O bound tasks
- **concurrent_limit**: 50-200 for I/O bound, fewer for CPU bound
- **Connection pooling**: Use 5-10 connections per application
- **DLQ monitoring**: Always monitor dead letter queues
- **Metrics**: Enable Prometheus for production monitoring

## 🤝 **Contributing**


We welcome contributions! Please see our [Contributing Guide](CONTRIBUTING.md) for details.

1. Fork the repository
2. Create a feature branch: `git checkout -b my-feature`
3. Make changes and add tests
4. Run tests: `cargo test`
5. Submit a pull request

## 📄 **License**


This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

## 🔗 **Links**


- **Documentation**: [docs.rs/rust-rabbit](https://docs.rs/rust-rabbit)
- **Crates.io**: [crates.io/crates/rust-rabbit](https://crates.io/crates/rust-rabbit)
- **GitHub**: [github.com/nghiaphamln/rust-rabbit](https://github.com/nghiaphamln/rust-rabbit)
- **Issues**: [github.com/nghiaphamln/rust-rabbit/issues](https://github.com/nghiaphamln/rust-rabbit/issues)

---

**Made with ❤️ for the Rust community**
use rust_rabbit::metrics::RustRabbitMetrics;

let metrics = RustRabbitMetrics::new()?;
let rabbit = RustRabbit::with_metrics(config, metrics.clone()).await?;

// Automatic metrics collection:
// - Message throughput (published/consumed per second)
// - Processing latency (P50, P90, P99)
// - Error rates (failed messages, connection errors)
// - Queue depths (pending messages)
// - Connection health (active, reconnections)
```

### **Circuit Breaker**


```rust
use rust_rabbit::circuit_breaker::CircuitBreakerConfig;

let config = RabbitConfig::builder()
    .connection_string("amqp://localhost:5672")
    .circuit_breaker(CircuitBreakerConfig {
        failure_threshold: 5,
        failure_window: Duration::from_secs(60),
        recovery_timeout: Duration::from_secs(30),
        success_threshold: 3,
        half_open_max_requests: 5,
    })
    .build();

// Circuit breaker automatically handles connection failures
```

## 📊 **Performance**


**RustRabbit v0.3.0 Benchmarks:**

| Metric | Value | Improvement vs v0.2.0 |
|--------|--------|-------------|
| **Throughput** | 75,000+ msgs/sec | +50% |
| **Latency (P99)** | < 8ms | -20% |
| **Memory Usage** | < 45MB baseline | -10% |
| **Connection Pool** | 10-100 connections | Stable |

**Advanced Pattern Performance:**

| Pattern | Throughput | Memory Overhead | Best Use Case |
|---------|------------|-----------------|---------------|
| **Request-Response** | 25,000 req/sec | +5MB | RPC, API calls |
| **Saga** | 10,000 flows/sec | +8MB | Distributed transactions |
| **Event Sourcing** | 50,000 events/sec | +15MB | CQRS, audit trails |
| **Priority Queue** | 60,000 msgs/sec | +2MB | Task prioritization |

*Benchmarks: Intel i7-10700K, 32GB RAM, RabbitMQ 3.12*

## 🛠️ **Configuration**


### **Builder Pattern Configuration**


```rust
use rust_rabbit::{RabbitConfig, consumer::ConsumerOptions};

// Environment-specific configs
let prod_config = RabbitConfig::builder()
    .connection_string("amqp://prod-server:5672")
    .connection_timeout(Duration::from_secs(30))
    .retry(|retry| retry.aggressive())
    .health(|health| health.frequent())
    .pool(|pool| pool.high_throughput())
    .build();

// Consumer configurations
let reliable_options = ConsumerOptions::builder("critical-orders")
    .consumer_tag("critical-processor")
    .minutes_retry()      // Auto-configure for reliability
    .prefetch_count(1)    // Process one at a time
    .build();

let high_throughput_options = ConsumerOptions::builder("bulk-orders")
    .consumer_tag("bulk-processor")
    .high_throughput()    // Optimize for speed
    .auto_declare_queue()
    .build();
```

## 🧪 **Testing**


RustRabbit includes comprehensive test coverage:

```bash
# Unit tests (58 tests)

cargo test --lib

# Integration tests with real RabbitMQ

docker-compose -f docker-compose.test.yml up -d
cargo test --test integration_example -- --test-threads=1

# Examples compilation

cargo check --examples

# Performance benchmarks

cargo bench
```

**Test Coverage:**
- ✅ End-to-end message flows
- ✅ Retry mechanisms with delayed exchange
- ✅ Health monitoring and recovery
- ✅ All advanced patterns (Phase 2)
- ✅ Concurrent processing scenarios
- ✅ Error handling and edge cases

## 📚 **Examples**


Comprehensive examples in the `examples/` directory:

```bash
# Core features

cargo run --example minutes_retry_preset        # NEW: One-line retry setup
cargo run --example simple_auto_consumer_example
cargo run --example retry_policy_demo

# Advanced patterns (Phase 2)

cargo run --example phase2_patterns_example     # Comprehensive demo
cargo run --example saga_example               # E-commerce workflow
cargo run --example event_sourcing_example     # Bank account CQRS

# Comparison examples

cargo run --example before_vs_after_setup      # Shows complexity reduction
```

## ✅ **Retry Best Practices & Common Pitfalls**


### **🚀 Production-Ready Patterns**


#### **1. Idempotent Message Processing**

```rust
use std::collections::HashSet;

struct IdempotentOrderHandler {
    processed_ids: Arc<Mutex<HashSet<String>>>,
}

#[async_trait]

impl MessageHandler<OrderMessage> for IdempotentOrderHandler {
    async fn handle(&self, message: OrderMessage, _context: MessageContext) -> MessageResult {
        // Check if already processed
        {
            let processed = self.processed_ids.lock().await;
            if processed.contains(&message.order_id) {
                warn!("Order {} already processed, skipping", message.order_id);
                return MessageResult::Ack;  // Don't process again
            }
        }
        
        match process_order(&message).await {
            Ok(_) => {
                // Mark as processed
                self.processed_ids.lock().await.insert(message.order_id.clone());
                MessageResult::Ack
            }
            Err(e) => {
                if is_retryable(&e) {
                    MessageResult::Retry
                } else {
                    MessageResult::Reject
                }
            }
        }
    }
}
```

#### **2. Graceful Shutdown with Retry**

```rust
use tokio::signal;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

#[tokio::main]

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let shutdown_flag = Arc::new(AtomicBool::new(false));
    let shutdown_flag_clone = shutdown_flag.clone();
    
    // Setup graceful shutdown
    tokio::spawn(async move {
        signal::ctrl_c().await.expect("Failed to listen for ctrl+c");
        info!("Shutdown signal received, finishing current messages...");
        shutdown_flag_clone.store(true, Ordering::Relaxed);
    });
    
    let handler = GracefulOrderHandler { shutdown_flag };
    let consumer = Consumer::new(connection_manager, consumer_options).await?;
    
    consumer.consume(handler).await?;
    Ok(())
}

struct GracefulOrderHandler {
    shutdown_flag: Arc<AtomicBool>,
}

#[async_trait]

impl MessageHandler<OrderMessage> for GracefulOrderHandler {
    async fn handle(&self, message: OrderMessage, _context: MessageContext) -> MessageResult {
        // Check shutdown flag
        if self.shutdown_flag.load(Ordering::Relaxed) {
            warn!("Shutdown in progress, requeuing message {}", message.order_id);
            return MessageResult::Requeue;  // Let another instance handle it
        }
        
        // Normal processing
        match process_order(&message).await {
            Ok(_) => MessageResult::Ack,
            Err(e) if is_retryable(&e) => MessageResult::Retry,
            Err(_) => MessageResult::Reject,
        }
    }
}
```

### **⚠️ Common Pitfalls to Avoid**


#### **❌ Pitfall 1: Auto-ACK with Retry Policy**

```rust
// 🚫 WRONG - Retry won't work!
let options = ConsumerOptions::builder("orders")
    .auto_ack(true)           // ← This breaks retry mechanism
    .retry_policy(retry)      // ← Will be ignored!
    .build();

// ✅ CORRECT - Manual ACK enables retry
let options = ConsumerOptions::builder("orders")
    .auto_ack(false)          // ← Must be false for retry to work
    .retry_policy(retry)      // ← Now retry works properly
    .build();
```

#### **❌ Pitfall 2: Poison Messages with Infinite Retry**

```rust
// 🚫 WRONG - Can create poison messages
#[async_trait]

impl MessageHandler<OrderMessage> for BadHandler {
    async fn handle(&self, message: OrderMessage, _context: MessageContext) -> MessageResult {
        match serde_json::from_str::<OrderMessage>(&message.data) {
            Ok(order) => process_order(order).await,
            Err(_) => MessageResult::Retry,  // ← Will retry forever on invalid JSON!
        }
    }
}

// ✅ CORRECT - Classify errors properly
#[async_trait]

impl MessageHandler<OrderMessage> for GoodHandler {
    async fn handle(&self, message: OrderMessage, context: MessageContext) -> MessageResult {
        let retry_count = context.get_retry_count();
        
        match serde_json::from_str::<OrderMessage>(&message.data) {
            Ok(order) => {
                match process_order(order).await {
                    Ok(_) => MessageResult::Ack,
                    Err(e) if is_retryable(&e) => MessageResult::Retry,
                    Err(_) => MessageResult::Reject,
                }
            }
            Err(_) => {
                error!("Invalid JSON format, sending to DLQ");
                MessageResult::Reject  // ← Don't retry parse errors
            }
        }
    }
}
```

#### **❌ Pitfall 3: Blocking Operations in Handler**

```rust
// 🚫 WRONG - Blocks event loop
#[async_trait]

impl MessageHandler<OrderMessage> for BlockingHandler {
    async fn handle(&self, message: OrderMessage, _context: MessageContext) -> MessageResult {
        // This blocks the async runtime!
        std::thread::sleep(Duration::from_secs(10));
        MessageResult::Ack
    }
}

// ✅ CORRECT - Use async operations
#[async_trait]  

impl MessageHandler<OrderMessage> for AsyncHandler {
    async fn handle(&self, message: OrderMessage, _context: MessageContext) -> MessageResult {
        // Non-blocking async delay
        tokio::time::sleep(Duration::from_secs(10)).await;
        MessageResult::Ack
    }
}
```

#### **❌ Pitfall 4: Missing Dead Letter Queue**

```rust
// 🚫 WRONG - Failed messages lost forever
let retry = RetryPolicy::builder()
    .max_retries(3)
    .initial_delay(Duration::from_millis(500))
    // Missing DLQ configuration!
    .build();

// ✅ CORRECT - Always configure DLQ
let retry = RetryPolicy::builder()
    .max_retries(3)
    .initial_delay(Duration::from_millis(500))
    .dead_letter_exchange("orders.dlx".to_string())
    .dead_letter_queue("orders.dlq".to_string())  // ← Failed messages go here
    .build();
```

### **📊 Performance Tuning Tips**


#### **Optimize Prefetch Count**

```rust
// High throughput, low latency messages
let options = ConsumerOptions::builder("fast_queue")
    .prefetch_count(100)    // Batch more messages
    .build();

// Heavy processing, reliable delivery
let options = ConsumerOptions::builder("heavy_queue")
    .prefetch_count(1)      // Process one at a time
    .build();
    
// Balanced approach
let options = ConsumerOptions::builder("balanced_queue")
    .prefetch_count(10)     // Good default for most cases
    .build();
```

#### **Monitor Retry Metrics**

```rust
use rust_rabbit::metrics::RustRabbitMetrics;

let metrics = RustRabbitMetrics::new();

// Monitor in your handler
#[async_trait]

impl MessageHandler<OrderMessage> for MetricsHandler {
    async fn handle(&self, message: OrderMessage, context: MessageContext) -> MessageResult {
        let start = Instant::now();
        let retry_count = context.get_retry_count();
        
        // Track retry counts
        metrics.retry_attempts.with_label_values(&[&retry_count.to_string()]).inc();
        
        let result = match process_order(&message).await {
            Ok(_) => {
                metrics.messages_processed_success.inc();
                MessageResult::Ack
            }
            Err(e) if is_retryable(&e) => {
                metrics.messages_retry.inc();
                MessageResult::Retry
            }
            Err(_) => {
                metrics.messages_dead_letter.inc();
                MessageResult::Reject
            }
        };
        
        // Track processing time
        metrics.processing_duration
            .observe(start.elapsed().as_secs_f64());
            
        result
    }
}
```

### **🔧 Debugging Checklist**


#### **When Retry Doesn't Work:**

1. ✅ Check `auto_ack: false` in ConsumerOptions
2. ✅ Verify retry policy is configured
3. ✅ Ensure handler returns `MessageResult::Retry`
4. ✅ Check RabbitMQ has x-delayed-message plugin
5. ✅ Verify exchange bindings exist

#### **When Messages Disappear:**

1. ✅ Check dead letter exchange/queue configuration
2. ✅ Monitor DLQ for rejected messages
3. ✅ Verify queue declarations (durable, auto-delete)
4. ✅ Check network connectivity
5. ✅ Review consumer exceptions

#### **Performance Issues:**

1. ✅ Tune prefetch_count based on message size/processing time
2. ✅ Monitor connection pool usage
3. ✅ Check for blocking operations in handlers
4. ✅ Review retry delays (too aggressive?)
5. ✅ Monitor queue depth and consumer lag

## 🗺️ **Roadmap**


### ✅ **Phase 1 (v0.2.0) - COMPLETED**

- Prometheus metrics integration
- Circuit breaker pattern
- Health monitoring
- Connection pooling

### ✅ **Phase 2 (v0.3.0) - COMPLETED**

- Request-Response pattern
- Saga pattern for distributed transactions  
- Event sourcing with CQRS
- Message deduplication
- Priority queues
- **Minutes retry preset** - Zero-config production setup

### 🔮 **Phase 3 (v0.4.0) - Enterprise**

- Multi-broker support with failover
- Message encryption at rest
- Schema registry integration
- Advanced routing patterns
- Performance optimizations

## 🤝 **Contributing**


We welcome contributions! Areas for improvement:

- 🐛 Bug fixes and performance improvements
- 📚 Documentation and examples
- ✨ New features from roadmap
- 🧪 Additional test coverage
- 📊 Benchmarks and optimizations

## 🆘 **Support**


- 📖 [Documentation](https://docs.rs/rust-rabbit)
- 💬 [GitHub Discussions](https://github.com/nghiaphamln/rust-rabbit/discussions)
- 🐛 [Issue Tracker](https://github.com/nghiaphamln/rust-rabbit/issues)
- 📧 Email: nghiaphamln3@gmail.com

## 📄 **License**


MIT License - see [LICENSE](LICENSE) file for details.

## 🙏 **Acknowledgments**


- Inspired by [MassTransit](https://masstransit-project.com/) for .NET
- Built on [lapin](https://github.com/amqp-rs/lapin) for AMQP protocol
- Powered by [Prometheus](https://prometheus.io/) for metrics

---

<div align="center">

**⭐ Star us on GitHub if RustRabbit helps your project! ⭐**

[GitHub](https://github.com/nghiaphamln/rust-rabbit) • [Crates.io](https://crates.io/crates/rust-rabbit) • [Docs.rs](https://docs.rs/rust-rabbit)

*Built with ❤️ for the Rust community*

</div>