Retries

max_attempts(n) caps deliveries; the count is the broker's delivery-count, or the framework's header on a copy this crate published.

The cap counts the deliveries of one message at the mount site, the first one included. A delivery whose message carries the AMQP header section reports the broker's delivery-count, which every retry increments. A message this crate published carries no header section, so the broker has counted nothing for it, and the count is RETRY_COUNT_HEADER instead. Under a cap, a retry of such a message is the runtime's copy, not the broker's redelivery: the runtime rejects the delivery and publishes it again to the subscription's address at once, with the header incremented. A delivery that reaches the cap is not retried: it goes to the dead-letter address, or is rejected where the registration names none. A retry on a registration with no cap stays the broker's own redelivery, and nothing limits it. The runtime rejects every original it replaces with a copy, so a broker that dead-letters rejected deliveries keeps those originals: ActiveMQ Artemis moves them to its DLQ address by default. The copies leave through the retry publisher of the registration.

Retries and dead letters caps the attempts of a refunds service.

Example

Five deliveries of one payment; the fifth failure goes to payments.dead.

Macros

examples/amqp_retry.rs

#[ruststream::app]
fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        // Five deliveries of one payment, the first included; the fifth failure goes to
        // "payments.dead" instead. Every copy leaves through `Publish`, stamped by `Retried`.
        b.include(charge)
            .max_attempts(nonzero!(5u32))
            .dead_letter("payments.dead")
            .out_retry(Publish)
            .transform(Retried);
        b.include(on_dead_letter);
    })
}

Manual

examples/manual/amqp_retry.rs

fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        // Five deliveries of one payment, the first included; the fifth failure goes to
        // "payments.dead" instead. Every copy leaves through `Publish`, stamped by `Retried`.
        b.include(subscriber(AmqpAddress::queue("payments"), Charge).build())
            .max_attempts(nonzero!(5u32))
            .dead_letter("payments.dead")
            .out_retry(Publish)
            .transform(Retried);
        b.include(subscriber(AmqpAddress::queue("payments.dead"), OnDeadLetter).build());
    })
}