Compare commits

..

52 Commits
pr22 ... 0.9.11

Author SHA1 Message Date
hsaturn
7f12ecfd6d Update mqtt_class_binder.ino 2022-12-28 20:29:33 +01:00
Francois BIOT
3ae1afec27 Release 0.9.10 2022-12-28 20:23:58 +01:00
Francois BIOT
9608ed9fdf Added example for MqttClassBinder 2022-12-28 20:22:31 +01:00
hsaturn
220e904ae9 Add MqttClassBinder 2022-12-28 19:34:29 +01:00
hsaturn
49b696315c Fix TINY_MQTT_DEBUG compilation 2022-12-28 19:30:16 +01:00
hsaturn
a9ebf31e6f Version 0.9.9 2022-12-24 02:01:27 +01:00
hsaturn
4b4eb0b684 Fix compilation error when not in debug 2022-12-23 18:06:35 +01:00
HSaturn
18ce34c458 :xDefault client is no more empty 2022-12-17 18:36:32 +01:00
Francois BIOT
70ca3787bb Better debugging code 2022-12-04 02:41:11 +01:00
Francois BIOT
396e3fde95 Moved MqttStreaming to TinyConsole 2022-12-04 02:30:37 +01:00
Francois BIOT
c913bc61bb tinymqtt-test : new commands 2022-12-04 02:03:28 +01:00
Francois BIOT
7af8e46b59 simple-client typo 2022-12-04 02:00:50 +01:00
Francois BIOT
0569bc6000 [tinytest] Removed auto commands and default topic 2022-12-04 00:25:43 +01:00
Francois BIOT
708a2b41dc tinymqtt-test enhancements 2022-12-03 23:09:20 +01:00
Francois BIOT
c4edfb6e40 fix bug in simple-broker example 2022-12-03 21:27:03 +01:00
Francois BIOT
cf724507e9 fix typo 2022-12-03 21:03:05 +01:00
Francois BIOT
744a590467 Replaced tabs by spaces 2022-12-03 20:47:02 +01:00
Francois BIOT
5a3e9bd90e Fix reboot bugs due to TinyConsole 2022-12-03 20:34:59 +01:00
hsaturn
d8b24adef7 [MqttClient] Fix compilation warnings 2022-12-03 20:25:03 +01:00
hsaturn
4726ff293c [MqttClient] Fix keep_alive decoding 2022-12-03 20:23:15 +01:00
hsaturn
2a4e84d827 [TinyMqtt.h] rework code 2022-12-01 08:07:25 +01:00
hsaturn
9ef47fa6a4 [MqttClient] Renamed parent to local_broker 2022-12-01 07:49:56 +01:00
hsaturn
3358340319 Fix unit tests 2022-11-30 20:52:29 +01:00
Francois BIOT
1fff9fd0e1 Use TinyConsole instead of Serial 2022-11-30 20:06:15 +01:00
hsaturn
d12096ef51 Ooops bad tag 2022-11-23 13:01:24 +01:00
hsaturn
ea56d21190 Merge branch 'main' of github.com:hsaturn/TinyMqtt into main 2022-11-23 12:56:35 +01:00
hsaturn
c802c895b6 Version 0.9.6 bump 2022-11-23 12:54:15 +01:00
hsaturn
074bca971f Update README.md 2022-11-23 12:51:12 +01:00
hsaturn
7bd9c27b89 Merge branch 'main' of github.com:hsaturn/TinyMqtt into main 2022-11-21 04:42:30 +01:00
hsaturn
3e73673302 [tinymqtt-test] Fix lot of errors
- TINY_MQTT_DEBUG was not set (no dump)
  - MqttClient::counter has disapeared (compilation error)
  - payload was badly displayed
  - broker/client names could be reused for client/brokers
2022-11-21 04:42:16 +01:00
hsaturn
4b12aaa198 Update README.md
Unit tests in features
2022-11-21 01:47:47 +01:00
hsaturn
5f9cab8992 Update README.md
Wildcards
2022-11-21 01:44:54 +01:00
hsaturn
96766f7091 [Topic] Wildcards added
+ wildcard added
  # wildcard added
  * wildcard added (but does not appear in mqtt specification...)
  $SYS messages compare is supported
2022-11-21 01:37:55 +01:00
hsaturn
354aec239f [tests] Re-added debug mode in tests
The EXTRA_FLAGS needs some clean before running tests.
This commit allows to compile twice for this situation, but it is not perfect
because EXTRA_CFLAGS can still cause some problems. The tests Makefile should
be able to detect this and to group builds and tests with each own EXTRA flags.
2022-10-31 02:12:15 +01:00
hsaturn
3839a0a830 [tests] Fix unit tests timed out
The debug-tests was changing the compilation (TINY_MQTT_DEBUG on)
Thus the other tests were too long to execute due to Serial output emulation.
2022-10-31 01:49:11 +01:00
hsaturn
0444a4c348 [tests] removed useless code in debug-tests.ino 2022-10-31 01:35:21 +01:00
hsaturn
73207e4745 Renamed TCP_ASYNC define to TINY_MQTT_ASYNC for name consistency. 2022-10-31 01:24:41 +01:00
hsaturn
b7d44445af [tests] Added a debug mode compilation phase (and fixed it) 2022-10-31 01:20:30 +01:00
hsaturn
cce6b2ecfc Update README.md
Added links for examples
2022-10-30 22:01:54 +01:00
hsaturn
883f1e27e6 Release 0.9.3 2022-10-30 21:47:58 +01:00
hsaturn
e7fc147424 Changed tab to spaces 2022-10-30 21:43:45 +01:00
hsaturn
2147b147fc [Readme] Some minor changes 2022-10-30 21:42:59 +01:00
hsaturn
f5e9a43461 [StringIndexer] Fix compare bug and moved strToIndex to private
strToIndex is dangerous because it can increment the use of a string, or
create a new string. This method should only be called by IndexedString.
2022-10-30 21:41:52 +01:00
hsaturn
cabb56fc8c [tests] Added -g3 for tests for debugging purposes 2022-10-30 20:44:44 +01:00
hsaturn
58786eb6d9 Release 0.9.2 2022-10-30 19:48:04 +01:00
hsaturn
776242b259 Release 0.9.1 2022-10-30 19:45:50 +01:00
hsaturn
a9d19c3218 [lib] Remove library.json 2022-10-30 18:31:46 +01:00
hsaturn
7bd299ec07 Use slim lint 2022-10-30 15:23:50 +01:00
hsaturn
107469cd78 Lint action, upgrade to latest 2022-10-30 15:13:42 +01:00
hsaturn
709e1fd567 Tabs changed to spaces (at least) 2022-10-30 13:26:33 +01:00
hsaturn
4eb8f18ebf [MqttMessage] Rewrite of the length encoding 2022-10-30 13:24:05 +01:00
hsaturn
d5d27c8020 Merge pull request #25 from hsaturn/pr22
was pr#22 : added reply to message subscribe and unsubscribe
2022-01-10 05:28:08 +01:00
31 changed files with 3130 additions and 2464 deletions

View File

@@ -17,6 +17,7 @@ jobs:
- name: Setup - name: Setup
run: | run: |
cd .. cd ..
git clone https://github.com/hsaturn/TinyConsole
git clone https://github.com/bxparks/EpoxyDuino git clone https://github.com/bxparks/EpoxyDuino
git clone https://github.com/bxparks/AceRoutine git clone https://github.com/bxparks/AceRoutine
git clone https://github.com/bxparks/AUnit git clone https://github.com/bxparks/AUnit

View File

@@ -1,26 +1,52 @@
name: Super-Linter ---
#################################
#################################
## Super Linter GitHub Actions ##
#################################
#################################
name: Lint Code Base
# Run this workflow every time a new commit pushed to your repository
# #
# Documentation:
# https://docs.github.com/en/actions/learn-github-actions/workflow-syntax-for-github-actions
#
#############################
# Start the job on all push #
#############################
on: push on: push
###############
# Set the Job #
###############
jobs: jobs:
# Set the job key. The key is displayed as the job name build:
# when a job name is not provided
super-lint:
# Name the Job # Name the Job
name: Lint code base name: Lint Code Base
# Set the type of machine to run on # Set the agent to run on
runs-on: ubuntu-latest runs-on: ubuntu-latest
##################
# Load all steps #
##################
steps: steps:
# Checks out a copy of your repository on the ubuntu-latest machine ##########################
- name: Checkout code # Checkout the code base #
uses: actions/checkout@v2 ##########################
- name: Checkout Code
uses: actions/checkout@v3
with:
# Full git history is needed to get a proper
# list of changed files within `super-linter`
fetch-depth: 0
# Runs the Super-Linter action ################################
- name: Run Super-Linter # Run Linter against code base #
uses: github/super-linter@v3 ################################
- name: Lint Code Base
uses: github/super-linter/slim@v4
env: env:
VALIDATE_ALL_CODEBASE: false
# Change to 'master' if your main branch differs
DEFAULT_BRANCH: main DEFAULT_BRANCH: main
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

View File

@@ -17,11 +17,20 @@ TinyMqtt is a small, fast and capable Mqtt Broker and Client for Esp8266 / Esp32
The max I've seen was 2k msg/s (1 client 1 subscription) The max I've seen was 2k msg/s (1 client 1 subscription)
- Act as as a mqtt broker and/or a mqtt client - Act as as a mqtt broker and/or a mqtt client
- Mqtt 3.1.1 / Qos 0 supported - Mqtt 3.1.1 / Qos 0 supported
- Wildcards supported (+ # $ and * (even if not part of the spec...))
- Standalone (can work without WiFi) (degraded/local mode) - Standalone (can work without WiFi) (degraded/local mode)
- Brokers can connect to another broker and becomes then a - Brokers can connect to another broker and becomes then a
proxy for clients that are connected to it. proxy for clients that are connected to it.
- zeroconf, this is a strange but very powerful mode where - zeroconf, this is a strange but very powerful mode where
all brokers tries to connect together on the same local network. all brokers tries to connect together on the same local network.
- small memory footprint (very efficient topic storage)
- long messages are supported (>127 bytes)
- TinyMQTT is largely unit tested, so once a bug is fixed, it is fixed forever
## Limitations
- Max of 255 different topics can be stored (change index_t type to allow more)
- No Qos because messages are not queued but immediately sent to clients
## Quickstart ## Quickstart
@@ -33,10 +42,11 @@ TinyMqtt is a small, fast and capable Mqtt Broker and Client for Esp8266 / Esp32
| Example | Description | | Example | Description |
| ------------------- | ------------------------------------------ | | ------------------- | ------------------------------------------ |
| client-without-wifi | standalone example | | [client-with-wifi](https://github.com/hsaturn/TinyMqtt/tree/main/examples/client-with-wifi/client-with-wifi.ino) | standalone example |
| simple-client | Connect the ESP to an external Mqtt broker | | [client-without-wifi](https://github.com/hsaturn/TinyMqtt/tree/main/examples/client-without-wifi/client-without-wifi.ino) | standalone example |
| simple-broker | Simple Mqtt broker with your ESP | | [simple-client](https://github.com/hsaturn/TinyMqtt/tree/main/examples/simple-client/simple-client.ino) | Connect the ESP to an external Mqtt broker |
| tinymqtt-test | Complex console example | | [simple-broker](https://github.com/hsaturn/TinyMqtt/tree/main/examples/simple-broker/simple-broker.ino) | Simple Mqtt broker with your ESP |
| [tinymqtt-test](https://github.com/hsaturn/TinyMqtt/tree/main/examples/tinymqtt-test/tinymqtt-test.ino) | Complex console example |
- tinymqtt-test : This is a complex sketch with a terminal console - tinymqtt-test : This is a complex sketch with a terminal console
that allows to add clients publish, connect etc with interpreted commands. that allows to add clients publish, connect etc with interpreted commands.
@@ -59,7 +69,7 @@ no need for having tons of clients (also RAM is the problem with many clients)
* ~~MqttClient auto re-subscribe (::resubscribe works bad on broker.emqx.io)~~ * ~~MqttClient auto re-subscribe (::resubscribe works bad on broker.emqx.io)~~
* MqttClient auto reconnection * MqttClient auto reconnection
* MqttClient user/password * MqttClient user/password
* Wildcards (I may implement only # as I'm not interrested by a clever and cpu consuming matching) * ~~Wildcards (I may implement only # as I'm not interrested by a clever and cpu consuming matching)~~
* I suspect that MqttClient::parent could be removed and replaced with a simple boolean * I suspect that MqttClient::parent could be removed and replaced with a simple boolean
(this'll need to rewrite a few functions) (this'll need to rewrite a few functions)

View File

@@ -0,0 +1,134 @@
#include <TinyMqtt.h> // https://github.com/hsaturn/TinyMqtt
#include <MqttClassBinder.h>
/**
* Example on how to bind a class:onPublish function
*
* Local broker that accept connections and two local clients
*
*
* +-----------------------------+
* | ESP |
* | +--------+ | 1883 <--- External client/s
* | +-------->| broker | | 1883 <--- External client/s
* | | +--------+ |
* | | ^ |
* | | | |
* | | | | -----
* | v v | ---
* | +----------+ +----------+ | -
* | | internal | | internal | +-------* Wifi
* | | client | | client | |
* | +----------+ +----------+ |
* | |
* +-----------------------------+
*
* 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
* - MqttClassBinder allows to mix together many mqtt sources
*
* cons - Takes more memory (48 more bytes for the one MqttClassBinder<Class>
* - a bit hard to understand
*
*/
const char *ssid = "";
const char *password = "";
std::string topic_b="sensor/btemp";
std::string topic_sender= "sensor/counter";
MqttBroker broker(1883);
MqttClient mqtt_a(&broker);
MqttClient mqtt_b(&broker);
MqttClient mqtt_sender(&broker);
class MqttReceiver: public MqttClassBinder<MqttReceiver>
{
public:
void onPublish(const MqttClient* source, const Topic& topic, const char* payload, size_t /* length */)
{
Serial
<< " * MqttReceiver received topic (" << topic.c_str() << ")"
<< " from (" << source->id() << "), "
<< " payload: (" << payload << ')' << 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);
if (strlen(ssid)==0)
Serial << "****** PLEASE EDIT THE EXAMPLE AND MODIFY ssid/password *************" << endl;
}
Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
broker.begin();
MqttReceiver* receiver = new MqttReceiver;
// receiver will receive both publication from two MqttClient
// (that could be connected to two different brokers)
MqttClassBinder<MqttReceiver>::onPublish(&mqtt_a, receiver);
MqttClassBinder<MqttReceiver>::onPublish(&mqtt_b, receiver);
mqtt_a.id("mqtt_a");
mqtt_b.id("mqtt_b");
mqtt_sender.id("sender");
mqtt_a.subscribe(topic_b);
mqtt_b.subscribe(topic_sender);
}
void loop()
{
broker.loop(); // Don't forget to add loop for every broker and clients
mqtt_a.loop();
mqtt_b.loop();
mqtt_sender.loop();
// ============= client A publish ================
{
static const int interval = 5000; // publishes every 5s (please avoid usage of delay())
static uint32_t timer = millis() + interval;
if (millis() > timer)
{
static int counter = 0;
Serial << "Sender is publishing " << topic_sender.c_str() << endl;
timer += interval;
mqtt_sender.publish(topic_sender, "sent by Sender, message #"+std::string(String(counter++).c_str()));
}
}
// ============= client B publish ================
{
static const int interval = 7000; // will send topic each 7s
static uint32_t timer = millis() + interval;
static int temperature;
if (millis() > timer)
{
Serial << "B is publishing " << topic_b.c_str() << endl;
timer += interval;
mqtt_b.publish(topic_b, "sent by B: temp="+std::string(String(16+temperature++%6).c_str()));
}
}
}

View File

@@ -26,7 +26,7 @@ void setup()
Serial.begin(115200); Serial.begin(115200);
if (strlen(ssid)==0) if (strlen(ssid)==0)
Serial << "****** PLEASE MODIFY ssid/password *************" << endl; Console << TinyConsole::red << "****** PLEASE MODIFY ssid/password *************" << endl;
WiFi.mode(WIFI_STA); WiFi.mode(WIFI_STA);
WiFi.begin(ssid, password); WiFi.begin(ssid, password);
@@ -35,10 +35,10 @@ void setup()
Serial << '.'; Serial << '.';
delay(500); delay(500);
} }
Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl; Console << TinyConsole::green << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
broker.begin(); broker.begin();
Serial << "Broker ready : " << WiFi.localIP() << " on port " << PORT << endl; Console << "Broker ready : " << WiFi.localIP() << " on port " << PORT << endl;
} }
void loop() void loop()

View File

@@ -1,4 +1,5 @@
#include "TinyMqtt.h" // https://github.com/hsaturn/TinyMqtt #include "TinyMqtt.h" // https://github.com/hsaturn/TinyMqtt
#include "TinyStreaming.h" // https://github.com/hsaturn/TinyConsole
/** Simple Client (The simplest configuration) /** Simple Client (The simplest configuration)
* *
@@ -55,7 +56,7 @@ void setup()
Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl; Serial << "Connected to " << ssid << "IP address: " << WiFi.localIP() << endl;
client.connect(BROKER, BROKER_PORT); // Put here your broker ip / port client.connect(BROKER, BROKER_PORT);
} }
void loop() void loop()

View File

@@ -1,5 +1,7 @@
// 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 <MqttStreaming.h> #include <TinyStreaming.h>
#if defined(ESP8266) #if defined(ESP8266)
#include <ESP8266mDNS.h> #include <ESP8266mDNS.h>
#elif defined(ESP32) #elif defined(ESP32)
@@ -10,9 +12,37 @@
#endif #endif
#include <sstream> #include <sstream>
#include <string>
#include <map> #include <map>
bool echo_on = true; bool echo_on = true;
auto green = TinyConsole::green;
auto red = TinyConsole::red;
auto white = TinyConsole::white;
auto cyan = TinyConsole::cyan;
auto yellow = TinyConsole::yellow;
auto magenta = TinyConsole::magenta;
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 = "";
struct free_broker
{
public:
free_broker(const char* s, uint16_t p, const char* comment) : url(s), port(p) {}
std::string url;
uint16_t port;
};
const std::map<std::string, free_broker> list =
{
{ "mqtthq", { "public.mqtthq.com" , 8083, "publish/subscribe" }},
{ "hivemq", { "broker.hivemq.com", 1883, "" }}
};
/** Very complex example /** Very complex example
* Console allowing to make any kind of test, * Console allowing to make any kind of test,
@@ -25,22 +55,22 @@ bool echo_on = true;
* TODO examples of scripts * TODO examples of scripts
*/ */
const char* ssid = "";
const char* password = "";
std::string topic="sensor/temperature";
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)
{ {
Serial << "--> " << srce->id().c_str() << ": ======> received " << topic.c_str(); Console << cyan << "--> " << srce->id().c_str() << ": received " << topic.c_str() << white;
if (payload) Serial << ", payload[" << length << "]=["; if (payload)
{
Console << ", payload[" << length << "]=[";
while(length--) while(length--)
{ {
const char c=*payload++; const char c=*payload++;
if (c!=10 and c!=13 and c <32) Serial << '?'; if (c<32)
Serial << *payload++; Console << '?';
else
Console << c;
}
Console << ']' << endl;
} }
Serial<< endl;
} }
std::map<std::string, MqttClient*> clients; std::map<std::string, MqttClient*> clients;
@@ -50,23 +80,26 @@ void setup()
{ {
WiFi.persistent(false); // https://github.com/esp8266/Arduino/issues/1054 WiFi.persistent(false); // https://github.com/esp8266/Arduino/issues/1054
Serial.begin(115200); Serial.begin(115200);
Console.begin(Serial);
Console.setPrompt("> ");
Console.setCallback(onCommand);
delay(500); delay(500);
Serial << endl << endl; Console.cls();
Serial << "***************************************************************" << endl; Console << endl << endl;
Serial << "* Welcome to the TinyMqtt console" << endl; Console << yellow
Serial << endl; << "***************************************************************" << endl;
Serial << "* The console allows to test all features of the libraries." << endl; Console << "* Welcome to the TinyMqtt console" << endl;
Serial << endl; Console << "* The console allows to test all features of the libraries." << endl;
Console << "* Enter help to view the list of commands." << endl;
Console << "***************************************************************" << endl;
Console << endl;
if (strlen(ssid)==0) if (strlen(ssid)==0)
Serial << "* WARNING: You may want to modify ssid/password in order" << endl Console << red << "* ERROR: You must modify ssid/password in order" << endl
<< " to reflect your Wifi configuration." << endl; << " to be able to connect to your Wifi network." << endl;
Serial << endl; Console << endl << white;
Serial << "* Enter help to view the list of commands." << endl;
Serial << "***************************************************************" << endl;
Serial << endl;
Serial << "Connecting to '" << ssid << "' "; Console << "Connecting to '" << ssid << "' ";
WiFi.mode(WIFI_STA); WiFi.mode(WIFI_STA);
WiFi.begin(ssid, password); WiFi.begin(ssid, password);
@@ -74,21 +107,23 @@ void setup()
WiFi.begin(ssid, password); WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) while (WiFi.status() != WL_CONNECTED)
{ Serial << '-'; delay(500); } { Console << '-'; delay(500); }
Serial << 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";
Serial << "Starting MDNS, name= " << name; Console << "Starting MDNS, name= " << name;
if (!MDNS.begin(name)) if (!MDNS.begin(name))
Serial << " error, not available." << endl; Console << " error, not available." << endl;
else else
Serial << " ok." << endl; Console << " ok." << endl;
MqttBroker* broker = new MqttBroker(1883); MqttBroker* broker = new MqttBroker(1883);
broker->begin(); broker->begin();
brokers["broker"] = broker; brokers["broker"] = broker;
if (Console.isTerm()) onCommand("every 333 view");
} }
std::string getword(std::string& str, const char* if_empty=nullptr, char sep=' '); std::string getword(std::string& str, const char* if_empty=nullptr, char sep=' ');
@@ -137,7 +172,7 @@ std::string getword(std::string& str, const char* if_empty/*=nullptr*/, char sep
{ {
sword.erase(0,1); sword.erase(0,1);
to = getint(sword); to = getint(sword);
if (sword[0]!=')') Serial << "Missing ')'" << endl; if (sword[0]!=')') Console << "Missing ')'" << endl;
} }
else else
{ {
@@ -148,7 +183,7 @@ std::string getword(std::string& str, const char* if_empty/*=nullptr*/, char sep
} }
else else
{ {
Serial << "Missing '('" << endl; Console << "Missing '('" << endl;
} }
} }
while(str[0]==' ') str.erase(0,1); while(str[0]==' ') str.erase(0,1);
@@ -191,15 +226,16 @@ std::string getip(std::string& str, const char* if_empty=nullptr, char sep=' ')
if (addr.length()) addr += '.'; if (addr.length()) addr += '.';
addr += s; addr += s;
} }
Console << "connect address: " << addr << endl;
return addr; return addr;
} }
std::map<std::string, std::string> vars; std::map<std::string, std::string> vars;
std::set<std::string> commands = { std::set<std::string> commands = {
"auto", "broker", "blink", "client", "connect", "broker", "blink", "client", "connect",
"create", "delete", "help", "interval", "create", "delete", "debug", "help", "interval",
"ls", "ip", "off", "on", "set", "list", "ls", "ip", "off", "on", "set",
"publish", "reset", "subscribe", "unsubscribe", "view", "echo", "every" "publish", "reset", "subscribe", "unsubscribe", "view", "echo", "every"
}; };
@@ -222,7 +258,7 @@ void convertToCommand(std::string& search)
search = matches; search = matches;
else if (count>1) else if (count>1)
{ {
Serial << "Ambiguous command: " << matches << endl; Console << "Ambiguous command: " << matches << endl;
search.clear(); search.clear();
} }
} }
@@ -260,123 +296,6 @@ void replaceVars(std::string& cmd)
} }
// publish at regular interval
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] interval [s]" << 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;
float temp=19;
};
std::map<MqttClient*, automatic*> automatic::autos;
bool compare(std::string s, const char* cmd) bool compare(std::string s, const char* cmd)
{ {
uint8_t p=0; uint8_t p=0;
@@ -400,13 +319,17 @@ struct Every
void dump() void dump()
{ {
Serial << (active ? "enabled " : "disabled "); if (active)
auto mill=millis(); Console << green << "enabled";
Serial << ms << "ms [" << cmd << "] next in ";
if (mill > next)
Serial << "now";
else else
Serial << next-mill << "ms"; Console << red << "disabled";
auto mill=millis();
Console << white << ms << "ms [" << cmd << "] next in ";
if (mill > next)
Console << "now";
else
Console << next-mill << "ms";
} }
}; };
@@ -418,6 +341,34 @@ int16_t blink;
std::vector<Every> everies; std::vector<Every> everies;
void onCommand(const std::string& command)
{
Console << endl;
std::string cmd=command;
if (cmd.substr(0,3)!="set") replaceVars(cmd);
eval(cmd);
Console << endl;
Console.prompt();
}
void clientConnect(MqttClient* client, std::string& cmd)
{
std::string remote = getword(cmd);
uint16_t port;
auto it=list.find(remote);
if (it != list.end())
{
Console << "Connecting to free broker: " << remote << endl;
remote = it->second.url;
port=it->second.port;
}
else
port=getint(cmd);
client->connect(remote.c_str(), port, getint(cmd, 60));
Console << (client->connected() ? "connected." : "not connected") << endl;
}
void eval(std::string& cmd) void eval(std::string& cmd)
{ {
while(cmd.length()) while(cmd.length())
@@ -446,7 +397,7 @@ void eval(std::string& cmd)
} }
else else
{ {
Serial << "Unknown class (" << s.c_str() << ")" << endl; Console << red << "Unknown class (" << s.c_str() << ")" << white << endl;
cmd.clear(); cmd.clear();
} }
} }
@@ -456,6 +407,22 @@ void eval(std::string& cmd)
if (s.length()) convertToCommand(s); if (s.length()) convertToCommand(s);
if (s.length()==0) if (s.length()==0)
{} {}
else if (compare(s, "debug"))
{
#if TINY_MQTT_DEBUG
TinyMqtt::debug = getint(cmd);
#else
Console << red << "TinyMqtt not compiled in debug" << endl;
#endif
}
else if (compare(s, "list"))
{
Console << "List of free servers" << endl;
for(const auto& fb: list)
{
Console << " " << fb.first << " : " << fb.second.url << ":" << fb.second.port << endl;
}
}
else if (compare(s, "delete")) else if (compare(s, "delete"))
{ {
if (client==nullptr && broker==nullptr) if (client==nullptr && broker==nullptr)
@@ -470,14 +437,14 @@ void eval(std::string& cmd)
broker = brokers[s]; broker = brokers[s];
} }
else else
Serial << "Unable to find (" << s.c_str() << ")" << endl; Console << red << "Unable to find (" << s.c_str() << ")" << white << endl;
} }
if (client) if (client)
{ {
for (auto it: clients) for (auto it: clients)
{ {
if (it.second != client) continue; if (it.second != client) continue;
Serial << "deleted" << endl; Console << "deleted" << endl;
delete (it.second); delete (it.second);
clients.erase(it.first); clients.erase(it.first);
break; break;
@@ -489,7 +456,7 @@ void eval(std::string& cmd)
for(auto it: brokers) for(auto it: brokers)
{ {
if (broker != it.second) continue; if (broker != it.second) continue;
Serial << "deleted" << endl; Console << "deleted" << endl;
delete (it.second); delete (it.second);
brokers.erase(it.first); brokers.erase(it.first);
break; break;
@@ -497,13 +464,13 @@ void eval(std::string& cmd)
cmd += " ls"; cmd += " ls";
} }
else else
Serial << "Nothing to delete" << endl; Console << "Nothing to delete" << endl;
} }
else if (broker) else if (broker)
{ {
if (compare(s,"connect")) if (compare(s,"connect"))
{ {
Serial << "NYI" << endl; Console << "NYI" << endl;
} }
else if (compare(s, "view")) else if (compare(s, "view"))
{ {
@@ -511,7 +478,7 @@ void eval(std::string& cmd)
} }
else else
{ {
Serial << "Unknown broker command (" << s << ")" << endl; Console << "Unknown broker command (" << s << ")" << endl;
s.clear(); s.clear();
} }
} }
@@ -519,20 +486,19 @@ void eval(std::string& cmd)
{ {
if (compare(s,"connect")) if (compare(s,"connect"))
{ {
client->connect(getip(cmd,"192.168.1.40").c_str(), getint(cmd, 1883), getint(cmd, 60)); clientConnect(client, cmd);
Serial << (client->connected() ? "connected." : "not connected") << endl;
} }
else if (compare(s,"publish")) else if (compare(s,"publish"))
{ {
retval = client->publish(getword(cmd, topic.c_str()), getword(cmd)); retval = client->publish(getword(cmd), getword(cmd));
} }
else if (compare(s,"subscribe")) else if (compare(s,"subscribe"))
{ {
client->subscribe(getword(cmd, topic.c_str())); client->subscribe(getword(cmd));
} }
else if (compare(s, "unsubscribe")) else if (compare(s, "unsubscribe"))
{ {
client->unsubscribe(getword(cmd, topic.c_str())); client->unsubscribe(getword(cmd));
} }
else if (compare(s, "view")) else if (compare(s, "view"))
{ {
@@ -540,7 +506,7 @@ void eval(std::string& cmd)
} }
else else
{ {
Serial << "Unknown client command (" << s << ")" << endl; Console << "Unknown client command (" << s << ")" << endl;
s.clear(); s.clear();
} }
} }
@@ -565,10 +531,10 @@ void eval(std::string& cmd)
echo_on = false; echo_on = false;
else else
{ {
Serial << s << ' '; Console << s << ' ';
while(cmd.length()) while(cmd.length())
{ {
Serial << getword(cmd) << ' '; Console << getword(cmd) << ' ';
} }
} }
} }
@@ -585,7 +551,7 @@ void eval(std::string& cmd)
every.next=millis()+ms; every.next=millis()+ms;
everies.push_back(every); everies.push_back(every);
every.dump(); every.dump();
Serial << endl; Console << endl;
cmd.clear(); cmd.clear();
} }
} }
@@ -600,7 +566,7 @@ void eval(std::string& cmd)
{ {
if (every.active != active) if (every.active != active)
{ {
Serial << "every #" << count << (active ? " on" :" off") << endl; Console << "every #" << count << (active ? " on" :" off") << endl;
every.active = active; every.active = active;
every.underrun = 0; every.underrun = 0;
} }
@@ -611,19 +577,19 @@ void eval(std::string& cmd)
else if (compare(cmd, "list") or cmd.length()==0) else if (compare(cmd, "list") or cmd.length()==0)
{ {
getword(cmd); getword(cmd);
Serial << "List of everies (ms=" << millis() << ")" << endl; Console << "List of everies (ms=" << millis() << ")" << endl;
uint8_t count=0; uint8_t count=0;
for(auto& every: everies) for(auto& every: everies)
{ {
Serial << count << ": "; Console << count << ": ";
every.dump(); every.dump();
Serial << endl; Console << endl;
count++; count++;
} }
} }
else if (compare(cmd, "remove")) else if (compare(cmd, "remove"))
{ {
Serial << "Removing..." << endl; Console << "Removing..." << endl;
getword(cmd); getword(cmd);
int8_t every=getint(cmd, -1); int8_t every=getint(cmd, -1);
if (every==-1 and compare(cmd, "last") and everies.size()) if (every==-1 and compare(cmd, "last") and everies.size())
@@ -641,10 +607,10 @@ void eval(std::string& cmd)
everies.erase(everies.begin()+every); everies.erase(everies.begin()+every);
} }
else else
Serial << "Bad colmmand" << endl; Console << "Bad colmmand" << endl;
} }
else else
Serial << "Bad command" << endl; Console << "Bad command" << endl;
} }
else if (compare(s, "blink")) else if (compare(s, "blink"))
{ {
@@ -655,7 +621,7 @@ void eval(std::string& cmd)
blink_ms_off[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); pinMode(blink_nr, OUTPUT);
blink_next[blink_nr] = millis(); blink_next[blink_nr] = millis();
Serial << "Blink " << blink_nr << ' ' << (blink_ms_on[blink_nr] ? "on" : "off") << endl; Console << "Blink " << blink_nr << ' ' << (blink_ms_on[blink_nr] ? "on" : "off") << endl;
if (blink_ms_on[blink_nr]) if (blink_ms_on[blink_nr])
blink |= 1<< blink_nr; blink |= 1<< blink_nr;
else else
@@ -664,16 +630,15 @@ void eval(std::string& cmd)
} }
} }
} }
else if (compare(s, "auto"))
{
automatic::command(client, cmd);
if (client == nullptr)
cmd.clear();
}
else if (compare(s, "broker")) else if (compare(s, "broker"))
{ {
std::string id=getword(cmd); std::string id=getword(cmd);
if (id.length() or brokers.find(id)!=brokers.end()) 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())
{ {
int port=getint(cmd, 0); int port=getint(cmd, 0);
if (port) if (port)
@@ -682,40 +647,56 @@ void eval(std::string& cmd)
broker->begin(); broker->begin();
brokers[id] = broker; brokers[id] = broker;
Serial << "new broker (" << id.c_str() << ")" << endl; Console << "new broker (" << id.c_str() << ")" << endl;
} }
else else
Serial << "Missing port" << endl; {
Console << "Missing port" << endl;
cmd.clear();
}
} }
else else
Serial << "Missing or existing broker name (" << id.c_str() << ")" << endl; {
cmd+=" ls"; Console << "Missing or existing broker name (" << id.c_str() << ")" << endl;
cmd.clear();
}
} }
else if (compare(s, "client")) else if (compare(s, "client"))
{ {
std::string id=getword(cmd); std::string id=getword(cmd);
if (id.length() or clients.find(id)!=clients.end()) if (brokers.find(id) != brokers.end())
{ {
s=getword(cmd); // broker name Console << "A broker have that name" << endl;
if (s=="" or brokers.find(s) != brokers.end()) cmd.clear();
}
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())
{ {
MqttBroker* broker = nullptr; MqttBroker* broker = nullptr;
if (s.length()) broker = brokers[s]; if (s.length()) broker = brokers[s];
MqttClient* client = new MqttClient(broker); MqttClient* client = new MqttClient(broker, id);
client->id(id);
clients[id]=client; clients[id]=client;
client->setCallback(onPublish); client->setCallback(onPublish);
client->subscribe(topic); if (list.find(s) != list.end())
Serial << "new client (" << id.c_str() << ", " << s.c_str() << ')' << endl; {
cmd=s+' '+cmd;
clientConnect(client, cmd);
}
Console << "new client (" << id.c_str() << ", " << s.c_str() << ')' << endl;
} }
else if (s.length()) else if (s.length())
{ {
Serial << " not found." << endl; Console << " not found." << endl;
cmd.clear();
} }
} }
else else
Serial << "Missing or existing client name" << endl; {
cmd+=" ls"; Console << "Missing or existing client name" << endl;
cmd.clear();
}
} }
else if (compare(s, "set")) else if (compare(s, "set"))
{ {
@@ -724,12 +705,12 @@ void eval(std::string& cmd)
{ {
for(auto it: vars) for(auto it: vars)
{ {
Serial << " " << it.first << " -> " << it.second << endl; Console << " " << it.first << " -> " << it.second << endl;
} }
} }
else if (commands.find(name) != commands.end()) else if (commands.find(name) != commands.end())
{ {
Serial << "Reserved keyword (" << name << ")" << endl; Console << "Reserved keyword (" << name << ")" << endl;
cmd.clear(); cmd.clear();
} }
else else
@@ -745,63 +726,76 @@ void eval(std::string& cmd)
} }
else if (compare(s, "ls") or compare(s, "view")) else if (compare(s, "ls") or compare(s, "view"))
{ {
Serial << "--< " << clients.size() << " client/s. >--" << endl; bool view = compare(s, "view");
if (view)
{
Console << save_cursor << magenta;
Console.gotoxy(1,1);
}
Console << "--< " << '/' << clients.size() << " client/s. >--" << erase_to_end << endl;
for(auto it: clients) for(auto it: clients)
{ {
it.second->dump(" "); it.second->dump(" ");
} }
Serial << "--< " << brokers.size() << " brokers/s. >--" << endl; Console << "--< " << brokers.size() << " brokers/s. >--" << erase_to_end << endl;
for(auto it: brokers) for(auto it: brokers)
{ {
Serial << " +-- '" << it.first.c_str() << "' " << it.second->clientsCount() << " client/s."<< endl; Console << " +-- '" << it.first.c_str() << "' " << it.second->clientsCount() << " client/s."<< erase_to_end << endl;
it.second->dump(" "); it.second->dump(" ");
} }
if (view)
{
Console.bg(white);
Console << erase_to_end << restore_cursor;
}
} }
else if (compare(s, "reset")) else if (compare(s, "reset"))
ESP.restart(); ESP.restart();
else if (compare(s, "ip")) else if (compare(s, "ip"))
Serial << "IP: " << WiFi.localIP() << endl; Console << "IP: " << WiFi.localIP() << endl;
else if (compare(s,"help")) else if (compare(s,"help"))
{ {
Serial << "syntax:" << endl; Console << "syntax:" << endl;
Serial << " MqttBroker:" << endl; Console << " MqttBroker:" << endl;
Serial << " broker {name} {port} : create a new broker" << endl; Console << " broker {broker_name} {port} : create a new broker" << endl;
Serial << endl; Console << " broker_name can be one of 'list'" << endl;
Serial << " MqttClient:" << endl; Console << " broker_name.delete : delete a broker (buggy)" << endl;
Serial << " client {name} {parent broker} : create a client then" << endl; Console << " broker_name.view : dump a broker" << endl;
Serial << " name.connect [ip] [port] [alive]" << endl; Console << endl;
Serial << " name.[un]subscribe [topic]" << endl; Console << " MqttClient:" << endl;
Serial << " name.publish [topic][payload]" << endl; Console << " client {name} {broker} : create a client then" << endl;
Serial << " name.view" << endl; Console << " name.connect [ip] [port] [alive]" << endl;
Serial << " name.delete" << endl; Console << " name.[un]subscribe topic" << endl;
Console << " name.publish topic [payload]" << endl;
automatic::help(); Console << " name.view" << endl;
Serial << endl; Console << " name.delete" << endl;
Serial << " help" << endl; Console << endl;
Serial << " blink [Dx on_ms off_ms]" << endl; Console << " list : list of free brokers (debug 1 advised)" << endl;
Serial << " ls / ip / reset" << endl; Console << " debug #" << endl;
Serial << " set [name][value]" << endl; Console << " list : get list of free brokers" << endl;
Serial << " ! repeat last command" << endl; Console << " blink [Dx on_ms off_ms] : make pin blink" << endl;
Serial << endl; Console << " ls / ip / reset" << endl;
Serial << " echo [on|off] or strings" << endl; Console << " set [name][value]" << endl;
Serial << " every ms [command]; every list; every remove [nr|all], every (on|off) [#]" << endl; Console << " ! repeat last command" << endl;
Serial << " on {output}; off {output}" << endl; Console << endl;
Serial << " $id : name of the client." << endl; Console << " echo [on|off] or strings" << endl;
Serial << " rnd[(min[,max])] random number." << endl; Console << " every ms [command]; every list; every remove [nr|all]; every (on|off) [#]" << endl;
Serial << " default topic is '" << topic.c_str() << "'" << endl; Console << " on {output}; off {output}" << endl;
Serial << endl; Console << " $id : name of the client." << endl;
Console << " rnd[(min[,max])] random number." << endl;
Console << endl;
} }
else else
{ {
while(s[0]==' ') s.erase(0,1); while(s[0]==' ') s.erase(0,1);
if (s.length()) if (s.length())
Serial << "Unknown command (" << s.c_str() << ")" << endl; Console << "Unknown command (" << s.c_str() << ")" << endl;
} }
if (retval != MqttOk) if (retval != MqttOk)
{ {
Serial << "# MQTT ERROR " << retval << endl; Console << "# MQTT ERROR " << retval << endl;
} }
} }
} }
@@ -823,7 +817,8 @@ void loop()
every.next += every.ms; every.next += every.ms;
if (ms > every.next and ms > every.underrun) if (ms > every.next and ms > every.underrun)
{ {
Serial << "Underrun every #" << e << ", " << (ms - every.next) << "ms late" << endl; every.next += every.ms;
Console << yellow << "Underrun every #" << e << ", " << (ms - every.next) << "ms late" << endl;
every.underrun = ms+5000; every.underrun = ms+5000;
} }
} }
@@ -851,45 +846,15 @@ void loop()
} }
static long count; static long count;
#if defined(ESP9266) #if defined(ESP9266)
MDNS.update(); MDNS.update();
#endif #endif
if (MqttClient::counter != count)
{
Serial << "# " << MqttClient::counter << endl;
count = MqttClient::counter;
}
for(auto it: brokers) for(auto it: brokers)
it.second->loop(); it.second->loop();
for(auto it: clients) for(auto it: clients)
it.second->loop(); it.second->loop();
automatic::loop(); Console.loop();
if (Serial.available())
{
static std::string cmd;
char c=Serial.read();
if (echo_on)
Serial << c;
if (c==10 or c==13)
{
Serial << "----------------[ " << cmd.c_str() << " ]--------------" << endl;
static std::string last_cmd;
if (cmd=="!")
cmd=last_cmd;
else
last_cmd=cmd;
if (cmd.substr(0,3)!="set") replaceVars(cmd);
eval(cmd);
}
else
{
cmd=cmd+c;
}
}
} }

View File

@@ -1,18 +0,0 @@
{
"name": "TinyMqtt",
"keywords": "ethernet, mqtt, m2m, iot",
"description": "MQTT is a lightweight messaging protocol ideal for small devices. This library allows to send and receive and host a broker for MQTT. It does support MQTT 3.1.1 with QOS=0 on ESP8266 and ESP32 WROOM platfrms.",
"repository": {
"type": "git",
"url": "https://github.com/hsaturn/TinyMqtt.git"
},
"version": "0.8.0",
"exclude": "",
"examples": "examples/*/*.ino",
"frameworks": "arduino",
"platforms": [
"atmelavr",
"espressif8266",
"espressif32"
]
}

View File

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

72
src/MqttClassBinder.h Normal file
View File

@@ -0,0 +1,72 @@
// MqttReceiver must implement onPublish(...)
template <class MqttReceiver>
class MqttClassBinder
{
public:
MqttClassBinder()
{
unregister(this);
}
~MqttClassBinder() { unregister(this); }
static void onUnpublished(MqttClient::CallBack handler)
{
unrouted_handler = handler;
}
static void onPublish(MqttClient* client, MqttReceiver* dest)
{
routes.insert(std::pair<MqttClient*, MqttReceiver*>(client, dest));
client->setCallback(onRoutePublish);
}
void onPublish(const MqttClient* client, const Topic& topic, const char* payload, size_t length)
{
static_cast<MqttReceiver*>(this)->MqttReceiver::onPublish(client, topic, payload, length);
}
static size_t size() { return routes.size(); }
static void reset() { routes.clear(); }
private:
static void onRoutePublish(const MqttClient* client, const Topic& topic, const char* payload, size_t length)
{
bool unrouted = true;
auto receivers = routes.equal_range(client);
for(auto it = receivers.first; it != receivers.second; ++it)
{
it->second->onPublish(client, topic, payload, length);
unrouted = false;
}
if (unrouted and unrouted_handler)
{
unrouted_handler(client, topic, payload, length);
}
}
private:
void unregister(MqttClassBinder<MqttReceiver>* which)
{
if (routes.size()==0) return; // bug in map stl
for(auto it=routes.begin(); it!=routes.end(); it++)
if (it->second == which)
{
routes.erase(it);
return;
}
}
static std::multimap<const MqttClient*, MqttClassBinder<MqttReceiver>*> routes;
static MqttClient::CallBack unrouted_handler;
};
template<class MqttReceiver>
std::multimap<const MqttClient*, MqttClassBinder<MqttReceiver>*> MqttClassBinder<MqttReceiver>::routes;
template<class MqttReceiver>
MqttClient::CallBack MqttClassBinder<MqttReceiver>::unrouted_handler = nullptr;

View File

@@ -1,413 +0,0 @@
/* MqttStreaming.h - Fork of Streaming.h adding std::string and with some minor fixes
* (I have to speek to the author in order to include my changes to his library if possible)
**/
/*
Streaming.h - Arduino library for supporting the << streaming operator
Copyright (c) 2010-2012 Mikal Hart. All rights reserved.
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.
This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
/*
Version 6 library changes
Copyright (c) 2019 Gazoodle. All rights reserved.
1. _BASED moved to template to remove type conversion to long and
sign changes which break int8_t and int16_t negative numbers.
The print implementation still upscales to long for it's internal
print routine.
2. _PAD added to allow padding & filling of characters to the stream
3. _WIDTH & _WIDTHZ added to allow width printing with space padding
and zero padding for numerics
4. Simple _FMT mechanism ala printf, but without the typeunsafetyness
and no internal buffers for replaceable stream printing
*/
#pragma once
#ifndef ARDUINO_STREAMING
#define ARDUINO_STREAMING
#if (defined(ARDUINO) && ARDUINO >= 100) || defined(EPOXY_DUINO)
#include "Arduino.h"
#else
#ifndef STREAMING_CONSOLE
#include "WProgram.h"
#endif
#endif
#include <string>
#if defined(ARDUINO_ARCH_AVR) || defined(ARDUINO_ARCH_MEGAAVR)
// No stl library, so need trivial version of std::is_signed ...
namespace std {
template<typename T>
struct is_signed { static const bool value = false; };
template<>
struct is_signed<int8_t> { static const bool value = true; };
template<>
struct is_signed<int16_t> { static const bool value = true; };
template<>
struct is_signed<int32_t> { static const bool value = true; };
};
#else
#include <type_traits>
#endif
#define STREAMING_LIBRARY_VERSION 6
#if !defined(typeof)
#define typeof(x) __typeof__(x)
#endif
// PrintBuffer implementation of Print, a small buffer to print in
// see its use with pad_float()
template <size_t N>
class PrintBuffer : public Print
{
size_t pos = 0;
char str[N] {};
public:
inline const char *operator() ()
{ return str; };
// inline void clear()
// { pos = 0; str[0] = '\0'; };
inline size_t write(uint8_t c)
{ return write(&c, 1); };
inline size_t write(const uint8_t *buffer, size_t size)
{
size_t s = std::min(size, N-1 - pos); // need a /0 left
if (s)
{
memcpy(&str[pos], buffer, s);
pos += s;
}
return s;
};
};
// Generic template
template<class T>
inline Print &operator <<(Print &stream, const T &arg)
{ stream.print(arg); return stream; }
// TODO sfinae maybe could do the trick ?
inline Print &operator <<(Print &stream, const std::string &str)
{ stream.print(str.c_str()); return stream; }
template<typename T>
struct _BASED
{
T val;
int base;
_BASED(T v, int b): val(v), base(b)
{}
};
#if ARDUINO >= 100
struct _BYTE_CODE
{
byte val;
_BYTE_CODE(byte v) : val(v)
{}
};
#define _BYTE(a) _BYTE_CODE(a)
inline Print &operator <<(Print &obj, const _BYTE_CODE &arg)
{ obj.write(arg.val); return obj; }
#else
#define _BYTE(a) _BASED<typeof(a)>(a, BYTE)
#endif
#define _HEX(a) _BASED<typeof(a)>(a, HEX)
#define _DEC(a) _BASED<typeof(a)>(a, DEC)
#define _OCT(a) _BASED<typeof(a)>(a, OCT)
#define _BIN(a) _BASED<typeof(a)>(a, BIN)
// Specialization for class _BASED
// Thanks to Arduino forum user Ben Combee who suggested this
// clever technique to allow for expressions like
// Serial << _HEX(a);
template<typename T>
inline Print &operator <<(Print &obj, const _BASED<T> &arg)
{ obj.print(arg.val, arg.base); return obj; }
#if ARDUINO >= 18 || defined(EPOXY_DUINO)
// Specialization for class _FLOAT
// Thanks to Michael Margolis for suggesting a way
// to accommodate Arduino 0018's floating point precision
// feature like this:
// Serial << _FLOAT(gps_latitude, 6); // 6 digits of precision
struct _FLOAT
{
double val; // only Print::print(double)
int digits;
_FLOAT(double v, int d): val(v), digits(d)
{}
};
inline Print &operator <<(Print &obj, const _FLOAT &arg)
{ obj.print(arg.val, arg.digits); return obj; }
#endif
// Specialization for enum _EndLineCode
// Thanks to Arduino forum user Paul V. who suggested this
// clever technique to allow for expressions like
// Serial << "Hello!" << endl;
enum _EndLineCode { endl };
inline Print &operator <<(Print &obj, _EndLineCode)
{ obj.println(); return obj; }
// Specialization for padding & filling, mainly utilized
// by the width printers
//
// Use like
// Serial << _PAD(10,' '); // Will output 10 spaces
// Serial << _PAD(4, '0'); // Will output 4 zeros
struct _PAD
{
int8_t width;
char chr;
_PAD(int8_t w, char c) : width(w), chr(c) {}
};
inline Print &operator <<(Print& stm, const _PAD &arg)
{
for(int8_t i = 0; i < arg.width; i++)
stm.print(arg.chr);
return stm;
}
// Specialization for width printing
//
// Use like Result
// -------- ------
// Serial << _WIDTH(1,5) " 1"
// Serial << _WIDTH(10,5) " 10"
// Serial << _WIDTH(100,5) " 100"
// Serial << _WIDTHZ(1,5) "00001"
//
// Great for times & dates, or hex dumps
//
// Serial << _WIDTHZ(hour,2) << ':' << _WIDTHZ(min,2) << ':' << _WIDTHZ(sec,2)
//
// for(int index=0; index<byte_array_size; index++)
// Serial << _WIDTHZ(_HEX(byte_array[index]))
template<typename T>
struct __WIDTH
{
const T val;
int8_t width;
char pad;
__WIDTH(const T& v, int8_t w, char p) : val(v), width(w), pad(p) {}
};
// Count digits in an integer of specific base
template<typename T>
inline uint8_t digits(T v, int8_t base = 10)
{
uint8_t digits = 0;
if ( std::is_signed<T>::value )
{
if ( v < 0 )
{
digits++;
v = -v; // v needs to be postive for the digits counter to work
}
}
do
{
v /= base;
digits++;
} while( v > 0 );
return digits;
}
// Generic get the width of a value in base 10
template<typename T>
inline uint8_t get_value_width(T val)
{ return digits(val); }
inline uint8_t get_value_width(const char * val)
{ return strlen(val); }
#ifdef ARDUINO
inline uint8_t get_value_width(const __FlashStringHelper * val)
{ return strlen_P(reinterpret_cast<const char *>(val)); }
#endif
// _BASED<T> get the width of a value
template<typename T>
inline uint8_t get_value_width(_BASED<T> b)
{ return digits(b.val, b.base); }
// Constructor wrapper to allow automatic template parameter deduction
template<typename T>
__WIDTH<T> _WIDTH(T val, int8_t width) { return __WIDTH<T>(val, width, ' '); }
template<typename T>
__WIDTH<T> _WIDTHZ(T val, int8_t width) { return __WIDTH<T>(val, width, '0'); }
// Operator overload to handle width printing.
template<typename T>
inline Print &operator <<(Print &stm, const __WIDTH<T> &arg)
{ stm << _PAD(arg.width - get_value_width(arg.val), arg.pad) << arg.val; return stm; }
// explicit Operator overload to handle width printing of _FLOAT, double and float
template<typename T>
inline Print &pad_float(Print &stm, const __WIDTH<T> &arg, const double val, const int digits = 2) // see Print::print(double, int = 2)
{
PrintBuffer<32> buf; // it's only ~45B on the stack, no allocation, leak or fragmentation
size_t size = buf.print(val, digits); // print in buf
return stm << _PAD(arg.width - size, arg.pad) << buf(); // pad and concat what's in buf
}
inline Print &operator <<(Print &stm, const __WIDTH<float> &arg)
{ return pad_float(stm, arg, arg.val); }
inline Print &operator <<(Print &stm, const __WIDTH<double> &arg)
{ return pad_float(stm, arg, arg.val); }
inline Print &operator <<(Print &stm, const __WIDTH<_FLOAT> &arg)
{ auto& f = arg.val; return pad_float(stm, arg, f.val, f.digits); }
// a less verbose _FLOATW for _WIDTH(_FLOAT)
#define _FLOATW(val, digits, width) _WIDTH<_FLOAT>(_FLOAT((val), (digits)), (width))
// Specialization for replacement formatting
//
// Designed to be similar to printf that everyone knows and loves/hates. But without
// the internal buffers and type agnosticism. This version only has placeholders in
// the format string, the actual values are supplied using the stream safe operators
// defined in this library.
//
// Use like this:
//
// Serial << FMT(F("Replace % with %"), 1, 2 )
// Serial << FMT("Time is %:%:%", _WIDTHZ(hours,2), _WIDTHZ(minutes,2), _WIDTHZ(seconds,2))
// Serial << FMT("Your score is %\\%", score); // Note the \\ to escape the % sign
// Ok, hold your hats. This is a foray into C++11's variadic template engine ...
inline char get_next_format_char(const char *& format_string)
{
char format_char = *format_string;
if ( format_char > 0 ) format_string++;
return format_char;
}
#ifdef ARDUINO
inline char get_next_format_char(const __FlashStringHelper*& format_string)
{
char format_char = pgm_read_byte(format_string);
if ( format_char > 0 ) format_string = reinterpret_cast<const __FlashStringHelper*>(reinterpret_cast<const char *>(format_string)+1);
return format_char;
}
#endif
template<typename Ft>
inline bool check_backslash(char& format_char, Ft& format_string)
{
if ( format_char == '\\')
{
format_char = get_next_format_char(format_string);
return true;
}
return false;
}
// The template tail printer helper
template<typename Ft, typename... Ts>
struct __FMT
{
Ft format_string;
__FMT(Ft f, Ts ... args) : format_string(f) {}
inline void tstreamf(Print& stm, Ft format) const
{
while(char c = get_next_format_char(format))
{
check_backslash(c, format);
if ( c )
stm.print(c);
}
}
};
// The variadic template helper
template<typename Ft, typename T, typename... Ts>
struct __FMT<Ft, T, Ts...> : __FMT<Ft, Ts...>
{
T val;
__FMT(Ft f, T t, Ts... ts) : __FMT<Ft, Ts...>(f, ts...), val(t) {}
inline void tstreamf(Print& stm, Ft format) const
{
while(char c = get_next_format_char(format))
{
if (!check_backslash(c, format))
{
if ( c == '%')
{
stm << val;
// Variadic recursion ... compiler rolls this out during
// template argument pack expansion
__FMT<Ft, Ts...>::tstreamf(stm, format);
return;
}
}
if (c)
stm.print(c);
}
}
};
// The actual operator should you only instanciate the FMT
// helper with a format string and no parameters
template<typename Ft, typename... Ts>
inline Print& operator <<(Print &stm, const __FMT<Ft, Ts...> &args)
{
args.tstreamf(stm, args.format_string);
return stm;
}
// The variadic stream helper
template<typename Ft, typename T, typename... Ts>
inline Print& operator <<(Print &stm, const __FMT<Ft, T, Ts...> &args)
{
args.tstreamf(stm, args.format_string);
return stm;
}
// As we don't have C++17, we can't get a constructor to use
// automatic argument deduction, but ... this little trick gets
// around that ...
template<typename Ft, typename... Ts>
__FMT<Ft, Ts...> _FMT(Ft format, Ts ... args) { return __FMT<Ft, Ts...>(format, args...); }
#endif

View File

@@ -1,3 +1,4 @@
// vim: ts=2 sw=2 expandtab
#pragma once #pragma once
#include <map> #include <map>
#include <string> #include <string>
@@ -26,30 +27,7 @@ class StringIndexer
#endif #endif
}; };
public: public:
using index_t=uint8_t; using index_t = uint8_t;
static index_t strToIndex(const char* str, uint8_t len)
{
for(auto it=strings.begin(); it!=strings.end(); it++)
{
if (strncmp(it->second.str.c_str(), str, len)==0)
{
it->second.used++;
return it->first;
}
}
for(index_t index=1; index; index++)
{
if (strings.find(index)==strings.end())
{
strings[index].str = std::string(str, len);
strings[index].used++;
// Serial << "Creating index " << index << " for (" << strings[index].str.c_str() << ") len=" << len << endl;
return index;
}
}
return 0; // TODO out of indexes
}
static const std::string& str(const index_t& index) static const std::string& str(const index_t& index)
{ {
@@ -82,6 +60,32 @@ class StringIndexer
static uint16_t count() { return strings.size(); } static uint16_t count() { return strings.size(); }
private: private:
friend class IndexedString;
// increment use of str or create a new index
static index_t strToIndex(const char* str, uint8_t len)
{
for(auto it=strings.begin(); it!=strings.end(); it++)
{
if (it->second.str.length() == len && strcmp(it->second.str.c_str(), str)==0)
{
it->second.used++;
return it->first;
}
}
for(index_t index=1; index; index++)
{
if (strings.find(index)==strings.end())
{
strings[index].str = std::string(str, len);
strings[index].used++;
// Serial << "Creating index " << index << " for (" << strings[index].str.c_str() << ") len=" << len << endl;
return index;
}
}
return 0; // TODO out of indexes
}
static std::map<index_t, StringCounter> strings; static std::map<index_t, StringCounter> strings;
}; };

View File

@@ -1,6 +1,16 @@
// vim: ts=2 sw=2 expandtab
#include "TinyMqtt.h" #include "TinyMqtt.h"
#include <sstream> #include <sstream>
#if TINY_MQTT_DEBUG
static auto cyan = TinyConsole::cyan;
static auto white = TinyConsole::white;
static auto red = TinyConsole::red;
static auto yellow = TinyConsole::yellow;
int TinyMqtt::debug=2;
#endif
#ifdef EPOXY_DUINO #ifdef EPOXY_DUINO
std::map<MqttMessage::Type, int> MqttClient::counters; std::map<MqttMessage::Type, int> MqttClient::counters;
#endif #endif
@@ -8,7 +18,7 @@
MqttBroker::MqttBroker(uint16_t port) MqttBroker::MqttBroker(uint16_t port)
{ {
server = new TcpServer(port); server = new TcpServer(port);
#ifdef TCP_ASYNC #ifdef TINY_MQTT_ASYNC
server->onClient(onClient, this); server->onClient(onClient, this);
#endif #endif
} }
@@ -23,10 +33,11 @@ MqttBroker::~MqttBroker()
} }
// private constructor used by broker only // private constructor used by broker only
MqttClient::MqttClient(MqttBroker* parent, TcpClient* new_client) MqttClient::MqttClient(MqttBroker* local_broker, TcpClient* new_client)
: parent(parent) : local_broker(local_broker)
{ {
#ifdef TCP_ASYNC debug("MqttClient private with broker");
#ifdef TINY_MQTT_ASYNC
client = new_client; client = new_client;
client->onData(onData, this); client->onData(onData, this);
// client->onConnect() TODO // client->onConnect() TODO
@@ -41,12 +52,12 @@ MqttClient::MqttClient(MqttBroker* parent, TcpClient* new_client)
#endif #endif
} }
MqttClient::MqttClient(MqttBroker* parent, const std::string& id) MqttClient::MqttClient(MqttBroker* local_broker, const std::string& id)
: parent(parent), clientId(id) : local_broker(local_broker), clientId(id)
{ {
client = nullptr; client = nullptr;
if (parent) parent->addClient(this); if (local_broker) local_broker->addClient(this);
} }
MqttClient::~MqttClient() MqttClient::~MqttClient()
@@ -64,59 +75,68 @@ void MqttClient::close(bool bSendDisconnect)
if (bSendDisconnect and client->connected()) if (bSendDisconnect and client->connected())
{ {
message.create(MqttMessage::Type::Disconnect); message.create(MqttMessage::Type::Disconnect);
message.hexdump("close");
message.sendTo(this); message.sendTo(this);
} }
client->stop(); client->stop();
} }
if (parent) if (local_broker)
{ {
parent->removeClient(this); local_broker->removeClient(this);
parent = nullptr; local_broker = nullptr;
} }
} }
void MqttClient::connect(MqttBroker* parentBroker) void MqttClient::connect(MqttBroker* local)
{ {
debug("MqttClient::connect_local");
close(); close();
parent = parentBroker; local_broker = local;
} }
void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka) void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
{ {
debug("MqttClient::connect"); debug("MqttClient::connect_to_host " << broker << ':' << port);
keep_alive = ka; keep_alive = ka;
close(); close();
if (client) delete client; if (client) delete client;
client = new TcpClient; client = new TcpClient;
debug("Trying to connect to " << broker.c_str() << ':' << port); #ifdef TINY_MQTT_ASYNC
#ifdef TCP_ASYNC
client->onData(onData, this); client->onData(onData, this);
client->onConnect(onConnect, this); client->onConnect(onConnect, this);
client->connect(broker.c_str(), port); client->connect(broker.c_str(), port, ka);
#else #else
if (client->connect(broker.c_str(), port)) if (client->connect(broker.c_str(), port))
{ {
debug("link established");
onConnect(this, client); onConnect(this, client);
} }
else
{
debug("unable to connect.");
}
#endif #endif
} }
void MqttBroker::addClient(MqttClient* client) void MqttBroker::addClient(MqttClient* client)
{ {
debug("MqttBroker::addClient");
clients.push_back(client); clients.push_back(client);
} }
void MqttBroker::connect(const std::string& host, uint16_t port) void MqttBroker::connect(const std::string& host, uint16_t port)
{ {
debug("MqttBroker::connect");
if (broker == nullptr) broker = new MqttClient; if (broker == nullptr) broker = new MqttClient;
broker->connect(host, port); broker->connect(host, port);
broker->parent = this; // Because connect removed the link broker->local_broker = this; // Because connect removed the link
} }
void MqttBroker::removeClient(MqttClient* remove) void MqttBroker::removeClient(MqttClient* remove)
{ {
debug("removeClient");
for(auto it=clients.begin(); it!=clients.end(); it++) for(auto it=clients.begin(); it!=clients.end(); it++)
{ {
auto client=*it; auto client=*it;
@@ -133,11 +153,12 @@ void MqttBroker::removeClient(MqttClient* remove)
return; return;
} }
} }
debug("Error cannot remove client"); // TODO should not occur debug(red << "Error cannot remove client"); // TODO should not occur
} }
void MqttBroker::onClient(void* broker_ptr, TcpClient* client) void MqttBroker::onClient(void* broker_ptr, TcpClient* client)
{ {
debug("MqttBroker::onClient");
MqttBroker* broker = static_cast<MqttBroker*>(broker_ptr); MqttBroker* broker = static_cast<MqttBroker*>(broker_ptr);
broker->addClient(new MqttClient(broker, client)); broker->addClient(new MqttClient(broker, client));
@@ -146,7 +167,7 @@ void MqttBroker::onClient(void* broker_ptr, TcpClient* client)
void MqttBroker::loop() void MqttBroker::loop()
{ {
#ifndef TCP_ASYNC #ifndef TINY_MQTT_ASYNC
WiFiClient client = server->available(); WiFiClient client = server->available();
if (client) if (client)
@@ -173,7 +194,7 @@ void MqttBroker::loop()
} }
else else
{ {
debug("Client " << client->id().c_str() << " Disconnected, parent=" << (dbg_ptr)client->parent); debug("Client " << client->id().c_str() << " Disconnected, local_broker=" << (dbg_ptr)client->local_broker);
// Note: deleting a client not added by the broker itself will probably crash later. // Note: deleting a client not added by the broker itself will probably crash later.
delete client; delete client;
break; break;
@@ -183,6 +204,7 @@ void MqttBroker::loop()
MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos) MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
{ {
debug("MqttBroker::subscribe");
if (broker && broker->connected()) if (broker && broker->connected())
{ {
return broker->subscribe(topic, qos); return broker->subscribe(topic, qos);
@@ -194,13 +216,13 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, Mqtt
{ {
MqttError retval = MqttOk; MqttError retval = MqttOk;
debug("publish "); debug("MqttBroker::publish");
int i=0; int i=0;
for(auto client: clients) for(auto client: clients)
{ {
i++; i++;
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << "brk_" << (broker && broker->connected() ? "con" : "dis") << Console << __LINE__ << " broker:" << (broker && broker->connected() ? "linked" : "alone") <<
" srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << i << ", local=" << client->isLocal() << ", con=" << client->connected() << endl; " srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << i << ", local=" << client->isLocal() << ", con=" << client->connected() << endl;
#endif #endif
bool doit = false; bool doit = false;
@@ -220,8 +242,8 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, Mqtt
{ {
doit = true; doit = true;
} }
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << ", doit=" << doit << ' '; Console << ", doit=" << doit << ' ';
#endif #endif
if (doit) retval = client->publishIfSubscribed(topic, msg); if (doit) retval = client->publishIfSubscribed(topic, msg);
@@ -242,16 +264,17 @@ bool MqttBroker::compareString(
void MqttMessage::getString(const char* &buff, uint16_t& len) void MqttMessage::getString(const char* &buff, uint16_t& len)
{ {
len = (buff[0]<<8)|(buff[1]); len = getSize(buff);
buff+=2; buff+=2;
} }
void MqttClient::clientAlive(uint32_t more_seconds) void MqttClient::clientAlive(uint32_t more_seconds)
{ {
debug("MqttClient::clientAlive");
if (keep_alive) if (keep_alive)
{ {
#ifdef EPOXY_DUINO #ifdef EPOXY_DUINO
alive=millis()+500000; alive=millis()+500000+0*more_seconds;
#else #else
alive=millis()+1000*(keep_alive+more_seconds); alive=millis()+1000*(keep_alive+more_seconds);
#endif #endif
@@ -264,11 +287,11 @@ void MqttClient::loop()
{ {
if (alive && (millis() > alive)) if (alive && (millis() > alive))
{ {
if (parent) if (local_broker)
{ {
debug("timeout client"); debug(red << "timeout client");
close(); close();
debug("closed"); debug(red << "closed");
} }
else if (client && client->connected()) else if (client && client->connected())
{ {
@@ -281,7 +304,7 @@ void MqttClient::loop()
// there is no need to send one PingReq per instance. // there is no need to send one PingReq per instance.
} }
} }
#ifndef TCP_ASYNC #ifndef TINY_MQTT_ASYNC
while(client && client->available()>0) while(client && client->available()>0)
{ {
message.incoming(client->read()); message.incoming(client->read());
@@ -297,24 +320,24 @@ void MqttClient::loop()
void MqttClient::onConnect(void *mqttclient_ptr, TcpClient*) void MqttClient::onConnect(void *mqttclient_ptr, TcpClient*)
{ {
MqttClient* mqtt = static_cast<MqttClient*>(mqttclient_ptr); MqttClient* mqtt = static_cast<MqttClient*>(mqttclient_ptr);
debug("cnx: connecting"); debug("MqttClient::onConnect");
MqttMessage msg(MqttMessage::Type::Connect); MqttMessage msg(MqttMessage::Type::Connect);
msg.add("MQTT",4); msg.add("MQTT",4);
msg.add(0x4); // Mqtt protocol version 3.1.1 msg.add(0x4); // Mqtt protocol version 3.1.1
msg.add(0x0); // Connect flags TODO user / name msg.add(0x0); // Connect flags TODO user / name
msg.add(0x00); // keep_alive msg.add((char)(mqtt->keep_alive >> 8)); // keep_alive
msg.add((char)mqtt->keep_alive); msg.add((char)(mqtt->keep_alive & 0xFF));
msg.add(mqtt->clientId); msg.add(mqtt->clientId);
debug("cnx: mqtt connecting"); debug("cnx: mqtt connecting");
msg.sendTo(mqtt); msg.sendTo(mqtt);
msg.reset(); msg.reset();
debug("cnx: mqtt sent " << (dbg_ptr)mqtt->parent); debug("cnx: mqtt sent " << (dbg_ptr)mqtt->local_broker);
mqtt->clientAlive(0); mqtt->clientAlive(0);
} }
#ifdef TCP_ASYNC #ifdef TINY_MQTT_ASYNC
void MqttClient::onData(void* client_ptr, TcpClient*, void* data, size_t len) void MqttClient::onData(void* client_ptr, TcpClient*, void* data, size_t len)
{ {
char* char_ptr = static_cast<char*>(data); char* char_ptr = static_cast<char*>(data);
@@ -354,29 +377,30 @@ void MqttClient::resubscribe()
MqttError MqttClient::subscribe(Topic topic, uint8_t qos) MqttError MqttClient::subscribe(Topic topic, uint8_t qos)
{ {
debug("subsribe(" << topic.c_str() << ")"); debug("MqttClient::subsribe(" << topic.c_str() << ")");
MqttError ret = MqttOk; MqttError ret = MqttOk;
subscriptions.insert(topic); subscriptions.insert(topic);
if (parent==nullptr) // remote broker if (local_broker==nullptr) // remote broker
{ {
return sendTopic(topic, MqttMessage::Type::Subscribe, qos); return sendTopic(topic, MqttMessage::Type::Subscribe, qos);
} }
else else
{ {
return parent->subscribe(topic, qos); return local_broker->subscribe(topic, qos);
} }
return ret; return ret;
} }
MqttError MqttClient::unsubscribe(Topic topic) MqttError MqttClient::unsubscribe(Topic topic)
{ {
debug("MqttClient::unsubscribe");
auto it=subscriptions.find(topic); auto it=subscriptions.find(topic);
if (it != subscriptions.end()) if (it != subscriptions.end())
{ {
subscriptions.erase(it); subscriptions.erase(it);
if (parent==nullptr) // remote broker if (local_broker==nullptr) // remote broker
{ {
return sendTopic(topic, MqttMessage::Type::UnSubscribe, 0); return sendTopic(topic, MqttMessage::Type::UnSubscribe, 0);
} }
@@ -386,6 +410,7 @@ MqttError MqttClient::unsubscribe(Topic topic)
MqttError MqttClient::sendTopic(const Topic& topic, MqttMessage::Type type, uint8_t qos) MqttError MqttClient::sendTopic(const Topic& topic, MqttMessage::Type type, uint8_t qos)
{ {
debug("MqttClient::sendTopic");
MqttMessage msg(type, 2); MqttMessage msg(type, 2);
// TODO manage packet identifier // TODO manage packet identifier
@@ -401,17 +426,8 @@ MqttError MqttClient::sendTopic(const Topic& topic, MqttMessage::Type type, uint
void MqttClient::processMessage(MqttMessage* mesg) void MqttClient::processMessage(MqttMessage* mesg)
{ {
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::Type::PingResp)
{
#ifdef NOT_ESP_CORE
Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << " ESP.getFreeHeap() "<< endl;
#else
Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl;
#endif
// mesg->hexdump("Incoming");
mesg->hexdump("Incoming"); mesg->hexdump("Incoming");
}
#endif #endif
auto header = mesg->getVHeader(); auto header = mesg->getVHeader();
const char* payload; const char* payload;
@@ -432,7 +448,7 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
} }
payload = header+10; payload = header+10;
mqtt_flags = header[7]; mqtt_flags = header[7];
keep_alive = (header[8]<<8)|(header[9]); keep_alive = MqttMessage::getSize(header+8);
if (strncmp("MQTT", header+2,4)) if (strncmp("MQTT", header+2,4))
{ {
debug("bad mqtt header"); debug("bad mqtt header");
@@ -461,18 +477,18 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
if (mqtt_flags & FlagUserName) if (mqtt_flags & FlagUserName)
{ {
mesg->getString(payload, len); mesg->getString(payload, len);
if (!parent->checkUser(payload, len)) break; if (not local_broker->checkUser(payload, len)) break;
payload += len; payload += len;
} }
if (mqtt_flags & FlagPassword) if (mqtt_flags & FlagPassword)
{ {
mesg->getString(payload, len); mesg->getString(payload, len);
if (!parent->checkPassword(payload, len)) break; if (not local_broker->checkPassword(payload, len)) break;
payload += len; payload += len;
} }
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << "Connected client:" << clientId.c_str() << ", keep alive=" << keep_alive << '.' << endl; Console << yellow << "Client " << clientId << " connected : keep alive=" << keep_alive << '.' << white << endl;
#endif #endif
bclose = false; bclose = false;
mqtt_connected=true; mqtt_connected=true;
@@ -507,12 +523,13 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
if (client) if (client)
{ {
uint16_t pingreq = MqttMessage::Type::PingResp; uint16_t pingreq = MqttMessage::Type::PingResp;
debug(cyan << "Ping response to client ");
client->write((const char*)(&pingreq), 2); client->write((const char*)(&pingreq), 2);
bclose = false; bclose = false;
} }
else else
{ {
debug("internal pingreq ?"); debug(red << "internal pingreq ?");
} }
break; break;
@@ -568,8 +585,8 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
break; break;
case MqttMessage::Type::Publish: case MqttMessage::Type::Publish:
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << "publish " << mqtt_connected << '/' << (long) client << endl; Console << "publish " << mqtt_connected << '/' << (long) client << endl;
#endif #endif
if (mqtt_connected or client == nullptr) if (mqtt_connected or client == nullptr)
{ {
@@ -578,28 +595,33 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
mesg->getString(payload, len); mesg->getString(payload, len);
Topic published(payload, len); Topic published(payload, len);
payload += len; payload += len;
// Serial << "Received Publish (" << published.str().c_str() << ") size=" << (int)len #if TINY_MQTT_DEBUG
Console << "Received Publish (" << published.str().c_str() << ") size=" << (int)len << endl;
#endif
// << '(' << std::string(payload, len).c_str() << ')' << " msglen=" << mesg->length() << endl; // << '(' << std::string(payload, len).c_str() << ')' << " msglen=" << mesg->length() << endl;
if (qos) payload+=2; // ignore packet identifier if any if (qos) payload+=2; // ignore packet identifier if any
len=mesg->end()-payload; len=mesg->end()-payload;
// TODO reset DUP // TODO reset DUP
// TODO reset RETAIN // TODO reset RETAIN
if (parent==nullptr or client==nullptr) // internal MqttClient receives publish if (local_broker==nullptr or client==nullptr) // internal MqttClient receives publish
{ {
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << (isSubscribedTo(published) ? "not" : "") << " subscribed.\n"; if (TinyMqtt::debug >= 2)
Serial << "has " << (callback ? "" : "no ") << " callback.\n"; {
Console << (isSubscribedTo(published) ? "not" : "") << " subscribed.\n";
Console << "has " << (callback ? "" : "no ") << " callback.\n";
}
#endif #endif
if (callback and isSubscribedTo(published)) if (callback and isSubscribedTo(published))
{ {
callback(this, published, payload, len); // TODO send the real payload callback(this, published, payload, len); // TODO send the real payload
} }
} }
else if (parent) // from outside to inside else if (local_broker) // from outside to inside
{ {
debug("publishing to parent"); debug("publishing to local_broker");
parent->publish(this, published, *mesg); local_broker->publish(this, published, *mesg);
} }
bclose = false; bclose = false;
} }
@@ -619,27 +641,83 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T
}; };
if (bclose) if (bclose)
{ {
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << "*************** Error msg 0x" << _HEX(mesg->type()); Console << red << "*************** Error msg 0x" << _HEX(mesg->type());
mesg->hexdump("-------ERROR ------"); mesg->hexdump("-------ERROR ------");
dump(); dump();
Serial << endl; Console << white << endl;
#endif #endif
close(); close();
} }
else else
{ {
clientAlive(parent ? 5 : 0); clientAlive(local_broker ? 5 : 0);
} }
} }
bool Topic::matches(const Topic& topic) const bool Topic::matches(const Topic& topic) const
{ {
if (getIndex() == topic.getIndex()) return true; if (getIndex() == topic.getIndex()) return true;
if (str() == topic.str()) return true; const char* p1 = c_str();
const char* p2 = topic.c_str();
if (p1 == p2) return true;
if (*p2 == '$' and *p1 != '$') return false;
while(*p1 and *p2)
{
if (*p1 == '+')
{
++p1;
if (*p1 and *p1!='/') return false;
if (*p1) ++p1;
while(*p2 and *p2++!='/');
}
else if (*p1 == '#')
{
if (*++p1==0) return true;
return false; return false;
}
else if (*p1 == '*')
{
const char c=*(p1+1);
if (c==0) return true;
if (c!='/') return false;
const char*p = p1+2;
while(*p and *p2)
{
if (*p == *p2)
{
if (*p==0) return true;
if (*p=='/')
{
p1=p;
break;
}
}
else
{
while(*p2 and *p2++!='/');
break;
}
++p;
++p2;
}
if (*p==0) return true;
}
else if (*p1 == *p2)
{
++p1;
++p2;
}
else
return false;
}
if (*p1=='/' and p1[1]=='#' and p1[2]==0) return true;
return *p1==0 and *p2==0;
} }
// publish from local client // 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)
{ {
@@ -648,9 +726,9 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa
msg.add(payload, pay_length, false); msg.add(payload, pay_length, false);
msg.complete(); msg.complete();
if (parent) if (local_broker)
{ {
return parent->publish(this, topic, msg); return local_broker->publish(this, topic, msg);
} }
else if (client) else if (client)
return msg.sendTo(this); return msg.sendTo(this);
@@ -663,7 +741,7 @@ MqttError MqttClient::publishIfSubscribed(const Topic& topic, MqttMessage& msg)
{ {
MqttError retval=MqttOk; MqttError retval=MqttOk;
debug("mqttclient publish " << subscriptions.size()); debug("mqttclient publishIfSubscribed " << topic.c_str() << ' ' << subscriptions.size());
if (isSubscribedTo(topic)) if (isSubscribedTo(topic))
{ {
if (client) if (client)
@@ -672,8 +750,8 @@ MqttError MqttClient::publishIfSubscribed(const Topic& topic, MqttMessage& msg)
{ {
processMessage(&msg); processMessage(&msg);
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << "Should call the callback ?\n"; Console << "Should call the callback ?\n";
#endif #endif
// callback(this, topic, nullptr, 0); // TODO Payload // callback(this, topic, nullptr, 0); // TODO Payload
} }
@@ -741,8 +819,8 @@ void MqttMessage::incoming(char in_byte)
break; break;
case Complete: case Complete:
default: default:
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << "Spurious " << _HEX(in_byte) << endl; Console << red << "Spurious " << _HEX(in_byte) << white << endl;
hexdump("spurious"); hexdump("spurious");
#endif #endif
reset(); reset();
@@ -768,12 +846,22 @@ void MqttMessage::add(const char* p, size_t len, bool addLength)
void MqttMessage::encodeLength() void MqttMessage::encodeLength()
{ {
debug("encodingLength");
if (state != Complete) if (state != Complete)
{ {
int length = buffer.size()-3; // 3 = 1 byte for header + 2 bytes for pre-reserved length field. int length = buffer.size()-3; // 3 = 1 byte for header + 2 bytes for pre-reserved length field.
if (length <= 0x7F)
{
buffer.erase(1,1);
buffer[1] = length;
vheader = 2;
}
else
{
buffer[1] = 0x80 | (length & 0x7F); buffer[1] = 0x80 | (length & 0x7F);
buffer[2] = (length >> 7); buffer[2] = (length >> 7);
vheader = 3; vheader = 3;
}
// We could check that buffer[2] < 128 (end of length encoding) // We could check that buffer[2] < 128 (end of length encoding)
state = Complete; state = Complete;
@@ -784,14 +872,14 @@ MqttError MqttMessage::sendTo(MqttClient* client)
{ {
if (buffer.size()) if (buffer.size())
{ {
debug("sending " << buffer.size() << " bytes"); debug(cyan << "sending " << buffer.size() << " bytes to " << client->id());
encodeLength(); encodeLength();
// hexdump("snd"); hexdump("Sending ");
client->write(&buffer[0], buffer.size()); client->write(&buffer[0], buffer.size());
} }
else else
{ {
debug("??? Invalid send"); debug(red << "??? Invalid send");
return MqttInvalidMessage; return MqttInvalidMessage;
} }
return MqttOk; return MqttOk;
@@ -800,7 +888,33 @@ MqttError MqttMessage::sendTo(MqttClient* client)
void MqttMessage::hexdump(const char* prefix) const void MqttMessage::hexdump(const char* prefix) const
{ {
(void)prefix; (void)prefix;
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
if (TinyMqtt::debug<2) return;
static std::map<Type, std::string> tts={
{ Connect, "Connect" },
{ ConnAck, "Connack" },
{ Publish, "Publish" },
{ PubAck, "Puback" },
{ Subscribe, "Subscribe" },
{ SubAck, "Suback" },
{ UnSubscribe, "Unsubscribe" },
{ UnSuback, "Unsuback" },
{ PingReq, "Pingreq" },
{ PingResp, "Pingresp" },
{ Disconnect, "Disconnect" }
};
std::string t("Unknown");
Type typ=static_cast<Type>(buffer[0] & 0xF0);
if (tts.find(typ) != tts.end())
t=tts[typ];
Console.fg(cyan);
#ifdef NOT_ESP_CORE
Console << "---> MESSAGE " << t << ' ' << _HEX(typ) << ' ' << " mem=???" << endl;
#else
Console << "---> MESSAGE " << t << ' ' << _HEX(typ) << ' ' << " mem=" << ESP.getFreeHeap() << endl;
#endif
Console.fg(white);
uint16_t addr=0; uint16_t addr=0;
const int bytes_per_row = 8; const int bytes_per_row = 8;
const char* hex_to_str = " | "; const char* hex_to_str = " | ";
@@ -808,19 +922,19 @@ void MqttMessage::hexdump(const char* prefix) const
const char* half_sep = " - "; const char* half_sep = " - ";
std::string ascii; std::string ascii;
Serial << prefix << " size(" << buffer.size() << "), state=" << state << endl; Console << prefix << " size(" << buffer.size() << "), state=" << state << endl;
for(const char chr: buffer) for(const char chr: buffer)
{ {
if ((addr % bytes_per_row) == 0) if ((addr % bytes_per_row) == 0)
{ {
if (ascii.length()) Serial << hex_to_str << ascii << separator << endl; if (ascii.length()) Console << hex_to_str << ascii << separator << endl;
if (prefix) Serial << prefix << separator; if (prefix) Console << prefix << separator;
ascii.clear(); ascii.clear();
} }
addr++; addr++;
if (chr<16) Serial << '0'; if (chr<16) Console << '0';
Serial << _HEX(chr) << ' '; Console << _HEX(chr) << ' ';
ascii += (chr<32 ? '.' : chr); ascii += (chr<32 ? '.' : chr);
if (ascii.length() == (bytes_per_row/2)) ascii += half_sep; if (ascii.length() == (bytes_per_row/2)) ascii += half_sep;
@@ -829,12 +943,12 @@ void MqttMessage::hexdump(const char* prefix) const
{ {
while(ascii.length() < bytes_per_row+strlen(half_sep)) while(ascii.length() < bytes_per_row+strlen(half_sep))
{ {
Serial << " "; // spaces per hexa byte Console << " "; // spaces per hexa byte
ascii += ' '; ascii += ' ';
} }
Serial << hex_to_str << ascii << separator; Console << hex_to_str << ascii << separator;
} }
Serial << endl; Console << endl;
#endif #endif
} }

View File

@@ -1,17 +1,22 @@
// vim: ts=2 sw=2 expandtab
#pragma once #pragma once
// TODO Should add a AUnit with both TCP_ASYNC and not TCP_ASYNC #ifndef TINY_MQTT_DEBUG
// #define TCP_ASYNC // Uncomment this to use ESPAsyncTCP instead of normal cnx #define TINY_MQTT_DEBUG 0
#endif
// TODO Should add a AUnit with both TINY_MQTT_ASYNC and not TINY_MQTT_ASYNC
// #define TINY_MQTT_ASYNC // Uncomment this to use ESPAsyncTCP instead of normal cnx
#if defined(ESP8266) || defined(EPOXY_DUINO) #if defined(ESP8266) || defined(EPOXY_DUINO)
#ifdef TCP_ASYNC #ifdef TINY_MQTT_ASYNC
#include <ESPAsyncTCP.h> #include <ESPAsyncTCP.h>
#else #else
#include <ESP8266WiFi.h> #include <ESP8266WiFi.h>
#endif #endif
#elif defined(ESP32) #elif defined(ESP32)
#include <WiFi.h> #include <WiFi.h>
#ifdef TCP_ASYNC #ifdef TINY_MQTT_ASYNC
#include <AsyncTCP.h> // https://github.com/me-no-dev/AsyncTCP #include <AsyncTCP.h> // https://github.com/me-no-dev/AsyncTCP
#endif #endif
#endif #endif
@@ -20,21 +25,35 @@
#else #else
#define dbg_ptr uint32_t #define dbg_ptr uint32_t
#endif #endif
#ifdef WIO_TERMINAL
// Uncommon board handling
// If you have a problem with this line, just remove it.
// Note: https://github.com/hsaturn/TinyMqtt/issues/41
#include <rpcWiFi.h>
#endif
#include <vector> #include <vector>
#include <set> #include <set>
#include <string> #include <string>
#include "StringIndexer.h" #include "StringIndexer.h"
#include <MqttStreaming.h>
// #define TINY_MQTT_DEBUG #define TINY_MQTT_DEFAULT_CLIENT_ID "Tiny"
#ifdef TINY_MQTT_DEBUG #include <TinyStreaming.h>
#define debug(what) { Serial << (int)__LINE__ << ' ' << what << endl; delay(100); } #if TINY_MQTT_DEBUG
#include <TinyConsole.h> // https://github.com/hsaturn/TinyConsole
struct TinyMqtt
{
static int debug;
};
#define debug(what) { if (TinyMqtt::debug>=1) Console << (int)__LINE__ << ' ' << what << TinyConsole::white << endl; delay(100); }
#else #else
#define debug(what) {} #define debug(what) {}
#endif #endif
#ifdef TCP_ASYNC #ifdef TINY_MQTT_ASYNC
using TcpClient = AsyncClient; using TcpClient = AsyncClient;
using TcpServer = AsyncServer; using TcpServer = AsyncServer;
#else #else
@@ -81,6 +100,7 @@ class MqttMessage
PingResp = 0xD0, PingResp = 0xD0,
Disconnect = 0xE0 Disconnect = 0xE0
}; };
enum __attribute__((packed)) State enum __attribute__((packed)) State
{ {
FixedHeader=0, FixedHeader=0,
@@ -92,6 +112,11 @@ class MqttMessage
Create=6 Create=6
}; };
static inline uint32_t getSize(const char* buffer)
{
const unsigned char* bun = (const unsigned char*)buffer;
return (*bun << 8) | bun[1]; }
MqttMessage() { reset(); } 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; }
void incoming(char byte); void incoming(char byte);
@@ -140,7 +165,6 @@ class MqttMessage
class MqttBroker; class MqttBroker;
class MqttClient class MqttClient
{ {
using CallBack = void (*)(const MqttClient* source, const Topic& topic, const char* payload, size_t payload_length);
enum __attribute__((packed)) Flags enum __attribute__((packed)) Flags
{ {
FlagUserName = 128, FlagUserName = 128,
@@ -152,26 +176,33 @@ class MqttClient
FlagReserved = 1 FlagReserved = 1
}; };
public: public:
/** Constructor. If broker is not null, this is the adress of a local broker.
using CallBack = void (*)(const MqttClient* source, const Topic& topic, const char* payload, size_t payload_length);
/** Constructor. Broker is the adress of a local broker if not null
If you want to connect elsewhere, leave broker null and use connect() **/ If you want to connect elsewhere, leave broker null and use connect() **/
MqttClient(MqttBroker* broker = nullptr, const std::string& id=""); MqttClient(MqttBroker* broker = nullptr, const std::string& id = TINY_MQTT_DEFAULT_CLIENT_ID);
MqttClient(const std::string& id) : MqttClient(nullptr, id){} MqttClient(const std::string& id) : MqttClient(nullptr, id){}
~MqttClient(); ~MqttClient();
void connect(MqttBroker* parent); void connect(MqttBroker* local_broker);
void connect(std::string broker, uint16_t port, uint16_t keep_alive = 10); void connect(std::string broker, uint16_t port, uint16_t keep_alive = 10);
// TODO it seems that connected returns true in tcp mode even if // TODO it seems that connected returns true in tcp mode even if
// no negociation occured (only if tcp link is established) // no negociation occurred
bool connected() { return bool connected()
(parent!=nullptr and client==nullptr) or {
(client and client->connected()); } return (local_broker!=nullptr and client==nullptr) or (client and client->connected());
}
void write(const char* buf, size_t length) void write(const char* buf, size_t length)
{ if (client) client->write(buf, length); } {
if (client) client->write(buf, length);
}
const std::string& id() const { return clientId; } const std::string& id() const { return clientId; }
void id(std::string& new_id) { clientId = new_id; } void id(const std::string& new_id) { clientId = new_id; }
/** Should be called in main loop() */ /** Should be called in main loop() */
void loop(); void loop();
@@ -179,8 +210,8 @@ class MqttClient
void setCallback(CallBack fun) void setCallback(CallBack fun)
{ {
callback=fun; callback=fun;
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Serial << "Callback set to " << (long)fun << endl; Console << TinyConsole::magenta << "Callback set to " << (long)fun << TinyConsole::white << endl;
if (callback) callback(this, "test/topic", "value", 5); if (callback) callback(this, "test/topic", "value", 5);
#endif #endif
}; };
@@ -203,44 +234,51 @@ class MqttClient
void dump(std::string indent="") void dump(std::string indent="")
{ {
(void)indent; (void)indent;
#ifdef TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
uint32_t ms=millis(); uint32_t ms=millis();
Serial << indent << "+-- " << '\'' << clientId.c_str() << "' " << (connected() ? " ON " : " OFF"); Console << indent << "+-- " << '\'' << clientId.c_str() << "' " << (connected() ? " ON " : " OFF");
Serial << ", alive=" << alive << '/' << ms << ", ka=" << keep_alive << ' '; Console << ", alive=" << alive << '/' << ms << ", ka=" << keep_alive << ' ';
Serial << (client && client->connected() ? "" : "dis") << "connected"; if (client)
{
if (client->connected())
Console << TinyConsole::green << "connected";
else
Console << TinyConsole::red << "disconnected";
Console << TinyConsole::white;
}
if (subscriptions.size()) if (subscriptions.size())
{ {
bool c = false; bool c = false;
Serial << " ["; Console << " [";
for(auto s: subscriptions) for(auto s: subscriptions)
(void)indent;
{ {
if (c) Serial << ", "; if (c) Console << ", ";
Serial << s.str().c_str(); Console << s.str().c_str();
c=true; c=true;
} }
Serial << ']'; Console << ']';
} }
Serial << endl; Console << TinyConsole::erase_to_end << endl;
#endif #endif
} }
#ifdef EPOXY_DUINO #ifdef EPOXY_DUINO
static std::map<MqttMessage::Type, int> counters; // Number of processed messages static std::map<MqttMessage::Type, int> counters; // Number of processed messages
#endif #endif
uint32_t keepAlive() const { return keep_alive; }
private: private:
// event when tcp/ip link established (real or fake) // event when tcp/ip link established (real or fake)
static void onConnect(void * client_ptr, TcpClient*); static void onConnect(void * client_ptr, TcpClient*);
#ifdef TCP_ASYNC #ifdef TINY_MQTT_ASYNC
static void onData(void* client_ptr, TcpClient*, void* data, size_t len); static void onData(void* client_ptr, TcpClient*, void* data, size_t len);
#endif #endif
MqttError sendTopic(const Topic& topic, MqttMessage::Type type, uint8_t qos); MqttError sendTopic(const Topic& topic, MqttMessage::Type type, uint8_t qos);
void resubscribe(); void resubscribe();
friend class MqttBroker; friend class MqttBroker;
MqttClient(MqttBroker* parent, TcpClient* client); MqttClient(MqttBroker* local_broker, TcpClient* client);
// republish a received publish if topic matches any in subscriptions // republish a received publish if topic matches any in subscriptions
MqttError publishIfSubscribed(const Topic& topic, MqttMessage& msg); MqttError publishIfSubscribed(const Topic& topic, MqttMessage& msg);
@@ -249,14 +287,13 @@ class MqttClient
bool mqtt_connected = false; bool mqtt_connected = false;
char mqtt_flags; char mqtt_flags;
uint32_t keep_alive = 60; uint32_t keep_alive = 30;
uint32_t alive; uint32_t alive;
MqttMessage message; MqttMessage message;
// TODO having a pointer on MqttBroker may produce larger binaries // connection to local broker, or link to the parent
// due to unecessary function linked if ever parent is not used // when MqttBroker uses MqttClient for each external connexion
// (this is the case when MqttBroker isn't used except here) MqttBroker* local_broker=nullptr;
MqttBroker* parent=nullptr; // connection to local broker
TcpClient* client=nullptr; // connection to remote broker TcpClient* client=nullptr; // connection to remote broker
std::set<Topic> subscriptions; std::set<Topic> subscriptions;
@@ -291,6 +328,8 @@ class MqttBroker
client->dump(indent); client->dump(indent);
} }
const std::vector<MqttClient*> getClients() const { return clients; }
private: private:
friend class MqttClient; friend class MqttClient;
@@ -306,17 +345,19 @@ class MqttBroker
MqttError subscribe(const Topic& topic, uint8_t qos); MqttError subscribe(const Topic& topic, uint8_t qos);
// For clients that are added not by the broker itself // For clients that are added not by the broker itself (local clients)
void addClient(MqttClient* client); void addClient(MqttClient* client);
void removeClient(MqttClient* client); void removeClient(MqttClient* client);
bool compareString(const char* good, const char* str, uint8_t str_len) const; bool compareString(const char* good, const char* str, uint8_t str_len) const;
std::vector<MqttClient*> clients; std::vector<MqttClient*> clients;
TcpServer* server;
private:
TcpServer* server = nullptr;
const char* auth_user = "guest"; const char* auth_user = "guest";
const char* auth_password = "guest"; const char* auth_password = "guest";
State state = Disconnected;
MqttClient* broker = nullptr; MqttClient* broker = nullptr;
State state = Disconnected;
}; };

View File

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

View File

@@ -5,7 +5,15 @@ tests:
$(MAKE) -C $$(dirname $$i) -j; \ $(MAKE) -C $$(dirname $$i) -j; \
done done
runtests: tests debugtest:
set -e; \
$(MAKE) clean; \
$(MAKE) -C debug-mode -j; \
debug-mode/debug-tests.out
runtests: debugtest
$(MAKE) clean
$(MAKE) tests
set -e; \ set -e; \
for i in *-tests/Makefile; do \ for i in *-tests/Makefile; do \
echo '==== Running:' $$(dirname $$i); \ echo '==== Running:' $$(dirname $$i); \

View File

@@ -0,0 +1,351 @@
// vim: ts=2 sw=2 expandtab
#include <Arduino.h>
#include <AUnit.h>
#include <TinyMqtt.h>
#include <MqttClassBinder.h>
#include <map>
#include <iostream>
#include <iomanip>
#include <sstream>
#include <string>
#include <iostream>
// --------------------- CUT HERE - MQTT MESSAGE ROUTER FILE ----------------------------
class TestReceiver : public MqttClassBinder<TestReceiver>
{
public:
TestReceiver(const char* name) : MqttClassBinder(), name_(name) {}
void onPublish(const MqttClient* /* source */, const Topic& topic, const char* payload, size_t /* length */)
{
Serial << "--> routed message received by " << name_ << ':' << topic.c_str() << " = " << payload << endl;
messages[name_]++;
}
private:
const std::string name_;
public:
static std::map<std::string, int> messages;
};
std::map<std::string, int> TestReceiver::messages;
static int unrouted = 0;
void onUnrouted(const MqttClient*, const Topic& topic, const char*, size_t)
{
Serial << "--> unrouted: " << topic.c_str() << endl;
unrouted++;
}
static std::string topic="sensor/temperature";
/**
* TinyMqtt network unit tests.
*
* No wifi connection unit tests.
* Checks with a local broker. Clients must connect to the local broker
**/
// if ascii_pos = 0, no ascii dump, else ascii dump starts after column ascii_pos
std::string bufferToHexa(const uint8_t* buffer, size_t length, char sep = 0, size_t ascii_pos = 0)
{
std::stringstream out;
std::string ascii;
std::string h("0123456789ABCDEF");
for(size_t i=0; i<length; i++)
{
uint8_t c = buffer[i];
out << h[ c >> 4] << h[ c & 0x0F ];
if (sep) out << sep;
if (ascii_pos)
{
if (c>=32)
ascii += c;
else
ascii +='.';
}
}
std::string ret(out.str());
if (ascii_pos)
{
while(ret.length() < ascii_pos)
ret += ' ';
ret +='[' + ascii + ']';
}
return ret;
}
void dumpMqttMessage(const uint8_t* buffer, size_t length)
{
std::map<int, std::string> pkt =
{ { MqttMessage::Unknown , "Unknown " },
{ MqttMessage::Connect , "Connect " },
{ MqttMessage::ConnAck , "ConnAck " },
{ MqttMessage::Publish , "Publish " },
{ MqttMessage::PubAck , "PubAck " },
{ MqttMessage::Subscribe , "Subscribe " },
{ MqttMessage::SubAck , "SubAck " },
{ MqttMessage::UnSubscribe , "Unsubscribe " },
{ MqttMessage::UnSuback , "UnSubAck " },
{ MqttMessage::PingReq , "PingReq " },
{ MqttMessage::PingResp , "PingResp " },
{ MqttMessage::Disconnect , "Disconnect " } };
std::cout << " | data sent " << std::setw(3) << length << " : ";
auto it = pkt.find(buffer[0] & 0xF0);
if (it == pkt.end())
std::cout << pkt[MqttMessage::Unknown];
else
std::cout << it->second;
std::cout << bufferToHexa(buffer, length, ' ', 60) << std::endl;
}
String toString(const IPAddress& ip)
{
return String(ip[0])+'.'+String(ip[1])+'.'+String(ip[2])+'.'+String(ip[3]);
}
MqttBroker broker(1883);
void reset_and_start_servers(int n, bool early_accept = true)
{
MqttClassBinder<TestReceiver>::reset();
TestReceiver::messages.clear();
unrouted = 0;
ESP8266WiFiClass::resetInstances();
ESP8266WiFiClass::earlyAccept = early_accept;
while(n)
{
ESP8266WiFiClass::selectInstance(n--);
WiFi.mode(WIFI_STA);
WiFi.begin("fake_ssid", "fake_pwd");
}
}
test(classbind_one_client_receives_the_message)
{
reset_and_start_servers(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
IPAddress ip_broker = WiFi.localIP();
// We have a 2nd ESP in order to test through wifi (opposed to local)
ESP8266WiFiClass::selectInstance(2);
MqttClient client;
client.connect(ip_broker.toString().c_str(), 1883);
broker.loop();
assertTrue(client.connected());
TestReceiver receiver("receiver");
MqttClassBinder<TestReceiver>::onPublish(&client, &receiver);
client.subscribe("a/b");
client.publish("a/b", "ab");
for (int i =0; i<10; i++)
{
client.loop();
broker.loop();
}
assertEqual(TestReceiver::messages["receiver"], 1);
assertEqual(unrouted, 0);
}
test(classbind_routes_should_be_empty_when_receiver_goes_out_of_scope)
{
reset_and_start_servers(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
IPAddress ip_broker = WiFi.localIP();
// We have a 2nd ESP in order to test through wifi (opposed to local)
ESP8266WiFiClass::selectInstance(2);
MqttClient client;
client.connect(ip_broker.toString().c_str(), 1883);
broker.loop();
assertTrue(client.connected());
// Make a receiver going out of scope
{
TestReceiver receiver("receiver");
MqttClassBinder<TestReceiver>::onPublish(&client, &receiver);
assertEqual(MqttClassBinder<TestReceiver>::size(), (size_t)1);
}
client.subscribe("a/b");
client.publish("a/b", "ab");
for (int i =0; i<10; i++)
{
client.loop();
broker.loop();
}
assertEqual(TestReceiver::messages["receiver"], 0);
assertEqual(MqttClassBinder<TestReceiver>::size(), (size_t)0);
}
test(classbind_publish_should_be_dispatched_to_many_receivers)
{
reset_and_start_servers(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
IPAddress ip_broker = WiFi.localIP();
// We have a 2nd ESP in order to test through wifi (opposed to local)
ESP8266WiFiClass::selectInstance(2);
MqttClient client;
client.connect(ip_broker.toString().c_str(), 1883);
broker.loop();
assertTrue(client.connected());
TestReceiver receiver_1("receiver_1");
TestReceiver receiver_2("receiver_2");
MqttClassBinder<TestReceiver>::onPublish(&client, &receiver_1);
MqttClassBinder<TestReceiver>::onPublish(&client, &receiver_2);
client.subscribe("a/b");
client.publish("a/b", "ab");
for (int i =0; i<10; i++)
{
client.loop();
broker.loop();
}
assertEqual(TestReceiver::messages["receiver_1"], 1);
assertEqual(TestReceiver::messages["receiver_2"], 1);
}
test(classbind_register_to_many_clients)
{
reset_and_start_servers(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
IPAddress ip_broker = WiFi.localIP();
// We have a 2nd ESP in order to test through wifi (opposed to local)
ESP8266WiFiClass::selectInstance(2);
MqttClient client_1;
client_1.connect(ip_broker.toString().c_str(), 1883);
broker.loop();
MqttClient client_2;
client_2.connect(ip_broker.toString().c_str(), 1883);
broker.loop();
assertTrue(client_1.connected());
assertTrue(client_2.connected());
TestReceiver receiver("receiver");
MqttClassBinder<TestReceiver>::onPublish(&client_1, &receiver);
MqttClassBinder<TestReceiver>::onPublish(&client_2, &receiver);
auto loop = [&client_1, &client_2, &broker]()
{
client_1.loop();
client_2.loop();
broker.loop();
};
client_1.subscribe("a/b");
client_2.subscribe("a/b");
// Ensure subscribptions are passed
for (int i =0; i<5; i++) loop();
client_1.publish("a/b", "from 1");
client_2.publish("a/b", "from 2");
// Ensure publishes are processed
for (int i =0; i<5; i++) loop();
assertEqual(TestReceiver::messages["receiver"], 4);
}
test(classbind_unrouted_fallback)
{
reset_and_start_servers(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
IPAddress ip_broker = WiFi.localIP();
// We have a 2nd ESP in order to test through wifi (opposed to local)
ESP8266WiFiClass::selectInstance(2);
MqttClient client;
client.connect(ip_broker.toString().c_str(), 1883);
broker.loop();
assertTrue(client.connected());
MqttClassBinder<TestReceiver>::onUnpublished(onUnrouted);
{
TestReceiver receiver("receiver");
MqttClassBinder<TestReceiver>::onPublish(&client, &receiver);
}
client.subscribe("a/b");
client.publish("a/b", "from 2");
// Ensure subscribptions are passed
for (int i =0; i<5; i++)
{
client.loop();
broker.loop();
}
assertEqual(TestReceiver::messages["receiver"], 0);
assertEqual(unrouted, 1);
}
test(classbind_should_cleanup_when_MqttClient_dies)
{
reset_and_start_servers(2, true);
TestReceiver receiver("receiver");
{
MqttClient client;
MqttClassBinder<TestReceiver>::onPublish(&client, &receiver);
assertEqual(MqttClassBinder<TestReceiver>::size(), (size_t)1);
}
assertEqual(MqttClassBinder<TestReceiver>::size(), (size_t)1);
}
//----------------------------------------------------------------------------
// setup() and loop()
void setup() {
/* delay(1000);
Serial.begin(115200);
while(!Serial);
*/
Serial.println("=============[ FAKE NETWORK TinyMqtt TESTS ]========================");
WiFi.mode(WIFI_STA);
WiFi.begin("network", "password");
}
void loop() {
aunit::TestRunner::run();
if (Serial.available()) ESP.reset();
}

13
tests/debug-mode/Makefile Normal file
View File

@@ -0,0 +1,13 @@
# See https://github.com/bxparks/EpoxyDuino for documentation about this
# Makefile to compile and run Arduino programs natively on Linux or MacOS.
EXTRA_CXXFLAGS=-g3 -O0 -DTINY_MQTT_DEBUG
# Remove flto flag from EpoxyDuino (too many <optimized out>)
CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics
APP_NAME := debug-tests
ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP TinyConsole
ARDUINO_LIB_DIRS := ../../../EspMock/libraries
EPOXY_CORE := EPOXY_CORE_ESP8266
include ../../../EpoxyDuino/EpoxyDuino.mk

View File

@@ -0,0 +1,15 @@
#include <Arduino.h>
#include <AUnit.h>
#include <TinyMqtt.h>
#include <map>
#include <iostream>
#include <iomanip>
#include <sstream>
#include <string>
// Only compilation check, so do nothing
void setup() {}
void loop() {
aunit::TestRunner::run();
}

View File

@@ -1,8 +1,10 @@
# See https://github.com/bxparks/EpoxyDuino for documentation about this # See https://github.com/bxparks/EpoxyDuino for documentation about this
# Makefile to compile and run Arduino programs natively on Linux or MacOS. # Makefile to compile and run Arduino programs natively on Linux or MacOS.
EXTRA_CXXFLAGS=-g3 -O0
APP_NAME := length-tests APP_NAME := length-tests
ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP TinyConsole
ARDUINO_LIB_DIRS := ../../../EspMock/libraries ARDUINO_LIB_DIRS := ../../../EspMock/libraries
EPOXY_CORE := EPOXY_CORE_ESP8266 EPOXY_CORE := EPOXY_CORE_ESP8266
include ../../../EpoxyDuino/EpoxyDuino.mk include ../../../EpoxyDuino/EpoxyDuino.mk

View File

@@ -1,3 +1,4 @@
// vim: ts=2 sw=2 expandtab
#include <AUnit.h> #include <AUnit.h>
#include <TinyMqtt.h> #include <TinyMqtt.h>
#include <map> #include <map>

View File

@@ -7,7 +7,7 @@ EXTRA_CXXFLAGS=-g3 -O0
CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics
APP_NAME := local-tests APP_NAME := local-tests
ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP TinyConsole
ARDUINO_LIB_DIRS := ../../../EspMock/libraries ARDUINO_LIB_DIRS := ../../../EspMock/libraries
EPOXY_CORE := EPOXY_CORE_ESP8266 EPOXY_CORE := EPOXY_CORE_ESP8266
include ../../../EpoxyDuino/EpoxyDuino.mk include ../../../EpoxyDuino/EpoxyDuino.mk

View File

@@ -7,7 +7,7 @@ EXTRA_CXXFLAGS=-g3 -O0
CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics
APP_NAME := network-tests APP_NAME := network-tests
ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP TinyConsole
ARDUINO_LIB_DIRS := ../../../EspMock/libraries ARDUINO_LIB_DIRS := ../../../EspMock/libraries
EPOXY_CORE := EPOXY_CORE_ESP8266 EPOXY_CORE := EPOXY_CORE_ESP8266
include ../../../EpoxyDuino/EpoxyDuino.mk include ../../../EpoxyDuino/EpoxyDuino.mk

View File

@@ -1,7 +1,13 @@
// vim: ts=2 sw=2 expandtab
#include <Arduino.h> #include <Arduino.h>
#include <AUnit.h> #include <AUnit.h>
#include <TinyMqtt.h> #include <TinyMqtt.h>
#include <map> #include <map>
#include <iostream>
#include <iomanip>
#include <sstream>
#include <string>
#include <iostream>
/** /**
* TinyMqtt network unit tests. * TinyMqtt network unit tests.
@@ -10,7 +16,60 @@
* Checks with a local broker. Clients must connect to the local broker * Checks with a local broker. Clients must connect to the local broker
**/ **/
using namespace std; // if ascii_pos = 0, no ascii dump, else ascii dump starts after column ascii_pos
std::string bufferToHexa(const uint8_t* buffer, size_t length, char sep = 0, size_t ascii_pos = 0)
{
std::stringstream out;
std::string ascii;
std::string h("0123456789ABCDEF");
for(size_t i=0; i<length; i++)
{
uint8_t c = buffer[i];
out << h[ c >> 4] << h[ c & 0x0F ];
if (sep) out << sep;
if (ascii_pos)
{
if (c>=32)
ascii += c;
else
ascii +='.';
}
}
std::string ret(out.str());
if (ascii_pos)
{
while(ret.length() < ascii_pos)
ret += ' ';
ret +='[' + ascii + ']';
}
return ret;
}
void dumpMqttMessage(const uint8_t* buffer, size_t length)
{
std::map<int, std::string> pkt =
{ { MqttMessage::Unknown , "Unknown " },
{ MqttMessage::Connect , "Connect " },
{ MqttMessage::ConnAck , "ConnAck " },
{ MqttMessage::Publish , "Publish " },
{ MqttMessage::PubAck , "PubAck " },
{ MqttMessage::Subscribe , "Subscribe " },
{ MqttMessage::SubAck , "SubAck " },
{ MqttMessage::UnSubscribe , "Unsubscribe " },
{ MqttMessage::UnSuback , "UnSubAck " },
{ MqttMessage::PingReq , "PingReq " },
{ MqttMessage::PingResp , "PingResp " },
{ MqttMessage::Disconnect , "Disconnect " } };
std::cout << " | data sent " << std::setw(3) << length << " : ";
auto it = pkt.find(buffer[0] & 0xF0);
if (it == pkt.end())
std::cout << pkt[MqttMessage::Unknown];
else
std::cout << it->second;
std::cout << bufferToHexa(buffer, length, ' ', 60) << std::endl;
}
String toString(const IPAddress& ip) String toString(const IPAddress& ip)
{ {
@@ -83,6 +142,41 @@ test(suback)
assertEqual(MqttClient::counters[MqttMessage::Type::SubAck], 1); assertEqual(MqttClient::counters[MqttMessage::Type::SubAck], 1);
} }
test(network_client_keep_alive_high)
{
const uint32_t keep_alive=1000;
start_servers(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
IPAddress broker_ip = WiFi.localIP();
ESP8266WiFiClass::selectInstance(2);
MqttClient client;
client.connect(broker_ip.toString().c_str(), 1883, keep_alive);
broker.loop();
assertTrue(broker.clientsCount() == 1);
assertTrue(client.connected());
MqttClient::counters[MqttMessage::Type::SubAck] = 0;
client.subscribe("a/b");
// TODO how to avoid these loops ???
broker.loop();
client.loop();
assertEqual(MqttClient::counters[MqttMessage::Type::SubAck], 1);
uint32_t sz = broker.getClients().size();
assertEqual(sz , (uint32_t)1);
uint32_t ka = broker.getClients()[0]->keepAlive();
assertEqual(ka, keep_alive);
}
test(network_client_to_broker_connexion) test(network_client_to_broker_connexion)
{ {
start_servers(2, true); start_servers(2, true);
@@ -207,8 +301,8 @@ test(network_publish_should_be_dispatched)
publisher.publish("a/c"); publisher.publish("a/c");
assertEqual(published.size(), (size_t)1); // 1 client has received something assertEqual(published.size(), (size_t)1); // 1 client has received something
assertEqual(published[""]["a/b"], 1); assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/b"], 1);
assertEqual(published[""]["a/c"], 2); assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/c"], 2);
} }
test(network_publish_should_be_dispatched_to_clients) test(network_publish_should_be_dispatched_to_clients)
@@ -253,7 +347,7 @@ test(network_unsubscribe)
publisher.publish("a/b"); // Those one, no (unsubscribed) publisher.publish("a/b"); // Those one, no (unsubscribed)
publisher.publish("a/b"); publisher.publish("a/b");
assertEqual(published[""]["a/b"], 1); // Only one publish has been received assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/b"], 1); // Only one publish has been received
} }
test(network_nocallback_when_destroyed) test(network_nocallback_when_destroyed)
@@ -299,7 +393,7 @@ test(network_hudge_payload)
MqttClient subscriber(&broker); MqttClient subscriber(&broker);
subscriber.setCallback(onPublish); subscriber.setCallback(onPublish);
subscriber.subscribe("a/b"); subscriber.subscribe("a/b"); // Note -> this does not send any byte .... (nowhere to send)
MqttClient publisher(&broker); MqttClient publisher(&broker);
publisher.publish("a/b", payload); // This publish is received publisher.publish("a/b", payload); // This publish is received
@@ -310,6 +404,47 @@ test(network_hudge_payload)
assertEqual(strcmp(payload, lastPayload), 0); assertEqual(strcmp(payload, lastPayload), 0);
} }
test(connack)
{
const bool view = false;
NetworkObserver check(
[this](const WiFiClient*, const uint8_t* buffer, size_t length)
{
if (view) dumpMqttMessage(buffer, length);
if (buffer[0] == MqttMessage::ConnAck)
{
std::string hex = bufferToHexa(buffer, length);
assertStringCaseEqual(hex.c_str(), "20020000");
}
}
);
start_servers(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
IPAddress broker_ip = WiFi.localIP();
ESP8266WiFiClass::selectInstance(2);
MqttClient client;
client.connect(broker_ip.toString().c_str(), 1883);
broker.loop();
assertTrue(broker.clientsCount() == 1);
assertTrue(client.connected());
MqttClient::counters[MqttMessage::Type::SubAck] = 0;
client.subscribe("a/b");
// TODO how to avoid these loops ???
broker.loop();
client.loop();
assertEqual(MqttClient::counters[MqttMessage::Type::SubAck], 1);
}
//---------------------------------------------------------------------------- //----------------------------------------------------------------------------
// setup() and loop() // setup() and loop()
void setup() { void setup() {

View File

@@ -7,7 +7,7 @@ EXTRA_CXXFLAGS=-g3 -O0
CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics
APP_NAME := nowifi-tests APP_NAME := nowifi-tests
ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP TinyConsole
ARDUINO_LIB_DIRS := ../../../EspMock/libraries ARDUINO_LIB_DIRS := ../../../EspMock/libraries
EPOXY_CORE := EPOXY_CORE_ESP8266 EPOXY_CORE := EPOXY_CORE_ESP8266
include ../../../EpoxyDuino/EpoxyDuino.mk include ../../../EpoxyDuino/EpoxyDuino.mk

View File

@@ -1,3 +1,4 @@
// vim: ts=2 sw=2 expandtab
#include <Arduino.h> #include <Arduino.h>
#include <AUnit.h> #include <AUnit.h>
#include <TinyMqtt.h> #include <TinyMqtt.h>
@@ -64,8 +65,8 @@ test(nowifi_publish_should_be_dispatched)
publisher.publish("a/c"); publisher.publish("a/c");
assertEqual(published.size(), (size_t)1); // 1 client has received something assertEqual(published.size(), (size_t)1); // 1 client has received something
assertEqual(published[""]["a/b"], 1); assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/b"], 1);
assertEqual(published[""]["a/c"], 2); assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/c"], 2);
} }
test(nowifi_publish_should_be_dispatched_to_clients) test(nowifi_publish_should_be_dispatched_to_clients)
@@ -93,6 +94,104 @@ test(nowifi_publish_should_be_dispatched_to_clients)
assertEqual(published["B"]["a/c"], 0); assertEqual(published["B"]["a/c"], 0);
} }
test(nowifi_subscribe_with_star_wildcard)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient subscriber(&broker, "A");
subscriber.setCallback(onPublish);
subscriber.subscribe("house/*/temp");
MqttClient publisher(&broker);
publisher.publish("house/bedroom/temp");
publisher.publish("house/kitchen/temp");
publisher.publish("house/living_room/tv/temp");
publisher.publish("building/location1/bedroom/temp");
assertEqual(published["A"]["house/bedroom/temp"], 1);
assertEqual(published["A"]["house/kitchen/temp"], 1);
assertEqual(published["A"]["house/living_room/tv/temp"], 1);
assertEqual(published["A"]["building/location1/bedroom/temp"], 0);
}
test(nowifi_subscribe_with_plus_wildcard)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient subscriber(&broker, "A");
subscriber.setCallback(onPublish);
subscriber.subscribe("house/+/temp");
MqttClient publisher(&broker);
publisher.publish("house/bedroom/temp");
publisher.publish("house/kitchen/temp");
publisher.publish("house/living_room/tv/temp");
publisher.publish("building/location1/bedroom/temp");
assertEqual(published["A"]["house/bedroom/temp"], 1);
assertEqual(published["A"]["house/kitchen/temp"], 1);
assertEqual(published["A"]["house/living_room/tv/temp"], 0);
assertEqual(published["A"]["building/location1/bedroom/temp"], 0);
}
test(nowifi_should_not_receive_sys_msg)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient subscriber(&broker, "A");
subscriber.setCallback(onPublish);
subscriber.subscribe("+/data");
MqttClient publisher(&broker);
publisher.publish("$SYS/data");
assertEqual(published["A"]["$SYS/data"], 0);
}
test(nowifi_subscribe_with_mixed_wildcards)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient subscriber(&broker, "A");
subscriber.setCallback(onPublish);
subscriber.subscribe("+/data/#");
MqttClient publisher(&broker);
publisher.publish("node1/data/update");
publisher.publish("node2/data/delta");
publisher.publish("node3/data");
assertEqual(published["A"]["node1/data/update"], 1);
assertEqual(published["A"]["node2/data/delta"], 1);
assertEqual(published["A"]["node3/data"], 1);
}
test(nowifi_unsubscribe_with_wildcards)
{
published.clear();
assertEqual(broker.clientsCount(), (size_t)0);
MqttClient subscriber(&broker, "A");
subscriber.setCallback(onPublish);
subscriber.subscribe("one/two/+");
subscriber.subscribe("one/two/three");
MqttClient publisher(&broker);
publisher.publish("one/two/three");
publisher.publish("one/two/four");
subscriber.unsubscribe("one/two/+");
publisher.publish("one/two/five");
assertEqual(published["A"]["one/two/three"], 1);
assertEqual(published["A"]["one/two/four"], 1);
assertEqual(published["A"]["one/two/five"], 0);
}
test(nowifi_unsubscribe) test(nowifi_unsubscribe)
{ {
published.clear(); published.clear();
@@ -110,7 +209,7 @@ test(nowifi_unsubscribe)
publisher.publish("a/b"); // Those one, no (unsubscribed) publisher.publish("a/b"); // Those one, no (unsubscribed)
publisher.publish("a/b"); publisher.publish("a/b");
assertEqual(published[""]["a/b"], 1); // Only one publish has been received assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/b"], 1); // Only one publish has been received
} }
test(nowifi_nocallback_when_destroyed) test(nowifi_nocallback_when_destroyed)

View File

@@ -1,8 +1,10 @@
# See https://github.com/bxparks/EpoxyDuino for documentation about this # See https://github.com/bxparks/EpoxyDuino for documentation about this
# Makefile to compile and run Arduino programs natively on Linux or MacOS. # Makefile to compile and run Arduino programs natively on Linux or MacOS.
EXTRA_CXXFLAGS=-g3 -O0
APP_NAME := string-indexer-tests APP_NAME := string-indexer-tests
ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsync ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsync TinyConsole
ARDUINO_LIB_DIRS := ../../../EspMock/libraries ARDUINO_LIB_DIRS := ../../../EspMock/libraries
EPOXY_CORE := EPOXY_CORE_ESP8266 EPOXY_CORE := EPOXY_CORE_ESP8266
include ../../../EpoxyDuino/EpoxyDuino.mk include ../../../EpoxyDuino/EpoxyDuino.mk

View File

@@ -1,3 +1,4 @@
// vim: ts=2 sw=2 expandtab
#include <Arduino.h> #include <Arduino.h>
#include <AUnit.h> #include <AUnit.h>
#include <StringIndexer.h> #include <StringIndexer.h>
@@ -62,6 +63,14 @@ test(indexer_same_strings_should_equal)
assertTrue(one == two); assertTrue(one == two);
} }
test(indexer_compare_strings_with_same_beginning)
{
IndexedString two("one_two");
IndexedString one("one");
assertNotEqual(one.getIndex(), two.getIndex());
}
test(indexer_indexed_operator_eq) test(indexer_indexed_operator_eq)
{ {
IndexedString one("one"); IndexedString one("one");

View File

@@ -0,0 +1,10 @@
# See https://github.com/bxparks/EpoxyDuino for documentation about this
# Makefile to compile and run Arduino programs natively on Linux or MacOS.
EXTRA_CXXFLAGS=-g3 -O0
APP_NAME := topic-tests
ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsync TinyConsole
ARDUINO_LIB_DIRS := ../../../EspMock/libraries
EPOXY_CORE := EPOXY_CORE_ESP8266
include ../../../EpoxyDuino/EpoxyDuino.mk

View File

@@ -0,0 +1,86 @@
// vim: ts=2 sw=2 expandtab
#include <Arduino.h>
#include <AUnit.h>
#include <TinyMqtt.h>
#include <map>
#include <iostream>
#define endl "\n"
/**
* TinyMqtt / StringIndexer unit tests.
*
**/
using namespace std;
bool testTopicMatch(const char* a, const char* b, bool expected)
{
Topic ta(a);
Topic tb(b);
bool match(ta.matches(tb));
cout << " " << ta.c_str() << ' ';
if (match != expected)
cout << (expected ? " should match " : " should not match ");
else
cout << (expected ? " matches " : " unmatches ");
cout << tb.c_str() << endl;
return expected == match;
}
test(topic_matches)
{
// matching
assertTrue(testTopicMatch("a/b/c" , "a/b/c" , true));
assertTrue(testTopicMatch("a/*/c" , "a/xyz/c" , true));
assertTrue(testTopicMatch("a/*/e" , "a/b/c/d/e" , true));
assertTrue(testTopicMatch("a/*" , "a/b/c/d/e" , true));
assertTrue(testTopicMatch("*/c" , "a/b/c" , true));
assertTrue(testTopicMatch("/*/c" , "/a/b/c" , true));
assertTrue(testTopicMatch("a/*" , "a/b/c/d" , true));
assertTrue(testTopicMatch("a/+/c" , "a/b/c" , true));
assertTrue(testTopicMatch("a/+/c/+/e", "a/b/c/d/e" , true));
assertTrue(testTopicMatch("a/*/c/+/e", "a/b/c/d/e" , true));
assertTrue(testTopicMatch("/+/b" , "/a/b" , true));
assertTrue(testTopicMatch("+" , "a" , true));
assertTrue(testTopicMatch("a/b/#" , "a/b/c/d" , true));
assertTrue(testTopicMatch("a/b/#" , "a/b" , true));
assertTrue(testTopicMatch("a/*/c" , "a/*/c" , true));
// not matching
assertTrue(testTopicMatch("a/b/c" , "a/b/d" , false));
assertTrue(testTopicMatch("a/b/c" , "a/b/d" , false));
assertTrue(testTopicMatch("a/*/e" , "a/b/c/d/f" , false));
assertTrue(testTopicMatch("a/+" , "a" , false));
assertTrue(testTopicMatch("a/+" , "a/b/d" , false));
assertTrue(testTopicMatch("a/+/" , "a/" , false));
// $SYS topics
assertTrue(testTopicMatch("+/any" , "$SYS/any" , false));
assertTrue(testTopicMatch("*/any" , "$SYS/any" , false));
assertTrue(testTopicMatch("$SYS/any" , "$SYS/any" , true));
assertTrue(testTopicMatch("$SYS/+/y" , "$SYS/a/y" , true));
assertTrue(testTopicMatch("$SYS/#" , "$SYS/a/y" , true));
// not valid
assertTrue(testTopicMatch("a/#/b" , "a/x/b" , false));
assertTrue(testTopicMatch("a+" , "a/b/d" , false));
assertTrue(testTopicMatch("a/b/#/d" , "a/b/c/d" , false));
}
//----------------------------------------------------------------------------
// setup() and loop()
void setup() {
delay(1000);
Serial.begin(115200);
while(!Serial);
Serial.println("=============[ TinyMqtt StringIndexer TESTS ]========================");
}
void loop() {
aunit::TestRunner::run();
// if (Serial.available()) ESP.reset();
}