From 4c7c4b466e2b5228a6098c9130cd6c3334315153 Mon Sep 17 00:00:00 2001 From: Matthias Bisping Date: Tue, 15 Mar 2022 09:46:39 +0100 Subject: [PATCH] set dead letter queue related params --- pyinfra/pyampq/queue_manager/pika_queue_manager.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/pyinfra/pyampq/queue_manager/pika_queue_manager.py b/pyinfra/pyampq/queue_manager/pika_queue_manager.py index da56823..a0f34ed 100644 --- a/pyinfra/pyampq/queue_manager/pika_queue_manager.py +++ b/pyinfra/pyampq/queue_manager/pika_queue_manager.py @@ -63,12 +63,19 @@ def attempts_remain(n_attempts, max_attempts): class PikaQueueManager(QueueManager): - def __init__(self, input_queue, output_queue): + def __init__(self, input_queue, output_queue, dead_letter_queue=None): super().__init__(input_queue, output_queue) + self.connection = get_connection() self.channel = self.connection.channel() - self.channel.queue_declare(input_queue) - self.channel.queue_declare(output_queue) + + if not dead_letter_queue: + dead_letter_queue = CONFIG.rabbitmq.queues.dead_letter + + args = {"x-dead-letter-exchange": "", "x-dead-letter-routing-key": dead_letter_queue} + + self.channel.queue_declare(input_queue, arguments=args, auto_delete=False, durable=False) + self.channel.queue_declare(output_queue, arguments=args, auto_delete=False, durable=False) def republish(self, body, n_current_attempts, frame): self.channel.basic_publish(