Compare commits

...

17 Commits
0.5.1 ... 0.7.0

Author SHA1 Message Date
hsaturn
82c5b971e9 Release 0.7.0 2021-03-28 23:34:54 +02:00
hsaturn
01998e74ec Test linter 2021-03-28 22:59:42 +02:00
hsaturn
5f46fd304c Release 0.7.0 2021-03-28 22:47:13 +02:00
hsaturn
213d637eaf Code cleaning 2021-03-28 21:31:10 +02:00
hsaturn
3bb2dd5a81 Code cleaning 2021-03-28 21:29:02 +02:00
hsaturn
7d9ab6381d Disconnect added (finally) 2021-03-28 21:28:06 +02:00
hsaturn
470cde62da Version 0.6.0 2021-03-27 10:26:41 +01:00
hsaturn
3fb9b6317d README.md modified (typo) 2021-03-27 02:00:49 +01:00
hsaturn
ee9ad93bfd README.md modified 2021-03-27 02:00:00 +01:00
hsaturn
0b1a932244 MqttClient resubscribe on reconnect 2021-03-27 01:55:03 +01:00
hsaturn
972759237c Speed and stability improved 2021-03-27 01:40:49 +01:00
hsaturn
cb00d7f82a MqttClient / UnSubscribe message implemented 2021-03-27 01:39:37 +01:00
hsaturn
0b735d22a5 Less debug as the code is more stable now 2021-03-27 01:38:32 +01:00
hsaturn
9178aac02c MqttClient client length augmented to 60 (was not passing MqttBox tests 2021-03-26 01:59:08 +01:00
hsaturn
c706fbcff2 Update readme 2021-03-26 01:52:31 +01:00
hsaturn
a0c41a0ccb Smart ip feature for connect 2021-03-26 01:51:30 +01:00
hsaturn
b780dcf99c test fix, was unable to use set when replacements occurs 2021-03-26 01:10:33 +01:00
10 changed files with 340 additions and 69 deletions

25
.github/workflows/superlinter.yml vendored Normal file
View File

@@ -0,0 +1,25 @@
name: Super-Linter
# Run this workflow every time a new commit pushed to your repository
on: push
jobs:
# Set the job key. The key is displayed as the job name
# when a job name is not provided
super-lint:
# Name the Job
name: Lint code base
# Set the type of machine to run on
runs-on: ubuntu-latest
steps:
# Checks out a copy of your repository on the ubuntu-latest machine
- name: Checkout code
uses: actions/checkout@v2
# Runs the Super-Linter action
- name: Run Super-Linter
uses: github/super-linter@v3
env:
DEFAULT_BRANCH: main
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

View File

@@ -2,14 +2,17 @@
![](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/badge/platform-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
ESP 8266 is a small, fast and capable Mqtt Broker and Client
## Features
- Very (very !!) fast broker I saw it re-sent 1000 topics per second for two
clients that had subscribed (payload ~15 bytes). No topic lost.
The max I've seen was 2k msg/s (1 client 1 subscription)
- Act as as a mqtt broker and/or a mqtt client
- Mqtt 3.1.1 / Qos 0 supported
- Standalone (can work without WiFi) (degraded/local mode)
@@ -26,10 +29,12 @@ no need for having tons of clients (also RAM is the problem with many clients)
* Test what is the real max number of clients for broker. As far as I saw, 1k is needed per client which would make more than 30 clients critical.
* ~~MqttMessage uses a buffer 256 bytes which is usually far than needed.~~
* ~~MqttClient does not support more than one subscription at time~~
* MqttClient auto re-subscribe
* ~~MqttClient auto re-subscribe (::resubscribe works bad on broker.emqx.io)~~
* MqttClient auto reconnection
* MqttClient does not callback payload...
* ~~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

View File

@@ -1,9 +1,9 @@
// vim: ts=30
// vim: ts=40
Exemple of commands that can be sent via the serial monitor to tinymqtt-test
----------------------------------------------------------------------------
Commands can usually be abbreviated to their first letters.
ex: cl for client, a / a.con / a.sub / a.p for publish.
--------
set name value set variable name to value (later replaced)
set name if no value, then var is erased
@@ -17,9 +17,12 @@ a.publish topic [payload] send a topic with a payload
a.subscribe topic subscribes to a topic
delete a destroy the client
----------------------------------------------------
* note, if 'server' is a number, then it replaces the end of the local ip.
i.e. if local ip is 192.168.1.10, connect 2.35 becomes 192.168.2.35
--------
example:
--------
client c
c.connect broker.emqx.io
@@ -30,5 +33,7 @@ c.publish topic 15
c.publish topic 20
macro exansion example
----------------------
set temp publish sensor/temperature
c.temp 20 -> c.publish sensor/temperature 20

View File

@@ -1,6 +1,7 @@
#define TINY_MQTT_DEBUG
#include <TinyMqtt.h> // https://github.com/hsaturn/TinyMqtt
#include <MqttStreaming.h>
#include <sstream>
#include <map>
/**
@@ -15,11 +16,6 @@
* cons - Takes more memory
* - a bit hard to understand
*
* This sounds crazy: a mqtt mqtt that do not need a broker !
* The use case arise when one ESP wants to publish topics and subscribe to them at the same time.
* Without broker, the ESP won't react to its own topics.
*
* TinyMqtt mqtt allows this use case to work.
*/
#include <my_credentials.h>
@@ -27,18 +23,26 @@
std::string topic="sensor/temperature";
void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
{ Serial << "--> " << srce->id().c_str() << ": ======> received " << topic.c_str() << endl; }
{
Serial << "--> " << srce->id().c_str() << ": ======> received " << topic.c_str();
if (payload) Serial << ", payload[" << length << "]=[";
while(length--)
{
const char c=*payload++;
if (c!=10 and c!=13 and c <32) Serial << '?';
Serial << *payload++;
}
Serial<< endl;
}
std::map<std::string, MqttClient*> clients;
std::map<std::string, MqttBroker*> brokers;
void setup()
{
Serial.begin(115200);
delay(500);
Serial << endl << endl << endl
<< "Demo started. Type help for more..." << endl
<< "Connecting to '" << ssid << "' ";
WiFi.mode(WIFI_STA);
@@ -48,6 +52,7 @@ void setup()
{ Serial << '-'; delay(500); }
Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
Serial << "Type help for more..." << endl;
MqttBroker* broker = new MqttBroker(1883);
broker->begin();
@@ -78,6 +83,45 @@ std::string getword(std::string& str, const char* if_empty=nullptr, char sep=' '
return sword;
}
bool isaddr(std::string s)
{
if (s.length()==0 or s.length()>3) return false;
for(char c: s)
if (c<'0' or c>'9') return false;
return true;
}
std::string getip(std::string& str, const char* if_empty=nullptr, char sep=' ')
{
std::string addr=getword(str, if_empty, sep);
std::string ip=addr;
std::vector<std::string> build;
while(ip.length())
{
std::string b=getword(ip,nullptr,'.');
if (isaddr(b) && build.size()<4)
{
build.push_back(b);
}
else
return addr;
}
IPAddress local=WiFi.localIP();
addr="";
while(build.size()!=4)
{
std::stringstream b;
b << (int)local[3-build.size()];
build.insert(build.begin(), b.str());
}
for(std::string s: build)
{
if (addr.length()) addr += '.';
addr += s;
}
return addr;
}
std::map<std::string, std::string> vars;
std::set<std::string> commands = {
@@ -270,6 +314,12 @@ using ClientFunction = void(*)(std::string& cmd, MqttClient* publish);
void loop()
{
static long count;
if (MqttClient::counter != count)
{
Serial << "# " << MqttClient::counter << endl;
count = MqttClient::counter;
}
for(auto it: brokers)
it.second->loop();
@@ -293,8 +343,7 @@ void loop()
else
last_cmd=cmd;
replaceVars(cmd);
Serial << "---------------@[ " << cmd.c_str() << " ]--------------" << endl;
if (cmd.substr(0,3)!="set") replaceVars(cmd);
while(cmd.length())
{
MqttError retval = MqttOk;
@@ -390,7 +439,7 @@ void loop()
{
if (compare(s,"connect"))
{
client->connect(getword(cmd,"192.168.1.40").c_str(), getint(cmd, 1883), getint(cmd, 60));
client->connect(getip(cmd,"192.168.1.40").c_str(), getint(cmd, 1883), getint(cmd, 60));
Serial << (client->connected() ? "connected." : "not connected") << endl;
}
else if (compare(s,"publish"))

View File

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

View File

@@ -1,5 +1,5 @@
name=TinyMqtt
version=0.5.1
version=0.7.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.

View File

@@ -104,7 +104,7 @@ class IndexedString
const std::string& str() const { return StringIndexer::str(index); }
const StringIndexer::index_t getIndex() const { return index; }
const StringIndexer::index_t& getIndex() const { return index; }
private:
StringIndexer::index_t index;

View File

@@ -41,14 +41,20 @@ MqttClient::~MqttClient()
{
close();
delete client;
Serial << "Client deleted" << endl;
}
void MqttClient::close()
void MqttClient::close(bool bSendDisconnect)
{
debug("close " << id().c_str());
mqtt_connected = false;
if (client)
{
if (bSendDisconnect and client->connected())
{
message.create(MqttMessage::Type::Disconnect);
message.sendTo(this);
}
client->stop();
}
@@ -120,7 +126,7 @@ void MqttBroker::loop()
// for(auto it=clients.begin(); it!=clients.end(); it++)
// use index because size can change during the loop
for(int i=0; i<clients.size(); i++)
for(size_t i=0; i<clients.size(); i++)
{
auto client = clients[i];
if (client->connected())
@@ -146,8 +152,10 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, Mqtt
for(auto client: clients)
{
i++;
#if TINY_MQTT_DEBUG
Serial << "brk_" << (broker && broker->connected() ? "con" : "dis") <<
" srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << i << ", local=" << client->isLocal() << ", con=" << client->connected() << endl;
#endif
bool doit = false;
if (broker && broker->connected()) // Broker is connected
{
@@ -166,7 +174,9 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, Mqtt
// All is allowed
doit = true;
}
#if TINY_MQTT_DEBUG
Serial << ", doit=" << doit << ' ';
#endif
if (doit) retval = client->publish(topic, msg);
debug("");
@@ -232,6 +242,26 @@ void MqttClient::loop()
}
}
void MqttClient::resubscribe()
{
// TODO resubscription limited to 256 bytes
if (subscriptions.size())
{
MqttMessage msg(MqttMessage::Type::Subscribe, 2);
// TODO manage packet identifier
msg.add(0);
msg.add(0);
for(auto topic: subscriptions)
{
msg.add(topic);
msg.add(0); // TODO qos
}
msg.sendTo(this); // TODO return value
}
}
MqttError MqttClient::subscribe(Topic topic, uint8_t qos)
{
debug("subsribe(" << topic.c_str() << ")");
@@ -248,7 +278,7 @@ MqttError MqttClient::subscribe(Topic topic, uint8_t qos)
msg.add(0);
msg.add(0);
msg.add(topic.str());
msg.add(topic);
msg.add(qos);
ret = msg.sendTo(this);
@@ -257,15 +287,18 @@ MqttError MqttClient::subscribe(Topic topic, uint8_t qos)
return ret;
}
long MqttClient::counter=0;
void MqttClient::processMessage()
{
std::string error;
std::string s;
counter++;
#if TINY_MQTT_DEBUG
if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessage::Type::PingResp)
{
Serial << "---> INCOMING " << _HEX(message.type()) << " client(" << (int)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
message.hexdump("Incoming");
// message.hexdump("Incoming");
}
#endif
auto header = message.getVHeader();
const char* payload;
uint16_t len;
@@ -295,20 +328,6 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
// 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;
@@ -348,9 +367,9 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
break;
case MqttMessage::Type::ConnAck:
// TODO what more on connack ?
mqtt_connected = true;
bclose = false;
resubscribe();
break;
case MqttMessage::Type::SubAck:
@@ -380,6 +399,7 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
break;
case MqttMessage::Type::Subscribe:
case MqttMessage::Type::UnSubscribe:
{
if (!mqtt_connected) break;
payload = header+2;
@@ -391,7 +411,15 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
debug( " topic (" << std::string(payload, len) << ')');
outstring("Subscribes", payload, len);
// subscribe(Topic(payload, len));
subscriptions.insert(Topic(payload, len));
Topic topic(payload, len);
if ((message.type() & 0XF0) == MqttMessage::Type::Subscribe)
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);
@@ -425,11 +453,21 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
{
callback(this, published, nullptr, 0); // TODO send the real payload
}
// TODO should send PUBACK
message.create(MqttMessage::Type::PubAck);
// TODO re-add packet identifier if any
message.sendTo(this);
bclose = false;
}
break;
case MqttMessage::Type::Disconnect:
// TODO should discard any will message
if (!mqtt_connected) break;
mqtt_connected = false;
close(false);
bclose=false;
break;
default:
bclose=true;
break;
@@ -437,7 +475,7 @@ if (message.type() != MqttMessage::Type::PingReq && message.type() != MqttMessag
if (bclose)
{
Serial << "*************** Error msg 0x" << _HEX(message.type());
if (error.length()) Serial << ':' << error.c_str();
message.hexdump("-------ERROR ------");
Serial << endl;
close();
}
@@ -477,10 +515,9 @@ MqttError MqttClient::publish(const Topic& topic, MqttMessage& msg)
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 << " match/send";
debug(" match client=" << (int32_t)client << ", topic " << topic.str().c_str() << ' ');
if (client)
{
retval = msg.sendTo(this);
@@ -490,7 +527,6 @@ MqttError MqttClient::publish(const Topic& topic, MqttMessage& msg)
callback(this, topic, nullptr, 0); // TODO Payload
}
}
Serial << endl;
}
return retval;
}
@@ -522,8 +558,15 @@ void MqttMessage::incoming(char in_byte)
vheader = buffer.length();
if (size==0)
state = Complete;
else if (size > 500) // TODO magic
{
state = Error;
}
else
{
buffer.reserve(size);
state = VariableHeader;
}
}
break;
case VariableHeader:
@@ -541,6 +584,7 @@ void MqttMessage::incoming(char in_byte)
case Complete:
default:
Serial << "Spurious " << _HEX(in_byte) << endl;
hexdump("spurious");
reset();
break;
}
@@ -574,11 +618,11 @@ void MqttMessage::encodeLength(char* msb, int length)
MqttError MqttMessage::sendTo(MqttClient* client)
{
if (buffer.size()>2)
if (buffer.size())
{
debug("sending " << buffer.size() << " bytes");
encodeLength(&buffer[1], buffer.size()-2);
hexdump("snd");
// hexdump("snd");
client->write(&buffer[0], buffer.size());
}
else

View File

@@ -5,11 +5,11 @@
#include "StringIndexer.h"
#include <MqttStreaming.h>
#define TINY_MQTT_DEBUG
#ifdef TINY_MQTT_DEBUG
#if 0
#define debug(what) { Serial << __LINE__ << ' ' << what << endl; delay(100); }
#define TINY_MQTT_DEBUG 1
#else
#define TINY_MQTT_DEBUG 0
#define debug(what) {}
#endif
@@ -39,15 +39,17 @@ class MqttMessage
public:
enum Type
{
Unknown = 0,
Connect = 0x10,
ConnAck = 0x20,
Publish = 0x30,
PubAck = 0x40,
Subscribe = 0x80,
SubAck = 0x90,
PingReq = 0xC0,
PingResp = 0xD0,
Unknown = 0,
Connect = 0x10,
ConnAck = 0x20,
Publish = 0x30,
PubAck = 0x40,
Subscribe = 0x80,
SubAck = 0x90,
UnSubscribe = 0xA0,
PingReq = 0xC0,
PingResp = 0xD0,
Disconnect = 0xE0
};
enum State
{
@@ -119,6 +121,7 @@ class MqttClient
};
public:
MqttClient(MqttBroker*);
MqttClient(MqttBroker* brk, const std::string& id) : MqttClient(brk) { clientId=id; }
MqttClient() : MqttClient(nullptr) {};
~MqttClient();
@@ -136,7 +139,7 @@ class MqttClient
void id(std::string& new_id) { clientId = new_id; }
void loop();
void close();
void close(bool bSendDisconnect=true);
void setCallback(CallBack fun) {callback=fun; };
// Publish from client to the world
@@ -152,17 +155,16 @@ class MqttClient
// TODO seems to be useless
bool isLocal() const { return client == nullptr; }
#ifdef TINY_MQTT_DEBUG
void dump()
{
uint32_t ms=millis();
Serial << "MqttClient (" << clientId.c_str() << ") p=" << (int32_t) parent
<< " c=" << (int32_t)client << (connected() ? " ON " : " OFF");
Serial << ", alive=" << (uint32_t)alive << '/' << ms << ", ka=" << keep_alive;
Serial << " cnx " << (client && client->connected());
Serial << " [";
Serial << (client && client->connected() ? "" : "dis") << "connected";
message.hexdump("entrant msg");
bool c=false;
Serial << " [";
for(auto s: subscriptions)
{
Serial << (c?", ": "")<< s.str().c_str();
@@ -172,9 +174,12 @@ class MqttClient
Serial << "]" << endl;
}
#endif
static long counter; // Number of messages sent
private:
void resubscribe();
friend class MqttBroker;
MqttClient(MqttBroker* parent, WiFiClient& client);
// republish a received publish if topic matches any in subscriptions
@@ -221,7 +226,8 @@ class MqttBroker
void connect(std::string host, uint32_t port=1883);
bool connected() const { return state == Connected; }
#ifdef TINY_MQTT_DEBUG
size_t clientsCount() const { return clients.size(); }
void dump()
{
Serial << clients.size() << " client/s" << endl;
@@ -231,7 +237,6 @@ class MqttBroker
client->dump();
}
}
#endif
private:
friend class MqttClient;

View File

@@ -0,0 +1,138 @@
#include <AUnit.h>
#include <TinyMqtt.h>
#include <map>
/**
* TinyMqtt local unit tests.
*
* No wifi connection unit tests.
**/
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(local_client_should_unregister_when_destroyed)
{
assertEqual(broker.clientsCount(), (size_t)0);
{
MqttClient client(&broker);
assertEqual(broker.clientsCount(), (size_t)1);
}
assertEqual(broker.clientsCount(), (size_t)0);
}
test(local_connect)
{
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient client(&broker);
assertTrue(client.connected());
assertEqual(broker.clientsCount(), (size_t)1);
}
test(local_publish_should_be_dispatched)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient subscriber(&broker);
subscriber.subscribe("a/b");
subscriber.subscribe("a/c");
subscriber.setCallback(onPublish);
MqttClient publisher(&broker);
publisher.publish("a/b");
publisher.publish("a/c");
publisher.publish("a/c");
assertEqual(published.size(), (size_t)1); // 1 client has received something
assertTrue(published[""]["a/b"] == 1);
assertTrue(published[""]["a/c"] == 2);
}
test(local_publish_should_be_dispatched_to_local_clients)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient subscriber_a(&broker, "A");
subscriber_a.setCallback(onPublish);
subscriber_a.subscribe("a/b");
subscriber_a.subscribe("a/c");
MqttClient subscriber_b(&broker, "B");
subscriber_b.setCallback(onPublish);
subscriber_b.subscribe("a/b");
MqttClient publisher(&broker);
publisher.publish("a/b");
publisher.publish("a/c");
assertEqual(published.size(), (size_t)2); // 2 clients have received something
assertTrue(published["A"]["a/b"] == 1);
assertTrue(published["A"]["a/c"] == 1);
assertTrue(published["B"]["a/b"] == 1);
assertTrue(published["B"]["a/c"] == 0);
}
test(local_unsubscribe)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient subscriber(&broker);
subscriber.setCallback(onPublish);
subscriber.subscribe("a/b");
MqttClient publisher(&broker);
publisher.publish("a/b");
// subscriber.unsubscribe("a/b"); TODO not yet implemented
publisher.publish("a/b");
assertTrue(published[""]["a/b"] == 1); // Only one publish has been received
}
test(local_nocallback_when_destroyed)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
{
MqttClient subscriber(&broker);
subscriber.setCallback(onPublish);
subscriber.subscribe("a/b");
}
MqttClient publisher(&broker);
publisher.publish("a/b");
assertEqual(published.size(), (size_t)0); // Only one publish has been received
}
//----------------------------------------------------------------------------
// setup() and loop()
void setup() {
delay(1000);
Serial.begin(115200);
while(!Serial);
Serial.println("=============[ NO WIFI CONNECTION TinyMqtt TESTS ]========================");
}
void loop() {
aunit::TestRunner::run();
if (Serial.available()) ESP.reset();
}