MQTT bridge: fix memory use, improve status reporting

- Main broker: only allocate _mqtt_client when custom broker configured
  (analyzer-only saves one PsychicMqttClient). Reconnect main broker after
  forced disconnect with 30s throttle; set last_attempt on disconnect so
  throttle applies and avoids reconnect storms on flaky WiFi.
- Analyzer clients: call disconnect() when WiFi transitions to disconnected
  so ESP-IDF frees MQTT buffers (it does not free on WiFi drop). Reduces
  fragmentation and Max drop after disconnect/reconnect cycles.
- get wifi.status: report WiFi uptime (Xd Xh Xm Xs) when WITH_MQTT_BRIDGE.
  Track connect time in bridge; backfill when already connected at first check.
- get mqtt.status: show msgs on/off, broker (connected/disconnected/n/a),
  analyzer US/EU (connected/disconnected/off), and queue count.

Note: PsychicMqttClient change (register event only on first client creation)
belongs in the library repo if committed separately.
This commit is contained in:
agessaman committed 2026-02-02 19:09:31 -08:00
1 parent 7fda48d5d6
commit 498566e6c9
22 files changed
+2880 -119

No files matched your search

+1
View File
@@ -0,0 +1 @@
{"type": "library", "name": "PsychicMqttClient", "version": "0.2.4", "spec": {"owner": "elims", "id": 17628, "name": "PsychicMqttClient", "requirements": null, "uri": null}}
+66
View File
@@ -0,0 +1,66 @@
# Changelog
All notable changes to this project will be documented in this file.
## [0.2.4] - Fixes
### Fixed
- Binary payloads truncated by strcpy and switched to memcpy [18](https://github.com/theelims/PsychicMqttClient/pull/18)
- Fixed broken link to Adafruit cert repository
### Changed
- Replaced `mqtt.eclipseprojects.io` with `broker.hivemq.com`
- Replaced debug tag with 🐙
## [0.2.3] - Various Fixes and Certificate Validation
### Fixed
- Issue [16](https://github.com/theelims/PsychicMqttClient/issues/16) was fixed.
### Added
- PR [17](https://github.com/theelims/PsychicMqttClient/pull/17) adds functionality to authenticate via client key + certificate.
## [0.2.2] - Various Fixes
### Fixed
- Issue [14](https://github.com/theelims/PsychicMqttClient/issues/14) was fixed.
- Issue [13](https://github.com/theelims/PsychicMqttClient/issues/13) was fixed.
### Added
- The function `forceStop()` was added to address issue [12](https://github.com/theelims/PsychicMqttClient/issues/12).
## [0.2.1] - Fixing Wildcard Behavior
### Changed
- Issue [6](https://github.com/theelims/PsychicMqttClient/issues/6) was fixed.
### Added
- Example for wildcards in topics.
## [0.2.0] - Compatible with Arduino 3 / ESP-IDF 5
### Added
- CI workflow
- ESP-IDF 5 compatibility
### Changed
- updated library.json
## [0.1.1] - Stability Fixes
### Fixed
- Fixed subscribe error when not connected
- Freezing of event-loop when disconnect() is called while the MQTT server is not connected
## [0.1.0] - Initial Release
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2024 elims
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+198
View File
@@ -0,0 +1,198 @@
# PsychicMqttClient
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Continuous Integration](https://github.com/theelims/PsychicMqttClient/actions/workflows/ci.yml/badge.svg)](https://github.com/theelims/PsychicMqttClient/actions/workflows/ci.yml)
[![PlatformIO Registry](https://badges.registry.platformio.org/packages/elims/library/PsychicMqttClient.svg)](https://registry.platformio.org/libraries/elims/PsychicMqttClient)
Fully featured async MQTT 3.1.1 client for ESP32 with support for SSL/TLS and MQTT over WS. Uses the ESP-IDF MQTT client library under the hood and adds a powerful but easy to use API on top of it. Supports MQTT over TCP, SSL with mbedtls, MQTT over Websocket and MQTT over Websocket Secure.
There are countless popular MQTT client libraries available for Arduino and ESP32. Like [AsyncMqttClient](https://github.com/marvinroger/async-mqtt-client) by Marvin Roger, [pubsubclient](https://github.com/knolleary/pubsubclient) by knolleary and [arduino-mqtt](https://github.com/256dpi/arduino-mqtt) by 256dpi. They are widely used, but all have their unique limitations. Like not supporting all QoS levels, limited message size and none of them has a practical support for SSL/TLS. Also MQTT over websocket is missing in all of them.
The API is very similar to [AsyncMqttClient](https://github.com/marvinroger/async-mqtt-client) for the ESP32 by Marvin Roger, so that this library can be used almost as a drop-in replacement. Only minor adjustments are necessary.
## Features
- Supports MQTT 3.1.1 with QoS 0, QoS 1 and QoS 2
- Compatible with all ESP32 variants (ESP32, ESP32-S2, ESP32-S3, ESP32-C3, ESP32-C6, ...)
- Supports Arduino 2 (ESP-IDF 4) and Arduino 3 (ESP-IDF 5)
- Supports MQTT over TCP and MQTT over websocket
- Full support for SSL/TSL encryption - for both MQTT over TCP and MQTT over WS
- No limitation in buffer size for transmit and receive messages. Multipart messages are reassembled.
- Fully asynchronous and non-blocking
- No dependencies to other libraries
- Flexible and powerful event-based API
- `onTopic()` event which calls a callback every time a message on a specific subscribed topic is received
- Handles reconnects automatically
- Automatically embeds a X509 Root CA Bundle into the binary on platformio
> [!IMPORTANT]
> This library does not compile with the Arduino IDE. It requires the build system of [platformio](https://platformio.org/) or [pioarduino](https://github.com/pioarduino/platform-espressif32).
## Usage
MQTT has never been easier to use. Instantiate the MQTT client, set the server URI, and subscribe to a topic with the `onTopic()` event handler. Easy to use with a lambda function where you can work with the received payload. Connect the MQTT client to the server and you're set. Publish works as expected from any other MQTT client.
```cpp
#include <PsychicMqttClient.h>
PsychicMqttClient mqttClient;
void setup()
{
// Setup Serial, WiFi, ...
mqttClient.setServer("mqtt://mqtt.eclipseprojects.io");
mqttClient.onTopic("your/topic", 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{
Serial.printf("Received Topic: %s\r\n", topic);
Serial.printf("Received Payload: %s\r\n", payload); });
mqttClient.connect();
}
void loop()
{
// Your logic
mqttClient.publish("your/topic", 0, 0, "Hello World!");
}
```
The client will handle all the connection details on its own. It will attempt to reconnect automatically, so you don't need to take care of this.
> [!IMPORTANT]
> No blocking code inside event handler functions. These must return fast. Especially no `delay()`, hardware functions or any other blocking code. Store the results and notify your main loop to take actions. If applicable use FreeRTOS toolings like queues or semaphores to protect against race conditions.
## SSL/TLS Encryption
Using SSL/TLS encryption can be a little bit tedious, but with this MQTT client this is exceptionally easy. You can include a single PEM certificate in the code, or create a bundle of certificates with a platformio script. These are embedded into the binary and offer you universal SSL/TLS support for a wide range of servers without headaches.
### Single CA Root Certificate
If you only need to connect to one specific MQTT server, which will never change, directly embedding the certificate into the source file is the easiest method.
You can get the root certificate directly from the browser. Open the domain of your MQTT server and verify the SSL encryption by pressing on the icon left to the URL. On Chrome press on `Connection is Secure` > `Certificate is Valid` to open the certificate dialog. (Works similar on all other browsers.)
![Chrome Screenshot](docs/resources/Browser_certificate.PNG)
In the Certificate Viewer select the Root Certificate at the top of the tree and press `Export` or any other way to download it. Open the downloaded file and copy the content into your source file. Format the string that it looks like shown below. Add the quotes and `\n` line breaks to keep the block format.
```cpp
const char *eclipse_root_ca = "-----BEGIN CERTIFICATE-----\n"
"MIIFazCCA1OgAwIBAgIRAIIQz7DSQONZRGPgu2OCiwAwDQYJKoZIhvcNAQELBQAw\n"
"TzELMAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2Vh\n"
"cmNoIEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDEwHhcNMTUwNjA0MTEwNDM4\n"
"WhcNMzUwNjA0MTEwNDM4WjBPMQswCQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJu\n"
"ZXQgU2VjdXJpdHkgUmVzZWFyY2ggR3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBY\n"
"MTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAK3oJHP0FDfzm54rVygc\n"
"h77ct984kIxuPOZXoHj3dcKi/vVqbvYATyjb3miGbESTtrFj/RQSa78f0uoxmyF+\n"
"0TM8ukj13Xnfs7j/EvEhmkvBioZxaUpmZmyPfjxwv60pIgbz5MDmgK7iS4+3mX6U\n"
"A5/TR5d8mUgjU+g4rk8Kb4Mu0UlXjIB0ttov0DiNewNwIRt18jA8+o+u3dpjq+sW\n"
"T8KOEUt+zwvo/7V3LvSye0rgTBIlDHCNAymg4VMk7BPZ7hm/ELNKjD+Jo2FR3qyH\n"
"B5T0Y3HsLuJvW5iB4YlcNHlsdu87kGJ55tukmi8mxdAQ4Q7e2RCOFvu396j3x+UC\n"
"B5iPNgiV5+I3lg02dZ77DnKxHZu8A/lJBdiB3QW0KtZB6awBdpUKD9jf1b0SHzUv\n"
"KBds0pjBqAlkd25HN7rOrFleaJ1/ctaJxQZBKT5ZPt0m9STJEadao0xAH0ahmbWn\n"
"OlFuhjuefXKnEgV4We0+UXgVCwOPjdAvBbI+e0ocS3MFEvzG6uBQE3xDk3SzynTn\n"
"jh8BCNAw1FtxNrQHusEwMFxIt4I7mKZ9YIqioymCzLq9gwQbooMDQaHWBfEbwrbw\n"
"qHyGO0aoSCqI3Haadr8faqU9GY/rOPNk3sgrDQoo//fb4hVC1CLQJ13hef4Y53CI\n"
"rU7m2Ys6xt0nUW7/vGT1M0NPAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNV\n"
"HRMBAf8EBTADAQH/MB0GA1UdDgQWBBR5tFnme7bl5AFzgAiIyBpY9umbbjANBgkq\n"
"hkiG9w0BAQsFAAOCAgEAVR9YqbyyqFDQDLHYGmkgJykIrGF1XIpu+ILlaS/V9lZL\n"
"ubhzEFnTIZd+50xx+7LSYK05qAvqFyFWhfFQDlnrzuBZ6brJFe+GnY+EgPbk6ZGQ\n"
"3BebYhtF8GaV0nxvwuo77x/Py9auJ/GpsMiu/X1+mvoiBOv/2X/qkSsisRcOj/KK\n"
"NFtY2PwByVS5uCbMiogziUwthDyC3+6WVwW6LLv3xLfHTjuCvjHIInNzktHCgKQ5\n"
"ORAzI4JMPJ+GslWYHb4phowim57iaztXOoJwTdwJx4nLCgdNbOhdjsnvzqvHu7Ur\n"
"TkXWStAmzOVyyghqpZXjFaH3pO3JLF+l+/+sKAIuvtd7u+Nxe5AW0wdeRlN8NwdC\n"
"jNPElpzVmbUq4JUagEiuTDkHzsxHpFKVK7q4+63SM1N95R1NbdWhscdCb+ZAJzVc\n"
"oyi3B43njTOQ5yOf+1CceWxG1bQVs5ZufpsMljq4Ui0/1lvh+wjChP4kqKOJ2qxq\n"
"4RgqsahDYVvTH9w7jXbyLeiNdd8XM2w9U/t7y0Ff/9yi0GE44Za4rF2LN9d11TPA\n"
"mRGunUHBcnWEvgJBQl9nJEiU0Zsnvgc/ubhPgXRR4Xq37Z0j4r7g1SgEEzwxA57d\n"
"emyPxgcYxn/eR44/KJ4EBs+lVDR3veyJm+kXQ99b21/+jh5Xos1AnX5iItreGCc=\n"
"-----END CERTIFICATE-----\n";
```
When setting the server URI just add the reference to the root CA certificate:
```cpp
mqttClient.setCACert(eclipse_root_ca);
```
That's it. Your MQTT connection is encrypted now.
### X509 CA Root Certificate Bundles
If you require universal connectivity to more then one server with different root certificate authorities you can use the python script in the `/scripts` folder. It will either download a standard set of the most popular root CA's or use a set of certificates in \*.PEM or \*.DEM file format located in the folder `/ssl_certs`. For the download either the Mozilla collection at [https://curl.se/ca/cacert.pem](https://curl.se/ca/cacert.pem) is used or a collection curated by [Adafruit](https://github.com/adafruit/certificates/) specifically adjusted for the constraints of embedded systems.
Copy the script from the library folder into your platformio project folder `./scripts` so that it can be found. In the `platformio.ini` add the following lines
```ini
extra_scripts = pre:scripts/generate_cert_bundle.py
; Source for SSL Cert Store can bei either downloaded from Mozilla with 'mozilla' ('https://curl.se/ca/cacert.pem')
; or from a curated Adafruit repository with 'adafruit' (https://raw.githubusercontent.com/adafruit/certificates/main/data/roots.pem)
; or complied from a 'folder' full of *.pem / *.dem files stored in the ./ssl_certs folder
board_ssl_cert_source = adafruit
board_build.embed_files = src/certs/x509_crt_bundle.bin
```
and configure `board_ssl_cert_source` to your needs. If you use your own collection copy the \*.PEM / \*.DEM certificate files to `./ssl_certs`. The platformio build system will automatically compile the certificates into a binary file and embed them into the final binary. This can be later accessed in your code by
```cpp
extern const uint8_t rootca_crt_bundle_start[] asm("_binary_src_certs_x509_crt_bundle_bin_start");
extern const uint8_t rootca_crt_bundle_end[] asm("_binary_src_certs_x509_crt_bundle_bin_end");
```
To include the bundle for the MQTT client simply call
```cpp
mqttClient.setCACertBundle(rootca_crt_bundle_start, rootca_crt_bundle_end - rootca_crt_bundle_start);
```
when configuring the MQTT client instance before connecting.
#### Together with WiFiClientSecure
If you use WiFiClientSecure in your application, it can be configured to use the X509 certificate bundle as well. In that case you should use WiFiClientSecure to instantiate the certificate bundle before calling `attachArduinoCACertBundle()`.
```cpp
WiFiClientSecure client;
#if ESP_ARDUINO_VERSION_MAJOR == 3
client.setCACertBundle(rootca_crt_bundle_start, rootca_crt_bundle_end - rootca_crt_bundle_start);
#else
client.setCACertBundle(rootca_crt_bundle_start);
#endif
// ...
mqttClient.attachArduinoCACertBundle();
```
Otherwise the bundle will be overwritten by the MQTT client with unwanted side effects.
> [!IMPORTANT]
> Currently there is a bug in mbedtls which prevents the proper certificate validation for all certificates signed by `Let's encrypt`. For this reason working directly with the ISRG Root X1 CA certificate or the bundle downloaded from Mozilla might result in SSL failing. You need to include the DST Root CA X3 certificate as well. Currently this is only done by the Adafruit repository. You can [read](https://github.com/adafruit/certificates/pull/1) here and [here](https://github.com/espressif/arduino-esp32/issues/8626) about the details.
## Advanced Usage
Check the [documentation](/documentation.md) or the commented header file and the `FullyFeatured` example for a complete list of all event handlers and configuration options. You can even get access to the ESP-IDF MQTT Clients' configuration object, should you need parameters not broken out to the API.
## License
MIT License
Copyright (c) 2025 elims
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
Binary file not shown.

After

Width:  |  Height:  |  Size: 38 KiB

@@ -0,0 +1,243 @@
/**
* PsychicMqttClient
*
* Fully Featured Example for a minimal MQTT client using the PsychicMqttClient library.
*
* Please change the ssid and pass to your actual WiFi credentials.
*
* This example uses the public MQTT broker broker.hivemq.com,
* registers a onTopic callback function subscribed to a unique topic
* and publishes a message to that topic. The echoed message payload will be
* printed to the serial monitor.
*
*/
#include <Arduino.h>
#include <WiFi.h>
#include <PsychicMqttClient.h>
const char ssid[] = "ssid"; // your network SSID (name)
const char pass[] = "pass"; // your network password
/**
* Load the root certificate bundle embedded by the PIO build process
*/
extern const uint8_t rootca_crt_bundle_start[] asm("_binary_src_certs_x509_crt_bundle_bin_start");
extern const uint8_t rootca_crt_bundle_end[] asm("_binary_src_certs_x509_crt_bundle_bin_end");
const char loremIpsum[] = "Lorem ipsum dolor sit amet, consetetur sadipscing elitr, "
"sed diam nonumy eirmod tempor invidunt ut labore et dolor"
"e magna aliquyam erat, sed diam voluptua.At vero eos et a"
"ccusam et justo duo dolores et ea rebum.Stet clita kasd g"
"ubergren, no sea takimata sanctus est Lorem ipsum dolor s"
"it amet.Lorem ipsum dolor sit amet, consetetur sadipscing"
" elitr, sed diam nonumy eirmod tempor invidunt ut labore "
"et dolore magna aliquyam erat, sed diam voluptua.At vero "
"eos et accusam et justo duo dolores et ea rebum.Stet clit"
"a kasd gubergren, no sea takimata sanctus est Lorem ipsum"
" dolor sit amet.Lorem ipsum dolor sit amet, consetetur sa"
"dipscing elitr, sed diam nonumy eirmod tempor invidunt ut"
" labore et dolore magna aliquyam erat, sed diam voluptua."
"At vero eos et accusam et justo duo dolores et ea rebum.S"
"tet clita kasd gubergren, no sea takimata sanctus est Lor"
"em ipsum dolor sit amet.Duis autem vel eum iriure dolor i"
"n hendrerit in vulputate velit esse molestie consequat, v"
"el illum dolore eu feugiat nulla facilisis at vero eros e"
"t accumsan et iusto odio dignissim qui blandit praesent l"
"uptatum zzril delenit augue duis dolore te feugait nulla "
"facilisi. Lorem ipsum dolor sit amet, consectetuer adipis"
"cing elit, sed diam nonummy nibh euismod tincidunt ut lao"
"reet dolore magna aliquam erat volutpat.Ut wisi enim ad m"
"inim veniam, quis nostrud exerci tation ullamcorper susci"
"pit lobortis nisl ut aliquip ex ea commodo consequat.Duis"
" autem vel eum iriure dolor in hendrerit in vulputate vel"
"it esse molestie consequat, vel illum dolore eu feugiat n"
"ulla facilisis at vero eros et accumsan et iusto odio dig"
"nissim qui blandit praesent luptatum zzril delenit augue "
"duis dolore te feugait nulla facilisi.Nam liber tempor cu"
"m soluta nobis eleifend option congue nihil imperdiet dom"
"ing id quod mazim placerat facer possim assum.Lorem ipsum"
" dolor sit amet, consectetuer adipiscing elit, sed diam n"
"onummy nibh euismod tincidunt ut laoreet dolore magna ali"
"quam erat volutpat.Ut wisi enim ad minim veniam, quis nos"
"trud exerci tation ullamcorper suscipit lobortis nisl ut "
"aliquip ex ea commodo consequat.Duis autem vel eum iriure"
" dolor in hendrerit in vulputate velit esse molestie cons"
"equat, vel illum dolore eu feugiat nulla facilisis.At ver"
"o eos et accusam et justo duo dolores et ea rebum.Stet cl"
"ita kasd gubergren, no sea takimata sanctus est Lorem ips"
"um dolor sit amet.Lorem ipsum dolor sit amet, consetetur "
"sadipscing elitr, sed diam nonumy eirmod tempor invidunt "
"ut labore et dolore magna aliquyam erat, sed diam voluptu"
"a.(End of Lorem Ipsum)";
/**
* Create a PsychicMqttClient object
*/
PsychicMqttClient mqttClient;
void onMqttConnect(bool sessionPresent)
{
Serial.println("Connected to MQTT.");
Serial.printf("Session present: %d\r\n", sessionPresent);
int packetIdSub = mqttClient.subscribe("test/lol", 2);
Serial.printf("Subscribing at QoS 2, packetId: %d\r\n", packetIdSub);
mqttClient.publish("test/lol", 0, true, "test 1");
Serial.println("Publishing at QoS 0");
int packetIdPub1 = mqttClient.publish("test/lol", 1, true, "test 2");
Serial.printf("Publishing at QoS 1, packetId: %d\r\n", packetIdPub1);
int packetIdPub2 = mqttClient.publish("test/lol", 2, true, "test 3");
Serial.printf("Publishing at QoS 2, packetId: %d\r\n", packetIdPub2);
}
void onMqttDisconnect(bool sessionPresent)
{
Serial.println("Disconnected from MQTT.");
}
void onMqttSubscribe(uint16_t packetId)
{
Serial.println("Subscribe acknowledged.");
Serial.printf(" packetId: %d\r\n", packetId);
}
void onMqttUnsubscribe(uint16_t packetId)
{
Serial.println("Unsubscribe acknowledged.");
Serial.printf(" packetId: %d\r\n", packetId);
}
void onMqttMessage(char *topic, char *payload, int retain, int qos, bool dup)
{
Serial.println("Message received.");
Serial.printf(" topic: %s\r\n", topic);
Serial.printf(" qos: %d\r\n", qos);
Serial.printf(" dup: %d\r\n", dup);
Serial.printf(" retain: %d\r\n", retain);
}
void onMqttPublish(uint16_t packetId)
{
Serial.println("Publish acknowledged.");
Serial.printf(" packetId: %d\r\n", packetId);
}
void onLongTopic(const char *topic, const char *payload, int retain, int qos, bool dup)
{
Serial.printf("Received Topic: %s\r\n", topic);
Serial.printf("Received Payload: %s\r\n", payload);
}
void setup()
{
Serial.begin(115200);
/**
* Connect to the WiFi network with the given ssid and pass.
*/
WiFi.begin(ssid, pass);
Serial.printf("Connecting to WiFi %s .", ssid);
while (WiFi.status() != WL_CONNECTED)
{
Serial.print(".");
delay(500);
}
Serial.printf("\r\nConnected, IP address: %s \r\n", WiFi.localIP().toString().c_str());
/**
* Connect to the open MQTT broker broker.hivemq.com with SSL/TLS enryption.
*/
mqttClient.setServer("mqtts://broker.hivemq.com");
#if ESP_ARDUINO_VERSION_MAJOR == 3
mqttClient.setCACertBundle(rootca_crt_bundle_start, rootca_crt_bundle_end - rootca_crt_bundle_start);
#else
mqttClient.setCACertBundle(rootca_crt_bundle_start);
#endif
mqttClient.setBufferSize(1024);
/**
* Lambda callback function for onTopic Event Handler
*
* Subscribes to the topic "{MAC-Address}/simple" with QoS 0.
* The lambda callback function will be called when a message is received.
*/
String simpleTopic = String(ESP.getEfuseMac()) + "/simple";
String longTopic = String(ESP.getEfuseMac()) + "/long";
/**
* Register the callback functions for the MQTT client
*/
mqttClient.onConnect(onMqttConnect);
mqttClient.onDisconnect(onMqttDisconnect);
mqttClient.onSubscribe(onMqttSubscribe);
mqttClient.onUnsubscribe(onMqttUnsubscribe);
mqttClient.onMessage(onMqttMessage);
mqttClient.onPublish(onMqttPublish);
/**
* Using a lambda callback function is a very convenient way to handle the
* received message. The function will be called when a message is received.
*
* The parameters are:
* - topic: The topic of the received message
* - payload: The payload of the received message
* - retain: The retain flag of the received message
* - qos: The qos of the received message
* - dup: The duplicate flag of the received message
*
* It is important not to do any heavy calculations, hardware access, delays or
* blocking code in the callback function.
*/
mqttClient.onTopic(simpleTopic.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{
Serial.printf("Received Topic: %s\r\n", topic);
Serial.printf("Received Payload: %s\r\n", payload); });
/**
* Using a dedicated callback function is a good way to handle the received message.
*/
mqttClient.onTopic(longTopic.c_str(), 0, onLongTopic);
/**
* Connect to the MQTT broker
*/
mqttClient.connect();
/**
* Wait blocking until the connection is established
*/
while (!mqttClient.connected())
{
delay(500);
}
/**
* Publish a message to the topic "{MAC-Address}/simple" with QoS 0 and retain flag 0.
*
* You can only publish messages after the connection is established.
*/
mqttClient.publish(simpleTopic.c_str(), 0, 0, "Hello World!");
delay(5000);
Serial.println("Unsubscribing from topic test/lol");
mqttClient.unsubscribe("test/lol");
delay(5000);
Serial.println("Publishing lorem ipsum...");
mqttClient.publish(longTopic.c_str(), 0, 0, loremIpsum);
}
void loop()
{
/**
* Nothing to do here, the onTopic callback function will be called when a message is received.
*/
}
@@ -0,0 +1,115 @@
/**
* PsychicMqttClient
*
* Simple Example for a minimal MQTT client using the PsychicMqttClient library.
*
* Please change the ssid and pass to your actual WiFi credentials.
*
* This example uses the public MQTT broker broker.hivemq.com,
* registers a onTopic callback function subscribed to a unique topic
* and publishes a message to that topic. The echoed message payload will be
* printed to the serial monitor.
*
*/
#include <Arduino.h>
#include <WiFi.h>
#include <PsychicMqttClient.h>
const char ssid[] = "ssid"; // your network SSID (name)
const char pass[] = "pass"; // your network password
/**
* Load the root certificate bundle embedded by the PIO build process
*/
extern const uint8_t rootca_crt_bundle_start[] asm("_binary_src_certs_x509_crt_bundle_bin_start");
extern const uint8_t rootca_crt_bundle_end[] asm("_binary_src_certs_x509_crt_bundle_bin_end");
/**
* Create a PsychicMqttClient object
*/
PsychicMqttClient mqttClient;
void setup()
{
Serial.begin(115200);
/**
* Connect to the WiFi network with the given ssid and pass.
*/
WiFi.begin(ssid, pass);
Serial.printf("Connecting to WiFi %s .", ssid);
while (WiFi.status() != WL_CONNECTED)
{
Serial.print(".");
delay(500);
}
Serial.printf("\r\nConnected, IP address: %s \r\n", WiFi.localIP().toString().c_str());
/**
* Connect to the open MQTT broker broker.hivemq.com with SSL/TLS enryption.
*/
mqttClient.setServer("mqtts://broker.hivemq.com");
#if ESP_ARDUINO_VERSION_MAJOR == 3
mqttClient.setCACertBundle(rootca_crt_bundle_start, rootca_crt_bundle_end - rootca_crt_bundle_start);
#else
mqttClient.setCACertBundle(rootca_crt_bundle_start);
#endif
/**
* Lambda callback function for onTopic Event Handler
*
* Subscribes to the topic "{MAC-Address}/simple" with QoS 0.
* The lambda callback function will be called when a message is received.
*/
String topic = String(ESP.getEfuseMac()) + "/simple";
mqttClient.onTopic(topic.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{
/**
* Using a lambda callback function is a very convenient way to handle the
* received message. The function will be called when a message is received.
*
* The parameters are:
* - topic: The topic of the received message
* - payload: The payload of the received message
* - retain: The retain flag of the received message
* - qos: The qos of the received message
* - dup: The duplicate flag of the received message
*
* It is important not to do any heavy calculations, hardware access, delays or
* blocking code in the callback function.
*/
Serial.printf("Received Topic: %s\r\n", topic);
Serial.printf("Received Payload: %s\r\n", payload); });
/**
* Connect to the MQTT broker
*/
mqttClient.connect();
/**
* Wait blocking until the connection is established
*/
while (!mqttClient.connected())
{
delay(500);
}
/**
* Publish a message to the topic "{MAC-Address}/simple" with QoS 0 and retain flag 0.
*
* You can only publish messages after the connection is established.
*/
mqttClient.publish(topic.c_str(), 0, 0, "Hello World!");
}
void loop()
{
/**
* Nothing to do here, the onTopic callback function will be called when a message is received.
*/
}
@@ -0,0 +1,164 @@
/**
* PsychicMqttClient
*
* Simple Example for a minimal MQTT client using the PsychicMqttClient library.
*
* Please change the ssid and pass to your actual WiFi credentials.
*
* This example uses the public MQTT broker broker.hivemq.com,
* registers a onTopic callback function subscribed to a unique topic
* and publishes a message to that topic. The echoed message payload will be
* printed to the serial monitor.
*
*/
#include <Arduino.h>
#include <WiFi.h>
#include <PsychicMqttClient.h>
#include <WiFiClientSecure.h>
const char ssid[] = "ssid"; // your network SSID (name)
const char pass[] = "pass"; // your network password
/**
* Load the root certificate bundle embedded by the build process
*/
extern const uint8_t rootca_crt_bundle_start[] asm("_binary_src_certs_x509_crt_bundle_bin_start");
extern const uint8_t rootca_crt_bundle_end[] asm("_binary_src_certs_x509_crt_bundle_bin_end");
/**
* Create a PsychicMqttClient object
*/
PsychicMqttClient mqttClient;
/**
* Create a WiFiClientSecure object
*/
WiFiClientSecure client;
const char *server = "www.howsmyssl.com";
void fetchData()
{
Serial.println("\nStarting connection to server...");
if (!client.connect(server, 443))
Serial.println("Connection failed!");
else
{
Serial.println("Connected to server!");
// Make a HTTP request:
client.println("GET https://www.howsmyssl.com/a/check HTTP/1.0");
client.println("Host: www.howsmyssl.com");
client.println("Connection: close");
client.println();
while (client.connected())
{
String line = client.readStringUntil('\n');
if (line == "\r")
{
Serial.println("headers received");
break;
}
}
// if there are incoming bytes available
// from the server, read them and print them:
while (client.available())
{
char c = client.read();
Serial.write(c);
}
client.stop();
Serial.println("\nStopping connection to server...");
}
}
void setup()
{
Serial.begin(115200);
/**
* Connect to the WiFi network with the given ssid and pass.
*/
WiFi.begin(ssid, pass);
Serial.printf("Connecting to WiFi %s .", ssid);
while (WiFi.status() != WL_CONNECTED)
{
Serial.print(".");
delay(500);
}
Serial.printf("\r\nConnected, IP address: %s \r\n", WiFi.localIP().toString().c_str());
/**
* Attach the root certificate bundle to the WiFiClientSecure object
* first and fetch some data from a server.
*/
#if ESP_ARDUINO_VERSION_MAJOR == 3
client.setCACertBundle(rootca_crt_bundle_start, rootca_crt_bundle_end - rootca_crt_bundle_start);
#else
client.setCACertBundle(rootca_crt_bundle_start);
#endif
fetchData();
/**
* Connect to the open MQTT broker broker.hivemq.com with SSL/TLS enryption.
*/
mqttClient.setServer("mqtts://broker.hivemq.com");
mqttClient.attachArduinoCACertBundle();
/**
* Lambda callback function for onTopic Event Handler
*
* Subscribes to the topic "{MAC-Address}/simple" with QoS 0.
* The lambda callback function will be called when a message is received.
*/
String topic = String(ESP.getEfuseMac()) + "/simple";
mqttClient.onTopic(topic.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{
/**
* Using a lambda callback function is a very convenient way to handle the
* received message. The function will be called when a message is received.
*
* The parameters are:
* - topic: The topic of the received message
* - payload: The payload of the received message
* - retain: The retain flag of the received message
* - qos: The qos of the received message
* - dup: The duplicate flag of the received message
*
* It is important not to do any heavy calculations, hardware access, delays or
* blocking code in the callback function.
*/
Serial.printf("Received Topic: %s\r\n", topic);
Serial.printf("Received Payload: %s\r\n", payload); });
/**
* Connect to the MQTT broker
*/
mqttClient.connect();
/**
* Wait blocking until the connection is established
*/
while (!mqttClient.connected())
{
delay(500);
}
/**
* Publish a message to the topic "{MAC-Address}/simple" with QoS 0 and retain flag 0.
*
* You can only publish messages after the connection is established.
*/
mqttClient.publish(topic.c_str(), 0, 0, "Hello World!");
}
void loop()
{
/**
* Nothing to do here, the onTopic callback function will be called when a message is received.
*/
}
@@ -0,0 +1,139 @@
/**
* PsychicMqttClient
*
* Simple Example for a minimal MQTT client using the PsychicMqttClient library.
*
* Please change the ssid and pass to your actual WiFi credentials.
*
* This example uses the public MQTT broker broker.hivemq.com,
* registers a onTopic callback function subscribed to a unique topic
* and publishes a message to that topic. The echoed message payload will be
* printed to the serial monitor.
*
*/
#include <Arduino.h>
#include <WiFi.h>
#include <PsychicMqttClient.h>
const char ssid[] = "ssid"; // your network SSID (name)
const char pass[] = "pass"; // your network password
/**
* ISRG Root X1 Certificate which is used by the public MQTT broker broker.hivemq.com
*/
const char *eclipse_root_ca = "-----BEGIN CERTIFICATE-----\n"
"MIIFazCCA1OgAwIBAgIRAIIQz7DSQONZRGPgu2OCiwAwDQYJKoZIhvcNAQELBQAw\n"
"TzELMAkGA1UEBhMCVVMxKTAnBgNVBAoTIEludGVybmV0IFNlY3VyaXR5IFJlc2Vh\n"
"cmNoIEdyb3VwMRUwEwYDVQQDEwxJU1JHIFJvb3QgWDEwHhcNMTUwNjA0MTEwNDM4\n"
"WhcNMzUwNjA0MTEwNDM4WjBPMQswCQYDVQQGEwJVUzEpMCcGA1UEChMgSW50ZXJu\n"
"ZXQgU2VjdXJpdHkgUmVzZWFyY2ggR3JvdXAxFTATBgNVBAMTDElTUkcgUm9vdCBY\n"
"MTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAK3oJHP0FDfzm54rVygc\n"
"h77ct984kIxuPOZXoHj3dcKi/vVqbvYATyjb3miGbESTtrFj/RQSa78f0uoxmyF+\n"
"0TM8ukj13Xnfs7j/EvEhmkvBioZxaUpmZmyPfjxwv60pIgbz5MDmgK7iS4+3mX6U\n"
"A5/TR5d8mUgjU+g4rk8Kb4Mu0UlXjIB0ttov0DiNewNwIRt18jA8+o+u3dpjq+sW\n"
"T8KOEUt+zwvo/7V3LvSye0rgTBIlDHCNAymg4VMk7BPZ7hm/ELNKjD+Jo2FR3qyH\n"
"B5T0Y3HsLuJvW5iB4YlcNHlsdu87kGJ55tukmi8mxdAQ4Q7e2RCOFvu396j3x+UC\n"
"B5iPNgiV5+I3lg02dZ77DnKxHZu8A/lJBdiB3QW0KtZB6awBdpUKD9jf1b0SHzUv\n"
"KBds0pjBqAlkd25HN7rOrFleaJ1/ctaJxQZBKT5ZPt0m9STJEadao0xAH0ahmbWn\n"
"OlFuhjuefXKnEgV4We0+UXgVCwOPjdAvBbI+e0ocS3MFEvzG6uBQE3xDk3SzynTn\n"
"jh8BCNAw1FtxNrQHusEwMFxIt4I7mKZ9YIqioymCzLq9gwQbooMDQaHWBfEbwrbw\n"
"qHyGO0aoSCqI3Haadr8faqU9GY/rOPNk3sgrDQoo//fb4hVC1CLQJ13hef4Y53CI\n"
"rU7m2Ys6xt0nUW7/vGT1M0NPAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNV\n"
"HRMBAf8EBTADAQH/MB0GA1UdDgQWBBR5tFnme7bl5AFzgAiIyBpY9umbbjANBgkq\n"
"hkiG9w0BAQsFAAOCAgEAVR9YqbyyqFDQDLHYGmkgJykIrGF1XIpu+ILlaS/V9lZL\n"
"ubhzEFnTIZd+50xx+7LSYK05qAvqFyFWhfFQDlnrzuBZ6brJFe+GnY+EgPbk6ZGQ\n"
"3BebYhtF8GaV0nxvwuo77x/Py9auJ/GpsMiu/X1+mvoiBOv/2X/qkSsisRcOj/KK\n"
"NFtY2PwByVS5uCbMiogziUwthDyC3+6WVwW6LLv3xLfHTjuCvjHIInNzktHCgKQ5\n"
"ORAzI4JMPJ+GslWYHb4phowim57iaztXOoJwTdwJx4nLCgdNbOhdjsnvzqvHu7Ur\n"
"TkXWStAmzOVyyghqpZXjFaH3pO3JLF+l+/+sKAIuvtd7u+Nxe5AW0wdeRlN8NwdC\n"
"jNPElpzVmbUq4JUagEiuTDkHzsxHpFKVK7q4+63SM1N95R1NbdWhscdCb+ZAJzVc\n"
"oyi3B43njTOQ5yOf+1CceWxG1bQVs5ZufpsMljq4Ui0/1lvh+wjChP4kqKOJ2qxq\n"
"4RgqsahDYVvTH9w7jXbyLeiNdd8XM2w9U/t7y0Ff/9yi0GE44Za4rF2LN9d11TPA\n"
"mRGunUHBcnWEvgJBQl9nJEiU0Zsnvgc/ubhPgXRR4Xq37Z0j4r7g1SgEEzwxA57d\n"
"emyPxgcYxn/eR44/KJ4EBs+lVDR3veyJm+kXQ99b21/+jh5Xos1AnX5iItreGCc=\n"
"-----END CERTIFICATE-----\n";
/**
* Create a PsychicMqttClient object
*/
PsychicMqttClient mqttClient;
void setup()
{
Serial.begin(115200);
/**
* Connect to the WiFi network with the given ssid and pass.
*/
WiFi.begin(ssid, pass);
Serial.printf("Connecting to WiFi %s .", ssid);
while (WiFi.status() != WL_CONNECTED)
{
Serial.print(".");
delay(500);
}
Serial.printf("\r\nConnected, IP address: %s \r\n", WiFi.localIP().toString().c_str());
/**
* Connect to the open MQTT broker broker.hivemq.com with SSL/TLS enryption.
*/
mqttClient.setServer("mqtts://broker.hivemq.com");
mqttClient.setCACert(eclipse_root_ca);
/**
* Lambda callback function for onTopic Event Handler
*
* Subscribes to the topic "{MAC-Address}/simple" with QoS 0.
* The lambda callback function will be called when a message is received.
*/
String topic = String(ESP.getEfuseMac()) + "/simple";
mqttClient.onTopic(topic.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{
/**
* Using a lambda callback function is a very convenient way to handle the
* received message. The function will be called when a message is received.
*
* The parameters are:
* - topic: The topic of the received message
* - payload: The payload of the received message
* - retain: The retain flag of the received message
* - qos: The qos of the received message
* - dup: The duplicate flag of the received message
*
* It is important not to do any heavy calculations, hardware access, delays or
* blocking code in the callback function.
*/
Serial.printf("Received Topic: %s\r\n", topic);
Serial.printf("Received Payload: %s\r\n", payload); });
/**
* Connect to the MQTT broker
*/
mqttClient.connect();
/**
* Wait blocking until the connection is established
*/
while (!mqttClient.connected())
{
delay(500);
}
/**
* Publish a message to the topic "{MAC-Address}/simple" with QoS 0 and retain flag 0.
*
* You can only publish messages after the connection is established.
*/
mqttClient.publish(topic.c_str(), 0, 0, "Hello World!");
}
void loop()
{
/**
* Nothing to do here, the onTopic callback function will be called when a message is received.
*/
}
@@ -0,0 +1,103 @@
/**
* PsychicMqttClient
*
* Simple Example for a minimal MQTT client using the PsychicMqttClient library.
*
* Please change the ssid and pass to your actual WiFi credentials.
*
* This example uses the public MQTT broker broker.hivemq.com,
* registers a onTopic callback function subscribed to a unique topic
* and publishes a message to that topic. The echoed message payload will be
* printed to the serial monitor.
*
*/
#include <Arduino.h>
#include <WiFi.h>
#include <PsychicMqttClient.h>
const char ssid[] = "ssid"; // your network SSID (name)
const char pass[] = "pass"; // your network password
/**
* Create a PsychicMqttClient object
*/
PsychicMqttClient mqttClient;
void setup()
{
Serial.begin(115200);
/**
* Connect to the WiFi network with the given ssid and pass.
*/
WiFi.begin(ssid, pass);
Serial.printf("Connecting to WiFi %s .", ssid);
while (WiFi.status() != WL_CONNECTED)
{
Serial.print(".");
delay(500);
}
Serial.printf("\r\nConnected, IP address: %s \r\n", WiFi.localIP().toString().c_str());
/**
* Connect to the open MQTT broker broker.hivemq.com.
*/
mqttClient.setServer("mqtt://broker.hivemq.com");
/**
* Lambda callback function for onTopic Event Handler
*
* Subscribes to the topic "{MAC-Address}/simple" with QoS 0.
* The lambda callback function will be called when a message is received.
*/
String topic = String(ESP.getEfuseMac()) + "/simple";
mqttClient.onTopic(topic.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{
/**
* Using a lambda callback function is a very convenient way to handle the
* received message. The function will be called when a message is received.
*
* The parameters are:
* - topic: The topic of the received message
* - payload: The payload of the received message
* - retain: The retain flag of the received message
* - qos: The qos of the received message
* - dup: The duplicate flag of the received message
*
* It is important not to do any heavy calculations, hardware access, delays or
* blocking code in the callback function.
*/
Serial.printf("Received Topic: %s\r\n", topic);
Serial.printf("Received Payload: %s\r\n", payload); });
/**
* Connect to the MQTT broker
*/
mqttClient.connect();
/**
* Wait blocking until the connection is established
*/
while (!mqttClient.connected())
{
delay(500);
}
/**
* Publish a message to the topic "{MAC-Address}/simple" with QoS 0 and retain flag 0.
*
* You can only publish messages after the connection is established.
*/
mqttClient.publish(topic.c_str(), 0, 0, "Hello World!");
}
void loop()
{
/**
* Nothing to do here, the onTopic callback function will be called when a message is received.
*/
}
@@ -0,0 +1,103 @@
/**
* PsychicMqttClient
*
* Simple Example for a minimal MQTT client using the PsychicMqttClient library.
*
* Please change the ssid and pass to your actual WiFi credentials.
*
* This example uses the public MQTT broker broker.hivemq.com,
* registers a onTopic callback function subscribed to a unique topic
* and publishes a message to that topic. The echoed message payload will be
* printed to the serial monitor.
*
*/
#include <Arduino.h>
#include <WiFi.h>
#include <PsychicMqttClient.h>
const char ssid[] = "ssid"; // your network SSID (name)
const char pass[] = "pass"; // your network password
/**
* Create a PsychicMqttClient object
*/
PsychicMqttClient mqttClient;
void setup()
{
Serial.begin(115200);
/**
* Connect to the WiFi network with the given ssid and pass.
*/
WiFi.begin(ssid, pass);
Serial.printf("Connecting to WiFi %s .", ssid);
while (WiFi.status() != WL_CONNECTED)
{
Serial.print(".");
delay(500);
}
Serial.printf("\r\nConnected, IP address: %s \r\n", WiFi.localIP().toString().c_str());
/**
* Connect to the open MQTT broker broker.hivemq.com over websocket.
*/
mqttClient.setServer("ws://broker.hivemq.com:80/mqtt");
/**
* Lambda callback function for onTopic Event Handler
*
* Subscribes to the topic "{MAC-Address}/simple" with QoS 0.
* The lambda callback function will be called when a message is received.
*/
String topic = String(ESP.getEfuseMac()) + "/simple";
mqttClient.onTopic(topic.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{
/**
* Using a lambda callback function is a very convenient way to handle the
* received message. The function will be called when a message is received.
*
* The parameters are:
* - topic: The topic of the received message
* - payload: The payload of the received message
* - retain: The retain flag of the received message
* - qos: The qos of the received message
* - dup: The duplicate flag of the received message
*
* It is important not to do any heavy calculations, hardware access, delays or
* blocking code in the callback function.
*/
Serial.printf("Received Topic: %s\r\n", topic);
Serial.printf("Received Payload: %s\r\n", payload); });
/**
* Connect to the MQTT broker
*/
mqttClient.connect();
/**
* Wait blocking until the connection is established
*/
while (!mqttClient.connected())
{
delay(500);
}
/**
* Publish a message to the topic "{MAC-Address}/simple" with QoS 0 and retain flag 0.
*
* You can only publish messages after the connection is established.
*/
mqttClient.publish(topic.c_str(), 0, 0, "Hello World!");
}
void loop()
{
/**
* Nothing to do here, the onTopic callback function will be called when a message is received.
*/
}
@@ -0,0 +1,144 @@
/**
* PsychicMqttClient
*
* Simple Example for a minimal MQTT client using the PsychicMqttClient library.
*
* Please change the ssid and pass to your actual WiFi credentials.
*
* This example uses the public MQTT broker broker.hivemq.com,
* registers a onTopic callback function subscribed to a unique topic
* and publishes a message to that topic. The echoed message payload will be
* printed to the serial monitor.
*
*/
#include <Arduino.h>
#include <WiFi.h>
#include <PsychicMqttClient.h>
#define RECEIVE_DELAY 5000
const char ssid[] = "ssid"; // your network SSID (name)
const char pass[] = "pass"; // your network password
PsychicMqttClient mqttClient;
/*
Setup a number of topics with and without wildcards.
*/
const String TestTopicA = String(ESP.getEfuseMac()) + "/test/A/0000/topic";
const String TestTopicAWildcard = String(ESP.getEfuseMac()) + "/test/A/+/topic";
const String TestTopicAReference = String(ESP.getEfuseMac()) + "/test/A/1111/topic";
const String TestTopicALong = String(ESP.getEfuseMac()) + "/test/A/0000/topic/long";
const String TestTopicB = String(ESP.getEfuseMac()) + "/test/B/topic/0000";
const String TestTopicBWildcard = String(ESP.getEfuseMac()) + "/test/B/topic/#";
const String TestTopicBReference = String(ESP.getEfuseMac()) + "/test/B/topic/1111";
const String TestTopicBLong = String(ESP.getEfuseMac()) + "/test/B/topic/0000/long";
const String TestTopicC = String(ESP.getEfuseMac()) + "/test/C/topic/0000";
const String TestTopicCWildcard = String(ESP.getEfuseMac()) + "/test/C/topic/+";
const String TestTopicCReference = String(ESP.getEfuseMac()) + "/test/C/topic/1111";
const String TestTopicCLong = String(ESP.getEfuseMac()) + "/test/C/topic/0000/long";
const String TestAllTopics = String(ESP.getEfuseMac()) + "/test/#";
const String TestTopic = String(ESP.getEfuseMac()) + "/test/topic";
void setup()
{
Serial.begin(115200);
WiFi.begin(ssid, pass);
Serial.printf("Connecting to WiFi %s .", ssid);
while (WiFi.status() != WL_CONNECTED)
{
Serial.print(".");
delay(500);
}
Serial.printf("\r\nConnected, IP address: %s \r\n", WiFi.localIP().toString().c_str());
mqttClient.setServer("mqtt://broker.hivemq.com");
Serial.printf("Subscribing to %s\r\n", TestTopicA.c_str());
mqttClient.onTopic(TestTopicA.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{ Serial.printf("Received %s: %s\r\n", TestTopicA.c_str(), topic); });
Serial.printf("Subscribing to %s\r\n", TestTopicAWildcard.c_str());
mqttClient.onTopic(TestTopicAWildcard.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{ Serial.printf("Received %s: %s\r\n", TestTopicAWildcard.c_str(), topic); });
Serial.printf("Subscribing to %s\r\n", TestTopicB.c_str());
mqttClient.onTopic(TestTopicB.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{ Serial.printf("Received %s: %s\r\n", TestTopicB.c_str(), topic); });
Serial.printf("Subscribing to %s\r\n", TestTopicBWildcard.c_str());
mqttClient.onTopic(TestTopicBWildcard.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{ Serial.printf("Received %s: %s\r\n", TestTopicBWildcard.c_str(), topic); });
Serial.printf("Subscribing to %s\r\n", TestTopicC.c_str());
mqttClient.onTopic(TestTopicC.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{ Serial.printf("Received %s: %s\r\n", TestTopicC.c_str(), topic); });
Serial.printf("Subscribing to %s\r\n", TestTopicCWildcard.c_str());
mqttClient.onTopic(TestTopicCWildcard.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{ Serial.printf("Received %s: %s\r\n", TestTopicCWildcard.c_str(), topic); });
Serial.printf("Subscribing to %s\r\n", TestAllTopics.c_str());
mqttClient.onTopic(TestAllTopics.c_str(), 0, [&](const char *topic, const char *payload, int retain, int qos, bool dup)
{ Serial.printf("Received %s: %s\r\n", TestAllTopics.c_str(), topic); });
mqttClient.connect();
while (!mqttClient.connected())
{
delay(500);
}
delay(RECEIVE_DELAY);
}
void loop()
{
Serial.println("Testing Wildcards:");
Serial.printf("\r\nPublishing to %s\r\n", TestTopicA.c_str());
mqttClient.publish(TestTopicA.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.printf("\r\nPublishing to %s\r\n", TestTopicAReference.c_str());
mqttClient.publish(TestTopicAReference.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.printf("\r\nPublishing to %s\r\n", TestTopicALong.c_str());
mqttClient.publish(TestTopicALong.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.printf("\r\nPublishing to %s\r\n", TestTopicB.c_str());
mqttClient.publish(TestTopicB.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.printf("\r\nPublishing to %s\r\n", TestTopicBReference.c_str());
mqttClient.publish(TestTopicBReference.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.printf("\r\nPublishing to %s\r\n", TestTopicBLong.c_str());
mqttClient.publish(TestTopicBLong.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.printf("\r\nPublishing to %s\r\n", TestTopicC.c_str());
mqttClient.publish(TestTopicC.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.printf("\r\nPublishing to %s\r\n", TestTopicCReference.c_str());
mqttClient.publish(TestTopicCReference.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.printf("\r\nPublishing to %s\r\n", TestTopicCLong.c_str());
mqttClient.publish(TestTopicCLong.c_str(), 0, 0, "test");
delay(RECEIVE_DELAY);
Serial.println("\r\nTest done.");
vTaskDelete(NULL);
}
+32
View File
@@ -0,0 +1,32 @@
{
"name": "PsychicMqttClient",
"version": "0.2.4-meshcore.1",
"description": "Fully featured async MQTT client for ESP32 with support for SSL/TLS and MQTT over WS. Uses the ESP-IDF MQTT client library under the hood and adds a powerful and easy to use API on top of it. (MeshCore: fixed memory leak in connect())",
"keywords": "iot, home, automation, async, mqtt, client, esp32, mqttclient, mqtt-client",
"repository": {
"type": "git",
"url": "https://github.com/theelims/PsychicMqttClient.git"
},
"authors": {
"name": "elims",
"email": "elims@gmx.net",
"maintainer": true
},
"license": "MIT",
"frameworks": "arduino",
"platforms": "espressif32",
"headers": "PsychicMQTTClient.h",
"export": {
"include": [
"docs",
"examples",
"scripts",
"src",
"library.json",
"CHANGELOG.md",
"LICENSE",
"platformio.ini",
"README.md"
]
}
}
+10
View File
@@ -0,0 +1,10 @@
name=PsychicMqttClient
version=0.2.4-meshcore.1
author=elims <elims@gmx.net>
maintainer=elims <elims@gmx.net>
sentence=Fully featured async MQTT client for ESP32 with support for SSL/TLS and MQTT over WS.
paragraph=Uses the ESP-IDF MQTT client library under the hood and adds a powerful and easy to use API on top of it.
category=Communication
url=https://github.com/theelims/PsychicMqttClient
architectures=esp32
license=MIT
+42
View File
@@ -0,0 +1,42 @@
[env]
framework = arduino
build_flags =
-Wall -Wextra
-D CONFIG_ARDUHAL_LOG_COLORS
-D CORE_DEBUG_LEVEL=ARDUHAL_LOG_LEVEL_VERBOSE
lib_deps =
upload_protocol = esptool
monitor_speed = 115200
monitor_filters = esp32_exception_decoder, log2file
extra_scripts = pre:scripts/generate_cert_bundle.py
; Source for SSL Cert Store can bei either downloaded from Mozilla with 'mozilla' ('https://curl.se/ca/cacert.pem')
; or from a curated Adafruit repository with 'adafruit' (https://raw.githubusercontent.com/adafruit/certificates/main/data/roots.pem)
; or complied from a 'folder' full of *.pem / *.dem files stored in the ./ssl_certs folder
board_ssl_cert_source = adafruit
board_build.embed_files = src/certs/x509_crt_bundle.bin
[platformio]
lib_dir = .
; src_dir = examples/FullyFeatured
; src_dir = examples/Simple
; src_dir = examples/Simple_WS
src_dir = examples/SSL_CA_Bundle
; src_dir = examples/SSL_CA_Bundle_WiFiClientSecure
; src_dir = examples/SSL_CA_Cert
; src_dir = examples/Wildcards
[env:arduino-2]
platform = espressif32@6.10.0
board = esp32dev
[env:arduino-3]
platform = espressif32
platform_packages=
platformio/framework-arduinoespressif32 @ https://github.com/espressif/arduino-esp32.git#3.0.4
platformio/framework-arduinoespressif32-libs @ https://github.com/espressif/arduino-esp32/releases/download/3.0.4/esp32-arduino-libs-3.0.4.zip
board = esp32dev
[env:pioarduino-3]
platform = https://github.com/pioarduino/platform-espressif32/releases/download/54.03.21/platform-espressif32.zip
board = esp32-s3-devkitc-1
@@ -0,0 +1,212 @@
#!/usr/bin/env python
#
# modified ESP32 x509 certificate bundle generation utility to run with platformio
#
# Converts PEM and DER certificates to a custom bundle format which stores just the
# subject name and public key to reduce space
#
# The bundle will have the format: number of certificates; crt 1 subject name length; crt 1 public key length;
# crt 1 subject name; crt 1 public key; crt 2...
#
# SPDX-FileCopyrightText: 2018-2022 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from __future__ import with_statement
from pathlib import Path
import os
import struct
import sys
import requests
from io import open
Import("env")
try:
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
except ImportError:
env.Execute("$PYTHONEXE -m pip install cryptography")
ca_bundle_bin_file = 'x509_crt_bundle.bin'
mozilla_cacert_url = 'https://curl.se/ca/cacert.pem'
adafruit_filtered_cacert_url = 'https://raw.githubusercontent.com/adafruit/certificates/main/data/roots-filtered.pem'
adafruit_full_cacert_url = 'https://raw.githubusercontent.com/adafruit/certificates/main/data/roots-full.pem'
certs_dir = Path("./ssl_certs")
binary_dir = Path("./src/certs")
quiet = False
def download_cacert_file(source):
if source == "mozilla":
response = requests.get(mozilla_cacert_url)
elif source == "adafruit":
response = requests.get(adafruit_filtered_cacert_url)
elif source == "adafruit-full":
response = requests.get(adafruit_full_cacert_url)
else:
raise InputError('Invalid certificate source')
if response.status_code == 200:
# Ensure the directory exists, create it if necessary
os.makedirs(certs_dir, exist_ok=True)
# Generate the full path to the output file
output_file = os.path.join(certs_dir, "cacert.pem")
# Write the certificate bundle to the output file with utf-8 encoding
with open(output_file, "w", encoding="utf-8") as f:
f.write(response.text)
status('Certificate bundle downloaded to: %s' % output_file)
else:
status('Failed to fetch the certificate bundle.')
def status(msg):
""" Print status message to stderr """
if not quiet:
critical(msg)
def critical(msg):
""" Print critical message to stderr """
sys.stderr.write('SSL Cert Store: ')
sys.stderr.write(msg)
sys.stderr.write('\n')
class CertificateBundle:
def __init__(self):
self.certificates = []
self.compressed_crts = []
if os.path.isfile(ca_bundle_bin_file):
os.remove(ca_bundle_bin_file)
def add_from_path(self, crts_path):
found = False
for file_path in os.listdir(crts_path):
found |= self.add_from_file(os.path.join(crts_path, file_path))
if found is False:
raise InputError('No valid x509 certificates found in %s' % crts_path)
def add_from_file(self, file_path):
try:
if file_path.endswith('.pem'):
status('Parsing certificates from %s' % file_path)
with open(file_path, 'r', encoding='utf-8') as f:
crt_str = f.read()
self.add_from_pem(crt_str)
return True
elif file_path.endswith('.der'):
status('Parsing certificates from %s' % file_path)
with open(file_path, 'rb') as f:
crt_str = f.read()
self.add_from_der(crt_str)
return True
except ValueError:
critical('Invalid certificate in %s' % file_path)
raise InputError('Invalid certificate')
return False
def add_from_pem(self, crt_str):
""" A single PEM file may have multiple certificates """
crt = ''
count = 0
start = False
for strg in crt_str.splitlines(True):
if strg == '-----BEGIN CERTIFICATE-----\n' and start is False:
crt = ''
start = True
elif strg == '-----END CERTIFICATE-----\n' and start is True:
crt += strg + '\n'
start = False
self.certificates.append(x509.load_pem_x509_certificate(crt.encode(), default_backend()))
count += 1
if start is True:
crt += strg
if count == 0:
raise InputError('No certificate found')
status('Successfully added %d certificates' % count)
def add_from_der(self, crt_str):
self.certificates.append(x509.load_der_x509_certificate(crt_str, default_backend()))
status('Successfully added 1 certificate')
def create_bundle(self):
# Sort certificates in order to do binary search when looking up certificates
self.certificates = sorted(self.certificates, key=lambda cert: cert.subject.public_bytes(default_backend()))
bundle = struct.pack('>H', len(self.certificates))
for crt in self.certificates:
""" Read the public key as DER format """
pub_key = crt.public_key()
pub_key_der = pub_key.public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)
""" Read the subject name as DER format """
sub_name_der = crt.subject.public_bytes(default_backend())
name_len = len(sub_name_der)
key_len = len(pub_key_der)
len_data = struct.pack('>HH', name_len, key_len)
bundle += len_data
bundle += sub_name_der
bundle += pub_key_der
return bundle
class InputError(RuntimeError):
def __init__(self, e):
super(InputError, self).__init__(e)
def main():
bundle = CertificateBundle()
try:
cert_source = env.GetProjectOption("board_ssl_cert_source")
if (cert_source == "mozilla" or cert_source == "adafruit"):
download_cacert_file(cert_source)
bundle.add_from_file(os.path.join(certs_dir, "cacert.pem"))
elif (cert_source == "folder"):
bundle.add_from_path(certs_dir)
except ValueError:
critical('Invalid configuration option: use \'board_ssl_cert_source\' parameter in platformio.ini' )
raise InputError('Invalid certificate')
status('Successfully added %d certificates in total' % len(bundle.certificates))
crt_bundle = bundle.create_bundle()
# Ensure the directory exists, create it if necessary
os.makedirs(binary_dir, exist_ok=True)
output_file = os.path.join(binary_dir, ca_bundle_bin_file)
with open(output_file, 'wb') as f:
f.write(crt_bundle)
status('Successfully created %s' % output_file)
try:
main()
except InputError as e:
print(e)
sys.exit(2)
@@ -0,0 +1,674 @@
#include "PsychicMqttClient.h"
static const char *TAG = "🐙";
static void log_error_if_nonzero(const char *message, int error_code)
{
if (error_code != 0)
{
ESP_LOGE(TAG, "Last error %s: 0x%x", message, error_code);
}
}
PsychicMqttClient::PsychicMqttClient() : _mqtt_cfg()
{
memset(&_mqtt_cfg, 0, sizeof(_mqtt_cfg));
}
PsychicMqttClient::~PsychicMqttClient()
{
disconnect();
esp_mqtt_client_destroy(_client);
// Free memory in _buffer and _topic
if (_buffer != nullptr)
{
free(_buffer);
_buffer = nullptr; // Set to nullptr to avoid dangling pointers
}
if (_topic != nullptr)
{
free(_topic);
_topic = nullptr; // Set to nullptr to avoid dangling pointers
}
// Free memory in _onMessageUserCallbacks
for (auto &callback : _onMessageUserCallbacks)
{
if (callback.topic != nullptr)
{
free(callback.topic); // Free the dynamically allocated topic
callback.topic = nullptr;
}
}
_onMessageUserCallbacks.clear(); // Clear the vector
}
PsychicMqttClient &PsychicMqttClient::setKeepAlive(int keepAlive)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.session.keepalive = keepAlive;
#else
_mqtt_cfg.keepalive = keepAlive;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setAutoReconnect(bool reconnect)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.network.disable_auto_reconnect = !reconnect;
#else
_mqtt_cfg.disable_auto_reconnect = !reconnect;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setClientId(const char *clientId)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.credentials.client_id = clientId;
#else
_mqtt_cfg.client_id = clientId;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setCleanSession(bool cleanSession)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.session.disable_clean_session = !cleanSession;
#else
_mqtt_cfg.disable_clean_session = !cleanSession;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setBufferSize(int bufferSize)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.buffer.size = bufferSize;
#else
_mqtt_cfg.buffer_size = bufferSize;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setTaskStackAndPriority(int stackSize, int priority)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.task.stack_size = stackSize;
_mqtt_cfg.task.priority = priority;
#else
_mqtt_cfg.task_stack = stackSize;
_mqtt_cfg.task_prio = priority;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setCACert(const char *rootCA, size_t rootCALen)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.broker.verification.certificate = rootCA;
_mqtt_cfg.broker.verification.certificate_len = rootCALen;
#else
_mqtt_cfg.cert_pem = rootCA;
_mqtt_cfg.cert_len = rootCALen;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setCACertBundle(const uint8_t *bundle, size_t bundleLen)
{
#if ESP_IDF_VERSION_MAJOR == 5
if (bundle != nullptr)
{
esp_crt_bundle_set(bundle, bundleLen);
_mqtt_cfg.broker.verification.crt_bundle_attach = esp_crt_bundle_attach;
}
else
{
esp_crt_bundle_detach(NULL);
_mqtt_cfg.broker.verification.crt_bundle_attach = NULL;
}
#else
if (bundle != nullptr)
{
arduino_esp_crt_bundle_set(bundle);
_mqtt_cfg.crt_bundle_attach = arduino_esp_crt_bundle_attach;
}
else
{
arduino_esp_crt_bundle_detach(NULL);
_mqtt_cfg.crt_bundle_attach = NULL;
}
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::attachArduinoCACertBundle(bool attach)
{
#if ESP_IDF_VERSION_MAJOR == 5
if (attach)
_mqtt_cfg.broker.verification.crt_bundle_attach = esp_crt_bundle_attach;
else
_mqtt_cfg.broker.verification.crt_bundle_attach = NULL;
#else
if (attach)
_mqtt_cfg.crt_bundle_attach = arduino_esp_crt_bundle_attach;
else
_mqtt_cfg.crt_bundle_attach = NULL;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setCredentials(const char *username, const char *password)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.credentials.username = username;
if (password != nullptr)
_mqtt_cfg.credentials.authentication.password = password;
#else
_mqtt_cfg.username = username;
if (password != nullptr)
_mqtt_cfg.password = password;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setClientCertificate(const char *clientCert, const char *clientKey,
size_t clientCertLen, size_t clientKeyLen)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.credentials.authentication.certificate = clientCert;
_mqtt_cfg.credentials.authentication.key = clientKey;
_mqtt_cfg.credentials.authentication.certificate_len = clientCertLen;
_mqtt_cfg.credentials.authentication.key_len = clientKeyLen;
#else
_mqtt_cfg.client_cert_pem = clientCert;
_mqtt_cfg.client_key_pem = clientKey;
_mqtt_cfg.client_cert_len = clientCertLen;
_mqtt_cfg.client_key_len = clientKeyLen;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setWill(const char *topic, uint8_t qos, bool retain, const char *payload, int length)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.session.last_will.topic = topic;
_mqtt_cfg.session.last_will.qos = qos;
_mqtt_cfg.session.last_will.retain = retain;
_mqtt_cfg.session.last_will.msg_len = length;
_mqtt_cfg.session.last_will.msg = payload;
#else
_mqtt_cfg.lwt_topic = topic;
_mqtt_cfg.lwt_qos = qos;
_mqtt_cfg.lwt_retain = retain;
_mqtt_cfg.lwt_msg_len = length;
_mqtt_cfg.lwt_msg = payload;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::setServer(const char *uri)
{
#if ESP_IDF_VERSION_MAJOR == 5
_mqtt_cfg.broker.address.uri = uri;
#else
_mqtt_cfg.uri = uri;
#endif
return *this;
}
PsychicMqttClient &PsychicMqttClient::onConnect(OnConnectUserCallback callback)
{
_onConnectUserCallbacks.push_back(callback);
return *this;
}
PsychicMqttClient &PsychicMqttClient::onDisconnect(OnDisconnectUserCallback callback)
{
_onDisconnectUserCallbacks.push_back(callback);
return *this;
}
PsychicMqttClient &PsychicMqttClient::onSubscribe(OnSubscribeUserCallback callback)
{
_onSubscribeUserCallbacks.push_back(callback);
return *this;
}
PsychicMqttClient &PsychicMqttClient::onUnsubscribe(OnUnsubscribeUserCallback callback)
{
_onUnsubscribeUserCallbacks.push_back(callback);
return *this;
}
PsychicMqttClient &PsychicMqttClient::onMessage(OnMessageUserCallback callback)
{
OnMessageUserCallback_t subscription = {nullptr, 0, callback};
_onMessageUserCallbacks.push_back(subscription);
return *this;
}
PsychicMqttClient &PsychicMqttClient::onTopic(const char *topic, int qos, OnMessageUserCallback callback)
{
OnMessageUserCallback_t subscription = {strcpy((char *)malloc(strlen(topic) + 1), topic), qos, callback};
_onMessageUserCallbacks.push_back(subscription);
if (_connected)
subscribe(topic, qos);
return *this;
}
PsychicMqttClient &PsychicMqttClient::onPublish(OnPublishUserCallback callback)
{
_onPublishUserCallbacks.push_back(callback);
return *this;
}
PsychicMqttClient &PsychicMqttClient::onError(OnErrorUserCallback callback)
{
_onErrorUserCallbacks.push_back(callback);
return *this;
}
bool PsychicMqttClient::connected()
{
return _connected;
}
void PsychicMqttClient::connect()
{
#if ESP_IDF_VERSION_MAJOR == 5
if (_mqtt_cfg.broker.address.uri == nullptr)
{
ESP_LOGE(TAG, "MQTT URI not set.");
return;
}
#else
if (_mqtt_cfg.uri == nullptr)
{
ESP_LOGE(TAG, "MQTT URI not set.");
return;
}
#endif
if (_client == nullptr)
{
_client = esp_mqtt_client_init(&_mqtt_cfg);
// Register event handler only once when client is first created
// to avoid memory leak from repeated registrations
esp_mqtt_client_register_event(_client, MQTT_EVENT_ANY, _onMqttEventStatic, this);
}
else
{
ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_set_config(_client, &_mqtt_cfg));
}
ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_start(_client));
ESP_LOGI(TAG, "MQTT client started.");
}
void PsychicMqttClient::disconnect()
{
if (_client == nullptr)
{
ESP_LOGW(TAG, "MQTT client not started.");
return;
}
if (_connected)
{
ESP_LOGI(TAG, "Disconnecting MQTT client.");
_stopMqttClient = false;
esp_mqtt_client_disconnect(_client);
// Wait for all disconnect events to be processed
while (!_stopMqttClient)
{
vTaskDelay(10 / portTICK_PERIOD_MS);
}
}
esp_mqtt_client_stop(_client);
ESP_LOGI(TAG, "MQTT client stopped.");
}
void PsychicMqttClient::forceStop()
{
if (_client == nullptr)
{
ESP_LOGW(TAG, "MQTT client not started.");
return;
}
if (_connected)
{
ESP_LOGI(TAG, "Forced stop MQTT client.");
}
ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_stop(_client));
_connected = false;
ESP_LOGI(TAG, "MQTT client forcefully stopped.");
}
int PsychicMqttClient::subscribe(const char *topic, int qos)
{
if (_connected)
{
ESP_LOGI(TAG, "Subscribing to topic %s with QoS %d", topic, qos);
return esp_mqtt_client_subscribe(_client, topic, qos);
}
else
{
ESP_LOGW(TAG, "MQTT client not connected. Dropping subscription to topic %s with QoS %d.", topic, qos);
return -1;
}
}
int PsychicMqttClient::unsubscribe(const char *topic)
{
ESP_LOGI(TAG, "Unsubscribing from topic %s", topic);
return esp_mqtt_client_unsubscribe(_client, topic);
}
int PsychicMqttClient::publish(const char *topic, int qos, bool retain, const char *payload, int length, bool async)
{
// drop message if not connected and QoS is 0
if (!connected() && qos == 0)
{
ESP_LOGW(TAG, "MQTT client not connected. Dropping message with QoS = 0.");
return -1;
}
if (async)
{
ESP_LOGV(TAG, "Enqueuing message to topic %s with QoS %d", topic, qos);
return esp_mqtt_client_enqueue(_client, topic, payload, length, qos, retain, true);
}
else
{
ESP_LOGV(TAG, "Publishing message to topic %s with QoS %d", topic, qos);
return esp_mqtt_client_publish(_client, topic, payload, length, qos, retain);
}
}
const char *PsychicMqttClient::getClientId()
{
#if ESP_IDF_VERSION_MAJOR == 5
return _mqtt_cfg.credentials.client_id;
#else
return _mqtt_cfg.client_id;
#endif
}
esp_mqtt_client_config_t *PsychicMqttClient::getMqttConfig()
{
return &_mqtt_cfg;
}
void PsychicMqttClient::_onMqttEventStatic(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data)
{
// Since this is a static function, we need to cast the first argument (void*) back to the class instance type
PsychicMqttClient *instance = (PsychicMqttClient *)handler_args;
instance->_onMqttEvent(base, event_id, event_data);
}
void PsychicMqttClient::_onMqttEvent(esp_event_base_t base, int32_t event_id, void *event_data)
{
ESP_LOGV(TAG, "Event dispatched from event loop base=%s, event_id=%d", base, event_id);
esp_mqtt_event_handle_t event = (esp_mqtt_event_handle_t)event_data;
switch (event_id)
{
case MQTT_EVENT_CONNECTED:
_connected = true;
_onConnect(event);
break;
case MQTT_EVENT_DISCONNECTED:
_connected = false;
_onDisconnect(event);
break;
case MQTT_EVENT_SUBSCRIBED:
_onSubscribe(event);
break;
case MQTT_EVENT_UNSUBSCRIBED:
_onUnsubscribe(event);
break;
case MQTT_EVENT_PUBLISHED:
_onPublish(event);
break;
case MQTT_EVENT_DATA:
_onMessage(event);
break;
case MQTT_EVENT_ERROR:
_connected = false;
_onError(event);
break;
default:
ESP_LOGI(TAG, "Other event id:%d", event->event_id);
break;
}
}
void PsychicMqttClient::_onConnect(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");
// Resubscribe to all topics
for (auto topic : _onMessageUserCallbacks)
{
if (topic.topic != nullptr)
subscribe(topic.topic, topic.qos);
}
for (auto callback : _onConnectUserCallbacks)
{
callback(event->session_present);
}
}
void PsychicMqttClient::_onDisconnect(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
for (auto callback : _onDisconnectUserCallbacks)
{
callback(event->session_present);
}
_stopMqttClient = true;
}
void PsychicMqttClient::_onSubscribe(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
for (auto callback : _onSubscribeUserCallbacks)
{
callback(event->msg_id);
}
}
void PsychicMqttClient::_onUnsubscribe(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
for (auto callback : _onUnsubscribeUserCallbacks)
{
callback(event->msg_id);
}
}
void PsychicMqttClient::_onMessage(esp_mqtt_event_handle_t &event)
{
// ESP_LOGI(TAG, "MQTT_EVENT_DATA");
// printf("MSG_ID=%d\r\n", event->msg_id);
// printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
// printf("DATA=%.*s\r\n", event->data_len, event->data);
// printf("DATA_LEN=%d\r\n", event->data_len);
// printf("TOTAL_DATA_LEN=%d\r\n", event->total_data_len);
// printf("CURRENT_DATA_OFFSET=%d\r\n", event->current_data_offset);
// Check if we are dealing with a simple message
if (event->total_data_len == event->data_len)
{
ESP_LOGV(TAG, "MQTT_EVENT_DATA_SINGLE");
// Copy the characters from data->data_ptr to c-string
char payload[event->data_len + 1];
memcpy(payload, (char *)event->data, event->data_len);
payload[event->data_len] = '\0';
ESP_LOGV(TAG, "Payload=%s", payload);
char topic[event->topic_len + 1];
memcpy(topic, (char *)event->topic, event->topic_len);
topic[event->topic_len] = '\0';
ESP_LOGV(TAG, "Topic=%s", topic);
for (auto callback : _onMessageUserCallbacks)
{
if (callback.topic == nullptr || _isTopicMatch(topic, callback.topic))
{
callback.callback(topic, payload, event->retain, event->qos, event->dup);
}
}
}
// Check if we are dealing with a first multipart message
else if (event->current_data_offset == 0)
{
ESP_LOGV(TAG, "MQTT_EVENT_DATA_MULTIPART_FIRST");
// Allocate memory for the buffer
_buffer = (char *)malloc(event->total_data_len + 1);
// Copy the characters from even->data to _buffer
memcpy(_buffer, (char *)event->data, event->data_len);
// Store the topic for later use, as it is only sent with the first message
_topic = (char *)malloc(event->topic_len + 1);
memcpy(_topic, (char *)event->topic, event->topic_len);
_topic[event->topic_len] = '\0';
}
// Check if we are on the last message
else if (event->current_data_offset + event->data_len == event->total_data_len)
{
ESP_LOGV(TAG, "MQTT_EVENT_DATA_MULTIPART_LAST");
// Copy the characters from even->data to _buffer
memcpy(_buffer + event->current_data_offset, (char *)event->data, event->data_len);
_buffer[event->total_data_len] = '\0';
ESP_LOGV(TAG, "Topic=%s", _topic);
ESP_LOGV(TAG, "Payload=%s", _buffer);
for (auto callback : _onMessageUserCallbacks)
{
if (callback.topic == nullptr || _isTopicMatch(_topic, callback.topic))
{
callback.callback(_topic, _buffer, event->retain, event->qos, event->dup);
}
}
// Free the memory
free(_buffer);
_buffer = nullptr;
free(_topic);
_topic = nullptr;
}
// Otherwise, we are in the middle of the message
else
{
// copy the characters from even->data to _buffer
memcpy(_buffer + event->current_data_offset, (char *)event->data, event->data_len);
ESP_LOGV(TAG, "MQTT_EVENT_DATA_MULTIPART");
}
}
void PsychicMqttClient::_onPublish(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
for (auto callback : _onPublishUserCallbacks)
{
callback(event->msg_id);
}
}
void PsychicMqttClient::_onError(esp_mqtt_event_handle_t &event)
{
ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
if (event->error_handle->error_type == MQTT_ERROR_TYPE_TCP_TRANSPORT)
{
log_error_if_nonzero("reported from esp-tls", event->error_handle->esp_tls_last_esp_err);
log_error_if_nonzero("reported from tls stack", event->error_handle->esp_tls_stack_err);
log_error_if_nonzero("captured as transport's socket errno", event->error_handle->esp_transport_sock_errno);
ESP_LOGI(TAG, "Last errno string (%s)", strerror(event->error_handle->esp_transport_sock_errno));
for (auto callback : _onErrorUserCallbacks)
{
callback(*event->error_handle);
}
}
}
bool PsychicMqttClient::_isTopicMatch(const char *topic, const char *subscription)
{
ESP_LOGD(TAG, "Match topic: %s with subscription: %s", topic, subscription);
String topicStr(topic);
String subscriptionStr(subscription);
// Check if the subscription is a pure wildcard
if (subscriptionStr == "#" || subscriptionStr == "+")
{
ESP_LOGV(TAG, "Subscription is a pure wildcard --> MATCH");
return true;
}
// Check if the topic is a simple match
if (topicStr == subscriptionStr)
{
ESP_LOGV(TAG, "Topic is a direct match --> MATCH");
return true;
}
// Split the topic and subscription into tokens
int topicIndex = 0;
int subscriptionIndex = 0;
int lastTopicIndex = topicStr.lastIndexOf('/');
int lastSubscriptionIndex = subscriptionStr.lastIndexOf('/');
String topicToken = topicStr.substring(topicIndex, topicStr.indexOf('/', topicIndex));
String subscriptionToken = subscriptionStr.substring(subscriptionIndex, subscriptionStr.indexOf('/', subscriptionIndex));
ESP_LOGV(TAG, "Initial topic token: %s, subscription token: %s", topicToken.c_str(), subscriptionToken.c_str());
ESP_LOGV(TAG, "Last topic index: %d, last subscription index: %d", lastTopicIndex, lastSubscriptionIndex);
while (topicToken.length() > 0 && subscriptionToken.length() > 0)
{
ESP_LOGV(TAG, "Comparing topic token: %s with subscription token: %s", topicToken.c_str(), subscriptionToken.c_str());
if (subscriptionToken == "#")
{
ESP_LOGV(TAG, "Subscription token is # --> MATCH");
return true;
}
if (subscriptionToken != "+" && topicToken != subscriptionToken)
{
ESP_LOGV(TAG, "Tokens do not match and subscription token is not + --> NO MATCH");
return false;
}
ESP_LOGV(TAG, "Current token index: topic: %d, subscription: %d", topicIndex, subscriptionIndex);
if (topicIndex == lastTopicIndex + 1 || subscriptionIndex == lastSubscriptionIndex + 1)
{
ESP_LOGV(TAG, "End of tokens. Topic: %s, Subscription: %s", topicStr.substring(topicIndex).c_str(), subscriptionStr.substring(subscriptionIndex).c_str());
bool match = ((subscriptionToken == "+" && topicIndex == lastTopicIndex + 1) || topicStr.substring(topicIndex) == subscriptionStr.substring(subscriptionIndex));
ESP_LOGV(TAG, "End of tokens. Match: %s", match ? "true" : "false");
return match;
}
topicIndex = topicStr.indexOf('/', topicIndex) + 1;
subscriptionIndex = subscriptionStr.indexOf('/', subscriptionIndex) + 1;
ESP_LOGV(TAG, "Next token index: topic: %d, subscription: %d", topicIndex, subscriptionIndex);
topicToken = topicStr.substring(topicIndex, topicStr.indexOf('/', topicIndex));
subscriptionToken = subscriptionStr.substring(subscriptionIndex, subscriptionStr.indexOf('/', subscriptionIndex));
ESP_LOGV(TAG, "Next topic token: %s, subscription token: %s", topicToken.c_str(), subscriptionToken.c_str());
}
return false;
}
@@ -0,0 +1,394 @@
#pragma once
/**
* PsychicMqttClient
*
* Fully featured async MQTT 3.1.1 client for ESP32 with support for SSL/TLS
* and MQTT over WS. Uses the ESP-IDF MQTT client library under the hood and
* adds a powerful but easy to use API on top of it. Supports MQTT over TCP,
* SSL with mbedtls, MQTT over Websocket and MQTT over Websocket Secure.
* https://github.com/theelims/PsychicMqttClient
*
* MIT License
*
* Copyright (c) 2024 elims
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include <functional>
#include <vector>
#include "Arduino.h"
#include "mqtt_client.h"
#include "esp_crt_bundle.h"
#define PSYCHIC_MQTT_CLIENT_VERSION_STR "0.2.1"
#define PSYCHIC_MQTT_CLIENT_VERSION_MAJOR 0
#define PSYCHIC_MQTT_CLIENT_VERSION_MINOR 2
#define PSYCHIC_MQTT_CLIENT_VERSION_PATCH 1
#ifndef ARDUINO_ARCH_ESP32
#error "This library only supports boards with an ESP32 processor."
#endif
// user callbacks
typedef std::function<void(bool sessionPresent)> OnConnectUserCallback;
typedef std::function<void(bool sessionPresent)> OnDisconnectUserCallback;
typedef std::function<void(int msgId)> OnSubscribeUserCallback;
typedef std::function<void(int msgId)> OnUnsubscribeUserCallback;
typedef std::function<void(char *topic, char *payload, int retain, int qos, bool dup)> OnMessageUserCallback;
typedef std::function<void(int msgId)> OnPublishUserCallback;
typedef std::function<void(esp_mqtt_error_codes_t error)> OnErrorUserCallback;
typedef struct
{
char *topic;
int qos;
OnMessageUserCallback callback;
} OnMessageUserCallback_t;
/**
* @class PsychicMqttClient
* @brief A class that wraps the ESP-IDF MQTT client and provides a more user friendly interface.
* The API is very similar to AsyncMqttClient for the ESP32 by Marvin Roger, so that this library can
* be used as a almost drop-in replacement.
*/
class PsychicMqttClient
{
public:
/**
* @brief Constructs a new instance of the PsychicMqttClient class.
*/
PsychicMqttClient();
/**
* @brief Destroys the instance of the PsychicMqttClient class.
*/
~PsychicMqttClient();
/**
* @brief Sets the keep alive interval in seconds for the MQTT connection.
*
* @param keepAlive The keep alive interval in seconds. Defaults to 120.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setKeepAlive(int keepAlive = 120);
/**
* @brief Sets the auto reconnect flag for the MQTT connection.
*
* @param reconnect The auto reconnect flag. Defaults to true.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setAutoReconnect(bool reconnect = true);
/**
* @brief Sets the client ID for the MQTT connection.
*
* @param clientId The client ID. Defaults to ESP32_%CHIPID% where
* %CHIPID% are the last 3 bytes of MAC address in hex format.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setClientId(const char *clientId);
/**
* @brief Sets the clean session flag for the MQTT connection.
*
* @param cleanSession The clean session flag. Defaults to true.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setCleanSession(bool cleanSession = true);
/**
* @brief Sets the size for the MQTT send/receive buffer. If messages exceed
* the buffer size, the message will be split into multiple chunks. Received m
* essages will be assembled into the original message.
*
* @param bufferSize The buffer size in bytes. Defaults to 1024.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setBufferSize(int bufferSize = 1024);
/**
* @brief Sets the task stack size and priority for the MQTT client task.
*
* @param stackSize The task stack size in bytes. Defaults to 6144.
* @param prio The task priority. Defaults to 5.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setTaskStackAndPriority(int stackSize, int prio);
/**
* @brief Sets the CA root certificate for the MQTT server.
*
* @param cert The certificate in PEM or DER format.
* @param certLen optional length of the certificate shouldn't cert be null-terminated.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setCACert(const char *rootCA, size_t rootCALen = 0);
/**
* @brief Sets a CA root certificate bundle for the MQTT server. Use this
* method if you have multiple CA root certificates and this is the only place
* using SSL/TLS. Otherwise use attachArduinoCACertBundle() to attach an existing
* certificate bundle.
*
* @param bundle The certificate bundle in PEM or DER format.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setCACertBundle(const uint8_t *bundle, size_t bundleLen = 0);
/**
* @brief Attaches an existing CA root certificate bundle for the MQTT server. Like if you
* already use WiFiClientSecure and want to use the same CA root certificate bundle for MQTT.
*
* @param attach Whether to attach or detach the CA root certificate bundle. Defaults to true.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &attachArduinoCACertBundle(bool attach = true);
/**
* @brief Sets the credentials for the MQTT connection.
*
* @param username The username for the MQTT connection. Defaults to nullptr.
* @param password The password for the MQTT connection. Defaults to nullptr.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setCredentials(const char *username, const char *password = nullptr);
/**
* @brief Authenticate via mutual X.509 certificate.
*
* @param clientCert The X.509 certificate issued from the same CA as the server cert.
* @param clientKey The private key for the X.509 certificate.
* @param clientCertLen Optional length of the client certificate. If the certificate is not null-terminated.
* @param clientKeyLen Optional length of the client key. If the key is not null-terminated.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setClientCertificate(const char *clientCert, const char *clientKey, size_t clientCertLen = 0,
size_t clientKeyLen = 0);
/**
* @brief Sets the last will and testament for the MQTT connection.
*
* @param topic The topic for the last will and testament.
* @param qos The QoS level for the last will and testament.
* @param retain The retain flag for the last will and testament.
* @param payload The payload for the last will and testament. Defaults to nullptr.
* @param length The length of the payload for the last will and testament. Defaults to 0.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setWill(const char *topic, uint8_t qos, bool retain, const char *payload = nullptr, int length = 0);
/**
* @brief Sets the MQTT server URI. Supports mqtt://, mqtts:// and ws://, wss:// as
* transport protocols. Fully supports SSL/TLS.
*
* Example: mqtt://162.168.10.1
* mqtt://mqtt.eclipseprojects.io
* mqtt://mqtt.eclipseprojects.io:1884
* mqtt://username:password@mqtt.eclipseprojects.io:1884
* mqtts://mqtt.eclipseprojects.io
* mqtts://mqtt.eclipseprojects.io:8884
* ws://mqtt.eclipseprojects.io:80/mqtt
* wss://mqtt.eclipseprojects.io:443/mqtt
*
* @param uri The MQTT server URI.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &setServer(const char *uri);
/**
* @brief Registers a callback function to be called when the MQTT client is connected.
*
* @param callback The callback function with the signature void(bool sessionPresent) to be registered.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &onConnect(OnConnectUserCallback callback);
/**
* @brief Registers a callback function to be called when the MQTT client is disconnected.
*
* @param callback The callback function with the signature void(bool sessionPresent) to
* be registered.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &onDisconnect(OnDisconnectUserCallback callback);
/**
* @brief Registers a callback function to be called when a topic is subscribed.
*
* @param callback The callback function with the signature void(int msgId) to be registered.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &onSubscribe(OnSubscribeUserCallback callback);
/**
* @brief Registers a callback function to be called when a topic is unsubscribed.
*
* @param callback The callback function with the signature void(int msgId) to be registered.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &onUnsubscribe(OnUnsubscribeUserCallback callback);
/**
* @brief Registers a callback function to be called when a message is received.
* Multipart messages will be reassembled into the original message.
*
* @param callback The callback function with the signature void(char *topic,
* char *payload, int msgId, int retain, int qos, bool dup) to be registered.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &onMessage(OnMessageUserCallback callback);
/**
* @brief Registers a callback function to be called when a message is
* received on a specific topic. Multipart messages will be
* reassembled into the original message. Fully supports MQTT Wildcards.
* Will automatically subscribe to all topics once the client is connected.
*
* @param topic The topic to listen for. MQTT Wildcards are fully supported.
* @param qos The QoS level to listen for.
* @param callback The callback function with the signature void(char *topic,
* char *payload, int msgId, int retain, int qos, bool dup) to be registered.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &onTopic(const char *topic, int qos, OnMessageUserCallback callback);
/**
* @brief Registers a callback function to be called when a message is published.
*
* @param callback The callback function with the signature void(int msgId) to be registered.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &onPublish(OnPublishUserCallback callback);
/**
* @brief Registers a callback function to be called when an error occurs.
*
* @param callback The callback function with the signature void(esp_mqtt_error_codes_t error) to be registered.
* @return A reference to the PsychicMqttClient instance.
*/
PsychicMqttClient &onError(OnErrorUserCallback callback);
/**
* @brief Checks if the MQTT client is connected.
*
* @return True if the client is connected, false otherwise.
*/
bool connected();
/**
* @brief Connects the MQTT client to the server.
*
* @note All parameters must be set before calling this method.
*/
void connect();
/**
* @brief Disconnects the MQTT client from the server.
* This call might be blocking until the client is stopped cleanly
*/
void disconnect();
/**
* @brief Forcefully stops the MQTT client and disconnects from the server.
* This does not trigger the onDisconnect callbacks.
*/
void forceStop();
/**
* @brief Subscribes to a topic. Server must be connected
* for a subscription to succeed.
*
* @param topic The topic to subscribe to.
* @param qos The QoS level for the subscription.
* @return Message ID on success, -1 on failure.
*/
int subscribe(const char *topic, int qos);
/**
* @brief Unsubscribes from a topic. Server must be connected
* for an unsubscription to succeed.
*
* @param topic The topic to unsubscribe from.
* @return Message ID on success, -1 on failure.
*/
int unsubscribe(const char *topic);
/**
* @brief Publishes a message to a topic.
*
* @param topic The topic to publish to.
* @param qos The QoS level (0-2) for the message.
* @param retain The retain flag for the message.
* @param payload The payload for the message. Defaults to nullptr.
* @param length The length of the payload. Defaults to 0.
* @param async Whether to enqueue the message for asynchronous publishing.
* Defaults to true. False means blocking until the message is published.
* @return Message ID on success, -1 on failure.
*/
int publish(const char *topic, int qos, bool retain, const char *payload = nullptr, int length = 0, bool async = true);
/**
* @brief Gets the client ID of the MQTT client.
*
* @return The client ID.
*/
const char *getClientId();
/**
* @brief Returns the ESP-IDF MQTT client config object of the MQTT client in case
* lower level access is needed.
*
* @return The config object.
*/
esp_mqtt_client_config_t *getMqttConfig();
private:
esp_mqtt_client_handle_t _client = nullptr;
esp_mqtt_client_config_t _mqtt_cfg;
esp_mqtt_error_codes_t _lastError;
bool _connected = false;
bool _stopMqttClient = false;
char *_buffer = nullptr;
char *_topic = nullptr;
static void _onMqttEventStatic(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data);
void _onMqttEvent(esp_event_base_t base, int32_t event_id, void *event_data);
bool _isTopicMatch(const char *topic, const char *subscription);
std::vector<OnConnectUserCallback> _onConnectUserCallbacks;
std::vector<OnDisconnectUserCallback> _onDisconnectUserCallbacks;
std::vector<OnSubscribeUserCallback> _onSubscribeUserCallbacks;
std::vector<OnUnsubscribeUserCallback> _onUnsubscribeUserCallbacks;
std::vector<OnMessageUserCallback_t> _onMessageUserCallbacks;
std::vector<OnPublishUserCallback> _onPublishUserCallbacks;
std::vector<OnErrorUserCallback> _onErrorUserCallbacks;
void _onBeforeConnect(esp_mqtt_event_handle_t &event_data, esp_mqtt_client_handle_t &client);
void _onConnect(esp_mqtt_event_handle_t &event_data);
void _onDisconnect(esp_mqtt_event_handle_t &event_data);
void _onSubscribe(esp_mqtt_event_handle_t &event_data);
void _onUnsubscribe(esp_mqtt_event_handle_t &event_data);
void _onMessage(esp_mqtt_event_handle_t &event_data);
void _onPublish(esp_mqtt_event_handle_t &event_data);
void _onError(esp_mqtt_event_handle_t &event_data);
};
Binary file not shown.
+23 -1
View File
@@ -641,7 +641,7 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch
} else if (memcmp(config, "mqtt.iata", 9) == 0) {
sprintf(reply, "> %s", _prefs->mqtt_iata);
} else if (memcmp(config, "mqtt.status", 11) == 0) {
sprintf(reply, "> %s", _prefs->mqtt_status_enabled ? "on" : "off");
MQTTBridge::formatMqttStatusReply(reply, 160, _prefs);
} else if (memcmp(config, "mqtt.packets", 12) == 0) {
sprintf(reply, "> %s", _prefs->mqtt_packets_enabled ? "on" : "off");
} else if (memcmp(config, "mqtt.raw", 8) == 0) {
@@ -677,6 +677,28 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch
}
if (status == WL_CONNECTED) {
sprintf(reply, "> %s, IP: %s, RSSI: %d dBm", status_str, WiFi.localIP().toString().c_str(), WiFi.RSSI());
#ifdef WITH_MQTT_BRIDGE
unsigned long connect_at = MQTTBridge::getWifiConnectedAtMillis();
if (connect_at != 0) {
unsigned long uptime_ms = millis() - connect_at;
unsigned long uptime_sec = uptime_ms / 1000;
unsigned long d = uptime_sec / 86400;
unsigned long h = (uptime_sec % 86400) / 3600;
unsigned long m = (uptime_sec % 3600) / 60;
unsigned long s = uptime_sec % 60;
size_t len = strlen(reply);
const size_t reply_remaining = 128; // caller provides buffer (e.g. 161 bytes), leave headroom
if (d > 0) {
snprintf(reply + len, reply_remaining, ", uptime: %lud %luh %lum %lus", d, h, m, s);
} else if (h > 0) {
snprintf(reply + len, reply_remaining, ", uptime: %luh %lum %lus", h, m, s);
} else if (m > 0) {
snprintf(reply + len, reply_remaining, ", uptime: %lum %lus", m, s);
} else {
snprintf(reply + len, reply_remaining, ", uptime: %lus", s);
}
}
#endif
} else {
sprintf(reply, "> %s (code: %d)", status_str, status);
}
+183 -118
View File
@@ -45,6 +45,48 @@ static bool isWiFiConfigValid(const NodePrefs* prefs) {
#ifdef WITH_MQTT_BRIDGE
// Time (millis()) when WiFi was last seen connected; 0 when disconnected. Used for get wifi.status uptime.
static unsigned long s_wifi_connected_at = 0;
// Singleton for formatMqttStatusReply (set in begin(), cleared in end())
static MQTTBridge* s_mqtt_bridge_instance = nullptr;
unsigned long MQTTBridge::getWifiConnectedAtMillis() {
return s_wifi_connected_at;
}
void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const NodePrefs* prefs) {
if (buf == nullptr || bufsize == 0) return;
const char* msgs = (prefs->mqtt_status_enabled) ? "on" : "off";
if (s_mqtt_bridge_instance == nullptr || !s_mqtt_bridge_instance->_initialized) {
snprintf(buf, bufsize, "> msgs: %s (bridge not running)", msgs);
return;
}
MQTTBridge* b = s_mqtt_bridge_instance;
const char* broker = "n/a";
if (b->_config_valid) {
broker = b->_cached_has_brokers ? "connected" : "disconnected";
}
const char* us = "off";
if (prefs->mqtt_analyzer_us_enabled) {
us = (b->_analyzer_us_client && b->_analyzer_us_client->connected()) ? "connected" : "disconnected";
}
const char* eu = "off";
if (prefs->mqtt_analyzer_eu_enabled) {
eu = (b->_analyzer_eu_client && b->_analyzer_eu_client->connected()) ? "connected" : "disconnected";
}
int q = 0;
#ifdef ESP_PLATFORM
if (b->_packet_queue_handle != nullptr) {
q = (int)uxQueueMessagesWaiting(b->_packet_queue_handle);
}
#else
q = b->_queue_count;
#endif
snprintf(buf, bufsize, "> msgs: %s, broker: %s, us: %s, eu: %s, queue: %d",
msgs, broker, us, eu, q);
}
MQTTBridge::MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity)
: BridgeBase(prefs, mgr, rtc), _mqtt_client(nullptr),
_active_brokers(0), _queue_count(0),
@@ -188,38 +230,33 @@ void MQTTBridge::begin() {
return;
}
// Initialize PsychicMqttClient (will be used by task)
_mqtt_client = new PsychicMqttClient();
// Optimize MQTT client configuration for memory efficiency
optimizeMqttClientConfig(_mqtt_client, false);
// Set up event callbacks for the main MQTT client
_mqtt_client->onConnect([this](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT broker connected");
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].enabled && !_brokers[i].connected) {
_brokers[i].connected = true;
_active_brokers++;
// Update cached broker status
_cached_has_brokers = isAnyBrokerConnected();
break;
// Create main MQTT client only when a custom broker is configured (saves RAM when using analyzer-only)
if (_config_valid) {
_mqtt_client = new PsychicMqttClient();
optimizeMqttClientConfig(_mqtt_client, false);
_mqtt_client->onConnect([this](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT broker connected");
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].enabled && !_brokers[i].connected) {
_brokers[i].connected = true;
_active_brokers++;
_cached_has_brokers = isAnyBrokerConnected();
break;
}
}
}
});
_mqtt_client->onDisconnect([this](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT broker disconnected");
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].connected) {
_brokers[i].connected = false;
_active_brokers--;
// Update cached broker status
_cached_has_brokers = isAnyBrokerConnected();
break;
});
_mqtt_client->onDisconnect([this](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT broker disconnected");
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].connected) {
_brokers[i].connected = false;
_active_brokers--;
_cached_has_brokers = isAnyBrokerConnected();
break;
}
}
}
});
});
}
// Set default broker from preferences or build flags
setBroker(0, _prefs->mqtt_server, _prefs->mqtt_port, _prefs->mqtt_username, _prefs->mqtt_password, true);
@@ -258,8 +295,10 @@ void MQTTBridge::begin() {
_packet_queue_handle = nullptr;
vSemaphoreDelete(_raw_data_mutex);
_raw_data_mutex = nullptr;
delete _mqtt_client;
_mqtt_client = nullptr;
if (_mqtt_client) {
delete _mqtt_client;
_mqtt_client = nullptr;
}
return;
}
@@ -271,36 +310,33 @@ void MQTTBridge::begin() {
WiFi.setAutoConnect(true);
WiFi.begin(_prefs->wifi_ssid, _prefs->wifi_password);
// Initialize PsychicMqttClient
_mqtt_client = new PsychicMqttClient();
optimizeMqttClientConfig(_mqtt_client, false);
// Set up event callbacks
_mqtt_client->onConnect([this](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT broker connected");
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].enabled && !_brokers[i].connected) {
_brokers[i].connected = true;
_active_brokers++;
// Update cached broker status
_cached_has_brokers = isAnyBrokerConnected();
break;
// Create main MQTT client only when a custom broker is configured (saves RAM when using analyzer-only)
if (_config_valid) {
_mqtt_client = new PsychicMqttClient();
optimizeMqttClientConfig(_mqtt_client, false);
_mqtt_client->onConnect([this](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT broker connected");
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].enabled && !_brokers[i].connected) {
_brokers[i].connected = true;
_active_brokers++;
_cached_has_brokers = isAnyBrokerConnected();
break;
}
}
}
});
_mqtt_client->onDisconnect([this](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT broker disconnected");
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].connected) {
_brokers[i].connected = false;
_active_brokers--;
// Update cached broker status
_cached_has_brokers = isAnyBrokerConnected();
break;
});
_mqtt_client->onDisconnect([this](bool sessionPresent) {
MQTT_DEBUG_PRINTLN("MQTT broker disconnected");
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].connected) {
_brokers[i].connected = false;
_active_brokers--;
_cached_has_brokers = isAnyBrokerConnected();
break;
}
}
}
});
});
}
setBroker(0, _prefs->mqtt_server, _prefs->mqtt_port, _prefs->mqtt_username, _prefs->mqtt_password, true);
_analyzer_us_enabled = _prefs->mqtt_analyzer_us_enabled;
@@ -310,11 +346,13 @@ void MQTTBridge::begin() {
#endif
_initialized = true;
s_mqtt_bridge_instance = this;
MQTT_DEBUG_PRINTLN("MQTT Bridge initialized");
}
void MQTTBridge::end() {
MQTT_DEBUG_PRINTLN("Stopping MQTT Bridge...");
s_mqtt_bridge_instance = nullptr;
#ifdef ESP_PLATFORM
// Delete FreeRTOS task first (it will clean up WiFi/MQTT connections)
@@ -344,11 +382,13 @@ void MQTTBridge::end() {
_raw_data_mutex = nullptr;
}
#else
// Disconnect from all brokers
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].enabled && _brokers[i].connected) {
_mqtt_client->disconnect();
_brokers[i].connected = false;
// Disconnect from all brokers (main client only exists when _config_valid)
if (_mqtt_client) {
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].enabled && _brokers[i].connected) {
_mqtt_client->disconnect();
_brokers[i].connected = false;
}
}
}
@@ -463,6 +503,11 @@ void MQTTBridge::mqttTaskLoop() {
unsigned long now = millis();
wl_status_t current_wifi_status = WiFi.status();
// Backfill WiFi connect time when already connected (e.g. boot with WiFi up) so get wifi.status shows uptime
if (current_wifi_status == WL_CONNECTED && s_wifi_connected_at == 0) {
s_wifi_connected_at = millis();
}
// Initialize last_wifi_status on first loop() call
if (!wifi_status_initialized) {
last_wifi_status = current_wifi_status;
@@ -477,6 +522,7 @@ void MQTTBridge::mqttTaskLoop() {
if (current_wifi_status == WL_CONNECTED) {
if (last_wifi_status != WL_CONNECTED) {
wifi_disconnected_time = 0;
s_wifi_connected_at = millis();
// Configure WiFi power management for efficient operation
wifi_ps_type_t ps_mode;
uint8_t ps_pref = _prefs->wifi_power_save;
@@ -499,10 +545,22 @@ void MQTTBridge::mqttTaskLoop() {
// NTP sync will be handled by _ntp_sync_pending flag from WiFi event handler
// This prevents multiple simultaneous syncs
}
// If already connected but we never recorded connect time (e.g. boot with WiFi up), set it now
if (s_wifi_connected_at == 0) {
s_wifi_connected_at = millis();
}
last_wifi_status = WL_CONNECTED;
} else {
if (last_wifi_status == WL_CONNECTED) {
wifi_disconnected_time = now;
s_wifi_connected_at = 0;
// Explicitly stop analyzer MQTT clients so ESP-IDF frees their buffers (it does not free on WiFi drop)
if (_analyzer_us_client) {
_analyzer_us_client->disconnect();
}
if (_analyzer_eu_client) {
_analyzer_eu_client->disconnect();
}
} else if (wifi_disconnected_time > 0) {
unsigned long disconnected_duration = now - wifi_disconnected_time;
@@ -727,6 +785,10 @@ void MQTTBridge::loop() {
unsigned long now = millis();
wl_status_t current_wifi_status = WiFi.status();
if (current_wifi_status == WL_CONNECTED && s_wifi_connected_at == 0) {
s_wifi_connected_at = millis();
}
// Initialize last_wifi_status on first loop() call
if (!wifi_status_initialized) {
last_wifi_status = current_wifi_status;
@@ -743,14 +805,26 @@ void MQTTBridge::loop() {
if (current_wifi_status == WL_CONNECTED) {
if (last_wifi_status != WL_CONNECTED) {
wifi_disconnected_time = 0;
s_wifi_connected_at = millis();
if (!_ntp_synced) {
syncTimeWithNTP();
}
}
if (s_wifi_connected_at == 0) {
s_wifi_connected_at = millis();
}
last_wifi_status = WL_CONNECTED;
} else {
if (last_wifi_status == WL_CONNECTED) {
wifi_disconnected_time = now;
s_wifi_connected_at = 0;
// Explicitly stop analyzer MQTT clients so ESP-IDF frees their buffers (it does not free on WiFi drop)
if (_analyzer_us_client) {
_analyzer_us_client->disconnect();
}
if (_analyzer_eu_client) {
_analyzer_eu_client->disconnect();
}
} else if (wifi_disconnected_time > 0) {
unsigned long disconnected_duration = now - wifi_disconnected_time;
@@ -903,7 +977,7 @@ void MQTTBridge::loop() {
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (_brokers[i].enabled) {
_brokers[i].connected = false;
_brokers[i].last_attempt = 0; // Allow immediate reconnect
_brokers[i].last_attempt = 0; // Allow immediate reconnect after full reinit
}
}
_active_brokers = 0;
@@ -1008,70 +1082,60 @@ void MQTTBridge::connectToBrokers() {
return;
}
// Throttle main broker reconnection to avoid rapid connect/disconnect loops when signal is flaky
static const unsigned long MAIN_BROKER_RECONNECT_THROTTLE_MS = 30000;
// For now, connect to the first enabled broker
// TODO: Implement multi-broker support with PsychicMqttClient
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
if (!_brokers[i].enabled) continue;
// Check if we need to attempt connection
// Allow immediate reconnect if last_attempt is 0 (was reset due to failure)
bool can_attempt = (_brokers[i].last_attempt == 0) ||
(millis() - _brokers[i].last_attempt > _brokers[i].reconnect_interval);
if (!_brokers[i].connected && can_attempt) {
MQTT_DEBUG_PRINTLN("Connecting to broker %d: %s:%d", i, _brokers[i].host, _brokers[i].port);
// Generate unique client ID
char client_id[32];
snprintf(client_id, sizeof(client_id), "%s_%d_%lu", _origin, i, millis());
// Only call connect() once for initial connection.
// After that, ESP-IDF's auto-reconnect handles reconnection automatically.
// If we forced disconnect (e.g. on publish failure), we must call connect() again
// since disconnect() stops the client; throttle to avoid reconnect storms when signal is bad.
if (!_brokers[i].initial_connect_done) {
MQTT_DEBUG_PRINTLN("Initial connection to broker %d: %s:%d", i, _brokers[i].host, _brokers[i].port);
// Set broker URI and connect using PsychicMqttClient API
char broker_uri[128];
snprintf(broker_uri, sizeof(broker_uri), "mqtt://%s:%d", _brokers[i].host, _brokers[i].port);
_mqtt_client->setServer(broker_uri);
// Set credentials if provided
if (strlen(_brokers[i].username) > 0) {
_mqtt_client->setCredentials(_brokers[i].username, _brokers[i].password);
}
// Ensure we're disconnected before attempting new connection
if (_mqtt_client->connected()) {
_mqtt_client->disconnect();
vTaskDelay(pdMS_TO_TICKS(100)); // Brief delay to allow disconnect to complete
}
// Connect to the broker (PsychicMqttClient uses async connection)
// ESP-IDF MQTT client has auto-reconnect enabled by default (10s retry)
_mqtt_client->connect();
// Update attempt timestamp
_brokers[i].initial_connect_done = true;
_brokers[i].last_attempt = millis();
MQTT_DEBUG_PRINTLN("Initiating connection to broker %d", i);
}
// Maintain connection and check for stale connections
if (_brokers[i].connected) {
// Check actual connection state - if it's stale, mark as disconnected and trigger reconnect
// PsychicMqttClient handles automatic reconnection internally, but we need to detect stale state
if (!_mqtt_client->connected()) {
MQTT_DEBUG_PRINTLN("Broker %d connection lost, marking for reconnect", i);
_brokers[i].connected = false;
_active_brokers--;
_brokers[i].last_attempt = 0; // Reset attempt time to allow immediate reconnect
// Update cached broker status
_cached_has_brokers = isAnyBrokerConnected();
}
// Removed aggressive 4-hour health check that was causing connection instability.
// The MQTT client library handles connection health internally, and forcing
// disconnections on healthy connections was causing hours of downtime.
} else {
// Not connected - ensure we attempt reconnection if enough time has passed
// Reset last_attempt if it's been too long (prevents getting stuck)
if (_brokers[i].last_attempt > 0 && (millis() - _brokers[i].last_attempt) > 300000) {
// Been trying for more than 5 minutes - reset to allow fresh attempt
_brokers[i].last_attempt = 0;
MQTT_DEBUG_PRINTLN("Initiated connection to broker %d (auto-reconnect will handle future reconnections)", i);
} else if (_mqtt_client && !_mqtt_client->connected()) {
// Main broker was stopped (e.g. after forced disconnect on publish failure). Reconnect with throttle.
unsigned long now = millis();
unsigned long reconnect_elapsed = (_brokers[i].last_attempt <= now)
? (now - _brokers[i].last_attempt)
: (ULONG_MAX - _brokers[i].last_attempt + now + 1);
if (reconnect_elapsed >= MAIN_BROKER_RECONNECT_THROTTLE_MS) {
MQTT_DEBUG_PRINTLN("Reconnecting to broker %d: %s:%d (throttled)", i, _brokers[i].host, _brokers[i].port);
char broker_uri[128];
snprintf(broker_uri, sizeof(broker_uri), "mqtt://%s:%d", _brokers[i].host, _brokers[i].port);
_mqtt_client->setServer(broker_uri);
if (strlen(_brokers[i].username) > 0) {
_mqtt_client->setCredentials(_brokers[i].username, _brokers[i].password);
}
_mqtt_client->connect();
_brokers[i].last_attempt = now;
}
}
// Note: Connection state (_brokers[i].connected) is updated by onConnect/onDisconnect callbacks
// We just update the cached status here for consistency
_cached_has_brokers = isAnyBrokerConnected();
}
// Update cached broker status after connection attempts
@@ -1342,7 +1406,7 @@ bool MQTTBridge::publishStatus() {
// (packet publishes handle this more gracefully)
_brokers[i].connected = false;
_active_brokers--;
_brokers[i].last_attempt = 0;
_brokers[i].last_attempt = millis(); // Throttle reconnection
_cached_has_brokers = isAnyBrokerConnected();
continue;
}
@@ -1376,7 +1440,7 @@ bool MQTTBridge::publishStatus() {
}
_brokers[i].connected = false;
_active_brokers--;
_brokers[i].last_attempt = 0; // Reset attempt time to allow immediate reconnect
_brokers[i].last_attempt = millis(); // Throttle reconnection (connectToBrokers will retry after MAIN_BROKER_RECONNECT_THROTTLE_MS)
// Update cached broker status
_cached_has_brokers = isAnyBrokerConnected();
}
@@ -1540,7 +1604,7 @@ void MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx,
}
_brokers[i].connected = false;
_active_brokers--;
_brokers[i].last_attempt = 0; // Reset attempt time to allow immediate reconnect
_brokers[i].last_attempt = millis(); // Throttle reconnection (connectToBrokers will retry after MAIN_BROKER_RECONNECT_THROTTLE_MS)
// Update cached broker status
_cached_has_brokers = isAnyBrokerConnected();
}
@@ -1650,7 +1714,7 @@ void MQTTBridge::publishRaw(mesh::Packet* packet) {
}
_brokers[i].connected = false;
_active_brokers--;
_brokers[i].last_attempt = 0; // Reset attempt time to allow immediate reconnect
_brokers[i].last_attempt = millis(); // Throttle reconnection (connectToBrokers will retry after MAIN_BROKER_RECONNECT_THROTTLE_MS)
// Update cached broker status
_cached_has_brokers = isAnyBrokerConnected();
}
@@ -1796,6 +1860,7 @@ void MQTTBridge::setBrokerDefaults() {
_brokers[i].qos = 0;
_brokers[i].enabled = false;
_brokers[i].connected = false;
_brokers[i].initial_connect_done = false;
_brokers[i].reconnect_interval = 5000; // 5 seconds
}
}
@@ -2520,7 +2585,7 @@ void MQTTBridge::maintainAnalyzerConnections() {
}
}
}
// Note: PsychicMqttClient handles automatic reconnection internally,
// but we need to ensure tokens are renewed before reconnection attempts
}
+13
View File
@@ -67,6 +67,7 @@ private:
uint8_t qos;
bool enabled;
bool connected;
bool initial_connect_done; // True after first connect() call - auto-reconnect handles the rest
unsigned long last_attempt;
unsigned long reconnect_interval;
};
@@ -257,6 +258,18 @@ public:
*/
static bool isConfigValid(const NodePrefs* prefs);
/**
* Time (millis()) when WiFi was last seen connected. 0 when disconnected or unknown.
* Used by get wifi.status to report uptime when WITH_MQTT_BRIDGE is defined.
*/
static unsigned long getWifiConnectedAtMillis();
/**
* Format an informative MQTT status line for get mqtt.status (msgs, broker, analyzers, queue).
* Writes into buf, at most bufsize bytes. Requires bridge to have been initialized (begin() called).
*/
static void formatMqttStatusReply(char* buf, size_t bufsize, const NodePrefs* prefs);
/**
* Check if MQTT bridge is ready to operate (has WiFi credentials)
*