RabbitMQ — Audited Consumer

Use this when you want to verify that trace IDs are preserved across retries on RabbitMQ. The handler returns Outcome::Retry on the first delivery; both delivery attempts emit an audit record with the same UUID — proving that trace identity survives the hold-queue round-trip. The pattern directly applies to payment processing, compliance logging, and any workflow where correlating retries to the original delivery matters.

Prerequisites

Docker (a RabbitMQ testcontainer is spun up automatically)
Cargo features: rabbitmq,audit

Run

cargo run --example rabbitmq_audited_consumer --features rabbitmq,audit

Expected output

topology declared

published payment

[handler] payment=PAY-001 amount=$49.99 attempt=1
[audit] {
  "trace_id": "550e8400-e29b-41d4-a716-446655440000",
  "outcome": "Retry",
  "duration_ms": 0,
  ...
}
[handler] payment=PAY-001 amount=$49.99 attempt=2
[audit] {
  "trace_id": "550e8400-e29b-41d4-a716-446655440000",
  "outcome": "Ack",
  "duration_ms": 0,
  ...
}
done

The trace_id is the same UUID on both delivery attempts.

Source

//! Audited consumer example — custom `AuditHandler` that writes to stdout.
//!
//! Demonstrates: `MessageHandlerExt::audited` wrapper, custom `AuditHandler`
//! implementation, trace ID propagation across retries.
//!
//! Spins up a RabbitMQ testcontainer automatically (requires a running
//! Docker daemon):
//!
//!     cargo run --example rabbitmq_audited_consumer --features rabbitmq,audit
 
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use shove::rabbitmq::RabbitMqConfig;
use shove::*;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::rabbitmq::RabbitMq as RabbitMqImage;
 
// ─── Message type ───────────────────────────────────────────────────────────
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PaymentEvent {
    payment_id: String,
    amount_cents: u64,
}
 
// ─── Topic ──────────────────────────────────────────────────────────────────
 
define_topic!(
    Payments,
    PaymentEvent,
    TopologyBuilder::new("ex-audited-payments")
        .hold_queue(Duration::from_secs(5))
        .dlq()
        .build()
);
 
// ─── Handler ────────────────────────────────────────────────────────────────
 
struct PaymentHandler;
 
impl MessageHandler<Payments> for PaymentHandler {
    type Context = ();
    async fn handle(&self, msg: PaymentEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[handler] payment={} amount=${:.2} attempt={}",
            msg.payment_id,
            msg.amount_cents as f64 / 100.0,
            metadata.retry_count + 1,
        );
        // Simulate a transient failure on first attempt to show trace ID
        // persisting across retries.
        if metadata.retry_count == 0 {
            Outcome::Retry
        } else {
            Outcome::Ack
        }
    }
}
 
// ─── Custom audit handler ───────────────────────────────────────────────────
 
/// Prints every audit record to stdout as JSON.
#[derive(Clone, Default)]
struct StdoutAuditHandler;
 
impl AuditHandler<Payments> for StdoutAuditHandler {
    async fn audit(&self, record: &AuditRecord<PaymentEvent>) -> Result<(), ShoveError> {
        let json = serde_json::to_string_pretty(record).map_err(ShoveError::Serialization)?;
        println!("[audit] {json}");
        Ok(())
    }
}
 
// ─── Main ───────────────────────────────────────────────────────────────────
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let container = RabbitMqImage::default().start().await?;
    let port = container.get_host_port_ipv4(5672).await?;
    let uri = format!("amqp://guest:guest@localhost:{port}/%2f");
    let broker = Broker::<RabbitMq>::new(RabbitMqConfig::new(&uri)).await?;
    broker.topology().declare::<Payments>().await?;
    println!("topology declared\n");
 
    // ── Publish a payment ──
    let publisher = broker.publisher().await?;
    let event = PaymentEvent {
        payment_id: "PAY-001".into(),
        amount_cents: 4999,
    };
    publisher.publish::<Payments>(&event).await?;
    println!("published payment\n");
 
    // ── Start audited consumer ──
    //
    // `audited(audit)` wraps the handler in the `Audited` decorator — the
    // supervisor sees a normal `MessageHandler<Payments>`, no API changes needed.
    let mut supervisor = broker.consumer_supervisor();
    supervisor.register::<Payments, _>(
        PaymentHandler.audited(StdoutAuditHandler),
        ConsumerOptions::<RabbitMq>::new().with_max_retries(3),
    )?;
 
    // Let it process (first attempt retries, second acks), then shut down.
    let outcome = supervisor
        .run_until_timeout(
            async {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(10)) => {}
                    _ = tokio::signal::ctrl_c() => {}
                }
            },
            Duration::from_secs(5),
        )
        .await;
 
    println!("done");
    std::process::exit(outcome.exit_code());
}

Walkthrough

Handler with deliberate retry

PaymentHandler inspects metadata.retry_count: when it is 0 (first delivery) it returns Outcome::Retry, simulating a transient failure. On the second delivery (retry_count >= 1) it returns Outcome::Ack. This two-attempt lifecycle is the minimal setup needed to observe trace-ID continuity across a retry cycle.

`StdoutAuditHandler` and `AuditRecord`

StdoutAuditHandler implements AuditHandler\<Payments\> and serialises the full AuditRecord\<PaymentEvent\> to pretty-printed JSON. The record includes trace_id (a UUID assigned when the message is first dispatched and carried through retries via AMQP headers), outcome (the value returned by the inner handler), duration_ms (wall-clock time the inner handler spent), and a copy of the original message. Printing the whole record as JSON makes it easy to inspect every field during development; swap in a DB writer or ShoveAuditHandler for production.

Wiring with `.audited()`

The call PaymentHandler.audited(StdoutAuditHandler) comes from MessageHandlerExt, which is blanket-implemented for all MessageHandler types when the audit feature is enabled. The resulting combinator satisfies MessageHandler\<Payments\>, so supervisor.register sees a single opaque handler — no API changes needed to add auditing.

Supervisor with `max_retries`

ConsumerOptions::<RabbitMq>::new().with_max_retries(3) ensures the consumer will retry up to three times before sending a message to the DLQ. In this example only one retry is needed, but setting a limit guards against runaway retry loops if the handler logic changes.

What to try next

Remove .with_max_retries(3) and change PaymentHandler to always Retry — the message loops until max_retries is reached (broker default), then lands in the DLQ; verify the audit record's outcome changes to "Reject" on the final delivery.
Replace StdoutAuditHandler with a struct that publishes to a second topic — this is the ShoveAuditHandler pattern for production audit pipelines.
Add a second payment with a different payment_id and observe that each gets a distinct trace_id.
See Guides: Audit for the full AuditRecord schema and production wiring patterns.