Python: Subscribing to MQTT topic

Introduction

The objective of this post is to explain how to connect to a MQTT broker and subscribe to a topic, using Python. For this example, we will be using paho-mqtt, which is a MQTT Python client library. The easiest way to install it is via pip, with the following command:

pip install paho-mqtt

As MQTT broker, we will use CloudMQTT, which offers a free plan, amongst other options. You can check here how to create an account and how to create a broker instance.

Since the basics on how to connect to the CloudMQTT broker were already covered in this previous post, those sections of the code will not be described in detail. Please read that post first for the basics of paho-mqtt.

The code

First, we need to import the client class from the MQTT library, to have access to all the functionality needed to connect to the broker and subscribe to topics. Additionally, we will also import Python’s time module, so we can have access to the sleep function.

We will also declare some auxiliary global variables. The first one will be used to control the initial state of connection to the broker, and the other ones will hold the information needed to connect to the broker. All the credentials needed for the connection should be obtained from the CloudMQTT instance information page.

import paho.mqtt.client as mqttClient
import time
 
Connected = False #global variable for the state of the connection
 
broker_address= "m11.cloudmqtt.com"
port = 12948
user = "yourUSer"
password = "yourPass"

Now, we will create a new client instance. We will pass as argument of the constructor a unique client identifier in format of a string. Next, we will call the username_pw_set method, which allows to specify a username and, optionally, a password. For the broker we are trying to connect to, the password is needed.

We also need to specify a on_connect callback function, which is called when the broker responds to the connection request. We will just assign the function for now and define the actual code latter.

Since we are going to subscribe to a topic, we will also specify a on_message callback function, which is called for every message received on a subscribed topic [1]. Again, we will define the actual function code latter.

client = mqttClient.Client("Python")               #create new instance
client.username_pw_set(user, password=password)    #set username and password
client.on_connect= on_connect                      #attach function to callback
client.on_message= on_message                      #attach function to callback

Finally, we will call the connect method, for establishing the connection to the broker. This is a blocking method. Then, we will call a method called loop_start, which will run a thread in background to handle the network connection and sending/receiving data.

Since establishing the connection may take a while, we will do a loop until the previously declared Connected variable is set to true. This will be done in our callback function, which we still need to specify.

Once the connection is established, we need to call the subscribe method, passing as input the topic that we want to subscribe to. In this case, we will subscribe to the “python/test” topic.

Note that this method can be called with a different set of arguments, and we are using it in its simple form. Also remember that when a message is received, it will be handled on a function we called on_message, yet to be defined.

client.connect(broker_address, port=port)  #connect to broker
client.loop_start()                        #start the loop
 
while Connected != True:    #Wait for connection
    time.sleep(0.1)
 
client.subscribe("python/test")

After subscribing to the topic, we will do an infinite loop with a small delay in each iteration, since the messages will be handled by a callback function. So, we don’t need to worry about polling anything.

We will run this loop on a try-except block, where the except block will catch a keyboard interrupt. This way, we can end the loop by sending a ctrl+C command on the Python shell.

Since the program will finish in the except block, we call the disconnect method, to disconnect from the broker. After that, we also need to call the loop_stop method, to end the previously mentioned background thread.

try:
    while True:
        time.sleep(1)
 
except KeyboardInterrupt:
    print "exiting"
    client.disconnect()
    client.loop_stop()

Now, we still need to declare the on_message callback function. As can be seen here, this function receives 3 arguments. We are only going to use the argument named in the previous link as message, which is an instance of the MQTTMessage class[1].

We will access a member of this class called payload, which we will use to get the actual message received from the subscribed topic. So, on the on_message function, we will print the message payload, as can be seen bellow.

def on_message(client, userdata, message):
    print "Message received: "  + message.payload

You can check the full source code bellow, which already includes the on_connect function declaration.

import paho.mqtt.client as mqttClient
import time
 
def on_connect(client, userdata, flags, rc):
 
    if rc == 0:
 
        print("Connected to broker")
 
        global Connected                #Use global variable
        Connected = True                #Signal connection 
 
    else:
 
        print("Connection failed")
 
def on_message(client, userdata, message):
    print "Message received: "  + message.payload
 
Connected = False   #global variable for the state of the connection
 
broker_address= "m11.cloudmqtt.com"  #Broker address
port = 12948                         #Broker port
user = "yourUser"                    #Connection username
password = "yourPassword"            #Connection password
 
client = mqttClient.Client("Python")               #create new instance
client.username_pw_set(user, password=password)    #set username and password
client.on_connect= on_connect                      #attach function to callback
client.on_message= on_message                      #attach function to callback
 
client.connect(broker_address, port=port)          #connect to broker
 
client.loop_start()        #start the loop
 
while Connected != True:    #Wait for connection
    time.sleep(0.1)
 
client.subscribe("python/test")
 
try:
    while True:
        time.sleep(1)
 
except KeyboardInterrupt:
    print "exiting"
    client.disconnect()
    client.loop_stop()

Testing the code

For the testing part, we can use IDLE, the Python IDE. So, after finishing the code, run it. As we defined, a message should be printed after the connection to the broker is established.

To test the subscription to the topic, we will be using an application that can publish to the topic where the Python program subscribed. This application will be MQTTlens. So, open it and connect to the broker. Then, publish a message to the “python/test” topic, as shown in figure 1.

You should get an output similar to figure 2 on IDLE, where the Python application is running.

 References

[1] https://github.com/eclipse/paho.mqtt.python#on_message

Related posts

• Python: Publishing messages to MQTT topic
• ESP8266: Connecting to MQTT broker

Technical details

• Python version: 2.7.8
• MQTTLens version: 0.0.13