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

Allow streaming of large messages #568

Merged
merged 5 commits into from
Nov 25, 2019
Merged

Allow streaming of large messages #568

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b623982
Allow streaming of large messages
turgu1 Jan 28, 2019
373c7d3
Update PubSubClient.h
turgu1 Jan 28, 2019
e7d9688
Update receive_spec.cpp
turgu1 Jan 29, 2019
7d6e409
Update receive_spec.cpp
turgu1 Jan 29, 2019
b25040a
Update receive_spec.cpp
turgu1 Jan 29, 2019
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
16 changes: 9 additions & 7 deletions src/PubSubClient.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,7 @@ boolean PubSubClient::connect(const char *id, const char *user, const char *pass
}
}
uint8_t llen;
uint16_t len = readPacket(&llen);
uint32_t len = readPacket(&llen);

if (len == 4) {
if (buffer[3] == 0) {
Expand Down Expand Up @@ -243,12 +243,12 @@ boolean PubSubClient::readByte(uint8_t * result, uint16_t * index){
return false;
}

uint16_t PubSubClient::readPacket(uint8_t* lengthLength) {
uint32_t PubSubClient::readPacket(uint8_t* lengthLength) {
uint16_t len = 0;
if(!readByte(buffer, &len)) return 0;
bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH;
uint32_t multiplier = 1;
uint16_t length = 0;
uint32_t length = 0;
uint8_t digit = 0;
uint16_t skip = 0;
uint8_t start = 0;
Expand Down Expand Up @@ -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 (this->stream) {
if (isPublish && len-*lengthLength-2>skip) {
if (isPublish && idx-*lengthLength-2>skip) {
this->stream->write(digit);
}
}
if (len < MQTT_MAX_PACKET_SIZE) {
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.
}

Expand Down
2 changes: 1 addition & 1 deletion src/PubSubClient.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ class PubSubClient : public Print {
unsigned long lastInActivity;
bool pingOutstanding;
MQTT_CALLBACK_SIGNATURE;
uint16_t readPacket(uint8_t*);
uint32_t readPacket(uint8_t*);
boolean readByte(uint8_t * result);
boolean readByte(uint8_t * result, uint16_t * index);
boolean write(uint8_t header, uint8_t* buf, uint16_t length);
Expand Down
4 changes: 2 additions & 2 deletions tests/src/receive_spec.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,7 @@ int test_drop_invalid_remaining_length_message() {


int test_receive_oversized_stream_message() {
IT("drops an oversized message");
IT("receive an oversized streamed message");
reset_callback();

Stream stream;
Expand Down Expand Up @@ -222,7 +222,7 @@ int test_receive_oversized_stream_message() {

IS_TRUE(callback_called);
IS_TRUE(strcmp(lastTopic,"topic")==0);
IS_TRUE(lastLength == length-9);
IS_TRUE(lastLength == MQTT_MAX_PACKET_SIZE-9);

IS_FALSE(stream.error());
IS_FALSE(shimClient.error());
Expand Down