Skip to main content

retry_after

retry_after(delay) asks for a redelivery no sooner than delay, for a message that is not ready yet.

AMQP has no per-message delay, so on a queue without a delay the runtime waits. It settles the delivery, holds a copy in the service and publishes it back to the queue when the delay is up. The copy goes out under the queue's name on the default exchange, which addresses the queue of that name. It carries RETRY_COUNT_HEADER raised by one. The copy lives in the service's memory, so it is at-most-once over the delay: a process that stops before the timer fires loses it. A retry cap counts these copies like immediate ones.

Example​

The first delivery finds the payment unsettled and asks for the refund again in 30 seconds. The copy comes back with the retry count, and the refund goes through.

examples/lapin_settlement_retry_after.rs
/// The first delivery finds the payment unsettled and asks for the refund again in 30 seconds.
/// The copy that comes back carries the retry count, and the refund goes through.
#[subscriber(RabbitQueue::new("refunds"))]
async fn issue_refund(refund: &Refund, ctx: &mut Context<'_>) -> HandlerOutcome {
    if ctx.headers().get_str(RETRY_COUNT_HEADER).is_none() {
        tracing::info!(refund.id, "payment not settled yet");
        return HandlerOutcome::retry_after(RECHECK);
    }
    tracing::info!(refund.id, "refunded");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("refunds", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(issue_refund);
        },
    )
}