Refactor mqtt connections (#6)

* Split into methods , add recurrent retry

* Fix arguments for disconnect method

* Rewrite nasty functions into nested classes

Author: bojleros

app/main.py

@@ -75,8 +75,8 @@ def __init__(self, serconf, ecconf, callback=None):
         self.rtu = None
         self.reinit_count = 3
         self.callback = callback
-        self.brightness = [255, 255, 255]
-        self.register = [255, 255, 255]
+        self.brightness = [255] * 3
+        self.register = [255] * 3
         self.lock = threading.Lock()
 
     def __del__(self):
@@ -183,10 +183,9 @@ def set_channel(self, client, userdata, message):
             # raise e
         else:
             time.sleep(0.02)
+            self.lock.release()
             if bool(self.callback):
-
                 self.callback(ch, payload)
-            self.lock.release()
 
 
 class Mqtt:
@@ -196,28 +195,79 @@ def __init__(self, mqconf, command_topics, state_topics, callback):
         self.ctopics = command_topics
         self.stopics = state_topics
         self.callback = callback
-        self.connhandlers = []
+        self.consumers = [None] * 3
 
     def __del__(self):
         msg("Stopping all mq connections")
-        for h in self.connhandlers:
-            h.loop_stop()
-            h.disconnect()
-
-    def _consume_topic(self, channel):
-        c = subscribe.Client()
-        c.on_message = self.callback
-        c.user_data_set({'channel': channel})
-        if self.mqconf['username'] is not None:
-            c.username_pw_set(self.mqconf['username'], password=self.mqconf['password'])
-        c.connect(self.mqconf['address'], port=self.mqconf['port'], keepalive=15)
-        c.subscribe(self.ctopics[channel], qos=self.mqconf['qos'])
-        c.loop_start()
-        self.connhandlers.append(c)
+        for h in self.consumers:
+            if bool(h):
+                del h
+
+    class Consumer:
+
+        def __init__(self, mqconf, channel, topic, msg_callback):
+            self.mqconf = mqconf
+            self.channel = str(channel)
+            self.topic = topic
+            self.msg_callback = msg_callback
+
+            self.conn = subscribe.Client()
+            self.conn.on_message = self.msg_callback
+            self.conn.on_connect = self._on_connect
+            self.conn.on_disconnect = self._on_disconnect
+            self.conn.user_data_set({'channel': self.channel})
+
+            if self.mqconf['username'] is not None:
+                self.conn.username_pw_set(self.mqconf['username'],
+                                          password=self.mqconf['password'])
+
+            self._connect()
+            # subscribe is executed via _on_connect
+            self.conn.loop_start()
+
+        def __del__(self):
+            if bool(self.conn):
+                msg("Channel%s : Closing connection" % self.channel)
+                self.conn.loop_stop()
+                self.conn.disconnect()
+                del self.conn
+
+        def _connect(self, depth=0):
+            try:
+                self.conn.connect(self.mqconf['address'], port=self.mqconf['port'], keepalive=15)
+            except Exception as e:
+                msg("Channel%s : Connection failed : %s" % (self.channel, str(e)))
+                depth += 1
+                if depth <= 60:
+                    msg("Channel%s : Waiting 10 seconds before reconnecting ..." % self.channel)
+                    time.sleep(10)
+                    self._connect(depth)
+                else:
+                    msg("Channel%s : Reconnecting was failing for too long ..." % self.channel)
+                    raise e
+
+        def _subscribe(self):
+            try:
+                self.conn.subscribe(self.topic, qos=self.mqconf['qos'])
+            except Exception as e:
+                msg("Channel%s: Subscription exception : %s" % (self.channel, str(e)))
+                raise e
+
+        def _on_connect(self, client, userdata, flags, rc):
+            msg("Channel%s : Connected" % self.channel)
+            self._subscribe()
+
+        def _on_disconnect(self, client, userdata, rc):
+            msg("Channel%s : Disconnected" % self.channel)
 
     def consume_all(self):
         for ch, topic in self.ctopics.items():
-            self._consume_topic(ch)
+            try:
+                c = self.Consumer(self.mqconf, ch, topic, self.callback)
+            except Exception as e:
+                raise e
+            else:
+                self.consumers[int(ch)] = c
 
     def postback(self, ch, payload):
         auth = None
@@ -227,7 +277,7 @@ def postback(self, ch, payload):
                     }
 
         # homeassistant lack proper typing
-        # on the other side json module is constantly puting double quotation marks around int !
+        # on the other side json module is constantly puting double quotation marks around ints ...
         payload_str = "{\"state\": \"%s\", \"brightness\": %s}" % (payload['state'], payload['brightness'])
 
         try: