Zhaopeng's Homepage

My tech notebook

MQTT communication via Cloud Broker

Posted on In

In my startup, we have worked on some AIoT (AI + IoT) projects and we’ve done some MVPs as demo to our client. In some certain project, we have used MQTT protocol for collecting sensors data and transporting them to other smart devices. So in this post, I would like to present a simple demo for building a MQTT communication via a broker on AWS cloud.

Alt Text

MQTT (Message Queuing Telemetry Transport) is an open OASIS and ISO standard (ISO/IEC 20922) lightweight, publish-subscribe network protocol that transports messages between devices. The protocol usually runs over TCP/IP; however, any network protocol that provides ordered, lossless, bi-directional connections can support MQTT. It is designed for connections with remote locations where a “small code footprint” is required or the network bandwidth is limited.

Cloud side

On AWS server, I use docker to run an emqx service which play a role as “broker”. 

1
sudo docker run --rm -ti --name emqx -p 18083:18083 -p 1883:1883 -e EMQX_ADMIN_PASSWORD="myPasswd" emqx/emqx:latest

Output of my putty terminal:
Alt Text

When this service is running, one can access to its administration page by visiting http://3.138.200.179:18083/#/:

Alt Text

Subscriber

I have a subcriber.py file which defines a mqtt client who connects to cloud broker with port 1883 and subscribes message named 'qiu_data'. 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import paho.mqtt.client as mqtt

## callback upon connection
def on_connect(client, userdata, flags, rc):
    print("Connected, result code: " + str(rc))

## callback upon arrival of msg
def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.payload))

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
# client.on_disconnect = on_disconnect

## I have not yet bindded a domain name for my AWS ECS
client.connect('3.138.200.179', 1883, 600) # 600ms: keepalive interval setting
client.subscribe('qiu_data', qos=0)
client.loop_forever() # keep running

Publisher

I have a publisher.py file which defines another mqtt client who connects to cloud broker with port 1883 and publish some data in 'qiu_data' messages. 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import paho.mqtt.client as mqtt
import json
import time

def on_connect(client, userdata, flags, rc):
    print("Connected with result code: " + str(rc))

def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.payload))

data = {
    "type": "IoT",
    "timestamp": time.time(),
    "msgId":"8ed7a307-0e82-9738-xxxx",
    "data":{
        "temp":23.5,
        "speed":46.8
    }
}

param = json.dumps(data)

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect('3.138.200.179', 1883, 600) # 600ms for keepalive
client.publish('qiu_data', payload='hello world', qos=0)
time.sleep(1)
client.publish("qiu_data", payload=param, qos=0)

Execution

In a terminal I type python subcriber.py, and in a second terminal I type python publisher.py. See the published topics have been received in the subscriber’s terminal:

Alt Text