From 1d1a7e8cfbcb1a16ce92ff026cf3923713395215 Mon Sep 17 00:00:00 2001
From: avollkopf <43980694+avollkopf@users.noreply.github.com>
Date: Tue, 27 Feb 2024 07:20:10 +0100
Subject: [PATCH] just add topic filters for mqttsensor instead of crating
 tasks to accommodate new aiomqtt single message queue

---
 cbpi/__init__.py                        |  2 +-
 cbpi/controller/satellite_controller.py | 33 ++++++-------------------
 cbpi/extension/mqtt_sensor/__init__.py  | 17 +++----------
 3 files changed, 12 insertions(+), 40 deletions(-)

diff --git a/cbpi/__init__.py b/cbpi/__init__.py
index 5285c9f..87f292b 100644
--- a/cbpi/__init__.py
+++ b/cbpi/__init__.py
@@ -1,3 +1,3 @@
-__version__ = "4.3.2.a10"
+__version__ = "4.4.0.a1"
 __codename__ = "Winter Storm"
 
diff --git a/cbpi/controller/satellite_controller.py b/cbpi/controller/satellite_controller.py
index d37ea2c..19f2dc5 100644
--- a/cbpi/controller/satellite_controller.py
+++ b/cbpi/controller/satellite_controller.py
@@ -157,29 +157,10 @@ class SatelliteController:
             except Exception as e:
                 self.logger.warning("Failed to send sensorupdate via mqtt: {}".format(e))
 
-    def subcribe(self, topic, method):
-        task = asyncio.create_task(self._subcribe(topic, method))
-        return task
-
-    async def _subcribe(self, topic, method):
-        self.error=False
-        while True:
-            try:
-                if self.client._connected.done():
-                    await self.client.subscribe(topic)
-                    async for message in self.client.messages:
-                        if message.topic.matches(topic):
-                            await method(message.payload.decode())
-            except asyncio.CancelledError:
-                # Cancel
-                self.logger.warning("Subscription {} Cancelled".format(topic))
-                self.error=True
-            except MqttError as e:
-                self.logger.error("Sub MQTT Exception: {}".format(e))
-            except Exception as e:
-                self.logger.error("Sub Exception: {}".format(e))
-            # wait before try to resubscribe
-            if self.error == True:
-                break
-            else:
-                await asyncio.sleep(5)
+    def subscribe(self, topic, method):
+        self.topic_filters.append((topic,method))
+        return True
+    
+    def unsubscribe(self, topic, method):
+        self.topic_filters.remove((topic,method))
+        return True
diff --git a/cbpi/extension/mqtt_sensor/__init__.py b/cbpi/extension/mqtt_sensor/__init__.py
index 47114cf..c5b0c0b 100644
--- a/cbpi/extension/mqtt_sensor/__init__.py
+++ b/cbpi/extension/mqtt_sensor/__init__.py
@@ -24,7 +24,7 @@ class MQTTSensor(CBPiSensor):
         self.payload_text = self.props.get("PayloadDictionary", None)
         if self.payload_text != None:
             self.payload_text = self.payload_text.split('.')
-        self.mqtt_task = self.cbpi.satellite.subcribe(self.Topic, self.on_message)
+        self.subscribed = self.cbpi.satellite.subscribe(self.Topic, self.on_message)
         self.value: float = 999
         self.timeout=int(self.props.get("Timeout", 60))
         self.starttime = time.time()
@@ -55,7 +55,7 @@ class MQTTSensor(CBPiSensor):
         pass
 
     async def on_message(self, message):
-        val = json.loads(message)
+        val = json.loads(message.payload.decode())
         try:
             if self.payload_text is not None:
                 for key in self.payload_text:
@@ -130,17 +130,8 @@ class MQTTSensor(CBPiSensor):
         return dict(value=self.value)
 
     async def on_stop(self):
-        was_cancelled=False
-        if not self.mqtt_task.done():
-            logging.info("Task not done -> cancelling")
-            was_cancelled = self.mqtt_task.cancel()
-        try:            
-            logging.info("Trying to call cancelled task")
-            await self.mqtt_task
-        except asyncio.CancelledError:
-            logging.info("Task has been Cancelled")
-            pass
-        logging.info("Task cancelled: {}".format(was_cancelled))
+        self.subscribed = self.cbpi.satellite.unsubscribe(self.Topic, self.on_message)
+
 
 def setup(cbpi):
     '''