Queue Arguments#

Beyond durable, exclusive and auto_delete, RabbitMQ configures a queue through optional arguments: a dictionary of x-* keys sent with the declaration. They control message TTL, queue length, overflow behavior, dead-lettering, priorities and more. The full list lives in the RabbitMQ documentation.

RabbitQueue forwards them through the arguments parameter:

from faststream.rabbit import RabbitQueue

queue = RabbitQueue(
    "orders",
    arguments={
        "x-message-ttl": 60_000,
        "x-max-length": 10_000,
        "x-overflow": "reject-publish",
    },
)

x-queue-type is set for you from the queue_type parameter, so don't put it into arguments. The accepted keys depend on that type, and RabbitQueue is typed accordingly: your type checker flags a key the chosen queue type doesn't support, and your editor can complete the rest.

Common Arguments#

Arguments that belong to the consumer rather than the queue (x-stream-offset, x-priority) go to the subscriber's consume_args instead, as the stream example shows.

Dead Letter Queue#

When a consumer rejects a message, or a message sits in a queue longer than its TTL, RabbitMQ can forward it to a dead letter exchange instead of dropping it. A queue bound to that exchange collects those messages, so you can inspect, alert on, or replay them later.

The dead-lettering rules live on the source queue as the x-dead-letter-exchange and x-dead-letter-routing-key arguments. The dead letter exchange and the queue bound to it are ordinary objects that you declare like any other:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

from faststream import FastStream, Logger
from faststream.exceptions import RejectMessage
from faststream.rabbit import (
    ExchangeType,
    RabbitBroker,
    RabbitExchange,
    RabbitMessage,
    RabbitQueue,
)

broker = RabbitBroker()
app = FastStream(broker)

dead_letter_exchange = RabbitExchange("orders-dlx", type=ExchangeType.DIRECT)
dead_letter_queue = RabbitQueue("orders-dead", routing_key="orders")

orders_queue = RabbitQueue(
    "orders",
    arguments={
        "x-dead-letter-exchange": "orders-dlx",
        "x-dead-letter-routing-key": "orders",
        "x-message-ttl": 60_000,
    },
)


@broker.subscriber(dead_letter_queue, dead_letter_exchange)
async def handle_dead_letter(
    order_id: str,
    msg: RabbitMessage,
    logger: Logger,
) -> None:
    reason = msg.headers["x-death"][0]["reason"]
    logger.warning("order %s dead-lettered: %s", order_id, reason)


@broker.subscriber(orders_queue)
async def handle_order(order_id: str, logger: Logger) -> None:
    if order_id.startswith("bad"):
        raise RejectMessage

    logger.info("processed %s", order_id)

A rejected order goes to orders-dlx with the routing key orders, lands in orders-dead and reaches handle_dead_letter as the very same message, plus an x-death header that says why it was dead-lettered (rejected, expired, maxlen or delivery_limit) and from which queue. Subscribers start in the order they are declared, each one declaring its queue and exchange and consuming right away, so declare the dead letter subscriber first: orders-dlx and orders-dead then exist before the first order can be rejected.

The same path is taken by any message that expires after x-message-ttl milliseconds, and by messages a handler fails on with an unhandled exception, since the default acknowledgement policy rejects those too.

Use a dead letter queue whenever a handler can fail for a reason a retry won't fix (a malformed payload, a business rule violation): requeuing such a message only makes the same handler fail again. For transient failures, use a quorum queue with x-delivery-limit and ack_policy=AckPolicy.NACK_ON_ERROR: a failed message is requeued and redelivered that many times first, and dead-lettered only after that. Under the default REJECT_ON_ERROR policy a failure rejects the message without requeue, so it is dead-lettered right away.