opcua-with-python/opcua2mqtt/MqttPublish.py

from threading import Event
from loguru import logger
from MqttBase import AbstractMqttPublisher
from AbstractDataObject import InvalidDataObjectException
from queue import Empty
import json

LOOP_SLICE = 0.1  # seconds

class MqttPublish(AbstractMqttPublisher):
    def __init__(self, config, stats, queue):
        super().__init__(config)
        self.queue = queue
        self.stats = stats
        self.topicPre = self.config["publishTopicPrefix"]

    def localLoop(self):
        while not self.killBill:
            try:
                dataObject = self.queue.get(timeout=LOOP_SLICE)
                topic = self.topicPre + '/' + dataObject.getTopicPart()
                payload = dataObject.getPayload()
                self.client.publish(topic, payload)
                logger.debug("mqtt message sent: {} -> {}".format(topic, payload))
                self.stats.incMqttRequests()
            except Empty:
                # just evaluate the killBill at the top of the loop again
                pass
            except InvalidDataObjectException as e:
                self.stats.incMqttErrors()
                logger.error(f"InvalidDataObjectException received in MQTT local loop: {e}")
            except Exception as e:
                self.stats.incMqttErrors()
                logger.error(f"Exception {type(e)} received in MQTT local loop: {e}")