Are you an LLM? Read llms.txt for a summary of the docs, or llms-full.txt for the full context.
Skip to content

RabbitMQ — Basic Pubsub

Use this as your starting point for RabbitMQ integration: it covers every non-sequenced topology shape (minimal queue, DLQ, escalating retry backoff, deferred delivery), all four Outcome variants, and all three publish variants against a real broker. The canonical reference for how topic definitions and handler outcomes map to RabbitMQ.

Prerequisites

  • Docker (a RabbitMQ testcontainer is spun up automatically — no manual docker run needed)
  • Cargo feature: rabbitmq

Run

cargo run --example rabbitmq_basic_pubsub --features rabbitmq

Expected output

topologies declared

messages published

[minimal] order=ORD-001 amount=$50.00 attempt=1
[minimal] order=ORD-002 amount=$99.00 attempt=1
[minimal] order=ORD-003 amount=$10.00 attempt=1
[minimal] order=ORD-004 amount=$25.00 attempt=1
[minimal] order=ORD-005 amount=$77.77 attempt=1
[dlq] rejecting order=ORD-001 → DLQ
[dlq] dead-letter: order=ORD-001 reason=rejected deaths=1
[retry] order=ORD-001 attempt=1
[retry]   → transient failure, will Retry
[retry] order=ORD-001 attempt=2
[retry]   → success on retry
[defer] order=ORD-001 attempt=1 redelivered=false
[defer]   → not ready yet, deferring to hold queue
[defer] order=ORD-001 attempt=1 redelivered=true
[defer]   → processing now
done

Source

//! Basic pub/sub examples covering all non-sequenced topology configurations.
//!
//! Demonstrates: `define_topic!`, manual `Topic` impl, all `Outcome` variants,
//! `publish`, `publish_with_headers`, `publish_batch`, and per-topic consumer
//! registration via `Broker<RabbitMq>::consumer_supervisor()`.
//!
//! Note: the earlier DLQ-consumer demo (`run_dlq::<DlqOrder>`) isn't wired
//! through the generic `ConsumerSupervisor<B>` wrapper yet — messages that
//! land in the DLQ here are simply left there for inspection.
//!
//! Spins up a RabbitMQ testcontainer automatically (requires a running
//! Docker daemon):
//!
//!     cargo run --example rabbitmq_basic_pubsub --features rabbitmq
 
use std::collections::HashMap;
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use shove::rabbitmq::RabbitMqConfig;
use shove::{
    Broker, ConsumerOptions, DeadMessageMetadata, JsonCodec, MessageHandler, MessageMetadata,
    Outcome, QueueTopology, RabbitMq, Topic, TopologyBuilder, define_topic,
};
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::rabbitmq::RabbitMq as RabbitMqImage;
 
// ─── Message type ───────────────────────────────────────────────────────────
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct OrderEvent {
    order_id: String,
    amount_cents: u64,
}
 
// ─── Topic definitions ──────────────────────────────────────────────────────
 
// 1. Minimal: just a queue. No DLQ, no hold queues.
//    Rejected messages are discarded with a warning log.
define_topic!(
    MinimalOrder,
    OrderEvent,
    TopologyBuilder::new("ex-minimal-orders")
        .hold_queue(Duration::from_secs(30))
        .build()
);
 
// 2. With DLQ: rejected messages land in a dead-letter queue for inspection.
define_topic!(
    DlqOrder,
    OrderEvent,
    TopologyBuilder::new("ex-dlq-orders").dlq().build()
);
 
// 3. With hold queues + DLQ: escalating retry backoff (5 s → 30 s → 120 s).
//    Messages that exceed max_retries are sent to DLQ.
define_topic!(
    RetryOrder,
    OrderEvent,
    TopologyBuilder::new("ex-retry-orders")
        .hold_queue(Duration::from_secs(5))
        .hold_queue(Duration::from_secs(30))
        .hold_queue(Duration::from_secs(120))
        .dlq()
        .build()
);
 
// 4. Manual `Topic` impl — for topics that use `Defer` (hold without retry
//    counter increment). `OnceLock` pattern shown explicitly.
struct ScheduledOrder;
 
impl Topic for ScheduledOrder {
    type Message = OrderEvent;
    type Codec = JsonCodec;
 
    fn topology() -> &'static QueueTopology {
        static TOPOLOGY: std::sync::OnceLock<QueueTopology> = std::sync::OnceLock::new();
        TOPOLOGY.get_or_init(|| {
            TopologyBuilder::new("ex-scheduled-orders")
                .hold_queue(Duration::from_secs(10))
                .dlq()
                .build()
        })
    }
}
 
// ─── Handlers ───────────────────────────────────────────────────────────────
 
// Acks every message.
struct AckHandler;
 
impl MessageHandler<MinimalOrder> for AckHandler {
    type Context = ();
    async fn handle(&self, msg: OrderEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[minimal] order={} amount=${:.2} attempt={}",
            msg.order_id,
            msg.amount_cents as f64 / 100.0,
            metadata.retry_count + 1,
        );
        Outcome::Ack
    }
}
 
// Rejects every message. Also implements handle_dead for DLQ processing.
struct RejectHandler;
 
impl MessageHandler<DlqOrder> for RejectHandler {
    type Context = ();
    async fn handle(&self, msg: OrderEvent, _metadata: MessageMetadata, _: &()) -> Outcome {
        println!("[dlq] rejecting order={} → DLQ", msg.order_id);
        Outcome::Reject
    }
 
    async fn handle_dead(&self, msg: OrderEvent, metadata: DeadMessageMetadata, _: &()) {
        println!(
            "[dlq] dead-letter: order={} reason={} deaths={}",
            msg.order_id,
            metadata.reason.as_deref().unwrap_or("unknown"),
            metadata.death_count,
        );
    }
}
 
// Retries once (simulates transient failure), then acks.
struct RetryHandler;
 
impl MessageHandler<RetryOrder> for RetryHandler {
    type Context = ();
    async fn handle(&self, msg: OrderEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[retry] order={} attempt={}",
            msg.order_id,
            metadata.retry_count + 1,
        );
        if metadata.retry_count == 0 {
            println!("[retry]   → transient failure, will Retry");
            Outcome::Retry
        } else {
            println!("[retry]   → success on retry");
            Outcome::Ack
        }
    }
}
 
// Defers the message on first delivery (re-delivers via hold queue without
// incrementing the retry counter), then acks.
struct DeferHandler;
 
impl MessageHandler<ScheduledOrder> for DeferHandler {
    type Context = ();
    async fn handle(&self, msg: OrderEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[defer] order={} attempt={} redelivered={}",
            msg.order_id,
            metadata.retry_count + 1,
            metadata.redelivered,
        );
        if metadata.retry_count == 0 && !metadata.redelivered {
            println!("[defer]   → not ready yet, deferring to hold queue");
            Outcome::Defer
        } else {
            println!("[defer]   → processing now");
            Outcome::Ack
        }
    }
}
 
// ─── Main ───────────────────────────────────────────────────────────────────
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Spin up a RabbitMQ testcontainer for the lifetime of this example.
    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?;
 
 
    // ── Declare all topologies ──
    let topology = broker.topology();
    topology.declare::<MinimalOrder>().await?;
    topology.declare::<DlqOrder>().await?;
    topology.declare::<RetryOrder>().await?;
    topology.declare::<ScheduledOrder>().await?;
    println!("topologies declared\n");
 
    // ── Publish ──
    let publisher = broker.publisher().await?;
 
    // Single publish
    let order = OrderEvent {
        order_id: "ORD-001".into(),
        amount_cents: 5000,
    };
    publisher.publish::<MinimalOrder>(&order).await?;
    publisher.publish::<DlqOrder>(&order).await?;
    publisher.publish::<RetryOrder>(&order).await?;
    publisher.publish::<ScheduledOrder>(&order).await?;
 
    // Publish with custom headers
    let mut headers = HashMap::new();
    headers.insert("x-source".into(), "example".into());
    headers.insert("x-priority".into(), "high".into());
    let order2 = OrderEvent {
        order_id: "ORD-002".into(),
        amount_cents: 9900,
    };
    publisher
        .publish_with_headers::<MinimalOrder>(&order2, headers)
        .await?;
 
    // Batch publish
    let batch = vec![
        OrderEvent {
            order_id: "ORD-003".into(),
            amount_cents: 1000,
        },
        OrderEvent {
            order_id: "ORD-004".into(),
            amount_cents: 2500,
        },
        OrderEvent {
            order_id: "ORD-005".into(),
            amount_cents: 7777,
        },
    ];
    publisher.publish_batch::<MinimalOrder>(&batch).await?;
    println!("messages published\n");
 
    // ── Start consumers via a single supervisor ──
    //
    // Each `register` spawns one tokio task running that topic's consumer.
    // The supervisor owns a shared shutdown token so ctrl-c or our timeout
    // stops every consumer in lock-step.
    let mut supervisor = broker.consumer_supervisor();
    supervisor.register::<MinimalOrder, _>(AckHandler, ConsumerOptions::<RabbitMq>::new())?;
    supervisor.register::<DlqOrder, _>(RejectHandler, ConsumerOptions::<RabbitMq>::new())?;
    supervisor.register::<RetryOrder, _>(
        RetryHandler,
        ConsumerOptions::<RabbitMq>::new()
            .with_max_retries(3)
            .with_prefetch_count(20),
    )?;
    supervisor.register::<ScheduledOrder, _>(DeferHandler, ConsumerOptions::<RabbitMq>::new())?;
 
    // Run until ctrl-c or a 5 s demo timeout, then drain for up to 5 s.
    let outcome = supervisor
        .run_until_timeout(
            async {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(5)) => {}
                    _ = tokio::signal::ctrl_c() => {}
                }
            },
            Duration::from_secs(5),
        )
        .await;
 
    println!("done");
    std::process::exit(outcome.exit_code());
}

Walkthrough

Four topic shapes with define_topic!

The example defines four topics using two different approaches. MinimalOrder, DlqOrder, and RetryOrder all use the define_topic! macro — a shorthand that generates the Topic impl, the OnceLock topology cell, and a zero-sized struct in one call. ScheduledOrder is declared manually to show the OnceLock pattern explicitly; this form is needed when the topic needs to use Outcome::Defer (which requires the hold queue TTL to be read from the topology). The RetryOrder topology chains three .hold_queue() calls to create escalating backoff stages: 5 s → 30 s → 120 s.

All four Outcome variants in action

  • AckHandler always returns Outcome::Ack — the message is removed from the queue.
  • RejectHandler returns Outcome::Reject — the message is immediately routed to the DLQ without using up a retry slot. It also implements handle_dead to process messages that arrive on the DLQ itself.
  • RetryHandler checks metadata.retry_count: on the first attempt it returns Outcome::Retry, which sends the message through the hold queue and back. On the second attempt it acks.
  • DeferHandler returns Outcome::Defer on first delivery when metadata.redelivered == false. Unlike Retry, Defer routes through the hold queue without incrementing the retry counter — the message comes back as if it were a fresh delivery.

Consumer supervisor

broker.consumer_supervisor() creates a ConsumerSupervisor\<RabbitMq\>. Each supervisor.register(handler, options) call spawns one AMQP channel/consumer for that topic. The supervisor owns a shared shutdown token, so run_until_timeout stops every registered consumer in lock-step and drains in-flight messages before returning. with_max_retries(3) on RetryOrder ensures messages that exhaust their retries are sent to the DLQ rather than looping indefinitely.

Publish variants

publisher.publish::<T>(&msg) sends a single message. publish_with_headers::<T>(&msg, headers) attaches custom AMQP headers (the example uses x-source and x-priority). publish_batch::<T>(&[msg]) sends multiple messages in a single channel operation — more efficient for bursts. All three variants share the same type-checked routing; the compiler rejects publishing the wrong message type for a given topic.

What to try next

  • Change Outcome::Retry to Outcome::Defer in RetryHandler — notice that metadata.retry_count no longer increments on re-delivery.
  • Add a fourth .hold_queue(Duration::from_secs(300)) to RetryOrder — messages get a fourth retry stage at 5 minutes.
  • Remove .dlq() from MinimalOrder — rejected messages are silently discarded (you will see a warning log from the framework).
  • See Guides: Retries for a full explanation of hold queues and retry limits.