allow queue name to be specified by libraries, add is_root option to configuration

also add nonroot handling for AMQP protocol

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

Author: Lance-Drane

CHANGELOG.md

@@ -2,8 +2,13 @@
 
 We follow [Common Changelog](https://common-changelog.org/) formatting for this document.
 
-## [0.9.0] - 2026-02-11
+## [0.9.1] - 2026-02-25
 
 Initial reorganization of SDK packaging.
 
-[0.9.0]: https://github.com/INTERSECT-SDK/python-sdk-common/releases/tag/0.9.0
+### Changed
+
+- Added new argument to `ControlPlaneManager.add_subscription_channel()` which specifies a queue name to use.
+- Add `is_root` option to `ControlPlaneConfig` in preparation for Registry Service.
+
+[0.9.1]: https://github.com/INTERSECT-SDK/python-sdk-common/releases/tag/0.9.1

pyproject.toml

@@ -18,7 +18,7 @@ readme = "README.md"
 license = { text = "BSD-3-Clause" }
 requires-python = ">=3.10,<4.0"
 keywords = ["intersect"]
-version = "0.9.0"
+version = "0.9.1a"
 classifiers = [
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3.10",

src/intersect_sdk_common/config.py

@@ -122,6 +122,21 @@ class ControlPlaneConfig:
     NOTE: INTERSECT currently only supports AMQP and MQTT.
     """
 
+    # TODO default this to False once the registry service is in place
+    is_root: bool = True
+    """
+    Whether or not the broker credentials are for connecting as a root user.
+
+    This should be True IF:
+      - You are a Core Service
+      - You are an SDK Client or Service, but your message broker is hosted locally.
+
+      This should be False IF:
+        - You are an SDK Client or Service, and the broker you're connected to is remote.
+
+    This is important for specific implementations; the Registry Service configures queues for microservices, but Core Services configure their own queues themselves.
+    """
+
 
 @dataclass
 class DataStoreConfig:

src/intersect_sdk_common/control_plane/brokers/amqp_client.py

@@ -96,6 +96,7 @@ def __init__(
         username: str,
         password: str,
         topics_to_handlers: Callable[[], dict[str, TopicHandler]],
+        is_root: bool,
     ) -> None:
         """The default constructor.
 
@@ -105,6 +106,7 @@ def __init__(
             username: username credentials for AMQP broker
             password: password credentials for AMQP broker
             topics_to_handlers: callback function which gets the topic to handler map from the channel manager
+            is_root: Whether or not the client can configure exchanges and queues themselves (core services), or if this must be delegated to a Core Service (SDK Clients/Services)
         """
         self._connection_params = pika.ConnectionParameters(
             host=host,
@@ -118,6 +120,8 @@ def __init__(
             retry_delay=0.5,
         )
 
+        self._is_root = is_root
+
         # The pika connection to the broker
         self._connection: pika.adapters.SelectConnection = None
         self._channel_in: Channel = None
@@ -225,7 +229,6 @@ def subscribe(self, topic: str, persist: bool) -> None:
         persist: If True, we will create an idempotent queue name which should persist
           even on broker or application shutdown. If False, we will allow the server to create a unique
           queue name, and the queue will be destroyed once the associated channel is closed.
-
         """
         topic = _hierarchy_2_amqp(topic)
         cb = functools.partial(
@@ -365,13 +368,17 @@ def _on_output_channel_open(self, channel: Channel) -> None:
         self._channel_out = channel
         cb = functools.partial(self._on_channel_closed, channel_num=channel_num)
         self._channel_out.add_on_close_callback(cb)
-        # producer flag should first make sure the exchange exists before publishing
-        channel.exchange_declare(
-            exchange=_INTERSECT_MESSAGE_EXCHANGE,
-            exchange_type='topic',
-            durable=True,
-            callback=lambda _frame: self._channel_flags.set_nth_flag(channel_num),
-        )
+        if self._is_root:
+            # if root, producer flag should first make sure the exchange exists before publishing
+            channel.exchange_declare(
+                exchange=_INTERSECT_MESSAGE_EXCHANGE,
+                exchange_type='topic',
+                durable=True,
+                callback=lambda _frame: self._channel_flags.set_nth_flag(channel_num),
+            )
+        else:
+            # nonroot producers can assume they may publish immediately
+            self._channel_flags.set_nth_flag(channel_num)
         logger.info('AMQP: output channel ready')
 
     # CONSUMER #
@@ -382,10 +389,26 @@ def _on_input_channel_open(self, channel: Channel) -> None:
         self._channel_flags.set_nth_flag(channel_num)
         cb_1 = functools.partial(self._on_channel_closed, channel_num=channel_num)
         self._channel_in.add_on_close_callback(cb_1)
-        cb_2 = functools.partial(self._on_exchange_declareok, channel=channel)
-        channel.exchange_declare(
-            exchange=_INTERSECT_MESSAGE_EXCHANGE, exchange_type='topic', durable=True, callback=cb_2
-        )
+        if self._is_root:
+            cb_2 = functools.partial(self._on_exchange_declareok, channel=channel)
+            channel.exchange_declare(
+                exchange=_INTERSECT_MESSAGE_EXCHANGE,
+                exchange_type='topic',
+                durable=True,
+                callback=cb_2,
+            )
+        else:
+            for topic, topic_handler in self._topics_to_handlers().items():
+                amqp_topic = _hierarchy_2_amqp(topic)
+                cb = functools.partial(
+                    self._setup_basic_qos,
+                    None,
+                    channel=channel,
+                    topic=amqp_topic,
+                    persist=topic_handler.topic_persist,
+                    queue_name=topic_handler.queue_name,
+                )
+                self._connection.ioloop.add_callback_threadsafe(cb)
 
     def _on_exchange_declareok(self, _unused_frame: Frame, channel: Channel) -> None:
         """Create a queue on the broker (called from AMQP).
@@ -404,7 +427,7 @@ 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),
+                queue_name=topic_handler.queue_name,
             )
             self._connection.ioloop.add_callback_threadsafe(cb)
 
@@ -448,7 +471,7 @@ def _on_queue_declareok(
         """
         queue_name = frame.method.queue
         cb = functools.partial(
-            self._on_queue_bindok,
+            self._setup_basic_qos,
             channel=channel,
             topic=topic,
             queue_name=queue_name,
@@ -461,9 +484,9 @@ def _on_queue_declareok(
             callback=cb,
         )
 
-    def _on_queue_bindok(
+    def _setup_basic_qos(
         self,
-        _unused_frame: Frame,
+        _unused_frame: Frame | None,
         channel: Channel,
         topic: str,
         queue_name: str,
@@ -472,7 +495,7 @@ def _on_queue_bindok(
         """Sets up basic QOS on the current channel.
 
         Args:
-            _unused_frame: AMQP response from binding to the queue. Ignored.
+            _unused_frame: AMQP response from binding to the queue. Ignored. Also explicitly set to "None" if called from nonroot configuration.
             channel: The Channel being instantiated.
             topic: Name of the topic on the broker.
             queue_name: The name of the queue on the AMQP broker.

src/intersect_sdk_common/control_plane/control_plane_manager.py

@@ -24,6 +24,7 @@ def _create_control_provider(
             username=config.username,
             password=config.password,
             topics_to_handlers=topic_handler_callback,
+            is_root=config.is_root,
         )
 
     # MQTT
@@ -44,7 +45,6 @@ class ControlPlaneManager:
     def __init__(
         self,
         control_configs: list[ControlPlaneConfig],
-        queue_name_generator: Callable[[str], str],
     ) -> None:
         """Basic constructor.
 
@@ -61,10 +61,9 @@ 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
+        self, channel: str, callbacks: set[MessageCallback], persist: bool, queue_name: str
     ) -> None:
         """Start listening for messages on a channel on all configured brokers.
 
@@ -79,14 +78,17 @@ def add_subscription_channel(
           callbacks: functions to call on subscribing to a message
           persist: if True, expect the associated message queue to live long; if False, it will only live the duration of the application.
             Any queue associated with a Service should always set this to True. Clients will need to subscribe to their own, temporary queues, and should set this to False.
+          queue_name: the name of the queue to subscribe to. This is generally hardcoded for Core Services, but autogenerated for SDK Services and Clients.
         """
         topic_handler = self._topics_to_handlers.get(channel)
         if topic_handler is None:
-            topic_handler = TopicHandler(persist, self._queue_name_generator)
+            topic_handler = TopicHandler(persist, queue_name)
             topic_handler.callbacks |= callbacks
             self._topics_to_handlers[channel] = topic_handler
         else:
             topic_handler.callbacks |= callbacks
+            topic_handler.topic_persist = persist
+            topic_handler.queue_name = queue_name
         if self.is_connected():
             for provider in self._control_providers:
                 provider.subscribe(channel, persist)

src/intersect_sdk_common/control_plane/topic_handler.py

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