MQTT and Pub/Sub Messaging

tags: appliot2022

MQTT using a client app

For the first experiments we will use an MQTT client that can be used in any OS, it's called MQTT-explorer:

You have to install it from here: http://mqtt-explorer.com

Connecting to a public broker

There are various public MQTT brokers, for example:

• broker.hivemq.com
• broker.mqttdashboard.com
• test.mosquitto.org

In this exercise we will use the one offered by HiveMQ (broker.hivemq.com). Fill in the data as indicated below:

Then click on ADVANCED, and add the topic test/#:

Now click on BACK, and then on CONNECT. You will start seeing something like this:

Where all this data coming from?

Some basic exercises

Let's start with an easy one. Click on the DISCONNECT button. Now add a subscription to the topic Spain/Valencia/UPV, then CONNECT once again.

Now, (1) write the same identical text for the topic (i.e., Spain/Valencia/UPV) in the topic field of the Publish section, select the raw option and write a text in the box below. The text can be whatever you want (e.g., Ciao!!). When you are done click on the Publish button.

You'll get something like the image below… plus all the messages written by all the other clients. With just one publish action you actually reached various devices!! (THis is called many-to-many communications!)

Quick tests:
a. What happens if you type spain/valencia/upv instead?
b. How would you build a messaging application with a group of friends?

Simple data collection using MQTT

You will now read the values of two LoRaWAN sensors that are periodically sending their data to the TTN Network Server from the GRC lab.

You have to use the MQTT Explore with the following parameters:

Broker: eu1.cloud.thethings.network:1883
Username: lopys2ttn@ttn
Password: NNSXS.A55Z2P4YCHH2RQ7ONQVXFCX2IPMPJQLXAPKQSWQ.A5AB4GALMW623GZMJEWNIVRQSMRMZF4CHDBTTEQYRAOFKBH35G2A
Topic: v3/+/devices/#

MQTT with python

All the code necessary for this Lab session is available here

. You can execute it either in your computer or online.

executing online. Create an account in https://repl.it

executing in your computer. You must have python3 and the paho-mqtt library installed ($ sudo pip3 install paho-mqtt)

The documentation of the MQTT Paho API is here: https://www.eclipse.org/paho/clients/python/docs/

A simple subscriber

File: sisub.py cointains the code of a simple python subscriber. This code connects to a public broker and subscribes to topic $SYS/#. Let's give it a try.

import paho.mqtt.client as mqtt

THE_BROKER = "test.mosquitto.org"
THE_TOPIC  = "$SYS/#"


# Callback function used when the client receives a CONNACK response from the broker.
def on_connect(client, userdata, flags, rc):
    print("connected to ", client._host, "port: ", client._port)
    print("flags: ", flags, "returned code: ", rc)

    client.subscribe(THE_TOPIC, qos=0)

# Callback function used when the client receives a message from the broker.
def on_message(client, userdata, msg):
    print("message received with topic: {} and payload: {}".format(msg.topic, str(msg.payload)))

if __name__ == "__main__":

	client = mqtt.Client(client_id="", 
                             clean_session=True, 
                             userdata=None, 
                             protocol=mqtt.MQTTv311, 
                             transport="tcp")

	client.on_connect = on_connect
	client.on_message = on_message

	client.username_pw_set(username=None, password=None)
	client.connect(THE_BROKER, port=1883, keepalive=60)

	# Blocking call that processes network traffic, dispatches callbacks and
	# handles reconnecting.
	client.loop_forever()

A simple producer

File: sipub.py cointains the code of a simple python producer. This code connects to a public broker and periodically publishes random values to topic "PMtest/rndvalue"

import random
import time

import paho.mqtt.client as mqtt

THE_BROKER = "test.mosquitto.org"
THE_TOPIC  = "PMtest/rndvalue"

# Callback function used when the client receives a CONNACK response from the broker.
def on_connect(client, userdata, flags, rc):
    print("connected to ", client._host, "port: ", client._port)
    print("flags: ", flags, "returned code: ", rc)

# Callback function used when a message is published.
def on_publish(client, userdata, mid):
    print("msg published (mid={})".format(mid))

if __name__ == "__main__":

	client = mqtt.Client(client_id="", 
			     clean_session=True, 
			     userdata=None, 
			     protocol=mqtt.MQTTv311, 
			     transport="tcp")

	client.on_connect = on_connect
	client.on_publish = on_publish

	client.username_pw_set(username=None, password=None)
	client.connect(THE_BROKER, port=1883, keepalive=60)

	client.loop_start()

	while True:

		msg_to_be_sent = random.randint(0, 100)
		print("publishing: ", msg_to_be_sent)
		client.publish(THE_TOPIC, 
					   payload=msg_to_be_sent, 
					   qos=0, 
					   retain=False)

		time.sleep(5)

	client.loop_stop()

What should be modified in sisub.py so that it can receive the data sent by sipub.py?

Getting data from TTN

Repeat the previous example in which, using the client http://mqtt-explorer.com we read data from TTN. In this case it is enough to print all the JSON that arrives.

The necessary parameters are:

Broker: eu1.cloud.thethings.network
Username: lopys2ttn@ttn
Password: NNSXS.A55Z2P4YCHH2RQ7ONQVXFCX2IPMPJQLXAPKQSWQ.A5AB4GALMW623GZMJEWNIVRQSMRMZF4CHDBTTEQYRAOFKBH35G2A
Topic: v3/+/devices/#

What should be modified in sisub.py so that it can receive the data from TTN?

Final labs

Starting an MQTT broker as a container

An MQTT broker can be executed as a Docker container. This fact provides flexibility when deploying these services, especially in edge devices, like Raspberry Pi or similar.
Searching for MQTT, you'll get:

$ docker search mqtt
NAME                                                  DESCRIPTION                                     STARS     OFFICIAL   AUTOMATED
eclipse-mosquitto                                     Eclipse Mosquitto is an open source message …   899       [OK]       
ncarlier/mqtt                                         MQTT Docker image based on Debian.              34                   [OK]
efrecon/mqtt-client                                   Minimal MQTT clients based on mosquitto         10                   [OK]
dersimn/mqtt-admin                                    Docker image for https://github.com/hobbyqua…   5                    [OK]
...        

The first one, eclipse-mosquitto, is the official image for the Mosquitto MQTT broker, probably the most widely used MQTT broker.

The third one, efrecon/mqtt-client is also a helpful image that allows having a handy client to test deployments.

The docker-compose.yml file below indicates how to start up an MQTT broker locally:

version: '3'

services:
    mosquitto:
        image: eclipse-mosquitto:latest
        container_name: broker1
        ports:
            - 1883:1883
        volumes:
            - ./mosquitto.conf:/mosquitto/config/mosquitto.conf

Executing docker compose up you will get something like:

$ docker compose up
[+] Running 1/0
 ⠿ Container broker1  Created                                               0.0s
Attaching to broker1
broker1  | 1644776311: mosquitto version 2.0.14 starting
broker1  | 1644776311: Config loaded from /mosquitto/config/mosquitto.conf.
broker1  | 1644776311: Opening ipv4 listen socket on port 1883.
broker1  | 1644776311: Opening ipv6 listen socket on port 1883.
broker1  | 1644776311: mosquitto version 2.0.14 running

giving:

$ docker ps -a
CONTAINER ID   IMAGE                      COMMAND                  CREATED              STATUS              PORTS                    NAMES
5f94617ba10f   eclipse-mosquitto:latest   "/docker-entrypoint.…"   About a minute ago   Up About a minute   0.0.0.0:1883->1883/tcp   broker1

this is obtained with a ./mosquitto/config/mosquitto.conf file as simple as this:

allow_anonymous true
listener 1883

Try using this broker using "MQTT explorer" as a subscriber and sipub.py as a publisher.

MQTT with micropythom

In the shared folder you can find examples of a simple publisher and a subscriber written in micropython for the LoPys.

For example, the code of a subscriber would be this:

from mqtt import MQTTClient
import time
import pycom

import ufun

wifi_ssid = "..."
wifi_passwd = '...'
broker_addr = "broker.hivemq.com"

def settimeout(duration):
   pass

def on_message(topic, msg):

    print("topic is: "+str(topic))
    print("msg is: "+str(msg))

### if __name__ == "__main__":

ufun.connect_to_wifi(wifi_ssid, wifi_passwd)

client = MQTTClient("PMtesting", broker_addr, 1883)
client.set_callback(on_message)

if not client.connect():
    print ("Connected to broker: "+broker_addr)
client.subscribe('test/#')

print('Waiting messages...')
while 1:
    client.wait_msg()