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

mbed OS5に対応したMilkcocoaライブラリのテストバージョンです。

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 }