Release 0.7.0

.github/workflows/superlinter.yml

@@ -0,0 +1,25 @@
+name: Super-Linter
+
+# Run this workflow every time a new commit pushed to your repository
+on: push
+
+jobs:
+  # Set the job key. The key is displayed as the job name
+  # when a job name is not provided
+  super-lint:
+    # Name the Job
+    name: Lint code base
+    # Set the type of machine to run on
+    runs-on: ubuntu-latest
+
+    steps:
+      # Checks out a copy of your repository on the ubuntu-latest machine
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      # Runs the Super-Linter action
+      - name: Run Super-Linter
+        uses: github/super-linter@v3
+        env:
+          DEFAULT_BRANCH: main
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -1 +1,2 @@
 *~
+src/my_credentials.h

README.md

@@ -2,14 +2,17 @@
 
 ![](https://img.shields.io/github/v/release/hsaturn/TinyMqtt)
 ![](https://img.shields.io/github/issues/hsaturn/TinyMqtt)
-![](https://img.shields.io/badge/paltform-ESP8266-green)
+![](https://img.shields.io/badge/platform-ESP8266-green)
 ![](https://img.shields.io/github/license/hsaturn/TinyMqtt)
 ![](https://img.shields.io/badge/Mqtt-%203.1.1-yellow)
 
-ESP 8266 is a small and very capable Mqtt Broker and Client
+ESP 8266 is a small, fast and capable Mqtt Broker and Client
 
 ## Features
 
+- Very (very !!) fast broker I saw it re-sent 1000 topics per second for two
+  clients that had subscribed (payload ~15 bytes). 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
 - Mqtt 3.1.1 / Qos 0 supported
 - Standalone (can work without WiFi) (degraded/local mode)
@@ -23,16 +26,18 @@ ESP 8266 is a small and very capable Mqtt Broker and Client
 * Implement zeroconf mode (needs async)
 * Add a max_clients in MqttBroker. Used with zeroconf, there will be
 no need for having tons of clients (also RAM is the problem with many clients)
-* Test what is the real max number of clients for broker. As far as I saw, 3k is needed per client which would make more than 10 clients critical.
+* Test what is the real max number of clients for broker. As far as I saw, 1k is needed per client which would make more than 30 clients critical.
-* MqttMessage uses a buffer 256 bytes which is usually far than needed.
+* ~~MqttMessage uses a buffer 256 bytes which is usually far than needed.~~
+* ~~MqttClient does not support more than one subscription at time~~
+* ~~MqttClient auto re-subscribe (::resubscribe works bad on broker.emqx.io)~~
 * MqttClient auto reconnection
-* MqttClient auto re-subscribe
+* ~~MqttClient unsubscribe~~
-* MqttClient does not callback payload...
+* MqttClient does not sent payload to callback...
 * MqttClient user/password
+* Wildcards (I may implement only # as I'm not interrested by a clever and cpu consuming matching)
 
 ## Quickstart
 
-* install [Streaming library](https://github.com/janelia-arduino/Streaming)
 * install [TinyMqtt library](https://github.com/hsaturn/TinyMqtt)
 * modify <libraries/TinyMqtt/src/my_credentials.h> (wifi setup)
 

examples/client-with-wifi/client-with-wifi.ino

@@ -1,5 +1,4 @@
 #include <TinyMqtt.h>   // https://github.com/hsaturn/TinyMqtt
-#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
 
 /** 
   * Local broker that accept connections and two local clients

examples/client-without-wifi/client-without-wifi.ino

@@ -1,5 +1,4 @@
 #include <TinyMqtt.h>   // https://github.com/hsaturn/TinyMqtt
-#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
 
 /** TinyMQTT allows a disconnected mode:
  * 
@@ -15,10 +14,10 @@ MqttBroker broker(1883);
 MqttClient mqtt_a(&broker);
 MqttClient mqtt_b(&broker);
 
-void onPublishA(const Topic& topic, const char* payload, size_t length)
+void onPublishA(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
 { Serial << "--> A Received " << topic.c_str() << endl; }
 
-void onPublishB(const Topic& topic, const char* payload, size_t length)
+void onPublishB(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
 { Serial << "--> B Received " << topic.c_str() << endl; }
 
 void setup()

examples/simple-broker/simple-broker.ino

@@ -1,5 +1,4 @@
 #include "TinyMqtt.h"   // https://github.com/hsaturn/TinyMqtt
-#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
 
 #include <my_credentials.h>
 

examples/simple-client/simple-client.ino

@@ -1,6 +1,5 @@
 #include <ESP8266WiFi.h>
 #include "TinyMqtt.h"   // https://github.com/hsaturn/TinyMqtt
-#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
 
 /** Simple Client
   *
@@ -43,7 +42,11 @@ void loop()
 
 	delay(1000);
 
-	temp += (random(100)>50 ? 0.1 : -0.1);
+	auto rnd=random(100);
+
+	if (rnd > 66) temp += 0.1;
+	else if (rnd < 33) temp -= 0.1;
+
 	client.publish("sensor/temperature", String(temp));
 
 }

examples/tinymqtt-test/commands.txt

@@ -0,0 +1,39 @@
+// vim: ts=40
+Exemple of commands that can be sent via the serial monitor to tinymqtt-test
+----------------------------------------------------------------------------
+Commands can usually be abbreviated to their first letters.
+ex: cl for client, a / a.con / a.sub / a.p for publish.
+--------
+
+set name value	set variable name to value (later replaced)
+set name	if no value, then var is erased
+set	view all vars
+reserved keywords are forbidden
+
+
+client a	starts a client (not connected no internal broker)
+a.connect [server][port][alive]	connects the client, default port=1883
+a.publish topic [payload]	send a topic with a payload
+a.subscribe topic	subscribes to a topic
+delete a	destroy the client
+
+* note, if 'server' is a number, then it replaces the end of the local ip.
+i.e. if local ip is 192.168.1.10, connect 2.35 becomes 192.168.2.35
+
+--------
+example:
+--------
+
+client c
+c.connect broker.emqx.io
+set topic sensor/temperature
+
+c.subscribe topic
+c.publish topic 15
+c.publish topic 20
+
+macro exansion example
+----------------------
+
+set temp publish sensor/temperature
+c.temp 20	-> c.publish sensor/temperature 20

examples/tinymqtt-test/tinymqtt-test.ino

@@ -1,6 +1,7 @@
 #define TINY_MQTT_DEBUG
-#include <TinyMqtt.h>   // https://github.com/hsaturn/TinyMqtt
+#include <TinyMqtt.h>       // https://github.com/hsaturn/TinyMqtt
-#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
+#include <MqttStreaming.h>
+#include <sstream>
 #include <map>
 
 /** 
@@ -15,11 +16,6 @@
   * cons - Takes more memory
 	*      - a bit hard to understand
 	*
-  * This sounds crazy: a mqtt mqtt that do not need a broker !
-  *  The use case arise when one ESP wants to publish topics and subscribe to them at the same time.
-  *  Without broker, the ESP won't react to its own topics.
-  *  
-  *  TinyMqtt mqtt allows this use case to work.
   */
 
 #include <my_credentials.h>
@@ -27,18 +23,26 @@
 std::string topic="sensor/temperature";
 
 void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
-{ Serial << "--> " << srce->id().c_str() << ": ======> received " << topic.c_str() << endl; }
+{
+	Serial << "--> " << srce->id().c_str() << ": ======> received " << topic.c_str();
+	if (payload) Serial << ", payload[" << length << "]=[";
+	while(length--)
+	{
+		const char c=*payload++;
+		if (c!=10 and c!=13 and c <32) Serial << '?';
+		Serial << *payload++;
+	}
+	Serial<< endl;
+}
 
 std::map<std::string, MqttClient*> clients;
 std::map<std::string, MqttBroker*> brokers;
 
-
 void setup()
 {
   Serial.begin(115200);
 	delay(500);
 	Serial << endl << endl << endl
-	<< "Demo started. Type help for more..." << endl
 	<< "Connecting to '" << ssid << "' ";
 
   WiFi.mode(WIFI_STA);
@@ -48,21 +52,22 @@ void setup()
 	{ Serial << '-'; delay(500); }
 
   Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
+	Serial << "Type help for more..." << endl;
 
 	MqttBroker* broker = new MqttBroker(1883);
 	broker->begin();
 	brokers["broker"] = broker;
 }
 
-int getint(std::string& str, const int if_empty=0, char sep=' ')
+int getint(std::string& str, const int if_empty=0)
 {
 	std::string sword;
-	while(str.length() && str[0]!=sep)
+	while(str.length() && str[0]>='0' && str[0]<='9')
 	{
 		sword += str[0]; str.erase(0,1);
 	}
-	while(str[0]==sep) str.erase(0,1);
+	while(str[0]==' ') str.erase(0,1);
-	if (if_empty and sword.length()==0) sword=if_empty;
+	if (if_empty and sword.length()==0) return if_empty;
 	return atoi(sword.c_str());
 }
 
@@ -78,6 +83,111 @@ std::string getword(std::string& str, const char* if_empty=nullptr, char sep=' '
 	return sword;
 }
 
+bool isaddr(std::string s)
+{
+	if (s.length()==0 or s.length()>3) return false;
+	for(char c: s)
+		if (c<'0' or c>'9') return false;
+	return true;
+}
+
+std::string getip(std::string& str, const char* if_empty=nullptr, char sep=' ')
+{
+	std::string addr=getword(str, if_empty, sep);
+	std::string ip=addr;
+	std::vector<std::string> build;
+	while(ip.length())
+	{
+		std::string b=getword(ip,nullptr,'.');
+		if (isaddr(b) && build.size()<4)
+		{
+			build.push_back(b);
+		}
+		else
+			return addr;
+	}
+	IPAddress local=WiFi.localIP();
+	addr="";
+	while(build.size()!=4)
+	{
+		std::stringstream b;
+		b << (int)local[3-build.size()];
+		build.insert(build.begin(), b.str());
+	}
+	for(std::string s: build)
+	{
+		if (addr.length()) addr += '.';
+		addr += s;
+	}
+	return addr;
+}
+
+std::map<std::string, std::string> vars;
+
+std::set<std::string> commands = {
+	"auto", "broker", "client", "connect",
+	"create", "delete", "help", "interval",
+	"ls", "ip", "off", "on", "set",
+	"publish", "reset", "subscribe", "view"
+};
+
+void getCommand(std::string& search)
+{
+	while(search[0]==' ') search.erase(0,1);
+	if (search.length()==0) return;
+	std::string matches;
+	int count=0;
+	for(std::string cmd: commands)
+	{
+		if (cmd.substr(0, search.length()) == search)
+		{
+			if (count) matches +=", ";
+			count++;
+			matches += cmd;
+		}
+	}
+	if (count==1)
+		search = matches;
+	else if (count>1)
+	{
+		Serial << "Ambiguous command: " << matches << endl;
+		search="";
+	}
+}
+
+void replace(const char* d, std::string& str, std::string srch, std::string to)
+{
+	if (d[0] && d[1])
+	{
+		srch=d[0]+srch+d[1];
+		to=d[0]+to+d[1];
+
+		size_t pos = 0;
+		while((pos=str.find(srch, pos)) != std::string::npos)
+		{
+			str.erase(pos, srch.length());
+			str.insert(pos, to);
+			pos += to.length();
+		}
+	}
+}
+
+void replaceVars(std::string& cmd)
+{
+	cmd = ' '+cmd+' ';
+
+	for(auto it: vars)
+	{
+		replace("..", cmd, it.first, it.second);
+		replace(". ", cmd, it.first, it.second);
+		replace(" .", cmd, it.first, it.second);
+		replace("  ", cmd, it.first, it.second);
+	}
+	cmd.erase(0, cmd.find_first_not_of(" "));
+	cmd.erase(cmd.find_last_not_of(" ")+1);
+}
+
+// publish at regular interval
 class automatic
 {
 	public:
@@ -179,6 +289,7 @@ class automatic
 		{
 					Serial << "    auto [$id] on/off" << endl;
 					Serial << "    auto [$id] view" << endl;
+					Serial << "    auto [$id] interval [s]" << endl;
 					Serial << "    auto [$id] create [millis] [topic]" << endl;
 		}
 
@@ -189,6 +300,7 @@ class automatic
 		std::string topic_;
 		bool bon=false;
 		static std::map<MqttClient*, automatic*> autos;
+		float temp=19;
 };
 std::map<MqttClient*, automatic*> automatic::autos;
 
@@ -202,6 +314,12 @@ using ClientFunction = void(*)(std::string& cmd, MqttClient* publish);
 
 void loop()
 {
+	static long count;
+	if (MqttClient::counter != count)
+	{
+		Serial << "# " << MqttClient::counter << endl;
+		count = MqttClient::counter;
+	}
 	for(auto it: brokers)
 		it.second->loop();
 
@@ -217,41 +335,53 @@ void loop()
 
 		if (c==10 or c==14)
 		{
+
 			Serial << "----------------[ " << cmd.c_str() << " ]--------------" << endl;
 			static std::string last_cmd;
 			if (cmd=="!")
 				cmd=last_cmd;
 			else
 				last_cmd=cmd;
+
+			if (cmd.substr(0,3)!="set") replaceVars(cmd);
 			while(cmd.length())
 			{
+				MqttError retval = MqttOk;
+
 				std::string s;
 				MqttBroker* broker = nullptr;
 				MqttClient* client = nullptr;
 
 				// client.function notation
 				// ("a.fun " becomes "fun a ")
-				if (cmd.find('.') != std::string::npos)
+				if (cmd.find('.') != std::string::npos && 
+						cmd.find('.') < cmd.find(' '))
 				{
 					s=getword(cmd, nullptr, '.');
 
-					if (clients.find(s) != clients.end())
+					if (s.length())
 					{
-						client = clients[s];
+						if (clients.find(s) != clients.end())
-					}
+						{
-					else if (brokers.find(s) != brokers.end())
+							client = clients[s];
-					{
+						}
-						broker = brokers[s];
+						else if (brokers.find(s) != brokers.end())
-					}
+						{
-					else
+							broker = brokers[s];
-					{
+						}
-						Serial << "Unknown class (" << s.c_str() << ")" << endl;
+						else
-						cmd="";
+						{
+							Serial << "Unknown class (" << s.c_str() << ")" << endl;
+							cmd="";
+						}
 					}
 				}
 				
 				s = getword(cmd);
-				if (compare(s, "delete"))
+				if (s.length()) getCommand(s);
+				if (s.length()==0)
+				{}
+				else if (compare(s, "delete"))
 				{
 					if (client==nullptr && broker==nullptr)
 					{
@@ -309,16 +439,14 @@ void loop()
 				{
 					if (compare(s,"connect"))
 					{
-						client->connect(getword(cmd,"192.168.1.40").c_str(), 1883);
+						client->connect(getip(cmd,"192.168.1.40").c_str(), getint(cmd, 1883), getint(cmd, 60));
 						Serial << (client->connected() ? "connected." : "not connected") << endl;
 					}
 					else if (compare(s,"publish"))
 					{
-						auto ok=client->publish(getword(cmd, topic.c_str()));
+						while (cmd[0]==' ') cmd.erase(0,1);
-						if (ok != MqttOk)
+						retval = client->publish(getword(cmd, topic.c_str()), cmd.c_str(), cmd.length());
-						{
+						cmd="";	// remove payload
-							Serial << "## ERROR " << ok << endl;
-						}
 					}
 					else if (compare(s,"subscribe"))
 					{
@@ -382,6 +510,32 @@ void loop()
 						Serial << "Missing or existing client name" << endl;
 					cmd+=" ls";
 				}
+				else if (compare(s, "set"))
+				{
+					std::string name(getword(cmd));
+					if (name.length()==0)
+					{
+						for(auto it: vars)
+						{
+							Serial << "  " << it.first << " -> " << it.second << endl;
+						}
+					}
+					else if (commands.find(name) != commands.end())
+					{
+						Serial << "Reserved keyword (" << name << ")" << endl;
+						cmd.clear();
+					}
+					else
+					{
+						if (cmd.length())
+						{
+							vars[name] = cmd;
+							cmd.clear();
+						}
+						else if (vars.find(name) != vars.end())
+							vars.erase(vars.find(name));
+					}
+				}
 				else if (compare(s, "ls") or compare(s, "view"))
 				{
 					Serial << "--< " << clients.size() << " client/s. >--" << endl;
@@ -409,9 +563,9 @@ void loop()
 					Serial << endl;
 					Serial << "  MqttClient:" << endl;
 					Serial << "    client {name} {parent broker} : create a client then" << endl;
-					Serial << "      name.connect  [ip]" << endl;
+					Serial << "      name.connect  [ip] [port] [alive]" << endl;
 					Serial << "      name.subscribe [topic]" << endl;
-					Serial << "      name.publish [topic]" << endl;
+					Serial << "      name.publish [topic][payload]" << endl;
 					Serial << "      name.view" << endl;
 					Serial << "      name.delete" << endl;
 
@@ -419,6 +573,7 @@ void loop()
 					Serial << endl;
 					Serial << "    help" << endl;
 					Serial << "    ls / ip / reset" << endl;
+					Serial << "    set [name][value]" << endl;
 					Serial << "    !  repeat last command" << endl;
 					Serial << endl;
 					Serial << "  $id : name of the client." << endl;
@@ -431,6 +586,11 @@ void loop()
 					if (s.length())
 						Serial << "Unknown command (" << s.c_str() << ")" << endl;
 				}
+
+				if (retval != MqttOk)
+				{
+					Serial << "## ERROR " << retval << endl;
+				}
 			}
 		}
 		else

library.json

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

library.properties

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

@@ -0,0 +1,412 @@
+/* MqttStreaming.h - Fork of Streaming.h adding std::string and with some minor fixes
+ * (I have to speek to the author in order to include my changes to his library if possible)
+ **/
+
+/*
+  Streaming.h - Arduino library for supporting the << streaming operator
+  Copyright (c) 2010-2012 Mikal Hart.  All rights reserved.
+
+  This library is free software; you can redistribute it and/or
+  modify it under the terms of the GNU Lesser General Public
+  License as published by the Free Software Foundation; either
+  version 2.1 of the License, or (at your option) any later version.
+
+  This library is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  Lesser General Public License for more details.
+
+  You should have received a copy of the GNU Lesser General Public
+  License along with this library; if not, write to the Free Software
+  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+*/
+
+/*
+  Version 6 library changes
+    Copyright (c) 2019 Gazoodle. All rights reserved.
+
+  1.  _BASED moved to template to remove type conversion to long and
+      sign changes which break int8_t and int16_t negative numbers. 
+      The print implementation still upscales to long for it's internal
+      print routine.
+
+  2.  _PAD added to allow padding & filling of characters to the stream
+
+  3.  _WIDTH & _WIDTHZ added to allow width printing with space padding
+      and zero padding for numerics
+
+  4.  Simple _FMT mechanism ala printf, but without the typeunsafetyness 
+      and no internal buffers for replaceable stream printing
+*/
+
+#ifndef ARDUINO_STREAMING
+#define ARDUINO_STREAMING
+
+#if defined(ARDUINO) && ARDUINO >= 100
+#include "Arduino.h"
+#else
+#ifndef STREAMING_CONSOLE
+#include "WProgram.h"
+#endif
+#endif
+
+#include <string>
+
+#if defined(ARDUINO_ARCH_AVR) || defined(ARDUINO_ARCH_MEGAAVR)
+// No stl library, so need trivial version of std::is_signed ...
+namespace std {
+template<typename T>
+  struct is_signed { static const bool value = false; };
+  template<>
+  struct is_signed<int8_t> { static const bool value = true; };
+  template<>
+  struct is_signed<int16_t> { static const bool value = true; };
+  template<>
+  struct is_signed<int32_t> { static const bool value = true; };
+};
+#else
+#include <type_traits>
+#endif
+
+#define STREAMING_LIBRARY_VERSION 6
+
+#if !defined(typeof)
+#define typeof(x) __typeof__(x)
+#endif
+
+// PrintBuffer implementation of Print, a small buffer to print in
+// see its use with pad_float()
+template <size_t N>
+class PrintBuffer : public Print
+{
+  size_t  pos = 0;
+  char    str[N] {};
+public:
+  inline const char *operator() () 
+  { return str; };
+  
+  // inline void clear() 
+  // { pos = 0; str[0] = '\0'; };
+  
+  inline size_t write(uint8_t c) 
+  { return write(&c, 1); };
+  
+  inline size_t write(const uint8_t *buffer, size_t size)
+  {
+    size_t s = std::min(size, N-1 - pos); // need a /0 left
+    if (s)
+    {
+      memcpy(&str[pos], buffer, s);
+      pos += s;
+    }
+    return s;
+  };
+};
+
+// Generic template
+template<class T>
+inline Print &operator <<(Print &stream, const T &arg)
+{ stream.print(arg); return stream; }
+
+// TODO sfinae maybe could do the trick ?
+inline Print &operator <<(Print &stream, const std::string &str)
+{ stream.print(str.c_str()); return stream; }
+
+template<typename T>
+struct _BASED
+{
+  T val;
+  int base;
+  _BASED(T v, int b): val(v), base(b)
+  {}
+};
+
+#if ARDUINO >= 100
+
+struct _BYTE_CODE
+{
+  byte val;
+  _BYTE_CODE(byte v) : val(v)
+  {}
+};
+#define _BYTE(a)    _BYTE_CODE(a)
+
+inline Print &operator <<(Print &obj, const _BYTE_CODE &arg)
+{ obj.write(arg.val); return obj; }
+
+#else
+
+#define _BYTE(a)    _BASED<typeof(a)>(a, BYTE)
+
+#endif
+
+#define _HEX(a)     _BASED<typeof(a)>(a, HEX)
+#define _DEC(a)     _BASED<typeof(a)>(a, DEC)
+#define _OCT(a)     _BASED<typeof(a)>(a, OCT)
+#define _BIN(a)     _BASED<typeof(a)>(a, BIN)
+
+// Specialization for class _BASED
+// Thanks to Arduino forum user Ben Combee who suggested this
+// clever technique to allow for expressions like
+//   Serial << _HEX(a);
+
+template<typename T>
+inline Print &operator <<(Print &obj, const _BASED<T> &arg)
+{ obj.print(arg.val, arg.base); return obj; }
+
+#if ARDUINO >= 18
+// Specialization for class _FLOAT
+// Thanks to Michael Margolis for suggesting a way
+// to accommodate Arduino 0018's floating point precision
+// feature like this:
+//   Serial << _FLOAT(gps_latitude, 6); // 6 digits of precision
+
+struct _FLOAT
+{
+  double val; // only Print::print(double)
+  int digits;
+  _FLOAT(double v, int d): val(v), digits(d)
+  {}
+};
+
+inline Print &operator <<(Print &obj, const _FLOAT &arg)
+{ obj.print(arg.val, arg.digits); return obj; }
+#endif
+
+// Specialization for enum _EndLineCode
+// Thanks to Arduino forum user Paul V. who suggested this
+// clever technique to allow for expressions like
+//   Serial << "Hello!" << endl;
+
+enum _EndLineCode { endl };
+
+inline Print &operator <<(Print &obj, _EndLineCode)
+{ obj.println(); return obj; }
+
+// Specialization for padding & filling, mainly utilized
+// by the width printers
+//
+//  Use like 
+//    Serial << _PAD(10,' ');   // Will output 10 spaces
+//    Serial << _PAD(4, '0');   // Will output 4 zeros
+struct _PAD
+{
+  int8_t width;
+  char   chr;
+  _PAD(int8_t w, char c) : width(w), chr(c) {}
+};
+
+inline Print &operator <<(Print& stm, const _PAD &arg)
+{
+  for(int8_t i = 0; i < arg.width; i++)
+    stm.print(arg.chr);
+  return stm;
+}
+
+// Specialization for width printing
+//
+//    Use like                            Result
+//    --------                            ------
+//    Serial << _WIDTH(1,5)               "    1"
+//    Serial << _WIDTH(10,5)              "   10"
+//    Serial << _WIDTH(100,5)             "  100"
+//    Serial << _WIDTHZ(1,5)              "00001"
+//
+//  Great for times & dates, or hex dumps
+//
+//    Serial << _WIDTHZ(hour,2) << ':' << _WIDTHZ(min,2) << ':' << _WIDTHZ(sec,2)
+//
+//    for(int index=0; index<byte_array_size; index++)
+//      Serial << _WIDTHZ(_HEX(byte_array[index]))
+
+template<typename T>
+struct __WIDTH
+{
+  const T val;
+  int8_t width;
+  char pad;
+  __WIDTH(const T& v, int8_t w, char p) : val(v), width(w), pad(p) {}
+};
+
+//  Count digits in an integer of specific base
+template<typename T>
+inline uint8_t digits(T v, int8_t base = 10)
+{
+  uint8_t digits = 0;
+  if ( std::is_signed<T>::value )
+  {
+    if ( v < 0 )
+    {
+      digits++;
+      v = -v; // v needs to be postive for the digits counter to work
+    }
+  }
+  do
+  {
+    v /= base;
+    digits++;
+  } while( v > 0 );
+  return digits;
+}
+
+// Generic get the width of a value in base 10
+template<typename T>
+inline uint8_t get_value_width(T val)
+{ return digits(val); }
+
+inline uint8_t get_value_width(const char * val)
+{ return strlen(val); }
+
+#ifdef ARDUINO
+inline uint8_t get_value_width(const __FlashStringHelper * val)
+{ return strlen_P(reinterpret_cast<const char *>(val)); }
+#endif
+
+// _BASED<T> get the width of a value
+template<typename T>
+inline uint8_t get_value_width(_BASED<T> b)
+{ return digits(b.val, b.base); }
+
+// Constructor wrapper to allow automatic template parameter deduction
+template<typename T>
+__WIDTH<T> _WIDTH(T val, int8_t width) { return __WIDTH<T>(val, width, ' '); }
+template<typename T>
+__WIDTH<T> _WIDTHZ(T val, int8_t width) { return __WIDTH<T>(val, width, '0'); }
+
+
+// Operator overload to handle width printing.
+template<typename T>
+inline Print &operator <<(Print &stm, const __WIDTH<T> &arg)
+{ stm << _PAD(arg.width - get_value_width(arg.val), arg.pad) << arg.val; return stm; }
+
+// explicit Operator overload to handle width printing of _FLOAT, double and float
+template<typename T>
+inline Print &pad_float(Print &stm, const __WIDTH<T> &arg, const double val, const int digits = 2) // see Print::print(double, int = 2)
+{
+  PrintBuffer<32> buf; // it's only ~45B on the stack, no allocation, leak or fragmentation
+  size_t size = buf.print(val, digits); // print in buf
+  return stm << _PAD(arg.width - size, arg.pad) << buf(); // pad and concat what's in buf
+}
+
+inline Print &operator <<(Print &stm, const __WIDTH<float>  &arg) 
+{ return pad_float(stm, arg, arg.val); }
+
+inline Print &operator <<(Print &stm, const __WIDTH<double> &arg) 
+{ return pad_float(stm, arg, arg.val); } 
+
+inline Print &operator <<(Print &stm, const __WIDTH<_FLOAT> &arg) 
+{ auto& f = arg.val; return pad_float(stm, arg, f.val, f.digits); }
+
+// a less verbose _FLOATW for _WIDTH(_FLOAT)
+#define _FLOATW(val, digits, width) _WIDTH<_FLOAT>(_FLOAT((val), (digits)), (width))
+
+// Specialization for replacement formatting
+//
+//  Designed to be similar to printf that everyone knows and loves/hates. But without
+//  the internal buffers and type agnosticism. This version only has placeholders in
+//  the format string, the actual values are supplied using the stream safe operators
+//  defined in this library.
+//
+//  Use like this:
+//
+//      Serial << FMT(F("Replace % with %"), 1, 2 )
+//      Serial << FMT("Time is %:%:%", _WIDTHZ(hours,2), _WIDTHZ(minutes,2), _WIDTHZ(seconds,2))
+//      Serial << FMT("Your score is %\\%", score);   // Note the \\ to escape the % sign
+
+// Ok, hold your hats. This is a foray into C++11's variadic template engine ...
+
+inline char get_next_format_char(const char *& format_string)
+{
+  char format_char = *format_string;
+  if ( format_char > 0 ) format_string++;
+  return format_char;
+}
+
+#ifdef ARDUINO
+inline char get_next_format_char(const __FlashStringHelper*& format_string)
+{
+  char format_char = pgm_read_byte(format_string);
+  if ( format_char > 0 ) format_string = reinterpret_cast<const __FlashStringHelper*>(reinterpret_cast<const char *>(format_string)+1);
+  return format_char;
+}
+#endif
+
+template<typename Ft>
+inline bool check_backslash(char& format_char, Ft& format_string)
+{
+  if ( format_char == '\\')
+  {
+    format_char = get_next_format_char(format_string);
+    return true;
+  }
+  return false;
+}
+
+// The template tail printer helper
+template<typename Ft, typename... Ts>
+struct __FMT
+{
+  Ft format_string;
+  __FMT(Ft f, Ts ... args) : format_string(f) {}
+  inline void tstreamf(Print& stm, Ft format) const
+  {
+    while(char c = get_next_format_char(format))
+    {
+      check_backslash(c, format);
+      if ( c )
+        stm.print(c);
+    }
+  }
+};
+
+// The variadic template helper
+template<typename Ft, typename T, typename... Ts>
+struct __FMT<Ft, T, Ts...> : __FMT<Ft, Ts...>
+{
+  T val;
+  __FMT(Ft f, T t, Ts... ts) : __FMT<Ft, Ts...>(f, ts...), val(t) {}
+  inline void tstreamf(Print& stm, Ft format) const
+  {
+    while(char c = get_next_format_char(format))
+    {
+      if (!check_backslash(c, format))
+      {
+        if ( c == '%')
+        {
+          stm << val;
+          // Variadic recursion ... compiler rolls this out during 
+          // template argument pack expansion
+          __FMT<Ft, Ts...>::tstreamf(stm, format);
+          return;
+        }
+      }
+      if (c)
+        stm.print(c);
+    }
+  }
+};
+
+// The actual operator should you only instanciate the FMT 
+// helper with a format string and no parameters
+template<typename Ft, typename... Ts>
+inline Print& operator <<(Print &stm, const __FMT<Ft, Ts...> &args)
+{
+    args.tstreamf(stm, args.format_string);
+    return stm;
+}
+
+// The variadic stream helper
+template<typename Ft, typename T, typename... Ts>
+inline Print& operator <<(Print &stm, const __FMT<Ft, T, Ts...> &args)
+{
+    args.tstreamf(stm, args.format_string);
+    return stm;
+}
+
+// As we don't have C++17, we can't get a constructor to use
+// automatic argument deduction, but ... this little trick gets
+// around that ...
+template<typename Ft, typename... Ts>
+__FMT<Ft, Ts...> _FMT(Ft format, Ts ... args) { return __FMT<Ft, Ts...>(format, args...); }
+
+#endif

src/StringIndexer.h

@@ -2,7 +2,6 @@
 #include <map>
 #include <string>
 #include <string.h>
-// #include <Streaming.h>
 #include <ESP8266WiFi.h>
 
 /***
@@ -105,7 +104,7 @@ class IndexedString
 
 		const std::string& str() const { return StringIndexer::str(index); }
 
-		const StringIndexer::index_t getIndex() const { return index; }
+		const StringIndexer::index_t& getIndex() const { return index; }
 
 	private:
 		StringIndexer::index_t index;

src/TinyMqtt.cpp

@@ -1,6 +1,5 @@
 #include "TinyMqtt.h"
 #include <sstream>
-#include <Streaming.h>
 
 void outstring(const char* prefix, const char*p, uint16_t len)
 {
@@ -42,14 +41,20 @@ MqttClient::~MqttClient()
 {
 	close();
 	delete client;
+	Serial << "Client deleted" << endl;
 }
 
-void MqttClient::close()
+void MqttClient::close(bool bSendDisconnect)
 {
 	debug("close " << id().c_str());
 	mqtt_connected = false;
 	if (client)
 	{
+		if (bSendDisconnect and client->connected())
+		{
+			message.create(MqttMessage::Type::Disconnect);
+			message.sendTo(this);
+		}
 		client->stop();
 	}
 
@@ -60,7 +65,7 @@ void MqttClient::close()
 	}
 }
 
-void MqttClient::connect(std::string broker, uint16_t port)
+void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
 {
 	debug("cnx: closing");
 	close();
@@ -70,18 +75,20 @@ void MqttClient::connect(std::string broker, uint16_t port)
 	if (client->connect(broker.c_str(), port))
 	{
 	  debug("cnx: connecting");
-		message.create(MqttMessage::Type::Connect);
+		MqttMessage msg(MqttMessage::Type::Connect);
-		message.add("MQTT",4);
+		msg.add("MQTT",4);
-		message.add(0x4);	// Mqtt protocol version 3.1.1
+		msg.add(0x4);	// Mqtt protocol version 3.1.1
-		message.add(0x0);	// Connect flags         TODO user / name
+		msg.add(0x0);	// Connect flags         TODO user / name
 
-		keep_alive = 1;			// TODO not configurable
+		keep_alive = ka;
-		message.add(0x00); // keep_alive
+		msg.add(0x00);			// keep_alive
-		message.add((char)keep_alive);
+		msg.add((char)keep_alive);
-		message.add(clientId);
+		msg.add(clientId);
 	  debug("cnx: mqtt connecting");
-		message.sendTo(this);
+		msg.sendTo(this);
+		msg.reset();
 	  debug("cnx: mqtt sent " << (int32_t)parent);
+
 		clientAlive(0);
 	}
 }
@@ -119,7 +126,7 @@ void MqttBroker::loop()
 
   // for(auto it=clients.begin(); it!=clients.end(); it++)
 	// use index because size can change during the loop
-	for(int i=0; i<clients.size(); i++)
+	for(size_t i=0; i<clients.size(); i++)
 	{
 		auto client = clients[i];
     if (client->connected())
@@ -145,26 +152,33 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, Mqtt
 	for(auto client: clients)
 	{
 		i++;
+#if TINY_MQTT_DEBUG
 		Serial << "brk_" << (broker && broker->connected() ? "con" : "dis") <<
-			 "	srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << i << ", local=" << client->isLocal() << ", con=" << client->connected();
+			 "	srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << i << ", local=" << client->isLocal() << ", con=" << client->connected() << endl;
+#endif
 		bool doit = false;
-		if (broker && broker->connected())	// Connected: R2 R3 R5 R6
+		if (broker && broker->connected())	// Broker is connected
 		{
 			//    ext broker -> clients or
 			// or clients -> ext broker
 			if (source == broker)	// broker -> clients
 				doit = true;
 			else									// clients -> broker
-				retval=broker->publish(topic, msg);
+			{
+				MqttError ret = broker->publish(topic, msg);
+				if (ret != MqttOk) retval = ret;
+			}
 		}
 		else // Disconnected: R7
 		{
 			// All is allowed
 			doit = true;
 		}
+#if TINY_MQTT_DEBUG
 		Serial << ", doit=" << doit << ' ';
+#endif
 
-		if (doit) client->publish(topic, msg);
+		if (doit) retval = client->publish(topic, msg);
 		debug("");
 	}
 	return retval;
@@ -208,6 +222,7 @@ void MqttClient::loop()
 		}
 		else if (client && client->connected())
 		{
+			debug("pingreq");
 			uint16_t pingreq = MqttMessage::Type::PingReq;
 			client->write((uint8_t*)(&pingreq), 2);
 			clientAlive(0);
@@ -227,11 +242,63 @@ void MqttClient::loop()
 	}
 }
 
+void MqttClient::resubscribe()
+{
+	// TODO resubscription limited to 256 bytes
+	if (subscriptions.size())
+	{
+		MqttMessage msg(MqttMessage::Type::Subscribe, 2);
+
+		// TODO manage packet identifier
+		msg.add(0);
+		msg.add(0);
+
+		for(auto topic: subscriptions)
+		{
+			msg.add(topic);
+			msg.add(0);		// TODO qos
+		}
+		msg.sendTo(this);	// TODO return value
+	}
+}
+
+MqttError MqttClient::subscribe(Topic topic, uint8_t qos)
+{
+	debug("subsribe(" << topic.c_str() << ")");
+	MqttError ret = MqttOk;
+
+ 	subscriptions.insert(topic);
+
+	if (parent==nullptr) // remote broker ?
+	{
+		debug("remote subscribe");
+		MqttMessage msg(MqttMessage::Type::Subscribe, 2);
+
+		// TODO manage packet identifier
+		msg.add(0);
+		msg.add(0);
+		
+		msg.add(topic);
+		msg.add(qos);
+		ret = msg.sendTo(this);
+		
+		// TODO we should wait (state machine) for SUBACK
+	}
+	return ret;
+}
+
+long MqttClient::counter=0;
+
 void MqttClient::processMessage()
 {
-	std::string error;
+	counter++;
-	std::string s;
+#if TINY_MQTT_DEBUG
-	// Serial << "---> INCOMING " << _HEX(message.type()) << ", mem=" << ESP.getFreeHeap() << endl;
+if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessage::Type::PingResp)
+{
+	Serial << "---> INCOMING " << _HEX(message.type()) << " client(" << (int)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
+	// message.hexdump("Incoming");
+}
+#endif
   auto header = message.getVHeader();
   const char* payload;
   uint16_t len;
@@ -261,20 +328,6 @@ void MqttClient::processMessage()
 
 			// ClientId
 			message.getString(payload, len);
-			debug("client id len=" << len);
-			if (len>30)
-			{
-				Serial << '(';
-				for(int i=0; i<30; i++)
-				{
-					if (i%5==0) Serial << ' ';
-					char c=*(header+i);
-					Serial << (c < 32 ? '.' : c);
-				}
-				Serial << " )" << endl;
-				debug("Bad client id length");
-				break;
-			}
 			clientId = std::string(payload, len);
 			payload += len;
 
@@ -305,11 +358,30 @@ void MqttClient::processMessage()
 			Serial << "Connected client:" << clientId.c_str() << ", keep alive=" << keep_alive << '.' << endl;
 			bclose = false;
 			mqtt_connected=true;
-			// Reuse received msg
+			{
-			message.create(MqttMessage::Type::Connack);
+				MqttMessage msg(MqttMessage::Type::ConnAck);
-			message.add(0);	// Session present (not implemented)
+				msg.add(0);	// Session present (not implemented)
-			message.add(0); // Connection accepted
+				msg.add(0); // Connection accepted
-			message.sendTo(this);
+				msg.sendTo(this);
+			}
+			break;
+
+		case MqttMessage::Type::ConnAck:
+			mqtt_connected = true;
+			bclose = false;
+			resubscribe();
+			break;
+
+		case MqttMessage::Type::SubAck:
+		case MqttMessage::Type::PubAck:
+			if (!mqtt_connected) break;
+			// Ignore acks
+			bclose = false;
+			break;
+
+		case MqttMessage::Type::PingResp:
+			// TODO: no PingResp is suspicious (server dead)
+			bclose = false;
 			break;
 
 		case MqttMessage::Type::PingReq:
@@ -327,14 +399,35 @@ void MqttClient::processMessage()
 			break;
 
 		case MqttMessage::Type::Subscribe:
-			if (!mqtt_connected) break;
+		case MqttMessage::Type::UnSubscribe:
-			payload = header+2;
+			{
-			message.getString(payload, len);	// Topic
+				if (!mqtt_connected) break;
-			outstring("Subscribes", payload, len);
+				payload = header+2;
 				
-			subscribe(Topic(payload, len));
+				debug("subscribe loop");
-			bclose = false;
+				while(payload < message.end())
-			// TODO SUBACK
+				{
+					message.getString(payload, len);	// Topic
+					debug( "  topic (" << std::string(payload, len) << ')');
+					outstring("Subscribes", payload, len);
+					// subscribe(Topic(payload, len));
+					Topic topic(payload, len);
+					if ((message.type() & 0XF0) == MqttMessage::Type::Subscribe)
+						subscriptions.insert(topic);
+					else
+					{
+						auto it=subscriptions.find(topic);
+						if (it != subscriptions.end())
+							subscriptions.erase(it);
+					}
+					payload += len;
+					uint8_t qos = *payload++;
+					debug("  qos=" << qos);
+				}
+				debug("end loop");
+				bclose = false;
+				// TODO SUBACK
+			}
 			break;
 
 		case MqttMessage::Type::Publish:
@@ -351,16 +444,28 @@ void MqttClient::processMessage()
 				if (qos) payload+=2;	// ignore packet identifier if any
 				// TODO reset DUP
 				// TODO reset RETAIN
-				debug("publishing to parent");
+				if (parent)
-				parent->publish(this, published, message);
+				{
-				// TODO should send PUBACK
+				  debug("publishing to parent");
+					parent->publish(this, published, message);
+				}
+				else if (callback && subscriptions.find(published)!=subscriptions.end())
+				{
+					callback(this, published, nullptr, 0);	// TODO send the real payload
+				}
+				message.create(MqttMessage::Type::PubAck);
+				// TODO re-add packet identifier if any
+				message.sendTo(this);
 				bclose = false;
 			}
 			break;
 
-		case MqttMessage::Type::PubAck:
+		case MqttMessage::Type::Disconnect:
+			// TODO should discard any will message
 			if (!mqtt_connected) break;
-			bclose = false;
+			mqtt_connected = false;
+			close(false);
+			bclose=false;
 			break;
 
 		default:
@@ -370,13 +475,13 @@ void MqttClient::processMessage()
 	if (bclose)
 	{
 		Serial << "*************** Error msg 0x" << _HEX(message.type());
-		if (error.length()) Serial << ':' << error.c_str();
+		message.hexdump("-------ERROR ------");
 		Serial << endl;
 		close();
   }
 	else
 	{
-		clientAlive(5);
+		clientAlive(parent ? 5 : 0);
 	}
 	message.reset();
 }
@@ -391,14 +496,13 @@ bool Topic::matches(const Topic& topic) const
 // publish from local client
 MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pay_length)
 {
-	MqttMessage msg;
+	MqttMessage msg(MqttMessage::Publish);
-	msg.create(MqttMessage::Publish);
 	msg.add(topic);
 	msg.add(payload, pay_length, false);
 	if (parent)
 		return parent->publish(this, topic, msg);
 	else if (client)
-		msg.sendTo(this);
+		return msg.sendTo(this);
 	else
 		return MqttNowhereToSend;
 }
@@ -411,20 +515,18 @@ MqttError MqttClient::publish(const Topic& topic, MqttMessage& msg)
 	debug("mqttclient publish " << subscriptions.size());
 	for(const auto& subscription: subscriptions)
 	{
-		Serial << " client=" << (int32_t)client << ", topic " << topic.str().c_str() << ' ';
 		if (subscription.matches(topic))
 		{
-			Serial << " match/send";
+			debug(" match client=" << (int32_t)client << ", topic " << topic.str().c_str() << ' ');
 			if (client)
 			{
-				msg.sendTo(this);
+				retval = msg.sendTo(this);
 			}
 			else if (callback)
 			{
 				callback(this, topic, nullptr, 0);	// TODO Payload
 			}
 		}
-		Serial << endl;
 	}
 	return retval;
 }
@@ -456,8 +558,15 @@ void MqttMessage::incoming(char in_byte)
 				vheader = buffer.length();
 				if (size==0)
 					state = Complete;
+				else if (size > 500)	// TODO magic
+				{
+					state = Error;
+				}
 				else
+				{
+					buffer.reserve(size);
 					state = VariableHeader;
+				}
 			}
 			break;
 		case VariableHeader:
@@ -475,6 +584,7 @@ void MqttMessage::incoming(char in_byte)
 		case Complete:
 		default:
 			Serial << "Spurious " << _HEX(in_byte) << endl;
+			hexdump("spurious");
 			reset();
 			break;
 	}
@@ -506,28 +616,58 @@ void MqttMessage::encodeLength(char* msb, int length)
 	} while (length);
 };
 
-void MqttMessage::sendTo(MqttClient* client)
+MqttError MqttMessage::sendTo(MqttClient* client)
 {
-	if (buffer.size()>2)
+	if (buffer.size())
 	{
+		debug("sending " << buffer.size() << " bytes");
 		encodeLength(&buffer[1], buffer.size()-2);
 		// hexdump("snd");
 		client->write(&buffer[0], buffer.size());
 	}
 	else
 	{
-		Serial << "??? Invalid send" << endl;
+		debug("??? Invalid send");
+		return MqttInvalidMessage;
 	}
+	return MqttOk;
 }
 
 void MqttMessage::hexdump(const char* prefix) const
 {
-	if (prefix) Serial << prefix << ' ';
+	uint16_t addr=0;
-	Serial << "size(" << buffer.size() << ") : ";
+	const int bytes_per_row = 8;
-		for(const char chr: buffer)
+	const char* hex_to_str = " | ";
+	const char* separator = hex_to_str;
+	const char* half_sep = " - ";
+	std::string ascii;
+
+	Serial << prefix << " size(" << buffer.size() << "), state=" << state << endl;
+
+	for(const char chr: buffer)
+	{
+		if ((addr % bytes_per_row) == 0)
 		{
-			if (chr<16) Serial << '0';
+			if (ascii.length()) Serial << hex_to_str << ascii << separator << endl;
-			Serial << _HEX(chr) << ' ';
+			if (prefix) Serial << prefix << separator;
+			ascii.clear();
 		}
-		Serial << endl;
+		addr++;
+		if (chr<16) Serial << '0';
+		Serial << _HEX(chr) << ' ';
+
+		ascii += (chr<32 ? '.' : chr);
+		if (ascii.length() == (bytes_per_row/2)) ascii += half_sep;
+	}
+	if (ascii.length())
+	{
+		while(ascii.length() < bytes_per_row+strlen(half_sep))
+		{
+			Serial << "   ";	// spaces per hexa byte
+			ascii += ' ';
+		}
+		Serial << hex_to_str << ascii << separator;
+	}
+
+	Serial << endl;
 }

src/TinyMqtt.h

@@ -3,11 +3,13 @@
 #include <set>
 #include <string>
 #include "StringIndexer.h"
+#include <MqttStreaming.h>
 
-#ifdef TINY_MQTT_DEBUG
+#if 0
-  #include <Streaming.h>
   #define debug(what) { Serial << __LINE__ << ' ' << what << endl; delay(100); }
+  #define TINY_MQTT_DEBUG 1
 #else
+  #define TINY_MQTT_DEBUG 0
   #define debug(what) {}
 #endif
 
@@ -15,6 +17,7 @@ enum MqttError
 {
 	MqttOk = 0,
 	MqttNowhereToSend=1,
+	MqttInvalidMessage=2,
 };
 
 class Topic : public IndexedString
@@ -36,14 +39,17 @@ class MqttMessage
 	public:
 		enum Type
 		{
-			Unknown   =    0,
+			Unknown     =    0,
-			Connect   = 0x10,
+			Connect     = 0x10,
-			Connack   = 0x20,
+			ConnAck     = 0x20,
-			Publish   = 0x30,
+			Publish     = 0x30,
-			PubAck    = 0x40,
+			PubAck      = 0x40,
-			Subscribe = 0x80,
+			Subscribe   = 0x80,
-			PingReq   = 0xC0,
+			SubAck      = 0x90,
-			PingResp  = 0xD0,
+			UnSubscribe = 0xA0,
+			PingReq     = 0xC0,
+			PingResp    = 0xD0,
+			Disconnect  = 0xE0
 		};
 		enum State
 		{
@@ -57,7 +63,7 @@ class MqttMessage
 		};
 
 		MqttMessage() { reset(); }
-		MqttMessage(Type t) { create(t); }
+		MqttMessage(Type t, uint8_t bits_d3_d0=0) { create(t); buffer[0] |= bits_d3_d0; }
 		void incoming(char byte);
 		void add(char byte) { incoming(byte); }
 		void add(const char* p, size_t len, bool addLength=true );
@@ -87,7 +93,7 @@ class MqttMessage
 			size=0;
 			state=Create;
 		}
-		void sendTo(MqttClient*);
+		MqttError sendTo(MqttClient*);
 		void hexdump(const char* prefix=nullptr) const;
 
 	private:
@@ -115,12 +121,13 @@ class MqttClient
 	};
 	public:
 		MqttClient(MqttBroker*);
+		MqttClient(MqttBroker* brk, const std::string& id) : MqttClient(brk) { clientId=id; }
 		MqttClient() : MqttClient(nullptr) {};
 
 		~MqttClient();
 
 		void connect(MqttBroker* parent);
-		void connect(std::string broker, uint16_t port);
+		void connect(std::string broker, uint16_t port, uint16_t ka=10);
 
 		bool connected() { return
 			(parent!=nullptr and client==nullptr) or
@@ -132,7 +139,7 @@ class MqttClient
 		void id(std::string& new_id) { clientId = new_id; }
 
 		void loop();
-		void close();
+		void close(bool bSendDisconnect=true);
 		void setCallback(CallBack fun) {callback=fun; };
 
 		// Publish from client to the world
@@ -141,30 +148,38 @@ class MqttClient
 		MqttError publish(const Topic& t, const std::string& s) { return publish(t,s.c_str(),s.length());}
 		MqttError publish(const Topic& t) { return publish(t, nullptr, 0);};
 
-		void subscribe(Topic topic) { subscriptions.insert(topic); }
+		MqttError subscribe(Topic topic, uint8_t qos=0);
-		void unsubscribe(Topic& topic);
+		MqttError unsubscribe(Topic& topic);
 
 		// connected to local broker
 		// TODO seems to be useless
 		bool isLocal() const { return client == nullptr; }
 
-#ifdef TINY_MQTT_DEBUG
 		void dump()
 		{
+			uint32_t ms=millis();
 			Serial << "MqttClient (" << clientId.c_str() << ") p=" << (int32_t) parent
 				<< " c=" << (int32_t)client << (connected() ? " ON " : " OFF"); 
-			Serial << " [";
+			Serial << ", alive=" << (uint32_t)alive << '/' << ms << ", ka=" << keep_alive;
+			Serial << (client && client->connected() ? "" : "dis") << "connected";
+		  message.hexdump("entrant msg");
 			bool c=false;
+			Serial << " [";
 			for(auto s: subscriptions)
 			{
 				Serial << (c?", ": "")<< s.str().c_str();
 				c=true;
 			}
+
+			
 			Serial << "]" << endl;
 		}
-#endif
+
+		static long counter;	// Number of messages sent
 
 	private:
+		void resubscribe();
+
 		friend class MqttBroker;
 		MqttClient(MqttBroker* parent, WiFiClient& client);
 		// republish a received publish if topic matches any in subscriptions
@@ -211,7 +226,8 @@ class MqttBroker
 		void connect(std::string host, uint32_t port=1883);
 		bool connected() const { return state == Connected; }
 
-#ifdef TINY_MQTT_DEBUG
+		size_t clientsCount() const { return clients.size(); }
+
 		void dump()
 		{
 			Serial << clients.size() << " client/s" << endl;
@@ -221,7 +237,6 @@ class MqttBroker
 				client->dump();
 			}
 		}
-#endif
 
 	private:
 		friend class MqttClient;

tests/tinymqtt-tests/tinymqtt-tests.ino

@@ -0,0 +1,138 @@
+#include <AUnit.h>
+#include <TinyMqtt.h>
+#include <map>
+
+/**
+  * TinyMqtt local unit tests.
+	*
+	* No wifi connection unit tests.
+	**/
+
+using namespace std;
+
+MqttBroker broker(1883);
+
+std::map<std::string, std::map<Topic, int>>	published;		// map[client_id] => map[topic] = count
+
+void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
+{
+	if (srce)
+		published[srce->id()][topic]++;
+}
+
+test(local_client_should_unregister_when_destroyed)
+{
+	assertEqual(broker.clientsCount(), (size_t)0);
+	{
+		MqttClient client(&broker);
+		assertEqual(broker.clientsCount(), (size_t)1);
+	}
+	assertEqual(broker.clientsCount(), (size_t)0);
+}
+
+test(local_connect)
+{
+	assertEqual(broker.clientsCount(), (size_t)0);
+
+	MqttClient client(&broker);
+	assertTrue(client.connected());
+	assertEqual(broker.clientsCount(), (size_t)1);
+}
+
+test(local_publish_should_be_dispatched)
+{
+	published.clear();
+	assertEqual(broker.clientsCount(), (size_t)0);
+
+	MqttClient subscriber(&broker);
+	subscriber.subscribe("a/b");
+	subscriber.subscribe("a/c");
+	subscriber.setCallback(onPublish);
+
+	MqttClient publisher(&broker);
+	publisher.publish("a/b");
+	publisher.publish("a/c");
+	publisher.publish("a/c");
+
+	assertEqual(published.size(), (size_t)1);	// 1 client has received something
+	assertTrue(published[""]["a/b"] == 1);
+	assertTrue(published[""]["a/c"] == 2);
+}
+
+test(local_publish_should_be_dispatched_to_local_clients)
+{
+	published.clear();
+	assertEqual(broker.clientsCount(), (size_t)0);
+
+	MqttClient subscriber_a(&broker, "A");
+	subscriber_a.setCallback(onPublish);
+	subscriber_a.subscribe("a/b");
+	subscriber_a.subscribe("a/c");
+
+	MqttClient subscriber_b(&broker, "B");
+	subscriber_b.setCallback(onPublish);
+	subscriber_b.subscribe("a/b");
+
+	MqttClient publisher(&broker);
+	publisher.publish("a/b");
+	publisher.publish("a/c");
+
+	assertEqual(published.size(), (size_t)2);	// 2 clients have received something
+	assertTrue(published["A"]["a/b"] == 1);
+	assertTrue(published["A"]["a/c"] == 1);
+	assertTrue(published["B"]["a/b"] == 1);
+	assertTrue(published["B"]["a/c"] == 0);
+}
+
+test(local_unsubscribe)
+{
+	published.clear();
+	assertEqual(broker.clientsCount(), (size_t)0);
+
+	MqttClient subscriber(&broker);
+	subscriber.setCallback(onPublish);
+	subscriber.subscribe("a/b");
+
+	MqttClient publisher(&broker);
+	publisher.publish("a/b");
+
+	// subscriber.unsubscribe("a/b");		TODO not yet implemented
+
+	publisher.publish("a/b");
+
+	assertTrue(published[""]["a/b"] == 1);	// Only one publish has been received
+}
+
+test(local_nocallback_when_destroyed)
+{
+	published.clear();
+	assertEqual(broker.clientsCount(), (size_t)0);
+
+	{
+		MqttClient subscriber(&broker);
+		subscriber.setCallback(onPublish);
+		subscriber.subscribe("a/b");
+	}
+
+	MqttClient publisher(&broker);
+	publisher.publish("a/b");
+
+	assertEqual(published.size(), (size_t)0);	// Only one publish has been received
+}
+
+
+//----------------------------------------------------------------------------
+// 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();
+}