Very promising async commit

AsyncTcp

.github/workflows/aunit.yml

@@ -0,0 +1,28 @@
+# See https://docs.github.com/en/actions/guides for documentation about GitHub
+# Actions.
+
+name: AUnit Tests
+
+# Run on all branches.
+on: [push]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-18.04
+
+    steps:
+    - uses: actions/checkout@v2
+
+    - name: Setup
+      run: |
+        cd ..
+        git clone https://github.com/bxparks/EpoxyDuino
+        git clone https://github.com/bxparks/AceRoutine
+        git clone https://github.com/bxparks/AUnit
+        git clone https://github.com/bxparks/AceCommon
+        git clone https://github.com/hsaturn/EspMock
+    - name: Verify tests
+      run: |
+        make -C tests
+        make -C tests runtests

.github/workflows/superlinter.yml

@@ -0,0 +1,26 @@
+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,2 +0,0 @@
-*~
-src/my_credentials.h

README.md

@@ -1,15 +1,18 @@
 # TinyMqtt
 
-![](https://img.shields.io/github/v/release/hsaturn/TinyMqtt)
+![Release](https://img.shields.io/github/v/release/hsaturn/TinyMqtt)
-![](https://img.shields.io/github/issues/hsaturn/TinyMqtt)
+![Issues](https://img.shields.io/github/issues/hsaturn/TinyMqtt)
-![](https://img.shields.io/badge/paltform-ESP8266-green)
+![Esp8266](https://img.shields.io/badge/platform-ESP8266-green)
-![](https://img.shields.io/github/license/hsaturn/TinyMqtt)
+![Gpl 3.0](https://img.shields.io/github/license/hsaturn/TinyMqtt)
-![](https://img.shields.io/badge/Mqtt-%203.1.1-yellow)
+![Mqtt 3.1.1](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)
@@ -18,29 +21,16 @@ ESP 8266 is a small and very capable Mqtt Broker and Client
 - zeroconf, this is a strange but very powerful mode where
   all brokers tries to connect together on the same local network.
 
-## TODO List
-* Use [Async library](https://github.com/me-no-dev/ESPAsyncTCP)
-* 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, 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.~~
-* ~~MqttClient does not support more than one subscription at time~~
-* MqttClient auto re-subscribe
-* MqttClient auto reconnection
-* MqttClient does not callback payload...
-* MqttClient user/password
-
 ## Quickstart
 
 * install [TinyMqtt library](https://github.com/hsaturn/TinyMqtt)
+  (you can use the Arduino library manager and search for TinyMqtt)
 * modify <libraries/TinyMqtt/src/my_credentials.h> (wifi setup)
 
 ## Examples
 
-
 | Example             | Description                                |
-| ---------------------------- | --------------------------------- |
+| ------------------- | ------------------------------------------ |
 | client-without-wifi | standalone example                         |
 | simple-client       | Connect the ESP to an external Mqtt broker |
 | simple-broker       | Simple Mqtt broker with your ESP           |
@@ -51,11 +41,25 @@ no need for having tons of clients (also RAM is the problem with many clients)
 
 ## Standalone mode (zeroconf)
 -> The zeroconf mode is not yet implemented
-zerofonf clients to connect to broker on local network.
+zeroconf clients to connect to broker on local network.
 
 In Zeroconf mode, each ESP is a a broker and scans the local network.
 After a while one ESP naturally becomes a 'master' and all ESP are connected together.
 No problem if the master dies, a new master will be choosen soon.
 
+## TODO List
+* Use [Async library](https://github.com/me-no-dev/ESPAsyncTCP)
+* 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)
+* Why not a 'global' TinyMqtt::loop() instead of having to call loop for all broker/clients instances
+* 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.
+* ~~MqttClient auto re-subscribe (::resubscribe works bad on broker.emqx.io)~~
+* MqttClient auto reconnection
+* MqttClient user/password
+* Wildcards (I may implement only # as I'm not interrested by a clever and cpu consuming matching)
+* I suspect that MqttClient::parent could be removed and replaced with a simple boolean
+  (this'll need to rewrite a few functions)
+
 ## License
 Gnu GPL 3.0, see [LICENSE](https://github.com/hsaturn/TinyMqtt/blob/main/LICENSE).

examples/tinymqtt-test/commands.txt

@@ -1,6 +1,15 @@
-// vim: ts=30
+// 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
@@ -8,4 +17,23 @@ 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,8 @@
-#define TINY_MQTT_DEBUG
 #include <TinyMqtt.h>       // https://github.com/hsaturn/TinyMqtt
 #include <MqttStreaming.h>
+#include <ESP8266mDNS.h>
+
+#include <sstream>
 #include <map>
 
 /**
@@ -15,11 +17,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 +24,27 @@
 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()
 {
+	WiFi.persistent(false); // https://github.com/esp8266/Arduino/issues/1054
   Serial.begin(115200);
 	delay(500);
 	Serial << endl << endl << endl
-	<< "Demo started. Type help for more..." << endl
 	<< "Connecting to '" << ssid << "' ";
 
   WiFi.mode(WIFI_STA);
@@ -48,6 +54,15 @@ void setup()
 	{ Serial << '-'; delay(500); }
 
   Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
+	Serial << "Type help for more..." << endl;
+
+  const char* name="tinytest";
+  Serial << "Starting MDNS, name= " << name;
+  if (!MDNS.begin(name))
+    Serial << "  error, not available." << endl;
+  else
+    Serial << " ok." << endl;
+
 
 	MqttBroker* broker = new MqttBroker(1883);
 	broker->begin();
@@ -78,6 +93,110 @@ 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", "blink", "client", "connect",
+	"create", "delete", "help", "interval",
+	"ls", "ip", "off", "on", "set",
+	"publish", "reset", "subscribe", "unsubscribe", "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
 {
@@ -180,6 +299,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;
 		}
 
@@ -202,8 +322,44 @@ bool compare(std::string s, const char* cmd)
 
 using ClientFunction = void(*)(std::string& cmd, MqttClient* publish);
 
+uint32_t blink_ms_on[16];
+uint32_t blink_ms_off[16];
+uint32_t blink_next[16];
+bool blink_state[16];
+int16_t blink;
 void loop()
 {
+	auto ms=millis();
+	int8_t out=1;
+	int16_t blink_bits = blink;
+	while(blink_bits)
+	{
+		if (blink_ms_on[out] and ms > blink_next[out])
+		{
+			if (blink_state[out])
+			{
+				blink_next[out] += blink_ms_on[out];
+				digitalWrite(out, LOW);
+			}
+			else
+			{
+				blink_next[out] += blink_ms_off[out];
+				digitalWrite(abs(out), HIGH);
+			}
+			blink_state[out] = not blink_state[out];
+		}
+		blink_bits >>=1;
+		out++;
+	}
+
+	static long count;
+  MDNS.update();
+
+	if (MqttClient::counter != count)
+	{
+		Serial << "# " << MqttClient::counter << endl;
+		count = MqttClient::counter;
+	}
 	for(auto it: brokers)
 		it.second->loop();
 
@@ -219,12 +375,15 @@ 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;
@@ -235,10 +394,13 @@ void loop()
 
 				// 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 (s.length())
+					{
 						if (clients.find(s) != clients.end())
 						{
 							client = clients[s];
@@ -253,9 +415,13 @@ void loop()
 							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)
 					{
@@ -273,11 +439,11 @@ void loop()
 					}
 					if (client)
 					{
-						clients.erase(s);
 						for (auto it: clients)
 						{
 							if (it.second != client) continue;
 							Serial << "deleted" << endl;
+							delete (it.second);
 							clients.erase(it.first);
 							break;
 						}
@@ -287,9 +453,9 @@ void loop()
 					{
 						for(auto it: brokers)
 						{
-							Serial << (int32_t)it.second << '/' << (int32_t)broker << endl;
 							if (broker != it.second) continue;
 							Serial << "deleted" << endl;
+							delete (it.second);
 							brokers.erase(it.first);
 							break;
 						}
@@ -313,7 +479,7 @@ void loop()
 				{
 					if (compare(s,"connect"))
 					{
-						client->connect(getword(cmd,"192.168.1.40").c_str(), getint(cmd, 1883), getint(cmd, 60));
+						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"))
@@ -326,11 +492,33 @@ void loop()
 					{
 						client->subscribe(getword(cmd, topic.c_str()));
 					}
+					else if (compare(s, "unsubscribe"))
+					{
+						client->unsubscribe(getword(cmd, topic.c_str()));
+					}
 					else if (compare(s, "view"))
 					{
 						client->dump();
 					}
 				}
+				else if (compare(s, "blink"))
+				{
+					uint8_t blink_nr = getint(cmd, 0);
+					if (blink_nr)
+					{
+						blink_ms_on[blink_nr]=getint(cmd, blink_ms_on[blink_nr]);
+						blink_ms_off[blink_nr]=getint(cmd, blink_ms_on[blink_nr]);
+						pinMode(blink_nr, OUTPUT);
+						blink_next[blink_nr] = millis();
+						Serial << "Blink " << blink_nr << ' ' << (blink_ms_on[blink_nr] ? "on" : "off") << endl;
+						if (blink_ms_on[blink_nr])
+							blink |= 1<< (blink_nr-1);
+						else
+						{
+							blink &= ~(1<<(blink_nr-1));
+						}
+					}
+				}
 				else if (compare(s, "auto"))
 				{
 					automatic::command(client, cmd);
@@ -384,6 +572,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;
@@ -412,7 +626,7 @@ void loop()
 					Serial << "  MqttClient:" << endl;
 					Serial << "    client {name} {parent broker} : create a client then" << endl;
 					Serial << "      name.connect  [ip] [port] [alive]" << endl;
-					Serial << "      name.subscribe [topic]" << endl;
+					Serial << "      name.[un]subscribe [topic]" << endl;
 					Serial << "      name.publish [topic][payload]" << endl;
 					Serial << "      name.view" << endl;
 					Serial << "      name.delete" << endl;
@@ -420,7 +634,9 @@ void loop()
 					automatic::help();
 					Serial << endl;
 					Serial << "    help" << endl;
+					Serial << "    blink [Dx on_ms off_ms]" << 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;

keywords.txt

@@ -9,13 +9,19 @@
 TinyMqtt    KEYWORD1
 
 MqttBroker  KEYWORD1
+connect		  KEYWORD2
+clientsCount KEYWORD2
 begin				KEYWORD2
 loop				KEYWORD2
+port				KEYWORD2
 
 MqttClient  KEYWORD1
+connect		  KEYWORD2
+connected	  KEYWORD2
 publish     KEYWORD2
 setCallback	KEYWORD2
 subscribe	  KEYWORD2
+unsubscribe	KEYWORD2
 
 Topic				KEYWORD1
 matches			KEYWORD2

library.json

@@ -1,12 +1,12 @@
 {
     "name": "TinyMqtt",
     "keywords": "ethernet, mqtt, m2m, iot",
-    "description": "MQTT is a lightweight messaging protocol ideal for small devices. This library allows you to send and receive MQTT messages. It does support MQTT 3.1.1 without any QOS.",
+    "description": "MQTT is a lightweight messaging protocol ideal for small devices. This library allows to send and receive MQTT messages. It does support MQTT 3.1.1 with QOS=0.",
     "repository": {
         "type": "git",
         "url": "https://github.com/hsaturn/TinyMqtt.git"
     },
-    "version": "0.4.0",
+    "version": "0.7.3",
     "exclude": "",
     "examples": "examples/*/*.ino",
     "frameworks": "arduino",

library.properties

@@ -1,10 +1,11 @@
 name=TinyMqtt
-version=0.4.0
+version=0.7.3
 author=Francois BIOT, HSaturn, <hsaturn@gmail.com>
 maintainer=Francois BIOT, HSaturn, <hsaturn@gmail.com>
 sentence=A tiny broker and client library for MQTT messaging.
-paragraph=MQTT is a lightweight messaging protocol ideal for small devices. This library allows you to send and receive MQTT messages and to jhost a broker in your ESP. It does support MQTT 3.1.1 without any QOS.
+paragraph=MQTT is a lightweight messaging protocol ideal for small devices. This library allows to send and receive MQTT messages and to host a broker in your ESP. It does support MQTT 3.1.1 with QoS=0.
 category=Communication
 url=https://github.com/hsaturn/TinyMqtt
 architectures=*
 includes=TinyMqtt.h
+depends=AsyncTCP

src/StringIndexer.h

@@ -2,7 +2,6 @@
 #include <map>
 #include <string>
 #include <string.h>
-#include <ESP8266WiFi.h>
 
 /***
  * Allows to store up to 255 different strings with one byte class
@@ -15,11 +14,21 @@ class StringIndexer
 		std::string str;
 		uint8_t used=0;
 		friend class StringIndexer;
+
+		#if EPOXY_DUINO
+		public:
+		// Workaround to avoid coredump in Indexer::release
+		// when destroying a Topic after the deletion of
+		// StringIndexer::strings map (which can occurs only with AUnit,
+		// never in the ESP itself, because ESP never ends)
+		// (I hate static vars)
+		~StringCounter() { used=255; }
+		#endif
 	};
 	public:
 		using index_t=uint8_t;
 
-		static const index_t strToIndex(const char* str, uint8_t len)
+		static index_t strToIndex(const char* str, uint8_t len)
 		{
 			for(auto it=strings.begin(); it!=strings.end(); it++)
 			{
@@ -29,7 +38,7 @@ class StringIndexer
 					return it->first;
 				}
 			}
-			for(index_t index=0; index<255; index++)
+			for(index_t index=1; index; index++)
 			{
 				if (strings.find(index)==strings.end())
 				{
@@ -70,6 +79,8 @@ class StringIndexer
 			}
 		}
 
+		static uint16_t count() { return strings.size(); }
+
 	private:
 		static std::map<index_t, StringCounter> strings;
 };
@@ -88,6 +99,8 @@ class IndexedString
 			index=StringIndexer::strToIndex(str, len);
 		}
 
+		IndexedString(const std::string& str) : IndexedString(str.c_str(), str.length()) {};
+
 		~IndexedString() { StringIndexer::release(index); }
 
 		IndexedString& operator=(const IndexedString& source)
@@ -102,9 +115,14 @@ class IndexedString
 			return i1.index < i2.index;
 		}
 
+		friend bool operator==(const IndexedString& i1, const IndexedString& i2)
+		{
+			return i1.index == i2.index;
+		}
+
 		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

@@ -9,8 +9,10 @@ void outstring(const char* prefix, const char*p, uint16_t len)
 	Serial << '\'' << endl;
 }
 
-MqttBroker::MqttBroker(uint16_t port) : server(port)
+MqttBroker::MqttBroker(uint16_t port)
 {
+	server = new AsyncServer(port);
+	server->onClient(onClient, this);
 }
 
 MqttBroker::~MqttBroker()
@@ -19,18 +21,21 @@ MqttBroker::~MqttBroker()
 	{
 		delete clients[0];
 	}
+	delete server;
 }
 
 // private constructor used by broker only
-MqttClient::MqttClient(MqttBroker* parent, WiFiClient& new_client)
+MqttClient::MqttClient(MqttBroker* parent, AsyncClient* new_client)
-	: parent(parent)
+	: parent(parent), client(new_client)
 {
-	client = new WiFiClient(new_client);
+	client->onData(onData, this);
+	// client->onConnect() TODO
+	// client->onDisconnect() TODO
 	alive = millis()+5000;	// client expires after 5s if no CONNECT msg
 }
 
-MqttClient::MqttClient(MqttBroker* parent)
+MqttClient::MqttClient(MqttBroker* parent, const std::string& id)
-	: parent(parent)
+	: parent(parent), clientId(id)
 {
 		client = nullptr;
 
@@ -43,12 +48,17 @@ MqttClient::~MqttClient()
 	delete client;
 }
 
-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();
 	}
 
@@ -64,23 +74,27 @@ void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
 	debug("cnx: closing");
 	close();
 	if (client) delete client;
-	client = new WiFiClient;
+	client = new AsyncClient;
 	debug("Trying to connect to " << broker.c_str() << ':' << port);
+	// TODO This may return immediately !!!
+	// TODO so I have to add onConnect and move this code to onConnect
+	// TODO also, as this is async now, I must take care of
+	// TODO the broker that may disconnect and delete the client immediately
 	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 = ka;			// 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);
-		message.reset();
+		msg.reset();
 	  debug("cnx: mqtt sent " << (int32_t)parent);
 
 		clientAlive(0);
@@ -92,6 +106,13 @@ void MqttBroker::addClient(MqttClient* client)
 		clients.push_back(client);
 }
 
+void MqttBroker::connect(const std::string& host, uint16_t port)
+{
+	if (broker == nullptr) broker = new MqttClient;
+	broker->connect(host, port);
+	broker->parent = this;	// Because connect removed the link
+}
+
 void MqttBroker::removeClient(MqttClient* remove)
 {
   for(auto it=clients.begin(); it!=clients.end(); it++)
@@ -99,6 +120,11 @@ void MqttBroker::removeClient(MqttClient* remove)
 		auto client=*it;
 		if (client==remove)
 		{
+			// TODO if this broker is connected to an external broker
+			// we have to unsubscribe remove's topics.
+			// (but doing this, check that other clients are not subscribed...)
+			// Unless -> we could receive useless messages
+			//        -> we are using (memory) one IndexedString plus its string for nothing.
 			debug("Remove " << clients.size());
 			clients.erase(it);
 			debug("Client removed " << clients.size());
@@ -108,19 +134,27 @@ void MqttBroker::removeClient(MqttClient* remove)
 	debug("Error cannot remove client");	// TODO should not occur
 }
 
+void MqttBroker::onClient(void* broker_ptr, AsyncClient* client)
+{
+	MqttBroker* broker = static_cast<MqttBroker*>(broker_ptr);
+
+	broker->addClient(new MqttClient(broker, client));
+	debug("New client #" << broker->clients.size());
+}
+
 void MqttBroker::loop()
 {
-  WiFiClient client = server.available();
+	if (broker)
-  
-  if (client)
 	{
-		addClient(new MqttClient(this, client));
+		// TODO should monitor broker's activity.
-		debug("New client (" << clients.size() << ')');
+		// 1 When broker disconnect and reconnect we have to re-subscribe
+		broker->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())
@@ -137,7 +171,16 @@ void MqttBroker::loop()
 	}
 }
 
-MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, MqttMessage& msg)
+MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
+{
+	if (broker && broker->connected())
+	{
+		return broker->subscribe(topic, qos);
+	}
+	return MqttNowhereToSend;
+}
+
+MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, const MqttMessage& msg) const
 {
 	MqttError retval = MqttOk;
 
@@ -146,29 +189,32 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, Mqtt
 	for(auto client: clients)
 	{
 		i++;
+#ifdef TINY_MQTT_DEBUG
 		Serial << "brk_" << (broker && broker->connected() ? "con" : "dis") <<
 			 "	srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << i << ", local=" << client->isLocal() << ", con=" << client->connected() << endl;
+#endif
 		bool doit = false;
-		if (broker && broker->connected())	// Broker is connected
+		if (broker && broker->connected())	// this (MqttBroker) is connected (to a external broker)
 		{
-			//    ext broker -> clients or
+			// ext_broker -> clients or clients -> ext_broker
-			// or clients -> ext broker
+			if (source == broker)	// external broker -> internal clients
-			if (source == broker)	// broker -> clients
 				doit = true;
-			else									// clients -> broker
+			else									// external clients -> this broker
 			{
-				MqttError ret = broker->publish(topic, msg);
+				// As this broker is connected to another broker, simply forward the msg
+				MqttError ret = broker->publishIfSubscribed(topic, msg);
 				if (ret != MqttOk) retval = ret;
 			}
 		}
-		else // Disconnected: R7
+		else // Disconnected
 		{
-			// All is allowed
 			doit = true;
 		}
+#ifdef TINY_MQTT_DEBUG
 		Serial << ", doit=" << doit << ' ';
+#endif
 
-		if (doit) retval = client->publish(topic, msg);
+		if (doit) retval = client->publishIfSubscribed(topic, msg);
 		debug("");
 	}
 	return retval;
@@ -214,21 +260,48 @@ void MqttClient::loop()
 		{
 			debug("pingreq");
 			uint16_t pingreq = MqttMessage::Type::PingReq;
-			client->write((uint8_t*)(&pingreq), 2);
+			client->write((const char*)(&pingreq), 2);
 			clientAlive(0);
 
 			// TODO when many MqttClient passes through a local browser
 			// there is no need to send one PingReq per instance.
 		}
 	}
+}
 
-	while(client && client->available()>0)
+void MqttClient::onData(void* client_ptr, AsyncClient*, void* data, size_t len)
+{
+	char* char_ptr = static_cast<char*>(data);
+	MqttClient* client=static_cast<MqttClient*>(client_ptr);
+	while(len>0)
 	{
-		message.incoming(client->read());
+		client->message.incoming(*char_ptr++);
-		if (message.type())
+		if (client->message.type())
 		{
-			processMessage();
+			client->processMessage(&client->message);
+			client->message.reset();
 		}
+		len--;
+	}
+}
+
+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
 	}
 }
 
@@ -239,39 +312,64 @@ MqttError MqttClient::subscribe(Topic topic, uint8_t qos)
 
  	subscriptions.insert(topic);
 
-	if (parent==nullptr) // remote broker ?
+	if (parent==nullptr) // remote broker
 	{
-		debug("remote subscribe");
+		return sendTopic(topic, MqttMessage::Type::Subscribe, qos);
-		MqttMessage msg(MqttMessage::Type::Subscribe, 2);
+	}
+	else
+	{
+		return parent->subscribe(topic, qos);
+	}
+	return ret;
+}
+
+MqttError MqttClient::unsubscribe(Topic topic)
+{
+	auto it=subscriptions.find(topic);
+	if (it != subscriptions.end())
+	{
+		subscriptions.erase(it);
+		if (parent==nullptr) // remote broker
+		{
+			return sendTopic(topic, MqttMessage::Type::UnSubscribe, 0);
+		}
+	}
+	return MqttOk;
+}
+
+MqttError MqttClient::sendTopic(const Topic& topic, MqttMessage::Type type, uint8_t qos)
+{
+	MqttMessage msg(type, 2);
 
 	// TODO manage packet identifier
 	msg.add(0);
 	msg.add(0);
 
-		msg.add(topic.str());
+	msg.add(topic);
 	msg.add(qos);
-		ret = msg.sendTo(this);
 
-		// TODO we should wait (state machine) for SUBACK
+	// TODO instead we should wait (state machine) for SUBACK / UNSUBACK ?
-	}
+	return msg.sendTo(this);
-	return ret;
 }
 
-void MqttClient::processMessage()
+long MqttClient::counter=0;
+
+void MqttClient::processMessage(const MqttMessage* mesg)
 {
-	std::string error;
+	counter++;
-	std::string s;
+#ifdef TINY_MQTT_DEBUG
-if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessage::Type::PingResp)
+if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::Type::PingResp)
 {
-	Serial << "---> INCOMING " << _HEX(message.type()) << " client(" << (int)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
+	Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (int)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
-	message.hexdump("Incoming");
+	// mesg->hexdump("Incoming");
 }
-  auto header = message.getVHeader();
+#endif
+  auto header = mesg->getVHeader();
   const char* payload;
   uint16_t len;
 	bool bclose=true;
 
-	switch(message.type() & 0XF0)
+	switch(mesg->type() & 0XF0)
 	{
 		case MqttMessage::Type::Connect:
 			if (mqtt_connected)
@@ -294,44 +392,30 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
 			}
 
 			// ClientId
-			message.getString(payload, len);
+			mesg->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;
 
 			if (mqtt_flags & FlagWill)	// Will topic
 			{
-				message.getString(payload, len);	// Will Topic
+				mesg->getString(payload, len);	// Will Topic
 				outstring("WillTopic", payload, len);
 				payload += len;
 
-				message.getString(payload, len);	// Will Message
+				mesg->getString(payload, len);	// Will Message
 				outstring("WillMessage", payload, len);
 				payload += len;
 			}
 			// FIXME forgetting credential is allowed (security hole)
 			if (mqtt_flags & FlagUserName)
 			{
-				message.getString(payload, len);
+				mesg->getString(payload, len);
 				if (!parent->checkUser(payload, len)) break;
 				payload += len;
 			}
 			if (mqtt_flags & FlagPassword)
 			{
-				message.getString(payload, len);
+				mesg->getString(payload, len);
 				if (!parent->checkPassword(payload, len)) break;
 				payload += len;
 			}
@@ -339,21 +423,22 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
 			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:
+		case MqttMessage::Type::ConnAck:
-			// TODO what more on connack ?
 			mqtt_connected = true;
 			bclose = false;
+			resubscribe();
 			break;
 
-		case MqttMessage::Type::Suback:
+		case MqttMessage::Type::SubAck:
-		case MqttMessage::Type::Puback:
+		case MqttMessage::Type::PubAck:
 			if (!mqtt_connected) break;
 			// Ignore acks
 			bclose = false;
@@ -369,7 +454,7 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
 			if (client)
 			{
 				uint16_t pingreq = MqttMessage::Type::PingResp;
-				client->write((uint8_t*)(&pingreq), 2);
+				client->write((const char*)(&pingreq), 2);
 			  bclose = false;
 			}
 			else
@@ -379,21 +464,32 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
 			break;
 
 		case MqttMessage::Type::Subscribe:
+		case MqttMessage::Type::UnSubscribe:
 			{
 				if (!mqtt_connected) break;
 				payload = header+2;
 				
-				debug("subscribe loop");
+				debug("un/subscribe loop");
-				while(payload < message.end())
+				while(payload < mesg->end())
 				{
-					message.getString(payload, len);	// Topic
+					mesg->getString(payload, len);	// Topic
 					debug( "  topic (" << std::string(payload, len) << ')');
-					outstring("Subscribes", payload, len);
+					outstring("  un/subscribes", payload, len);
 					// subscribe(Topic(payload, len));
-					subscriptions.insert(Topic(payload, len));
+					Topic topic(payload, len);
 					payload += len;
+					if ((mesg->type() & 0XF0) == MqttMessage::Type::Subscribe)
+					{
 						uint8_t qos = *payload++;
-					debug("  qos=" << qos);
+						if (qos != 0) debug("Unsupported QOS" << qos << endl);
+						subscriptions.insert(topic);
+					}
+					else
+					{
+						auto it=subscriptions.find(topic);
+						if (it != subscriptions.end())
+							subscriptions.erase(it);
+					}
 				}
 				debug("end loop");
 				bclose = false;
@@ -402,36 +498,42 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
 			break;
 
 		case MqttMessage::Type::Publish:
-			if (!mqtt_connected) break;
+			if (mqtt_connected or client == nullptr)
 			{
-				uint8_t qos = message.type() & 0x6;
+				uint8_t qos = mesg->type() & 0x6;
 				payload = header;
-				message.getString(payload, len);
+				mesg->getString(payload, len);
 				Topic published(payload, len);
 				payload += len;
-				len=message.end()-payload;
 				// Serial << "Received Publish (" << published.str().c_str() << ") size=" << (int)len
-				// << '(' << std::string(payload, len).c_str() << ')'	<< " msglen=" << message.length() << endl; 
+				// << '(' << std::string(payload, len).c_str() << ')'	<< " msglen=" << mesg->length() << endl;
 				if (qos) payload+=2;	// ignore packet identifier if any
+				len=mesg->end()-payload;
 				// TODO reset DUP
 				// TODO reset RETAIN
-				if (parent)
+
+				if (client==nullptr)	// internal MqttClient receives publish
+				{
+					if (callback and isSubscribedTo(published))
+					{
+						callback(this, published, payload, len);	// TODO send the real payload
+					}
+				}
+				else if (parent) // from outside to inside
 				{
 				  debug("publishing to parent");
-					parent->publish(this, published, message);
+					parent->publish(this, published, *mesg);
 				}
-				else if (callback && subscriptions.find(published)!=subscriptions.end())
-				{
-					callback(this, published, nullptr, 0);	// TODO send the real payload
-				}
-				// TODO should send PUBACK
 				bclose = false;
 			}
 			break;
 
-		case MqttMessage::Type::PubAck:
+		case MqttMessage::Type::Disconnect:
+			// TODO should discard any will msg
 			if (!mqtt_connected) break;
-			bclose = false;
+			mqtt_connected = false;
+			close(false);
+			bclose=false;
 			break;
 
 		default:
@@ -440,8 +542,9 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
 	};
 	if (bclose)
 	{
-		Serial << "*************** Error msg 0x" << _HEX(message.type());
+		Serial << "*************** Error msg 0x" << _HEX(mesg->type());
-		if (error.length()) Serial << ':' << error.c_str();
+		mesg->hexdump("-------ERROR ------");
+		dump();
 		Serial << endl;
 		close();
   }
@@ -449,7 +552,6 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
 	{
 		clientAlive(parent ? 5 : 0);
 	}
-	message.reset();
 }
 
 bool Topic::matches(const Topic& topic) const
@@ -462,12 +564,14 @@ 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);
+	msg.complete();
 	if (parent)
+	{
 		return parent->publish(this, topic, msg);
+	}
 	else if (client)
 		return msg.sendTo(this);
 	else
@@ -475,31 +579,33 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa
 }
 
 // republish a received publish if it matches any in subscriptions
-MqttError MqttClient::publish(const Topic& topic, MqttMessage& msg)
+MqttError MqttClient::publishIfSubscribed(const Topic& topic, const MqttMessage& msg)
 {
 	MqttError retval=MqttOk;
 
 	debug("mqttclient publish " << subscriptions.size());
-	for(const auto& subscription: subscriptions)
+	if (isSubscribedTo(topic))
 	{
-		Serial << " client=" << (int32_t)client << ", topic " << topic.str().c_str() << ' ';
-		if (subscription.matches(topic))
-		{
-			Serial << " match/send";
 		if (client)
-			{
 			retval = msg.sendTo(this);
-			}
+		else
-			else if (callback)
 		{
-				callback(this, topic, nullptr, 0);	// TODO Payload
+			processMessage(&msg);
+			// callback(this, topic, nullptr, 0);	// TODO Payload
 		}
 	}
-		Serial << endl;
-	}
 	return retval;
 }
 
+bool MqttClient::isSubscribedTo(const Topic& topic) const
+{
+	for(const auto& subscription: subscriptions)
+		if (subscription.matches(topic))
+			return true;
+
+	return false;
+}
+
 void MqttMessage::reset()
 {
 	buffer.clear();
@@ -527,9 +633,16 @@ 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:
 		case PayLoad:
@@ -546,6 +659,7 @@ void MqttMessage::incoming(char in_byte)
 		case Complete:
 		default:
 			Serial << "Spurious " << _HEX(in_byte) << endl;
+			hexdump("spurious");
 			reset();
 			break;
 	}
@@ -560,13 +674,14 @@ void MqttMessage::add(const char* p, size_t len, bool addLength)
 {
 	if (addLength)
 	{
+		buffer.reserve(buffer.length()+addLength+2);
 		incoming(len>>8);
 		incoming(len & 0xFF);
 	}
 	while(len--) incoming(*p++);
 }
 
-void MqttMessage::encodeLength(char* msb, int length)
+void MqttMessage::encodeLength(char* msb, int length) const
 {
 	do
 	{
@@ -577,13 +692,19 @@ void MqttMessage::encodeLength(char* msb, int length)
 	} while (length);
 };
 
-MqttError MqttMessage::sendTo(MqttClient* client)
+void MqttMessage::complete()
 {
-	if (buffer.size()>2)
+		encodeLength(&buffer[1], buffer.size()-2);
+		state = Complete;
+}
+
+MqttError MqttMessage::sendTo(MqttClient* client) const
+{
+	if (buffer.size())
 	{
 		debug("sending " << buffer.size() << " bytes");
 		encodeLength(&buffer[1], buffer.size()-2);
-		hexdump("snd");
+		// hexdump("snd");
 		client->write(&buffer[0], buffer.size());
 	}
 	else

src/TinyMqtt.h

@@ -1,11 +1,13 @@
+#pragma once
 #include <ESP8266WiFi.h>
+#include <ESPAsyncTCP.h>
 #include <vector>
 #include <set>
 #include <string>
 #include "StringIndexer.h"
 #include <MqttStreaming.h>
 
-#define TINY_MQTT_DEBUG
+// #define TINY_MQTT_DEBUG
 
 #ifdef TINY_MQTT_DEBUG
   #define debug(what) { Serial << __LINE__ << ' ' << what << endl; delay(100); }
@@ -41,13 +43,15 @@ class MqttMessage
 		{
 			Unknown     =    0,
 			Connect     = 0x10,
-			Connack   = 0x20,
+			ConnAck     = 0x20,
 			Publish     = 0x30,
 			PubAck      = 0x40,
 			Subscribe   = 0x80,
-			Suback    = 0x90,
+			SubAck      = 0x90,
+			UnSubscribe = 0xA0,
 			PingReq     = 0xC0,
 			PingResp    = 0xD0,
+			Disconnect  = 0xE0
 		};
 		enum State
 		{
@@ -70,6 +74,7 @@ class MqttMessage
 		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();
 
@@ -77,7 +82,6 @@ class MqttMessage
 		// output buff+=2, len=length(str)
 		static void getString(const char* &buff, uint16_t& len);
 
-
 		Type type() const
 		{
 			return state == Complete ? static_cast<Type>(buffer[0]) : Unknown;
@@ -86,18 +90,18 @@ class MqttMessage
 		void create(Type type)
 		{
 			buffer=(char)type;
-			buffer+='\0';
+			buffer+='\0';		// reserved for msg length
 			vheader=2;
 			size=0;
 			state=Create;
 		}
-		MqttError sendTo(MqttClient*);
+		MqttError sendTo(MqttClient*) const;
 		void hexdump(const char* prefix=nullptr) const;
 
 	private:
-		void encodeLength(char* msb, int length);
+		void encodeLength(char* msb, int length) const;
 
-		std::string buffer;
+		mutable std::string buffer;	// mutable -> sendTo()
 		uint8_t vheader;
 		uint16_t size;	// bytes left to receive
 		State state;
@@ -118,13 +122,15 @@ class MqttClient
 		FlagReserved = 1
 	};
 	public:
-		MqttClient(MqttBroker*);
+	  /** Constructor. If broker is not null, this is the adress of a local broker.
-		MqttClient() : MqttClient(nullptr) {};
+		    If you want to connect elsewhere, leave broker null and use connect() **/
+		MqttClient(MqttBroker* broker = nullptr, const std::string& id="");
+		MqttClient(const std::string& id) : MqttClient(nullptr, id){}
 
 		~MqttClient();
 
 		void connect(MqttBroker* parent);
-		void connect(std::string broker, uint16_t port, uint16_t ka=10);
+		void connect(std::string broker, uint16_t port, uint16_t keep_alive = 10);
 
 		bool connected() { return
 			(parent!=nullptr and client==nullptr) or
@@ -135,8 +141,9 @@ class MqttClient
 		const std::string& id() const { return clientId; }
 		void id(std::string& new_id) { clientId = new_id; }
 
+		/** Should be called in main loop() */
 		void loop();
-		void close();
+		void close(bool bSendDisconnect=true);
 		void setCallback(CallBack fun) {callback=fun; };
 
 		// Publish from client to the world
@@ -146,23 +153,22 @@ class MqttClient
 		MqttError publish(const Topic& t) { return publish(t, nullptr, 0);};
 
 		MqttError subscribe(Topic topic, uint8_t qos=0);
-		MqttError unsubscribe(Topic& topic);
+		MqttError unsubscribe(Topic topic);
+		bool isSubscribedTo(const Topic& topic) const;
 
 		// 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
+			Serial << "MqttClient (" << clientId.c_str() << ") " << (connected() ? " ON " : " OFF");
-				<< " c=" << (int32_t)client << (connected() ? " ON " : " OFF"); 
+			Serial << ", alive=" << alive << '/' << ms << ", ka=" << keep_alive;
-			Serial << ", alive=" << (uint32_t)alive << '/' << ms << ", ka=" << keep_alive;
+			Serial << (client && client->connected() ? "" : "dis") << "connected";
-			Serial << " cnx " << (client && client->connected());
-			Serial << " [";
             message.hexdump("entrant msg");
 			bool c=false;
+			Serial << " [";
 			for(auto s: subscriptions)
 			{
 				Serial << (c?", ": "")<< s.str().c_str();
@@ -172,20 +178,26 @@ class MqttClient
 			
 			Serial << "]" << endl;
 		}
-#endif
+
+		/** Count the number of messages that have been sent **/
+		static long counter;
 
 	private:
+		static void onData(void* client_ptr, AsyncClient*, void* data, size_t len);
+		MqttError sendTopic(const Topic& topic, MqttMessage::Type type, uint8_t qos);
+		void resubscribe();
+
 		friend class MqttBroker;
-		MqttClient(MqttBroker* parent, WiFiClient& client);
+		MqttClient(MqttBroker* parent, AsyncClient* client);
 		// republish a received publish if topic matches any in subscriptions
-		MqttError publish(const Topic& topic, MqttMessage& msg);
+		MqttError publishIfSubscribed(const Topic& topic, const MqttMessage& msg);
 
 		void clientAlive(uint32_t more_seconds);
-		void processMessage();
+		void processMessage(const MqttMessage* message);
 
 		bool mqtt_connected = false;
 		char mqtt_flags;
-		uint32_t keep_alive;
+		uint32_t keep_alive = 60;
 		uint32_t alive;
 		MqttMessage message;
 
@@ -194,7 +206,7 @@ class MqttClient
 		// (this is the case when MqttBroker isn't used except here)
 		MqttBroker* parent=nullptr;		// connection to local broker
 
-		WiFiClient* client=nullptr;		// connection to mqtt client or to remote broker
+		AsyncClient* client=nullptr;		// connection to mqtt client or to remote broker
 		std::set<Topic>	subscriptions;
 		std::string clientId;
 		CallBack callback = nullptr;
@@ -213,15 +225,14 @@ class MqttBroker
 		MqttBroker(uint16_t port);
 		~MqttBroker();
 
-		void begin() { server.begin(); }
+		void begin() { server->begin(); }
 		void loop();
 
-		uint8_t port() const { return server.port(); }
+		void connect(const std::string& host, uint16_t port=1883);
-
-		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;
@@ -231,11 +242,11 @@ class MqttBroker
 				client->dump();
 			}
 		}
-#endif
 
 	private:
 		friend class MqttClient;
 
+		static void onClient(void*, AsyncClient*);
 		bool checkUser(const char* user, uint8_t len) const
 		{ return compareString(auth_user, user, len); }
 
@@ -243,7 +254,9 @@ class MqttBroker
 		{ return compareString(auth_password, password, len); }
 
 
-		MqttError publish(const MqttClient* source, const Topic& topic, MqttMessage& msg);
+		MqttError publish(const MqttClient* source, const Topic& topic, const MqttMessage& msg) const;
+
+		MqttError subscribe(const Topic& topic, uint8_t qos);
 
 		// For clients that are added not by the broker itself
 		void addClient(MqttClient* client);
@@ -251,7 +264,7 @@ class MqttBroker
 
 		bool compareString(const char* good, const char* str, uint8_t str_len) const;
 		std::vector<MqttClient*>	clients;
-		WiFiServer server;
+		AsyncServer* server;
 
 		const char* auth_user = "guest";
 		const char* auth_password = "guest";

tests/Makefile

@@ -0,0 +1,20 @@
+tests:
+	set -e; \
+	for i in *-tests/Makefile; do \
+		echo '==== Making:' $$(dirname $$i); \
+		$(MAKE) -C $$(dirname $$i) -j; \
+	done
+
+runtests: tests
+	set -e; \
+	for i in *-tests/Makefile; do \
+		echo '==== Running:' $$(dirname $$i); \
+		$$(dirname $$i)/$$(dirname $$i).out; \
+	done
+
+clean:
+	set -e; \
+	for i in *-tests/Makefile; do \
+		echo '==== Cleaning:' $$(dirname $$i); \
+		$(MAKE) -C $$(dirname $$i) clean; \
+	done

tests/local-tests/Makefile

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

tests/local-tests/local-tests.ino

@@ -0,0 +1,152 @@
+#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(local_client_should_unregister_when_destroyed)
+{
+return;
+	assertEqual(broker.clientsCount(), (size_t)0);
+	{
+		MqttClient client;
+		assertEqual(broker.clientsCount(), (size_t)0);	// Ensure client is not yet connected
+		client.connect("127.0.0.1", 1883);
+		assertEqual(broker.clientsCount(), (size_t)1);	// Ensure client is now connected
+	}
+	assertEqual(broker.clientsCount(), (size_t)0);
+}
+
+#if 0
+test(local_connect)
+{
+	assertEqual(broker.clientsCount(), (size_t)0);
+
+	MqttClient client;
+	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;
+	subscriber.subscribe("a/b");
+	subscriber.subscribe("a/c");
+	subscriber.setCallback(onPublish);
+
+	MqttClient publisher;
+	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("A");
+	subscriber_a.setCallback(onPublish);
+	subscriber_a.subscribe("a/b");
+	subscriber_a.subscribe("a/c");
+
+	MqttClient subscriber_b("B");
+	subscriber_b.setCallback(onPublish);
+	subscriber_b.subscribe("a/b");
+
+	MqttClient publisher;
+	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;
+	subscriber.setCallback(onPublish);
+	subscriber.subscribe("a/b");
+
+	MqttClient publisher;
+	publisher.publish("a/b");
+
+	subscriber.unsubscribe("a/b");
+
+	publisher.publish("a/b");
+	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 publisher;
+	{
+		MqttClient subscriber;
+		subscriber.setCallback(onPublish);
+		subscriber.subscribe("a/b");
+		publisher.publish("a/b");
+	}
+
+	publisher.publish("a/b");
+
+	assertEqual(published.size(), (size_t)1);	// Only one publish has been received
+}
+#endif
+
+//----------------------------------------------------------------------------
+// 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/nowifi-tests/Makefile

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

tests/nowifi-tests/nowifi-tests.ino

@@ -0,0 +1,166 @@
+#include <AUnit.h>
+#include <TinyMqtt.h>
+#include <map>
+
+/**
+  * TinyMqtt nowifi unit tests.
+	*
+	* No wifi connection unit tests.
+  * Checks with a local broker. Clients must connect to the local broker
+	**/
+
+using namespace std;
+
+MqttBroker broker(1883);
+
+std::map<std::string, std::map<Topic, int>>	published;		// map[client_id] => map[topic] = count
+
+char* lastPayload = nullptr;
+size_t lastLength;
+
+void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
+{
+	if (srce)
+		published[srce->id()][topic]++;
+	
+	if (lastPayload) free(lastPayload);
+  lastPayload = strdup(payload);
+	lastLength = length;
+}
+
+test(nowifi_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(nowifi_connect)
+{
+	assertEqual(broker.clientsCount(), (size_t)0);
+
+	MqttClient client(&broker);
+	assertTrue(client.connected());
+	assertEqual(broker.clientsCount(), (size_t)1);
+}
+
+test(nowifi_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
+	assertEqual(published[""]["a/b"], 1);
+	assertEqual(published[""]["a/c"], 2);
+}
+
+test(nowifi_publish_should_be_dispatched_to_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");		// A and B should receive this
+	publisher.publish("a/c");		// A should receive this 
+
+	assertEqual(published.size(), (size_t)2);	// 2 clients have received something
+	assertEqual(published["A"]["a/b"], 1);
+	assertEqual(published["A"]["a/c"], 1);
+	assertEqual(published["B"]["a/b"], 1);
+	assertEqual(published["B"]["a/c"], 0);
+}
+
+test(nowifi_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");		// This publish is received
+
+	subscriber.unsubscribe("a/b");
+
+	publisher.publish("a/b");		// Those one, no (unsubscribed)
+	publisher.publish("a/b");
+
+	assertEqual(published[""]["a/b"], 1);	// Only one publish has been received
+}
+
+test(nowifi_nocallback_when_destroyed)
+{
+	published.clear();
+	assertEqual(broker.clientsCount(), (size_t)0);
+
+	MqttClient publisher(&broker);
+
+	{
+		MqttClient subscriber(&broker);
+		subscriber.setCallback(onPublish);
+		subscriber.subscribe("a/b");
+		publisher.publish("a/b");
+	}
+
+	publisher.publish("a/b");
+
+	assertEqual(published.size(), (size_t)1);	// Only one publish has been received
+}
+
+test(nowifi_payload_nullptr)
+{
+	published.clear();
+
+	const char* payload="abcd";
+
+	MqttClient subscriber(&broker);
+	subscriber.setCallback(onPublish);
+	subscriber.subscribe("a/b");
+
+	MqttClient publisher(&broker);
+	publisher.publish("a/b", payload, strlen(payload));		// This publish is received
+
+  // coming from MqttClient::publish(...)
+  assertEqual(payload, lastPayload);
+	assertEqual(lastLength, (size_t)4);
+}
+
+//----------------------------------------------------------------------------
+// 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/result.json

@@ -0,0 +1,6 @@
+{
+    "schemaVersion" : 1,
+    "label" : "tests",
+    "message" : "Message content",
+    "color": "red"
+}

tests/result.yaml

@@ -0,0 +1,2 @@
+result: 1
+insert: "passed"

tests/string-indexer-tests/Makefile

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

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

@@ -0,0 +1,109 @@
+#include <AUnit.h>
+#include <StringIndexer.h>
+#include <map>
+
+/**
+	* TinyMqtt / StringIndexer unit tests.
+	*
+	**/
+
+using namespace std;
+
+test(indexer_empty)
+{
+  assertEqual(StringIndexer::count(), 0);
+}
+
+test(indexer_strings_deleted_should_empty_indexer)
+{
+  assertTrue(StringIndexer::count()==0);
+	{
+		IndexedString one("one");
+    assertEqual(StringIndexer::count(), 1);
+		IndexedString two("two");
+    assertEqual(StringIndexer::count(), 2);
+		IndexedString three("three");
+    assertEqual(StringIndexer::count(), 3);
+		IndexedString four("four");
+    assertEqual(StringIndexer::count(), 4);
+	}
+  assertEqual(StringIndexer::count(), 0);
+}
+
+test(indexer_same_strings_count_as_one)
+{
+	IndexedString one  ("one");
+	IndexedString two  ("one");
+	IndexedString three("one");
+	IndexedString fourt("one");
+
+	assertEqual(StringIndexer::count(), 1);
+}
+
+test(indexer_size_of_indexed_string)
+{
+	assertEqual(sizeof(IndexedString), (size_t)1);
+}
+
+test(indexer_different_strings_are_different)
+{
+	IndexedString one("one");
+	IndexedString two("two");
+
+	assertFalse(one == two);
+}
+
+test(indexer_same_strings_should_equal)
+{
+	IndexedString one("one");
+	IndexedString two("one");
+
+	assertTrue(one == two);
+}
+
+test(indexer_indexed_operator_eq)
+{
+	IndexedString one("one");
+	{
+		IndexedString same = one;
+		assertTrue(one == same);
+		assertEqual(StringIndexer::count(), 1);
+  }
+	assertEqual(StringIndexer::count(), 1);
+}
+
+test(indexer_get_string)
+{
+	std::string sone("one");
+	IndexedString one(sone);
+
+  assertTrue(sone==one.str());
+}
+
+test(indexer_get_index)
+{
+	IndexedString one1("one");
+	IndexedString one2("one");
+	IndexedString two1("two");
+	IndexedString two2("two");
+
+  assertTrue(one1.getIndex() == one2.getIndex());
+	assertTrue(two1.getIndex() == two2.getIndex());
+	assertTrue(one1.getIndex() != two1.getIndex());
+}
+
+//----------------------------------------------------------------------------
+// setup() and loop()
+void setup() {
+	delay(1000);
+	Serial.begin(115200);
+	while(!Serial);
+
+	Serial.println("=============[ TinyMqtt StringIndexer TESTS ]========================");
+}
+
+void loop() {
+	aunit::TestRunner::run();
+
+	if (Serial.available()) ESP.reset();
+}