Release 0.4.0

Fix too long time

README.md

@@ -1,2 +1,61 @@
 # TinyMqtt
-ESP 8266 Small footprint Mqtt Broker and Client
+
+![](https://img.shields.io/github/v/release/hsaturn/TinyMqtt)
+![](https://img.shields.io/github/issues/hsaturn/TinyMqtt)
+![](https://img.shields.io/badge/paltform-ESP8266-green)
+![](https://img.shields.io/github/license/hsaturn/TinyMqtt)
+![](https://img.shields.io/badge/Mqtt-%203.1.1-yellow)
+
+ESP 8266 is a small and very capable Mqtt Broker and Client
+
+## Features
+
+- 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)
+- Brokers can connect to another broker and becomes then a
+  proxy for clients that are connected to it.
+- 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, 3k is needed per client which would make more than 10 clients critical.
+* MqttMessage uses a buffer 256 bytes which is usually far than needed.
+* MqttClient auto reconnection
+* MqttClient auto re-subscribe
+* MqttClient does not callback payload...
+* MqttClient user/password
+
+## Quickstart
+
+* install [Streaming library](https://github.com/janelia-arduino/Streaming)
+* install [TinyMqtt library](https://github.com/hsaturn/TinyMqtt)
+* modify <libraries/TinyMqtt/src/my_credentials.h> (wifi setup)
+
+## Examples
+
+
+| 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           |
+| tinymqtt-test       | Complex console example                    |
+
+- tinymqtt-test : This is a complex sketch with a terminal console
+  that allows to add clients publish, connect etc with interpreted commands.
+
+## Standalone mode (zeroconf)
+-> The zeroconf mode is not yet implemented
+zerofonf 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.
+
+## License
+Gnu GPL 3.0, see [LICENSE](https://github.com/hsaturn/TinyMqtt/blob/main/LICENSE).

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

@@ -0,0 +1,88 @@
+#include <TinyMqtt.h>   // https://github.com/hsaturn/TinyMqtt
+#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
+
+/** 
+  * Local broker that accept connections and two local clients
+  *
+  * pros - Reduces internal latency (when publish is received by the same ESP)
+  *      - Reduces wifi traffic
+  *      - 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
+	*
+  * 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>
+
+std::string topic="sensor/temperature";
+
+MqttBroker broker(1883);
+
+MqttClient mqtt_a(&broker);
+MqttClient mqtt_b(&broker);
+
+void onPublishA(const MqttClient* source, const Topic& topic, const char* payload, size_t length)
+{ Serial << endl << "---------> A Received " << topic.c_str() << endl; }
+
+void onPublishB(const MqttClient* source, const Topic& topic, const char* payload, size_t length)
+{ Serial << endl << "---------> B Received " << topic.c_str() << endl; }
+
+void setup()
+{
+  Serial.begin(115200);
+	delay(500);
+	Serial << "Clients with wifi " << endl;
+
+  WiFi.mode(WIFI_STA);
+  WiFi.begin(ssid, password);
+
+  while (WiFi.status() != WL_CONNECTED) {   Serial << '-'; delay(500); }
+
+  Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
+
+	broker.begin();
+ 
+  mqtt_a.setCallback(onPublishA);
+  mqtt_a.subscribe(topic);
+
+  mqtt_b.setCallback(onPublishB);
+  mqtt_b.subscribe(topic);
+}
+
+void loop()
+{
+	broker.loop();
+
+  mqtt_a.loop();
+  mqtt_b.loop();
+
+  // ============= client A publish ================
+	static const int intervalA = 5000;  // publishes every 5s
+	static uint32_t timerA = millis() + intervalA;
+	
+	if (millis() > timerA)
+	{
+		Serial << "A is publishing " << topic.c_str() << endl;
+		timerA += intervalA;
+		mqtt_a.publish(topic);
+	}
+
+  // ============= client B publish ================
+	static const int intervalB = 7000;	// will send topic each 7s
+	static uint32_t timerB = millis() + intervalB;
+	
+	if (millis() > timerB)
+	{
+		Serial << "B is publishing " << topic.c_str() << endl;
+		timerB += intervalB;
+		mqtt_b.publish(topic, std::string(String(15+millis()%10).c_str()));
+	}
+}

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

@@ -0,0 +1,64 @@
+#include <TinyMqtt.h>   // https://github.com/hsaturn/TinyMqtt
+#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
+
+/** TinyMQTT allows a disconnected mode:
+ * 
+ *  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.
+ *  
+ */
+
+std::string topic="sensor/temperature";
+
+MqttBroker broker(1883);
+MqttClient mqtt_a(&broker);
+MqttClient mqtt_b(&broker);
+
+void onPublishA(const Topic& topic, const char* payload, size_t length)
+{ Serial << "--> A Received " << topic.c_str() << endl; }
+
+void onPublishB(const Topic& topic, const char* payload, size_t length)
+{ Serial << "--> B Received " << topic.c_str() << endl; }
+
+void setup()
+{
+  Serial.begin(115200);
+  delay(500);
+  Serial << "init" << endl;
+
+  mqtt_a.setCallback(onPublishA);
+  mqtt_a.subscribe(topic);
+
+  mqtt_b.setCallback(onPublishB);
+  mqtt_b.subscribe(topic);
+}
+
+void loop()
+{
+	broker.loop();
+  mqtt_a.loop();
+  mqtt_b.loop();
+
+  // ============= client A publish ================
+	static const int intervalA = 5000;
+	static uint32_t timerA = millis() + intervalA;
+	
+	if (millis() > timerA)
+	{
+		Serial << "A is publishing " << topic.c_str() << endl;
+		timerA += intervalA;
+		mqtt_a.publish(topic);
+	}
+
+  // ============= client B publish ================
+	static const int intervalB = 3000;	// will send topic each 5000 ms
+	static uint32_t timerB = millis() + intervalB;
+	
+	if (millis() > timerB)
+	{
+		Serial << "B is publishing " << topic.c_str() << endl;
+		timerB += intervalB;
+		mqtt_b.publish(topic);
+	}
+}

examples/simple-broker/simple-broker.ino

@@ -1,9 +1,7 @@
-#include <ESP8266WiFi.h>
 #include "TinyMqtt.h"   // https://github.com/hsaturn/TinyMqtt
 #include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
 
-const char *ssid = ;      // Put here your wifi SSID ("ssid")
-const char *password = ;  // Put here your Wifi password ("pwd")
+#include <my_credentials.h>
 
 #define PORT 1883
 MqttBroker broker(PORT);
@@ -16,7 +14,6 @@ void setup()
   WiFi.begin(ssid, password);
  
   while (WiFi.status() != WL_CONNECTED) {   
-    delay(500);
     Serial << '.';
     delay(500);
   }

examples/simple-client/simple-client.ino

@@ -0,0 +1,49 @@
+#include <ESP8266WiFi.h>
+#include "TinyMqtt.h"   // https://github.com/hsaturn/TinyMqtt
+#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
+
+/** Simple Client
+  *
+	* This is the simplest Mqtt client configuration
+	*
+	* pro  - small memory footprint (both ram and flash)
+	*      - very simple to setup and use
+	*
+	* cons - cannot work without wifi connection
+  *      - stop working if broker is down
+	*      - local publishes takes more time (because they go outside)
+	*/
+
+#include <my_credentials.h>
+
+static float temp=19;
+static MqttClient client;
+
+void setup() 
+{
+  Serial.begin(115200);
+	delay(500);
+	Serial << "Simple clients with wifi " << endl;
+
+  WiFi.mode(WIFI_STA);
+  WiFi.begin(ssid, password);
+ 
+  while (WiFi.status() != WL_CONNECTED)
+	{ delay(500); Serial << '.'; }
+
+  Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;  
+
+	client.connect("192.168.1.40", 1883);	// Put here your broker ip / port
+
+}
+
+void loop()
+{
+	client.loop();
+
+	delay(1000);
+
+	temp += (random(100)>50 ? 0.1 : -0.1);
+	client.publish("sensor/temperature", String(temp));
+
+}

examples/tinymqtt-test/tinymqtt-test.ino

@@ -0,0 +1,441 @@
+#define TINY_MQTT_DEBUG
+#include <TinyMqtt.h>   // https://github.com/hsaturn/TinyMqtt
+#include <Streaming.h>  // https://github.com/janelia-arduino/Streaming
+#include <map>
+
+/** 
+  * Console allowing to make any kind of test.
+  *
+  * pros - Reduces internal latency (when publish is received by the same ESP)
+  *      - Reduces wifi traffic
+  *      - 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
+	*
+  * 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>
+
+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; }
+
+std::map<std::string, MqttClient*> clients;
+std::map<std::string, MqttBroker*> brokers;
+
+
+void setup()
+{
+  Serial.begin(115200);
+	delay(500);
+	Serial << endl << endl << endl
+	<< "Demo started. Type help for more..." << endl
+	<< "Connecting to '" << ssid << "' ";
+
+  WiFi.mode(WIFI_STA);
+  WiFi.begin(ssid, password);
+
+  while (WiFi.status() != WL_CONNECTED)
+	{ Serial << '-'; delay(500); }
+
+  Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
+
+	MqttBroker* broker = new MqttBroker(1883);
+	broker->begin();
+	brokers["broker"] = broker;
+}
+
+int getint(std::string& str, const int if_empty=0, char sep=' ')
+{
+	std::string sword;
+	while(str.length() && str[0]!=sep)
+	{
+		sword += str[0]; str.erase(0,1);
+	}
+	while(str[0]==sep) str.erase(0,1);
+	if (if_empty and sword.length()==0) sword=if_empty;
+	return atoi(sword.c_str());
+}
+
+std::string getword(std::string& str, const char* if_empty=nullptr, char sep=' ')
+{
+	std::string sword;
+	while(str.length() && str[0]!=sep)
+	{
+		sword += str[0]; str.erase(0,1);
+	}
+	while(str[0]==sep) str.erase(0,1);
+	if (if_empty and sword.length()==0) return if_empty;
+	return sword;
+}
+
+class automatic
+{
+	public:
+		automatic(MqttClient* clt, uint32_t intervl)
+		: client(clt), topic_(::topic)
+		{
+			interval(intervl);
+		  autos[clt] = this;
+		}
+
+		void interval(uint32_t new_interval)
+		{
+			interval_ = new_interval;
+			if (interval_<1000) interval_=1000;
+			timer_ = millis() + interval_;
+		}
+
+		void loop_()
+		{
+			if (!bon) return;
+			if (interval_ && millis() > timer_)
+			{
+				Serial << "AUTO PUBLISH " << interval_ << endl;
+				timer_ += interval_;
+				client->publish(topic_, std::string(String(15+millis()%10).c_str()));
+			}
+		}
+
+		void topic(std::string new_topic) { topic_ = new_topic; }
+
+		static void loop()
+		{
+			for(auto it: autos)
+				it.second->loop_();
+		}
+
+		static void command(MqttClient* who, std::string cmd)
+		{
+			automatic* autop = nullptr;
+			if (autos.find(who) != autos.end())
+			{
+				autop=autos[who];
+			}
+			std::string s = getword(cmd);
+			if (compare(s, "create"))
+			{
+				std::string seconds=getword(cmd, "10000");
+				if (autop) delete autop;
+				std::string top = getword(cmd, ::topic.c_str());
+				autos[who] = new automatic(who, atol(seconds.c_str()));
+				autos[who]->topic(top);
+				autos[who]->bon=true;
+				Serial << "New auto (" << seconds.c_str() << " topic:" << top.c_str() << ')' << endl;
+			}
+			else if (autop)
+			{
+				while(s.length())
+				{
+					if (s=="on")
+					{
+						autop->bon = true;
+						autop->interval(autop->interval_);
+					}
+					else if (s=="off")
+						autop->bon=false;
+					else if (s=="interval")
+					{
+						int32_t i=getint(cmd);
+						if (i)
+							autop->interval(atol(s.c_str()));
+						else
+							Serial << "Bad value" << endl;
+					}
+				  else if (s=="view")
+					{
+						Serial << "  automatic "
+							<< (int32_t)autop->client
+							<< " interval " << autop->interval_
+							<< (autop->bon ? " on" : " off") << endl;
+					}
+					else
+					{
+						Serial << "Unknown auto command (" << s.c_str() << ")" << endl;
+						break;
+					}
+					s=getword(cmd);
+				}
+			}
+			else if (who==nullptr)
+			{
+				for(auto it: autos)
+					command(it.first, s+' '+cmd);
+			}
+			else
+				Serial << "what ? (" << s.c_str() << ")" << endl;
+		}
+
+		static void help()
+		{
+					Serial << "    auto [$id] on/off" << endl;
+					Serial << "    auto [$id] view" << endl;
+					Serial << "    auto [$id] create [millis] [topic]" << endl;
+		}
+
+	private:
+		MqttClient* client;
+		uint32_t interval_;
+		uint32_t timer_;
+		std::string topic_;
+		bool bon=false;
+		static std::map<MqttClient*, automatic*> autos;
+};
+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;
+	return strncmp(cmd, s.c_str(), s.length())==0;
+}
+
+using ClientFunction = void(*)(std::string& cmd, MqttClient* publish);
+
+void loop()
+{
+	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;
+			while(cmd.length())
+			{
+				std::string s;
+				MqttBroker* broker = nullptr;
+				MqttClient* client = nullptr;
+
+				// client.function notation
+				// ("a.fun " becomes "fun a ")
+				if (cmd.find('.') != std::string::npos)
+				{
+					s=getword(cmd, nullptr, '.');
+
+					if (clients.find(s) != clients.end())
+					{
+						client = clients[s];
+					}
+					else if (brokers.find(s) != brokers.end())
+					{
+						broker = brokers[s];
+					}
+					else
+					{
+						Serial << "Unknown class (" << s.c_str() << ")" << endl;
+						cmd="";
+					}
+				}
+				
+				s = getword(cmd);
+				if (compare(s, "delete"))
+				{
+					if (client==nullptr && broker==nullptr)
+					{
+						s = getword(cmd);
+						if (clients.find(s) != clients.end())
+						{
+							client = clients[s];
+						}
+						else if (brokers.find(s) != brokers.end())
+						{
+							broker = brokers[s];
+						}
+						else
+							Serial << "Unable to find (" << s.c_str() << ")" << endl;
+					}
+					if (client)
+					{
+						clients.erase(s);
+						for (auto it: clients)
+						{
+							if (it.second != client) continue;
+							Serial << "deleted" << endl;
+							clients.erase(it.first);
+							break;
+						}
+						cmd += " ls";
+					}
+					else if (broker)
+					{
+						for(auto it: brokers)
+						{
+							Serial << (int32_t)it.second << '/' << (int32_t)broker << endl;
+							if (broker != it.second) continue;
+							Serial << "deleted" << endl;
+							brokers.erase(it.first);
+							break;
+						}
+						cmd += " ls";
+					}
+					else
+						Serial << "Nothing to delete" << endl;
+				}
+				else if (broker)
+				{
+					if (compare(s,"connect"))
+				  {
+						Serial << "NYI" << endl;
+				  }
+					else if (compare(s, "view"))
+					{
+						broker->dump();
+					}
+			  }
+				else if (client)
+				{
+					if (compare(s,"connect"))
+					{
+						client->connect(getword(cmd,"192.168.1.40").c_str(), 1883);
+						Serial << (client->connected() ? "connected." : "not connected") << endl;
+					}
+					else if (compare(s,"publish"))
+					{
+						auto ok=client->publish(getword(cmd, topic.c_str()));
+						if (ok != MqttOk)
+						{
+							Serial << "## ERROR " << ok << endl;
+						}
+					}
+					else if (compare(s,"subscribe"))
+					{
+						client->subscribe(getword(cmd, topic.c_str()));
+					}
+					else if (compare(s, "view"))
+					{
+						client->dump();
+					}
+				}
+				else if (compare(s, "auto"))
+				{
+					automatic::command(client, cmd);
+					if (client == nullptr)
+						cmd.clear();
+				}
+				else if (compare(s, "broker"))
+				{
+					std::string id=getword(cmd);
+					if (id.length() or brokers.find(id)!=brokers.end())
+					{
+						int port=getint(cmd, 0);
+						if (port)
+						{
+							MqttBroker* broker = new MqttBroker(port);
+							broker->begin();
+
+							brokers[id] = broker;
+							Serial << "new broker (" << id.c_str() << ")" << endl;
+						}
+						else
+							Serial << "Missing port" << endl;
+					}
+					else
+						Serial << "Missing or existing broker name (" << id.c_str() << ")" << endl;
+					cmd+=" ls";
+				}
+				else if (compare(s, "client"))
+				{
+					std::string id=getword(cmd);
+					if (id.length() or clients.find(id)!=clients.end())
+					{
+						s=getword(cmd);	// broker name
+						if (s=="" or brokers.find(s) != brokers.end())
+						{
+							MqttBroker* broker = nullptr;
+							if (s.length()) broker = brokers[s];
+							MqttClient* client = new MqttClient(broker);
+							client->id(id);
+							clients[id]=client;
+							client->setCallback(onPublish);
+							client->subscribe(topic);
+							Serial << "new client (" << id.c_str() << ", " << s.c_str() << ')' << endl;
+						}
+						else if (s.length())
+						{
+							Serial << " not found." << endl;
+						}
+					}
+					else
+						Serial << "Missing or existing client name" << endl;
+					cmd+=" ls";
+				}
+				else if (compare(s, "ls") or compare(s, "view"))
+				{
+					Serial << "--< " << clients.size() << " client/s. >--" << endl;
+					for(auto it: clients)
+					{
+						Serial << "  "; it.second->dump();
+					}
+
+					Serial << "--< " << brokers.size() << " brokers/s. >--" << endl;
+					for(auto it: brokers)
+					{
+						Serial << " ==[ Broker: " << it.first.c_str() << " ]== ";
+						it.second->dump();
+					}
+				}
+				else if (compare(s, "reset"))
+					ESP.restart();
+				else if (compare(s, "ip"))
+					Serial << "IP: " << WiFi.localIP() << endl;
+				else if (compare(s,"help"))
+				{
+					Serial << "syntax:" << endl;
+					Serial << "  MqttBroker:" << endl;
+					Serial << "    broker {name} {port} : create a new broker" << endl;
+					Serial << endl;
+					Serial << "  MqttClient:" << endl;
+					Serial << "    client {name} {parent broker} : create a client then" << endl;
+					Serial << "      name.connect  [ip]" << endl;
+					Serial << "      name.subscribe [topic]" << endl;
+					Serial << "      name.publish [topic]" << endl;
+					Serial << "      name.view" << endl;
+					Serial << "      name.delete" << endl;
+
+					automatic::help();
+					Serial << endl;
+					Serial << "    help" << endl;
+					Serial << "    ls / ip / reset" << endl;
+					Serial << "    !  repeat last command" << endl;
+					Serial << endl;
+					Serial << "  $id : name of the client." << endl;
+					Serial << "  default topic is '" << topic.c_str() << "'" << endl;
+					Serial << endl;
+				}
+				else
+				{
+					while(s[0]==' ') s.erase(0,1);
+					if (s.length())
+						Serial << "Unknown command (" << s.c_str() << ")" << endl;
+				}
+			}
+		}
+		else
+		{
+			cmd=cmd+c;
+		}
+	}
+}

keywords.txt

@@ -3,16 +3,23 @@
 #######################################
 
 #######################################
-# Datatypes (KEYWORD1)
+# Datatypes and functions
 #######################################
 
-MqttBroker	KEYWORD1
+TinyMqtt    KEYWORD1
+
+MqttBroker  KEYWORD1
+begin				KEYWORD2
+loop				KEYWORD2
+
 MqttClient  KEYWORD1
+publish     KEYWORD2
+setCallback	KEYWORD2
+subscribe	  KEYWORD2
 
-#######################################
-# Methods and Functions (KEYWORD2)
-#######################################
-
+Topic				KEYWORD1
+matches			KEYWORD2
+c_str				KEYWORD2
 
 #######################################
 # Constants (LITERAL1)

library.json

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

library.properties

@@ -1,9 +1,10 @@
 name=TinyMqtt
-version=0.1
-author=HSaturn <hsaturn@gmail.com>
-maintainer=HSaturn <hsaturn@gmail.com>
+version=0.4.0
+author=Francois BIOT, HSaturn, <hsaturn@gmail.com>
+maintainer=Francois BIOT, HSaturn, <hsaturn@gmail.com>
 sentence=A tiny broker and client library for MQTT messaging.
-paragraph=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.
+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.
 category=Communication
 url=https://github.com/hsaturn/TinyMqtt
 architectures=*
+includes=TinyMqtt.h

src/StringIndexer.h

@@ -2,7 +2,7 @@
 #include <map>
 #include <string>
 #include <string.h>
-#include <Streaming.h>
+// #include <Streaming.h>
 #include <ESP8266WiFi.h>
 
 /***
@@ -36,7 +36,7 @@ class StringIndexer
 				{
 					strings[index].str = std::string(str, len);
 					strings[index].used++;
-					Serial << "Creating index " << index << " for (" << strings[index].str.c_str() << ") len=" << len << endl;
+					// Serial << "Creating index " << index << " for (" << strings[index].str.c_str() << ") len=" << len << endl;
 					return index;
 				}
 			}
@@ -66,7 +66,7 @@ class StringIndexer
 				if (it->second.used == 0)
 				{
 					strings.erase(it);
-					Serial << "Removing string(" << it->second.str.c_str() << ") size=" << strings.size() << endl;
+					// Serial << "Removing string(" << it->second.str.c_str() << ") size=" << strings.size() << endl;
 				}
 			}
 		}

src/TinyMqtt.cpp

@@ -10,32 +10,101 @@ 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(port)
 {
 }
 
-MqttCnx::MqttCnx(MqttBroker* parent, WiFiClient& new_client)
-	: parent(parent),
-	mqtt_connected(false)
+MqttBroker::~MqttBroker()
 {
-	client = new_client ? new WiFiClient(new_client) : nullptr;
-	clientAlive();
+	while(clients.size())
+	{
+		delete clients[0];
+	}
 }
 
-MqttCnx::~MqttCnx()
+// private constructor used by broker only
+MqttClient::MqttClient(MqttBroker* parent, WiFiClient& new_client)
+	: parent(parent)
+{
+	client = new WiFiClient(new_client);
+	alive = millis()+5000;	// client expires after 5s if no CONNECT msg
+}
+
+MqttClient::MqttClient(MqttBroker* parent)
+	: parent(parent)
+{
+		client = nullptr;
+
+		if (parent) parent->addClient(this);
+}
+
+MqttClient::~MqttClient()
 {
 	close();
+	delete client;
 }
 
-void MqttCnx::close()
+void MqttClient::close()
 {
+	debug("close " << id().c_str());
+	mqtt_connected = false;
 	if (client)
 	{
 		client->stop();
-		delete client;
-		client = nullptr;
 	}
+
+	if (parent)
+	{
+		parent->removeClient(this);
+		parent=nullptr;
+	}
+}
+
+void MqttClient::connect(std::string broker, uint16_t port)
+{
+	debug("cnx: closing");
+	close();
+	if (client) delete client;
+	client = new WiFiClient;
+	debug("Trying to connect to " << broker.c_str() << ':' << port);
+	if (client->connect(broker.c_str(), port))
+	{
+	  debug("cnx: connecting");
+		message.create(MqttMessage::Type::Connect);
+		message.add("MQTT",4);
+		message.add(0x4);	// Mqtt protocol version 3.1.1
+		message.add(0x0);	// Connect flags         TODO user / name
+
+		keep_alive = 1;			// TODO not configurable
+		message.add(0x00); // keep_alive
+		message.add((char)keep_alive);
+		message.add(clientId);
+	  debug("cnx: mqtt connecting");
+		message.sendTo(this);
+	  debug("cnx: mqtt sent " << (int32_t)parent);
+		clientAlive(0);
+	}
+}
+
+void MqttBroker::addClient(MqttClient* client)
+{
+		clients.push_back(client);
+}
+
+void MqttBroker::removeClient(MqttClient* remove)
+{
+  for(auto it=clients.begin(); it!=clients.end(); it++)
+	{
+		auto client=*it;
+		if (client==remove)
+		{
+			debug("Remove " << clients.size());
+			clients.erase(it);
+			debug("Client removed " << clients.size());
+			return;
+		}
+	}
+	debug("Error cannot remove client");	// TODO should not occur
 }
 
 void MqttBroker::loop()
@@ -44,31 +113,61 @@ void MqttBroker::loop()
   
   if (client)
 	{
-		clients.push_back(new MqttCnx(this, client));
-		Serial << "New client (" << clients.size() << ')' << endl;
+		addClient(new MqttClient(this, client));
+		debug("New client (" << clients.size() << ')');
 	}
 
-  for(auto it=clients.begin(); it!=clients.end(); it++)
+  // 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++)
 	{
-		auto client=*it;
-    if(client->connected())
+		auto client = clients[i];
+    if (client->connected())
     {
 			client->loop();
 		}
 		else
 		{
-      Serial << "Client " << client->id().c_str() << "  Disconnected" << endl;
-			clients.erase(it);
+      debug("Client " << client->id().c_str() << "  Disconnected, parent=" << (int32_t)client->parent);
+			// Note: deleting a client not added by the broker itself will probably crash later.
 			delete client;
 			break;
     }
 	}
 }
 
-void MqttBroker::publish(const Topic& topic, MqttMessage& msg)
+MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, MqttMessage& msg)
 {
+	MqttError retval = MqttOk;
+
+	debug("publish ");
+	int i=0;
 	for(auto client: clients)
-		client->publish(topic, msg);
+	{
+		i++;
+		Serial << "brk_" << (broker && broker->connected() ? "con" : "dis") <<
+			 "	srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << i << ", local=" << client->isLocal() << ", con=" << client->connected();
+		bool doit = false;
+		if (broker && broker->connected())	// Connected: R2 R3 R5 R6
+		{
+			//    ext broker -> clients or
+			// or clients -> ext broker
+			if (source == broker)	// broker -> clients
+				doit = true;
+			else									// clients -> broker
+				retval=broker->publish(topic, msg);
+		}
+		else // Disconnected: R7
+		{
+			// All is allowed
+			doit = true;
+		}
+		Serial << ", doit=" << doit << ' ';
+
+		if (doit) client->publish(topic, msg);
+		debug("");
+	}
+	return retval;
 }
 
 bool MqttBroker::compareString(
@@ -81,28 +180,41 @@ bool MqttBroker::compareString(
 	return *good==0;
 }
 
-void MqttMessage::getString(char* &buffer, uint16_t& len)
+void MqttMessage::getString(const char* &buff, uint16_t& len)
 {
-	len = (buffer[0]<<8)|(buffer[1]);
-	buffer+=2;
+	len = (buff[0]<<8)|(buff[1]);
+	buff+=2;
 }
 
-void MqttCnx::clientAlive()
+void MqttClient::clientAlive(uint32_t more_seconds)
 {
 	if (keep_alive)
 	{
-		alive=millis()+1000*(keep_alive+5);
+		alive=millis()+1000*(keep_alive+more_seconds);
 	}
 	else
 		alive=0;
 }
 
-void MqttCnx::loop()
+void MqttClient::loop()
 {
 	if (alive && (millis() > alive))
 	{
-		Serial << "timeout client" << endl;
-		close();
+		if (parent)
+		{
+			debug("timeout client");
+			close();
+			debug("closed");
+		}
+		else if (client && client->connected())
+		{
+			uint16_t pingreq = MqttMessage::Type::PingReq;
+			client->write((uint8_t*)(&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)
@@ -115,32 +227,58 @@ void MqttCnx::loop()
 	}
 }
 
-void MqttCnx::processMessage()
+void MqttClient::processMessage()
 {
 	std::string error;
 	std::string s;
 	// Serial << "---> INCOMING " << _HEX(message.type()) << ", mem=" << ESP.getFreeHeap() << endl;
   auto header = message.getVHeader();
-  char* payload;
+  const char* payload;
   uint16_t len;
 	bool bclose=true;
 
 	switch(message.type() & 0XF0)
 	{
 		case MqttMessage::Type::Connect:
-			if (mqtt_connected) break;
+			if (mqtt_connected)
+			{
+				debug("already connected");
+				break;
+			}
 			payload = header+10;
-			flags = header[7];
+			mqtt_flags = header[7];
 			keep_alive = (header[8]<<8)|(header[9]);
-			if (strncmp("MQTT", header+2,4)) break;
-			if (header[6]!=0x04) break;	// Level 3.1.1
+			if (strncmp("MQTT", header+2,4))
+			{
+				debug("bad mqtt header");
+				break;
+			}
+			if (header[6]!=0x04)
+			{
+				debug("unknown level");
+				break;	// Level 3.1.1
+			}
 
 			// ClientId
 			message.getString(payload, len);
+			debug("client id len=" << len);
+			if (len>30)
+			{
+				Serial << '(';
+				for(int i=0; i<30; i++)
+				{
+					if (i%5==0) Serial << ' ';
+					char c=*(header+i);
+					Serial << (c < 32 ? '.' : c);
+				}
+				Serial << " )" << endl;
+				debug("Bad client id length");
+				break;
+			}
 			clientId = std::string(payload, len);
 			payload += len;
 
-			if (flags & FlagWill)	// Will topic
+			if (mqtt_flags & FlagWill)	// Will topic
 			{
 				message.getString(payload, len);	// Will Topic
 				outstring("WillTopic", payload, len);
@@ -150,13 +288,14 @@ void MqttCnx::processMessage()
 				outstring("WillMessage", payload, len);
 				payload += len;
 			}
-			if (flags & FlagUserName)
+			// FIXME forgetting credential is allowed (security hole)
+			if (mqtt_flags & FlagUserName)
 			{
 				message.getString(payload, len);
 				if (!parent->checkUser(payload, len)) break;
 				payload += len;
 			}
-			if (flags & FlagPassword)
+			if (mqtt_flags & FlagPassword)
 			{
 				message.getString(payload, len);
 				if (!parent->checkPassword(payload, len)) break;
@@ -174,10 +313,17 @@ void MqttCnx::processMessage()
 			break;
 
 		case MqttMessage::Type::PingReq:
-			message.create(MqttMessage::Type::PingResp);
-			message.add(0);
-			message.sendTo(this);
-			bclose = false;
+			if (!mqtt_connected) break;
+			if (client)
+			{
+				uint16_t pingreq = MqttMessage::Type::PingResp;
+				client->write((uint8_t*)(&pingreq), 2);
+			  bclose = false;
+			}
+			else
+			{
+				debug("internal pingreq ?");
+			}
 			break;
 
 		case MqttMessage::Type::Subscribe:
@@ -186,7 +332,7 @@ void MqttCnx::processMessage()
 			message.getString(payload, len);	// Topic
 			outstring("Subscribes", payload, len);
 
-			subscriptions.insert(Topic(payload, len));
+			subscribe(Topic(payload, len));
 			bclose = false;
 			// TODO SUBACK
 			break;
@@ -205,7 +351,8 @@ void MqttCnx::processMessage()
 				if (qos) payload+=2;	// ignore packet identifier if any
 				// TODO reset DUP
 				// TODO reset RETAIN
-				parent->publish(published, message);
+				debug("publishing to parent");
+				parent->publish(this, published, message);
 				// TODO should send PUBACK
 				bclose = false;
 			}
@@ -229,7 +376,7 @@ void MqttCnx::processMessage()
   }
 	else
 	{
-		clientAlive();
+		clientAlive(5);
 	}
 	message.reset();
 }
@@ -241,29 +388,57 @@ bool Topic::matches(const Topic& topic) const
 	return false;
 }
 
-void MqttCnx::publish(const Topic& topic, MqttMessage& msg)
+// publish from local client
+MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pay_length)
 {
+	MqttMessage msg;
+	msg.create(MqttMessage::Publish);
+	msg.add(topic);
+	msg.add(payload, pay_length, false);
+	if (parent)
+		return parent->publish(this, topic, msg);
+	else if (client)
+		msg.sendTo(this);
+	else
+		return MqttNowhereToSend;
+}
+
+// republish a received publish if it matches any in subscriptions
+MqttError MqttClient::publish(const Topic& topic, MqttMessage& msg)
+{
+	MqttError retval=MqttOk;
+
+	debug("mqttclient publish " << subscriptions.size());
 	for(const auto& subscription: subscriptions)
 	{
+		Serial << " client=" << (int32_t)client << ", topic " << topic.str().c_str() << ' ';
 		if (subscription.matches(topic))
 		{
-			// Serial << "Republishing " << topic.str().c_str() << " to " << clientId.c_str() << endl;
-			msg.sendTo(this);
+			Serial << " match/send";
+			if (client)
+			{
+				msg.sendTo(this);
+			}
+			else if (callback)
+			{
+				callback(this, topic, nullptr, 0);	// TODO Payload
+			}
 		}
+		Serial << endl;
 	}
+	return retval;
 }
 
 void MqttMessage::reset()
 {
-	curr=buffer;
-	*curr=0;	// Type Unknown
+	buffer.clear();
 	state=FixedHeader;
 	size=0;
 }
 
 void MqttMessage::incoming(char in_byte)
 {
-	*curr++ = in_byte;
+	buffer += in_byte;
 	switch(state)
 	{
 		case FixedHeader:
@@ -278,7 +453,7 @@ void MqttMessage::incoming(char in_byte)
 			}
 			else if ((in_byte & 0x80) == 0)
 			{
-				vheader = curr;
+				vheader = buffer.length();
 				if (size==0)
 					state = Complete;
 				else
@@ -299,18 +474,27 @@ void MqttMessage::incoming(char in_byte)
 			break;
 		case Complete:
 		default:
-			curr--;
 			Serial << "Spurious " << _HEX(in_byte) << endl;
-			state = Error;
+			reset();
 			break;
 	}
-	if (curr-buffer > 250)
+	if (buffer.length() > MaxBufferLength)	// TODO magic 256 ?
 	{
-		Serial << "Too much incoming bytes." << endl;
-		curr=buffer;
+		debug("Too long " << state);
+		reset();
 	}
 }
 
+void MqttMessage::add(const char* p, size_t len, bool addLength)
+{
+	if (addLength)
+	{
+		incoming(len>>8);
+		incoming(len & 0xFF);
+	}
+	while(len--) incoming(*p++);
+}
+
 void MqttMessage::encodeLength(char* msb, int length)
 {
 	do
@@ -322,31 +506,28 @@ void MqttMessage::encodeLength(char* msb, int length)
 	} while (length);
 };
 
-void MqttMessage::sendTo(MqttCnx* client)
+void MqttMessage::sendTo(MqttClient* client)
 {
-	if (curr-buffer-2 >= 0)
+	if (buffer.size()>2)
 	{
-		encodeLength(buffer+1, curr-buffer-2);
+		encodeLength(&buffer[1], buffer.size()-2);
 		// hexdump("snd");
-		client->write(buffer, curr-buffer);
+		client->write(&buffer[0], buffer.size());
 	}
 	else
 	{
 		Serial << "??? Invalid send" << endl;
-		Serial << (long)end() << "-" << (long)buffer << endl;
 	}
 }
 
 void MqttMessage::hexdump(const char* prefix) const
 {
 	if (prefix) Serial << prefix << ' ';
-	Serial << (long)buffer << "-" << (long)curr << " : ";
-		const char* p=buffer;
-		while(p!=curr)
+	Serial << "size(" << buffer.size() << ") : ";
+		for(const char chr: buffer)
 		{
-			if (*p<16) Serial << '0';
-			Serial << _HEX(*p) << ' ';
-			p++;
+			if (chr<16) Serial << '0';
+			Serial << _HEX(chr) << ' ';
 		}
 		Serial << endl;
 }

src/TinyMqtt.h

@@ -4,20 +4,35 @@
 #include <string>
 #include "StringIndexer.h"
 
-#define MaxBufferLength 255
+#ifdef TINY_MQTT_DEBUG
+  #include <Streaming.h>
+  #define debug(what) { Serial << __LINE__ << ' ' << what << endl; delay(100); }
+#else
+  #define debug(what) {}
+#endif
+
+enum MqttError
+{
+	MqttOk = 0,
+	MqttNowhereToSend=1,
+};
 
 class Topic : public IndexedString
 {
 	public:
 		Topic(const char* s, uint8_t len) : IndexedString(s,len){}
 		Topic(const char* s) : Topic(s, strlen(s)) {}
+		Topic(const std::string s) : Topic(s.c_str(), s.length()){};
+
+		const char* c_str() const { return str().c_str(); }
 
 		bool matches(const Topic&) const;
 };
 
-class MqttCnx;
+class MqttClient;
 class MqttMessage
 {
+	const uint16_t MaxBufferLength = 255;
 	public:
 		enum Type
 		{
@@ -45,15 +60,18 @@ class MqttMessage
 		MqttMessage(Type t) { create(t); }
 		void incoming(char byte);
 		void add(char byte) { incoming(byte); }
-		char* getVHeader() const { return vheader; }
-		char* end() const { return curr; }
-		uint16_t length() const { return curr-buffer; }
+		void add(const char* p, size_t len, bool addLength=true );
+		void add(const std::string& s) { add(s.c_str(), s.length()); }
+		void add(const Topic& t) { add(t.str()); }
+		const char* end() const { return &buffer[0]+buffer.size(); }
+		const char* getVHeader() const { return &buffer[vheader]; }
+		uint16_t length() const { return buffer.size(); }
 
 		void reset();
 
 		// buff is MSB/LSB/STRING
 		// output buff+=2, len=length(str)
-		void getString(char* &buff, uint16_t& len);
+		static void getString(const char* &buff, uint16_t& len);
 
 
 		Type type() const
@@ -63,98 +81,171 @@ class MqttMessage
 
 		void create(Type type)
 		{
-			buffer[0]=type;
-			curr=buffer+2;
-			vheader=curr;
+			buffer=(char)type;
+			buffer+='\0';
+			vheader=2;
 			size=0;
 			state=Create;
 		}
-		void sendTo(MqttCnx*);
+		void sendTo(MqttClient*);
 		void hexdump(const char* prefix=nullptr) const;
 
 	private:
 		void encodeLength(char* msb, int length);
 
-	  char buffer[256];	// TODO 256 ?
-		char* vheader;
-		char* curr;
+		std::string buffer;
+		uint8_t vheader;
 		uint16_t size;	// bytes left to receive
 		State state;
 };
 
 class MqttBroker;
-class MqttCnx
+class MqttClient
 {
+	using CallBack = void (*)(const MqttClient* source, const Topic& topic, const char* payload, size_t payload_length);
 	enum Flags
 	{
 		FlagUserName = 128,
 		FlagPassword = 64,
 		FlagWillRetain = 32,	// unsupported
 		FlagWillQos = 16 | 8, // unsupported
-		FlagWill = 4,			// unsupported
+		FlagWill = 4,			    // unsupported
 		FlagCleanSession = 2,	// unsupported
 		FlagReserved = 1
 	};
 	public:
-		MqttCnx(MqttBroker* parent, WiFiClient& client);
+		MqttClient(MqttBroker*);
+		MqttClient() : MqttClient(nullptr) {};
 
-		~MqttCnx();
+		~MqttClient();
 
-		bool connected() { return client && client->connected(); }
+		void connect(MqttBroker* parent);
+		void connect(std::string broker, uint16_t port);
+
+		bool connected() { return
+			(parent!=nullptr and client==nullptr) or
+			(client and client->connected()); }
 		void write(const char* buf, size_t length)
 		{ if (client) client->write(buf, length); }
 
 		const std::string& id() const { return clientId; }
+		void id(std::string& new_id) { clientId = new_id; }
 
 		void loop();
 		void close();
-		void publish(const Topic& topic, MqttMessage& msg);
+		void setCallback(CallBack fun) {callback=fun; };
+
+		// Publish from client to the world
+		MqttError publish(const Topic&, const char* payload, size_t pay_length);
+		MqttError publish(const Topic& t, const String& s) { return publish(t, s.c_str(), s.length()); }
+		MqttError publish(const Topic& t, const std::string& s) { return publish(t,s.c_str(),s.length());}
+		MqttError publish(const Topic& t) { return publish(t, nullptr, 0);};
+
+		void subscribe(Topic topic) { subscriptions.insert(topic); }
+		void unsubscribe(Topic& topic);
+
+		// connected to local broker
+		// TODO seems to be useless
+		bool isLocal() const { return client == nullptr; }
+
+#ifdef TINY_MQTT_DEBUG
+		void dump()
+		{
+			Serial << "MqttClient (" << clientId.c_str() << ") p=" << (int32_t) parent
+				<< " c=" << (int32_t)client << (connected() ? " ON " : " OFF"); 
+			Serial << " [";
+			bool c=false;
+			for(auto s: subscriptions)
+			{
+				Serial << (c?", ": "")<< s.str().c_str();
+				c=true;
+			}
+			Serial << "]" << endl;
+		}
+#endif
 
 	private:
-		void clientAlive();
+		friend class MqttBroker;
+		MqttClient(MqttBroker* parent, WiFiClient& client);
+		// republish a received publish if topic matches any in subscriptions
+		MqttError publish(const Topic& topic, MqttMessage& msg);
+
+		void clientAlive(uint32_t more_seconds);
 		void processMessage();
 
-		char flags;
+		bool mqtt_connected = false;
+		char mqtt_flags;
 		uint32_t keep_alive;
 		uint32_t alive;
-		bool mqtt_connected;
-		WiFiClient* client;
 		MqttMessage message;
-		MqttBroker* parent;
+
+		// TODO	having a pointer on MqttBroker may produce larger binaries
+		// due to unecessary function linked if ever parent is not used
+		// (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
 		std::set<Topic>	subscriptions;
 		std::string clientId;
-};
-
-class MqttClient
-{
-	public:
-		MqttClient(IPAddress broker) : broker_ip(broker) {}
-
-	protected:
-		IPAddress broker_ip;
+		CallBack callback = nullptr;
 };
 
 class MqttBroker
 {
+	enum State
+	{
+		Disconnected,	// Also the initial state
+		Connecting,		// connect and sends a fake publish to avoid circular cnx
+		Connected,		// this->broker is connected and circular cnx avoided
+	};
 	public:
+	  // TODO limit max number of clients
 		MqttBroker(uint16_t port);
+		~MqttBroker();
 
 		void begin() { server.begin(); }
 		void loop();
 
+		uint8_t port() const { return server.port(); }
+
+		void connect(std::string host, uint32_t port=1883);
+		bool connected() const { return state == Connected; }
+
+#ifdef TINY_MQTT_DEBUG
+		void dump()
+		{
+			Serial << clients.size() << " client/s" << endl;
+			for(auto client: clients)
+			{
+				Serial << "  ";
+				client->dump();
+			}
+		}
+#endif
+
+	private:
+		friend class MqttClient;
+
 		bool checkUser(const char* user, uint8_t len) const
 		{ return compareString(auth_user, user, len); }
 
 		bool checkPassword(const char* password, uint8_t len) const
 		{ return compareString(auth_password, password, len); }
 
-		void publish(const Topic& topic, MqttMessage& msg);
 
-	private:
+		MqttError publish(const MqttClient* source, const Topic& topic, MqttMessage& msg);
+
+		// For clients that are added not by the broker itself
+		void addClient(MqttClient* client);
+		void removeClient(MqttClient* client);
+
 		bool compareString(const char* good, const char* str, uint8_t str_len) const;
-		std::vector<MqttCnx*>	clients;
+		std::vector<MqttClient*>	clients;
 		WiFiServer server;
 
 		const char* auth_user = "guest";
 		const char* auth_password = "guest";
+		State state = Disconnected;
+
+		MqttClient* broker = nullptr;
 };

src/my_credentials.h

@@ -0,0 +1,2 @@
+const char *ssid     = "YOUR-SSID-HERE";
+const char *password = "YOUR-PASSWORD-HERE";