mathieu/Adafruit_MQTT_Library
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Handle publish packets when they're not the expected packet type

Browse Source
This commit is contained in:
Lucas Jandrew 2020-02-23 22:42:43 -05:00
parent 259cba0837
commit 5ea3e3ba72
2 changed files with 609 additions and 561 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

283
Adafruit_MQTT.cpp
View File

@ -47,15 +47,17 @@ int strncasecmp(const char * str1, const char * str2, int len) {
void printBuffer(uint8_t *buffer, uint16_t len) { void printBuffer(uint8_t *buffer, uint16_t len) {
DEBUG_PRINTER.print('\t'); DEBUG_PRINTER.print('\t');
for (uint16_t i=0; i<len; i++) { for (uint16_t i = 0; i < len; i++) {
if (isprint(buffer[i])) if (isprint(buffer[i])) {
DEBUG_PRINTER.write(buffer[i]); DEBUG_PRINTER.write(buffer[i]);
else } else {
DEBUG_PRINTER.print(" "); DEBUG_PRINTER.print(" ");
}
DEBUG_PRINTER.print(F(" [0x")); DEBUG_PRINTER.print(F(" [0x"));
if (buffer[i] < 0x10) if (buffer[i] < 0x10) {
DEBUG_PRINTER.print("0"); DEBUG_PRINTER.print("0");
DEBUG_PRINTER.print(buffer[i],HEX); }
DEBUG_PRINTER.print(buffer[i], HEX);
DEBUG_PRINTER.print("], "); DEBUG_PRINTER.print("], ");
if (i % 8 == 7) { if (i % 8 == 7) {
DEBUG_PRINTER.print("\n\t"); DEBUG_PRINTER.print("\n\t");
@ -74,7 +76,7 @@ static uint8_t *stringprint(uint8_t *p, char *s) {
} }
*/ */
static uint8_t *stringprint(uint8_t *p, const char *s, uint16_t maxlen=0) { static uint8_t *stringprint(uint8_t *p, const char *s, uint16_t maxlen = 0) {
// If maxlen is specified (has a non-zero value) then use it as the maximum // If maxlen is specified (has a non-zero value) then use it as the maximum
// length of the source string to write to the buffer. Otherwise write // length of the source string to write to the buffer. Otherwise write
// the entire source string. // the entire source string.
@ -87,10 +89,12 @@ static uint8_t *stringprint(uint8_t *p, const char *s, uint16_t maxlen=0) {
Serial.write(pgm_read_byte(s+i)); Serial.write(pgm_read_byte(s+i));
} }
*/ */
p[0] = len >> 8; p++; p[0] = len >> 8;
p[0] = len & 0xFF; p++; p++;
strncpy((char *)p, s, len); p[0] = len & 0xFF;
return p+len; p++;
strncpy((char *) p, s, len);
return p + len;
} }
@ -108,7 +112,7 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server,
password = pass; password = pass;
// reset subscriptions // reset subscriptions
for (uint8_t i=0; i<MAXSUBSCRIPTIONS; i++) { for (uint8_t i = 0; i < MAXSUBSCRIPTIONS; i++) {
subscriptions[i] = 0; subscriptions[i] = 0;
} }
@ -133,7 +137,7 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server,
password = pass; password = pass;
// reset subscriptions // reset subscriptions
for (uint8_t i=0; i<MAXSUBSCRIPTIONS; i++) { for (uint8_t i = 0; i < MAXSUBSCRIPTIONS; i++) {
subscriptions[i] = 0; subscriptions[i] = 0;
} }
@ -148,37 +152,46 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server,
int8_t Adafruit_MQTT::connect() { int8_t Adafruit_MQTT::connect() {
// Connect to the server. // Connect to the server.
if (!connectServer()) if (!connectServer()) {
return -1; return -1;
}
// Construct and send connect packet. // Construct and send connect packet.
uint8_t len = connectPacket(buffer); uint8_t len = connectPacket(buffer);
if (!sendPacket(buffer, len)) if (!sendPacket(buffer, len)) {
return -1; return -1;
}
// Read connect response packet and verify it // Read connect response packet and verify it
len = readFullPacket(buffer, MAXBUFFERSIZE, CONNECT_TIMEOUT_MS); len = readFullPacket(buffer, MAXBUFFERSIZE, CONNECT_TIMEOUT_MS);
if (len != 4) if (len != 4) {
return -1; return -1;
if ((buffer[0] != (MQTT_CTRL_CONNECTACK << 4)) || (buffer[1] != 2)) }
if ((buffer[0] != (MQTT_CTRL_CONNECTACK << 4)) || (buffer[1] != 2)) {
return -1; return -1;
if (buffer[3] != 0) }
if (buffer[3] != 0) {
return buffer[3]; return buffer[3];
}
// Setup subscriptions once connected. // Setup subscriptions once connected.
for (uint8_t i=0; i<MAXSUBSCRIPTIONS; i++) { for (uint8_t i = 0; i < MAXSUBSCRIPTIONS; i++) {
// Ignore subscriptions that aren't defined. // Ignore subscriptions that aren't defined.
if (subscriptions[i] == 0) continue; if (subscriptions[i] == 0) {
continue;
}
boolean success = false; boolean success = false;
for (uint8_t retry=0; (retry<3) && !success; retry++) { // retry until we get a suback for (uint8_t retry = 0; (retry < 3) && !success; retry++) { // retry until we get a suback
// Construct and send subscription packet. // Construct and send subscription packet.
uint8_t len = subscribePacket(buffer, subscriptions[i]->topic, subscriptions[i]->qos); uint8_t len = subscribePacket(buffer, subscriptions[i]->topic, subscriptions[i]->qos);
if (!sendPacket(buffer, len)) if (!sendPacket(buffer, len)) {
return -1; return -1;
}
if(MQTT_PROTOCOL_LEVEL < 3) // older versions didn't suback if (MQTT_PROTOCOL_LEVEL < 3) { // older versions didn't suback
break; break;
}
// Check for SUBACK if using MQTT 3.1.1 or higher // Check for SUBACK if using MQTT 3.1.1 or higher
// TODO: The Server is permitted to start sending PUBLISH packets matching the // TODO: The Server is permitted to start sending PUBLISH packets matching the
@ -190,14 +203,13 @@ int8_t Adafruit_MQTT::connect() {
break; break;
} }
} }
if (! success) return -2; // failed to sub for some reason if (!success) return -2; // failed to sub for some reason
} }
return 0; return 0;
} }
int8_t Adafruit_MQTT::connect(const char *user, const char *pass) int8_t Adafruit_MQTT::connect(const char *user, const char *pass) {
{
username = user; username = user;
password = pass; password = pass;
return connect(); return connect();
@ -206,20 +218,22 @@ int8_t Adafruit_MQTT::connect(const char *user, const char *pass)
uint16_t Adafruit_MQTT::processPacketsUntil(uint8_t *buffer, uint8_t waitforpackettype, uint16_t timeout) { uint16_t Adafruit_MQTT::processPacketsUntil(uint8_t *buffer, uint8_t waitforpackettype, uint16_t timeout) {
uint16_t len; uint16_t len;
while(true) { while (true) {
len = readFullPacket(buffer, MAXBUFFERSIZE, timeout); len = readFullPacket(buffer, MAXBUFFERSIZE, timeout);
if(len == 0){ if (len == 0) {
break; break;
} }
if ((buffer[0] >> 4) == waitforpackettype) uint8_t packetType = (buffer[0] >> 4);
{ if (packetType == waitforpackettype) {
return len; return len;
} } else {
else // if (packetType == MQTT_CTRL_PUBLISH) {
{ // handleSubscriptionPacket(len);
// } else {
ERROR_PRINTLN(F("Dropped a packet")); ERROR_PRINTLN(F("Dropped a packet"));
// }
} }
} }
return 0; return 0;
@ -235,7 +249,8 @@ uint16_t Adafruit_MQTT::readFullPacket(uint8_t *buffer, uint16_t maxsize, uint16
rlen = readPacket(pbuff, 1, timeout); rlen = readPacket(pbuff, 1, timeout);
if (rlen != 1) return 0; if (rlen != 1) return 0;
DEBUG_PRINT(F("Packet Type:\t")); DEBUG_PRINTBUFFER(pbuff, rlen); DEBUG_PRINT(F("Packet Type:\t"));
DEBUG_PRINTBUFFER(pbuff, rlen);
pbuff++; pbuff++;
uint32_t value = 0; uint32_t value = 0;
@ -251,37 +266,49 @@ uint16_t Adafruit_MQTT::readFullPacket(uint8_t *buffer, uint16_t maxsize, uint16
intermediate *= multiplier; intermediate *= multiplier;
value += intermediate; value += intermediate;
multiplier *= 128; multiplier *= 128;
if (multiplier > (128UL*128UL*128UL)) { if (multiplier > (128UL * 128UL * 128UL)) {
DEBUG_PRINT(F("Malformed packet len\n")); DEBUG_PRINT(F("Malformed packet len\n"));
return 0; return 0;
} }
} while (encodedByte & 0x80); } while (encodedByte & 0x80);
DEBUG_PRINT(F("Packet Length:\t")); DEBUG_PRINTLN(value); DEBUG_PRINT(F("Packet Length:\t"));
DEBUG_PRINTLN(value);
if (value > (maxsize - (pbuff-buffer) - 1)) { if (value > (maxsize - (pbuff - buffer) - 1)) {
DEBUG_PRINTLN(F("Packet too big for buffer")); DEBUG_PRINTLN(F("Packet too big for buffer"));
rlen = readPacket(pbuff, (maxsize - (pbuff-buffer) - 1), timeout); rlen = readPacket(pbuff, (maxsize - (pbuff - buffer) - 1), timeout);
} else { } else {
rlen = readPacket(pbuff, value, timeout); rlen = readPacket(pbuff, value, timeout);
} }
//DEBUG_PRINT(F("Remaining packet:\t")); DEBUG_PRINTBUFFER(pbuff, rlen); //DEBUG_PRINT(F("Remaining packet:\t")); DEBUG_PRINTBUFFER(pbuff, rlen);
return ((pbuff - buffer)+rlen); return ((pbuff - buffer) + rlen);
} }
const __FlashStringHelper* Adafruit_MQTT::connectErrorString(int8_t code) { const __FlashStringHelper *Adafruit_MQTT::connectErrorString(int8_t code) {
switch (code) { switch (code) {
case 1: return F("The Server does not support the level of the MQTT protocol requested"); case 1:
case 2: return F("The Client identifier is correct UTF-8 but not allowed by the Server"); return F("The Server does not support the level of the MQTT protocol requested");
case 3: return F("The MQTT service is unavailable"); case 2:
case 4: return F("The data in the user name or password is malformed"); return F("The Client identifier is correct UTF-8 but not allowed by the Server");
case 5: return F("Not authorized to connect"); case 3:
case 6: return F("Exceeded reconnect rate limit. Please try again later."); return F("The MQTT service is unavailable");
case 7: return F("You have been banned from connecting. Please contact the MQTT server administrator for more details."); case 4:
case -1: return F("Connection failed"); return F("The data in the user name or password is malformed");
case -2: return F("Failed to subscribe"); case 5:
default: return F("Unknown error"); return F("Not authorized to connect");
case 6:
return F("Exceeded reconnect rate limit. Please try again later.");
case 7:
return F(
"You have been banned from connecting. Please contact the MQTT server administrator for more details.");
case -1:
return F("Connection failed");
case -2:
return F("Failed to subscribe");
default:
return F("Unknown error");
} }
} }
@ -289,8 +316,9 @@ bool Adafruit_MQTT::disconnect() {
// Construct and send disconnect packet. // Construct and send disconnect packet.
uint8_t len = disconnectPacket(buffer); uint8_t len = disconnectPacket(buffer);
if (! sendPacket(buffer, len)) if (!sendPacket(buffer, len)) {
DEBUG_PRINTLN(F("Unable to send disconnect packet")); DEBUG_PRINTLN(F("Unable to send disconnect packet"));
}
return disconnectServer(); return disconnectServer();
@ -298,7 +326,7 @@ bool Adafruit_MQTT::disconnect() {
bool Adafruit_MQTT::publish(const char *topic, const char *data, uint8_t qos, bool retain) { bool Adafruit_MQTT::publish(const char *topic, const char *data, uint8_t qos, bool retain) {
return publish(topic, (uint8_t*)(data), strlen(data), qos, retain); return publish(topic, (uint8_t *) (data), strlen(data), qos, retain);
} }
bool Adafruit_MQTT::publish(const char *topic, uint8_t *data, uint16_t bLen, uint8_t qos, bool retain) { bool Adafruit_MQTT::publish(const char *topic, uint8_t *data, uint16_t bLen, uint8_t qos, bool retain) {
@ -348,16 +376,17 @@ bool Adafruit_MQTT::will(const char *topic, const char *payload, uint8_t qos, ui
bool Adafruit_MQTT::subscribe(Adafruit_MQTT_Subscribe *sub) { bool Adafruit_MQTT::subscribe(Adafruit_MQTT_Subscribe *sub) {
uint8_t i; uint8_t i;
// see if we are already subscribed // see if we are already subscribed
for (i=0; i<MAXSUBSCRIPTIONS; i++) { for (i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == sub) { if (subscriptions[i] == sub) {
DEBUG_PRINTLN(F("Already subscribed")); DEBUG_PRINTLN(F("Already subscribed"));
return true; return true;
} }
} }
if (i==MAXSUBSCRIPTIONS) { // add to subscriptionlist if (i == MAXSUBSCRIPTIONS) { // add to subscriptionlist
for (i=0; i<MAXSUBSCRIPTIONS; i++) { for (i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == 0) { if (subscriptions[i] == 0) {
DEBUG_PRINT(F("Added sub ")); DEBUG_PRINTLN(i); DEBUG_PRINT(F("Added sub "));
DEBUG_PRINTLN(i);
subscriptions[i] = sub; subscriptions[i] = sub;
return true; return true;
} }
@ -372,7 +401,7 @@ bool Adafruit_MQTT::unsubscribe(Adafruit_MQTT_Subscribe *sub) {
uint8_t i; uint8_t i;
// see if we are already subscribed // see if we are already subscribed
for (i=0; i<MAXSUBSCRIPTIONS; i++) { for (i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] == sub) { if (subscriptions[i] == sub) {
@ -382,12 +411,12 @@ bool Adafruit_MQTT::unsubscribe(Adafruit_MQTT_Subscribe *sub) {
uint8_t len = unsubscribePacket(buffer, subscriptions[i]->topic); uint8_t len = unsubscribePacket(buffer, subscriptions[i]->topic);
// sending unsubscribe failed // sending unsubscribe failed
if (! sendPacket(buffer, len)) if (!sendPacket(buffer, len))
return false; return false;
// if QoS for this subscription is 1 or 2, we need // if QoS for this subscription is 1 or 2, we need
// to wait for the unsuback to confirm unsubscription // to wait for the unsuback to confirm unsubscription
if(subscriptions[i]->qos > 0 && MQTT_PROTOCOL_LEVEL > 3) { if (subscriptions[i]->qos > 0 && MQTT_PROTOCOL_LEVEL > 3) {
// wait for UNSUBACK // wait for UNSUBACK
len = readFullPacket(buffer, MAXBUFFERSIZE, CONNECT_TIMEOUT_MS); len = readFullPacket(buffer, MAXBUFFERSIZE, CONNECT_TIMEOUT_MS);
@ -414,33 +443,30 @@ void Adafruit_MQTT::processPackets(int16_t timeout) {
uint32_t elapsed = 0, endtime, starttime = millis(); uint32_t elapsed = 0, endtime, starttime = millis();
while (elapsed < (uint32_t)timeout) { while (elapsed < (uint32_t) timeout) {
Adafruit_MQTT_Subscribe *sub = readSubscription(timeout - elapsed); Adafruit_MQTT_Subscribe *sub = readSubscription(timeout - elapsed);
if (sub) { if (sub) {
//Serial.println("**** sub packet received"); //Serial.println("**** sub packet received");
if (sub->callback_uint32t != NULL) { if (sub->callback_uint32t != NULL) {
// huh lets do the callback in integer mode // huh lets do the callback in integer mode
uint32_t data = 0; uint32_t data = 0;
data = atoi((char *)sub->lastread); data = atoi((char *) sub->lastread);
//Serial.print("*** calling int callback with : "); Serial.println(data); //Serial.print("*** calling int callback with : "); Serial.println(data);
sub->callback_uint32t(data); sub->callback_uint32t(data);
} } else if (sub->callback_double != NULL) {
else if (sub->callback_double != NULL) {
// huh lets do the callback in doublefloat mode // huh lets do the callback in doublefloat mode
double data = 0; double data = 0;
data = atof((char *)sub->lastread); data = atof((char *) sub->lastread);
//Serial.print("*** calling double callback with : "); Serial.println(data); //Serial.print("*** calling double callback with : "); Serial.println(data);
sub->callback_double(data); sub->callback_double(data);
} } else if (sub->callback_buffer != NULL) {
else if (sub->callback_buffer != NULL) {
// huh lets do the callback in buffer mode // huh lets do the callback in buffer mode
//Serial.print("*** calling buffer callback with : "); Serial.println((char *)sub->lastread); //Serial.print("*** calling buffer callback with : "); Serial.println((char *)sub->lastread);
sub->callback_buffer((char *)sub->lastread, sub->datalen); sub->callback_buffer((char *) sub->lastread, sub->datalen);
} } else if (sub->callback_io != NULL) {
else if (sub->callback_io != NULL) {
// huh lets do the callback in io mode // huh lets do the callback in io mode
//Serial.print("*** calling io instance callback with : "); Serial.println((char *)sub->lastread); //Serial.print("*** calling io instance callback with : "); Serial.println((char *)sub->lastread);
((sub->io_mqtt)->*(sub->callback_io))((char *)sub->lastread, sub->datalen); ((sub->io_mqtt)->*(sub->callback_io))((char *) sub->lastread, sub->datalen);
} }
} }
@ -454,24 +480,35 @@ void Adafruit_MQTT::processPackets(int16_t timeout) {
} }
Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) { Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) {
uint16_t i, topiclen, datalen;
// Check if data is available to read. // Check if data is available to read.
uint16_t len = readFullPacket(buffer, MAXBUFFERSIZE, timeout); // return one full packet uint16_t len = readFullPacket(buffer, MAXBUFFERSIZE, timeout); // return one full packet
if (!len) return handleSubscriptionPacket(len);
}
Adafruit_MQTT_Subscribe *Adafruit_MQTT::handleSubscriptionPacket(uint16_t len) {
uint16_t i, topiclen, datalen;
if (!len) {
return NULL; // No data available, just quit. return NULL; // No data available, just quit.
DEBUG_PRINT("Packet len: "); DEBUG_PRINTLN(len); }
DEBUG_PRINT("Packet len: ");
DEBUG_PRINTLN(len);
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
if (len<3) return NULL; if (len < 3) {
if ((buffer[0] & 0xF0) != (MQTT_CTRL_PUBLISH) << 4) return NULL; return NULL;
}
if ((buffer[0] & 0xF0) != (MQTT_CTRL_PUBLISH) << 4) {
return NULL;
}
// Parse out length of packet. // Parse out length of packet.
topiclen = buffer[3]; topiclen = buffer[3];
DEBUG_PRINT(F("Looking for subscription len ")); DEBUG_PRINTLN(topiclen); DEBUG_PRINT(F("Looking for subscription len "));
DEBUG_PRINTLN(topiclen);
// Find subscription associated with this packet. // Find subscription associated with this packet.
for (i=0; i<MAXSUBSCRIPTIONS; i++) { for (i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i]) { if (subscriptions[i]) {
// Skip this subscription if its name length isn't the same as the // Skip this subscription if its name length isn't the same as the
// received topic name. // received topic name.
@ -479,22 +516,25 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) {
continue; continue;
// Stop if the subscription topic matches the received topic. Be careful // Stop if the subscription topic matches the received topic. Be careful
// to make comparison case insensitive. // to make comparison case insensitive.
if (strncasecmp((char*)buffer+4, subscriptions[i]->topic, topiclen) == 0) { if (strncasecmp((char *) buffer + 4, subscriptions[i]->topic, topiclen) == 0) {
DEBUG_PRINT(F("Found sub #")); DEBUG_PRINTLN(i); DEBUG_PRINT(F("Found sub #"));
DEBUG_PRINTLN(i);
break; break;
} }
} }
} }
if (i==MAXSUBSCRIPTIONS) return NULL; // matching sub not found ??? if (i == MAXSUBSCRIPTIONS) {
return NULL; // matching sub not found ???
}
uint8_t packet_id_len = 0; uint8_t packet_id_len = 0;
uint16_t packetid = 0; uint16_t packetid = 0;
// Check if it is QoS 1, TODO: we dont support QoS 2 // Check if it is QoS 1, TODO: we dont support QoS 2
if ((buffer[0] & 0x6) == 0x2) { if ((buffer[0] & 0x6) == 0x2) {
packet_id_len = 2; packet_id_len = 2;
packetid = buffer[topiclen+4]; packetid = buffer[topiclen + 4];
packetid <<= 8; packetid <<= 8;
packetid |= buffer[topiclen+5]; packetid |= buffer[topiclen + 5];
} }
// zero out the old data // zero out the old data
@ -502,21 +542,22 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) {
datalen = len - topiclen - packet_id_len - 4; datalen = len - topiclen - packet_id_len - 4;
if (datalen > SUBSCRIPTIONDATALEN) { if (datalen > SUBSCRIPTIONDATALEN) {
datalen = SUBSCRIPTIONDATALEN-1; // cut it off datalen = SUBSCRIPTIONDATALEN - 1; // cut it off
} }
// extract out just the data, into the subscription object itself // extract out just the data, into the subscription object itself
memmove(subscriptions[i]->lastread, buffer+4+topiclen+packet_id_len, datalen); memmove(subscriptions[i]->lastread, buffer + 4 + topiclen + packet_id_len, datalen);
subscriptions[i]->datalen = datalen; subscriptions[i]->datalen = datalen;
DEBUG_PRINT(F("Data len: ")); DEBUG_PRINTLN(datalen); DEBUG_PRINT(F("Data len: "));
DEBUG_PRINT(F("Data: ")); DEBUG_PRINTLN((char *)subscriptions[i]->lastread); DEBUG_PRINTLN(datalen);
DEBUG_PRINT(F("Data: "));
DEBUG_PRINTLN((char *) subscriptions[i]->lastread);
if ((MQTT_PROTOCOL_LEVEL > 3) &&(buffer[0] & 0x6) == 0x2) { if ((MQTT_PROTOCOL_LEVEL > 3) && (buffer[0] & 0x6) == 0x2) {
uint8_t ackpacket[4]; uint8_t ackpacket[4];
// Construct and send puback packet. // Construct and send puback packet.
uint8_t len = pubackPacket(ackpacket, packetid); uint8_t len = pubackPacket(ackpacket, packetid);
if (!sendPacket(ackpacket, len)) if (!sendPacket(ackpacket, len)) DEBUG_PRINT(F("Failed"));
DEBUG_PRINT(F("Failed"));
} }
// return the valid matching subscription // return the valid matching subscription
@ -535,14 +576,16 @@ bool Adafruit_MQTT::ping(uint8_t num) {
while (num--) { while (num--) {
// Construct and send ping packet. // Construct and send ping packet.
uint8_t len = pingPacket(buffer); uint8_t len = pingPacket(buffer);
if (!sendPacket(buffer, len)) if (!sendPacket(buffer, len)) {
continue; continue;
}
// Process ping reply. // Process ping reply.
len = processPacketsUntil(buffer, MQTT_CTRL_PINGRESP, PING_TIMEOUT_MS); processPacketsUntil(buffer, MQTT_CTRL_PINGRESP, PING_TIMEOUT_MS);
if (buffer[0] == (MQTT_CTRL_PINGRESP << 4)) if (buffer[0] == (MQTT_CTRL_PINGRESP << 4)) {
return true; return true;
} }
}
return false; return false;
} }
@ -560,7 +603,7 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
// fixed header, connection messsage no flags // fixed header, connection messsage no flags
p[0] = (MQTT_CTRL_CONNECT << 4) | 0x0; p[0] = (MQTT_CTRL_CONNECT << 4) | 0x0;
p+=2; p += 2;
// fill in packet[1] last // fill in packet[1] last
#if MQTT_PROTOCOL_LEVEL == 3 #if MQTT_PROTOCOL_LEVEL == 3
@ -568,7 +611,7 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
#elif MQTT_PROTOCOL_LEVEL == 4 #elif MQTT_PROTOCOL_LEVEL == 4
p = stringprint(p, "MQTT"); p = stringprint(p, "MQTT");
#else #else
#error "MQTT level not supported" #error "MQTT level not supported"
#endif #endif
p[0] = MQTT_PROTOCOL_LEVEL; p[0] = MQTT_PROTOCOL_LEVEL;
@ -582,12 +625,12 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
p[0] |= MQTT_CONN_WILLFLAG; p[0] |= MQTT_CONN_WILLFLAG;
if(will_qos == 1) if (will_qos == 1)
p[0] |= MQTT_CONN_WILLQOS_1; p[0] |= MQTT_CONN_WILLQOS_1;
else if(will_qos == 2) else if (will_qos == 2)
p[0] |= MQTT_CONN_WILLQOS_2; p[0] |= MQTT_CONN_WILLQOS_2;
if(will_retain == 1) if (will_retain == 1)
p[0] |= MQTT_CONN_WILLRETAIN; p[0] |= MQTT_CONN_WILLRETAIN;
} }
@ -603,7 +646,7 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
p[0] = MQTT_CONN_KEEPALIVE & 0xFF; p[0] = MQTT_CONN_KEEPALIVE & 0xFF;
p++; p++;
if(MQTT_PROTOCOL_LEVEL == 3) { if (MQTT_PROTOCOL_LEVEL == 3) {
p = stringprint(p, clientid, 23); // Limit client ID to first 23 characters. p = stringprint(p, clientid, 23); // Limit client ID to first 23 characters.
} else { } else {
if (pgm_read_byte(clientid) != 0) { if (pgm_read_byte(clientid) != 0) {
@ -631,7 +674,7 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
len = p - packet; len = p - packet;
packet[1] = len-2; // don't include the 2 bytes of fixed header data packet[1] = len - 2; // don't include the 2 bytes of fixed header data
DEBUG_PRINTLN(F("MQTT connect packet:")); DEBUG_PRINTLN(F("MQTT connect packet:"));
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
return len; return len;
@ -642,12 +685,12 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
uint16_t Adafruit_MQTT::publishPacket(uint8_t *packet, const char *topic, uint16_t Adafruit_MQTT::publishPacket(uint8_t *packet, const char *topic,
uint8_t *data, uint16_t bLen, uint8_t qos, bool retain) { uint8_t *data, uint16_t bLen, uint8_t qos, bool retain) {
uint8_t *p = packet; uint8_t *p = packet;
uint16_t len=0; uint16_t len = 0;
// calc length of non-header data // calc length of non-header data
len += 2; // two bytes to set the topic size len += 2; // two bytes to set the topic size
len += strlen(topic); // topic length len += strlen(topic); // topic length
if(qos > 0) { if (qos > 0) {
len += 2; // qos packet id len += 2; // qos packet id
} }
len += bLen; // payload length len += bLen; // payload length
@ -661,47 +704,46 @@ uint16_t Adafruit_MQTT::publishPacket(uint8_t *packet, const char *topic,
uint8_t encodedByte = len % 128; uint8_t encodedByte = len % 128;
len /= 128; len /= 128;
// if there are more data to encode, set the top bit of this byte // if there are more data to encode, set the top bit of this byte
if ( len > 0 ) { if (len > 0) {
encodedByte |= 0x80; encodedByte |= 0x80;
} }
p[0] = encodedByte; p[0] = encodedByte;
p++; p++;
} while ( len > 0 ); } while (len > 0);
// topic comes before packet identifier // topic comes before packet identifier
p = stringprint(p, topic); p = stringprint(p, topic);
// add packet identifier. used for checking PUBACK in QOS > 0 // add packet identifier. used for checking PUBACK in QOS > 0
if(qos > 0) { if (qos > 0) {
p[0] = (packet_id_counter >> 8) & 0xFF; p[0] = (packet_id_counter >> 8) & 0xFF;
p[1] = packet_id_counter & 0xFF; p[1] = packet_id_counter & 0xFF;
p+=2; p += 2;
// increment the packet id // increment the packet id
packet_id_counter++; packet_id_counter++;
} }
memmove(p, data, bLen); memmove(p, data, bLen);
p+= bLen; p += bLen;
len = p - packet; len = p - packet;
DEBUG_PRINTLN(F("MQTT publish packet:")); DEBUG_PRINTLN(F("MQTT publish packet:"));
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
return len; return len;
} }
uint8_t Adafruit_MQTT::subscribePacket(uint8_t *packet, const char *topic, uint8_t Adafruit_MQTT::subscribePacket(uint8_t *packet, const char *topic, uint8_t qos) {
uint8_t qos) {
uint8_t *p = packet; uint8_t *p = packet;
uint16_t len; uint16_t len;
p[0] = MQTT_CTRL_SUBSCRIBE << 4 | MQTT_QOS_1 << 1; p[0] = MQTT_CTRL_SUBSCRIBE << 4 | MQTT_QOS_1 << 1;
// fill in packet[1] last // fill in packet[1] last
p+=2; p += 2;
// packet identifier. used for checking SUBACK // packet identifier. used for checking SUBACK
p[0] = (packet_id_counter >> 8) & 0xFF; p[0] = (packet_id_counter >> 8) & 0xFF;
p[1] = packet_id_counter & 0xFF; p[1] = packet_id_counter & 0xFF;
p+=2; p += 2;
// increment the packet id // increment the packet id
packet_id_counter++; packet_id_counter++;
@ -712,14 +754,13 @@ uint8_t Adafruit_MQTT::subscribePacket(uint8_t *packet, const char *topic,
p++; p++;
len = p - packet; len = p - packet;
packet[1] = len-2; // don't include the 2 bytes of fixed header data packet[1] = len - 2; // don't include the 2 bytes of fixed header data
DEBUG_PRINTLN(F("MQTT subscription packet:")); DEBUG_PRINTLN(F("MQTT subscription packet:"));
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
return len; return len;
} }
uint8_t Adafruit_MQTT::unsubscribePacket(uint8_t *packet, const char *topic) { uint8_t Adafruit_MQTT::unsubscribePacket(uint8_t *packet, const char *topic) {
uint8_t *p = packet; uint8_t *p = packet;
@ -727,12 +768,12 @@ uint8_t Adafruit_MQTT::unsubscribePacket(uint8_t *packet, const char *topic) {
p[0] = MQTT_CTRL_UNSUBSCRIBE << 4 | 0x1; p[0] = MQTT_CTRL_UNSUBSCRIBE << 4 | 0x1;
// fill in packet[1] last // fill in packet[1] last
p+=2; p += 2;
// packet identifier. used for checking UNSUBACK // packet identifier. used for checking UNSUBACK
p[0] = (packet_id_counter >> 8) & 0xFF; p[0] = (packet_id_counter >> 8) & 0xFF;
p[1] = packet_id_counter & 0xFF; p[1] = packet_id_counter & 0xFF;
p+=2; p += 2;
// increment the packet id // increment the packet id
packet_id_counter++; packet_id_counter++;
@ -740,7 +781,7 @@ uint8_t Adafruit_MQTT::unsubscribePacket(uint8_t *packet, const char *topic) {
p = stringprint(p, topic); p = stringprint(p, topic);
len = p - packet; len = p - packet;
packet[1] = len-2; // don't include the 2 bytes of fixed header data packet[1] = len - 2; // don't include the 2 bytes of fixed header data
DEBUG_PRINTLN(F("MQTT unsubscription packet:")); DEBUG_PRINTLN(F("MQTT unsubscription packet:"));
DEBUG_PRINTBUFFER(buffer, len); DEBUG_PRINTBUFFER(buffer, len);
return len; return len;
@ -776,11 +817,13 @@ uint8_t Adafruit_MQTT::disconnectPacket(uint8_t *packet) {
// Adafruit_MQTT_Publish Definition //////////////////////////////////////////// // Adafruit_MQTT_Publish Definition ////////////////////////////////////////////
Adafruit_MQTT_Publish::Adafruit_MQTT_Publish(Adafruit_MQTT *mqttserver, Adafruit_MQTT_Publish::Adafruit_MQTT_Publish(Adafruit_MQTT *mqttserver,
const char *feed, uint8_t q) { const char *feed, uint8_t
q) {
mqtt = mqttserver; mqtt = mqttserver;
topic = feed; topic = feed;
qos = q; qos = q;
} }
bool Adafruit_MQTT_Publish::publish(int32_t i) { bool Adafruit_MQTT_Publish::publish(int32_t i) {
char payload[12]; char payload[12];
ltoa(i, payload, 10); ltoa(i, payload, 10);
@ -804,7 +847,8 @@ bool Adafruit_MQTT_Publish::publish(const char *payload) {
} }
//publish buffer of arbitrary length //publish buffer of arbitrary length
bool Adafruit_MQTT_Publish::publish(uint8_t *payload, uint16_t bLen) { bool Adafruit_MQTT_Publish::publish(uint8_t *payload, uint16_t
bLen) {
return mqtt->publish(topic, payload, bLen, qos); return mqtt->publish(topic, payload, bLen, qos);
} }
@ -813,7 +857,8 @@ bool Adafruit_MQTT_Publish::publish(uint8_t *payload, uint16_t bLen) {
// Adafruit_MQTT_Subscribe Definition ////////////////////////////////////////// // Adafruit_MQTT_Subscribe Definition //////////////////////////////////////////
Adafruit_MQTT_Subscribe::Adafruit_MQTT_Subscribe(Adafruit_MQTT *mqttserver, Adafruit_MQTT_Subscribe::Adafruit_MQTT_Subscribe(Adafruit_MQTT *mqttserver,
const char *feed, uint8_t q) { const char *feed, uint8_t
q) {
mqtt = mqttserver; mqtt = mqttserver;
topic = feed; topic = feed;
qos = q; qos = q;
@ -839,7 +884,7 @@ void Adafruit_MQTT_Subscribe::setCallback(SubscribeCallbackBufferType cb) {
void Adafruit_MQTT_Subscribe::setCallback(AdafruitIO_MQTT *io, SubscribeCallbackIOType cb) { void Adafruit_MQTT_Subscribe::setCallback(AdafruitIO_MQTT *io, SubscribeCallbackIOType cb) {
callback_io = cb; callback_io = cb;
io_mqtt= io; io_mqtt = io;
} }
void Adafruit_MQTT_Subscribe::removeCallback(void) { void Adafruit_MQTT_Subscribe::removeCallback(void) {

3
Adafruit_MQTT.h
View File

@ -195,6 +195,9 @@ class Adafruit_MQTT {
// that subscribe should be called first for each topic that receives messages! // that subscribe should be called first for each topic that receives messages!
Adafruit_MQTT_Subscribe *readSubscription(int16_t timeout=0); Adafruit_MQTT_Subscribe *readSubscription(int16_t timeout=0);
// Handle any data coming in for subscriptions and fires them off to the appropriate callback
Adafruit_MQTT_Subscribe *handleSubscriptionPacket(uint16_t len);
void processPackets(int16_t timeout); void processPackets(int16_t timeout);
// Ping the server to ensure the connection is still alive. // Ping the server to ensure the connection is still alive.