Compare commits
17 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
82c5b971e9 | ||
|
|
01998e74ec | ||
|
|
5f46fd304c | ||
|
|
213d637eaf | ||
|
|
3bb2dd5a81 | ||
|
|
7d9ab6381d | ||
|
|
470cde62da | ||
|
|
3fb9b6317d | ||
|
|
ee9ad93bfd | ||
|
|
0b1a932244 | ||
|
|
972759237c | ||
|
|
cb00d7f82a | ||
|
|
0b735d22a5 | ||
|
|
9178aac02c | ||
|
|
c706fbcff2 | ||
|
|
a0c41a0ccb | ||
|
|
b780dcf99c |
25
.github/workflows/superlinter.yml
vendored
Normal file
25
.github/workflows/superlinter.yml
vendored
Normal 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 }}
|
||||
13
README.md
13
README.md
@@ -2,14 +2,17 @@
|
||||
|
||||

|
||||

|
||||

|
||||

|
||||

|
||||

|
||||
|
||||
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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"))
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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;
|
||||
|
||||
102
src/TinyMqtt.cpp
102
src/TinyMqtt.cpp
@@ -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,9 +558,16 @@ 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:
|
||||
case PayLoad:
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -46,8 +46,10 @@ class MqttMessage
|
||||
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;
|
||||
|
||||
138
tests/tinymqtt-tests/tinymqtt-tests.ino
Normal file
138
tests/tinymqtt-tests/tinymqtt-tests.ino
Normal 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();
|
||||
}
|
||||
Reference in New Issue
Block a user