RabbitQueue

RabbitQueue consumes a classic queue: its durability, exclusivity, auto-delete, bindings and the settings shared with quorum queues.

RabbitQueue::new names the queue. Three settings belong to the classic queue alone:

Setting	Default	What it does
`durable`	`true`	the queue survives a broker restart
`exclusive`	`false`	the queue belongs to the connection that declared it and goes with it
`auto_delete`	`false`	the server deletes the queue once its last consumer leaves

RabbitMQ 4 refuses a transient queue that other connections may share, so durable(false) comes with exclusive(true). An exclusive queue suits a per-instance feed, such as a cache that every instance refreshes from a fanout exchange.

The other settings are shared with RabbitQuorumQueue: bind and bind_with join the queue to exchanges, dead_letter_exchange and dead_letter_routing_key give it a route for rejected messages, prefetch and batch_wait shape its consumption, delay parks delayed retries on the server, and argument passes anything else the queue declaration takes. Bindings, the dead-letter route and the arguments reach the server when the broker declares the topology.

A classic queue keeps no count of a message's deliveries. A retry puts the message back on the queue, and its next delivery carries the redelivered flag, which a handler reads through the Redelivered key.

Example

The order queue is bound to a topic exchange and dead-letters to orders.failed; an order that fails once is placed on its redelivery. The price queue belongs to this instance alone.

Macros
Manual

examples/lapin_subscriptions_rabbit_queue.rs
/// A durable queue fed by the topic exchange `events`; a dropped order leaves through
/// `orders.failed`. A delivery the server already handed out once is placed on its second try.
#[subscriber(RabbitQueue::new("orders")
    .bind(RabbitExchange::topic("events"), "order.*")
    .dead_letter_exchange("orders.failed"))]
async fn place(order: &Order, Ctx(redelivered): Ctx<Redelivered>) -> HandlerOutcome {
    if !redelivered && order.flaky {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, redelivered, "order placed");
    HandlerOutcome::ack()
}

/// A queue of this instance alone: it lives as long as the connection that declared it.
#[subscriber(RabbitQueue::new("prices.instance-a")
    .durable(false)
    .exclusive(true)
    .auto_delete(true)
    .bind(RabbitExchange::fanout("prices"), ""))]
async fn refresh(price: &Price) -> HandlerOutcome {
    tracing::info!(price.sku, price.cents, "price cached");
    HandlerOutcome::ack()
}

examples/manual/lapin_subscriptions_rabbit_queue.rs
/// A delivery the server already handed out once is placed on its second try.
struct Place;

impl Handle<Order, (), (), AmqpContext> for Place {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, AmqpContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let redelivered = ctx.context(Redelivered);
        if !redelivered && order.flaky {
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(order.id, redelivered, "order placed");
        ready(Ok(()))
    }
}

/// A durable queue fed by the topic exchange `events`; a dropped order leaves through
/// `orders.failed`.
fn orders() -> RabbitQueue {
    RabbitQueue::new("orders")
        .bind(RabbitExchange::topic("events"), "order.*")
        .dead_letter_exchange("orders.failed")
}

struct Refresh;

impl Handle<Price> for Refresh {
    fn handle(
        &self,
        price: &Price,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(price.sku, price.cents, "price cached");
        ready(Ok(()))
    }
}

/// A queue of this instance alone: it lives as long as the connection that declared it.
fn prices() -> RabbitQueue {
    RabbitQueue::new("prices.instance-a")
        .durable(false)
        .exclusive(true)
        .auto_delete(true)
        .bind(RabbitExchange::fanout("prices"), "")
}

Example​

Example