[fixes] Timeout and broker to broker modifications

README.md

@@ -24,6 +24,7 @@ TinyMqtt is a small, fast and capable Mqtt Broker and Client for Esp8266 / Esp32
 
 ## Features
 
+- Async Wifi compatible (me-no-dev/ESPAsyncTCP@^1.2.2)
 - Very fast broker I saw it re-sent 1000 topics per second for two
   clients that had subscribed (payload ~15 bytes ESP8266). No topic lost.
   The max I've seen was 2k msg/s (1 client 1 subscription)

bump_version.sh

@@ -1,16 +1,17 @@
 #!/bin/bash
 current_version=$(git describe --tags --abbrev=0)
-cp library.json.skeleton library.json
-while ifs= read -r line; do
-  name=$(echo "$line" | sed "s/=.*//g")
-  value=$(echo "$line" | cut -d= -f 2 | sed 's/"//g')
-  sed -i "s/#$name/$value/g" library.json
-done < library.properties
-
+if [ "$1" == "-d" ]; then
+  do=0
+  shift
+else
+  do=1
+fi
 if [ "$1" == "" ]; then
 	echo
-	echo "Syntax: $0 {new_version}"
+	echo "Syntax: $0 [-d] {new_version}"
 	echo
+	echo "  -d : dry run, generate json and update properties but do not run git commands"
+	echo ""
 	echo "  Current version: $current_version"
 	echo
 else
@@ -23,15 +24,35 @@ else
 		grep $current_version library.properties
 		if [ "$?" == "0" ]; then
 			sed -i "s/$current_version/$1/" library.properties
-			if [ 0 == 1 ]; then
+
+      cp library.json.skeleton library.json
+      while ifs= read -r line; do
+        name=$(echo "$line" | sed "s/=.*//g")
+        value=$(echo "$line" | cut -d= -f 2 | sed 's/"//g')
+        echo "  Replacing $name in json"
+        if [ "$name" == "depends" ]; then
+          depends=$(echo "$value" | sed "s/,/ /g")
+          echo "  Depends=$depends"
+        fi
+        echo "  " sed -i "s@#$name@$value@g" library.json
+        sed -i "s@#$name@$value@g" library.json
+      done < library.properties
+      deps=""
+      for depend in $depends; do
+        if [ "$deps" != "" ]; then
+          deps="$deps, "
+        fi
+        deps="$deps'$depend' : '*'"
+      done
+      sed -i "s@#dependencies@$deps@g" library.json
+      sed -i "s/'/\"/g" library.json
+      if [ "$do" == "1" ]; then
         git tag $1
         git add library.properties
         git add library.json
         git commit -m "Release $1"
         git push
         git push --tags
-      else
-        echo "No git operation made"
       fi
 		else
 			echo "Current version does not match library.property version, aborting"

examples/tinymqtt-test/tinymqtt-test.ino

@@ -1,22 +1,20 @@
 // vim: ts=2 sw=2 expandtab smartindent
 #include <TinyConsole.h>
-#include <TinyMqtt.h>       // https://github.com/hsaturn/TinyMqtt
+#include <TinyMqtt.h> // https://github.com/hsaturn/TinyMqtt
 #include <TinyStreaming.h>
 #if defined(ESP8266)
-  #include <ESP8266mDNS.h>
+#include <ESP8266mDNS.h>
 #elif defined(ESP32)
-  #include <WiFi.h>
-  #include <ESPmDNS.h>
+#include <WiFi.h>
+#include <ESPmDNS.h>
 #else
-  #error Unsupported platform
+#error Unsupported platform
 #endif
 
 #include <sstream>
 #include <string>
 #include <map>
 
-using string = TinyString;
-
 bool echo_on = true;
 auto green = TinyConsole::green;
 auto red = TinyConsole::red;
@@ -28,45 +26,47 @@ auto save_cursor = TinyConsole::save_cursor;
 auto restore_cursor = TinyConsole::restore_cursor;
 auto erase_to_end = TinyConsole::erase_to_end;
 
-const char* ssid = "";
-const char* password = "";
+const char *ssid = "Freebox-786A2F";
+const char *password = "usurpavi8dalum64lumine?";
+
+void onCommand(const string &command);
+void eval(string &cmd);
 
 struct free_broker
 {
-  public:
-    free_broker(const char* s, uint16_t p, const char* /* comment */) : url(s), port(p) {}
+public:
+  free_broker(const char *s, uint16_t p, const char * /* comment */) : url(s), port(p) {}
 
-    string url;
-    uint16_t port;
+  string url;
+  uint16_t port;
 };
 
 const std::map<string, free_broker> list =
-{
-  { "mqtthq", { "public.mqtthq.com" , 8083, "publish/subscribe" }},
-  { "hivemq", { "broker.hivemq.com", 1883, "" }}
-};
+    {
+        {"mqtthq", {"public.mqtthq.com", 8083, "publish/subscribe"}},
+        {"hivemq", {"broker.hivemq.com", 1883, ""}}};
 
 /** Very complex example
-  * Console allowing to make any kind of test,
-  * even some stress tests.
-  *
-  * Upload the sketch, the use the terminal.
-  * Press H for mini help.
-  *
-  * tested with mqtt-spy-0.5.4
-  * TODO examples of scripts
-  */
+ * Console allowing to make any kind of test,
+ * even some stress tests.
+ *
+ * Upload the sketch, the use the terminal.
+ * Press H for mini help.
+ *
+ * tested with mqtt-spy-0.5.4
+ * TODO examples of scripts
+ */
 
-void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
+void onPublish(const MqttClient *srce, const Topic &topic, const char *payload, size_t length)
 {
   Console << cyan << "--> " << srce->id().c_str() << ": received " << topic.c_str() << white;
   if (payload)
   {
     Console << ", payload[" << length << "]=[";
-    while(length--)
+    while (length--)
     {
-      const char c=*payload++;
-      if (c<32)
+      const char c = *payload++;
+      if (c < 32)
         Console << '?';
       else
         Console << c;
@@ -75,8 +75,8 @@ void onPublish(const MqttClient* srce, const Topic& topic, const char* payload,
   }
 }
 
-std::map<string, MqttClient*> clients;
-std::map<string, MqttBroker*> brokers;
+std::map<string, MqttClient *> clients;
+std::map<string, MqttBroker *> brokers;
 
 void setup()
 {
@@ -88,7 +88,8 @@ void setup()
   delay(500);
 
   Console.cls();
-  Console << endl << endl;
+  Console << endl
+          << endl;
   Console << yellow
           << "***************************************************************" << endl;
   Console << "*  Welcome to the TinyMqtt console" << endl;
@@ -96,10 +97,11 @@ void setup()
   Console << "*  Enter help to view the list of commands." << endl;
   Console << "***************************************************************" << endl;
   Console << endl;
-  if (strlen(ssid)==0)
+  if (strlen(ssid) == 0)
     Console << red << "*  ERROR: You must modify ssid/password in order" << endl
-                   << "          to be able to connect to your Wifi network." << endl;
-  Console << endl << white;
+            << "          to be able to connect to your Wifi network." << endl;
+  Console << endl
+          << white;
 
   Console << "Connecting to '" << ssid << "' ";
 
@@ -109,123 +111,139 @@ void setup()
   WiFi.begin(ssid, password);
 
   while (WiFi.status() != WL_CONNECTED)
-  { Console << '-'; delay(500); }
+  {
+    Console << '-';
+    delay(500);
+  }
 
-  Console << endl << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
+  Console << endl
+          << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
 
-  const char* name="tinytest";
+  const char *name = "tinytest";
   Console << "Starting MDNS, name= " << name;
   if (!MDNS.begin(name))
     Console << "  error, not available." << endl;
   else
     Console << " ok." << endl;
 
-
-  MqttBroker* broker = new MqttBroker(1883);
+  MqttBroker *broker = new MqttBroker(1883);
   broker->begin();
   brokers["broker"] = broker;
 
-  if (Console.isTerm()) onCommand("every 333 view");
+  if (Console.isTerm())
+    onCommand("every 333 view");
+
+  Console.prompt();
 }
 
-string getword(string& str, const char* if_empty=nullptr, char sep=' ');
+string getword(string &str, const char *if_empty = nullptr, char sep = ' ');
 
-int getint(string& str, const int if_empty=0)
+int getint(string &str, const int if_empty = 0)
 {
-  string str2=str;
+  string str2 = str;
   string sword = getword(str);
   if (sword[0] and isdigit(sword[0]))
   {
-    int ret=atoi(sword.c_str());
-    while(isdigit(sword[0]) or sword[0]==' ') sword.erase(0,1);
-    if (sword.length()) str = sword+' '+str;
+    int ret = atoi(sword.c_str());
+    while (isdigit(sword[0]) or sword[0] == ' ')
+      sword.erase(0, 1);
+    if (sword.length())
+      str = sword + ' ' + str;
     return ret;
   }
-  str=str2;
+  str = str2;
   return if_empty;
 }
 
-string getword(string& str, const char* if_empty/*=nullptr*/, char sep/*=' '*/)
+string getword(string &str, const char *if_empty /*=nullptr*/, char sep /*=' '*/)
 {
-  char quote=(str[0]=='"' or str[0]=='\'' ? str[0] : 0);
-  if (quote) str.erase(0,1);
+  char quote = (str[0] == '"' or str[0] == '\'' ? str[0] : 0);
+  if (quote)
+    str.erase(0, 1);
   string sword;
-  while(str.length() and (str[0]!=sep or quote))
+  while (str.length() and (str[0] != sep or quote))
   {
-    if (str[0]==quote)
+    if (str[0] == quote)
     {
-      str.erase(0,1);
+      str.erase(0, 1);
       break;
     }
     sword += str[0];
-    str.erase(0,1);
+    str.erase(0, 1);
   }
-  while(str[0]==sep) str.erase(0,1);
-  if (if_empty and sword.length()==0) return if_empty;
-  if (quote==false and sword.length()>=4 and sword.substr(0,3)=="rnd")
+  while (str[0] == sep)
+    str.erase(0, 1);
+  if (if_empty and sword.length() == 0)
+    return if_empty;
+  if (quote == false and sword.length() >= 4 and sword.substr(0, 3) == "rnd")
   {
-    sword.erase(0,3);
-    if (sword[0]=='(')
+    sword.erase(0, 3);
+    if (sword[0] == '(')
     {
       int to = 100;
-      sword.erase(0,1);
-      int from=getint(sword);
-      if (sword[0]==',')
+      sword.erase(0, 1);
+      int from = getint(sword);
+      if (sword[0] == ',')
       {
-        sword.erase(0,1);
+        sword.erase(0, 1);
         to = getint(sword);
-        if (sword[0]!=')') Console << "Missing ')'" << endl;
+        if (sword[0] != ')')
+          Console << "Missing ')'" << endl;
       }
       else
       {
-        to=from;
-        from=0;
+        to = from;
+        from = 0;
       }
-      return String(random(from,to)).c_str();
+      return String(random(from, to)).c_str();
     }
     else
     {
       Console << "Missing '('" << endl;
     }
   }
-  while(str[0]==' ') str.erase(0,1);
+  while (str[0] == ' ')
+    str.erase(0, 1);
   return sword;
 }
 
 bool isaddr(string s)
 {
-  if (s.length()==0 or s.length()>3) return false;
-  for(char c: s)
-    if (c<'0' or c>'9') return false;
+  if (s.length() == 0 or s.length() > 3)
+    return false;
+  for (char c : s)
+    if (c < '0' or c > '9')
+      return false;
   return true;
 }
 
-string getip(string& str, const char* if_empty=nullptr, char sep=' ')
+string getip(string &str, const char *if_empty = nullptr, char sep = ' ')
 {
-  string addr=getword(str, if_empty, sep);
-  string ip=addr;
+  string addr = getword(str, if_empty, sep);
+  string ip = addr;
   std::vector<string> build;
-  while(ip.length())
+  while (ip.length())
   {
-    string b=getword(ip,nullptr,'.');
-    if (isaddr(b) && build.size()<4)
+    string b = getword(ip, nullptr, '.');
+    if (isaddr(b) && build.size() < 4)
     {
       build.push_back(b);
     }
     else
       return addr;
   }
-  IPAddress local=WiFi.localIP();
+  IPAddress local = WiFi.localIP();
   addr.clear();
-  while(build.size()!=4)
+  while (build.size() != 4)
   {
     std::stringstream b;
-    b << (int)local[3-build.size()];
+    b << (int)local[3 - build.size()];
     build.insert(build.begin(), b.str().c_str());
   }
-  for(string s: build)
+  for (string s : build)
   {
-    if (addr.length()) addr += '.';
+    if (addr.length())
+      addr += '.';
     addr += s;
   }
   Console << "connect address: " << addr << endl;
@@ -235,58 +253,60 @@ string getip(string& str, const char* if_empty=nullptr, char sep=' ')
 std::map<string, string> vars;
 
 std::set<string> commands = {
-  "broker", "blink", "client", "connect",
-  "create", "delete", "debug", "help", "interval",
-  "list", "ls", "ip", "off", "on", "set",
-  "publish", "reset", "subscribe", "unsubscribe", "view", "echo", "every"
-};
+    "broker", "blink", "client", "connect",
+    "create", "delete", "debug", "help", "interval",
+    "list", "ls", "ip", "off", "on", "set",
+    "publish", "reset", "subscribe", "unsubscribe", "view", "echo", "every"};
 
-void convertToCommand(string& search)
+void convertToCommand(string &search)
 {
-  while(search[0]==' ') search.erase(0,1);
-  if (search.length()==0) return;
+  while (search[0] == ' ')
+    search.erase(0, 1);
+  if (search.length() == 0)
+    return;
   string matches;
-  int count=0;
-  for(string cmd: commands)
+  int count = 0;
+  for (string cmd : commands)
   {
     if (cmd.substr(0, search.length()) == search)
     {
-      if (count) matches +=", ";
+      if (count)
+        matches += ", ";
       count++;
       matches += cmd;
     }
   }
-  if (count==1)
+  if (count == 1)
     search = matches;
-  else if (count>1)
+  else if (count > 1)
   {
     Console << "Ambiguous command: " << matches << endl;
     search.clear();
   }
 }
 
-void replace(const char* d, string& str, string srch, string to)
+void replace(const char *d, string &str, string srch, string to)
 {
   if (d[0] && d[1])
   {
-    srch=d[0]+srch+d[1];
-    to=d[0]+to+d[1];
+    srch = d[0] + srch + d[1];
+    to = d[0] + to + d[1];
 
     size_t pos = 0;
-    while((pos=str.find(srch, pos)) != string::npos)
+    while ((pos = str.find(srch, pos)) != string::npos)
     {
       str.erase(pos, srch.length());
       str.insert(pos, to);
-      pos += to.length()-1;
+      pos += to.length() - 1;
     }
   }
 }
 
-void replaceVars(string& cmd)
+void replaceVars(string &cmd)
 {
-  cmd = ' '+cmd+' ';
+  cmd = ' ' + cmd + ' ';
 
-  for(auto it: vars)
+  for (auto it : vars)
   {
     replace("..", cmd, it.first, it.second);
     replace(". ", cmd, it.first, it.second);
@@ -294,30 +314,31 @@ void replaceVars(string& cmd)
     replace("  ", cmd, it.first, it.second);
   }
   cmd.erase(0, cmd.find_first_not_of(' '));
-  cmd.erase(cmd.find_last_not_of(' ')+1);
-
+  cmd.erase(cmd.find_last_not_of(' ') + 1);
 }
 
-bool compare(string s, const char* cmd)
+bool compare(string s, const char *cmd)
 {
-  uint8_t p=0;
-  while(s[p++]==*cmd++)
+  uint8_t p = 0;
+  while (s[p++] == *cmd++)
   {
-    if (*cmd==0 or s[p]==0) return true;
-    if (s[p]==' ') return true;
+    if (*cmd == 0 or s[p] == 0)
+      return true;
+    if (s[p] == ' ')
+      return true;
   }
   return false;
 }
 
-using ClientFunction = void(*)(string& cmd, MqttClient* publish);
+using ClientFunction = void (*)(string &cmd, MqttClient *publish);
 
 struct Every
 {
   string cmd;
   uint32_t ms;
   uint32_t next;
-  uint32_t underrun=0;
-  bool active=true;
+  uint32_t underrun = 0;
+  bool active = true;
 
   void dump()
   {
@@ -326,12 +347,12 @@ struct Every
     else
       Console << red << "disabled";
 
-    auto mill=millis();
+    auto mill = millis();
     Console << white << ms << "ms [" << cmd << "] next in ";
     if (mill > next)
       Console << "now";
     else
-      Console << next-mill << "ms";
+      Console << next - mill << "ms";
   }
 };
 
@@ -343,49 +364,49 @@ int16_t blink;
 
 std::vector<Every> everies;
 
-void onCommand(const string& command)
+void onCommand(const string &command)
 {
   Console << endl;
-  string cmd=command;
-  if (cmd.substr(0,3)!="set") replaceVars(cmd);
+  string cmd = command;
+  if (cmd.substr(0, 3) != "set")
+    replaceVars(cmd);
   eval(cmd);
   Console << endl;
-  Console.prompt();
 }
 
-void clientConnect(MqttClient* client, string& cmd)
+void clientConnect(MqttClient *client, string &cmd)
 {
   string remote = getword(cmd);
   uint16_t port;
-  auto it=list.find(remote);
+  auto it = list.find(remote);
   if (it != list.end())
   {
     Console << "Connecting to free broker: " << remote << endl;
     remote = it->second.url;
-    port=it->second.port;
+    port = it->second.port;
   }
   else
-    port=getint(cmd);
+    port = getint(cmd);
 
   client->connect(remote.c_str(), port, getint(cmd, 60));
   Console << (client->connected() ? "connected." : "not connected") << endl;
 }
 
-void eval(string& cmd)
+void eval(string &cmd)
 {
-  while(cmd.length())
+  while (cmd.length())
   {
     MqttError retval = MqttOk;
 
     string s;
-    MqttBroker* broker = nullptr;
-    MqttClient* client = nullptr;
+    MqttBroker *broker = nullptr;
+    MqttClient *client = nullptr;
 
     // client.function notation
     if (cmd.find('.') != string::npos &&
         cmd.find('.') < cmd.find(' '))
     {
-      s=getword(cmd, nullptr, '.');
+      s = getword(cmd, nullptr, '.');
 
       if (s.length())
       {
@@ -406,9 +427,11 @@ void eval(string& cmd)
     }
 
     s = getword(cmd);
-    if (s.length()) convertToCommand(s);
-    if (s.length()==0)
-    {}
+    if (s.length())
+      convertToCommand(s);
+    if (s.length() == 0)
+    {
+    }
     else if (compare(s, "debug"))
     {
 #if TINY_MQTT_DEBUG
@@ -420,14 +443,14 @@ void eval(string& cmd)
     else if (compare(s, "list"))
     {
       Console << "List of free servers" << endl;
-      for(const auto& fb: list)
+      for (const auto &fb : list)
       {
         Console << "  " << fb.first << " : " << fb.second.url << ":" << fb.second.port << endl;
       }
     }
     else if (compare(s, "delete"))
     {
-      if (client==nullptr && broker==nullptr)
+      if (client == nullptr && broker == nullptr)
       {
         s = getword(cmd);
         if (clients.find(s) != clients.end())
@@ -443,9 +466,10 @@ void eval(string& cmd)
       }
       if (client)
       {
-        for (auto it: clients)
+        for (auto it : clients)
         {
-          if (it.second != client) continue;
+          if (it.second != client)
+            continue;
           Console << "deleted" << endl;
           delete (it.second);
           clients.erase(it.first);
@@ -455,9 +479,10 @@ void eval(string& cmd)
       }
       else if (broker)
       {
-        for(auto it: brokers)
+        for (auto it : brokers)
         {
-          if (broker != it.second) continue;
+          if (broker != it.second)
+            continue;
           Console << "deleted" << endl;
           delete (it.second);
           brokers.erase(it.first);
@@ -470,7 +495,7 @@ void eval(string& cmd)
     }
     else if (broker)
     {
-      if (compare(s,"connect"))
+      if (compare(s, "connect"))
       {
         Console << "NYI" << endl;
       }
@@ -486,15 +511,15 @@ void eval(string& cmd)
     }
     else if (client)
     {
-      if (compare(s,"connect"))
+      if (compare(s, "connect"))
       {
         clientConnect(client, cmd);
       }
-      else if (compare(s,"publish"))
+      else if (compare(s, "publish"))
       {
         retval = client->publish(getword(cmd), getword(cmd));
       }
-      else if (compare(s,"subscribe"))
+      else if (compare(s, "subscribe"))
       {
         client->subscribe(getword(cmd));
       }
@@ -514,27 +539,27 @@ void eval(string& cmd)
     }
     else if (compare(s, "on"))
     {
-      uint8_t pin=getint(cmd, 2);
+      uint8_t pin = getint(cmd, 2);
       pinMode(pin, OUTPUT);
       digitalWrite(pin, HIGH);
     }
     else if (compare(s, "off"))
     {
-      uint8_t pin=getint(cmd, 2);
+      uint8_t pin = getint(cmd, 2);
       pinMode(pin, OUTPUT);
       digitalWrite(pin, LOW);
     }
     else if (compare(s, "echo"))
     {
-      s=getword(cmd);
-      if (s=="on")
+      s = getword(cmd);
+      if (s == "on")
         echo_on = true;
-      else if (s=="off")
+      else if (s == "off")
         echo_on = false;
       else
       {
         Console << s << ' ';
-        while(cmd.length())
+        while (cmd.length())
         {
           Console << getword(cmd) << ' ';
         }
@@ -548,9 +573,9 @@ void eval(string& cmd)
         if (cmd.length())
         {
           Every every;
-          every.ms=ms;
-          every.cmd=cmd;
-          every.next=millis()+ms;
+          every.ms = ms;
+          every.cmd = cmd;
+          every.next = millis() + ms;
           everies.push_back(every);
           every.dump();
           Console << endl;
@@ -559,16 +584,16 @@ void eval(string& cmd)
       }
       else if (compare(cmd, "off") or compare(cmd, "on"))
       {
-        bool active=getword(cmd)=="on";
-        uint8_t ever=getint(cmd, 100);
-        uint8_t count=0;
-        for(auto& every: everies)
+        bool active = getword(cmd) == "on";
+        uint8_t ever = getint(cmd, 100);
+        uint8_t count = 0;
+        for (auto &every : everies)
         {
-          if (count==ever or (ever==100))
+          if (count == ever or (ever == 100))
           {
             if (every.active != active)
             {
-              Console << "every #" << count << (active ? " on" :" off") << endl;
+              Console << "every #" << count << (active ? " on" : " off") << endl;
               every.active = active;
               every.underrun = 0;
             }
@@ -576,12 +601,12 @@ void eval(string& cmd)
           count++;
         }
       }
-      else if (compare(cmd, "list") or cmd.length()==0)
+      else if (compare(cmd, "list") or cmd.length() == 0)
       {
         getword(cmd);
         Console << "List of everies (ms=" << millis() << ")" << endl;
-        uint8_t count=0;
-        for(auto& every: everies)
+        uint8_t count = 0;
+        for (auto &every : everies)
         {
           Console << count << ": ";
           every.dump();
@@ -593,20 +618,20 @@ void eval(string& cmd)
       {
         Console << "Removing..." << endl;
         getword(cmd);
-        int8_t every=getint(cmd, -1);
-        if (every==-1 and compare(cmd, "last") and everies.size())
+        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);
+          everies.erase(everies.begin() + everies.size() - 1);
         }
-        else if (every==-1 and compare(cmd, "all"))
+        else if (every == -1 and compare(cmd, "all"))
         {
           getword(cmd);
           everies.clear();
         }
         else if (everies.size() > (uint8_t)every)
         {
-          everies.erase(everies.begin()+every);
+          everies.erase(everies.begin() + every);
         }
         else
           Console << "Bad colmmand" << endl;
@@ -619,33 +644,33 @@ void eval(string& cmd)
       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]);
+        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();
         Console << "Blink " << blink_nr << ' ' << (blink_ms_on[blink_nr] ? "on" : "off") << endl;
         if (blink_ms_on[blink_nr])
-          blink |= 1<< blink_nr;
+          blink |= 1 << blink_nr;
         else
         {
-          blink &= ~(1<< blink_nr);
+          blink &= ~(1 << blink_nr);
         }
       }
     }
     else if (compare(s, "broker"))
     {
-      string id=getword(cmd);
+      string id = getword(cmd);
       if (clients.find(id) != clients.end())
       {
         Console << "A client already have that name" << endl;
         cmd.clear();
       }
-      else if (id.length() or brokers.find(id)!=brokers.end())
+      else if (id.length() or brokers.find(id) != brokers.end())
       {
-        int port=getint(cmd, 0);
+        int port = getint(cmd, 0);
         if (port)
         {
-          MqttBroker* broker = new MqttBroker(port);
+          MqttBroker *broker = new MqttBroker(port);
           broker->begin();
 
           brokers[id] = broker;
@@ -665,25 +690,26 @@ void eval(string& cmd)
     }
     else if (compare(s, "client"))
     {
-      string id=getword(cmd);
+      string id = getword(cmd);
       if (brokers.find(id) != brokers.end())
       {
         Console << "A broker have that name" << endl;
         cmd.clear();
       }
-      else if (id.length() or clients.find(id)!=clients.end())
+      else if (id.length() or clients.find(id) != clients.end())
       {
-        s=getword(cmd);  // broker
-        if (s=="" or brokers.find(s) != brokers.end() or list.find(s) != list.end())
+        s = getword(cmd); // broker
+        if (s == "" or brokers.find(s) != brokers.end() or list.find(s) != list.end())
         {
-          MqttBroker* broker = nullptr;
-          if (s.length()) broker = brokers[s];
-          MqttClient* client = new MqttClient(broker, id);
-          clients[id]=client;
+          MqttBroker *broker = nullptr;
+          if (s.length())
+            broker = brokers[s];
+          MqttClient *client = new MqttClient(broker, id);
+          clients[id] = client;
           client->setCallback(onPublish);
           if (list.find(s) != list.end())
           {
-            cmd=s+' '+cmd;
+            cmd = s + ' ' + cmd;
             clientConnect(client, cmd);
           }
           Console << "new client (" << id.c_str() << ", " << s.c_str() << ')' << endl;
@@ -703,9 +729,9 @@ void eval(string& cmd)
     else if (compare(s, "set"))
     {
       string name(getword(cmd));
-      if (name.length()==0)
+      if (name.length() == 0)
       {
-        for(auto it: vars)
+        for (auto it : vars)
         {
           Console << "  " << it.first << " -> " << it.second << endl;
         }
@@ -732,18 +758,18 @@ void eval(string& cmd)
       if (view)
       {
         Console << save_cursor << magenta;
-        Console.gotoxy(1,1);
+        Console.gotoxy(1, 1);
       }
       Console << "--< " << '/' << clients.size() << " client/s. >--" << erase_to_end << endl;
-      for(auto it: clients)
+      for (auto it : clients)
       {
         it.second->dump("  ");
       }
 
       Console << "--< " << brokers.size() << " brokers/s. >--" << erase_to_end << endl;
-      for(auto it: brokers)
+      for (auto it : brokers)
       {
-        Console << "  +-- '" << it.first.c_str() << "' " << it.second->clientsCount() << " client/s."<< erase_to_end << endl;
+        Console << "  +-- '" << it.first.c_str() << "' " << it.second->clientsCount() << " client/s." << erase_to_end << endl;
         it.second->dump("     ");
       }
       if (view)
@@ -756,7 +782,7 @@ void eval(string& cmd)
       ESP.restart();
     else if (compare(s, "ip"))
       Console << "IP: " << WiFi.localIP() << endl;
-    else if (compare(s,"help"))
+    else if (compare(s, "help"))
     {
       Console << "syntax:" << endl;
       Console << "  MqttBroker:" << endl;
@@ -790,7 +816,8 @@ void eval(string& cmd)
     }
     else
     {
-      while(s[0]==' ') s.erase(0,1);
+      while (s[0] == ' ')
+        s.erase(0, 1);
       if (s.length())
         Console << "Unknown command (" << s.c_str() << ")" << endl;
     }
@@ -804,14 +831,15 @@ void eval(string& cmd)
 
 void loop()
 {
-  auto ms=millis();
-  int8_t out=0;
+  auto ms = millis();
+  int8_t out = 0;
   int16_t blink_bits = blink;
-  uint8_t e=0;
+  uint8_t e = 0;
 
-  for(auto& every: everies)
+  for (auto &every : everies)
   {
-    if (not every.active) continue;
+    if (not every.active)
+      continue;
     if (every.ms && every.cmd.length() && ms > every.next)
     {
       string cmd(every.cmd);
@@ -820,14 +848,15 @@ void loop()
       if (ms > every.next and ms > every.underrun)
       {
         every.next += every.ms;
-        Console << yellow << "Underrun every #" << e << ", " <<  (ms - every.next) << "ms late" << endl;
-        every.underrun = ms+5000;
+        Console << yellow << "Underrun every #" << e << ", " << (ms - every.next) << "ms late" << endl;
+        Console.fg(white);
+        every.underrun = ms + 5000;
       }
     }
     e++;
   }
 
-  while(blink_bits)
+  while (blink_bits)
   {
     if (blink_ms_on[out] and ms > blink_next[out])
     {
@@ -843,7 +872,7 @@ void loop()
       }
       blink_state[out] = not blink_state[out];
     }
-    blink_bits >>=1;
+    blink_bits >>= 1;
     out++;
   }
 
@@ -851,10 +880,10 @@ void loop()
   MDNS.update();
 #endif
 
-  for(auto it: brokers)
+  for (auto it : brokers)
     it.second->loop();
 
-  for(auto it: clients)
+  for (auto it : clients)
     it.second->loop();
 
   Console.loop();

library.json

@@ -6,7 +6,9 @@
         "type": "git",
         "url": "https://github.com/hsaturn/TinyMqtt.git"
     },
-    "version": "0.9.15",
+    "dependencies":
+    { "hsaturn/TinyConsole" : "*" },
+    "version": "0.9.19",
     "exclude": "",
     "examples": "examples/*/*.ino",
     "frameworks": "arduino",

library.json.skeleton

@@ -6,6 +6,8 @@
         "type": "git",
         "url": "https://github.com/hsaturn/TinyMqtt.git"
     },
+    "dependencies":
+    { #dependencies },
     "version": "#version",
     "exclude": "",
     "examples": "examples/*/*.ino",

library.properties

@@ -1,5 +1,5 @@
 name=TinyMqtt
-version=0.9.16
+version=0.9.19
 author=Francois BIOT, HSaturn, <hsaturn@gmail.com>
 maintainer=Francois BIOT <hsaturn@gmail.com>
 sentence=A tiny broker and client library for MQTT messaging.
@@ -7,5 +7,5 @@ paragraph=MQTT is a lightweight messaging protocol. This library allows to host
 category=Communication
 url=https://github.com/hsaturn/TinyMqtt
 architectures=*
-depends=TinyConsole
+depends=hsaturn/TinyConsole
 includes=TinyMqtt.h

src/StringIndexer.h

@@ -107,6 +107,8 @@ class IndexedString
       index = source.index;
     }
 
+    IndexedString(IndexedString&& i) : index(i.index) {}
+
     IndexedString(const char* str, uint8_t len)
     {
       index=StringIndexer::strToIndex(str, len);

src/TinyMqtt.cpp

@@ -15,8 +15,10 @@ int TinyMqtt::debug=2;
   std::map<MqttMessage::Type, int> MqttClient::counters;
 #endif
 
-MqttBroker::MqttBroker(uint16_t port)
+MqttBroker::MqttBroker(uint16_t port, uint8_t max_retain_size)
 {
+  debug("New broker" << port);
+  retain_size = max_retain_size;
   server = new TcpServer(port);
 #ifdef TINY_MQTT_ASYNC
   server->onClient(onClient, this);
@@ -116,7 +118,7 @@ void MqttClient::connect(string broker, uint16_t port, uint16_t ka)
 #ifdef TINY_MQTT_ASYNC
   tcp_client->onData(onData, this);
   tcp_client->onConnect(onConnect, this);
-  tcp_client->connect(broker.c_str(), port, ka);
+  tcp_client->connect(broker.c_str(), port);
 #else
   if (tcp_client->connect(broker.c_str(), port))
   {
@@ -142,6 +144,7 @@ void MqttBroker::connect(const string& host, uint16_t port)
   if (remote_broker == nullptr) remote_broker = new MqttClient;
   remote_broker->connect(host, port);
   remote_broker->local_broker = this;  // Because connect removed the link
+  // TODO shouldn't we resubscribe to all client subscriptions ?
 }
 
 void MqttBroker::removeClient(MqttClient* remove)
@@ -211,9 +214,19 @@ void MqttBroker::loop()
   }
 }
 
-MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
+// Obvioulsy called when the broker is connected to another broker.
+MqttError MqttBroker::subscribe(MqttClient* client, const Topic& topic, uint8_t qos)
 {
-  debug("MqttBroker::subscribe");
+  debug("MqttBroker::subscribe to " << topic.str() << ", retained=" << retained.size() );
+  for(auto& [retained_topic, retain]: retained)
+  {
+    debug("  retained: " << retained_topic.str());
+    if (topic.matches(retained_topic))
+    {
+      debug("  -> sending");
+      client->publishIfSubscribed(retained_topic, retain.msg);
+    }
+  }
   if (remote_broker && remote_broker->connected())
   {
     return remote_broker->subscribe(topic, qos);
@@ -221,10 +234,12 @@ MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
   return MqttNowhereToSend;
 }
 
-MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, MqttMessage& msg) const
+MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, MqttMessage& msg)
 {
   MqttError retval = MqttOk;
 
+  retain(topic, msg);
+
   debug("MqttBroker::publish");
   int i=0;
   for(auto client: clients)
@@ -294,25 +309,26 @@ void MqttClient::clientAlive(uint32_t more_seconds)
 
 void MqttClient::loop()
 {
-  if (keep_alive && (millis() >= alive))
+  if (keep_alive && (millis() >= alive - 5000))
   {
-    if (local_broker)
-    {
-      debug(red << "timeout client");
-      close();
-      debug(red << "closed");
-    }
-    else if (tcp_client && tcp_client->connected())
+    if (tcp_client && tcp_client->connected())
     {
       debug("pingreq");
-      uint16_t pingreq = MqttMessage::Type::PingReq;
-      tcp_client->write((const char*)(&pingreq), 2);
+      static MqttMessage pingreq(MqttMessage::Type::PingReq);
+      pingreq.sendTo(this);
       clientAlive(0);
 
       // TODO when many MqttClient passes through a local broker
       // there is no need to send one PingReq per instance.
     }
+    else if (local_broker)
+    {
+      debug(red << "timeout client");
+      close();
+      debug(red << "closed");
+    }
   }
+
 #ifndef TINY_MQTT_ASYNC
   while(tcp_client && tcp_client->available()>0)
   {
@@ -389,15 +405,15 @@ MqttError MqttClient::subscribe(Topic topic, uint8_t qos)
   debug("MqttClient::subsribe(" << topic.c_str() << ")");
   MqttError ret = MqttOk;
 
-   subscriptions.insert(topic);
+  subscriptions.insert(topic);
 
-  if (local_broker==nullptr) // remote broker
+  if (local_broker==nullptr) // connected to a remote broker
   {
     return sendTopic(topic, MqttMessage::Type::Subscribe, qos);
   }
   else
   {
-    return local_broker->subscribe(topic, qos);
+    return local_broker->subscribe(this, topic, qos);
   }
   return ret;
 }
@@ -568,7 +584,7 @@ void MqttClient::processMessage(MqttMessage* mesg)
             }
             else
               qoss.push_back(qos);
-            subscriptions.insert(topic);
+            subscribe(topic);
           }
           else
           {
@@ -618,8 +634,8 @@ void MqttClient::processMessage(MqttMessage* mesg)
           #if TINY_MQTT_DEBUG
             if (TinyMqtt::debug >= 2)
             {
-              Console << (isSubscribedTo(published) ? "not" : "") << " subscribed.\n";
-              Console << "has " << (callback ? "" : "no ") << " callback.\n";
+              Console << (isSubscribedTo(published) ? "not" : "") << " subscribed.\r\n";
+              Console << "has " << (callback ? "" : "no ") << " callback.\r\n";
             }
           #endif
           if (callback and isSubscribedTo(published))
@@ -728,9 +744,9 @@ bool Topic::matches(const Topic& topic) const
 
 
 // publish from local client
-MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pay_length)
+MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pay_length, bool retain)
 {
-  MqttMessage msg(MqttMessage::Publish);
+  MqttMessage msg(MqttMessage::Publish, retain ? 1 : 0);
   msg.add(topic);
   msg.add(payload, pay_length, false);
   msg.complete();
@@ -894,6 +910,35 @@ MqttError MqttMessage::sendTo(MqttClient* client)
   return MqttOk;
 }
 
+void MqttBroker::retainDrop()
+{
+  if (retained.size() >= retain_size)
+  {
+    std::map<Topic, Retain>::iterator oldest = retained.begin();
+    auto it = oldest;
+    while(++it != retained.end())
+    {
+      if (oldest->second.timestamp > it->second.timestamp)
+        oldest = it;
+    }
+    retained.erase(oldest);
+  }
+}
+
+void MqttBroker::retain(const Topic& topic, const MqttMessage& msg)
+{
+  debug("MqttBroker::retain msg_type=" << _HEX(msg.type()));
+  if (retain_size==0 or msg.type() != MqttMessage::Publish) return;
+  if (msg.flags() & 1)  // flag RETAIN
+  {
+    debug("  retaining " << topic.str());
+    if (retained.find(topic) == retained.end()) retainDrop();
+    // FIXME if payload size == 0 remove message from retained
+    Retain r(micros(), msg);
+    retained.insert({ topic, std::move(r)});
+  }
+}
+
 void MqttMessage::hexdump(const char* prefix) const
 {
   (void)prefix;

src/TinyMqtt.h

@@ -121,7 +121,10 @@ class MqttMessage
       return (*bun << 8) | bun[1]; }
 
     MqttMessage() { reset(); }
-    MqttMessage(Type t, uint8_t bits_d3_d0=0) { create(t); buffer[0] |= bits_d3_d0; }
+    MqttMessage(Type t, uint8_t bits_d3_d0=0) { create(t); buffer[0] |= (bits_d3_d0 & 0xF); }
+    MqttMessage(const MqttMessage& m)
+      : buffer(m.buffer), vheader(m.vheader), size(m.size), state(m.state) {}
+
     void incoming(char byte);
     void add(char byte) { incoming(byte); }
     void add(const char* p, size_t len, bool addLength=true );
@@ -156,6 +159,15 @@ class MqttMessage
     MqttError sendTo(MqttClient*);
     void hexdump(const char* prefix=nullptr) const;
 
+    MqttMessage& operator = (MqttMessage&& m)
+    {
+      buffer = std::move(m.buffer);
+      vheader = m.vheader;
+      size = m.size;
+      state = m.state;
+      return *this;
+    }
+
   private:
     void encodeLength();
 
@@ -228,11 +240,11 @@ class MqttClient
     };
 
     // Publish from client to the world
-    MqttError publish(const Topic&, const char* payload, size_t pay_length);
-    MqttError publish(const Topic& t, const char* payload) { return publish(t, payload, strlen(payload)); }
-    MqttError publish(const Topic& t, const String& s) { return publish(t, s.c_str(), s.length()); }
-    MqttError publish(const Topic& t, const string& s) { return publish(t,s.c_str(),s.length());}
-    MqttError publish(const Topic& t) { return publish(t, nullptr, 0);};
+    MqttError publish(const Topic&, const char* payload, size_t pay_length, bool retain=false);
+    MqttError publish(const Topic& t, const char* payload, bool retain=false) { return publish(t, payload, strlen(payload), retain); }
+    MqttError publish(const Topic& t, const String& s, bool retain=false) { return publish(t, s.c_str(), s.length(), retain); }
+    MqttError publish(const Topic& t, const string& s, bool retain=false) { return publish(t,s.c_str(),s.length(), retain);}
+    MqttError publish(const Topic& t, bool retain=false) { return publish(t, nullptr, 0, retain);};
 
     MqttError subscribe(Topic topic, uint8_t qos=0);
     MqttError unsubscribe(Topic topic);
@@ -317,15 +329,9 @@ class MqttClient
 
 class MqttBroker
 {
-  enum __attribute__((packed)) 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(uint16_t port, uint8_t retain_size=0);
     ~MqttBroker();
 
     void begin() { server->begin(); }
@@ -334,9 +340,10 @@ class MqttBroker
     /** Connect the broker to a parent broker */
     void connect(const string& host, uint16_t port=1883);
     /** returns true if connected to another broker */
-    bool connected() const { return state == Connected; }
+    bool connected() const { return remote_broker ? remote_broker->connected() : false; }
 
     size_t clientsCount() const { return clients.size(); }
+    void retain(uint8_t size) { retain_size = size; }
 
     void dump(string indent="")
     {
@@ -356,10 +363,9 @@ class MqttBroker
     bool checkPassword(const char* password, uint8_t len) const
     { return compareString(auth_password, password, len); }
 
+    MqttError publish(const MqttClient* source, const Topic& topic, MqttMessage& msg);
 
-    MqttError publish(const MqttClient* source, const Topic& topic, MqttMessage& msg) const;
-
-    MqttError subscribe(const Topic& topic, uint8_t qos);
+    MqttError subscribe(MqttClient*, const Topic& topic, uint8_t qos);
 
     // For clients that are added not by the broker itself (local clients)
     void addClient(MqttClient* client);
@@ -375,5 +381,25 @@ class MqttBroker
     const char* auth_password = "guest";
     MqttClient* remote_broker = nullptr;
 
-    State state = Disconnected;
+    void retain(const Topic& topic, const MqttMessage& msg);
+    void retainDrop();
+
+    struct Retain
+    {
+      Retain(unsigned long ts, const MqttMessage& m) : timestamp(ts), msg(m) {}
+      Retain(Retain&& r) : timestamp(r.timestamp), msg(std::move(r.msg)) {}
+
+      Retain& operator=(Retain&& r)
+      {
+        timestamp = r.timestamp;
+        msg = std::move(r.msg);
+        return *this;
+      }
+
+      unsigned long timestamp;
+      MqttMessage msg;
+    };
+
+    std::map<Topic, Retain> retained;
+    uint8_t retain_size;
 };