Update PubSubClient 2.7 (#1224)

Author: tekka007

drivers/PubSubClient/CHANGES.txt

@@ -1,8 +1,16 @@
+2.7
+   * Fix remaining-length handling to prevent buffer overrun
+   * Add large-payload API - beginPublish/write/publish/endPublish
+   * Add yield call to improve reliability on ESP
+   * Add Clean Session flag to connect options
+   * Add ESP32 support for functional callback signature
+   * Various other fixes
+
 2.4
    * Add MQTT_SOCKET_TIMEOUT to prevent it blocking indefinitely
      whilst waiting for inbound data
    * Fixed return code when publishing >256 bytes
-   
+
 2.3
    * Add publish(topic,payload,retained) function
 

drivers/PubSubClient/PubSubClient.cpp

@@ -17,15 +17,13 @@ PubSubClient::PubSubClient()
 	this->stream = NULL;
 	setCallback(NULL);
 }
-
 // cppcheck-suppress uninitMemberVar
 PubSubClient::PubSubClient(Client& client)
 {
 	this->_state = MQTT_DISCONNECTED;
 	setClient(client);
 	this->stream = NULL;
 }
-
 // cppcheck-suppress uninitMemberVar
 PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client)
 {
@@ -63,7 +61,6 @@ PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATUR
 	setClient(client);
 	setStream(stream);
 }
-
 // cppcheck-suppress uninitMemberVar
 PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client)
 {
@@ -101,7 +98,6 @@ PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE,
 	setClient(client);
 	setStream(stream);
 }
-
 // cppcheck-suppress uninitMemberVar
 PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client)
 {
@@ -143,22 +139,29 @@ PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGN
 
 bool PubSubClient::connect(const char *id)
 {
-	return connect(id,NULL,NULL,0,0,0,0);
+	return connect(id,NULL,NULL,0,0,0,0,1);
 }
 
 bool PubSubClient::connect(const char *id, const char *user, const char *pass)
 {
-	return connect(id,user,pass,0,0,0,0);
+	return connect(id,user,pass,0,0,0,0,1);
 }
 
 bool PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos,
                            bool willRetain, const char* willMessage)
 {
-	return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage);
+	return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage,1);
 }
 
 bool PubSubClient::connect(const char *id, const char *user, const char *pass,
                            const char* willTopic, uint8_t willQos, bool willRetain, const char* willMessage)
+{
+	return connect(id,user,pass,willTopic,willQos,willRetain,willMessage,1);
+}
+
+bool PubSubClient::connect(const char *id, const char *user, const char *pass,
+                           const char* willTopic, uint8_t willQos, bool willRetain, const char* willMessage,
+                           bool cleanSession)
 {
 	if (!connected()) {
 		int result = 0;
@@ -171,7 +174,7 @@ bool PubSubClient::connect(const char *id, const char *user, const char *pass,
 		if (result == 1) {
 			nextMsgId = 1;
 			// Leave room in the buffer for header and variable length field
-			uint16_t length = 5;
+			uint16_t length = MQTT_MAX_HEADER_SIZE;
 			unsigned int j;
 
 #if MQTT_VERSION == MQTT_VERSION_3_1
@@ -187,9 +190,12 @@ bool PubSubClient::connect(const char *id, const char *user, const char *pass,
 
 			uint8_t v;
 			if (willTopic) {
-				v = 0x06|(willQos<<3)|(willRetain<<5);
+				v = 0x04|(willQos<<3)|(willRetain<<5);
 			} else {
-				v = 0x02;
+				v = 0x00;
+			}
+			if (cleanSession) {
+				v = v|0x02;
 			}
 
 			if(user != NULL) {
@@ -204,20 +210,26 @@ bool PubSubClient::connect(const char *id, const char *user, const char *pass,
 
 			buffer[length++] = ((MQTT_KEEPALIVE) >> 8);
 			buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF);
+
+			CHECK_STRING_LENGTH(length,id)
 			length = writeString(id,buffer,length);
 			if (willTopic) {
+				CHECK_STRING_LENGTH(length,willTopic)
 				length = writeString(willTopic,buffer,length);
+				CHECK_STRING_LENGTH(length,willMessage)
 				length = writeString(willMessage,buffer,length);
 			}
 
 			if(user != NULL) {
+				CHECK_STRING_LENGTH(length,user)
 				length = writeString(user,buffer,length);
 				if(pass != NULL) {
+					CHECK_STRING_LENGTH(length,pass)
 					length = writeString(pass,buffer,length);
 				}
 			}
 
-			write(MQTTCONNECT,buffer,length-5);
+			write(MQTTCONNECT,buffer,length-MQTT_MAX_HEADER_SIZE);
 
 			lastInActivity = lastOutActivity = millis();
 
@@ -256,6 +268,7 @@ bool PubSubClient::readByte(uint8_t * result)
 {
 	uint32_t previousMillis = millis();
 	while(!_client->available()) {
+		yield();
 		uint32_t currentMillis = millis();
 		if(currentMillis - previousMillis >= ((int32_t) MQTT_SOCKET_TIMEOUT * 1000)) {
 			return false;
@@ -291,6 +304,12 @@ uint16_t PubSubClient::readPacket(uint8_t* lengthLength)
 	uint8_t start = 0;
 
 	do {
+		if (len == 5) {
+			// Invalid remaining length encoding - kill the connection
+			_state = MQTT_DISCONNECTED;
+			_client->stop();
+			return 0;
+		}
 		if(!readByte(&digit)) {
 			return 0;
 		}
@@ -364,16 +383,15 @@ bool PubSubClient::loop()
 				uint8_t type = buffer[0]&0xF0;
 				if (type == MQTTPUBLISH) {
 					if (callback) {
-						uint16_t tl = (buffer[llen+1]<<8)+buffer[llen+2];
+						uint16_t tl = (buffer[llen+1]<<8)+buffer[llen+2]; /* topic length in bytes */
+						memmove(buffer+llen+2,buffer+llen+3,tl); /* move topic inside buffer 1 byte to front */
+						buffer[llen+2+tl] = 0; /* end the topic as a 'C' string with \x00 */
+						char *topic = (char*) buffer+llen+2;
 						uint8_t *payload;
-						char topic[tl+1];
-						for (uint16_t i=0; i<tl; i++) {
-							topic[i] = buffer[llen+3+i];
-						}
-						topic[tl] = 0;
 						// msgId only present for QOS>0
 						if ((buffer[0]&0x06) == MQTTQOS1) {
-							uint16_t msgId = (buffer[llen+3+tl]<<8)+buffer[llen+3+tl+1];
+							uint16_t msgId = 0;
+							msgId = (buffer[llen+3+tl]<<8)+buffer[llen+3+tl+1];
 							payload = buffer+llen+3+tl+2;
 							callback(topic,payload,len-llen-3-tl-2);
 
@@ -396,6 +414,9 @@ bool PubSubClient::loop()
 				} else if (type == MQTTPINGRESP) {
 					pingOutstanding = false;
 				}
+			} else if (!connected()) {
+				// readPacket has closed the connection
+				return false;
 			}
 		}
 		return true;
@@ -422,12 +443,12 @@ bool PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned i
                            bool retained)
 {
 	if (connected()) {
-		if (MQTT_MAX_PACKET_SIZE < 5 + 2+strlen(topic) + plength) {
+		if (MQTT_MAX_PACKET_SIZE < MQTT_MAX_HEADER_SIZE + 2+strlen(topic) + plength) {
 			// Too long
 			return false;
 		}
 		// Leave room in the buffer for header and variable length field
-		uint16_t length = 5;
+		uint16_t length = MQTT_MAX_HEADER_SIZE;
 		length = writeString(topic,buffer,length);
 		uint16_t i;
 		for (i=0; i<plength; i++) {
@@ -437,11 +458,16 @@ bool PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned i
 		if (retained) {
 			header |= 1;
 		}
-		return write(header,buffer,length-5);
+		return write(header,buffer,length-MQTT_MAX_HEADER_SIZE);
 	}
 	return false;
 }
 
+bool PubSubClient::publish_P(const char* topic, const char* payload, bool retained)
+{
+	return publish_P(topic, (const uint8_t*)payload, strlen(payload), retained);
+}
+
 bool PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned int plength,
                              bool retained)
 {
@@ -487,12 +513,46 @@ bool PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned
 	return rc == tlen + 4 + plength;
 }
 
-bool PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length)
+bool PubSubClient::beginPublish(const char* topic, unsigned int plength, bool retained)
+{
+	if (connected()) {
+		// Send the header and variable length field
+		uint16_t length = MQTT_MAX_HEADER_SIZE;
+		length = writeString(topic,buffer,length);
+		uint8_t header = MQTTPUBLISH;
+		if (retained) {
+			header |= 1;
+		}
+		size_t hlen = buildHeader(header, buffer, plength+length-MQTT_MAX_HEADER_SIZE);
+		uint16_t rc = _client->write(buffer+(MQTT_MAX_HEADER_SIZE-hlen),length-(MQTT_MAX_HEADER_SIZE-hlen));
+		lastOutActivity = millis();
+		return (rc == (length-(MQTT_MAX_HEADER_SIZE-hlen)));
+	}
+	return false;
+}
+
+int PubSubClient::endPublish()
+{
+	return 1;
+}
+
+size_t PubSubClient::write(uint8_t data)
+{
+	lastOutActivity = millis();
+	return _client->write(data);
+}
+
+size_t PubSubClient::write(const uint8_t *buffer, size_t size)
+{
+	lastOutActivity = millis();
+	return _client->write(buffer,size);
+}
+
+size_t PubSubClient::buildHeader(uint8_t header, uint8_t* buf, uint16_t length)
 {
 	uint8_t lenBuf[4];
 	uint8_t llen = 0;
 	uint8_t pos = 0;
-	uint16_t rc;
 	uint16_t len = length;
 	do {
 		uint8_t digit;
@@ -507,12 +567,19 @@ bool PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length)
 
 	buf[4-llen] = header;
 	for (int i=0; i<llen; i++) {
-		buf[5-llen+i] = lenBuf[i];
+		buf[MQTT_MAX_HEADER_SIZE-llen+i] = lenBuf[i];
 	}
+	return llen+1; // Full header size is variable length bit plus the 1-byte fixed header
+}
+
+bool PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length)
+{
+	uint16_t rc;
+	uint8_t hlen = buildHeader(header, buf, length);
 
 #ifdef MQTT_MAX_TRANSFER_SIZE
-	uint8_t* writeBuf = buf+(4-llen);
-	uint16_t bytesRemaining = length+1+llen;  //Match the length type
+	uint8_t* writeBuf = buf+(MQTT_MAX_HEADER_SIZE-hlen);
+	uint16_t bytesRemaining = length+hlen;  //Match the length type
 	uint8_t bytesToWrite;
 	bool result = true;
 	while((bytesRemaining > 0) && result) {
@@ -524,9 +591,9 @@ bool PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length)
 	}
 	return result;
 #else
-	rc = _client->write(buf+(4-llen),length+1+llen);
+	rc = _client->write(buf+(MQTT_MAX_HEADER_SIZE-hlen),length+hlen);
 	lastOutActivity = millis();
-	return (rc == 1+llen+length);
+	return (rc == hlen+length);
 #endif
 }
 
@@ -537,7 +604,6 @@ bool PubSubClient::subscribe(const char* topic)
 
 bool PubSubClient::subscribe(const char* topic, uint8_t qos)
 {
-	// original: if (qos < 0 || qos > 1) { (qos is uint8_t, hence qos < 0 impossible, tekka)
 	if (qos > 1) {
 		return false;
 	}
@@ -547,7 +613,7 @@ bool PubSubClient::subscribe(const char* topic, uint8_t qos)
 	}
 	if (connected()) {
 		// Leave room in the buffer for header and variable length field
-		uint16_t length = 5;
+		uint16_t length = MQTT_MAX_HEADER_SIZE;
 		nextMsgId++;
 		if (nextMsgId == 0) {
 			nextMsgId = 1;
@@ -556,7 +622,7 @@ bool PubSubClient::subscribe(const char* topic, uint8_t qos)
 		buffer[length++] = (nextMsgId & 0xFF);
 		length = writeString((char*)topic, buffer,length);
 		buffer[length++] = qos;
-		return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-5);
+		return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-MQTT_MAX_HEADER_SIZE);
 	}
 	return false;
 }
@@ -568,15 +634,15 @@ bool PubSubClient::unsubscribe(const char* topic)
 		return false;
 	}
 	if (connected()) {
-		uint16_t length = 5;
+		uint16_t length = MQTT_MAX_HEADER_SIZE;
 		nextMsgId++;
 		if (nextMsgId == 0) {
 			nextMsgId = 1;
 		}
 		buffer[length++] = (nextMsgId >> 8);
 		buffer[length++] = (nextMsgId & 0xFF);
 		length = writeString(topic, buffer,length);
-		return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-5);
+		return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-MQTT_MAX_HEADER_SIZE);
 	}
 	return false;
 }
@@ -587,6 +653,7 @@ void PubSubClient::disconnect()
 	buffer[1] = 0;
 	_client->write(buffer,2);
 	_state = MQTT_DISCONNECTED;
+	_client->flush();
 	_client->stop();
 	lastInActivity = lastOutActivity = millis();
 }
@@ -644,7 +711,6 @@ PubSubClient& PubSubClient::setServer(const char * domain, uint16_t port)
 	this->port = port;
 	return *this;
 }
-
 // cppcheck-suppress passedByValue
 PubSubClient& PubSubClient::setCallback(MQTT_CALLBACK_SIGNATURE)
 {