Release 0.7.5

Modify dump() functions

.github/workflows/ci.yml

@@ -0,0 +1,18 @@
+name: "CI"
+on:
+  
+jobs:
+  ci:
+    runs-on: ubuntu-20.04
+    steps:
+    - name: Checkout this repository
+      uses: actions/checkout@v2.3.4
+    - name: Cache for arduino-ci
+      uses: actions/cache@v2.1.3
+      with:
+        path: |
+          ~/.arduino15
+        key: ${{ runner.os }}-arduino
+    - name: Install nix
+      uses: cachix/install-nix-action@v12
+    - run: nix-shell -I nixpkgs=channel:nixpkgs-unstable -p arduino-ci --run "arduino-ci"

.gitignore

@@ -1,5 +0,0 @@
-*~
-src/my_credentials.h
-*.o
-*.swp
-*.out

README.md

@@ -1,17 +1,19 @@
 # TinyMqtt
 
 ![Release](https://img.shields.io/github/v/release/hsaturn/TinyMqtt)
+[![AUnit Tests](https://github.com/hsaturn/TinyMqtt/actions/workflows/aunit.yml/badge.svg)](https://github.com/hsaturn/TinyMqtt/actions/workflows/aunit.yml)
 ![Issues](https://img.shields.io/github/issues/hsaturn/TinyMqtt)
 ![Esp8266](https://img.shields.io/badge/platform-ESP8266-green)
+![Esp32](https://img.shields.io/badge/platform-ESP32-green)
 ![Gpl 3.0](https://img.shields.io/github/license/hsaturn/TinyMqtt)
 ![Mqtt 3.1.1](https://img.shields.io/badge/Mqtt-%203.1.1-yellow)
 
-ESP 8266 is a small, fast and capable Mqtt Broker and Client
+TinyMqtt is a small, fast and capable Mqtt Broker and Client for Esp8266 / Esp32 / Esp WROOM
 
 ## 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.
+  clients that had subscribed (payload ~15 bytes ESP8266). No topic lost.
   The max I've seen was 2k msg/s (1 client 1 subscription)
 - Act as as a mqtt broker and/or a mqtt client
 - Mqtt 3.1.1 / Qos 0 supported
@@ -21,32 +23,16 @@ ESP 8266 is a small, fast and 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)
-* 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.
-* ~~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 unsubscribe~~
-* 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 [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           |
@@ -57,11 +43,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/client-with-wifi/client-with-wifi.ino

@@ -3,6 +3,23 @@
 /** 
   * Local broker that accept connections and two local clients
   *
+  * 
+  *  +-----------------------------+
+  *  | ESP                         |
+  *  |                 +--------+  | 1883 <--- External client/s
+  *  |       +-------->| broker |  | 1883 <--- External client/s
+  *  |       |         +--------+  |
+  *  |       |             ^       |
+  *  |       |             |       |
+  *  |       |             |       |     -----
+  *  |       v             v       |      ---  
+  *  | +----------+  +----------+  |       -  
+  *  | | internal |  | internal |  +-------*  Wifi
+  *  | | client   |  | client   |  |          
+  *  | +----------+  +----------+  |          
+  *  |                             |
+  *  +-----------------------------+
+  * 
   * pros - Reduces internal latency (when publish is received by the same ESP)
   *      - Reduces wifi traffic
   *      - No need to have an external broker
@@ -12,11 +29,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>
@@ -58,7 +70,7 @@ void setup()
 
 void loop()
 {
-	broker.loop();
+  broker.loop();  // Don't forget to add loop for every broker and clients
 
   mqtt_a.loop();
   mqtt_b.loop();

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

@@ -1,11 +1,29 @@
 #include <TinyMqtt.h>   // https://github.com/hsaturn/TinyMqtt
 
 /** TinyMQTT allows a disconnected mode:
+  * 
+  *  +-----------------------------+
+  *  | ESP                         |
+  *  |                 +--------+  |
+  *  |       +-------->| broker |  |
+  *  |       |         +--------+  |
+  *  |       |             ^       |
+  *  |       |             |       |
+  *  |       v             v       |
+  *  | +----------+  +----------+  |
+  *  | | internal |  | internal |  |
+  *  | | client   |  | client   |  |          
+  *  | +----------+  +----------+  |          
+  *  |                             |
+  *  +-----------------------------+
   *
   *  In this example, local clients A and B are talking together, no need to be connected.
+  *
   *  A single ESP can use this to be able to comunicate with itself with the power
   *  of MQTT, and once connected still continue to work with others.
   *
+  *  The broker may still be conected if wifi is on.
+  *  
   */
 
 std::string topic="sensor/temperature";

examples/simple-broker/simple-broker.ino

@@ -5,6 +5,16 @@
 #define PORT 1883
 MqttBroker broker(PORT);
 
+/** Basic Mqtt Broker
+  *
+  *  +-----------------------------+
+  *  | ESP                         |
+  *  |       +--------+            | 
+  *  |       | broker |            | 1883 <--- External client/s
+  *  |       +--------+            |
+  *  |                             |
+  *  +-----------------------------+
+  */
 void setup() 
 {
   Serial.begin(115200);

examples/simple-client/simple-client.ino

@@ -1,9 +1,23 @@
-#include <ESP8266WiFi.h>
 #include "TinyMqtt.h"   // https://github.com/hsaturn/TinyMqtt
 
-/** Simple Client
+/** Simple Client (The simplest configuration)
   *
-	* This is the simplest Mqtt client configuration
+  *                             
+  *                   +--------+       
+  *           +------>| broker |<--- <  Other client
+  *           |       +--------+       
+  *           |                               
+  *   +-----------------+
+  *   | ESP   |         |           
+  *   | +----------+    |          
+  *   | | internal |    |              
+  *   | | client   |    |          
+  *   | +----------+    |          
+  *   |                 |
+  *   +-----------------+
+  * 
+	* 1 - edit my_credentials.h to setup wifi essid/password
+	* 2 - change BROKER values (or keep emqx.io test broker)
 	*
 	* pro  - small memory footprint (both ram and flash)
 	*      - very simple to setup and use
@@ -13,6 +27,9 @@
 	*      - local publishes takes more time (because they go outside)
 	*/
 
+const char* BROKER = "broker.emqx.io";
+const uint16_t BROKER_PORT = 1883;
+
 #include <my_credentials.h>
 
 static float temp=19;
@@ -32,8 +49,7 @@ void setup()
 
   Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;  
 
-	client.connect("192.168.1.40", 1883);	// Put here your broker ip / port
+	client.connect(BROKER, BROKER_PORT);	// Put here your broker ip / port
-
 }
 
 void loop()

examples/tinymqtt-test/tinymqtt-test.ino

@@ -1,21 +1,24 @@
-#define TINY_MQTT_DEBUG
 #include <TinyMqtt.h>       // https://github.com/hsaturn/TinyMqtt
 #include <MqttStreaming.h>
+#if defined(ESP8266)
+  #include <ESP8266mDNS.h>
+#elif defined(ESP32)
+  #include <ESPmDNS.h>
+#else
+  #error Unsupported platform
+#endif
+
 #include <sstream>
 #include <map>
 
-/** 
+/** Very complex example
   * Console allowing to make any kind of test.
   *
-  * pros - Reduces internal latency (when publish is received by the same ESP)
+	* Upload the sketch, the use the terminal.
-  *      - Reduces wifi traffic
+	* Press H for mini help.
-  *      - No need to have an external broker
-	*      - can still report to a 'main' broker (TODO see documentation that have to be written)
-	*      - accepts external clients
-  *
-  * cons - Takes more memory
-	*      - a bit hard to understand
 	*
+  * tested with mqtt-spy-0.5.4
+	* TODO examples of scripts
   */
 
 #include <my_credentials.h>
@@ -55,6 +58,14 @@ void setup()
   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();
 	brokers["broker"] = broker;
@@ -108,7 +119,7 @@ std::string getip(std::string& str, const char* if_empty=nullptr, char sep=' ')
 			return addr;
 	}
 	IPAddress local=WiFi.localIP();
-	addr="";
+	addr.clear();
 	while(build.size()!=4)
 	{
 		std::stringstream b;
@@ -126,13 +137,13 @@ std::string getip(std::string& str, const char* if_empty=nullptr, char sep=' ')
 std::map<std::string, std::string> vars;
 
 std::set<std::string> commands = {
-	"auto", "broker", "client", "connect",
+	"auto", "broker", "blink", "client", "connect",
 	"create", "delete", "help", "interval",
 	"ls", "ip", "off", "on", "set",
-	"publish", "reset", "subscribe", "view"
+	"publish", "reset", "subscribe", "unsubscribe", "view", "every"
 };
 
-void getCommand(std::string& search)
+void convertToCommand(std::string& search)
 {
 	while(search[0]==' ') search.erase(0,1);
 	if (search.length()==0) return;
@@ -152,7 +163,7 @@ void getCommand(std::string& search)
 	else if (count>1)
 	{
 		Serial << "Ambiguous command: " << matches << endl;
-		search="";
+		search.clear();
 	}
 }
 
@@ -307,44 +318,47 @@ std::map<MqttClient*, automatic*> automatic::autos;
 
 bool compare(std::string s, const char* cmd)
 {
-	if (s.length()==0 or s.length()>strlen(cmd))  return false;
+	uint8_t p=0;
-	return strncmp(cmd, s.c_str(), s.length())==0;
+	while(s[p++]==*cmd++)
+	{
+		if (*cmd==0 or s[p]==0) return true;
+		if (s[p]==' ') return true;
+	}
+	return false;
 }
 
 using ClientFunction = void(*)(std::string& cmd, MqttClient* publish);
 
-void loop()
+struct Every
 {
-	static long count;
+	std::string cmd;
-	if (MqttClient::counter != count)
+	uint32_t ms;
+	uint32_t next;
+	uint32_t underrun=0;
+	bool active=true;
+
+	void dump()
 	{
-		Serial << "# " << MqttClient::counter << endl;
+		Serial << (active ? "enabled  " : "disabled ");
-		count = MqttClient::counter;
+		auto mill=millis();
-	}
+		Serial << ms << "ms [" << cmd << "] next in ";
-	for(auto it: brokers)
+		if (mill > next)
-		it.second->loop();
+			Serial << "now";
-
-	for(auto it: clients)
-		it.second->loop();
-
-	automatic::loop();
-
-	if (Serial.available())
-	{
-		static std::string cmd;
-		char c=Serial.read();
-
-		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;
+			Serial << next-mill << "ms";
+	}
+};
 
-			if (cmd.substr(0,3)!="set") replaceVars(cmd);
+uint32_t blink_ms_on[16];
+uint32_t blink_ms_off[16];
+uint32_t blink_next[16];
+bool blink_state[16];
+int16_t blink;
+
+std::vector<Every> everies;
+
+void eval(std::string& cmd)
+{
 	while(cmd.length())
 	{
 		MqttError retval = MqttOk;
@@ -354,7 +368,6 @@ void loop()
 		MqttClient* client = nullptr;
 
 		// client.function notation
-				// ("a.fun " becomes "fun a ")
 		if (cmd.find('.') != std::string::npos &&
 				cmd.find('.') < cmd.find(' '))
 		{
@@ -373,13 +386,13 @@ void loop()
 				else
 				{
 					Serial << "Unknown class (" << s.c_str() << ")" << endl;
-							cmd="";
+					cmd.clear();
 				}
 			}
 		}
 
 		s = getword(cmd);
-				if (s.length()) getCommand(s);
+		if (s.length()) convertToCommand(s);
 		if (s.length()==0)
 		{}
 		else if (compare(s, "delete"))
@@ -435,6 +448,11 @@ void loop()
 			{
 				broker->dump();
 			}
+			else
+			{
+				Serial << "Unknown broker command (" << s << ")" << endl;
+				s.clear();
+			}
 		}
 		else if (client)
 		{
@@ -447,16 +465,145 @@ void loop()
 			{
 				while (cmd[0]==' ') cmd.erase(0,1);
 				retval = client->publish(getword(cmd, topic.c_str()), cmd.c_str(), cmd.length());
-						cmd="";	// remove payload
+				cmd.clear();	// remove payload
 			}
 			else if (compare(s,"subscribe"))
 			{
 				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
+			{
+				Serial << "Unknown client command (" << s << ")" << endl;
+				s.clear();
+			}
+		}
+		else if (compare(s, "on"))
+		{
+			uint8_t pin=getint(cmd, 2);
+			pinMode(pin, OUTPUT);
+			digitalWrite(pin, HIGH);
+		}
+		else if (compare(s, "off"))
+		{
+			uint8_t pin=getint(cmd, 2);
+			pinMode(pin, OUTPUT);
+			digitalWrite(pin, LOW);
+		}
+		else if (compare(s, "every"))
+		{
+			uint32_t ms = getint(cmd, 0);
+			if (ms)
+			{
+				if (cmd.length())
+				{
+					Every every;
+					every.ms=ms;
+					every.cmd=cmd;
+					every.next=millis()+ms;
+					everies.push_back(every);
+					every.dump();
+					Serial << endl;
+					cmd.clear();
+				}
+			}
+			else if (compare(cmd, "off") or compare(cmd, "on"))
+			{
+				bool active=getword(cmd)=="on";
+				uint8_t ever;
+				if (compare(cmd, "all"))
+					ever=100;
+				else
+					ever=getint(cmd, 99);
+				uint8_t count=0;
+				if (ever == 99)
+				{
+					Serial << "Missing every number" << endl;
+				}
+				else
+				{
+					for(auto& every: everies)
+					{
+						if (count==ever or (ever==100))
+						{
+							if (every.active != active)
+							{
+								every.active = active;
+								every.underrun = 0;
+							}
+							ever = 99;
+							break;
+						}
+						count++;
+					}
+					if (ever != 99)
+					{
+						Serial << "Every not found" << endl;
+					}
+				}
+			}
+			else if (compare(cmd, "list") or cmd.length()==0)
+			{
+				getword(cmd);
+				Serial << "List of everies (ms=" << millis() << ")" << endl;
+				uint8_t count=0;
+				for(auto& every: everies)
+				{
+					Serial << count << ": ";
+					every.dump();
+					Serial << endl;
+					count++;
+				}
+			}
+			else if (compare(cmd, "remove"))
+			{
+				Serial << "Removing..." << endl;
+				getword(cmd);
+				int8_t every=getint(cmd, -1);
+				if (every==-1 and compare(cmd, "last") and everies.size())
+				{
+					getword(cmd);
+					everies.erase(everies.begin()+everies.size()-1);
+				}
+				else if (every==-1 and compare(cmd, "all"))
+				{
+					getword(cmd);
+					everies.clear();
+				}
+				else if (everies.size() > (uint8_t)every)
+				{
+					everies.erase(everies.begin()+every);
+				}
+				else
+					Serial << "Bad colmmand" << endl;
+			}
+			else
+				Serial << "Bad command" << endl;
+		}
+		else if (compare(s, "blink"))
+		{
+			int8_t blink_nr = getint(cmd, -1);
+			if (blink_nr >= 0)
+			{
+				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;
+				else
+				{
+					blink &= ~(1<< blink_nr);
+				}
+			}
 		}
 		else if (compare(s, "auto"))
 		{
@@ -542,14 +689,14 @@ void loop()
 			Serial << "--< " << clients.size() << " client/s. >--" << endl;
 			for(auto it: clients)
 			{
-						Serial << "  "; it.second->dump();
+				it.second->dump("  ");
 			}
 
 			Serial << "--< " << brokers.size() << " brokers/s. >--" << endl;
 			for(auto it: brokers)
 			{
-						Serial << " ==[ Broker: " << it.first.c_str() << " ]== ";
+				Serial << "  +-- '" << it.first.c_str() << "' " << it.second->clientsCount() << " client/s."<< endl;
-						it.second->dump();
+				it.second->dump("     ");
 			}
 		}
 		else if (compare(s, "reset"))
@@ -565,7 +712,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;
@@ -573,10 +720,13 @@ 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 << "  every ms [command]; every list; every remove [nr|all], every [on|off] #" << endl;
+			Serial << "  on {output}; off {output}" << endl;
 			Serial << "  $id : name of the client." << endl;
 			Serial << "  default topic is '" << topic.c_str() << "'" << endl;
 			Serial << endl;
@@ -590,9 +740,89 @@ void loop()
 
 		if (retval != MqttOk)
 		{
-					Serial << "## ERROR " << retval << endl;
+			Serial << "# MQTT ERROR " << retval << endl;
 		}
 	}
+}
+
+void loop()
+{
+	auto ms=millis();
+	int8_t out=0;
+	int16_t blink_bits = blink;
+	uint8_t e=0;
+
+	for(auto& every: everies)
+	{
+		if (not every.active) continue;
+		if (every.ms && every.cmd.length() && ms > every.next)
+		{
+			std::string cmd(every.cmd);
+			eval(cmd);
+			every.next += every.ms;
+			if (ms > every.next and ms > every.underrun)
+			{
+				Serial << "Underrun every #" << e << ", " <<  (ms - every.next) << "ms late" << endl;
+				every.underrun = ms+5000;
+			}
+		}
+		e++;
+	}
+
+	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;
+  #if defined(ESP9266)
+  MDNS.update();
+  #endif
+
+	if (MqttClient::counter != count)
+	{
+		Serial << "# " << MqttClient::counter << endl;
+		count = MqttClient::counter;
+	}
+	for(auto it: brokers)
+		it.second->loop();
+
+	for(auto it: clients)
+		it.second->loop();
+
+	automatic::loop();
+
+	if (Serial.available())
+	{
+		static std::string cmd;
+		char c=Serial.read();
+
+		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);
+			eval(cmd);
 		}
 		else
 		{

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 to send and receive MQTT messages. It does support MQTT 3.1.1 without QOS=0.",
+    "description": "MQTT is a lightweight messaging protocol ideal for small devices. This library allows to send and receive and host a broker for MQTT. It does support MQTT 3.1.1 with QOS=0 on ESP8266 and ESP32 WROOM platfrms.",
     "repository": {
         "type": "git",
         "url": "https://github.com/hsaturn/TinyMqtt.git"
     },
-    "version": "0.7.3",
+    "version": "0.7.5",
     "exclude": "",
     "examples": "examples/*/*.ino",
     "frameworks": "arduino",

library.properties

@@ -1,11 +1,11 @@
 name=TinyMqtt
-version=0.7.3
+version=0.7.5
 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 to send and receive MQTT messages and to host a broker in your ESP. It does support MQTT 3.1.1 without QoS=0.
+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 8266 and 32 WROOM. It does support MQTT 3.1.1 with QoS=0.
 category=Communication
 url=https://github.com/hsaturn/TinyMqtt
 architectures=*
 includes=TinyMqtt.h
-depends=
+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
@@ -29,7 +28,7 @@ class StringIndexer
 	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++)
 			{

src/TinyMqtt.cpp

@@ -9,8 +9,12 @@ 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 TcpServer(port);
+#ifdef TCP_ASYNC
+	server->onClient(onClient, this);
+#endif
 }
 
 MqttBroker::~MqttBroker()
@@ -19,14 +23,21 @@ MqttBroker::~MqttBroker()
 	{
 		delete clients[0];
 	}
-	server.close();
+	delete server;
 }
 
 // private constructor used by broker only
-MqttClient::MqttClient(MqttBroker* parent, WiFiClient& new_client)
+MqttClient::MqttClient(MqttBroker* parent, TcpClient* new_client)
 	: parent(parent)
 {
-	client = new WiFiClient(new_client);
+#ifdef TCP_ASYNC
+	client = new_client;
+	client->onData(onData, this);
+	// client->onConnect() TODO
+	// client->onDisconnect() TODO
+#else
+	client = new WiFiClient(*new_client);
+#endif
 	alive = millis()+5000;	// client expires after 5s if no CONNECT msg
 }
 
@@ -68,29 +79,22 @@ void MqttClient::close(bool bSendDisconnect)
 void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
 {
 	debug("cnx: closing");
+	keep_alive = ka;
 	close();
 	if (client) delete client;
-	client = new WiFiClient;
+	client = new TcpClient;
+
 	debug("Trying to connect to " << broker.c_str() << ':' << port);
+#ifdef TCP_ASYNC
+	client->onData(onData, this);
+	client->onConnect(onConnect, this);
+	client->connect(broker.c_str(), port);
+#else
 	if (client->connect(broker.c_str(), port))
 	{
-	  debug("cnx: connecting");
+		onConnect(this, client);
-		MqttMessage msg(MqttMessage::Type::Connect);
-		msg.add("MQTT",4);
-		msg.add(0x4);	// Mqtt protocol version 3.1.1
-		msg.add(0x0);	// Connect flags         TODO user / name
-
-		keep_alive = ka;
-		msg.add(0x00);			// keep_alive
-		msg.add((char)keep_alive);
-		msg.add(clientId);
-	  debug("cnx: mqtt connecting");
-		msg.sendTo(this);
-		msg.reset();
-	  debug("cnx: mqtt sent " << (int32_t)parent);
-
-		clientAlive(0);
 	}
+#endif
 }
 
 void MqttBroker::addClient(MqttClient* client)
@@ -126,10 +130,24 @@ void MqttBroker::removeClient(MqttClient* remove)
 	debug("Error cannot remove client");	// TODO should not occur
 }
 
+void MqttBroker::onClient(void* broker_ptr, TcpClient* client)
+{
+	MqttBroker* broker = static_cast<MqttBroker*>(broker_ptr);
+
+	broker->addClient(new MqttClient(broker, client));
+	debug("New client");
+}
+
 void MqttBroker::loop()
 {
-  WiFiClient client = server.available();
+#ifndef TCP_ASYNC
+  WiFiClient client = server->available();
 
+  if (client)
+	{
+		onClient(this, &client);
+	}
+#endif
 	if (broker)
 	{
 		// TODO should monitor broker's activity.
@@ -137,11 +155,6 @@ void MqttBroker::loop()
 		broker->loop();
 	}
 
-  if (client)
-	{
-		addClient(new MqttClient(this, client));
-		debug("New client (" << clients.size() << ')');
-	}
 
   // for(auto it=clients.begin(); it!=clients.end(); it++)
 	// use index because size can change during the loop
@@ -154,7 +167,7 @@ void MqttBroker::loop()
 		}
 		else
 		{
-      debug("Client " << client->id().c_str() << "  Disconnected, parent=" << (int32_t)client->parent);
+      debug("Client " << client->id().c_str() << "  Disconnected, parent=" << (dbg_ptr)client->parent);
 			// Note: deleting a client not added by the broker itself will probably crash later.
 			delete client;
 			break;
@@ -168,6 +181,7 @@ MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
 	{
 		return broker->subscribe(topic, qos);
 	}
+	return MqttNowhereToSend;
 }
 
 MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, const MqttMessage& msg) const
@@ -179,7 +193,7 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, cons
 	for(auto client: clients)
 	{
 		i++;
-#if TINY_MQTT_DEBUG
+#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
@@ -200,7 +214,7 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, cons
 		{
 			doit = true;
 		}
-#if TINY_MQTT_DEBUG
+#ifdef TINY_MQTT_DEBUG
 		Serial << ", doit=" << doit << ' ';
 #endif
 
@@ -250,14 +264,14 @@ 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
+			// TODO when many MqttClient passes through a local broker
 			// there is no need to send one PingReq per instance.
 		}
 	}
-
+#ifndef TCP_ASYNC
 	while(client && client->available()>0)
 	{
 		message.incoming(client->read());
@@ -267,8 +281,47 @@ void MqttClient::loop()
 			message.reset();
 		}
 	}
+#endif
 }
 
+void MqttClient::onConnect(void *mqttclient_ptr, TcpClient*)
+{
+	MqttClient* mqtt = static_cast<MqttClient*>(mqttclient_ptr);
+	debug("cnx: connecting");
+	MqttMessage msg(MqttMessage::Type::Connect);
+	msg.add("MQTT",4);
+	msg.add(0x4);	// Mqtt protocol version 3.1.1
+	msg.add(0x0);	// Connect flags         TODO user / name
+
+	msg.add(0x00);			// keep_alive
+	msg.add((char)mqtt->keep_alive);
+	msg.add(mqtt->clientId);
+	debug("cnx: mqtt connecting");
+	msg.sendTo(mqtt);
+	msg.reset();
+	debug("cnx: mqtt sent " << (dbg_ptr)mqtt->parent);
+
+	mqtt->clientAlive(0);
+}
+
+#ifdef TCP_ASYNC
+void MqttClient::onData(void* client_ptr, TcpClient*, void* data, size_t len)
+{
+	char* char_ptr = static_cast<char*>(data);
+	MqttClient* client=static_cast<MqttClient*>(client_ptr);
+	while(len>0)
+	{
+		client->message.incoming(*char_ptr++);
+		if (client->message.type())
+		{
+			client->processMessage(&client->message);
+			client->message.reset();
+		}
+		len--;
+	}
+}
+#endif
+
 void MqttClient::resubscribe()
 {
 	// TODO resubscription limited to 256 bytes
@@ -341,10 +394,10 @@ long MqttClient::counter=0;
 void MqttClient::processMessage(const MqttMessage* mesg)
 {
 	counter++;
-#if TINY_MQTT_DEBUG
+#ifdef TINY_MQTT_DEBUG
 if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::Type::PingResp)
 {
-	Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (int)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
+	Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
 	// mesg->hexdump("Incoming");
 }
 #endif
@@ -438,7 +491,7 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
 			if (client)
 			{
 				uint16_t pingreq = MqttMessage::Type::PingResp;
-				client->write((uint8_t*)(&pingreq), 2);
+				client->write((const char*)(&pingreq), 2);
 			  bclose = false;
 			}
 			else
@@ -453,25 +506,27 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
 				if (!mqtt_connected) break;
 				payload = header+2;
 				
-				debug("subscribe loop");
+				debug("un/subscribe loop");
 				while(payload < mesg->end())
 				{
 					mesg->getString(payload, len);	// Topic
 					debug( "  topic (" << std::string(payload, len) << ')');
-					outstring("Subscribes", payload, len);
+					outstring("  un/subscribes", payload, len);
 					// subscribe(Topic(payload, len));
 					Topic topic(payload, len);
+					payload += len;
 					if ((mesg->type() & 0XF0) == MqttMessage::Type::Subscribe)
+					{
+						uint8_t qos = *payload++;
+						if (qos != 0) debug("Unsupported QOS" << qos << endl);
 						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;
@@ -479,6 +534,11 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
 			}
 			break;
 
+		case MqttMessage::Type::UnSuback:
+			if (!mqtt_connected) break;
+			bclose = false;
+			break;
+
 		case MqttMessage::Type::Publish:
 			if (mqtt_connected or client == nullptr)
 			{
@@ -656,6 +716,7 @@ 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);
 	}

src/TinyMqtt.h

@@ -1,18 +1,48 @@
-#include <ESP8266WiFi.h>
+#pragma once
+
+// TODO Should add a AUnit with both TCP_ASYNC and not TCP_ASYNC
+// #define TCP_ASYNC	// Uncomment this to use ESPAsyncTCP instead of normal cnx
+
+#if defined(ESP8266) || defined(EPOXY_DUINO)
+	#ifdef TCP_ASYNC
+		#include <ESPAsyncTCP.h>
+  #else
+    #include <ESP8266WiFi.h>
+  #endif
+#elif defined(ESP32)
+	#ifdef TCP_ASYNC
+	    #include <AsyncTCP.h> // https://github.com/me-no-dev/AsyncTCP
+    #else
+	    #include <WiFi.h>
+    #endif
+#endif
+#ifdef EPOXY_DUINO
+  #define dbg_ptr uint64_t
+#else
+  #define dbg_ptr uint32_t
+#endif
 #include <vector>
 #include <set>
 #include <string>
 #include "StringIndexer.h"
 #include <MqttStreaming.h>
 
-#if 0
+// #define TINY_MQTT_DEBUG
+
+#ifdef TINY_MQTT_DEBUG
   #define debug(what) { Serial << __LINE__ << ' ' << what << endl; delay(100); }
-  #define TINY_MQTT_DEBUG 1
 #else
-  #define TINY_MQTT_DEBUG 0
   #define debug(what) {}
 #endif
 
+#ifdef TCP_ASYNC
+  using TcpClient = AsyncClient;
+  using TcpServer = AsyncServer;
+#else
+  using TcpClient = WiFiClient;
+  using TcpServer = WiFiServer;
+#endif
+
 enum MqttError
 {
 	MqttOk = 0,
@@ -47,6 +77,7 @@ class MqttMessage
 			Subscribe   = 0x80,
 			SubAck      = 0x90,
 			UnSubscribe = 0xA0,
+			UnSuback	= 0xB0,
 			PingReq     = 0xC0,
 			PingResp    = 0xD0,
 			Disconnect  = 0xE0
@@ -80,23 +111,15 @@ 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;
 		}
 
-		// shouldn't exist because it breaks constness :-(
-		// but this saves memory so ...
-		void changeType(Type type) const
-		{
-			buffer[0] = type;
-		}
-
 		void create(Type type)
 		{
 			buffer=(char)type;
-			buffer+='\0';
+			buffer+='\0';		// reserved for msg length
 			vheader=2;
 			size=0;
 			state=Create;
@@ -166,34 +189,40 @@ class MqttClient
 		// TODO seems to be useless
 		bool isLocal() const { return client == nullptr; }
 
-		void dump()
+		void dump(std::string indent="")
 		{
 			uint32_t ms=millis();
-			Serial << "MqttClient (" << clientId.c_str() << ") " << (connected() ? " ON " : " OFF");
+			Serial << indent << "+-- " << '\'' << clientId.c_str() << "' " << (connected() ? " ON " : " OFF");
-			Serial << ", alive=" << alive << '/' << ms << ", ka=" << keep_alive;
+			Serial << ", alive=" << alive << '/' << ms << ", ka=" << keep_alive << ' ';
 			Serial << (client && client->connected() ? "" : "dis") << "connected";
-            message.hexdump("entrant msg");
+			if (subscriptions.size())
-			bool c=false;
+			{
+				bool c = false;
 				Serial << " [";
 				for(auto s: subscriptions)
 				{
-				Serial << (c?", ": "")<< s.str().c_str();
+					if (c) Serial << ", ";
+					Serial << s.str().c_str();
 					c=true;
 				}
-
+				Serial << ']';
-			
+			}
-			Serial << "]" << endl;
+			Serial << endl;
 		}
 
 		/** Count the number of messages that have been sent **/
 		static long counter;
 
 	private:
+		static void onConnect(void * client_ptr, TcpClient*);
+#ifdef TCP_ASYNC
+		static void onData(void* client_ptr, TcpClient*, void* data, size_t len);
+#endif
 		MqttError sendTopic(const Topic& topic, MqttMessage::Type type, uint8_t qos);
 		void resubscribe();
 
 		friend class MqttBroker;
-		MqttClient(MqttBroker* parent, WiFiClient& client);
+		MqttClient(MqttBroker* parent, TcpClient* client);
 		// republish a received publish if topic matches any in subscriptions
 		MqttError publishIfSubscribed(const Topic& topic, const MqttMessage& msg);
 
@@ -202,7 +231,7 @@ class MqttClient
 
 		bool mqtt_connected = false;
 		char mqtt_flags;
-		uint32_t keep_alive;
+		uint32_t keep_alive = 60;
 		uint32_t alive;
 		MqttMessage message;
 
@@ -211,7 +240,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
+		TcpClient* client=nullptr;		// connection to mqtt client or to remote broker
 		std::set<Topic>	subscriptions;
 		std::string clientId;
 		CallBack callback = nullptr;
@@ -230,29 +259,24 @@ class MqttBroker
 		MqttBroker(uint16_t port);
 		~MqttBroker();
 
-		void begin() { server.begin(); }
+		void begin() { server->begin(); }
 		void loop();
 
-		uint16_t port() const { return server.port(); }
-
 		void connect(const std::string& host, uint16_t port=1883);
 		bool connected() const { return state == Connected; }
 
 		size_t clientsCount() const { return clients.size(); }
 
-		void dump()
+		void dump(std::string indent="")
 		{
-			Serial << clients.size() << " client/s" << endl;
 			for(auto client: clients)
-			{
+				client->dump(indent);
-				Serial << "  ";
-				client->dump();
-			}
 		}
 
 	private:
 		friend class MqttClient;
 
+		static void onClient(void*, TcpClient*);
 		bool checkUser(const char* user, uint8_t len) const
 		{ return compareString(auth_user, user, len); }
 
@@ -270,7 +294,7 @@ class MqttBroker
 
 		bool compareString(const char* good, const char* str, uint8_t str_len) const;
 		std::vector<MqttClient*>	clients;
-		WiFiServer server;
+		TcpServer* server;
 
 		const char* auth_user = "guest";
 		const char* auth_password = "guest";

tests/Makefile

@@ -5,7 +5,7 @@ tests:
 		$(MAKE) -C $$(dirname $$i) -j; \
 	done
 
-runtests:
+runtests: tests
 	set -e; \
 	for i in *-tests/Makefile; do \
 		echo '==== Running:' $$(dirname $$i); \

tests/local-tests/Makefile

@@ -3,4 +3,5 @@
 
 APP_NAME := local-tests
 ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock
-include ../../../EpoxyDuino/EpoxyDuino.mk
+ESP_LIBS = ESP8266WiFi ESPAsyncTCP
+include ../../../EspMock/EspMock.mk

tests/local-tests/local-tests.ino

@@ -6,7 +6,7 @@
 	* TinyMqtt local unit tests.
 	*
 	* Clients are connected to pseudo remote broker
-	* The remote will be 127.0.0.1:1883
+	* 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
 	**/

tests/nowifi-tests/Makefile

@@ -3,4 +3,5 @@
 
 APP_NAME := nowifi-tests
 ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock
-include ../../../EpoxyDuino/EpoxyDuino.mk
+ESP_LIBS = ESP8266WiFi ESPAsyncTCP
+include ../../../EspMock/EspMock.mk

tests/nowifi-tests/nowifi-tests.ino

@@ -15,14 +15,16 @@ MqttBroker broker(1883);
 
 std::map<std::string, std::map<Topic, int>>	published;		// map[client_id] => map[topic] = count
 
-const char* lastPayload;
+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]++;
-  lastPayload = payload;
+	
+	if (lastPayload) free(lastPayload);
+  lastPayload = strdup(payload);
 	lastLength = length;
 }
 
@@ -131,7 +133,6 @@ test(nowifi_nocallback_when_destroyed)
 
 test(nowifi_payload_nullptr)
 {
-  return; // FIXME
 	published.clear();
 
 	const char* payload="abcd";

tests/string-indexer-tests/Makefile

@@ -3,4 +3,5 @@
 
 APP_NAME := string-indexer-tests
 ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock
-include ../../../EpoxyDuino/EpoxyDuino.mk
+ESP_LIBS = ESP8266WiFi ESPAsyncTCP
+include ../../../EspMock/EspMock.mk

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

@@ -1,28 +1,14 @@
 #include <AUnit.h>
-#include <TinyMqtt.h>
+#include <StringIndexer.h>
 #include <map>
 
 /**
-	* TinyMqtt local unit tests.
+	* TinyMqtt / StringIndexer unit tests.
 	*
-	* Clients are connected to pseudo remote broker
-	* The remote will be 127.0.0.1:1883
-	* 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
-
-void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
-{
-	if (srce)
-		published[srce->id()][topic]++;
-}
-
 test(indexer_empty)
 {
   assertEqual(StringIndexer::count(), 0);