fix wildcard handling for protocols

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

Author: Lance-Drane

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 We follow [Common Changelog](https://common-changelog.org/) formatting for this document.
 
+## Unreleased
+
+### Fixed
+
+- Made support for wildcards in the channel parameter of `ControlPlaneManager.add_subscription_channel()` consistent across protocols
+
 ## [0.9.1] - 2026-02-25
 
 Initial reorganization of SDK packaging.

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]],
+        find_wildcard_handler: Callable[[str], TopicHandler | None],
         is_root: bool,
     ) -> None:
         """The default constructor.
@@ -106,6 +107,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
+            find_wildcard_handler: callback function which gets the wildcard topic handler 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(
@@ -131,6 +133,8 @@ def __init__(
 
         # Callback to the topics_to_handler list inside of
         self._topics_to_handlers = topics_to_handlers
+        self._find_wildcard_handler = find_wildcard_handler
+
         # mapping of topics to callables which can unsubscribe from the topic
         self._topics_to_consumer_tags: dict[str, _ConsumerTagInfo] = {}
         self._consumer_tags_to_threads: dict[str, threading.Thread] = {}
@@ -595,6 +599,9 @@ def _consume_message(
 
         tth_key = _amqp_2_hierarchy(basic_deliver.routing_key)
         topic_handler = self._topics_to_handlers().get(tth_key)
+        if not topic_handler:
+            # we may have included the topic in one of our wildcard handlers
+            topic_handler = self._find_wildcard_handler(tth_key)
         if topic_handler:
             consumer_tag_info = self._topics_to_consumer_tags.get(basic_deliver.routing_key)
             if not consumer_tag_info:

src/intersect_sdk_common/control_plane/brokers/mqtt_client.py

@@ -33,6 +33,13 @@
 _MQTT_MAX_RETRIES = 10
 
 
+# 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_mqtt(hierarchy: str) -> str:
+    """Take the hierarchy string format saved in the Service and map it to the MQTT topic format. Currently just covers wildcards."""
+    return hierarchy.replace('#', '+')
+
+
 class MQTTClient(BrokerClient):
     """Client for performing broker actions backed by a MQTT broker.
 
@@ -55,6 +62,7 @@ def __init__(
         username: str,
         password: str,
         topics_to_handlers: Callable[[], dict[str, TopicHandler]],
+        find_wildcard_handler: Callable[[str], TopicHandler | None],
         uid: str | None = None,
     ) -> None:
         """The default constructor.
@@ -65,6 +73,7 @@ def __init__(
             username: username credentials for MQTT broker
             password: password credentials for MQTT broker
             topics_to_handlers: callback function which gets the topic to handler map from the channel manager
+            find_wildcard_handler: callback function which gets the wildcard topic handler from the channel manager
             uid: A string representing the unique id to identify the client.
         """
         # Unique id for the MQTT broker to associate this client with
@@ -89,6 +98,7 @@ def __init__(
 
         # ConnectionManager callable state
         self._topics_to_handlers = topics_to_handlers
+        self._find_wildcard_handler = find_wildcard_handler
 
         # MQTT v3.1.1 automatically downgrades a QOS which is too high (good), but MQTT v5 will terminate the connection (bad)
         # see https://github.com/rabbitmq/rabbitmq-server/discussions/11842
@@ -155,8 +165,9 @@ def publish(
         props = Properties(PacketTypes.PUBLISH)  # type: ignore[no-untyped-call]
         props.ContentType = content_type
         props.UserProperty = list(headers.items())
+        resolved_topic = _hierarchy_2_mqtt(topic)
         self._connection.publish(
-            topic, payload, qos=self._max_supported_qos if persist else 0, properties=props
+            resolved_topic, payload, qos=self._max_supported_qos if persist else 0, properties=props
         )
 
     def subscribe(self, topic: str, persist: bool) -> None:
@@ -166,16 +177,18 @@ def subscribe(self, topic: str, persist: bool) -> None:
             topic: Topic to subscribe to.
             persist: Determine if the associated message queue of the topic is long-lived (True) or not (False)
         """
+        resolved_topic = _hierarchy_2_mqtt(topic)
         # NOTE: RabbitMQ only works with QOS of 1 and 0, and seems to convert QOS2 to QOS1
-        self._connection.subscribe(topic, qos=2 if persist else 0, properties=None)
+        self._connection.subscribe(resolved_topic, qos=2 if persist else 0, properties=None)
 
     def unsubscribe(self, topic: str) -> None:
         """Unsubscribe from a topic over the pre-existing connection.
 
         Args:
           topic: Topic to unsubscribe from.
         """
-        self._connection.unsubscribe(topic)
+        resolved_topic = _hierarchy_2_mqtt(topic)
+        self._connection.unsubscribe(resolved_topic)
 
     def _on_message(
         self,
@@ -190,7 +203,11 @@ def _on_message(
           userdata: MQTT user data
           message: MQTT message
         """
+        # NOTE: should not need to convert the MQTT topic to the protocol agnostic representation, as we only change the wildcard format (which won't show up as the message topic)
         topic_handler = self._topics_to_handlers().get(message.topic)
+        if not topic_handler:
+            # we may have included the topic in one of our wildcard handlers
+            topic_handler = self._find_wildcard_handler(message.topic)
         # Note that if we return prior to the callback, there will be no reply message
         if not topic_handler:
             logger.warning('Incompatible message topic %s, rejecting message', message.topic)

src/intersect_sdk_common/control_plane/control_plane_manager.py

@@ -1,17 +1,30 @@
 """Wrapper around interacting with the control plane, which may be multiple brokers."""
 
+import re
 from collections.abc import Callable
 
 from ..config import ControlPlaneConfig
+from ..exceptions import IntersectSetupError
 from ..logger import logger
 from .brokers.broker_client import BrokerClient
 from .definitions import MessageCallback
 from .topic_handler import TopicHandler
 
+_CHANNEL_REGEX = re.compile(r'^[a-zA-Z0-9*/-]+[a-zA-Z0-9*#/-]$')
+"""Allow for these patterns:
+
+- alphanumeric characters
+- hyphens
+- `/` (topic separator; note that this is '.' on the broker)
+- `*` (wildcard for exactly one word)
+- `#` (wildcard for any number of words, but can ONLY appear at the end of the topic string, see section 4.7.1.2 at https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718107)
+"""
+
 
 def _create_control_provider(
     config: ControlPlaneConfig,
     topic_handler_callback: Callable[[], dict[str, TopicHandler]],
+    find_wildcard_handler: Callable[[str], TopicHandler | None],
 ) -> BrokerClient:
     if config.protocol == 'amqp0.9.1':
         from .brokers.amqp_client import (  # noqa: PLC0415 (lazy load all AMQP modules)
@@ -24,6 +37,7 @@ def _create_control_provider(
             username=config.username,
             password=config.password,
             topics_to_handlers=topic_handler_callback,
+            find_wildcard_handler=find_wildcard_handler,
             is_root=config.is_root,
         )
 
@@ -36,6 +50,7 @@ def _create_control_provider(
         username=config.username,
         password=config.password,
         topics_to_handlers=topic_handler_callback,
+        find_wildcard_handler=find_wildcard_handler,
     )
 
 
@@ -53,38 +68,53 @@ def __init__(
         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)
+            _create_control_provider(
+                config, self.get_subscription_channels, self.get_wildcard_topic_handler
+            )
             for config in control_configs
         ]
 
         # flag which indicates if we SHOULD be connected.
         self._ready = False
         # topics_to_handlers are managed here and transcend connections/disconnections to the broker
         self._topics_to_handlers: dict[str, TopicHandler] = {}
+        self._wildcards: dict[str, TopicHandler] = {}
 
     def add_subscription_channel(
         self, channel: str, callbacks: set[MessageCallback], persist: bool, queue_name: str
     ) -> None:
         """Start listening for messages on a channel on all configured brokers.
 
-        Use the format ${TOPIC}/${TOPIC}/${TOPIC} ... for the channel name, protocol implementations
-        are responsible for converting the string to a valid channel.
+        The provided channel value should be PROTOCOL AGNOSTIC. Use the following symbols when constructing a topic:
+        - `/` - topic separator
+        - `*` - exactly one word
+        - `#` - any number of words (however, this can only be used as the LAST character)
+
+        Protocol implementations are responsible for converting the string to a valid channel.
 
         This function should usually only be called before you've connected,
         but it's okay to call it after connecting. (Mostly used for Clients.)
 
         Params:
-          channel: string of the channel which we should start listening to
+          channel: string of the channel which we should start listening to, remember to follow the above protocol agnostic convention
           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)
+        # this currently follows the MQTT/AMQP convention for wildcards
+        # '*' = exactly one word (AMQP), '+' = exactly one word (MQTT), '#' = any number of words (AMQP and MQTT)
+        if not _CHANNEL_REGEX.match(channel):
+            msg = f'Channel {channel} is not valid, it must be alphanumeric and can only contain the following special characters: * # / -'
+            raise IntersectSetupError(msg)
+        is_wildcard = '*' in channel or channel.endswith('#')
+        dict_to_change = self._wildcards if is_wildcard else self._topics_to_handlers
+
+        topic_handler = dict_to_change.get(channel)
         if topic_handler is None:
-            topic_handler = TopicHandler(persist, queue_name)
+            topic_handler = TopicHandler(persist, queue_name, channel if is_wildcard else None)
             topic_handler.callbacks |= callbacks
-            self._topics_to_handlers[channel] = topic_handler
+            dict_to_change[channel] = topic_handler
         else:
             topic_handler.callbacks |= callbacks
             topic_handler.topic_persist = persist
@@ -103,23 +133,38 @@ def remove_subscription_channel(self, channel: str) -> bool:
         try:
             del self._topics_to_handlers[channel]
         except KeyError:
-            return False
-        else:
-            if self.is_connected():
-                for provider in self._control_providers:
-                    provider.unsubscribe(channel)
-            return True
+            try:
+                del self._wildcards[channel]
+            except KeyError:
+                return False
+        if self.is_connected():
+            for provider in self._control_providers:
+                provider.unsubscribe(channel)
+        return True
 
     def get_subscription_channels(self) -> dict[str, TopicHandler]:
         """Get the subscription channels.
 
-        Note that this function gets accessed as a callback from the direct broker implementations.
+        These channels cannot be wildcards, and incoming topics must match the channel topics exactly. Note that this function gets accessed as a callback from the direct broker implementations.
 
         Returns:
           the dictionary of topics to topic information
         """
         return self._topics_to_handlers
 
+    def get_wildcard_topic_handler(self, topic: str) -> TopicHandler | None:
+        """Get the wildcard topic handler for a given topic, if it exists.
+
+        This is an inefficient lookup, so should only be used if we fail to find a non-wildcard match for an incoming topic. Note that this function gets accessed as a callback from the direct broker implementations.
+
+        Returns:
+          the topic handler associated with the topic, if it exists; None otherwise
+        """
+        for topic_handler in self._wildcards.values():
+            if topic_handler.does_topic_match(topic):
+                return topic_handler
+        return None
+
     def connect(self) -> None:
         """Connect to all configured brokers.
 

src/intersect_sdk_common/control_plane/topic_handler.py

@@ -1,5 +1,7 @@
 """Attributes associated with a specific pub/sub topic."""
 
+import re
+
 from .definitions import MessageCallback
 
 
@@ -17,13 +19,45 @@ class TopicHandler:
     queue_name: str
     """The name of the queue to subscribe to for this topic."""
 
-    def __init__(self, topic_persist: bool, queue_name: str) -> None:
+    _regex_pattern: re.Pattern[str] | None = None
+    """If this is a wildcard topic, the regex pattern to match incoming topics against (if no wildcards, this will be None)."""
+
+    def __init__(
+        self, topic_persist: bool, queue_name: str, regex_pattern: str | None = None
+    ) -> None:
         """Initialize a TopicHandler instance.
 
         Args:
             topic_persist: Whether the topic queue is expected to persist on the message broker.
             queue_name: The name of the queue to subscribe to for this topic.
+            regex_pattern: If this is a wildcard topic, the regex pattern to match incoming topics against (if no wildcards, leave as None).
         """
         self.callbacks = set()
         self.topic_persist = topic_persist
         self.queue_name = queue_name
+
+        if regex_pattern:
+            regex_builder = ['^']
+            for char in regex_pattern:
+                if char == '*':
+                    # match a single word (do not include the separator)
+                    regex_builder.append('[a-zA-Z0-9-]+')
+                elif char == '#':
+                    # match any sequence of 0 or more words (make sure to allow for the separator)
+                    regex_builder.append('[a-zA-Z0-9/-]*')
+                elif char == '/':
+                    # in-memory topics use '/' as the separator, but on the broker '.' is used
+                    regex_builder.append('\\.')
+                else:
+                    # ordinary character
+                    regex_builder.append(char)
+            regex_builder.append('$')
+            self._regex_pattern = re.compile(''.join(regex_builder))
+        else:
+            self._regex_pattern = None
+
+    def does_topic_match(self, topic: str) -> bool:
+        """Check if a given topic could be caught by the TopicHandler's topic regex. Only used for wildcard topics."""
+        if self._regex_pattern:
+            return bool(self._regex_pattern.match(topic))
+        return False

src/intersect_sdk_common/exceptions.py

@@ -17,3 +17,10 @@ class IntersectApplicationError(IntersectError):
 
     This exception should strictly be used for control flow - it should NEVER be a fatal exception
     """
+
+
+class IntersectSetupError(IntersectError):
+    """Exception thrown when there is an error in setting up the configuration of a Service/Client, PRIOR to publishing or handling a message.
+
+    This should be considered a fatal exception.
+    """

tests/unit/test_topic_handler.py

@@ -0,0 +1,18 @@
+from intersect_sdk_common.control_plane.topic_handler import TopicHandler
+
+
+def test_topic_handler_more_complex_wildcard():
+    # test that the topic handler correctly matches topics based on its wildcard pattern
+    topic_handler = TopicHandler(
+        topic_persist=False, queue_name='irrelevant', regex_pattern='a/*/c/#'
+    )
+    assert (
+        topic_handler.does_topic_match('a/b/c') is not None
+    )  # * should match b, # should match nothing
+    assert (
+        topic_handler.does_topic_match('a/x/c/d/e') is not None
+    )  # * should match x, # should match d/e
+    assert not topic_handler.does_topic_match(
+        'a/b/d'
+    )  # * should match b but then c doesn't match d
+    assert not topic_handler.does_topic_match('x/b/c')  # a doesn't match x