wn/pubsubclient
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Allow streaming of large messages

Browse Source 
These changes are required to allow for the transmission of large messages through a connected stream. The changes do not have an impact on the class interface and habitual behavior. In particular, it will enable the use of OTA through a stream hooked through the setStream() class method. I've designed such a stream to demonstrate the functionality: https://github.com/turgu1/mqtt_ota_example.git

Guy
This commit is contained in:
Guy Turcotte 2019-01-28 18:02:37 -05:00 committed by GitHub
parent 26ce89fa47
commit b6239823fb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/PubSubClient.cpp
View File

@ -197,7 +197,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
} }
} }
uint8_t llen; uint8_t llen;
uint16_t len = readPacket(&llen); uint32_t len = readPacket(&llen);
if (len == 4) { if (len == 4) {
if (buffer[3] == 0) { if (buffer[3] == 0) {
@ -243,12 +243,12 @@ boolean PubSubClient::readByte(uint8_t * result, uint16_t * index){
return false; return false;
} }
uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { uint32_t PubSubClient::readPacket(uint8_t* lengthLength) {
uint16_t len = 0; uint16_t len = 0;
if(!readByte(buffer, &len)) return 0; if(!readByte(buffer, &len)) return 0;
bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH; bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH;
uint32_t multiplier = 1; uint32_t multiplier = 1;
uint16_t length = 0; uint32_t length = 0;
uint8_t digit = 0; uint8_t digit = 0;
uint16_t skip = 0; uint16_t skip = 0;
uint8_t start = 0; uint8_t start = 0;
@ -279,20 +279,22 @@ uint16_t PubSubClient::readPacket(uint8_t* lengthLength) {
} }
} }
for (uint16_t i = start;i<length;i++) { uint32_t idx = len;
for (uint32_t i = start;i<length;i++) {
if(!readByte(&digit)) return 0; if(!readByte(&digit)) return 0;
if (this->stream) { if (this->stream) {
if (isPublish && len-*lengthLength-2>skip) { if (isPublish && idx-*lengthLength-2>skip) {
this->stream->write(digit); this->stream->write(digit);
} }
} }
if (len < MQTT_MAX_PACKET_SIZE) { if (len < MQTT_MAX_PACKET_SIZE) {
buffer[len] = digit; buffer[len] = digit;
}
len++; len++;
} }
idx++;
}
if (!this->stream && len > MQTT_MAX_PACKET_SIZE) { if (!this->stream && idx > MQTT_MAX_PACKET_SIZE) {
len = 0; // This will cause the packet to be ignored. len = 0; // This will cause the packet to be ignored.
} }