dead_letter_exchange

dead_letter_exchange gives a queue its own route for the messages it rejects: the x-dead-letter-exchange argument.

A drop settles with basic.reject without requeue, and the queue publishes the message to that exchange. The route belongs to the queue, so it takes every message the queue dead-letters, whatever the cause. Besides a rejection, RabbitMQ dead-letters a message whose TTL expired, one pushed out by a length limit, and one that spent a quorum queue's delivery limit. The message keeps its routing key, unless dead_letter_routing_key names another (x-dead-letter-routing-key). The dead-letter exchange routes like any other exchange: a queue bound to it under that key collects the messages. RabbitQuorumQueue takes the same two settings.

Both settings are queue arguments. The queue gets them when the service declares it, with declare_topology(true) on the broker. A queue the operator provisions carries the arguments it was declared with. The dead_letter(name) step of the mount chain is a different thing: the destination of a registration's retry cap.

Example

An invoice with a negative amount is dropped. The queue dead-letters it to invoices.dlx, where a second subscription reads it from the bound queue.

Macros
Manual

examples/lapin_settlement_dead_letter.rs
/// An invoice with a negative amount is rejected for good. The queue dead-letters it to
/// `invoices.dlx` under the routing key `invoices.rejected`.
#[subscriber(RabbitQueue::new("invoices")
    .dead_letter_exchange("invoices.dlx")
    .dead_letter_routing_key("invoices.rejected"))]
async fn book(invoice: &Invoice) -> HandlerOutcome {
    if invoice.amount_cents < 0 {
        tracing::warn!(invoice.id, "negative amount, rejected");
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

/// The rejected invoices wait here, bound to the dead-letter exchange under that key.
#[subscriber(RabbitQueue::new("invoices.rejected")
    .bind(RabbitExchange::direct("invoices.dlx"), "invoices.rejected"))]
async fn review(invoice: &Invoice) -> HandlerOutcome {
    tracing::error!(invoice.id, "invoice needs a review");
    HandlerOutcome::ack()
}

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

examples/manual/lapin_settlement_dead_letter.rs
/// An invoice with a negative amount is rejected for good. The queue dead-letters it to
/// `invoices.dlx` under the routing key `invoices.rejected`.
struct Book;

impl Handle<Invoice> for Book {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if invoice.amount_cents < 0 {
            tracing::warn!(invoice.id, "negative amount, rejected");
            return ready(Err(HandlerOutcome::drop()));
        }
        ready(Ok(()))
    }
}

/// The rejected invoices wait here, bound to the dead-letter exchange under that key.
struct Review;

impl Handle<Invoice> for Review {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::error!(invoice.id, "invoice needs a review");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        let invoices = RabbitQueue::new("invoices")
            .dead_letter_exchange("invoices.dlx")
            .dead_letter_routing_key("invoices.rejected");
        let rejected = RabbitQueue::new("invoices.rejected")
            .bind(RabbitExchange::direct("invoices.dlx"), "invoices.rejected");
        b.include(subscriber(invoices, Book).build());
        b.include(subscriber(rejected, Review).build());
    })
}

Example​

Example