avoid declaring the wrong queue, drop unreachable block

Author: Bogdanp

dramatiq/brokers/rabbitmq.py

@@ -24,7 +24,6 @@
 from threading import Event, local
 
 import pika
-from pika.exceptions import ChannelClosedByBroker
 
 from ..broker import Broker, Consumer, MessageProxy
 from ..common import current_millis, dq_name, q_name, xq_name
@@ -324,7 +323,7 @@ def enqueue(self, message, *, delay=None):
         attempts = 1
         while True:
             try:
-                self.declare_queue(queue_name, ensure=True)
+                self.declare_queue(message.queue_name, ensure=True)
                 self.logger.debug("Enqueueing message %r on queue %r.", message.message_id, queue_name)
                 self.emit_before("enqueue", message, delay)
                 self.channel.basic_publish(
@@ -345,13 +344,6 @@ def enqueue(self, message, *, delay=None):
                 # next caller/attempt may initiate new ones of each.
                 del self.connection
 
-                # When a queue has been deleted, attempt to get it
-                # redeclared by removing it from the known set. The next
-                # time a message is enqueued -- which could be when we
-                # retry this block -- it will be redeclared.
-                if isinstance(e, ChannelClosedByBroker) and e.reply_code == 404:
-                    self.queues.remove(q_name(queue_name))
-
                 attempts += 1
                 if attempts > MAX_ENQUEUE_ATTEMPTS:
                     raise ConnectionClosed(e) from None

tests/test_rabbitmq.py

@@ -1,6 +1,4 @@
 import os
-import random
-import string
 import time
 from threading import Event
 from unittest.mock import Mock, patch