Junichi Katsu / Mbed OS mbed-os-example-milkcocoa

Important changes to repositories hosted on mbed.com

Mbed hosted mercurial repositories are deprecated and are due to be permanently deleted in July 2026.

To keep a copy of this software download the repository Zip archive or clone locally using Mercurial.

It is also possible to export all your personal repositories from the account settings page.

Fork of mbed-os-example-mbed5-blinky by mbed-os-examples

Milkcocoa/MQTT/MQTTPacket/samples/publish-subscribe.txt@24:6ba1245bf049, 2017-01-24 (annotated)

Committer:
jksoft
Date:
Tue Jan 24 13:41:36 2017 +0000
Revision:
24:6ba1245bf049
??????????

Who changed what in which revision?

UserRevisionLine numberNew contents of line
jksoft 24:6ba1245bf049 1 #include "MQTTPacket.h"
jksoft 24:6ba1245bf049 2
jksoft 24:6ba1245bf049 3 #include <errno.h>
jksoft 24:6ba1245bf049 4 #include <stdlib.h>
jksoft 24:6ba1245bf049 5 #include <string.h>
jksoft 24:6ba1245bf049 6
jksoft 24:6ba1245bf049 7 #include "EthernetInterface.h"
jksoft 24:6ba1245bf049 8
jksoft 24:6ba1245bf049 9
jksoft 24:6ba1245bf049 10 TCPSocketConnection mysock;
jksoft 24:6ba1245bf049 11
jksoft 24:6ba1245bf049 12 int getdata(char* buf, int count)
jksoft 24:6ba1245bf049 13 {
jksoft 24:6ba1245bf049 14 return mysock.receive(buf, (size_t)count);
jksoft 24:6ba1245bf049 15 }
jksoft 24:6ba1245bf049 16
jksoft 24:6ba1245bf049 17 int toStop = 0;
jksoft 24:6ba1245bf049 18
jksoft 24:6ba1245bf049 19
jksoft 24:6ba1245bf049 20 int main()
jksoft 24:6ba1245bf049 21 {
jksoft 24:6ba1245bf049 22 MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
jksoft 24:6ba1245bf049 23 int rc = 0;
jksoft 24:6ba1245bf049 24 char buf[200];
jksoft 24:6ba1245bf049 25 int buflen = sizeof(buf);
jksoft 24:6ba1245bf049 26 int msgid = 1;
jksoft 24:6ba1245bf049 27 MQTTString topicString = MQTTString_initializer;
jksoft 24:6ba1245bf049 28 int req_qos = 0;
jksoft 24:6ba1245bf049 29 char* payload = "mypayload";
jksoft 24:6ba1245bf049 30 int payloadlen = strlen(payload);
jksoft 24:6ba1245bf049 31 int len = 0;
jksoft 24:6ba1245bf049 32 EthernetInterface eth;
jksoft 24:6ba1245bf049 33
jksoft 24:6ba1245bf049 34 eth.init(); //Use DHCP
jksoft 24:6ba1245bf049 35 eth.connect();
jksoft 24:6ba1245bf049 36
jksoft 24:6ba1245bf049 37 rc = mysock.connect("m2m.eclipse.org", 1883);
jksoft 24:6ba1245bf049 38
jksoft 24:6ba1245bf049 39 data.clientID.cstring = "SendReceive mbed MQTT ";
jksoft 24:6ba1245bf049 40 data.keepAliveInterval = 20;
jksoft 24:6ba1245bf049 41 data.cleansession = 1;
jksoft 24:6ba1245bf049 42
jksoft 24:6ba1245bf049 43 mysock.set_blocking(true, 1000); /* 1 second Timeout */
jksoft 24:6ba1245bf049 44
jksoft 24:6ba1245bf049 45 len = MQTTSerialize_connect(buf, buflen, &data);
jksoft 24:6ba1245bf049 46 rc = mysock.send(buf, len);
jksoft 24:6ba1245bf049 47
jksoft 24:6ba1245bf049 48 /* wait for connack */
jksoft 24:6ba1245bf049 49 if (MQTTPacket_read(buf, buflen, getdata) == CONNACK)
jksoft 24:6ba1245bf049 50 {
jksoft 24:6ba1245bf049 51 int connack_rc;
jksoft 24:6ba1245bf049 52
jksoft 24:6ba1245bf049 53 if (MQTTDeserialize_connack(&connack_rc, buf, buflen) != 1 || connack_rc != 0)
jksoft 24:6ba1245bf049 54 {
jksoft 24:6ba1245bf049 55 printf("Unable to connect, return code %d\n", connack_rc);
jksoft 24:6ba1245bf049 56 goto exit;
jksoft 24:6ba1245bf049 57 }
jksoft 24:6ba1245bf049 58 }
jksoft 24:6ba1245bf049 59 else
jksoft 24:6ba1245bf049 60 goto exit;
jksoft 24:6ba1245bf049 61
jksoft 24:6ba1245bf049 62 /* subscribe */
jksoft 24:6ba1245bf049 63 topicString.cstring = "substopic";
jksoft 24:6ba1245bf049 64 len = MQTTSerialize_subscribe(buf, buflen, 0, msgid, 1, &topicString, &req_qos);
jksoft 24:6ba1245bf049 65
jksoft 24:6ba1245bf049 66 rc = mysock.send(buf, len);
jksoft 24:6ba1245bf049 67 if (MQTTPacket_read(buf, buflen, getdata) == SUBACK) /* wait for suback */
jksoft 24:6ba1245bf049 68 {
jksoft 24:6ba1245bf049 69 int submsgid;
jksoft 24:6ba1245bf049 70 int subcount;
jksoft 24:6ba1245bf049 71 int granted_qos;
jksoft 24:6ba1245bf049 72
jksoft 24:6ba1245bf049 73 rc = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf, buflen);
jksoft 24:6ba1245bf049 74 if (granted_qos != 0)
jksoft 24:6ba1245bf049 75 {
jksoft 24:6ba1245bf049 76 printf("granted qos != 0, %d\n", granted_qos);
jksoft 24:6ba1245bf049 77 goto exit;
jksoft 24:6ba1245bf049 78 }
jksoft 24:6ba1245bf049 79 }
jksoft 24:6ba1245bf049 80 else
jksoft 24:6ba1245bf049 81 goto exit;
jksoft 24:6ba1245bf049 82
jksoft 24:6ba1245bf049 83 topicString.cstring = "pubtopic";
jksoft 24:6ba1245bf049 84 while (!toStop)
jksoft 24:6ba1245bf049 85 {
jksoft 24:6ba1245bf049 86 if (MQTTPacket_read(buf, buflen, getdata) == PUBLISH)
jksoft 24:6ba1245bf049 87 {
jksoft 24:6ba1245bf049 88 int dup;
jksoft 24:6ba1245bf049 89 int qos;
jksoft 24:6ba1245bf049 90 int retained;
jksoft 24:6ba1245bf049 91 int msgid;
jksoft 24:6ba1245bf049 92 int payloadlen_in;
jksoft 24:6ba1245bf049 93 char* payload_in;
jksoft 24:6ba1245bf049 94 int rc;
jksoft 24:6ba1245bf049 95 MQTTString receivedTopic;
jksoft 24:6ba1245bf049 96
jksoft 24:6ba1245bf049 97 rc = MQTTDeserialize_publish(&dup, &qos, &retained, &msgid, &receivedTopic,
jksoft 24:6ba1245bf049 98 &payload_in, &payloadlen_in, buf, buflen);
jksoft 24:6ba1245bf049 99 printf("message arrived %.*s\n", payloadlen_in, payload_in);
jksoft 24:6ba1245bf049 100 }
jksoft 24:6ba1245bf049 101
jksoft 24:6ba1245bf049 102 printf("publishing reading\n");
jksoft 24:6ba1245bf049 103 len = MQTTSerialize_publish(buf, buflen, 0, 0, 0, 0, topicString, payload, payloadlen);
jksoft 24:6ba1245bf049 104 rc = mysock.send(buf, len);
jksoft 24:6ba1245bf049 105 }
jksoft 24:6ba1245bf049 106
jksoft 24:6ba1245bf049 107 printf("disconnecting\n");
jksoft 24:6ba1245bf049 108 len = MQTTSerialize_disconnect(buf, buflen);
jksoft 24:6ba1245bf049 109 rc = mysock.send(buf, len);
jksoft 24:6ba1245bf049 110
jksoft 24:6ba1245bf049 111 exit:
jksoft 24:6ba1245bf049 112 eth.disconnect();
jksoft 24:6ba1245bf049 113
jksoft 24:6ba1245bf049 114 return 0;
jksoft 24:6ba1245bf049 115 }