I am developing a helper app to clean the retained messages on my Mosquitto MQTT service. The problem I have is how to process the queue once with Paho MQTT.
I know how to
- block the thread (
loop_forever())
- query the queue asynchronously (
loop_start()/ loop_stop())
What I would like to do is to process the queue once and quit (and process each of the topics gathered this way, which are (most likely) retained messages)
loop() was the most promising:
Call regularly to process network events. This call waits in select()
until the network socket is available for reading or writing, if
appropriate, then handles the incoming/outgoing data.
Unfortunately I do not see any topics when using it.
Right now my code starts a thread, waits 2 seconds and stops it. It does the job but I would like to understand how to do that cleanly though a one-pass processing:
import paho.mqtt.client as mqtt
import time
class MQTT:
def __init__(self):
print("initializing app")
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.client.connect("mqtt.example.com", 1883, 60)
self.client.loop_start()
time.sleep(2)
self.client.loop_stop()
def on_connect(self, client, userdata, flags, rc):
print("connected to MQTT with result code " + str(rc))
self.client.subscribe("#")
def on_message(self, client, userdata, msg):
# EDIT: added a check for actually retained messages
if msg.retain:
print(f"removing retained {msg.topic}")
self.client.publish(msg.topic, retain=True)
MQTT()
