Merge branch 'issue_2_broken_large_payloads' into main

README.md

@@ -12,7 +12,7 @@ TinyMqtt is a small, fast and capable Mqtt Broker and Client for Esp8266 / Esp32
 
 ## Features
 
-- Very (very !!) fast broker I saw it re-sent 1000 topics per second for two
+- Very fast broker I saw it re-sent 1000 topics per second for two
   clients that had subscribed (payload ~15 bytes ESP8266). No topic lost.
   The max I've seen was 2k msg/s (1 client 1 subscription)
 - Act as as a mqtt broker and/or a mqtt client

library.json

@@ -6,7 +6,7 @@
         "type": "git",
         "url": "https://github.com/hsaturn/TinyMqtt.git"
     },
-    "version": "0.7.9",
+    "version": "0.8.0",
     "exclude": "",
     "examples": "examples/*/*.ino",
     "frameworks": "arduino",

library.properties

@@ -1,5 +1,5 @@
 name=TinyMqtt
-version=0.7.9
+version=0.8.0
 author=Francois BIOT, HSaturn, <hsaturn@gmail.com>
 maintainer=Francois BIOT, HSaturn, <hsaturn@gmail.com>
 sentence=A tiny broker and client library for MQTT messaging.

src/TinyMqtt.cpp

@@ -38,7 +38,7 @@ MqttClient::MqttClient(MqttBroker* parent, TcpClient* new_client)
 #else
 	client = new WiFiClient(*new_client);
 #endif
-	alive = millis()+5000;	// client expires after 5s if no CONNECT msg
+	alive = millis()+5000;	// TODO MAGIC client expires after 5s if no CONNECT msg
 }
 
 MqttClient::MqttClient(MqttBroker* parent, const std::string& id)
@@ -190,7 +190,7 @@ MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
 	return MqttNowhereToSend;
 }
 
-MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, const MqttMessage& msg) const
+MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, MqttMessage& msg) const
 {
 	MqttError retval = MqttOk;
 
@@ -397,7 +397,7 @@ MqttError MqttClient::sendTopic(const Topic& topic, MqttMessage::Type type, uint
 
 long MqttClient::counter=0;
 
-void MqttClient::processMessage(const MqttMessage* mesg)
+void MqttClient::processMessage(MqttMessage* mesg)
 {
 	counter++;
 #ifdef TINY_MQTT_DEBUG
@@ -627,6 +627,8 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa
 	MqttMessage msg(MqttMessage::Publish);
 	msg.add(topic);
 	msg.add(payload, pay_length, false);
+	msg.complete();
+
 	if (parent)
 	{
 		return parent->publish(this, topic, msg);
@@ -638,7 +640,7 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa
 }
 
 // republish a received publish if it matches any in subscriptions
-MqttError MqttClient::publishIfSubscribed(const Topic& topic, const MqttMessage& msg)
+MqttError MqttClient::publishIfSubscribed(const Topic& topic, MqttMessage& msg)
 {
 	MqttError retval=MqttOk;
 
@@ -682,29 +684,23 @@ void MqttMessage::incoming(char in_byte)
 	switch(state)
 	{
 		case FixedHeader:
-			size=0;
+			size=MaxBufferLength;
 			state = Length;
 			break;
 		case Length:
-		  if (size==0)
+
+		  if (size==MaxBufferLength)
         size = in_byte & 0x7F;
-		  else if (size<128)
-		    size += static_cast<uint16_t>(in_byte & 0x7F)<<7;
 		  else
-		    state = Error;  // Really don't want to handle msg with length > 16k
+		    size += static_cast<uint16_t>(in_byte & 0x7F)<<7;
+
 			if (size > MaxBufferLength)
-			{
 				state = Error;
-			}
 			else if ((in_byte & 0x80) == 0)
 			{
 				vheader = buffer.length();
 				if (size==0)
 					state = Complete;
-				else if (size > 500)	// TODO magic
-				{
-					state = Error;
-				}
 				else
 				{
 					buffer.reserve(size);
@@ -749,35 +745,21 @@ void MqttMessage::add(const char* p, size_t len, bool addLength)
 	while(len--) incoming(*p++);
 }
 
-void MqttMessage::encodeLength() const
+void MqttMessage::encodeLength()
 {
 	if (state != Complete)
 	{
-		int length = buffer.size()-2;	// 1 byte for header, 1 byte for pre-reserved length field.
+		int length = buffer.size()-3;	// 3 = 1 byte for header + 2 bytes for pre-reserved length field.
-		std::string::size_type ins=1;
+    buffer[1] = 0x80 | (length & 0x7F);
-		do
+    buffer[2] = (length >> 7);
-		{
+    vheader = 3;
-			uint8_t encoded(length & 0x7F);
-			length >>=7;
-			if (length) encoded |= 0x80;
       
-			if (ins==1)
+    // We could check that buffer[2] < 128 (end of length encoding)
-				buffer[ins]=encoded;
+    state = Complete;
-			else
+  }
-				buffer.insert(ins, 1, encoded);
-			// On pourrait optimiser, cet insert est couteux, il faudrait en fait non pas
-			// insérer, mais réserver 4 octets  pour les remplir
-			// plus tard avec ke fixed header et la taille.
-			// Cela changerait en revanche le début du message qui ne serait plus
-			// buffer[0], mais buffer[0..3] selon la taille du message.
-			
-			++ins;
-		} while (length);
-		state = Complete;
-	}
 };
 
-MqttError MqttMessage::sendTo(MqttClient* client) const
+MqttError MqttMessage::sendTo(MqttClient* client)
 {
 	if (buffer.size())
 	{

src/TinyMqtt.h

@@ -64,7 +64,7 @@ class Topic : public IndexedString
 class MqttClient;
 class MqttMessage
 {
-	const uint16_t MaxBufferLength = 4096;  //hard limit: 16k
+	const uint16_t MaxBufferLength = 4096;  //hard limit: 16k due to size decoding
 	public:
 		enum Type
 		{
@@ -101,6 +101,7 @@ class MqttMessage
 		void add(const Topic& t) { add(t.str()); }
 		const char* end() const { return &buffer[0]+buffer.size(); }
 		const char* getVHeader() const { return &buffer[vheader]; }
+		void complete() { encodeLength(); }
 
 		void reset();
 
@@ -116,21 +117,22 @@ class MqttMessage
 		void create(Type type)
 		{
 			buffer=(char)type;
-			buffer+='\0';		// reserved for msg length
+			buffer+='\0';		// reserved for msg length byte 1/2
-			vheader=2;
+			buffer+='\0';		// reserved for msg length byte 2/2 (fixed)
+			vheader=3;      // Should never change
 			size=0;
 			state=Create;
 		}
-		MqttError sendTo(MqttClient*) const;
+		MqttError sendTo(MqttClient*);
 		void hexdump(const char* prefix=nullptr) const;
 
 	private:
-		void encodeLength() const;
+		void encodeLength();
 
-		mutable std::string buffer;	// mutable -> sendTo()
+		std::string buffer;
 		uint8_t vheader;
 		uint16_t size;	// bytes left to receive
-		mutable State state;	// mutable -> encodeLength()
+		State state;
 };
 
 class MqttBroker;
@@ -215,8 +217,7 @@ class MqttClient
 			Serial << endl;
 		}
 
-		/** Count the number of messages that have been sent **/
+		static long counter;  // Number of processed messages
-		static long counter;
 
 	private:
 
@@ -231,10 +232,10 @@ class MqttClient
 		friend class MqttBroker;
 		MqttClient(MqttBroker* parent, TcpClient* client);
 		// republish a received publish if topic matches any in subscriptions
-		MqttError publishIfSubscribed(const Topic& topic, const MqttMessage& msg);
+		MqttError publishIfSubscribed(const Topic& topic, MqttMessage& msg);
 
 		void clientAlive(uint32_t more_seconds);
-		void processMessage(const MqttMessage* message);
+		void processMessage(MqttMessage* message);
 
 		bool mqtt_connected = false;
 		char mqtt_flags;
@@ -291,7 +292,7 @@ class MqttBroker
 		{ return compareString(auth_password, password, len); }
 
 
-		MqttError publish(const MqttClient* source, const Topic& topic, const MqttMessage& msg) const;
+		MqttError publish(const MqttClient* source, const Topic& topic, MqttMessage& msg) const;
 
 		MqttError subscribe(const Topic& topic, uint8_t qos);
 

tests/howto

@@ -0,0 +1,6 @@
+cd TinyMqtt/tests/../..
+git clone https://github.com/hsaturn/EspMock.git
+git clone https://github.com/bxparks/AUnit.git
+git clone https://github.com/bxparks/EpoxyDuino.git
+cd TinyMqtt/tests
+make

tests/nowifi-tests/Makefile

@@ -1,6 +1,11 @@
 # See https://github.com/bxparks/EpoxyDuino for documentation about this
 # Makefile to compile and run Arduino programs natively on Linux or MacOS.
 
+EXTRA_CXXFLAGS=-g3 -O0
+
+# Remove flto flag from EpoxyDuino (too many <optimized out>)
+CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics
+
 APP_NAME := nowifi-tests
 ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi  ESPAsyncTCP
 ARDUINO_LIB_DIRS := ../../../EspMock/libraries

tests/nowifi-tests/nowifi-tests.ino

@@ -132,7 +132,7 @@ test(nowifi_nocallback_when_destroyed)
 	assertEqual(published.size(), (size_t)1);	// Only one publish has been received
 }
 
-test(nowifi_payload_nullptr)
+test(nowifi_small_payload)
 {
 	published.clear();
 
@@ -150,6 +150,22 @@ test(nowifi_payload_nullptr)
 	assertEqual(lastLength, (size_t)4);
 }
 
+test(nowifi_hudge_payload)
+{
+  const char* payload="This payload is hudge, just because its length exceeds 127. Thus when encoding length, we have to encode it on two bytes at min. This should not prevent the message from being encoded and decoded successfully !";
+
+  MqttClient subscriber(&broker);
+  subscriber.setCallback(onPublish);
+  subscriber.subscribe("a/b");
+
+	MqttClient publisher(&broker);
+	publisher.publish("a/b", payload);		// This publish is received
+
+  // onPublish should have filled lastPayload and lastLength
+  assertEqual(payload, lastPayload);
+	assertEqual(lastLength, strlen(payload));
+}
+
 //----------------------------------------------------------------------------
 // setup() and loop()
 void setup() {