Vergil Cola / Mbed OS HelloMQTT Featured

A Threaded Secure MQTT Client example. Uses MBED TLS for SSL/TLS connection. QoS0 only for now. Example has been tested with K64F connected via Ethernet.

Dependencies:   FP MQTTPacket

Fork of HelloMQTT by MQTT

main.cpp@21:a7506c90aa84, 2017-03-21 (annotated)

Committer:
vpcola
Date:
Tue Mar 21 03:32:27 2017 +0000
Revision:
21:a7506c90aa84
Parent:
20:49c9daf2b0ff
Child:
22:826657a00c44
test

Who changed what in which revision?

UserRevisionLine numberNew contents of line
icraggs 1:a1d5c7a6acbc 1 /*******************************************************************************
icraggs 17:0811bdbdd78a 2 * Copyright (c) 2014, 2015 IBM Corp.
icraggs 1:a1d5c7a6acbc 3 *
icraggs 1:a1d5c7a6acbc 4 * All rights reserved. This program and the accompanying materials
icraggs 1:a1d5c7a6acbc 5 * are made available under the terms of the Eclipse Public License v1.0
icraggs 1:a1d5c7a6acbc 6 * and Eclipse Distribution License v1.0 which accompany this distribution.
icraggs 1:a1d5c7a6acbc 7 *
icraggs 1:a1d5c7a6acbc 8 * The Eclipse Public License is available at
icraggs 1:a1d5c7a6acbc 9 * http://www.eclipse.org/legal/epl-v10.html
icraggs 1:a1d5c7a6acbc 10 * and the Eclipse Distribution License is available at
icraggs 1:a1d5c7a6acbc 11 * http://www.eclipse.org/org/documents/edl-v10.php.
icraggs 1:a1d5c7a6acbc 12 *
icraggs 1:a1d5c7a6acbc 13 * Contributors:
icraggs 1:a1d5c7a6acbc 14 * Ian Craggs - initial API and implementation and/or initial documentation
icraggs 17:0811bdbdd78a 15 * Ian Craggs - make sure QoS2 processing works, and add device headers
icraggs 1:a1d5c7a6acbc 16 *******************************************************************************/
Jan Jongboom 20:49c9daf2b0ff 17
icraggs 2:638c854c0695 18 /**
icraggs 2:638c854c0695 19 This is a sample program to illustrate the use of the MQTT Client library
icraggs 2:638c854c0695 20 on the mbed platform. The Client class requires two classes which mediate
icraggs 2:638c854c0695 21 access to system interfaces for networking and timing. As long as these two
icraggs 2:638c854c0695 22 classes provide the required public programming interfaces, it does not matter
icraggs 2:638c854c0695 23 what facilities they use underneath. In this program, they use the mbed
icraggs 2:638c854c0695 24 system libraries.
Jan Jongboom 20:49c9daf2b0ff 25
icraggs 2:638c854c0695 26 */
Jan Jongboom 20:49c9daf2b0ff 27
icraggs 17:0811bdbdd78a 28 #define MQTTCLIENT_QOS2 1
vpcola 21:a7506c90aa84 29 #include "mbed.h"
vpcola 21:a7506c90aa84 30 #include "rtos.h"
Jan Jongboom 20:49c9daf2b0ff 31 #include "easy-connect.h"
Jan Jongboom 20:49c9daf2b0ff 32 #include "MQTTNetwork.h"
Jan Jongboom 20:49c9daf2b0ff 33 #include "MQTTmbed.h"
icraggs 2:638c854c0695 34 #include "MQTTClient.h"
icraggs 2:638c854c0695 35
vpcola 21:a7506c90aa84 36 Serial pc(USBTX, USBRX, 115200);
icraggs 2:638c854c0695 37 int arrivedcount = 0;
icraggs 2:638c854c0695 38
vpcola 21:a7506c90aa84 39 Thread msgSender;
vpcola 21:a7506c90aa84 40
vpcola 21:a7506c90aa84 41 static MemoryPool<MQTT::Message, 16> pool;
vpcola 21:a7506c90aa84 42 static Queue<MQTT::Message, 16> queue;
icraggs 8:a3e3113054a1 43
icraggs 9:5beb8609e9f7 44 void messageArrived(MQTT::MessageData& md)
icraggs 2:638c854c0695 45 {
icraggs 9:5beb8609e9f7 46 MQTT::Message &message = md.message;
vpcola 21:a7506c90aa84 47 printf("Message arrived: qos %d, retained %d, dup %d, packetid %d\r\n", message.qos, message.retained, message.dup, message.id);
vpcola 21:a7506c90aa84 48 printf("Payload %.*s\r\n", message.payloadlen, (char*)message.payload);
icraggs 2:638c854c0695 49 ++arrivedcount;
icraggs 2:638c854c0695 50 }
icraggs 0:0cae29831d01 51
vpcola 21:a7506c90aa84 52 void SendDataThread()
vpcola 21:a7506c90aa84 53 {
vpcola 21:a7506c90aa84 54 unsigned int i;
vpcola 21:a7506c90aa84 55 while(true)
vpcola 21:a7506c90aa84 56 {
vpcola 21:a7506c90aa84 57 MQTT::Message * message = pool.alloc();
vpcola 21:a7506c90aa84 58 char * buff = new char[sizeof(char) * 100];
vpcola 21:a7506c90aa84 59
vpcola 21:a7506c90aa84 60 sprintf(buff, "message test %d", i);
vpcola 21:a7506c90aa84 61 message->qos = MQTT::QOS0;
vpcola 21:a7506c90aa84 62 message->retained = false;
vpcola 21:a7506c90aa84 63 message->dup = false;
vpcola 21:a7506c90aa84 64 message->payload = (void*)buff;
vpcola 21:a7506c90aa84 65 message->payloadlen = strlen(buff)+1;
vpcola 21:a7506c90aa84 66
vpcola 21:a7506c90aa84 67 // publish the message to mqtt
vpcola 21:a7506c90aa84 68 queue.put(message);
vpcola 21:a7506c90aa84 69 i++;
vpcola 21:a7506c90aa84 70
vpcola 21:a7506c90aa84 71 Thread::wait(2000);
vpcola 21:a7506c90aa84 72 }
vpcola 21:a7506c90aa84 73 }
icraggs 2:638c854c0695 74
icraggs 2:638c854c0695 75 int main(int argc, char* argv[])
Jan Jongboom 20:49c9daf2b0ff 76 {
Jan Jongboom 20:49c9daf2b0ff 77 float version = 0.6;
icraggs 2:638c854c0695 78 char* topic = "mbed-sample";
Jan Jongboom 20:49c9daf2b0ff 79
vpcola 21:a7506c90aa84 80 printf("HelloMQTT: version is %.2f\r\n", version);
Jan Jongboom 20:49c9daf2b0ff 81
Jan Jongboom 20:49c9daf2b0ff 82 NetworkInterface* network = easy_connect(true);
Jan Jongboom 20:49c9daf2b0ff 83 if (!network) {
Jan Jongboom 20:49c9daf2b0ff 84 return -1;
Jan Jongboom 20:49c9daf2b0ff 85 }
Jan Jongboom 20:49c9daf2b0ff 86
Jan Jongboom 20:49c9daf2b0ff 87 MQTTNetwork mqttNetwork(network);
Jan Jongboom 20:49c9daf2b0ff 88
Jan Jongboom 20:49c9daf2b0ff 89 MQTT::Client<MQTTNetwork, Countdown> client = MQTT::Client<MQTTNetwork, Countdown>(mqttNetwork);
Jan Jongboom 20:49c9daf2b0ff 90
Jan Jongboom 20:49c9daf2b0ff 91 const char* hostname = "m2m.eclipse.org";
icraggs 6:e4c690c45021 92 int port = 1883;
vpcola 21:a7506c90aa84 93 printf("Connecting to %s:%d\r\n", hostname, port);
Jan Jongboom 20:49c9daf2b0ff 94 int rc = mqttNetwork.connect(hostname, port);
icraggs 6:e4c690c45021 95 if (rc != 0)
vpcola 21:a7506c90aa84 96 printf("rc from TCP connect is %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 97
Jan Jongboom 20:49c9daf2b0ff 98 MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
icraggs 6:e4c690c45021 99 data.MQTTVersion = 3;
icraggs 8:a3e3113054a1 100 data.clientID.cstring = "mbed-sample";
icraggs 12:086a9314e8a5 101 data.username.cstring = "testuser";
icraggs 12:086a9314e8a5 102 data.password.cstring = "testpassword";
icraggs 16:28d062c5522b 103 if ((rc = client.connect(data)) != 0)
vpcola 21:a7506c90aa84 104 printf("rc from MQTT connect is %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 105
icraggs 19:7f7aba7a4a8b 106 if ((rc = client.subscribe(topic, MQTT::QOS2, messageArrived)) != 0)
vpcola 21:a7506c90aa84 107 printf("rc from MQTT subscribe is %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 108
vpcola 21:a7506c90aa84 109 // Start the data source
vpcola 21:a7506c90aa84 110 msgSender.start(SendDataThread);
vpcola 21:a7506c90aa84 111
vpcola 21:a7506c90aa84 112 while(true)
vpcola 21:a7506c90aa84 113 {
vpcola 21:a7506c90aa84 114 osEvent evt = queue.get(10);
vpcola 21:a7506c90aa84 115 if (evt.status == osEventMessage) {
vpcola 21:a7506c90aa84 116 printf("Message arrived from main thread ...\r\n");
vpcola 21:a7506c90aa84 117 // Unpack the message
vpcola 21:a7506c90aa84 118 MQTT::Message * message = (MQTT::Message *)evt.value.p;
vpcola 21:a7506c90aa84 119
vpcola 21:a7506c90aa84 120 printf("Publishing message to MQTT ...\r\n");
vpcola 21:a7506c90aa84 121 // Push to mqtt
vpcola 21:a7506c90aa84 122 int rc = client.publish(topic, *message);
vpcola 21:a7506c90aa84 123 if (rc < 0)
vpcola 21:a7506c90aa84 124 printf("Error sending mqtt message \r\n");
vpcola 21:a7506c90aa84 125 else
vpcola 21:a7506c90aa84 126 printf("Message published ...\r\n");
vpcola 21:a7506c90aa84 127
vpcola 21:a7506c90aa84 128 printf("Deleting payload ...\r\n");
vpcola 21:a7506c90aa84 129 // Delete payload
vpcola 21:a7506c90aa84 130 delete [] message->payload;
vpcola 21:a7506c90aa84 131
vpcola 21:a7506c90aa84 132 printf("Deleting pool allocation ...\r\n");
vpcola 21:a7506c90aa84 133 // Don't forget this!
vpcola 21:a7506c90aa84 134 pool.free(message);
vpcola 21:a7506c90aa84 135 }
vpcola 21:a7506c90aa84 136
vpcola 21:a7506c90aa84 137 printf("MQTT client yeild ...\r\n");
vpcola 21:a7506c90aa84 138 if (client.yield(100) != MQTT::SUCCESS)
vpcola 21:a7506c90aa84 139 {
vpcola 21:a7506c90aa84 140 printf("Yield error, client disconnected? ...\r\n");
vpcola 21:a7506c90aa84 141 break;
vpcola 21:a7506c90aa84 142 }
vpcola 21:a7506c90aa84 143 printf("MQTT client yeild successful ...\r\n");
vpcola 21:a7506c90aa84 144 }
Jan Jongboom 20:49c9daf2b0ff 145
icraggs 17:0811bdbdd78a 146 if ((rc = client.unsubscribe(topic)) != 0)
vpcola 21:a7506c90aa84 147 printf("rc from unsubscribe was %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 148
icraggs 8:a3e3113054a1 149 if ((rc = client.disconnect()) != 0)
vpcola 21:a7506c90aa84 150 printf("rc from disconnect was %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 151
Jan Jongboom 20:49c9daf2b0ff 152 mqttNetwork.disconnect();
Jan Jongboom 20:49c9daf2b0ff 153
vpcola 21:a7506c90aa84 154 printf("Version %.2f: finish %d msgs\r\n", version, arrivedcount);
Jan Jongboom 20:49c9daf2b0ff 155
icraggs 0:0cae29831d01 156 return 0;
icraggs 0:0cae29831d01 157 }