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@22:826657a00c44, 2017-03-21 (annotated)

Committer:
vpcola
Date:
Tue Mar 21 12:57:07 2017 +0000
Revision:
22:826657a00c44
Parent:
21:a7506c90aa84
Child:
23:06fac173529e
second commit;

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
vpcola 22:826657a00c44 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"
vpcola 22:826657a00c44 32 #include "MQTTLogging.h"
Jan Jongboom 20:49c9daf2b0ff 33 #include "MQTTNetwork.h"
Jan Jongboom 20:49c9daf2b0ff 34 #include "MQTTmbed.h"
icraggs 2:638c854c0695 35 #include "MQTTClient.h"
icraggs 2:638c854c0695 36
vpcola 21:a7506c90aa84 37 Serial pc(USBTX, USBRX, 115200);
icraggs 2:638c854c0695 38 int arrivedcount = 0;
icraggs 2:638c854c0695 39
vpcola 21:a7506c90aa84 40 Thread msgSender;
vpcola 21:a7506c90aa84 41
vpcola 21:a7506c90aa84 42 static MemoryPool<MQTT::Message, 16> pool;
vpcola 21:a7506c90aa84 43 static Queue<MQTT::Message, 16> queue;
icraggs 8:a3e3113054a1 44
icraggs 9:5beb8609e9f7 45 void messageArrived(MQTT::MessageData& md)
icraggs 2:638c854c0695 46 {
icraggs 9:5beb8609e9f7 47 MQTT::Message &message = md.message;
vpcola 21:a7506c90aa84 48 printf("Message arrived: qos %d, retained %d, dup %d, packetid %d\r\n", message.qos, message.retained, message.dup, message.id);
vpcola 22:826657a00c44 49 printf("Payload [%.*s]\r\n", message.payloadlen, (char*)message.payload);
icraggs 2:638c854c0695 50 ++arrivedcount;
icraggs 2:638c854c0695 51 }
icraggs 0:0cae29831d01 52
vpcola 21:a7506c90aa84 53 void SendDataThread()
vpcola 21:a7506c90aa84 54 {
vpcola 21:a7506c90aa84 55 unsigned int i;
vpcola 21:a7506c90aa84 56 while(true)
vpcola 21:a7506c90aa84 57 {
vpcola 21:a7506c90aa84 58 MQTT::Message * message = pool.alloc();
vpcola 21:a7506c90aa84 59 char * buff = new char[sizeof(char) * 100];
vpcola 21:a7506c90aa84 60
vpcola 21:a7506c90aa84 61 sprintf(buff, "message test %d", i);
vpcola 21:a7506c90aa84 62 message->qos = MQTT::QOS0;
vpcola 21:a7506c90aa84 63 message->retained = false;
vpcola 21:a7506c90aa84 64 message->dup = false;
vpcola 21:a7506c90aa84 65 message->payload = (void*)buff;
vpcola 21:a7506c90aa84 66 message->payloadlen = strlen(buff)+1;
vpcola 21:a7506c90aa84 67
vpcola 21:a7506c90aa84 68 // publish the message to mqtt
vpcola 21:a7506c90aa84 69 queue.put(message);
vpcola 21:a7506c90aa84 70 i++;
vpcola 21:a7506c90aa84 71
vpcola 21:a7506c90aa84 72 Thread::wait(2000);
vpcola 21:a7506c90aa84 73 }
vpcola 21:a7506c90aa84 74 }
icraggs 2:638c854c0695 75
icraggs 2:638c854c0695 76 int main(int argc, char* argv[])
Jan Jongboom 20:49c9daf2b0ff 77 {
Jan Jongboom 20:49c9daf2b0ff 78 float version = 0.6;
icraggs 2:638c854c0695 79 char* topic = "mbed-sample";
Jan Jongboom 20:49c9daf2b0ff 80
vpcola 21:a7506c90aa84 81 printf("HelloMQTT: version is %.2f\r\n", version);
Jan Jongboom 20:49c9daf2b0ff 82
Jan Jongboom 20:49c9daf2b0ff 83 NetworkInterface* network = easy_connect(true);
Jan Jongboom 20:49c9daf2b0ff 84 if (!network) {
Jan Jongboom 20:49c9daf2b0ff 85 return -1;
Jan Jongboom 20:49c9daf2b0ff 86 }
Jan Jongboom 20:49c9daf2b0ff 87
Jan Jongboom 20:49c9daf2b0ff 88 MQTTNetwork mqttNetwork(network);
Jan Jongboom 20:49c9daf2b0ff 89
Jan Jongboom 20:49c9daf2b0ff 90 MQTT::Client<MQTTNetwork, Countdown> client = MQTT::Client<MQTTNetwork, Countdown>(mqttNetwork);
Jan Jongboom 20:49c9daf2b0ff 91
Jan Jongboom 20:49c9daf2b0ff 92 const char* hostname = "m2m.eclipse.org";
icraggs 6:e4c690c45021 93 int port = 1883;
vpcola 21:a7506c90aa84 94 printf("Connecting to %s:%d\r\n", hostname, port);
Jan Jongboom 20:49c9daf2b0ff 95 int rc = mqttNetwork.connect(hostname, port);
icraggs 6:e4c690c45021 96 if (rc != 0)
vpcola 21:a7506c90aa84 97 printf("rc from TCP connect is %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 98
Jan Jongboom 20:49c9daf2b0ff 99 MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
icraggs 6:e4c690c45021 100 data.MQTTVersion = 3;
icraggs 8:a3e3113054a1 101 data.clientID.cstring = "mbed-sample";
icraggs 12:086a9314e8a5 102 data.username.cstring = "testuser";
icraggs 12:086a9314e8a5 103 data.password.cstring = "testpassword";
icraggs 16:28d062c5522b 104 if ((rc = client.connect(data)) != 0)
vpcola 21:a7506c90aa84 105 printf("rc from MQTT connect is %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 106
vpcola 22:826657a00c44 107 if ((rc = client.subscribe(topic, MQTT::QOS0, messageArrived)) != 0)
vpcola 21:a7506c90aa84 108 printf("rc from MQTT subscribe is %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 109
vpcola 22:826657a00c44 110 // Start the data producer
vpcola 21:a7506c90aa84 111 msgSender.start(SendDataThread);
vpcola 21:a7506c90aa84 112
vpcola 21:a7506c90aa84 113 while(true)
vpcola 21:a7506c90aa84 114 {
vpcola 21:a7506c90aa84 115 osEvent evt = queue.get(10);
vpcola 21:a7506c90aa84 116 if (evt.status == osEventMessage) {
vpcola 22:826657a00c44 117 //printf("Message arrived from main thread ...\r\n");
vpcola 21:a7506c90aa84 118 // Unpack the message
vpcola 21:a7506c90aa84 119 MQTT::Message * message = (MQTT::Message *)evt.value.p;
vpcola 21:a7506c90aa84 120
vpcola 21:a7506c90aa84 121 printf("Publishing message to MQTT ...\r\n");
vpcola 21:a7506c90aa84 122 // Push to mqtt
vpcola 22:826657a00c44 123 rc = client.publish(topic, *message);
vpcola 21:a7506c90aa84 124 if (rc < 0)
vpcola 22:826657a00c44 125 printf("Error sending mqtt message rc = %d\r\n", rc);
vpcola 21:a7506c90aa84 126 else
vpcola 21:a7506c90aa84 127 printf("Message published ...\r\n");
vpcola 21:a7506c90aa84 128
vpcola 22:826657a00c44 129 //printf("Deleting payload ...\r\n");
vpcola 21:a7506c90aa84 130 // Delete payload
vpcola 21:a7506c90aa84 131 delete [] message->payload;
vpcola 21:a7506c90aa84 132
vpcola 22:826657a00c44 133 //printf("Deleting pool allocation ...\r\n");
vpcola 21:a7506c90aa84 134 // Don't forget this!
vpcola 21:a7506c90aa84 135 pool.free(message);
vpcola 21:a7506c90aa84 136 }
vpcola 21:a7506c90aa84 137
vpcola 22:826657a00c44 138 //printf("MQTT client yield ...\r\n");
vpcola 21:a7506c90aa84 139 if (client.yield(100) != MQTT::SUCCESS)
vpcola 21:a7506c90aa84 140 {
vpcola 21:a7506c90aa84 141 printf("Yield error, client disconnected? ...\r\n");
vpcola 21:a7506c90aa84 142 break;
vpcola 21:a7506c90aa84 143 }
vpcola 22:826657a00c44 144 //printf("MQTT client yield successful ...\r\n");
vpcola 21:a7506c90aa84 145 }
Jan Jongboom 20:49c9daf2b0ff 146
icraggs 17:0811bdbdd78a 147 if ((rc = client.unsubscribe(topic)) != 0)
vpcola 21:a7506c90aa84 148 printf("rc from unsubscribe was %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 149
icraggs 8:a3e3113054a1 150 if ((rc = client.disconnect()) != 0)
vpcola 21:a7506c90aa84 151 printf("rc from disconnect was %d\r\n", rc);
Jan Jongboom 20:49c9daf2b0ff 152
Jan Jongboom 20:49c9daf2b0ff 153 mqttNetwork.disconnect();
Jan Jongboom 20:49c9daf2b0ff 154
vpcola 21:a7506c90aa84 155 printf("Version %.2f: finish %d msgs\r\n", version, arrivedcount);
Jan Jongboom 20:49c9daf2b0ff 156
icraggs 0:0cae29831d01 157 return 0;
icraggs 0:0cae29831d01 158 }