fix common connection issues for MQTT to support pub/sub

Author: tipatterson-dev

src/oshconnect/csapi4py/default_api_helpers.py

@@ -137,6 +137,29 @@ def retrieve_resource(self, res_type: APIResourceTypes, res_id: str = None, pare
                                                 headers=req_headers)
         return api_request.make_request()
 
+    def get_resource(self, resource_type: APIResourceTypes, resource_id: str = None,
+                     subresource_type: APIResourceTypes = None,
+                     req_headers: dict = None):
+
+        """
+        Helper to get resources by type, specifically by id, and optionally a sub-resource collection of a specified resource.
+        :param resource_type:
+        :param resource_id:
+        :param subresource_type:
+        :param req_headers:
+        :return:
+        """
+        if req_headers is None:
+            req_headers = {}
+        base_api_url = self.get_api_root_url()
+        resource_type_str = resource_type_to_endpoint(resource_type)
+        res_id_str = f'/{resource_id}' if resource_id else ""
+        sub_res_type_str = f'/{resource_type_to_endpoint(subresource_type)}' if subresource_type else ""
+        complete_url = f'{base_api_url}/{resource_type_str}{res_id_str}{sub_res_type_str}'
+        api_request = ConnectedSystemAPIRequest(url=complete_url, request_method='GET', auth=self.get_helper_auth(),
+                                                headers=req_headers)
+        return api_request.make_request()
+
     def update_resource(self, res_type: APIResourceTypes, res_id: str, json_data: any, parent_res_id: str = None,
                         from_collection: bool = False, url_endpoint: str = None, req_headers: dict = None):
         """

src/oshconnect/csapi4py/mqtt.py

@@ -2,7 +2,7 @@
 
 
 class MQTTCommClient:
-    def __init__(self, url, port=1883, username=None, password=None, path='mqtt', client_id="", transport='tcp'):
+    def __init__(self, url, port=1883, username=None, password=None, path='mqtt', client_id_suffix="", transport='tcp'):
         """
     Wraps a paho mqtt client to provide a simple interface for interacting with the mqtt server that is customized
     for this library.
@@ -17,10 +17,10 @@ def __init__(self, url, port=1883, username=None, password=None, path='mqtt', cl
         self.__url = url
         self.__port = port
         self.__path = path
-        self.__client_id = client_id
+        self.__client_id = f'oscapy_mqtt-{client_id_suffix}'
         self.__transport = transport
 
-        self.__client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
+        self.__client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=self.__client_id)
 
         if self.__transport == 'websockets':
             self.__client.ws_set_options(path=self.__path)

src/oshconnect/eventbus.py

@@ -12,4 +12,4 @@ class EventBus(ABC):
     """
     A base class for an event bus system.
     """
-    _deque: collections.deque
+    _deque: collections.deque

src/oshconnect/oshconnectapi.py

@@ -121,7 +121,9 @@ def discover_datastreams(self):
             datastreams = list(
                 map(lambda ds: Datastream(parent_node=system.get_parent_node(), id=ds.ds_id, datastream_resource=ds),
                     res_datastreams))
-            datastreams = [ds.set_parent_resource_id(system.get_underlying_resource().system_id) for ds in datastreams]
+            for ds in datastreams:
+                ds.set_parent_resource_id(system.get_underlying_resource().system_id)
+            # datastreams = [ds.set_parent_resource_id(system.get_underlying_resource().system_id) for ds in datastreams]
             self._datastreams.extend(datastreams)
 
     def discover_systems(self, nodes: list[str] = None):

src/oshconnect/streamableresource.py

@@ -149,7 +149,9 @@ def __init__(self, protocol: str, address: str, port: int,
         if kwargs.get('enable_mqtt'):
             if kwargs.get('mqtt_port') is not None:
                 self._mqtt_port = kwargs.get('mqtt_port')
-            self._mqtt_client = MQTTCommClient(url=self.address, port=self._mqtt_port)
+            self._mqtt_client = MQTTCommClient(url=self.address, port=self._mqtt_port, client_id_suffix=uuid.uuid4().hex,)
+            self._mqtt_client.connect()
+            self._mqtt_client.start()
             # self._mqtt_client = MQTTCommClient(url=self.address + self.server_root, port=self._mqtt_port,
             #                                    username=username, password=password, )
 
@@ -249,6 +251,7 @@ class Status(Enum):
     STOPPING = "stopping"
     STOPPED = "stopped"
 
+
 class StreamableModes(Enum):
     PUSH = "push"
     PULL = "pull"
@@ -340,7 +343,12 @@ def init_mqtt(self):
             logging.warning(f"No MQTT client configured for streamable resource {self._id}.")
             return
 
-        self.get_mqtt_topic()
+        self._mqtt_client.set_on_subscribe(self._default_on_subscribe)
+
+        # self.get_mqtt_topic()
+
+    def _default_on_subscribe(self, client, userdata, mid, granted_qos, properties):
+        print("OSH Subscribed: "+str(mid)+" "+str(granted_qos))
 
     def get_mqtt_topic(self, subresource: APIResourceTypes | None = None):
         """
@@ -468,7 +476,7 @@ def subscribe_mqtt(self, topic: str, qos: int = 0):
         if self._mqtt_client is None:
             logging.warning(f"No MQTT client configured for streamable resource {self._id}.")
             return
-        self._mqtt_client.subscribe(topic, qos=qos, msg_callback=self._message_handler)
+        self._mqtt_client.subscribe(topic, qos=qos, msg_callback=self._mqtt_sub_callback)
 
     def _publish_mqtt(self, topic, payload):
         if self._mqtt_client is None:
@@ -502,6 +510,11 @@ def publish(self, payload):
         pass
 
 
+    def _mqtt_sub_callback(self, client, userdata, msg):
+        print(f"Received MQTT message on topic {msg.topic}: {msg.payload}")
+        self._msg_reader_queue.put_nowait(msg.payload)
+
+
 class System(StreamableResource[SystemResource]):
     name: str
     label: str
@@ -534,8 +547,10 @@ def __init__(self, name: str, label: str, urn: str, parent_node: Node, **kwargs)
         # self.underlying_resource = self._sys_resource
 
     def discover_datastreams(self) -> list[DatastreamResource]:
-        res = self._parent_node.get_api_helper().retrieve_resource(
-            APIResourceTypes.DATASTREAM, req_headers={})
+        # res = self._parent_node.get_api_helper().retrieve_resource(
+        #     APIResourceTypes.DATASTREAM, req_headers={})
+        res = self._parent_node.get_api_helper().get_resource(APIResourceTypes.SYSTEM, self._resource_id,
+                                                              APIResourceTypes.DATASTREAM)
         datastream_json = res.json()['items']
         ds_resources = []
 
@@ -702,19 +717,20 @@ def insert_observation_dict(self, obs_data: dict):
     def start(self):
         super().start()
         if self._mqtt_client is not None:
-            self._mqtt_client.connect()
-            self._mqtt_client.start()
-            self.subscribe_mqtt(self._topic)
+            # self._mqtt_client.connect()
+
             if self._connection_mode is StreamableModes.PULL or self._connection_mode is StreamableModes.BIDIRECTIONAL:
                 self._mqtt_client.subscribe(self._topic, msg_callback=self._mqtt_sub_callback)
+            else:
+                try:
+                    loop = asyncio.get_event_loop()
+                    loop.create_task(self._write_to_mqtt())
+                except Exception as e:
+                    # TODO: Use logging instead of print
+                    print(traceback.format_exc())
+                    print(f"Error starting MQTT write task: {e}")
 
-        try:
-            loop = asyncio.get_event_loop()
-            loop.create_task(self._write_to_mqtt())
-        except Exception as e:
-            # TODO: Use logging instead of print
-            print(traceback.format_exc())
-            print(f"Error starting MQTT write task: {e}")
+            # self._mqtt_client.start()
 
     def init_mqtt(self):
         super().init_mqtt()
@@ -731,9 +747,9 @@ def _queue_push(self, msg):
     def _queue_pop(self):
         return self._msg_reader_queue.get_nowait()
 
-    def _mqtt_sub_callback(self, client, userdata, msg):
-        print(f"MQTT Message received on topic {msg.topic}: {msg.payload}")
-        self._queue_push(msg.payload)
+    # def _mqtt_sub_callback(self, client, userdata, msg):
+    #     print(f"MQTT Message received on topic {msg.topic}: {msg.payload}")
+    #     self._queue_push(msg.payload)
 
     def insert(self, data: dict):
         # self._queue_push(data)