cleanup

Author: sanderegg

packages/service-library/src/servicelib/rabbitmq/_client.py

@@ -157,6 +157,7 @@ async def subscribe(
         message_handler: MessageHandler,
         *,
         exclusive_queue: bool = True,
+        non_exclusive_queue_name: str | None = None,
         topics: list[str] | None = None,
         message_ttl: NonNegativeInt = RABBIT_QUEUE_MESSAGE_DEFAULT_TTL_MS,
         unexpected_error_retry_delay_s: float = _DEFAULT_UNEXPECTED_ERROR_RETRY_DELAY_S,
@@ -167,6 +168,8 @@ async def subscribe(
             receive the incoming messages
         - exclusive_queue: False means that only one instance of this application will
             reveice the incoming message
+        - non_exclusive_queue_name: if exclusive_queue is False, then this name will be used. If None
+            it will use the exchange_name.
 
         NOTE: ``message_ttl` is also a soft timeout: if the handler does not finish processing
         the message before this is reached the message will be redelivered!
@@ -192,7 +195,7 @@ async def subscribe(
             aio_pika.exceptions.ChannelPreconditionFailed: In case an existing exchange with
             different type is used
         Returns:
-            queue name
+            tuple of queue name and consumer tag mapping
         """
 
         assert self._channel_pool  # nosec
@@ -224,6 +227,7 @@ async def subscribe(
                 self.client_name,
                 exchange_name,
                 exclusive_queue=exclusive_queue,
+                non_exclusive_queue_name=non_exclusive_queue_name,
                 message_ttl=message_ttl,
                 arguments={"x-dead-letter-exchange": delayed_exchange_name},
             )
@@ -243,6 +247,7 @@ async def subscribe(
                 self.client_name,
                 delayed_exchange_name,
                 exclusive_queue=exclusive_queue,
+                non_exclusive_queue_name=non_exclusive_queue_name,
                 message_ttl=int(unexpected_error_retry_delay_s * 1000),
                 arguments={"x-dead-letter-exchange": exchange.name},
             )
@@ -271,6 +276,7 @@ async def add_topics(
                 self.client_name,
                 exchange_name,
                 exclusive_queue=True,
+                non_exclusive_queue_name=None,
                 arguments={
                     "x-dead-letter-exchange": _DELAYED_EXCHANGE_NAME.format(
                         exchange_name=exchange_name
@@ -296,6 +302,7 @@ async def remove_topics(
                 self.client_name,
                 exchange_name,
                 exclusive_queue=True,
+                non_exclusive_queue_name=None,
                 arguments={
                     "x-dead-letter-exchange": _DELAYED_EXCHANGE_NAME.format(
                         exchange_name=exchange_name

packages/service-library/src/servicelib/rabbitmq/_utils.py

@@ -68,6 +68,7 @@ async def declare_queue(
     exchange_name: str,
     *,
     exclusive_queue: bool,
+    non_exclusive_queue_name: str | None,
     arguments: dict[str, Any] | None = None,
     message_ttl: NonNegativeInt = RABBIT_QUEUE_MESSAGE_DEFAULT_TTL_MS,
 ) -> aio_pika.abc.AbstractRobustQueue:
@@ -78,11 +79,11 @@ async def declare_queue(
         "durable": True,
         "exclusive": exclusive_queue,
         "arguments": default_arguments,
-        "name": f"{get_rabbitmq_client_unique_name(client_name)}_{exchange_name}_exclusive",
+        "name": f"{get_rabbitmq_client_unique_name(client_name)}_{non_exclusive_queue_name or exchange_name}_exclusive",
     }
     if not exclusive_queue:
         # NOTE: setting a name will ensure multiple instance will take their data here
-        queue_parameters |= {"name": exchange_name}
+        queue_parameters |= {"name": non_exclusive_queue_name or exchange_name}
 
     # NOTE: if below line raises something similar to ``ChannelPreconditionFailed: PRECONDITION_FAILED``
     # most likely someone changed the signature of the queues (parameters etc...)