plugin.mqtt.MqttSender Class Reference

Encapsulates the connection and delivery to an MQTT broker. More...

Public Member Functions
	__init__ (self, broker_address, broker_port, client_id, username=None, password=None, keepalive=60, qos=0, retain=False, status_topic=None, queue_size=200, max_retries=5, initial_delay=2, max_delay=60)
	enqueue (self, topic, json_payload)
	Buffers a message for delivery.
	is_alive (self)
	Checks whether the worker thread is still running.
	shutdown (self)
	Graceful shutdown with queue drain (analogous to TelegramSender.shutdown)

Data Fields
	broker_address
	broker_port
	keepalive
	qos
	retain
	status_topic
	max_retries
	initial_delay
	max_delay
	client

Protected Member Functions
	_connect (self)
	Establishes the connection to the broker asynchronously and starts the network thread.
	_on_connect (self, client, userdata, flags, reason_code, properties=None)
	paho-mqtt callback on (re-)connect
	_on_disconnect (self, client, userdata, disconnect_flags, reason_code, properties=None)
	paho-mqtt callback on connection loss
	_requeue (self, topic, json_payload, retry_count)
	Puts a message back into the queue after a failed delivery attempt.
	_worker_loop (self)
	Processes the queue sequentially: connected -> send, otherwise buffer & wait.

Protected Attributes
	_stop_event
	_connected
	_msg_queue
	_worker

Detailed Description

Encapsulates the connection and delivery to an MQTT broker.

Every message is handed off to an internal queue and processed by a dedicated worker thread (analogous to TelegramSender in telegram.py). If the broker is currently unreachable, messages simply stay in the queue and are delivered once the connection is restored (zero-packet-loss within the bounds of the configured queue size). If the queue is full, the OLDEST message is dropped so that new alarms are not blocked - this is logged explicitly.

Constructor & Destructor Documentation

__init__()

plugin.mqtt.MqttSender.__init__	(		self,
			broker_address,
			broker_port,
			client_id,
			username = None,
			password = None,
			keepalive = 60,
			qos = 0,
			retain = False,
			status_topic = None,
			queue_size = 200,
			max_retries = 5,
			initial_delay = 2,
			max_delay = 60
	)

                 queue_size=200, max_retries=5, initial_delay=2, max_delay=60):
        self._stop_event = threading.Event()
 
        self.broker_address = broker_address
        self.broker_port = broker_port
        self.keepalive = keepalive
        self.qos = qos
        self.retain = retain
        self.status_topic = status_topic
        self.max_retries = max_retries
        self.initial_delay = initial_delay
        self.max_delay = max_delay
 
        self._connected = False
        self._msg_queue = queue.Queue(maxsize=queue_size)
 
        self.client = mqtt.Client(
            client_id=client_id,
            callback_api_version=mqtt.CallbackAPIVersion.VERSION2
        )
        if username:
            self.client.username_pw_set(username, password)
 
        # Last Will: published by the broker itself if the connection drops
        # ungracefully (e.g. host crash) - must be set before connect().
        if self.status_topic:
            self.client.will_set(self.status_topic, payload="offline", qos=1, retain=True)
 
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
 
        # built-in reconnect mechanism of paho with exponential backoff
        self.client.reconnect_delay_set(min_delay=1, max_delay=30)
 
        self._worker = threading.Thread(target=self._worker_loop, daemon=True)
        self._worker.start()
 
        self._connect()
 

Member Function Documentation

enqueue()

plugin.mqtt.MqttSender.enqueue	(		self,
			topic,
			json_payload
	)

Buffers a message for delivery.

Parameters

topic	MQTT topic
json_payload	already serialized JSON string

    def enqueue(self, topic, json_payload):
        r"""!Buffers a message for delivery.
 
        @param topic: MQTT topic
        @param json_payload: already serialized JSON string"""
        try:
            self._msg_queue.put_nowait((topic, json_payload, 0))
        except queue.Full:
            try:
                dropped_topic, _, _ = self._msg_queue.get_nowait()
                self._msg_queue.put_nowait((topic, json_payload, 0))
                logging.warning("MQTT Plugin: buffer full (max %d) - dropping oldest buffered message (topic '%s') in favor of the new one",
                                self._msg_queue.maxsize, dropped_topic)
            except queue.Empty:  # pragma: no cover - race, practically unreachable
                logging.error("MQTT Plugin: message for topic '%s' discarded (buffer full)", topic)
 

is_alive()

plugin.mqtt.MqttSender.is_alive

(

self

)

Checks whether the worker thread is still running.

Returns

True or False

    def is_alive(self):
        r"""!Checks whether the worker thread is still running.
 
        @return True or False"""
        return self._worker.is_alive()
 

shutdown()

plugin.mqtt.MqttSender.shutdown

(

self

)

Graceful shutdown with queue drain (analogous to TelegramSender.shutdown)

    def shutdown(self):
        r"""!Graceful shutdown with queue drain (analogous to TelegramSender.shutdown)"""
        logging.info("MQTT Plugin: shutting down connection...")
        self._stop_event.set()
 
        timeout = time.time() + 5
        while not self._msg_queue.empty() and time.time() < timeout:
            time.sleep(0.1)
 
        remaining = self._msg_queue.qsize()
        if remaining > 0:
            logging.warning("MQTT Plugin: %d unsent messages discarded during shutdown", remaining)
 
        if self.status_topic and self._connected:
            try:
                info = self.client.publish(self.status_topic, payload="offline", qos=1, retain=True)
                info.wait_for_publish(timeout=2)
            except Exception:
                pass
 
        self._worker.join(timeout=5)
        try:
            self.client.loop_stop()
            self.client.disconnect()
        except Exception:
            pass
        logging.debug("MQTT Plugin: connection closed")
 

_connect()

plugin.mqtt.MqttSender._connect (   self )

protected

Establishes the connection to the broker asynchronously and starts the network thread.

    def _connect(self):
        r"""!Establishes the connection to the broker asynchronously and starts the network thread"""
        try:
            self.client.connect_async(self.broker_address, self.broker_port, keepalive=self.keepalive)
            self.client.loop_start()
        except Exception as e:
            logging.error("MQTT Plugin: failed to connect to %s:%d: %s", self.broker_address, self.broker_port, e)
 

_on_connect()

plugin.mqtt.MqttSender._on_connect (   self,   client,   userdata,   flags,   reason_code,   properties = None  )

protected

paho-mqtt callback on (re-)connect

    def _on_connect(self, client, userdata, flags, reason_code, properties=None):
        r"""!paho-mqtt callback on (re-)connect"""
        rc = getattr(reason_code, "value", reason_code)
        if rc == 0:
            logging.info("MQTT Plugin: connected to %s:%d", self.broker_address, self.broker_port)
            self._connected = True
            if self.status_topic:
                client.publish(self.status_topic, payload="online", qos=1, retain=True)
        else:
            logging.error("MQTT Plugin: connection refused (reason_code=%s)", reason_code)
            self._connected = False
 

_on_disconnect()

plugin.mqtt.MqttSender._on_disconnect (   self,   client,   userdata,   disconnect_flags,   reason_code,   properties = None  )

protected

paho-mqtt callback on connection loss

    def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties=None):
        r"""!paho-mqtt callback on connection loss"""
        logging.warning("MQTT Plugin: connection lost (reason_code=%s) - buffered messages are waiting for reconnect", reason_code)
        self._connected = False
 

_requeue()

plugin.mqtt.MqttSender._requeue (   self,   topic,   json_payload,   retry_count  )

protected

Puts a message back into the queue after a failed delivery attempt.

    def _requeue(self, topic, json_payload, retry_count):
        r"""!Puts a message back into the queue after a failed delivery attempt"""
        try:
            self._msg_queue.put_nowait((topic, json_payload, retry_count))
        except queue.Full:
            logging.error("MQTT Plugin: message for topic '%s' discarded while requeuing (buffer full)", topic)
 

_worker_loop()

plugin.mqtt.MqttSender._worker_loop (   self )

protected

Processes the queue sequentially: connected -> send, otherwise buffer & wait.

    def _worker_loop(self):
        r"""!Processes the queue sequentially: connected -> send, otherwise buffer & wait"""
        delay = self.initial_delay
        while not self._stop_event.is_set():
            try:
                topic, json_payload, retry_count = self._msg_queue.get(timeout=1)
            except queue.Empty:
                continue
 
            if not self._connected:
                logging.debug("MQTT Plugin: not connected - message for topic '%s' stays buffered", topic)
                self._requeue(topic, json_payload, retry_count)
                time.sleep(min(delay, self.max_delay))
                continue
 
            try:
                result = self.client.publish(topic, json_payload, qos=self.qos, retain=self.retain)
                if result.rc != mqtt.MQTT_ERR_SUCCESS:
                    raise RuntimeError(f"publish() returned error code {result.rc}")
 
                result.wait_for_publish(timeout=5)
                if not result.is_published():
                    raise RuntimeError("no delivery confirmation within 5s")
 
                logging.debug("MQTT Plugin: message published on topic '%s' (MID=%s, QoS=%d)", topic, result.mid, self.qos)
                delay = self.initial_delay
 
            except Exception as e:
                if retry_count >= self.max_retries:
                    logging.error("MQTT Plugin: message for topic '%s' permanently discarded after %d attempts: %s", topic, self.max_retries, e)
                else:
                    logging.warning("MQTT Plugin: delivery attempt for topic '%s' failed (%d/%d): %s", topic, retry_count + 1, self.max_retries, e)
                    self._requeue(topic, json_payload, retry_count + 1)
                    time.sleep(min(delay, self.max_delay))
                    delay = min(delay * 2, self.max_delay)
 
 
# ===========================
# BoswatchPlugin-Class
# ===========================
 

Field Documentation

_stop_event

plugin.mqtt.MqttSender._stop_event

protected

broker_address

plugin.mqtt.MqttSender.broker_address

broker_port

plugin.mqtt.MqttSender.broker_port

keepalive

plugin.mqtt.MqttSender.keepalive

qos

plugin.mqtt.MqttSender.qos

retain

plugin.mqtt.MqttSender.retain

status_topic

plugin.mqtt.MqttSender.status_topic

max_retries

plugin.mqtt.MqttSender.max_retries

initial_delay

plugin.mqtt.MqttSender.initial_delay

max_delay

plugin.mqtt.MqttSender.max_delay

_connected

plugin.mqtt.MqttSender._connected

protected

_msg_queue

plugin.mqtt.MqttSender._msg_queue

protected

client

plugin.mqtt.MqttSender.client

_worker

plugin.mqtt.MqttSender._worker

protected