delivery_limit

delivery_limit(n) sets a quorum queue's x-delivery-limit: how many times the queue returns a message before it carries it away.

The argument counts returns, so the queue delivers a message n + 1 times: delivery_limit(4) delivers it five times. A quorum queue counts a delivery the handler rejected and a delivery whose consumer went away without settling it. Once the count passes the limit, the queue sends the message to its dead-letter exchange, or drops it when it has none. The server applies the limit itself: it holds while no service is running, and the service publishes nothing. A quorum queue without a limit of its own takes RabbitMQ's default of 20 returns. A returned message carries the queue's count in the x-delivery-count header.

A registration states the same through the mount chain: max_attempts(n) and dead_letter(name). On a quorum queue the pair becomes an x-delivery-limit of n - 1 and a dead-letter route to the queue name, written over the descriptor's limit. The pair reaches the queue only when the service declares it, so the broker sets declare_topology(true). On a broker that declares nothing the subscription refuses to start, and so does half of the pair. out_retry(policy) does not compile on a quorum queue, because no copy leaves this process.

Example

The provider refuses a charge every time. The queue delivers it five times and then moves it to charges.dlx, where a classic queue keeps it.

Macros
Manual

examples/lapin_settlement_delivery_limit.rs
/// The provider refuses this charge every time. The queue returns it four times, so it is
/// delivered five times, and then the queue moves it to `charges.dlx`.
#[subscriber(RabbitQuorumQueue::new("charges")
    .delivery_limit(4)
    .dead_letter_exchange("charges.dlx"))]
async fn take_payment(charge: &Charge) -> HandlerOutcome {
    tracing::warn!(charge.id, "provider refused the charge");
    HandlerOutcome::retry()
}

/// The charges the queue gave up on.
#[subscriber(RabbitQueue::new("charges.parked").bind(RabbitExchange::fanout("charges.dlx"), ""))]
async fn park(charge: &Charge) -> HandlerOutcome {
    tracing::error!(charge.id, "charge parked");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(take_payment);
        b.include(park);
    })
}

examples/manual/lapin_settlement_delivery_limit.rs
/// The provider refuses this charge every time. The queue returns it four times, so it is
/// delivered five times, and then the queue moves it to `charges.dlx`.
struct TakePayment;

impl Handle<Charge> for TakePayment {
    fn handle(
        &self,
        charge: &Charge,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::warn!(charge.id, "provider refused the charge");
        ready(Err(HandlerOutcome::retry()))
    }
}

/// The charges the queue gave up on.
struct Park;

impl Handle<Charge> for Park {
    fn handle(
        &self,
        charge: &Charge,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::error!(charge.id, "charge parked");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        let charges = RabbitQuorumQueue::new("charges")
            .delivery_limit(4)
            .dead_letter_exchange("charges.dlx");
        let parked =
            RabbitQueue::new("charges.parked").bind(RabbitExchange::fanout("charges.dlx"), "");
        b.include(subscriber(charges, TakePayment).build());
        b.include(subscriber(parked, Park).build());
    })
}

Example​

Example