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

If you already run RabbitMQ, this backend slots in with zero infrastructure changes. It covers the full feature surface: hold-queue retry topologies, dead-letter routing, coordinated consumer groups with queue-depth autoscaling, per-key FIFO delivery via consistent-hash exchange routing, and optional AMQP transactional mode for exactly-once routing on handlers with irreversible side effects.

What you need

A RabbitMQ broker 3.13+ with the management plugin enabled. The autoscaler reads queue depth from the management HTTP API, so the plugin is required for consumer_group() autoscaling.

For local dev:

docker run --rm -p 5672:5672 -p 15672:15672 rabbitmq:3-management

This exposes AMQP on 5672 and the management UI / API on 15672.

Install

cargo add shove --features rabbitmq

For exactly-once transactional routing (see below):

cargo add shove --features rabbitmq-transactional

Connect

    let broker = Broker::<RabbitMq>::new(RabbitMqConfig::new(&uri)).await?;

RabbitMqConfig::new takes a standard AMQP URL. TLS and authentication are configured via the URL (e.g. amqps://user:pass@host:5671).

Management API

Autoscaling needs queue depth from the RabbitMQ management plugin. Attach credentials via .with_management(...):

use shove::rabbitmq::{ManagementConfig, RabbitMqConfig};

let config = RabbitMqConfig::new("amqp://guest:guest@localhost:5672")
    .with_management(
        ManagementConfig::new("http://localhost:15672", "guest", "guest")
            .with_vhost("my-vhost")              // non-default vhost
            .with_tls_ca_cert("/path/to/ca.pem") // custom CA for self-signed mgmt certs
            .with_tls_skip_verify(),             // dev only — disables verification
    );

When management is None, Broker::queue_stats_provider().snapshot(...) and RabbitMqAutoscalerBackend return a Topology error directing the caller to configure it. The consumer_supervisor() path does not need the plugin; only autoscaling does.

Declare topology

    let topology = broker.topology();
    topology.declare::<MinimalOrder>().await?;
    topology.declare::<DlqOrder>().await?;
    topology.declare::<RetryOrder>().await?;
    topology.declare::<ScheduledOrder>().await?;

topology().declare::<T>() creates the main queue, any hold queues, and the DLQ on the broker. Idempotent — safe to call on every startup.

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?;

publisher().await? returns a Publisher<RabbitMq> that can publish single messages, messages with custom headers, or batches.

Consume

    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;

consumer_supervisor() fans out one consumer task per registered topic. For coordinated groups with autoscaling, use consumer_group() instead — see the section below.

Sequenced delivery

Messages for the same key stay in order. RabbitMQ enforces per-key ordering via a consistent-hash exchange and single-active-consumer (SAC) sub-queues. When you declare a sequenced topic, shove provisions a consistent-hash exchange ({queue}-seq-hash) and N sub-queues each configured with SAC and prefetch=1. The publisher routes each message by T::sequence_key(); the consumer processes one message at a time per shard, which guarantees that messages for the same key are never processed concurrently.

The shard count is configurable via routing_shards(N) in TopologyBuilder (default 8). More shards increase parallelism across different keys at the cost of more queues on the broker.

See Sequenced Topics for the full ordering model, and Sequenced Pubsub example for a runnable walkthrough.

Consumer groups + autoscaling

For non-sequenced topics, a RabbitMQ consumer group is N consumers competing on the main queue under standard work-queue semantics — the broker hands each message to one consumer. (Single-active-consumer is reserved for sequenced sub-queues, where one consumer per shard is required to preserve ordering — see Sequenced delivery above.) Call broker.consumer_group() and register topics with a ConsumerGroupConfig:

use shove::rabbitmq::ConsumerGroupConfig as RabbitMqGroupConfig;
use shove::{Broker, ConsumerGroupConfig, RabbitMq};

let mut group = broker.consumer_group();
group
    .register::<OrderTopic, _>(
        ConsumerGroupConfig::new(
            RabbitMqGroupConfig::new(1..=8)
                .with_prefetch_count(20)
                .with_max_retries(5),
        ),
        || MyHandler,
    )
    .await?;

The autoscaler reads queue depth from the RabbitMQ management HTTP API and scales the number of active consumers within the min..=max range. The management plugin must be running and reachable for autoscaling to work.

Connection pooling

When max_consumers > 50, the group automatically pools AMQP connections — one per ~50 workers, rounded up. Lapin parses every inbound delivery on a connection through a single reader task, so a high-throughput group of 200 consumers on one connection bottlenecks on that task. The pool is transparent; all clones of the underlying RabbitMqClient share broker config and reconnect logic.

Stuck-retry eviction

If a transient broker error kills the consumer mid-republish, a delivery can land in the per-consumer AwaitingRetry map and never escape. ConsumerOptions::<RabbitMq>::new().with_hold_queue_timeout(Duration) evicts any entry that has been waiting longer than the timeout and requeues it to the main queue. The eviction ticker fires at half the configured timeout, so a 30 s setting checks every 15 s. Default: no eviction — set it explicitly to enable.

Reconnect bounds

ConsumerOptions::<RabbitMq>::new().with_max_reconnect_attempts(N) caps how many times the consumer redials after a connection drop before surfacing a Connection error. Default: unlimited (retries forever).

See the Consumer Groups example for a full runnable walkthrough.

Exactly-once mode

Enable with --features rabbitmq-transactional. This wraps the publish-to-hold-queue + ack in an AMQP transaction (tx_select / tx_commit), eliminating the publish-then-ack race condition. The trade-off is approximately 10–15× lower throughput per channel. Reserve this for handlers with irreversible side effects where duplicate delivery is unacceptable.

See the Exactly-Once example and the Exactly-Once guide.

Gotchas

  • Management plugin is required for autoscaler queue-depth polling. Attach credentials via RabbitMqConfig::with_management(ManagementConfig::new(...)) — when absent, queue_stats_provider().snapshot(...) and the autoscaler return a Topology error rather than silently using a no-op provider.
  • TLS and authentication are configured via the AMQP URL passed to RabbitMqConfig::new(...). Use amqps:// for TLS; embed credentials in the URL or configure them via environment variables before constructing the config.
  • lapin handles the AMQP connection. rustls--ring is the default TLS feature pulled in by shove. Downstream crates that need a different TLS stack (e.g. vendored OpenSSL) can activate alternative rdkafka or lapin feature flags without changing shove's API.
  • Auto-reconnect is built in at the consumer level. Transient broker failures — network blips, broker restarts — recover automatically without operator intervention.
  • Exactly-once throughput cost (~10–15×) means it should only be enabled for topics that genuinely require it. The standard at-least-once mode is far faster and handlers should be made idempotent instead where possible.

Examples

See also