Fix issue_2 broken payload

Fix unit tests for EpoxyDuino

library.json

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

library.properties

@@ -1,5 +1,5 @@
 name=TinyMqtt
-version=0.7.6
+version=0.7.9
 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/MqttStreaming.h

@@ -42,7 +42,7 @@
 #ifndef ARDUINO_STREAMING
 #define ARDUINO_STREAMING
 
-#if defined(ARDUINO) && ARDUINO >= 100
+#if (defined(ARDUINO) && ARDUINO >= 100) || defined(EPOXY_DUINO)
 #include "Arduino.h"
 #else
 #ifndef STREAMING_CONSOLE
@@ -154,7 +154,7 @@ template<typename T>
 inline Print &operator <<(Print &obj, const _BASED<T> &arg)
 { obj.print(arg.val, arg.base); return obj; }
 
-#if ARDUINO >= 18
+#if ARDUINO >= 18 || defined(EPOXY_DUINO)
 // Specialization for class _FLOAT
 // Thanks to Michael Margolis for suggesting a way
 // to accommodate Arduino 0018's floating point precision

src/TinyMqtt.cpp

@@ -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");
@@ -397,8 +403,13 @@ void MqttClient::processMessage(const MqttMessage* mesg)
 #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
@@ -609,7 +627,6 @@ 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);
@@ -633,6 +650,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
 		}
 	}
@@ -665,7 +686,12 @@ void MqttMessage::incoming(char in_byte)
 			state = Length;
 			break;
 		case Length:
-			size = (size<<7) + (in_byte & 0x7F);
+		  if (size==0)
+        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
 			if (size > MaxBufferLength)
 			{
 				state = Error;
@@ -705,7 +731,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();
@@ -716,36 +742,47 @@ 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() const
 {
-	do
+	if (state != Complete)
 	{
-		uint8_t encoded(length & 0x7F);
+		int length = buffer.size()-2;	// 1 byte for header, 1 byte for pre-reserved length field.
-		length >>=7;
+		std::string::size_type ins=1;
-		if (length) encoded |= 0x80;
+		do
-		*msb++ = encoded;
+		{
-	} while (length);
+			uint8_t encoded(length & 0x7F);
-};
+			length >>=7;
+			if (length) encoded |= 0x80;
 
-void MqttMessage::complete()
+			if (ins==1)
-{
+				buffer[ins]=encoded;
-		encodeLength(&buffer[1], buffer.size()-2);
+			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
 {
 	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 = 255;
+	const uint16_t MaxBufferLength = 4096;  //hard limit: 16k
 	public:
 		enum Type
 		{
@@ -101,8 +101,6 @@ 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 reset();
 
@@ -127,12 +125,12 @@ class MqttMessage
 		void hexdump(const char* prefix=nullptr) const;
 
 	private:
-		void encodeLength(char* msb, int length) const;
+		void encodeLength() const;
 
 		mutable std::string buffer;	// mutable -> sendTo()
 		uint8_t vheader;
 		uint16_t size;	// bytes left to receive
-		State state;
+		mutable State state;	// mutable -> encodeLength()
 };
 
 class MqttBroker;
@@ -172,7 +170,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);
@@ -214,6 +219,8 @@ class MqttClient
 		static long counter;
 
 	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);
@@ -240,7 +247,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;

tests/length-tests.todo/Makefile

@@ -0,0 +1,8 @@
+# See https://github.com/bxparks/EpoxyDuino for documentation about this
+# Makefile to compile and run Arduino programs natively on Linux or MacOS.
+
+APP_NAME := length-tests
+ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi  ESPAsyncTCP
+ARDUINO_LIB_DIRS := ../../../EspMock/libraries
+EPOXY_CORE := EPOXY_CORE_ESP8266
+include ../../../EpoxyDuino/EpoxyDuino.mk

tests/length-tests.todo/length-tests.ino

@@ -0,0 +1,53 @@
+#include <AUnit.h>
+#include <TinyMqtt.h>
+#include <map>
+
+/**
+	* TinyMqtt local unit tests.
+	*
+	* Clients are connected to pseudo remote broker
+	* The remote should be 127.0.0.1:1883 <--- But this does not work due to Esp network limitations
+	* We are using 127.0.0.1 because this is simpler to test with a single ESP
+	* Also, this will allow to mock and thus run Action on github
+	**/
+
+using namespace std;
+
+MqttBroker broker(1883);
+
+std::map<std::string, std::map<Topic, int>>	published;		// map[client_id] => map[topic] = count
+
+const char* lastPayload;
+size_t lastLength;
+
+void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
+{
+	if (srce)
+		published[srce->id()][topic]++;
+  lastPayload = payload;
+	lastLength = length;
+}
+
+test(length_decode_greater_than_127)
+{
+  // TODO WRITE THIS TEST
+  // The test should verify than a mqtt message with more than 127 bytes is correctly decoded
+	assertEqual(1,2);
+}
+
+
+//----------------------------------------------------------------------------
+// setup() and loop()
+void setup() {
+	delay(1000);
+	Serial.begin(115200);
+	while(!Serial);
+
+	Serial.println("=============[ NO WIFI CONNECTION TinyMqtt TESTS ]========================");
+}
+
+void loop() {
+	aunit::TestRunner::run();
+
+	if (Serial.available()) ESP.reset();
+}

tests/local-tests/local-tests.ino

@@ -1,3 +1,4 @@
+#include <Arduino.h>
 #include <AUnit.h>
 #include <TinyMqtt.h>
 #include <map>

tests/nowifi-tests/nowifi-tests.ino

@@ -1,3 +1,4 @@
+#include <Arduino.h>
 #include <AUnit.h>
 #include <TinyMqtt.h>
 #include <map>

tests/string-indexer-tests/string-indexer-tests.ino

@@ -1,3 +1,4 @@
+#include <Arduino.h>
 #include <AUnit.h>
 #include <StringIndexer.h>
 #include <map>