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)
@@ -59,7 +59,7 @@ void MqttClient::close(bool bSendDisconnect)
 {
 	debug("close " << id().c_str());
 	mqtt_connected = false;
-	if (client)
+	if (client)	// connected to a remote broker
 	{
 		if (bSendDisconnect and client->connected())
 		{
@@ -72,10 +72,16 @@ void MqttClient::close(bool bSendDisconnect)
 	if (parent)
 	{
 		parent->removeClient(this);
-		parent=nullptr;
+		parent = nullptr;
 	}
 }
 
+void MqttClient::connect(MqttBroker* parentBroker)
+{
+	close();
+	parent = parentBroker;
+}
+
 void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
 {
 	debug("cnx: closing");
@@ -184,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;
 
@@ -391,14 +397,19 @@ 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
 if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::Type::PingResp)
 {
-	Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
+	#ifdef NOT_ESP_CORE
+		Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << " ESP.getFreeHeap() "<< endl;
+	#else
+		Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
+	#endif
 	// mesg->hexdump("Incoming");
+	mesg->hexdump("Incoming");
 }
 #endif
   auto header = mesg->getVHeader();
@@ -540,6 +551,9 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
 			break;
 
 		case MqttMessage::Type::Publish:
+			#ifdef TINY_MQTT_DEBUG
+				Serial << "publish " << mqtt_connected << '/' << (long) client << endl;
+			#endif
 			if (mqtt_connected or client == nullptr)
 			{
 				uint8_t qos = mesg->type() & 0x6;
@@ -554,8 +568,12 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
 				// TODO reset DUP
 				// TODO reset RETAIN
 
-				if (client==nullptr)	// internal MqttClient receives publish
+				if (parent==nullptr or client==nullptr)	// internal MqttClient receives publish
 				{
+					#ifdef TINY_MQTT_DEBUG
+						Serial << (isSubscribedTo(published) ? "not" : "") << " subscribed.\n";
+						Serial << "has " << (callback ? "" : "no ") << " callback.\n";
+					#endif
 					if (callback and isSubscribedTo(published))
 					{
 						callback(this, published, payload, len);	// TODO send the real payload
@@ -610,6 +628,7 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa
 	msg.add(topic);
 	msg.add(payload, pay_length, false);
 	msg.complete();
+
 	if (parent)
 	{
 		return parent->publish(this, topic, msg);
@@ -621,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;
 
@@ -633,6 +652,10 @@ MqttError MqttClient::publishIfSubscribed(const Topic& topic, const MqttMessage&
 		else
 		{
 			processMessage(&msg);
+
+			#ifdef TINY_MQTT_DEBUG
+				Serial << "Should call the callback ?\n";
+			#endif
 			// callback(this, topic, nullptr, 0);	// TODO Payload
 		}
 	}
@@ -661,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);
@@ -710,7 +727,7 @@ void MqttMessage::incoming(char in_byte)
 			reset();
 			break;
 	}
-	if (buffer.length() > MaxBufferLength)	// TODO magic 256 ?
+	if (buffer.length() > MaxBufferLength)
 	{
 		debug("Too long " << state);
 		reset();
@@ -721,36 +738,33 @@ void MqttMessage::add(const char* p, size_t len, bool addLength)
 {
 	if (addLength)
 	{
-		buffer.reserve(buffer.length()+addLength+2);
+		buffer.reserve(buffer.length()+2);
 		incoming(len>>8);
 		incoming(len & 0xFF);
 	}
 	while(len--) incoming(*p++);
 }
 
-void MqttMessage::encodeLength(char* msb, int length) const
+void MqttMessage::encodeLength()
 {
-	do
+	if (state != Complete)
 	{
-		uint8_t encoded(length & 0x7F);
-		length >>=7;
-		if (length) encoded |= 0x80;
-		*msb++ = encoded;
-	} while (length);
+		int length = buffer.size()-3;	// 3 = 1 byte for header + 2 bytes for pre-reserved length field.
+    buffer[1] = 0x80 | (length & 0x7F);
+    buffer[2] = (length >> 7);
+    vheader = 3;
+      
+    // We could check that buffer[2] < 128 (end of length encoding)
+    state = Complete;
+  }
 };
 
-void MqttMessage::complete()
-{
-		encodeLength(&buffer[1], buffer.size()-2);
-		state = Complete;
-}
-
-MqttError MqttMessage::sendTo(MqttClient* client) const
+MqttError MqttMessage::sendTo(MqttClient* client)
 {
 	if (buffer.size())
 	{
 		debug("sending " << buffer.size() << " bytes");
-		encodeLength(&buffer[1], buffer.size()-2);
+		encodeLength();
 		// hexdump("snd");
 		client->write(&buffer[0], 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,8 +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]; }
-		uint16_t length() const { return buffer.size(); }
-		void complete();
+		void complete() { encodeLength(); }
 
 		void reset();
 
@@ -118,18 +117,19 @@ class MqttMessage
 		void create(Type type)
 		{
 			buffer=(char)type;
-			buffer+='\0';		// reserved for msg length
-			vheader=2;
+			buffer+='\0';		// reserved for msg length byte 1/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(char* msb, int length) const;
+		void encodeLength();
 
-		mutable std::string buffer;	// mutable -> sendTo()
+		std::string buffer;
 		uint8_t vheader;
 		uint16_t size;	// bytes left to receive
 		State state;
@@ -172,7 +172,14 @@ class MqttClient
 		/** Should be called in main loop() */
 		void loop();
 		void close(bool bSendDisconnect=true);
-		void setCallback(CallBack fun) {callback=fun; };
+		void setCallback(CallBack fun)
+		{
+			callback=fun;
+			#ifdef TINY_MQTT_DEBUG
+				Serial << "Callback set to " << (long)fun << endl;
+				if (callback) callback(this, "test/topic", "value", 5);
+			#endif
+		};
 
 		// Publish from client to the world
 		MqttError publish(const Topic&, const char* payload, size_t pay_length);
@@ -210,10 +217,11 @@ class MqttClient
 			Serial << endl;
 		}
 
-		/** Count the number of messages that have been sent **/
-		static long counter;
+		static long counter;  // Number of processed messages
 
 	private:
+
+		// event when tcp/ip link established (real or fake)
 		static void onConnect(void * client_ptr, TcpClient*);
 #ifdef TCP_ASYNC
 		static void onData(void* client_ptr, TcpClient*, void* data, size_t len);
@@ -224,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;
@@ -240,7 +248,7 @@ class MqttClient
 		// (this is the case when MqttBroker isn't used except here)
 		MqttBroker* parent=nullptr;		// connection to local broker
 
-		TcpClient* client=nullptr;		// connection to mqtt client or to remote broker
+		TcpClient* client=nullptr;		// connection to remote broker
 		std::set<Topic>	subscriptions;
 		std::string clientId;
 		CallBack callback = nullptr;
@@ -284,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() {