# rabbit_warren 🐇
[](https://crates.io/crates/rabbit_warren)
[](https://docs.rs/rabbit_warren)
[](./LICENSE)
A robust, ergonomic, and production-ready RabbitMQ client library for Rust, built on top of [`lapin`](https://crates.io/crates/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`:
```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
```rust
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
```rust
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.
```rust
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:
```rust
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
```rust
// 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:
```rust
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:
```rust
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);
```
</details>
## Custom Ack Strategies
If your application has a custom error enum, implement `IntoAckAction` to centralize retry logic:
```rust
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:
```rust
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`](https://crates.io/crates/testcontainers) to spin up an ephemeral RabbitMQ instance. Ensure you have Docker installed and running, then execute:
```bash
cargo test --all-features
```
## License
Licensed under either of [Apache License, Version 2.0](LICENSE-APACHE) or [MIT license](LICENSE-MIT) at your option.