Zoltan Hudak / Mbed OS MQTT_Hello

MQTT client to test the ENC28J60-EMAC on NUCLEO-F446RE.

Dependencies:   ENC28J60-EMAC

MQTT/MQTTPacket/samples/publish-subscribe.txt@5:d9570dbf2f82, 2021-03-29 (annotated)

Committer:
hudakz
Date:
Mon Mar 29 09:32:44 2021 +0000
Revision:
5:d9570dbf2f82
Parent:
0:238f0d0c0ba3 
MQTT client to test the ENC28J60-EMAC on NUCLEO-F446RE.

Who changed what in which revision?

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