Pidbox: support queue_exclusive / queue_durable to fix deprecation (RabbitMQ 4.x) (#2338)

* Pidbox: add queue_exclusive / queue_durable flags

* Update kombu/pidbox.py

* update pidbox documentation

* update documentation

---------

Co-authored-by: ghirailghiro <ghirardellim2@gmail.com>
Co-authored-by: Asif Saif Uddin {"Auvi":"অভি"} <auvipy@gmail.com>

Author: ghirailghiro

docs/reference/kombu.pidbox.rst

@@ -70,6 +70,20 @@
         .. automethod:: get_reply_queue
         .. automethod:: get_queue
 
+    Mailbox Options
+    ~~~~~~~~~~~~~~~
+
+    .. versionadded:: 5.6.0
+
+    The `Mailbox` supports several configuration options that affect
+    the behavior of its exchanges and queues.
+
+    - ``durable``: If True, declares durable exchanges that survive broker restarts.
+    - ``exclusive``: If True, declares exclusive exchanges (usable by only one connection).
+
+    These provide finer control over broker-side behavior and are useful
+    in production environments where queue durability matters.\
+
     Node
     ----
 
@@ -91,4 +105,3 @@
         .. automethod:: handle
         .. automethod:: handle_message
         .. automethod:: reply
-

kombu/pidbox.py

@@ -181,6 +181,7 @@ def __init__(self, namespace,
                  type='direct', connection=None, clock=None,
                  accept=None, serializer=None, producer_pool=None,
                  queue_ttl=None, queue_expires=None,
+                 queue_durable=False, queue_exclusive=False,
                  reply_queue_ttl=None, reply_queue_expires=10.0):
         self.namespace = namespace
         self.connection = connection
@@ -193,9 +194,16 @@ def __init__(self, namespace,
         self.serializer = self.serializer if serializer is None else serializer
         self.queue_ttl = queue_ttl
         self.queue_expires = queue_expires
+        self.queue_durable = queue_durable
+        self.queue_exclusive = queue_exclusive
         self.reply_queue_ttl = reply_queue_ttl
         self.reply_queue_expires = reply_queue_expires
         self._producer_pool = producer_pool
+        if queue_exclusive and queue_durable:
+            raise ValueError(
+                "queue_exclusive and queue_durable cannot both be True "
+                "(exclusive queues are automatically deleted and cannot be durable).",
+            )
 
     def __call__(self, connection):
         bound = copy(self)
@@ -236,8 +244,9 @@ def get_reply_queue(self):
             f'{oid}.{self.reply_exchange.name}',
             exchange=self.reply_exchange,
             routing_key=oid,
-            durable=False,
-            auto_delete=True,
+            durable=self.queue_durable,
+            exclusive=self.queue_exclusive,
+            auto_delete=not self.queue_durable,
             expires=self.reply_queue_expires,
             message_ttl=self.reply_queue_ttl,
         )
@@ -250,8 +259,9 @@ def get_queue(self, hostname):
         return Queue(
             f'{hostname}.{self.namespace}.pidbox',
             exchange=self.exchange,
-            durable=False,
-            auto_delete=True,
+            durable=self.queue_durable,
+            exclusive=self.queue_exclusive,
+            auto_delete=not self.queue_durable,
             expires=self.queue_expires,
             message_ttl=self.queue_ttl,
         )

t/unit/test_pidbox.py

@@ -342,6 +342,38 @@ def get_next(self, consumer):
         if m:
             return m.payload
 
+    def test_mailbox_queue_exclusive(self):
+        mbox = pidbox.Mailbox(
+            'flagbox_ex',
+            queue_exclusive=True,
+            queue_durable=False,
+        )(self.connection)
+
+        for q in (mbox.get_queue('worker1'), mbox.get_reply_queue()):
+            assert q.exclusive is True
+            assert q.durable is False
+            assert q.auto_delete is True
+
+    def test_mailbox_queue_durable(self):
+        mbox = pidbox.Mailbox(
+            'flagbox_dur',
+            queue_exclusive=False,
+            queue_durable=True,
+        )(self.connection)
+
+        for q in (mbox.get_queue('worker1'), mbox.get_reply_queue()):
+            assert q.durable is True
+            assert q.exclusive is False
+            assert q.auto_delete is False
+
+    def test_mailbox_invalid_flag_combo(self):
+        with pytest.raises(ValueError):
+            pidbox.Mailbox(
+                'flagbox_bad',
+                queue_exclusive=True,
+                queue_durable=True,
+            )(self.connection)
+
 
 GLOBAL_PIDBOX = pidbox.Mailbox('global_unittest_mailbox')