Queues and topology¶

A RabbitQueue descriptor names the queue a handler consumes and describes what that queue is expected to look like: durability, queue type, exchange bindings, prefetch, and raw x-* arguments. A descriptor sits directly in the #[subscriber(...)] decorator:

use ruststream_lapin::{AMQPValue, Delay, LapinBroker, QueueType, RabbitExchange, RabbitQueue};

#[derive(Debug, Deserialize)]
struct OrderPlaced {
    id: u64,
}

// One queue, fed by a topic exchange: every `order.*` event lands here. The queue survives
// restarts (durable is the default) and dead-letters rejected messages.
#[subscriber(RabbitQueue::new("orders")
    .queue_type(QueueType::Quorum)
    .bind(RabbitExchange::topic("events"), "order.*")
    .dead_letter_exchange("dead-letters")
    .prefetch(16))]
async fn on_order(event: &OrderPlaced) -> HandlerResult {
    println!("order event {}", event.id);
    HandlerResult::Ack
}

The bare-string form #[subscriber("orders")] is shorthand for RabbitQueue::new("orders"): a durable, shared, non-auto-delete queue consumed as-is.

Declaration is an opt-in¶

Descriptors describe the EXPECTED topology. By default nothing is created on the broker: a missing queue is a subscribe error, because managing infrastructure is the user's job, not the framework's. A service that owns its queues opts in per broker:

#[ruststream::app]
fn app() -> impl App {
    // declare_topology is off by default; this service owns its queues, so it opts in.
    // default_queue_type applies to descriptors that do not pick a type themselves.
    let broker = LapinBroker::new("amqp://localhost:5672")
        .declare_topology(true)
        .default_queue_type(QueueType::Quorum)
        .prefetch(64);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(on_order);
        b.include(on_bounded);
        b.include(on_charge);
    })
}

With declaration enabled, subscribing declares the bound exchanges (except the built-in amq.* ones and the default exchange), the queue, and the bindings, in that order. Declaration is idempotent as long as the descriptor matches what exists; AMQP refuses to redeclare an entity with different properties (PRECONDITION_FAILED).

Queue types¶

RabbitMQ picks the queue implementation at declaration time via x-queue-type, and the type of an existing queue can never change. The descriptor exposes it as a typed option:

.queue_type(QueueType::Classic) - the classic single-node implementation.
.queue_type(QueueType::Quorum) - Raft-replicated; must stay durable (the crate rejects a quorum descriptor marked non-durable instead of letting the broker fail the declare).

A broker-wide .default_queue_type(..) applies to descriptors that do not pick a type; with neither set, no x-queue-type is sent and the server default applies.

Note that RabbitMQ 4 denies transient (non-durable) non-exclusive queues by default: keep the durable default unless the queue is .exclusive(true).

Prefetch¶

.prefetch(n) on the broker sets the per-subscription basic.qos window: at most n deliveries in flight unacknowledged. This is the back-pressure valve for the subscriber stream - consuming slower slows the broker's pushes instead of buffering without bound. A descriptor overrides it per queue with .prefetch(n). Without either, the server imposes no limit.

Delivery metadata¶

AmqpContext carries the AMQP delivery metadata that is not part of the payload or the headers: the exchange, the routing key, the redelivered flag, and the channel-local delivery tag. Each field has a zero-sized key in context::keys, readable two ways: as an extractor parameter (Ctx(routing_key): Ctx<RoutingKey> - the key names its context, so no ctx parameter is needed), or through a declared ctx: &mut Context<'_, AmqpContext> parameter with ctx.context(KEY):

// Delivery metadata arrives as extractor parameters: each key names the context it reads, so
// the handler declares exactly the fields it needs and no ctx parameter at all.
#[subscriber(RabbitQueue::new("orders-audit"))]
async fn audit(
    order: &Order,
    Ctx(routing_key): Ctx<RoutingKey>,
    Ctx(redelivered): Ctx<Redelivered>,
) -> HandlerResult {
    println!(
        "order {} came via {routing_key} (redelivered: {redelivered})",
        order.id
    );
    HandlerResult::Ack
}

Keyed worker lanes¶

A subscription can dispatch across several worker lanes with workers(n, by_key), keeping deliveries that share a key on the same lane (ordered per key, parallel across keys):

// Eight lanes, keyed: two orders for the same tenant never process concurrently (per-tenant
// ordering), while different tenants run in parallel. `by_key` reads the partition key the
// producer set on the message.
#[subscriber(RabbitQueue::new("orders"), workers(8, by_key))]
async fn on_order(order: &Order) -> HandlerResult {
    println!("order {} for tenant {}", order.id, order.tenant);
    HandlerResult::Ack
}

The key comes from the PARTITION_KEY_HEADER (amqp-partition-key): the producer sets it on the outgoing message's headers, and the crate reads it back through the Partitioned capability. AMQP itself does not interpret the header, so it is a pure client-side convention - unrelated to server-side hash routing (see the consistent-hash exchange below).

Dead-letter¶

.dead_letter_exchange("dlx") (plus optionally .dead_letter_routing_key(..)) sets the queue's native dead-letter target. A handler that drops a message settles with basic.reject(requeue = false), which routes it there; no extra machinery is involved.

Delayed retry¶

A handler that returns HandlerResult::retry_after(delay) asks for redelivery no sooner than delay - the not-ready-yet case, where an immediate requeue would just spin. By default the runtime handles this with its broker-agnostic fallback (the delayed copy waits in the service process, at-most-once over the window). .delay(..) makes it native instead: the message parks in a broker waiting queue with a per-message TTL and dead-letters back to the origin queue when the TTL fires, so the delayed copy lives on the broker.

// `.delay(..)` makes `retry_after` native: a delayed message parks in a broker waiting queue for
// the delay, then dead-letters back here - durable, off the service process. The waiting queue
// (`charges.retry` by default) is declared under `declare_topology`.
#[subscriber(RabbitQueue::new("charges").delay(Delay::dlx_ttl()))]
async fn on_charge(event: &OrderPlaced) -> HandlerResult {
    if event.id == 0 {
        // Not ready yet: come back in 30s instead of spinning on an immediate requeue.
        return HandlerResult::retry_after(Duration::from_secs(30));
    }
    HandlerResult::Ack
}

The waiting queue (<queue>.retry by default, or Delay::dlx_ttl_named(..)) is infrastructure: it is declared only under declare_topology(true), otherwise provision it yourself. Because a classic queue only releases expired messages from its head, use one waiting queue per delay class (or a quorum queue) when delays vary widely.

For workloads with widely mixed delays, the plugin-dme feature offers Delay::plugin_dme(), which routes redeliveries through the delayed-message-exchange plugin instead: the message carries an x-delay header and the plugin holds each one independently, so a short delay never waits behind a long one. It needs the plugin enabled on the broker, hence the feature gate.

Consistent-hash exchange (plugin)¶

For server-side fan-out - spreading one stream across several queues by hashing the routing key - the plugin-consistent-hash feature exposes RabbitExchange::consistent_hash(..), lowering to the rabbitmq_consistent_hash_exchange plugin's exchange type. Each queue binds with its integer weight as the routing key, and the broker splits the hash space proportionally:

use ruststream_lapin::{LapinBroker, RabbitExchange, RabbitQueue};

#[derive(Debug, Deserialize)]
struct Order {
    id: u64,
}

// Two shards bound to one consistent-hash exchange: the broker splits the hash space by weight
// (the binding key), so orders spread evenly and each key always lands on the same shard.
#[subscriber(RabbitQueue::new("orders-shard-a")
    .bind(RabbitExchange::consistent_hash("orders-by-key"), "1"))]
async fn shard_a(order: &Order) -> HandlerResult {
    println!("shard a: order {}", order.id);
    HandlerResult::Ack
}

#[subscriber(RabbitQueue::new("orders-shard-b")
    .bind(RabbitExchange::consistent_hash("orders-by-key"), "1"))]
async fn shard_b(order: &Order) -> HandlerResult {
    println!("shard b: order {}", order.id);
    HandlerResult::Ack
}

Consistent-hash routing happens on the broker (across queues); it is the server-side counterpart to client-side keyed worker lanes (across lanes in one consumer). Enable the plugin on the broker before using it; the feature is off by default because the plugin is not part of a stock RabbitMQ.

Raw arguments¶

Anything the descriptor does not model rides through verbatim:

// Anything the descriptor does not model rides through verbatim as a raw `x-*` argument.
#[subscriber(RabbitQueue::new("bounded")
    .argument("x-message-ttl", AMQPValue::LongLongInt(60_000))
    .argument("x-max-length", AMQPValue::LongLongInt(100_000)))]
async fn on_bounded(event: &OrderPlaced) -> HandlerResult {
    println!("bounded order {}", event.id);
    HandlerResult::Ack
}

AMQPValue and FieldTable are re-exported from the crate for exactly this purpose.