PubSub: use channel definition as prefix if defined, update docs (#1196)

Author: tomkralidis

docs/pubsub.rst

@@ -28,6 +28,7 @@ Example directive:
            type: mqtt
            url: mqtt://localhost:1883
            channel: messages/a/data  # optional
+           show_link: false  # default true
 
 HTTP
 ----
@@ -41,6 +42,7 @@ Example directive:
            type: http
            url: https://ntfy.sh
            channel: messages-a-data  # optional
+           show_link: true  # default true
 
 .. note::
 
@@ -51,6 +53,8 @@ Example directive:
 
 .. note::
 
-   If no ``channel`` is defined, the relevant OGC API endpoint is used.
+   If a ``channel`` is defined, it is used as a prefix to the relevant OGC API endpoint is used.
+
+   If a ``channel`` is not defined, only the relevant OGC API endpoint is used.
 
 .. _`OGC API Publish-Subscribe Workflow - Part 1: Core`: https://docs.ogc.org/DRAFTS/25-030.html

pycsw/broker/http.py

@@ -87,6 +87,9 @@ def pub(self, channel: str, message: str, qos: int = 1) -> bool:
         LOGGER.debug(f'Publishing to broker {self.broker_safe_url}')
         LOGGER.debug(f'Channel: {channel}')
         LOGGER.debug(f'Message: {message}')
+        LOGGER.debug(f'Santizing channel for HTTP')
+        channel = channel.replace('/', '-')
+        LOGGER.debug(f'Santized channel: {channel}')
 
         url = f'{self.broker}/{channel}'
 

pycsw/ogc/pubsub/__init__.py

@@ -48,7 +48,11 @@ def publish_message(pubsub_client, action: str, collection: str = None,
     :returns: `bool` of whether message publishing was successful
     """
 
-    channel = pubsub_client.channel or f'collections/{collection}'
+    if pubsub_client.channel is not None:
+        channel = f'{pubsub_client.channel}/collections/{collection}'
+    else:
+        channel = f'collections/{collection}'
+
     type_ = f'org.ogc.api.collection.item.{action}'
 
     if action in ['create', 'update']: