This MQTT testing tool is a simple Python script that will list topics that have a retained message and optionally deletes the retained message.
The script can be use from the command prompt by supplying options.
How it Works
The utility is designed to be run from the command line.
The first part of the script gets the command line input using getopt
The script requires only two parameters which are the broker IP or domain name and topic.
The script checks for a valid topic before continuing.
Now we initialize some flags using the Initialise_client_object function and the callbacks using the Initialise_clients function.
The script then tries to connect to the broker and uses a try except block to catch connection errors as explained in detail in understanding client connections.
Now we start a loop to process callbacks.
It then waits is loop for the connection to be acknowledged.
def on_connect(client, userdata, flags, rc): if rc==0: client.connected_flag=True print("connected sending ",inputs["loops"]," messages ",\ inputs["loop_delay"]," second Intervals") else: client.bad_connection_flag=True if rc==5: print("broker requires authentication")
Notice the use of the flags that are set and reset in the on_connect callback.
while not client.connected_flag and not client.bad_connection_flag: time.sleep(.25)
If the connection acknowledgement is bad the script exits if it is good it then it subscribes to a topic.
It then waits in a loop for the subscription to be acknowledged.
while not client.suback_flag: #wait for subscribe to be acknowledged time.sleep(.25)
Now we subscribe to the topic and wait for the subscribe to be acknowledged
while not client.suback_flag: #wait for subscribe to be acknowledged time.sleep(.25) if sleep_count>40: #give up print("Subscribe failure quitting") client.loop_stop() sys.exit() sleep_count+=1
We then wait a set period as all of the work is now being handled by the on_message callback.
The message.retained variable contains the retained flag which is 1 for a retained message.
If it is retained it stores the topic.
After the delay expires the script will print out the list of topics with retained messages and then if the clear flag is set it will clear the retained flag.
if inputs["clear_retained"]: if len(retained_topics)>0: verbose=False clear_retained(retained_topics)
This function clears the retained flag by publishing a empty message (msg=””) with the retained flag set.
def clear_retained(retained): #accepts single topic or list msg="" if isinstance(retained,str): client.publish(retained,msg,qos=QOS0,retain=RETAIN) else: try: for t in retained: client.publish(t,msg,qos=QOS0,retain=RETAIN) print ("Clearing retaind on ",msg,"topic -",t," qos=",QOS0," retain=",RETAIN) except: Print("problems with topic") return -1
Run with the -h option to list options.
c:>python retained-messsage.py -h
Note: the -c and -v options no longer require a parameter i.e use -c and not -c True
Create Retained Messages
For testing you may need to create some retained messages this script will do that.( create-retained-messsage.py)
The script can be run from command prompt but it is easier to edit, and run in the IDE or from the command prompt with no options.
- MQTT Test Tools
- Simple Python MQTT Data Logger
- Simple Controllable MQTT Sensor Simulator in Python
- How to Send a File Using MQTT Using a Python Script