initial

.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+*~
+.*~
+

MqttBase.py

@@ -0,0 +1,75 @@
+import paho.mqtt.client as mqtt
+from loguru import logger
+import threading
+import ssl
+
+
+
+def mqttOnConnectCallback(client, userdata, flags, rc):
+    userdata.onConnect()
+
+def mqttOnMessageCallback(client, userdata, message):
+    userdata.onMessage(message.topic, message.payload)
+
+def mqttOnDisconnectCallback(client, userdata, rc):
+    userdata.onDisconnect(rc)
+
+
+class AbstractMqttPublisher(threading.Thread):
+    def __init__(self, config):
+        super().__init__()
+
+        self.config = config["mqtt"]
+
+        self.client = mqtt.Client(userdata=self)
+
+        # consider this flag in the localLoop
+        self.killBill = False
+        self.killEvent = threading.Event()
+
+    def run(self):
+        self.client.on_message = mqttOnMessageCallback
+        self.client.on_connect = mqttOnConnectCallback
+        self.client.on_disconnect = mqttOnDisconnectCallback
+
+        if ("login" in self.config) and ("password" in self.config):
+            self.client.username_pw_set(self.config["login"], self.config["password"])
+        
+        if ("ca" in self.config) and ("cert" in self.config) and ("key" in self.config):
+            self.client.tls_set(
+                ca_certs=self.config["ca"],
+                certfile=self.config["cert"],
+                keyfile=self.config["key"],
+                cert_reqs=ssl.CERT_REQUIRED,
+                tls_version=ssl.PROTOCOL_TLSv1_2, 
+                ciphers=None # this does not mean "no cipher" but it means "default ciphers"
+            )
+
+        self.client.connect(self.config["broker"], int(self.config["port"]))
+        self.client.loop_start()
+        logger.info("mqtt loop started")
+
+        self.localLoop()
+    
+    def localLoop(self):
+        raise NotImplementedError()
+
+    def stop(self):
+        self.client.loop_stop()
+        logger.info("mqtt loop stopped")
+        
+        self.killBill = True
+        logger.info("kill flag set")
+
+        self.killEvent.set()
+        logger.info("kill events triggered")
+
+    def onConnect(self):
+        logger.info("mqtt connected")
+    
+    def onDisconnect(self, rc):
+        logger.warning("mqtt disconnect, rc: {}".format(rc))
+
+    def onMessage(self, topic, payload):
+        logger.warning("mqtt unexpected message received: {} -> {}".format(topic, str(payload)))
+

TestPublish.py

@@ -0,0 +1,19 @@
+from threading import Event
+from loguru import logger
+from MqttBase import AbstractMqttPublisher
+
+
+class TestPublish(AbstractMqttPublisher):
+    def __init__(self, config):
+        super().__init__(config)
+
+    def localLoop(self):
+        cnt = 0
+        while not self.killBill:
+            cnt += 1
+            topic = self.config["publishTopic"]
+            payload = str(cnt)
+            self.client.publish(topic, payload)
+            logger.warning("mqtt message sent: {} -> {}".format(topic, payload))
+
+            self.killEvent.wait(timeout=float(self.config["publishPeriod"]))

TestSubscribe.py

@@ -0,0 +1,20 @@
+from MqttBase import AbstractMqttPublisher
+from loguru import logger
+from time import sleep
+
+
+class TestSubscribe(AbstractMqttPublisher):
+    def __init__(self, config):
+        super().__init__(config)
+
+    def localLoop(self):
+        while not self.killBill:
+            sleep(60.0)
+
+    def onMessage(self, topic, payload):
+        logger.warning("mqtt message received: {} -> {}".format(topic, str(payload)))
+
+    def onConnect(self):
+        logger.info("mqtt connected")
+        self.client.subscribe("{}".format(self.config["subscribeTopic"]))
+        logger.info("subscribed")

config.ini

@@ -0,0 +1,9 @@
+[mqtt]
+broker = broker.mainscnt.eu
+port = 8883
+ca = c:\Users/dehottgw\key\isrgrootx1.pem
+cert = c:\Users\dehottgw\key\broker.mainscnt.eu\wn-mainscnt-broker-client.crt
+key = c:\Users\dehottgw\key\broker.mainscnt.eu\wn-mainscnt-broker-client.pem
+subscribeTopic = test1
+publishTopic = test2
+publishPeriod = 5

example1.py

@@ -0,0 +1,56 @@
+from TestPublish import TestPublish
+from TestSubscribe import TestSubscribe
+from loguru import logger
+import argparse
+import configparser
+import threading
+
+
+deathBell = threading.Event()
+
+def exceptHook(args):
+    global deathBell
+    logger.error("Exception in thread caught: {}".format(args))
+    deathBell.set()
+    logger.error("rang the death bell")
+
+
+logger.info("example1 starting")
+
+parser = argparse.ArgumentParser(description="example1")
+parser.add_argument('--config', '-f',
+                    help='Config file, default is $pwd/config.ini',
+                    required=False,
+                    default='./config.ini')
+args = parser.parse_args()
+
+config = configparser.ConfigParser()
+config.read(args.config)
+
+testSubscribeThread = TestSubscribe(config)
+testSubscribeThread.start()
+logger.info("testSubscribe started")
+
+testPublishThread = TestPublish(config)
+testPublishThread.start()
+logger.info("testPublish started")
+
+threading.excepthook = exceptHook
+logger.info("Threading excepthook set")
+
+logger.info("example1 is running")
+
+
+deathBell.wait()
+logger.error("example1 is dying")
+
+testSubscribeThread.stop()
+testPublishThread.stop()
+
+testSubscribeThread.join()
+logger.error("testSubscribe joined")
+
+testPublishThread.join()
+logger.error("testPublish joined")
+
+logger.error("example1 is terminated")