Very promising async commit

AsyncTcp

.gitignore

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

README.md

@@ -21,32 +21,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 +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/tinymqtt-test.ino

@@ -1,6 +1,7 @@
-#define TINY_MQTT_DEBUG
 #include <TinyMqtt.h>       // https://github.com/hsaturn/TinyMqtt
 #include <MqttStreaming.h>
+#include <ESP8266mDNS.h>
+
 #include <sstream>
 #include <map>
 
@@ -55,6 +56,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;
@@ -126,10 +135,10 @@ 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"
 };
 
 void getCommand(std::string& search)
@@ -313,9 +322,39 @@ 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;
@@ -453,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);
@@ -565,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;
@@ -573,6 +634,7 @@ 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;

library.json

@@ -1,7 +1,7 @@
 {
     "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 MQTT messages. It does support MQTT 3.1.1 with QOS=0.",
     "repository": {
         "type": "git",
         "url": "https://github.com/hsaturn/TinyMqtt.git"

library.properties

@@ -3,9 +3,9 @@ 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 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. 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,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,14 +21,16 @@ MqttBroker::~MqttBroker()
 	{
 		delete clients[0];
 	}
-	server.close();
+	delete server;
 }
 
 // private constructor used by broker only
-MqttClient::MqttClient(MqttBroker* parent, WiFiClient& new_client)
-	: parent(parent)
+MqttClient::MqttClient(MqttBroker* parent, AsyncClient* new_client)
+	: 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
 }
 
@@ -70,8 +74,12 @@ 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");
@@ -126,10 +134,16 @@ 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)
 	{
 		// TODO should monitor broker's activity.
@@ -137,11 +151,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
@@ -168,6 +177,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 +189,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 +210,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,22 +260,28 @@ 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());
-		if (message.type())
+		client->message.incoming(*char_ptr++);
+		if (client->message.type())
 		{
-			processMessage(&message);
-			message.reset();
+			client->processMessage(&client->message);
+			client->message.reset();
 		}
+		len--;
 	}
 }
 
@@ -341,7 +357,7 @@ 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;
@@ -438,7 +454,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 +469,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;
@@ -656,6 +674,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,15 +1,17 @@
+#pragma once
 #include <ESP8266WiFi.h>
+#include <ESPAsyncTCP.h>
 #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
 
@@ -80,23 +82,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;
@@ -189,11 +183,12 @@ class MqttClient
 		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 publishIfSubscribed(const Topic& topic, const MqttMessage& msg);
 
@@ -202,7 +197,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 +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;
@@ -230,11 +225,9 @@ 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; }
 
@@ -253,6 +246,7 @@ class MqttBroker
 	private:
 		friend class MqttClient;
 
+		static void onClient(void*, AsyncClient*);
 		bool checkUser(const char* user, uint8_t len) const
 		{ return compareString(auth_user, user, len); }
 
@@ -270,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

@@ -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);