retry_after

AMQP 1.0 has no delayed redelivery, so retry_after(delay) drops the delivery and publishes a copy to the subscription's own address after the delay.

The runtime rejects the delivery, waits out the delay and publishes a copy of the message, payload and headers, with RETRY_COUNT_HEADER incremented. The copy goes to the subscription's address, because one AMQP node is both what a receiver attaches to and what a sender publishes to, so the mount site names no destination. A bare name and every descriptor answer the same way. On a queue the copy competes for the receivers like any other message. On a topic every subscription on the address receives the copy, not only the one that asked for it. The copy waits in the service's memory, so the retry is at-most-once over the delay: a process that stops before the timer fires loses it. The rejected original is the broker's to handle: ActiveMQ Artemis moves it to its DLQ address by default, beside the copy that comes back. A cap reads the header of the copy, so it holds across delays. In a test the delay runs on the test's clock: tb.advance(delay) brings the copy back.

Example

A rate-limited charge comes back after 30 seconds.

Macros
Manual

examples/amqp_retry.rs
/// A rate-limited charge comes back after the delay; a dropped connection is retried at once.
#[subscriber(AmqpAddress::queue("payments"))]
async fn charge(payment: &Payment) -> HandlerOutcome {
    match charge_card(payment).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(ChargeError::RateLimited) => HandlerOutcome::retry_after(RECHECK),
        Err(ChargeError::Disconnected) => HandlerOutcome::retry(),
    }
}

examples/manual/amqp_retry.rs
/// A rate-limited charge comes back after the delay; a dropped connection is retried at once.
struct Charge;

impl Handle<Payment> for Charge {
    async fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        match charge_card(payment).await {
            Ok(()) => Ok(()),
            Err(ChargeError::RateLimited) => Err(HandlerOutcome::retry_after(RECHECK)),
            Err(ChargeError::Disconnected) => Err(HandlerOutcome::retry()),
        }
    }
}

Example​

Example