rabbit_warren 🐇

English | Русский

A robust, ergonomic, and production-ready RabbitMQ client library for Rust, built on top of lapin.

It abstracts away typical boilerplate (Publisher Confirms, the mandatory flag, automatic ack/nack strategies, and distributed tracing), allowing developers to focus purely on business logic.

Features

• Safe Publishing: Automatically enables Publisher Confirms and uses the mandatory flag to detect unroutable messages.
• Smart Consuming: Automatic ack on success. On error, you declaratively choose the strategy: Requeue (with built-in backoff to prevent infinite retry loops) or Discard.
• Distributed Tracing: Built-in OpenTelemetry context propagation (injecting/extracting trace headers from AMQP properties).
• Zero Boilerplate: No need to manually manage channels, ackers, or nack options in your application code.
• JSON Helpers: Built-in publish_json and deserialize_delivery utilities.

Installation

Add this to your Cargo.toml:

[dependencies]
rabbit_warren = "0.1" # Replace with the latest version
tokio = { version = "1", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }

Note: You do not need to add lapin to your Cargo.toml. All necessary types (e.g., Delivery, FieldTable) are re-exported by this crate.

Usage Examples

1. Initialization

use rabbit_warren::RmqClient;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let url = "amqp://guest:guest@127.0.0.1:5672";
    // Connect and set prefetch count to 10
    let client = RmqClient::connect(url, 10).await?;
    
    // Declare topology
    client.declare_exchange("my_exchange", rabbit_warren::ExchangeKind::Topic, true).await?;
    client.declare_queue("my_queue", true).await?;
    client.bind_queue("my_queue", "my_exchange", "routing.key").await?;

    Ok(())
}

2. Publishing Messages

use serde::Serialize;

#[derive(Serialize)]
struct MyEvent {
    id: u32,
    payload: String,
}

async fn publish_event(client: &RmqClient) -> anyhow::Result<()> {
    let event = MyEvent { id: 1, payload: "hello".into() };
    
    // Automatically serializes to JSON, sets delivery_mode=2 (persistent), 
    // and waits for publisher confirmation.
    client.publish_json("my_exchange", "routing.key", &event, None).await?;
    
    Ok(())
}

3. Consuming Messages

The library provides a ResultExt trait, allowing you to declaratively specify what should happen upon an error, without cluttering your handler with ack/nack logic.

use rabbit_warren::{ProcessingError, ResultExt};
use serde::Deserialize;

#[derive(Deserialize)]
struct MyEvent {
    id: u32,
    payload: String,
}

async fn start_consuming(client: &RmqClient) {
    client.consume("my_queue", |delivery| async move {
        // 1. Parsing. If it fails, requeue (transient issue).
        let event: MyEvent = rabbit_warren::deserialize_delivery(&delivery)
            .map_err(|e| ProcessingError::retryable(e))?;

        // 2. Business logic. 
        // If it's a permanent business error (e.g., invalid data), discard to avoid poison pills.
        if event.id == 0 {
            return Err(ProcessingError::permanent("Invalid ID".to_string()));
        }

        // 3. Simulate a transient error (e.g., DB timeout). 
        // The library automatically pauses (backoff) and requeues the message.
        if event.id == 99 {
            return Err(ProcessingError::retryable("Database temporarily unavailable".to_string()));
        }

        println!("Successfully processed event: {}", event.id);
        
        // Returning Ok(()) automatically acknowledges (ack) the message.
        Ok(())
    });
}

4. Ergonomic Error Handling (ResultExt)

You can use extension methods directly in your call chain to explicitly dictate the message's fate on specific errors:

use rabbit_warren::{RmqClient, ResultExt};

client.consume("my_queue", |delivery| async move {
    // Deserialization error -> requeue
    let msg: MyMessage = rabbit_warren::deserialize_delivery(&delivery).requeue_on_err()?;
    
    // Validation error -> discard
    validate(&msg).discard_on_err()?;
    
    // Database error -> requeue
    save_to_db(&msg).await.requeue_on_err()?;
    
    Ok(())
});

5. Consumer Refactoring: Before vs. After

This library replaces dozens of lines of manual channel management, while let loops, and explicit ack/nack calls with pure business logic.

Before raw lapin boilerplate

// 1. Verbose topology declaration
channel.exchange_declare(&exchange, ExchangeKind::Topic, ExchangeDeclareOptions { durable: true, ..Default::default() }, FieldTable::default()).await?;
channel.queue_declare(&queue_name, QueueDeclareOptions { durable: true, ..Default::default() }, FieldTable::default()).await?;
channel.queue_bind(&queue_name, &exchange, &routing_key, QueueBindOptions::default(), FieldTable::default()).await?;

let mut consumer = channel.basic_consume(&queue_name, "consumer-tag", BasicConsumeOptions::default(), FieldTable::default()).await?;

// 2. Manual lifecycle and stream management
tokio::spawn(async move {
    while let Some(delivery_result) = consumer.next().await {
        let delivery = delivery_result.unwrap();
        // ... manual payload parsing ...
        
        // 3. Manual ack/nack management (high risk of poison pills!)
        match process_message(delivery).await {
            Ok(_) => delivery.ack(BasicAckOptions::default()).await.unwrap(),
            Err(_) => delivery.nack(BasicNackOptions { requeue: true, ..Default::default() }).await.unwrap(),
        }
    }
});

After using rabbit_warren

First, centrally define the error handling strategy for your app:

use rabbit_warren::{AckAction, IntoAckAction};

impl IntoAckAction for AppError {
    fn ack_action(&self) -> AckAction {
        match self {
            AppError::InvalidJson(_) => AckAction::Discard, // Permanent -> Discard
            AppError::DatabaseError(_) => AckAction::Requeue, // Transient -> Requeue
        }
    }
}

Then, write clean, business-focused consumer code:

let handler = move |delivery: Delivery| {
    async move {
        let payload = rabbit_warren::payload_as_utf8(&delivery)?;
        let order: Order = serde_json::from_str(payload)?;
        processor.process(order).await?; 
        Ok(())
    }
};

// The library handles the loop, tracing, and ack/nack automatically
rmq_client.consume(&queue_name, handler);

Custom Ack Strategies

If your application has a custom error enum, implement IntoAckAction to centralize retry logic:

use rabbit_warren::{AckAction, IntoAckAction};

#[derive(Debug, thiserror::Error)]
enum AppError {
    #[error("Database connection lost")]
    DbConnection,
    #[error("Validation failed: {0}")]
    Validation(String),
}

impl IntoAckAction for AppError {
    fn ack_action(&self) -> AckAction {
        match self {
            AppError::DbConnection => AckAction::Requeue, 
            AppError::Validation(_) => AckAction::Discard, 
        }
    }
}

Note: By default, any error type not implementing IntoAckAction is treated as AckAction::Requeue to prevent accidental data loss.

Distributed Tracing

The library integrates seamlessly with the tracing ecosystem. The consume method automatically creates an info_span for every message.

For full OpenTelemetry support, extract and inject context into AMQP headers:

use rabbit_warren::FieldTable;
use opentelemetry::global;

let mut headers = FieldTable::default();
global::get_text_map_propagator(|propagator| {
    // propagator.inject_context(&context, &mut carrier);
});
client.publish_json("exchange", "key", &msg, Some(headers)).await?;

Testing

Integration tests use testcontainers to spin up an ephemeral RabbitMQ instance. Ensure you have Docker installed and running, then execute:

cargo test --all-features

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.