Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support publishing and receiving large messages. #193

Merged
merged 4 commits into from
May 14, 2021
Merged

Support publishing and receiving large messages. #193

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2ea8a0b
Support publishing and receiving large messages. Send multiple 250 by…
May 13, 2021
93a35d1
fix typo
May 13, 2021
c782cc4
Merge branch 'master' of https://github.com/adafruit/Adafruit_MQTT_Li…
May 14, 2021
db003d3
Addressing comments from PR #193 with merge from PR#166. Clang-format…
May 14, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 29 additions & 25 deletions Adafruit_MQTT.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I changed the signature here to explicitly take a uint32_t so that the comparisons are consistent. This doesn't change the fact that the max packet length is limited to 65536, but that is checked in readFullPacket

/* 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,
Expand Down Expand Up @@ -233,7 +249,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;
Copy link
Author

@xdylanm xdylanm May 13, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Packets larger than 255 bytes would overflow rlen.

Copy link
Member

@brentru brentru May 13, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This does solve an issue I was having today with a very large binary payload, thank you.

Correctly identify topic start position for >127byte messages (may also be addressed in PR166 -- happy to merge that fix in if preferred).

@flavio-fernandes Is this implemented in #166?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hi @brentru . No, this fix is related but not exactly what I was trying to address in #166 . The #166 was done to avoid corrupting memory should we attempt to
publish more data than what is available by MAXBUFFERSIZE. I think the function introduced in this PR (topicOffsetFromLenth) is what #166 introduces, so we definitely should combine that. @xdylanm how about you incorporate that into this pr? You can take the credit; I do not care for that. ;) Basically, we need this logic in addition to what you are doing here: d6f07ed#diff-a20445c19076bfb61fa2cf57b03325851600fc2dc25391f7e97f2de072fda081R701

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@flavio-fernandes in my own build, I originally had a simpler version of the length test
uint16_t const topicoffset = (len > 127 ? 1 : 0)
but I liked your function approach better, so I followed that pattern and credit to you =). If #166 is close to being merged, I'm happy to just rebase off of that when it's merged and update my PR. Per your note, you're trying to solve a different problem, which should certainly get resolved.

That said, I realized that this solution does not support packets > 65535 bytes since the lengths are all stored in uint16_t, and I'm still not sure what the right solution is there.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Understood. Yes, you correctly point out the fact that we cannot go beyond 65535 since we are using uint16.

I will defer to @brentru on how we want to proceed here, but my opinion is to go ahead with PR/166 as is,
and then have you leverage the function packetAdditionalLen in PR/193 and potentially improve on the uint16 limitation. Sounds reasonable, @brentru ?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This sounds OK and I agree. - @xdylanm, see the comment below about using clang-format so we can approve this PR.


// read the packet type:
rlen = readPacket(pbuff, 1, timeout);
Expand Down Expand Up @@ -267,7 +283,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)) {
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

value is uint32_t so it can handle the actual packet length. I made the cast explicit here in the conditional, but the key thing is that it's checked and for packets larger than 65536 (maxsize is uint16_t), you can't read past that size, and there is a debug message there.

DEBUG_PRINTLN(F("Packet too big for buffer"));
rlen = readPacket(pbuff, (maxsize - (pbuff - buffer) - 1), timeout);
} else {
Expand Down Expand Up @@ -492,7 +509,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);

Expand All @@ -505,8 +524,8 @@ 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);
break;
Expand All @@ -521,21 +540,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);
Expand Down Expand Up @@ -669,21 +688,6 @@ uint8_t Adafruit_MQTT::connectPacket(uint8_t *packet) {
return 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
static uint16_t packetAdditionalLen(uint16_t currLen) {
/* Increase length field based on current length */
if (currLen < 128)
return 0;
if (currLen < 16384)
return 1;
if (currLen < 2097151)
return 2;
return 3;
}

// 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,
Expand Down
5 changes: 3 additions & 2 deletions Adafruit_MQTT_Client.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,17 +83,18 @@ uint16_t Adafruit_MQTT_Client::readPacket(uint8_t *buffer, uint16_t maxlen,

bool Adafruit_MQTT_Client::sendPacket(uint8_t *buffer, uint16_t len) {
uint16_t ret = 0;

uint16_t offset = 0;
while (len > 0) {
if (client->connected()) {
// send 250 bytes at most at a time, can adjust this later based on Client

uint16_t sendlen = len > 250 ? 250 : len;
// Serial.print("Sending: "); Serial.println(sendlen);
ret = client->write(buffer, sendlen);
ret = client->write(buffer + offset, sendlen);
DEBUG_PRINT(F("Client sendPacket returned: "));
DEBUG_PRINTLN(ret);
len -= ret;
offset += ret;

if (ret != sendlen) {
DEBUG_PRINTLN("Failed to send packet.");
Expand Down
2 changes: 1 addition & 1 deletion library.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
name=Adafruit MQTT Library
version=2.2.0
version=2.3.0
author=Adafruit
maintainer=Adafruit <[email protected]>
sentence=MQTT library that supports the FONA, ESP8266, Yun, and generic Arduino Client hardware.
Expand Down