arduino/Adafruit_MQTT_Library
1
0
Fork 0
mirror of https://github.com/adafruit/Adafruit_MQTT_Library.git synced 2025-07-21 18:22:06 +03:00
Code Activity

Merge in KEEPALIVE check.

Browse Source
This commit is contained in:
dlizotte-uwo
2023-07-05 17:39:49 -04:00
parent 7b2061ad85 0c57bf6c1d
commit 847f561215
14 changed files with 501 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

220
Adafruit_MQTT.cpp
View File

@ -96,6 +96,22 @@ static uint8_t *stringprint(uint8_t *p, const char *s, uint16_t maxlen = 0) {
return p + len;
}
// packetAdditionalLen is a helper function used to figure out
// how bigger the payload needs to be in order to account for
// its variable length field. As per
// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Table_2.4_Size
// See also readFullPacket
static uint16_t packetAdditionalLen(uint32_t currLen) {
/* Increase length field based on current length */
if (currLen < 128) // 7-bits
return 0;
if (currLen < 16384) // 14-bits
return 1;
if (currLen < 2097152) // 21-bits
return 2;
return 3;
}
// Adafruit_MQTT Definition ////////////////////////////////////////////////////
Adafruit_MQTT::Adafruit_MQTT(const char *server, uint16_t port, const char *cid,
@ -117,6 +133,7 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server, uint16_t port, const char *cid,
will_retain = 0;
packet_id_counter = 1; // MQTT spec forbids packet id of 0 if QOS=1
keepAliveInterval = MQTT_CONN_KEEPALIVE;
}
Adafruit_MQTT::Adafruit_MQTT(const char *server, uint16_t port,
@ -138,6 +155,8 @@ Adafruit_MQTT::Adafruit_MQTT(const char *server, uint16_t port,
will_retain = 0;
packet_id_counter = 1; // MQTT spec forbids packet id of 0 if QOS=1
keepAliveInterval = MQTT_CONN_KEEPALIVE;
}
int8_t Adafruit_MQTT::connect() {
@ -202,6 +221,32 @@ int8_t Adafruit_MQTT::connect(const char *user, const char *pass) {
return connect();
}
void Adafruit_MQTT::processSubscriptionPacket(Adafruit_MQTT_Subscribe *sub) {
if (sub->callback_uint32t != NULL) {
// execute callback in integer mode
uint32_t data = 0;
data = atoi((char *)sub->lastread);
sub->callback_uint32t(data);
} else if (sub->callback_double != NULL) {
// execute callback in doublefloat mode
double data = 0;
data = atof((char *)sub->lastread);
sub->callback_double(data);
} else if (sub->callback_buffer != NULL) {
// execute callback in buffer mode
sub->callback_buffer((char *)sub->lastread, sub->datalen);
} else if (sub->callback_io != NULL) {
// execute callback in io mode
((sub->io_mqtt)->*(sub->callback_io))((char *)sub->lastread, sub->datalen);
} else {
DEBUG_PRINTLN(
"ERROR: Subscription packet did not have an associated callback");
return;
}
// mark subscription message as "read""
sub->new_message = false;
}
uint16_t Adafruit_MQTT::processPacketsUntil(uint8_t *buffer,
uint8_t waitforpackettype,
uint16_t timeout) {
@ -219,7 +264,9 @@ uint16_t Adafruit_MQTT::processPacketsUntil(uint8_t *buffer,
return len;
} else {
if (packetType == MQTT_CTRL_PUBLISH) {
handleSubscriptionPacket(len);
Adafruit_MQTT_Subscribe *sub = handleSubscriptionPacket(len);
if (sub)
processSubscriptionPacket(sub);
} else {
ERROR_PRINTLN(F("Dropped a packet"));
}
@ -233,7 +280,7 @@ uint16_t Adafruit_MQTT::readFullPacket(uint8_t *buffer, uint16_t maxsize,
// will read a packet and Do The Right Thing with length
uint8_t *pbuff = buffer;
uint8_t rlen;
uint16_t rlen;
// read the packet type:
rlen = readPacket(pbuff, 1, timeout);
@ -267,7 +314,8 @@ uint16_t Adafruit_MQTT::readFullPacket(uint8_t *buffer, uint16_t maxsize,
DEBUG_PRINT(F("Packet Length:\t"));
DEBUG_PRINTLN(value);
if (value > (maxsize - (pbuff - buffer) - 1)) {
// maxsize is limited to 65536 by 16-bit unsigned
if (value > uint32_t(maxsize - (pbuff - buffer) - 1)) {
DEBUG_PRINTLN(F("Packet too big for buffer"));
rlen = readPacket(pbuff, (maxsize - (pbuff - buffer) - 1), timeout);
} else {
@ -316,26 +364,29 @@ bool Adafruit_MQTT::disconnect() {
return disconnectServer();
}
bool Adafruit_MQTT::publish(const char *topic, const char *data, uint8_t qos) {
return publish(topic, (uint8_t *)(data), strlen(data), qos);
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);
}
bool Adafruit_MQTT::publish(const char *topic, uint8_t *data, uint16_t bLen,
uint8_t qos) {
uint8_t qos, bool retain) {
// Construct and send publish packet.
uint16_t len = publishPacket(buffer, topic, data, bLen, qos);
uint16_t len = publishPacket(buffer, topic, data, bLen, qos,
(uint16_t)sizeof(buffer), retain);
if (!sendPacket(buffer, len))
return false;
// If QOS level is high enough verify the response packet.
if (qos > 0) {
len = readFullPacket(buffer, MAXBUFFERSIZE, PUBLISH_TIMEOUT_MS);
len = processPacketsUntil(buffer, MQTT_CTRL_PUBACK, PUBLISH_TIMEOUT_MS);
DEBUG_PRINT(F("Publish QOS1+ reply:\t"));
DEBUG_PRINTBUFFER(buffer, len);
if (len != 4)
return false;
if ((buffer[0] >> 4) != MQTT_CTRL_PUBACK)
return false;
uint16_t packnum = buffer[2];
packnum <<= 8;
packnum |= buffer[3];
@ -366,6 +417,25 @@ bool Adafruit_MQTT::will(const char *topic, const char *payload, uint8_t qos,
return true;
}
/***************************************************************************/
/*!
@brief Sets the connect packet's KeepAlive Interval, in seconds. This
function MUST be called prior to connect().
@param keepAlive
Maximum amount of time without communication between the
client and the MQTT broker, in seconds.
@returns True if called prior to connect(), False otherwise.
*/
/***************************************************************************/
bool Adafruit_MQTT::setKeepAliveInterval(uint16_t keepAlive) {
if (connected()) {
DEBUG_PRINT(F("keepAlive defined after connection established."));
return false;
}
keepAliveInterval = keepAlive;
return true;
}
bool Adafruit_MQTT::subscribe(Adafruit_MQTT_Subscribe *sub) {
uint8_t i;
// see if we are already subscribed
@ -437,27 +507,8 @@ void Adafruit_MQTT::processPackets(int16_t timeout) {
while (elapsed < (uint32_t)timeout) {
Adafruit_MQTT_Subscribe *sub = readSubscription(timeout - elapsed);
if (sub) {
if (sub->callback_uint32t != NULL) {
// huh lets do the callback in integer mode
uint32_t data = 0;
data = atoi((char *)sub->lastread);
sub->callback_uint32t(data);
} else if (sub->callback_double != NULL) {
// huh lets do the callback in doublefloat mode
double data = 0;
data = atof((char *)sub->lastread);
sub->callback_double(data);
} else if (sub->callback_buffer != NULL) {
// huh lets do the callback in buffer mode
sub->callback_buffer((char *)sub->lastread, sub->datalen);
} else if (sub->callback_io != NULL) {
// huh lets do the callback in io mode
((sub->io_mqtt)->*(sub->callback_io))((char *)sub->lastread,
sub->datalen);
}
}
if (sub)
processSubscriptionPacket(sub);
// keep track over elapsed time
endtime = millis();
if (endtime < starttime) {
@ -467,10 +518,32 @@ void Adafruit_MQTT::processPackets(int16_t timeout) {
}
}
Adafruit_MQTT_Subscribe *Adafruit_MQTT::readSubscription(int16_t timeout) {
// Check if data is available to read.
uint16_t len =
readFullPacket(buffer, MAXBUFFERSIZE, timeout); // return one full packet
return handleSubscriptionPacket(len);
// Sync or Async subscriber with message
Adafruit_MQTT_Subscribe *s = 0;
// Check if are unread messages
for (uint8_t i = 0; i < MAXSUBSCRIPTIONS; i++) {
if (subscriptions[i] && subscriptions[i]->new_message) {
s = subscriptions[i];
break;
}
}
// not unread message
if (!s) {
// Check if data is available to read.
uint16_t len = readFullPacket(buffer, MAXBUFFERSIZE,
timeout); // return one full packet
s = handleSubscriptionPacket(len);
}
// it there is a message, mark it as not pending
if (s) {
s->new_message = false;
}
return s;
}
Adafruit_MQTT_Subscribe *Adafruit_MQTT::handleSubscriptionPacket(uint16_t len) {
@ -491,7 +564,9 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::handleSubscriptionPacket(uint16_t len) {
}
// Parse out length of packet.
topiclen = buffer[3];
uint16_t const topicoffset = packetAdditionalLen(len);
uint16_t const topicstart = topicoffset + 4;
topiclen = buffer[3 + topicoffset];
DEBUG_PRINT(F("Looking for subscription len "));
DEBUG_PRINTLN(topiclen);
@ -504,10 +579,16 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::handleSubscriptionPacket(uint16_t len) {
continue;
// Stop if the subscription topic matches the received topic. Be careful
// to make comparison case insensitive.
if (strncasecmp((char *)buffer + 4, subscriptions[i]->topic, topiclen) ==
0) {
if (strncasecmp((char *)buffer + topicstart, subscriptions[i]->topic,
topiclen) == 0) {
DEBUG_PRINT(F("Found sub #"));
DEBUG_PRINTLN(i);
if (subscriptions[i]->new_message) {
DEBUG_PRINTLN(F("Lost previous message"));
} else {
subscriptions[i]->new_message = true;
}
break;
}
}
@ -520,21 +601,21 @@ Adafruit_MQTT_Subscribe *Adafruit_MQTT::handleSubscriptionPacket(uint16_t len) {
// Check if it is QoS 1, TODO: we dont support QoS 2
if ((buffer[0] & 0x6) == 0x2) {
packet_id_len = 2;
packetid = buffer[topiclen + 4];
packetid = buffer[topiclen + topicstart];
packetid <<= 8;
packetid |= buffer[topiclen + 5];
packetid |= buffer[topiclen + topicstart + 1];
}
// zero out the old data
memset(subscriptions[i]->lastread, 0, SUBSCRIPTIONDATALEN);
datalen = len - topiclen - packet_id_len - 4;
datalen = len - topiclen - packet_id_len - topicstart;
if (datalen > SUBSCRIPTIONDATALEN) {
datalen = SUBSCRIPTIONDATALEN - 1; // cut it off
}
// 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 + topicstart + topiclen + packet_id_len, datalen);
subscriptions[i]->datalen = datalen;
DEBUG_PRINT(F("Data len: "));
DEBUG_PRINTLN(datalen);
@ -629,9 +710,9 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
p[0] |= MQTT_CONN_PASSWORDFLAG;
p++;
p[0] = MQTT_CONN_KEEPALIVE >> 8;
p[0] = keepAliveInterval >> 8;
p++;
p[0] = MQTT_CONN_KEEPALIVE & 0xFF;
p[0] = keepAliveInterval & 0xFF;
p++;
if (MQTT_PROTOCOL_LEVEL == 3) {
@ -671,8 +752,8 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
// as per
// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718040
uint16_t Adafruit_MQTT::publishPacket(uint8_t *packet, const char *topic,
uint8_t *data, uint16_t bLen,
uint8_t qos) {
uint8_t *data, uint16_t bLen, uint8_t qos,
uint16_t maxPacketLen, bool retain) {
uint8_t *p = packet;
uint16_t len = 0;
@ -682,10 +763,27 @@ uint16_t Adafruit_MQTT::publishPacket(uint8_t *packet, const char *topic,
if (qos > 0) {
len += 2; // qos packet id
}
len += bLen; // payload length
// Calculate additional bytes for length field (if any)
uint16_t additionalLen = packetAdditionalLen(len + bLen);
// Payload remaining length. When maxPacketLen provided is 0, let's
// assume buffer is big enough. Fingers crossed.
// 2 + additionalLen: header byte + remaining length field (from 1 to 4 bytes)
// len = topic size field + value (string)
// bLen = buffer size
if (!(maxPacketLen == 0 ||
(len + bLen + 2 + additionalLen <= maxPacketLen))) {
// If we make it here, we got a pickle: the payload is not going
// to fit in the packet buffer. Instead of corrupting memory, let's
// do something less damaging by reducing the bLen to what we are
// able to accomodate. Alternatively, consider using a bigger
// maxPacketLen.
bLen = maxPacketLen - (len + 2 + packetAdditionalLen(maxPacketLen));
}
len += bLen; // remaining len excludes header byte & length field
// Now you can start generating the packet!
p[0] = MQTT_CTRL_PUBLISH << 4 | qos << 1;
p[0] = MQTT_CTRL_PUBLISH << 4 | qos << 1 | (retain ? 1 : 0);
p++;
// fill in packet[1] last
@ -810,33 +908,34 @@ Adafruit_MQTT_Publish::Adafruit_MQTT_Publish(Adafruit_MQTT *mqttserver,
topic = feed;
qos = q;
}
bool Adafruit_MQTT_Publish::publish(int32_t i) {
bool Adafruit_MQTT_Publish::publish(int32_t i, bool retain) {
char payload[12];
ltoa(i, payload, 10);
return mqtt->publish(topic, payload, qos);
return mqtt->publish(topic, payload, qos, retain);
}
bool Adafruit_MQTT_Publish::publish(uint32_t i) {
bool Adafruit_MQTT_Publish::publish(uint32_t i, bool retain) {
char payload[11];
ultoa(i, payload, 10);
return mqtt->publish(topic, payload, qos);
return mqtt->publish(topic, payload, qos, retain);
}
bool Adafruit_MQTT_Publish::publish(double f, uint8_t precision) {
bool Adafruit_MQTT_Publish::publish(double f, uint8_t precision, bool retain) {
char payload[41]; // Need to technically hold float max, 39 digits and minus
// sign.
dtostrf(f, 0, precision, payload);
return mqtt->publish(topic, payload, qos);
return mqtt->publish(topic, payload, qos, retain);
}
bool Adafruit_MQTT_Publish::publish(const char *payload) {
return mqtt->publish(topic, payload, qos);
bool Adafruit_MQTT_Publish::publish(const char *payload, bool retain) {
return mqtt->publish(topic, payload, qos, retain);
}
// publish buffer of arbitrary length
bool Adafruit_MQTT_Publish::publish(uint8_t *payload, uint16_t bLen) {
return mqtt->publish(topic, payload, bLen, qos);
bool Adafruit_MQTT_Publish::publish(uint8_t *payload, uint16_t bLen,
bool retain) {
return mqtt->publish(topic, payload, bLen, qos, retain);
}
// Adafruit_MQTT_Subscribe Definition //////////////////////////////////////////
@ -852,6 +951,7 @@ Adafruit_MQTT_Subscribe::Adafruit_MQTT_Subscribe(Adafruit_MQTT *mqttserver,
callback_double = 0;
callback_io = 0;
io_mqtt = 0;
new_message = false;
}
void Adafruit_MQTT_Subscribe::setCallback(SubscribeCallbackUInt32Type cb) {