have ControlPlaneManager take in queue name generator from SDK/core-service

Signed-off-by: Lance-Drane <Lance-Drane@users.noreply.github.com>

Author: Lance-Drane

src/intersect_sdk_common/control_plane/brokers/amqp_client.py

@@ -11,7 +11,6 @@
 
 import functools
 import threading
-from hashlib import sha384
 from typing import TYPE_CHECKING
 
 import pika
@@ -63,19 +62,6 @@ def __repr__(self) -> str:
         return f'{self.consumer_tag} -- OBTAINED: {self.consumer_tag_obtained()}'
 
 
-def _get_queue_name(routing_key: str) -> str:
-    """Generate a valid queue name from the routing key.
-
-    We want to always be able to generate the same queue name from the routing key every time,
-    so we don't use UUIDs or want the broker to generate a key name.
-
-    We must also keep the length under 128 characters.
-
-    See https://www.rabbitmq.com/docs/queues#names for a complete reference.
-    """
-    return sha384(routing_key.encode()).hexdigest()
-
-
 # TODO we should be handling hierarchy parts as a list of strings until they get to the client
 # this will be a breaking change, so only add it when ready to break
 def _hierarchy_2_amqp(hierarchy: str) -> str:
@@ -418,10 +404,11 @@ def _on_exchange_declareok(self, _unused_frame: Frame, channel: Channel) -> None
                 channel=channel,
                 topic=amqp_topic,
                 persist=topic_handler.topic_persist,
+                queue_name=topic_handler.queue_name_generator(topic),
             )
             self._connection.ioloop.add_callback_threadsafe(cb)
 
-    def _create_queue(self, channel: Channel, topic: str, persist: bool) -> None:
+    def _create_queue(self, channel: Channel, topic: str, persist: bool, queue_name: str) -> None:
         """Create a queue on the broker.
 
         This can be called directly from the AMQP Client if the subscribed connection already has a Channel it's listening to.
@@ -432,12 +419,13 @@ def _create_queue(self, channel: Channel, topic: str, persist: bool) -> None:
             persist: boolean value to determine how we manage the queue.
               If True, this queue will persist forever, even on application or broker shutdown, and we need a persistent name.
               If False, we will generate a temporary queue using the broker's naming scheme.
+            queue_name: The name of the queue to create, if persist is True.
         """
         cb = functools.partial(
             self._on_queue_declareok, channel=channel, topic=topic, persist=persist
         )
         channel.queue_declare(
-            queue=_get_queue_name(topic)
+            queue=queue_name
             if persist
             else '',  # if we're transient, let the broker generate a name for us
             durable=persist,

src/intersect_sdk_common/control_plane/control_plane_manager.py

@@ -44,10 +44,13 @@ class ControlPlaneManager:
     def __init__(
         self,
         control_configs: list[ControlPlaneConfig],
+        queue_name_generator: Callable[[str], str],
     ) -> None:
         """Basic constructor.
 
         Some interaction with message brokers can change based on whether or not a Service or a Client is calling it.
+
+        queue_name_generator should be a hardcoded value for Core Services, the SDK should provide its own function to generate queue names.
         """
         self._control_providers = [
             _create_control_provider(config, self.get_subscription_channels)
@@ -58,6 +61,7 @@ def __init__(
         self._ready = False
         # topics_to_handlers are managed here and transcend connections/disconnections to the broker
         self._topics_to_handlers: dict[str, TopicHandler] = {}
+        self._queue_name_generator = queue_name_generator
 
     def add_subscription_channel(
         self, channel: str, callbacks: set[MessageCallback], persist: bool
@@ -78,7 +82,7 @@ def add_subscription_channel(
         """
         topic_handler = self._topics_to_handlers.get(channel)
         if topic_handler is None:
-            topic_handler = TopicHandler(persist)
+            topic_handler = TopicHandler(persist, self._queue_name_generator)
             topic_handler.callbacks |= callbacks
             self._topics_to_handlers[channel] = topic_handler
         else:

src/intersect_sdk_common/control_plane/topic_handler.py

@@ -1,5 +1,7 @@
 """Attributes associated with a specific pub/sub topic."""
 
+from collections.abc import Callable
+
 from .definitions import MessageCallback
 
 
@@ -14,11 +16,16 @@ class TopicHandler:
     topic_persist: bool
     """Whether or not a topic queue is expected to persist on the message broker."""
 
-    def __init__(self, topic_persist: bool) -> None:
+    queue_name_generator: Callable[[str], str]
+    """A pointer to the function which generates the queue name.."""
+
+    def __init__(self, topic_persist: bool, queue_name_generator: Callable[[str], str]) -> None:
         """Initialize a TopicHandler instance.
 
         Args:
             topic_persist: Whether the topic queue is expected to persist on the message broker.
+            queue_name_generator: A pointer to the function which generates the queue name.
         """
         self.callbacks = set()
         self.topic_persist = topic_persist
+        self.queue_name_generator = queue_name_generator