Réception de messages MQTT

La programmation de la réception des messages se fait en deux étapes :

• Programmer une fonction callback qui effectuera les traitements voulus à chaque réception d'un message,
• Programmer les abonnements aux topics qui nous intéressent.

La fonction callback

Au début de notre programme, on définit une fonction callback qui sera ensuite appelée une fois que la connexion aura été réellement effectuée. La fonction callback sera appelée avec les 4 arguments suivant (dont vous pouvez changer les noms si vous le souhaitez) :

• client

  Ce premier paramètre est une copie de l'objet de gestion de MQTT utilisé pour cette connexion

• userdata

  Ce deuxième paramètre est la structure de données utilisateur que l'on a passée à la connexion

• msg

  Ce troisième paramètre est l'objet message composé de plusieurs champs :

  • topic

    Ce champ est une chaine de caractères contenant le topic sur lequel le message a été reçu.

  • payload

    Ce champ est le message binaire qui a été reçu. On peut avoir besoin de le convertir en chaine de caractères (via la méthode decode("utf8")) ou effectuer dessus un décodage JSON si le message transmis est sérialisé en JSON.

  • qos

    Ce champ est la qualité de service utilisée pour transmettre le message.

On remplacera, les lignes de 3 à 7, par le traitement que l'on souhaite effectuer après la connexion. On retrouve les autres lignes qui correspondent à la structure de la fonction callback "on_connect" déjà vue précédemment.

def on_message(client,userdata,message):
	try:
		# corps du traitement qui sera exécuté
		#
		print("Received message '" + message.payload.decode("utf8")
		+ "' on topic '" + message.topic
		+ "' with QoS " + str(message.qos))
	except Exception as e: 
		print(e)
		quit(0)
	

L'abonnement aux topics

La meilleure place pour ajouter est dans le callback on_connect car en cas de reconnexion les abonnements seront ré-effectués. On utilise, pour cela, la fonction subscribe() de l'objet "client" récupéré dans le callback on_connect. Cette fonction prend, en premier argument, le topic et un second argument optionnel correspond à la QOS souhaitée (par défaut si on ne précise pas c'est 0). On peut également passer en unique paramètre une liste de tuple (topic,qos) ; ce qui permet de souscrire plusieurs topics avec un seul appel de fonction.

# -*- coding:Utf-8 -*-
import paho.mqtt.client as mqtt

def on_connect(client,userdata,flags,result):
	try:
		if (result==0):
			client.subscribe([("/junia/+/temp",2),("/junia/+/hum",2)])
		else:
			print("erreur de connexion")
			quit(0)
	except Exception as e: 
		print(e)
		quit(0)

def on_message(client,userdata,message):
	try:
		print("Received message '" + message.payload.decode("utf8")
		+ "' on topic '" + message.topic
		+ "' with QoS " + str(message.qos))
	except Exception as e: 
		print(e)

mqttc = mqtt.Client()
mqttc.on_connect = on_connect
mqttc.on_message = on_message
mqttc.connect("test.mosquitto.org", 1883, 60)

mqttc.loop_forever()
	

Remarque : la ligne 7 pourrait être remplacée par les deux lignes suivantes.

			client.subscribe("/junia/+/temp",2)
			client.subscribe("/junia/+/hum",2)
	

Filtrage des messages suivant le topic

Il existe deux solutions pour séparer les traitements en fonction du topic sur lequel on a reçu le message. Soit on teste dans la fonction callback on_message (commune à tous les topics) le topic dans une série de structures "if", soit on définit une fonction callback spécifique à chaque topic (ou série de topics) que l'on souhaite gérer indépendamment des autres via la fonction message_callback_add(). Cette fonction prend en premier paramètre une chaine de caractères (qui définit le topic avec son éventuel wildcard) et en second paramètre le nom de la fonction callback que l'on désire y associer.

Exemple de filtrage via message_callback_add()

# -*- coding:Utf-8 -*-
import paho.mqtt.client as mqtt

def on_connect(client,userdata,flags,result):
	try:
		if (result==0):
			client.subscribe([("/junia/+/temp",2),("/junia/+/hum",2)])
			client.message_callback_add("/junia/+/temp",on_message_temp)
			client.message_callback_add("/junia/+/hum",on_message_hum)
		else:
			print("erreur de connexion")
			quit(0)
	except Exception as e: 
		print(e)
		quit(0)

def on_message_temp(client,userdata,message):
	try:
		print("température : ", message.payload.decode("utf8"))
	except Exception as e: 
		print(e)

def on_message_hum(client,userdata,message):
	try:
		print("humidité : ", message.payload.decode("utf8"))
	except Exception as e: 
		print(e)

mqttc = mqtt.Client()
mqttc.on_connect = on_connect
mqttc.connect("test.mosquitto.org", 1883, 60)

mqttc.loop_forever()