Skip to main content

Dead letters

dead_letter(address) names the address a spent delivery is published to; a cap without one rejects it, and the broker's own dead-letter policy applies.

It follows max_attempts(n) at the mount site. When the delivery that fails is the n-th one, the runtime rejects it and publishes a copy to the dead-letter address instead of the subscription's. The copy keeps the payload and the headers, and RETRY_COUNT_HEADER on it says how many times the message was delivered. It leaves through the registration's retry publisher, so the steps that follow out_retry apply to it too. The dead-letter address is an ordinary address: a subscription of this service, or of another one, reads it. A cap without dead_letter rejects the spent delivery, and a warning names the subscription. Where a rejected delivery goes is the broker's policy: ActiveMQ Artemis moves it to its DLQ address by default. The original of a dead-letter copy is rejected as well, so on Artemis the spent delivery also reaches DLQ. In process a rejected delivery leaves the transport, so a test asserts the copy on the named address, as the example's test does.

Retries and dead letters reads the dead letters of a refunds service.

Example​

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);
    })
}

The dead-letter address is read by an ordinary subscription:

examples/amqp_retry.rs
/// The dead-letter address is read by an ordinary subscription. The framework's retry-count
/// header says how many times the payment was delivered before it was carried here.
#[subscriber(AmqpAddress::queue("payments.dead"))]
async fn on_dead_letter(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(payment.id, deliveries, "payment dead-lettered");
    HandlerOutcome::ack()
}