Mirror/AI-on-the-edge-device
1
0
Fork 0
mirror of https://github.com/jomjol/AI-on-the-edge-device.git synced 2026-01-27 12:50:39 +03:00
Code Issues Packages Projects Releases Wiki Activity

test1

Browse Source
This commit is contained in:
michael
2026-01-17 02:49:32 +01:00
parent a1ccda2e88
commit 4905663933
283 changed files with 32074 additions and 15759 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
code/components/jomjol_mqtt/CMakeLists.txt
View File

@@ -1,5 +1,5 @@
FILE(GLOB_RECURSE app_sources ${CMAKE_CURRENT_SOURCE_DIR}/*.*)
idf_component_register(SRCS ${app_sources}
INCLUDE_DIRS "."
REQUIRES esp_timer esp-tflite-micro mqtt jomjol_tfliteclass jomjol_helper jomjol_mqtt jomjol_wlan json)
INCLUDE_DIRS "." "../../include"
REQUIRES esp_timer esp-tflite-micro mqtt jomjol_tfliteclass jomjol_helper jomjol_mqtt jomjol_network json)

596
code/components/jomjol_mqtt/interface_mqtt.cpp
View File

@@ -1,86 +1,71 @@
#ifdef ENABLE_MQTT
#include "defines.h"
#include "interface_mqtt.h"
#include "esp_log.h"
#if DEBUG_DETAIL_ON
#include "esp_timer.h"
#endif
#include "connect_wlan.h"
#include "mqtt_client.h"
#include <esp_log.h>
#include <esp_timer.h>
#include <mqtt_client.h>
#include <cJSON.h>
#include "Helper.h"
#include "connect_wifi_sta.h"
#include "read_network_config.h"
#include "ClassLogFile.h"
#include "MainFlowControl.h"
#include "cJSON.h"
#include "../../include/defines.h"
#if DEBUG_DETAIL_ON
#include "esp_timer.h"
#endif
static const char *TAG = "MQTT IF";
std::map<std::string, std::function<void()>>* connectFunktionMap = NULL;
std::map<std::string, std::function<bool(std::string, char*, int)>>* subscribeFunktionMap = NULL;
std::map<std::string, std::function<void()>> *connectFunktionMap = NULL;
std::map<std::string, std::function<bool(std::string, char *, int)>> *subscribeFunktionMap = NULL;
esp_mqtt_client_handle_t client = NULL;
std::string caCert = "";
std::string clientCert = "";
std::string clientKey = "";
int failedOnRound = -1;
int MQTTReconnectCnt = 0;
esp_mqtt_event_id_t esp_mqtt_ID = MQTT_EVENT_ANY;
// ESP_EVENT_ANY_ID
bool mqtt_enabled = false;
bool mqtt_configOK = false;
bool mqtt_initialized = false;
bool mqtt_connected = false;
esp_mqtt_client_handle_t client = NULL;
std::string uri, client_id, lwt_topic, lwt_connected, lwt_disconnected, user, password, maintopic, domoticz_in_topic;
std::string caCert, clientCert, clientKey;
bool validateServerCert = true;
int keepalive;
bool SetRetainFlag;
void (*callbackOnConnected)(std::string, bool) = NULL;
bool MQTTPublish(std::string _key, std::string _content, int qos, bool retained_flag)
bool MQTTPublish(std::string _key, std::string _content, int qos, bool retained_flag)
{
if (!mqtt_enabled) { // MQTT sevice not started / configured (MQTT_Init not called before)
if (!mqtt_controll_config.mqtt_enabled)
{
// MQTT sevice not started / configured (MQTT_Init not called before)
return false;
}
if (failedOnRound == getCountFlowRounds()) { // we already failed in this round, do not retry until the next round
if (failedOnRound == getCountFlowRounds())
{
// we already failed in this round, do not retry until the next round
return true; // Fail quietly
}
#ifdef DEBUG_DETAIL_ON
LogFile.WriteHeapInfo("MQTT Publish");
#endif
MQTT_Init(); // Re-Init client if not initialized yet/anymore
if (mqtt_initialized && mqtt_connected) {
#ifdef DEBUG_DETAIL_ON
long long int starttime = esp_timer_get_time();
#endif
if (mqtt_controll_config.mqtt_initialized && mqtt_controll_config.mqtt_connected)
{
int msg_id = esp_mqtt_client_publish(client, _key.c_str(), _content.c_str(), 0, qos, retained_flag);
#ifdef DEBUG_DETAIL_ON
ESP_LOGD(TAG, "Publish msg_id %d in %lld ms", msg_id, (esp_timer_get_time() - starttime)/1000);
#endif
if (msg_id == -1) {
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Failed to publish topic '" + _key + "', re-trying...");
#ifdef DEBUG_DETAIL_ON
starttime = esp_timer_get_time();
#endif
if (msg_id == -1)
{
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Failed to publish topic '" + _key + "', re-trying...");
msg_id = esp_mqtt_client_publish(client, _key.c_str(), _content.c_str(), 0, qos, retained_flag);
#ifdef DEBUG_DETAIL_ON
ESP_LOGD(TAG, "Publish msg_id %d in %lld ms", msg_id, (esp_timer_get_time() - starttime)/1000);
#endif
if (msg_id == -1) {
if (msg_id == -1)
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Failed to publish topic '" + _key + "', skipping all MQTT publishings in this round!");
failedOnRound = getCountFlowRounds();
return false;
}
}
if (_content.length() > 80) { // Truncate message if too long
if (_content.length() > 80)
{
// Truncate message if too long
_content.resize(80);
_content.append("..");
}
@@ -88,208 +73,216 @@ bool MQTTPublish(std::string _key, std::string _content, int qos, bool retained_
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Published topic: " + _key + ", content: " + _content);
return true;
}
else {
else
{
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Publish skipped. Client not initalized or not connected. (topic: " + _key + ")");
return false;
}
}
static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event) {
static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event)
{
std::string topic = "";
switch (event->event_id) {
case MQTT_EVENT_BEFORE_CONNECT:
mqtt_initialized = true;
break;
case MQTT_EVENT_CONNECTED:
switch (event->event_id)
{
case MQTT_EVENT_BEFORE_CONNECT:
mqtt_controll_config.mqtt_initialized = true;
break;
case MQTT_EVENT_CONNECTED:
MQTTReconnectCnt = 0;
mqtt_controll_config.mqtt_initialized = true;
mqtt_controll_config.mqtt_connected = true;
MQTTconnected();
break;
case MQTT_EVENT_DISCONNECTED:
mqtt_controll_config.mqtt_connected = false;
MQTTReconnectCnt++;
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Disconnected, trying to reconnect");
if (MQTTReconnectCnt >= 5)
{
MQTTReconnectCnt = 0;
mqtt_initialized = true;
mqtt_connected = true;
MQTTconnected();
break;
case MQTT_EVENT_DISCONNECTED:
mqtt_connected = false;
MQTTReconnectCnt++;
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Disconnected, trying to reconnect");
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Disconnected, multiple reconnect attempts failed, still retrying...");
}
break;
if (MQTTReconnectCnt >= 5) {
MQTTReconnectCnt = 0;
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Disconnected, multiple reconnect attempts failed, still retrying...");
}
break;
case MQTT_EVENT_SUBSCRIBED:
ESP_LOGD(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGD(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_PUBLISHED:
ESP_LOGD(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_DATA:
ESP_LOGD(TAG, "MQTT_EVENT_DATA");
ESP_LOGD(TAG, "TOPIC=%.*s", event->topic_len, event->topic);
ESP_LOGD(TAG, "DATA=%.*s", event->data_len, event->data);
topic.assign(event->topic, event->topic_len);
if (subscribeFunktionMap != NULL) {
if (subscribeFunktionMap->find(topic) != subscribeFunktionMap->end()) {
ESP_LOGD(TAG, "call subcribe function for topic %s", topic.c_str());
(*subscribeFunktionMap)[topic](topic, event->data, event->data_len);
}
} else {
ESP_LOGW(TAG, "no handler available\r\n");
}
break;
case MQTT_EVENT_ERROR:
// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718033 --> chapter 3.2.2.3
case MQTT_EVENT_SUBSCRIBED:
ESP_LOGD(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
break;
// The server does not support the level of the MQTT protocol requested by the client
// NOTE: Only protocol 3.1.1 is supported (refer to setting in sdkconfig)
if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_PROTOCOL) {
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, unacceptable protocol version (0x01)");
}
// The client identifier is correct UTF-8 but not allowed by the server
// e.g. clientID empty (cannot be the case -> default set in firmware)
else if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_ID_REJECTED) {
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, identifier rejected (0x02)");
}
// The network connection has been made but the MQTT service is unavailable
else if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_SERVER_UNAVAILABLE) {
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, Server unavailable (0x03)");
}
// The data in the user name or password is malformed
else if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_BAD_USERNAME) {
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, malformed data in username or password (0x04)");
}
// The client is not authorized to connect
else if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_NOT_AUTHORIZED) {
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, not authorized. Check username/password (0x05)");
}
else {
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Other event id:" + event->error_handle->connect_return_code);
ESP_LOGE(TAG, "Other event id:%d", event->error_handle->connect_return_code);
}
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGD(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
break;
#ifdef DEBUG_DETAIL_ON
ESP_LOGD(TAG, "MQTT_EVENT_ERROR - esp_mqtt_error_codes:");
ESP_LOGD(TAG, "error_type:%d", event->error_handle->error_type);
ESP_LOGD(TAG, "connect_return_code:%d", event->error_handle->connect_return_code);
ESP_LOGD(TAG, "esp_transport_sock_errno:%d", event->error_handle->esp_transport_sock_errno);
ESP_LOGD(TAG, "esp_tls_last_esp_err:%d", event->error_handle->esp_tls_last_esp_err);
ESP_LOGD(TAG, "esp_tls_stack_err:%d", event->error_handle->esp_tls_stack_err);
ESP_LOGD(TAG, "esp_tls_cert_verify_flags:%d", event->error_handle->esp_tls_cert_verify_flags);
#endif
case MQTT_EVENT_PUBLISHED:
ESP_LOGD(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
break;
break;
default:
ESP_LOGD(TAG, "Other event id:%d", event->event_id);
break;
case MQTT_EVENT_DATA:
ESP_LOGD(TAG, "MQTT_EVENT_DATA");
ESP_LOGD(TAG, "TOPIC=%.*s", event->topic_len, event->topic);
ESP_LOGD(TAG, "DATA=%.*s", event->data_len, event->data);
topic.assign(event->topic, event->topic_len);
if (subscribeFunktionMap != NULL)
{
if (subscribeFunktionMap->find(topic) != subscribeFunktionMap->end())
{
ESP_LOGD(TAG, "call subcribe function for topic %s", topic.c_str());
(*subscribeFunktionMap)[topic](topic, event->data, event->data_len);
}
}
else
{
ESP_LOGW(TAG, "no handler available\r\n");
}
break;
case MQTT_EVENT_ERROR:
// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718033 --> chapter 3.2.2.3
// The server does not support the level of the MQTT protocol requested by the client
// NOTE: Only protocol 3.1.1 is supported (refer to setting in sdkconfig)
if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_PROTOCOL)
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, unacceptable protocol version (0x01)");
}
// The client identifier is correct UTF-8 but not allowed by the server
// e.g. clientID empty (cannot be the case -> default set in firmware)
else if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_ID_REJECTED)
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, identifier rejected (0x02)");
}
// The network connection has been made but the MQTT service is unavailable
else if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_SERVER_UNAVAILABLE)
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, Server unavailable (0x03)");
}
// The data in the user name or password is malformed
else if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_BAD_USERNAME)
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, malformed data in username or password (0x04)");
}
// The client is not authorized to connect
else if (event->error_handle->connect_return_code == MQTT_CONNECTION_REFUSE_NOT_AUTHORIZED)
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Connection refused, not authorized. Check username/password (0x05)");
}
else
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Other event id:" + event->error_handle->connect_return_code);
ESP_LOGE(TAG, "Other event id:%d", event->error_handle->connect_return_code);
}
break;
default:
ESP_LOGD(TAG, "Other event id:%d", event->event_id);
break;
}
return ESP_OK;
}
static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data) {
static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data)
{
ESP_LOGD(TAG, "Event dispatched from event loop base=%s, event_id=%d", base, (int)event_id);
mqtt_event_handler_cb((esp_mqtt_event_handle_t) event_data);
mqtt_event_handler_cb((esp_mqtt_event_handle_t)event_data);
}
bool MQTT_Configure(std::string _mqttURI, std::string _clientid, std::string _user, std::string _password,
std::string _maintopic, std::string _domoticz_in_topic, std::string _lwt, std::string _lwt_connected, std::string _lwt_disconnected,
std::string _cacertfilename, bool _validateServerCert, std::string _clientcertfilename, std::string _clientkeyfilename,
int _keepalive, bool _SetRetainFlag, void *_callbackOnConnected) {
if ((_mqttURI.length() == 0) || (_maintopic.length() == 0) || (_clientid.length() == 0))
bool MQTT_Configure(void *_callbackOnConnected)
{
if ((mqtt_controll_config.uri.length() == 0) || (mqtt_controll_config.maintopic.length() == 0) || (mqtt_controll_config.clientname.length() == 0))
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Init aborted! Config error (URI, MainTopic or ClientID missing)");
return false;
}
uri = _mqttURI;
client_id = _clientid;
lwt_topic = _maintopic + "/" + _lwt;
lwt_connected = _lwt_connected;
lwt_disconnected = _lwt_disconnected;
keepalive = _keepalive;
SetRetainFlag = _SetRetainFlag;
maintopic = _maintopic;
domoticz_in_topic = _domoticz_in_topic;
callbackOnConnected = ( void (*)(std::string, bool) )(_callbackOnConnected);
callbackOnConnected = (void (*)(std::string, bool))(_callbackOnConnected);
if (_clientcertfilename.length() && _clientkeyfilename.length()) {
std::ifstream cert_ifs(_clientcertfilename);
if (cert_ifs.is_open()) {
if (mqtt_controll_config.clientCertFilename.length() && mqtt_controll_config.clientKeyFilename.length())
{
std::ifstream cert_ifs(mqtt_controll_config.clientCertFilename);
if (cert_ifs.is_open())
{
std::string cert_content((std::istreambuf_iterator<char>(cert_ifs)), (std::istreambuf_iterator<char>()));
clientCert = cert_content;
cert_ifs.close();
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "using clientCert: " + _clientcertfilename);
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "using clientCert: " + mqtt_controll_config.clientCertFilename);
}
else {
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "could not open clientCert: " + _clientcertfilename);
else
{
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "could not open clientCert: " + mqtt_controll_config.clientCertFilename);
}
std::ifstream key_ifs(_clientkeyfilename);
if (key_ifs.is_open()) {
std::ifstream key_ifs(mqtt_controll_config.clientKeyFilename);
if (key_ifs.is_open())
{
std::string key_content((std::istreambuf_iterator<char>(key_ifs)), (std::istreambuf_iterator<char>()));
clientKey = key_content;
key_ifs.close();
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "using clientKey: " + _clientkeyfilename);
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "using clientKey: " + mqtt_controll_config.clientKeyFilename);
}
else {
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "could not open clientKey: " + _clientkeyfilename);
else
{
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "could not open clientKey: " + mqtt_controll_config.clientKeyFilename);
}
}
if (_cacertfilename.length()) {
std::ifstream ca_ifs(_cacertfilename);
if (ca_ifs.is_open()) {
if (mqtt_controll_config.caCertFilename.length())
{
std::ifstream ca_ifs(mqtt_controll_config.caCertFilename);
if (ca_ifs.is_open())
{
std::string content((std::istreambuf_iterator<char>(ca_ifs)), (std::istreambuf_iterator<char>()));
caCert = content;
ca_ifs.close();
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "using caCert: " + _cacertfilename);
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "using caCert: " + mqtt_controll_config.caCertFilename);
}
else {
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "could not open caCert: " + _cacertfilename);
else
{
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "could not open caCert: " + mqtt_controll_config.caCertFilename);
}
}
validateServerCert = _validateServerCert;
#ifdef __HIDE_PASSWORD
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG,
"URI: " + mqtt_controll_config.uri + ", clientname: " + mqtt_controll_config.clientname +
", user: " + mqtt_controll_config.user + ", password: XXXXXXXX, maintopic: " + mqtt_controll_config.maintopic +
", last-will-topic: " + mqtt_controll_config.lwt_topic +
", keepAlive: " + std::to_string(mqtt_controll_config.keepAlive) +
", RetainFlag: " + std::to_string(mqtt_controll_config.retainFlag));
#else
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG,
"URI: " + mqtt_controll_config.uri + ", clientname: " + mqtt_controll_config.clientname +
", user: " + mqtt_controll_config.user + ", password: " + mqtt_controll_config.password +
", maintopic: " + mqtt_controll_config.maintopic + ", last-will-topic: " + mqtt_controll_config.lwt_topic +
", keepAlive: " + std::to_string(mqtt_controll_config.keepAlive) +
", RetainFlag: " + std::to_string(mqtt_controll_config.retainFlag));
#endif
if (_user.length() && _password.length()){
user = _user;
password = _password;
}
#ifdef __HIDE_PASSWORD
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "URI: " + uri + ", clientname: " + client_id + ", user: " + user + ", password: XXXXXXXX, maintopic: "
+ maintopic + ", last-will-topic: " + lwt_topic + ", keepAlive: " + std::to_string(keepalive) + ", RetainFlag: " + std::to_string(SetRetainFlag));
#else
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "URI: " + uri + ", clientname: " + client_id + ", user: " + user + ", password: " + password + ", maintopic: "
+ maintopic + ", last-will-topic: " + lwt_topic + ", keepAlive: " + std::to_string(keepalive) + ", RetainFlag: " + std::to_string(SetRetainFlag));
#endif
mqtt_configOK = true;
mqtt_controll_config.mqtt_configOK = true;
return true;
}
int MQTT_Init() {
if (mqtt_initialized) {
int MQTT_Init()
{
if (mqtt_controll_config.mqtt_initialized)
{
return 0;
}
if (mqtt_configOK) {
mqtt_enabled = true;
} else {
if (mqtt_controll_config.mqtt_configOK)
{
mqtt_controll_config.mqtt_enabled = true;
}
else
{
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Init called, but client is not yet configured.");
return 0;
}
if (!getWIFIisConnected()) {
if (!get_wifi_sta_is_connected())
{
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Init called, but WIFI is not yet connected.");
return 0;
}
@@ -297,31 +290,33 @@ int MQTT_Init() {
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "Init");
MQTTdestroy_client(false);
esp_mqtt_client_config_t mqtt_cfg = { };
esp_mqtt_client_config_t mqtt_cfg = {};
mqtt_cfg.broker.address.uri = uri.c_str();
mqtt_cfg.credentials.client_id = client_id.c_str();
mqtt_cfg.network.disable_auto_reconnect = false; // Reconnection routine active (Default: false)
mqtt_cfg.network.reconnect_timeout_ms = 15000; // Try to reconnect to broker (Default: 10000ms)
mqtt_cfg.network.timeout_ms = 20000; // Network Timeout (Default: 10000ms)
mqtt_cfg.session.message_retransmit_timeout = 3000; // Time after message resent when broker not acknowledged (QoS1, QoS2)
mqtt_cfg.session.last_will.topic = lwt_topic.c_str();
mqtt_cfg.broker.address.uri = mqtt_controll_config.uri.c_str();
mqtt_cfg.credentials.client_id = mqtt_controll_config.clientname.c_str();
mqtt_cfg.network.disable_auto_reconnect = false; // Reconnection routine active (Default: false)
mqtt_cfg.network.reconnect_timeout_ms = 15000; // Try to reconnect to broker (Default: 10000ms)
mqtt_cfg.network.timeout_ms = 20000; // Network Timeout (Default: 10000ms)
mqtt_cfg.session.message_retransmit_timeout = 3000; // Time after message resent when broker not acknowledged (QoS1, QoS2)
mqtt_cfg.session.last_will.topic = mqtt_controll_config.lwt_topic.c_str();
mqtt_cfg.session.last_will.retain = 1;
mqtt_cfg.session.last_will.msg = lwt_disconnected.c_str();
mqtt_cfg.session.last_will.msg_len = (int)(lwt_disconnected.length());
mqtt_cfg.session.keepalive = keepalive;
mqtt_cfg.buffer.size = 2048; // size of MQTT send/receive buffer
mqtt_cfg.session.last_will.msg = mqtt_controll_config.lwt_disconnected.c_str();
mqtt_cfg.session.last_will.msg_len = (int)(mqtt_controll_config.lwt_disconnected.length());
mqtt_cfg.session.keepalive = mqtt_controll_config.keepAlive;
mqtt_cfg.buffer.size = 2048; // size of MQTT send/receive buffer
if (caCert.length()) {
if (caCert.length())
{
mqtt_cfg.broker.verification.certificate = caCert.c_str();
mqtt_cfg.broker.verification.certificate_len = caCert.length() + 1;
// Skip any validation of server certificate CN field, this reduces the
// security of TLS and makes the *MQTT* client susceptible to MITM attacks
mqtt_cfg.broker.verification.skip_cert_common_name_check = !validateServerCert;
mqtt_cfg.broker.verification.skip_cert_common_name_check = !mqtt_controll_config.validateServerCert;
}
if (clientCert.length() && clientKey.length()) {
if (clientCert.length() && clientKey.length())
{
mqtt_cfg.credentials.authentication.certificate = clientCert.c_str();
mqtt_cfg.credentials.authentication.certificate_len = clientCert.length() + 1;
@@ -329,77 +324,79 @@ int MQTT_Init() {
mqtt_cfg.credentials.authentication.key_len = clientKey.length() + 1;
}
if (user.length() && password.length()){
mqtt_cfg.credentials.username = user.c_str();
mqtt_cfg.credentials.authentication.password = password.c_str();
if (mqtt_controll_config.user.length() && mqtt_controll_config.password.length())
{
mqtt_cfg.credentials.username = mqtt_controll_config.user.c_str();
mqtt_cfg.credentials.authentication.password = mqtt_controll_config.password.c_str();
}
#ifdef DEBUG_DETAIL_ON
LogFile.WriteHeapInfo("MQTT Client Init");
#endif
client = esp_mqtt_client_init(&mqtt_cfg);
if (client)
{
esp_err_t ret = esp_mqtt_client_register_event(client, esp_mqtt_ID, mqtt_event_handler, client);
esp_err_t ret = esp_mqtt_client_register_event(client, mqtt_controll_config.esp_mqtt_ID, mqtt_event_handler, client);
if (ret != ESP_OK)
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Could not register event (ret=" + std::to_string(ret) + ")!");
mqtt_initialized = false;
mqtt_controll_config.mqtt_initialized = false;
return -1;
}
#ifdef DEBUG_DETAIL_ON
LogFile.WriteHeapInfo("MQTT Client Start");
#endif
ret = esp_mqtt_client_start(client);
if (ret != ESP_OK)
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Client start failed (retval=" + std::to_string(ret) + ")!");
mqtt_initialized = false;
mqtt_controll_config.mqtt_initialized = false;
return -1;
}
else {
else
{
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "Client started, waiting for established connection...");
mqtt_initialized = true;
mqtt_controll_config.mqtt_initialized = true;
return 1;
}
}
else
{
LogFile.WriteToFile(ESP_LOG_ERROR, TAG, "Init failed, no handle created!");
mqtt_initialized = false;
mqtt_controll_config.mqtt_initialized = false;
return -1;
}
}
void MQTTdestroy_client(bool _disable = false) {
if (client) {
if (mqtt_connected) {
MQTTdestroySubscribeFunction();
void MQTTdestroy_client(bool _disable = false)
{
if (client)
{
if (mqtt_controll_config.mqtt_connected)
{
MQTTdestroySubscribeFunction();
esp_mqtt_client_disconnect(client);
mqtt_connected = false;
mqtt_controll_config.mqtt_connected = false;
}
esp_mqtt_client_stop(client);
esp_mqtt_client_destroy(client);
client = NULL;
mqtt_initialized = false;
mqtt_controll_config.mqtt_initialized = false;
}
if (_disable) // Disable MQTT service, avoid restart with MQTTPublish
mqtt_configOK = false;
if (_disable)
{
// Disable MQTT service, avoid restart with MQTTPublish
mqtt_controll_config.mqtt_configOK = false;
}
}
bool getMQTTisEnabled() {
return mqtt_enabled;
bool getMQTTisEnabled()
{
return mqtt_controll_config.mqtt_enabled;
}
bool getMQTTisConnected() {
return mqtt_connected;
bool getMQTTisConnected()
{
return mqtt_controll_config.mqtt_connected;
}
bool mqtt_handler_flow_start(std::string _topic, char* _data, int _data_len)
bool mqtt_handler_flow_start(std::string _topic, char *_data, int _data_len)
{
ESP_LOGD(TAG, "Handler called: topic %s, data %.*s", _topic.c_str(), _data_len, _data);
@@ -407,47 +404,60 @@ bool mqtt_handler_flow_start(std::string _topic, char* _data, int _data_len)
return ESP_OK;
}
bool mqtt_handler_set_prevalue(std::string _topic, char* _data, int _data_len)
bool mqtt_handler_set_prevalue(std::string _topic, char *_data, int _data_len)
{
//ESP_LOGD(TAG, "Handler called: topic %s, data %.*s", _topic.c_str(), _data_len, _data);
//example: {"numbersname": "main", "value": 12345.1234567}
// ESP_LOGD(TAG, "Handler called: topic %s, data %.*s", _topic.c_str(), _data_len, _data);
// example: {"numbersname": "main", "value": 12345.1234567}
if (_data_len > 0) { // Check if data length > 0
if (_data_len > 0)
{
// Check if data length > 0
cJSON *jsonData = cJSON_Parse(_data);
cJSON *numbersname = cJSON_GetObjectItemCaseSensitive(jsonData, "numbersname");
cJSON *value = cJSON_GetObjectItemCaseSensitive(jsonData, "value");
if (cJSON_IsString(numbersname) && (numbersname->valuestring != NULL)) { // Check if numbersname is valid
if (cJSON_IsNumber(value)) { // Check if value is a number
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "handler_set_prevalue called: numbersname: " + std::string(numbersname->valuestring) +
", value: " + std::to_string(value->valuedouble));
if (flowctrl.UpdatePrevalue(std::to_string(value->valuedouble), std::string(numbersname->valuestring), true)) {
if (cJSON_IsString(numbersname) && (numbersname->valuestring != NULL))
{
// Check if numbersname is valid
if (cJSON_IsNumber(value))
{
// Check if value is a number
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "handler_set_prevalue called: numbersname: " + std::string(numbersname->valuestring) + ", value: " + std::to_string(value->valuedouble));
if (flowctrl.UpdatePrevalue(std::to_string(value->valuedouble), std::string(numbersname->valuestring), true))
{
cJSON_Delete(jsonData);
return ESP_OK;
}
}
else {
else
{
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "handler_set_prevalue: value not a valid number (\"value\": 12345.12345)");
}
}
else {
else
{
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "handler_set_prevalue: numbersname not a valid string (\"numbersname\": \"main\")");
}
cJSON_Delete(jsonData);
}
else {
else
{
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "handler_set_prevalue: handler called, but no data received");
}
return ESP_FAIL;
}
void MQTTconnected(){
if (mqtt_connected) {
void MQTTconnected()
{
if (mqtt_controll_config.mqtt_connected)
{
LogFile.WriteToFile(ESP_LOG_INFO, TAG, "Connected to broker");
if (connectFunktionMap != NULL) {
for(std::map<std::string, std::function<void()>>::iterator it = connectFunktionMap->begin(); it != connectFunktionMap->end(); ++it) {
if (connectFunktionMap != NULL)
{
for (std::map<std::string, std::function<void()>>::iterator it = connectFunktionMap->begin(); it != connectFunktionMap->end(); ++it)
{
it->second();
ESP_LOGD(TAG, "call connect function %s", it->first.c_str());
}
@@ -456,58 +466,71 @@ void MQTTconnected(){
// Subcribe to topics
// Note: Further subsriptions are handled in GPIO class
//*****************************************
std::function<bool(std::string topic, char* data, int data_len)> subHandler1 = mqtt_handler_flow_start;
MQTTregisterSubscribeFunction(maintopic + "/ctrl/flow_start", subHandler1); // subcribe to maintopic/ctrl/flow_start
std::function<bool(std::string topic, char *data, int data_len)> subHandler1 = mqtt_handler_flow_start;
MQTTregisterSubscribeFunction(mqtt_controll_config.maintopic + "/ctrl/flow_start", subHandler1); // subcribe to maintopic/ctrl/flow_start
std::function<bool(std::string topic, char* data, int data_len)> subHandler2 = mqtt_handler_set_prevalue;
MQTTregisterSubscribeFunction(maintopic + "/ctrl/set_prevalue", subHandler2); // subcribe to maintopic/ctrl/set_prevalue
std::function<bool(std::string topic, char *data, int data_len)> subHandler2 = mqtt_handler_set_prevalue;
MQTTregisterSubscribeFunction(mqtt_controll_config.maintopic + "/ctrl/set_prevalue", subHandler2); // subcribe to maintopic/ctrl/set_prevalue
if (subscribeFunktionMap != NULL) {
for(std::map<std::string, std::function<bool(std::string, char*, int)>>::iterator it = subscribeFunktionMap->begin(); it != subscribeFunktionMap->end(); ++it) {
if (subscribeFunktionMap != NULL)
{
for (std::map<std::string, std::function<bool(std::string, char *, int)>>::iterator it = subscribeFunktionMap->begin(); it != subscribeFunktionMap->end(); ++it)
{
int msg_id = esp_mqtt_client_subscribe(client, it->first.c_str(), 0);
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "topic " + it->first + " subscribe successful");
}
}
/* Send Static Topics and Homeassistant Discovery */
if (callbackOnConnected) { // Call onConnected callback routine --> mqtt_server
callbackOnConnected(maintopic, SetRetainFlag);
if (callbackOnConnected)
{
// Call onConnected callback routine --> mqtt_server
callbackOnConnected(mqtt_controll_config.maintopic, mqtt_controll_config.retainFlag);
}
}
}
void MQTTregisterConnectFunction(std::string name, std::function<void()> func){
void MQTTregisterConnectFunction(std::string name, std::function<void()> func)
{
ESP_LOGD(TAG, "MQTTregisteronnectFunction %s\r\n", name.c_str());
if (connectFunktionMap == NULL) {
if (connectFunktionMap == NULL)
{
connectFunktionMap = new std::map<std::string, std::function<void()>>();
}
if ((*connectFunktionMap)[name] != NULL) {
if ((*connectFunktionMap)[name] != NULL)
{
ESP_LOGW(TAG, "connect function %s already registred", name.c_str());
return;
}
(*connectFunktionMap)[name] = func;
if (mqtt_connected) {
if (mqtt_controll_config.mqtt_connected)
{
func();
}
}
void MQTTunregisterConnectFunction(std::string name){
void MQTTunregisterConnectFunction(std::string name)
{
ESP_LOGD(TAG, "unregisterConnnectFunction %s\r\n", name.c_str());
if ((connectFunktionMap != NULL) && (connectFunktionMap->find(name) != connectFunktionMap->end())) {
if ((connectFunktionMap != NULL) && (connectFunktionMap->find(name) != connectFunktionMap->end()))
{
connectFunktionMap->erase(name);
}
}
void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::string, char*, int)> func){
void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::string, char *, int)> func)
{
ESP_LOGD(TAG, "registerSubscribeFunction %s", topic.c_str());
if (subscribeFunktionMap == NULL) {
subscribeFunktionMap = new std::map<std::string, std::function<bool(std::string, char*, int)>>();
if (subscribeFunktionMap == NULL)
{
subscribeFunktionMap = new std::map<std::string, std::function<bool(std::string, char *, int)>>();
}
if ((*subscribeFunktionMap)[topic] != NULL) {
if ((*subscribeFunktionMap)[topic] != NULL)
{
ESP_LOGW(TAG, "topic %s already registered for subscription", topic.c_str());
return;
}
@@ -515,10 +538,14 @@ void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::st
(*subscribeFunktionMap)[topic] = func;
}
void MQTTdestroySubscribeFunction(){
if (subscribeFunktionMap != NULL) {
if (mqtt_connected) {
for(std::map<std::string, std::function<bool(std::string, char*, int)>>::iterator it = subscribeFunktionMap->begin(); it != subscribeFunktionMap->end(); ++it) {
void MQTTdestroySubscribeFunction()
{
if (subscribeFunktionMap != NULL)
{
if (mqtt_controll_config.mqtt_connected)
{
for (std::map<std::string, std::function<bool(std::string, char *, int)>>::iterator it = subscribeFunktionMap->begin(); it != subscribeFunktionMap->end(); ++it)
{
int msg_id = esp_mqtt_client_unsubscribe(client, it->first.c_str());
ESP_LOGD(TAG, "topic %s unsubscribe successful, msg_id=%d", it->first.c_str(), msg_id);
}
@@ -529,4 +556,3 @@ void MQTTdestroySubscribeFunction(){
subscribeFunktionMap = NULL;
}
}
#endif //ENABLE_MQTT

14
code/components/jomjol_mqtt/interface_mqtt.h
View File

@@ -1,5 +1,3 @@
#ifdef ENABLE_MQTT
#pragma once
#ifndef INTERFACE_MQTT_H
@@ -9,23 +7,19 @@
#include <map>
#include <functional>
bool MQTT_Configure(std::string _mqttURI, std::string _clientid, std::string _user, std::string _password,
std::string _maintopic, std::string _domoticz_in_topic, std::string _lwt, std::string _lwt_connected, std::string _lwt_disconnected,
std::string _cacertfilename, bool _validateServerCert, std::string _clientcertfilename, std::string _clientkeyfilename,
int _keepalive, bool SetRetainFlag, void *callbackOnConnected);
bool MQTT_Configure(void *callbackOnConnected);
int MQTT_Init();
void MQTTdestroy_client(bool _disable);
bool MQTTPublish(std::string _key, std::string _content, int qos, bool retained_flag = 1); // retained Flag as Standart
bool MQTTPublish(std::string _key, std::string _content, int qos, bool retained_flag = 1); // retained Flag as Standart
bool getMQTTisEnabled();
bool getMQTTisConnected();
void MQTTregisterConnectFunction(std::string name, std::function<void()> func);
void MQTTunregisterConnectFunction(std::string name);
void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::string, char*, int)> func);
void MQTTregisterSubscribeFunction(std::string topic, std::function<bool(std::string, char *, int)> func);
void MQTTdestroySubscribeFunction();
void MQTTconnected();
#endif //INTERFACE_MQTT_H
#endif //#ENABLE_MQTT
#endif // INTERFACE_MQTT_H

305
code/components/jomjol_mqtt/mqtt_outbox.c
View File

@@ -1,305 +0,0 @@
/* This is a modification of https://github.com/espressif/esp-mqtt/blob/master/lib/mqtt_outbox.c
* to use the PSRAM instead of the internal heap.
*/
#include "mqtt_outbox.h"
#include <stdlib.h>
#include <string.h>
#include "sys/queue.h"
#include "esp_log.h"
#include "esp_heap_caps.h"
/* Enable this to use the PSRAM for MQTT Publishing.
* This saves 10 kBytes of RAM, see https://github.com/jomjol/AI-on-the-edge-device/pull/2113
* However we can run into PSRAM fragmentation issues, leading to insufficient large blocks to load the model.
* See https://github.com/jomjol/AI-on-the-edge-device/issues/2200 */
#define USE_PSRAM
#ifdef CONFIG_MQTT_CUSTOM_OUTBOX
static const char *TAG = "outbox";
typedef struct outbox_item {
char *buffer;
int len;
int msg_id;
int msg_type;
int msg_qos;
outbox_tick_t tick;
pending_state_t pending;
STAILQ_ENTRY(outbox_item) next;
} outbox_item_t;
STAILQ_HEAD(outbox_list_t, outbox_item);
outbox_handle_t outbox_init(void)
{
#ifdef USE_PSRAM
outbox_handle_t outbox = heap_caps_calloc(1, sizeof(struct outbox_list_t), MALLOC_CAP_8BIT | MALLOC_CAP_SPIRAM);
#else
outbox_handle_t outbox = calloc(1, sizeof(struct outbox_list_t));
#endif
//ESP_MEM_CHECK(TAG, outbox, return NULL);
STAILQ_INIT(outbox);
return outbox;
}
outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, outbox_message_handle_t message, outbox_tick_t tick)
{
#ifdef USE_PSRAM
outbox_item_handle_t item = heap_caps_calloc(1, sizeof(outbox_item_t), MALLOC_CAP_8BIT | MALLOC_CAP_SPIRAM);
#else
outbox_item_handle_t item = calloc(1, sizeof(outbox_item_t));
#endif
//ESP_MEM_CHECK(TAG, item, return NULL);
item->msg_id = message->msg_id;
item->msg_type = message->msg_type;
item->msg_qos = message->msg_qos;
item->tick = tick;
item->len = message->len + message->remaining_len;
item->pending = QUEUED;
#ifdef USE_PSRAM
item->buffer = heap_caps_malloc(message->len + message->remaining_len, MALLOC_CAP_8BIT | MALLOC_CAP_SPIRAM);
#else
item->buffer = malloc(message->len + message->remaining_len);
#endif
/*ESP_MEM_CHECK(TAG, item->buffer, {
free(item);
return NULL;
});*/
memcpy(item->buffer, message->data, message->len);
if (message->remaining_data) {
memcpy(item->buffer + message->len, message->remaining_data, message->remaining_len);
}
STAILQ_INSERT_TAIL(outbox, item, next);
ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%d, size=%d", message->msg_id, message->msg_type, message->len + message->remaining_len, outbox_get_size(outbox));
return item;
}
outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id)
{
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
if (item->msg_id == msg_id) {
return item;
}
}
return NULL;
}
outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox, pending_state_t pending, outbox_tick_t *tick)
{
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
if (item->pending == pending) {
if (tick) {
*tick = item->tick;
}
return item;
}
}
return NULL;
}
esp_err_t outbox_delete_item(outbox_handle_t outbox, outbox_item_handle_t item_to_delete)
{
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
if (item == item_to_delete) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
#ifdef USE_PSRAM
heap_caps_free(item->buffer);
heap_caps_free(item);
#else
free(item->buffer);
free(item);
#endif
return ESP_OK;
}
}
return ESP_FAIL;
}
uint8_t *outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos)
{
if (item) {
*len = item->len;
*msg_id = item->msg_id;
*msg_type = item->msg_type;
*qos = item->msg_qos;
return (uint8_t *)item->buffer;
}
return NULL;
}
esp_err_t outbox_delete(outbox_handle_t outbox, int msg_id, int msg_type)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (item->msg_id == msg_id && (0xFF & (item->msg_type)) == msg_type) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
#ifdef USE_PSRAM
heap_caps_free(item->buffer);
heap_caps_free(item);
#else
free(item->buffer);
free(item);
#endif
ESP_LOGD(TAG, "DELETED msgid=%d, msg_type=%d, remain size=%d", msg_id, msg_type, outbox_get_size(outbox));
return ESP_OK;
}
}
return ESP_FAIL;
}
esp_err_t outbox_delete_msgid(outbox_handle_t outbox, int msg_id)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (item->msg_id == msg_id) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
#ifdef USE_PSRAM
heap_caps_free(item->buffer);
heap_caps_free(item);
#else
free(item->buffer);
free(item);
#endif
}
}
return ESP_OK;
}
esp_err_t outbox_set_pending(outbox_handle_t outbox, int msg_id, pending_state_t pending)
{
outbox_item_handle_t item = outbox_get(outbox, msg_id);
if (item) {
item->pending = pending;
return ESP_OK;
}
return ESP_FAIL;
}
pending_state_t outbox_item_get_pending(outbox_item_handle_t item)
{
if (item) {
return item->pending;
}
return QUEUED;
}
esp_err_t outbox_set_tick(outbox_handle_t outbox, int msg_id, outbox_tick_t tick)
{
outbox_item_handle_t item = outbox_get(outbox, msg_id);
if (item) {
item->tick = tick;
return ESP_OK;
}
return ESP_FAIL;
}
esp_err_t outbox_delete_msgtype(outbox_handle_t outbox, int msg_type)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (item->msg_type == msg_type) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
#ifdef USE_PSRAM
heap_caps_free(item->buffer);
heap_caps_free(item);
#else
free(item->buffer);
free(item);
#endif
}
}
return ESP_OK;
}
int outbox_delete_single_expired(outbox_handle_t outbox, outbox_tick_t current_tick, outbox_tick_t timeout)
{
int msg_id = -1;
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
if (current_tick - item->tick > timeout) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
#ifdef USE_PSRAM
heap_caps_free(item->buffer);
#else
free(item->buffer);
#endif
msg_id = item->msg_id;
#ifdef USE_PSRAM
heap_caps_free(item);
#else
free(item);
#endif
return msg_id;
}
}
return msg_id;
}
int outbox_delete_expired(outbox_handle_t outbox, outbox_tick_t current_tick, outbox_tick_t timeout)
{
int deleted_items = 0;
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (current_tick - item->tick > timeout) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
#ifdef USE_PSRAM
heap_caps_free(item->buffer);
heap_caps_free(item);
#else
free(item->buffer);
free(item);
#endif
deleted_items ++;
}
}
return deleted_items;
}
int outbox_get_size(outbox_handle_t outbox)
{
int siz = 0;
outbox_item_handle_t item;
STAILQ_FOREACH(item, outbox, next) {
// Suppressing "use after free" warning as this could happen only if queue is in inconsistent state
// which never happens if STAILQ interface used
siz += item->len; // NOLINT(clang-analyzer-unix.Malloc)
}
return siz;
}
void outbox_delete_all_items(outbox_handle_t outbox)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
#ifdef USE_PSRAM
heap_caps_free(item->buffer);
heap_caps_free(item);
#else
free(item->buffer);
free(item);
#endif
}
}
void outbox_destroy(outbox_handle_t outbox)
{
outbox_delete_all_items(outbox);
#ifdef USE_PSRAM
heap_caps_free(outbox);
#else
free(outbox);
#endif
}
#endif /* CONFIG_MQTT_CUSTOM_OUTBOX */

66
code/components/jomjol_mqtt/mqtt_outbox.h
View File

@@ -1,66 +0,0 @@
/* This is an adaption of https://github.com/espressif/esp-mqtt/blob/master/lib/include/mqtt_outbox.h
* This file is subject to the terms and conditions defined in
* file 'LICENSE', which is part of this source code package.
* Tuan PM <tuanpm at live dot com>
*/
#ifndef _MQTT_OUTOBX_H_
#define _MQTT_OUTOBX_H_
//#include "platform.h"
#include "esp_err.h"
#ifdef __cplusplus
extern "C" {
#endif
struct outbox_item;
typedef struct outbox_list_t *outbox_handle_t;
typedef struct outbox_item *outbox_item_handle_t;
typedef struct outbox_message *outbox_message_handle_t;
typedef long long outbox_tick_t;
typedef struct outbox_message {
uint8_t *data;
int len;
int msg_id;
int msg_qos;
int msg_type;
uint8_t *remaining_data;
int remaining_len;
} outbox_message_t;
typedef enum pending_state {
QUEUED,
TRANSMITTED,
ACKNOWLEDGED,
CONFIRMED
} pending_state_t;
outbox_handle_t outbox_init(void);
outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, outbox_message_handle_t message, outbox_tick_t tick);
outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox, pending_state_t pending, outbox_tick_t *tick);
outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id);
uint8_t *outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos);
esp_err_t outbox_delete(outbox_handle_t outbox, int msg_id, int msg_type);
esp_err_t outbox_delete_msgid(outbox_handle_t outbox, int msg_id);
esp_err_t outbox_delete_msgtype(outbox_handle_t outbox, int msg_type);
esp_err_t outbox_delete_item(outbox_handle_t outbox, outbox_item_handle_t item);
int outbox_delete_expired(outbox_handle_t outbox, outbox_tick_t current_tick, outbox_tick_t timeout);
/**
* @brief Deletes single expired message returning it's message id
*
* @return msg id of the deleted message, -1 if no expired message in the outbox
*/
int outbox_delete_single_expired(outbox_handle_t outbox, outbox_tick_t current_tick, outbox_tick_t timeout);
esp_err_t outbox_set_pending(outbox_handle_t outbox, int msg_id, pending_state_t pending);
pending_state_t outbox_item_get_pending(outbox_item_handle_t item);
esp_err_t outbox_set_tick(outbox_handle_t outbox, int msg_id, outbox_tick_t tick);
int outbox_get_size(outbox_handle_t outbox);
void outbox_destroy(outbox_handle_t outbox);
void outbox_delete_all_items(outbox_handle_t outbox);
#ifdef __cplusplus
}
#endif
#endif

410
code/components/jomjol_mqtt/server_mqtt.cpp
View File

@@ -1,391 +1,371 @@
#ifdef ENABLE_MQTT
#include "defines.h"
#include "server_mqtt.h"
#include <string>
#include <sstream>
#include <iomanip>
#include <vector>
#include <esp_log.h>
#include "defines.h"
#include "Helper.h"
#include "connect_wifi_sta.h"
#include "read_network_config.h"
#include "esp_log.h"
#include "ClassLogFile.h"
#include "connect_wlan.h"
#include "read_wlanini.h"
#include "server_mqtt.h"
#include "ClassFlowMQTT.h"
#include "interface_mqtt.h"
#include "time_sntp.h"
#include "../../include/defines.h"
#include "basic_auth.h"
static const char *TAG = "MQTT SERVER";
extern const char* libfive_git_version(void);
extern const char* libfive_git_revision(void);
extern const char* libfive_git_branch(void);
extern const char *libfive_git_version(void);
extern const char *libfive_git_revision(void);
extern const char *libfive_git_branch(void);
extern std::string getFwVersion(void);
std::vector<NumberPost*>* NUMBERS;
bool HomeassistantDiscovery = false;
std::string meterType = "";
std::string valueUnit = "";
std::string timeUnit = "";
std::string rateUnit = "Unit/Minute";
float roundInterval; // Minutes
int keepAlive = 0; // Seconds
bool retainFlag;
static std::string maintopic, domoticzintopic;
std::vector<NumberPost *> *NUMBERS;
bool sendingOf_DiscoveryAndStaticTopics_scheduled = true; // Set it to true to make sure it gets sent at least once after startup
void mqttServer_setParameter(std::vector<NumberPost*>* _NUMBERS, int _keepAlive, float _roundInterval) {
void mqttServer_setParameter(std::vector<NumberPost *> *_NUMBERS)
{
NUMBERS = _NUMBERS;
keepAlive = _keepAlive;
roundInterval = _roundInterval;
}
void mqttServer_setMeterType(std::string _meterType, std::string _valueUnit, std::string _timeUnit,std::string _rateUnit) {
meterType = _meterType;
valueUnit = _valueUnit;
timeUnit = _timeUnit;
rateUnit = _rateUnit;
}
/**
* Takes any multi-level MQTT-topic and returns the last topic level as nodeId
* see https://www.hivemq.com/blog/mqtt-essentials-part-5-mqtt-topics-best-practices/ for details about MQTT topics
*/
std::string createNodeId(std::string &topic) {
*/
std::string createNodeId(std::string &topic)
{
auto splitPos = topic.find_last_of('/');
return (splitPos == std::string::npos) ? topic : topic.substr(splitPos + 1);
}
bool sendHomeAssistantDiscoveryTopic(std::string group, std::string field,
std::string name, std::string icon, std::string unit, std::string deviceClass, std::string stateClass, std::string entityCategory,
int qos) {
bool sendHomeAssistantDiscoveryTopic(std::string group, std::string field, std::string name, std::string icon, std::string unit, std::string deviceClass, std::string stateClass, std::string entityCategory, int qos)
{
std::string version = std::string(libfive_git_version());
if (version == "") {
if (version == "")
{
version = std::string(libfive_git_branch()) + " (" + std::string(libfive_git_revision()) + ")";
}
std::string topicFull;
std::string configTopic;
std::string payload;
std::string component;
std::string configTopic = field;
configTopic = field;
if (group != "" && (*NUMBERS).size() > 1) { // There is more than one meter, prepend the group so we can differentiate them
if (group != "" && (*NUMBERS).size() > 1)
{
// There is more than one meter, prepend the group so we can differentiate them
configTopic = group + "_" + field;
name = group + " " + name;
}
if (field == "problem") { // Special case: Binary sensor which is based on error topic
std::string component;
if (field == "problem")
{
// Special case: Binary sensor which is based on error topic
component = "binary_sensor";
}
else if (field == "flowstart") { // Special case: Button
else if (field == "flowstart")
{
// Special case: Button
component = "button";
}
else {
else
{
component = "sensor";
}
/**
/**
* homeassistant needs the MQTT discovery topic according to the following structure:
* <discovery_prefix>/<component>/[<node_id>/]<object_id>/config
* if the main topic is embedded in a nested structure, we just use the last part as node_id
* if the main topic is embedded in a nested structure, we just use the last part as node_id
* This means a maintopic "home/test/watermeter" is transformed to the discovery topic "homeassistant/sensor/watermeter/..."
*/
std::string node_id = createNodeId(maintopic);
topicFull = "homeassistant/" + component + "/" + node_id + "/" + configTopic + "/config";
/* See https://www.home-assistant.io/docs/mqtt/discovery/ */
payload = string("{") +
"\"~\": \"" + maintopic + "\"," +
"\"unique_id\": \"" + maintopic + "-" + configTopic + "\"," +
"\"object_id\": \"" + maintopic + "_" + configTopic + "\"," + // Default entity ID; required for HA <= 2025.10
"\"default_entity_id\": \"" + component + "." + maintopic + "_" + configTopic + "\"," + // Default entity ID; required in HA >=2026.4
"\"name\": \"" + name + "\"," +
"\"icon\": \"mdi:" + icon + "\",";
*/
std::string node_id = createNodeId(mqtt_controll_config.maintopic);
topicFull = mqtt_controll_config.discoveryprefix + "/" + component + "/" + node_id + "/" + configTopic + "/config";
if (group != "") {
if (field == "problem") { // Special case: Binary sensor which is based on error topic
/* See https://www.home-assistant.io/docs/mqtt/discovery/ */
std::string payload = string("{") + "\"~\": \"" + mqtt_controll_config.maintopic + "\"," +
"\"unique_id\": \"" + mqtt_controll_config.maintopic + "-" + configTopic + "\"," +
"\"object_id\": \"" + mqtt_controll_config.maintopic + "_" + configTopic + "\"," + // Default entity ID; required for HA <= 2025.10
"\"default_entity_id\": \"" + component + "." + mqtt_controll_config.maintopic + "_" + configTopic + "\"," + // Default entity ID; required in HA >=2026.4
"\"name\": \"" + name + "\"," + "\"icon\": \"mdi:" + icon + "\",";
if (group != "")
{
if (field == "problem")
{
// Special case: Binary sensor which is based on error topic
payload += "\"state_topic\": \"~/" + group + "/error\",";
payload += "\"value_template\": \"{{ 'OFF' if 'no error' in value else 'ON'}}\",";
}
else {
else
{
payload += "\"state_topic\": \"~/" + group + "/" + field + "\",";
}
}
else {
if (field == "problem") { // Special case: Binary sensor which is based on error topic
else
{
if (field == "problem")
{
// Special case: Binary sensor which is based on error topic
payload += "\"state_topic\": \"~/error\",";
payload += "\"value_template\": \"{{ 'OFF' if 'no error' in value else 'ON'}}\",";
}
else if (field == "flowstart") { // Special case: Button
else if (field == "flowstart")
{
// Special case: Button
payload += "\"cmd_t\":\"~/ctrl/flow_start\","; // Add command topic
}
else {
else
{
payload += "\"state_topic\": \"~/" + field + "\",";
}
}
if (unit != "") {
if (unit != "")
{
payload += "\"unit_of_meas\": \"" + unit + "\",";
}
if (deviceClass != "") {
if (deviceClass != "")
{
payload += "\"device_class\": \"" + deviceClass + "\",";
}
if (stateClass != "") {
if (stateClass != "")
{
payload += "\"state_class\": \"" + stateClass + "\",";
}
}
if (entityCategory != "") {
if (entityCategory != "")
{
payload += "\"entity_category\": \"" + entityCategory + "\",";
}
}
payload +=
"\"availability_topic\": \"~/" + std::string(LWT_TOPIC) + "\"," +
"\"payload_available\": \"" + LWT_CONNECTED + "\"," +
payload +=
"\"availability_topic\": \"~/" + std::string(LWT_TOPIC) + "\"," +
"\"payload_available\": \"" + LWT_CONNECTED + "\"," +
"\"payload_not_available\": \"" + LWT_DISCONNECTED + "\",";
payload += string("\"device\": {") +
"\"identifiers\": [\"" + maintopic + "\"]," +
"\"name\": \"" + maintopic + "\"," +
"\"model\": \"Meter Digitizer\"," +
"\"manufacturer\": \"AI on the Edge Device\"," +
"\"sw_version\": \"" + version + "\"," +
"\"configuration_url\": \"http://" + *getIPAddress() + "\"" +
"}" +
"}";
payload += string("\"device\": {") + "\"identifiers\": [\"" + mqtt_controll_config.maintopic + "\"]," +
"\"name\": \"" + mqtt_controll_config.maintopic + "\"," +
"\"model\": \"Meter Digitizer\"," +
"\"manufacturer\": \"AI on the Edge Device\"," +
"\"sw_version\": \"" + version + "\"," +
"\"configuration_url\": \"http://" + network_config.ipaddress + "\"" + "}" + "}";
return MQTTPublish(topicFull, payload, qos, true);
}
bool MQTThomeassistantDiscovery(int qos) {
bool allSendsSuccessed = false;
if (!getMQTTisConnected()) {
bool MQTThomeassistantDiscovery(int qos)
{
if (!getMQTTisConnected())
{
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Unable to send Homeassistant Discovery Topics, we are not connected to the MQTT broker!");
return false;
}
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Publishing Homeassistant Discovery topics (Meter Type: '" + meterType + "', Value Unit: '" + valueUnit + "' , Rate Unit: '" + rateUnit + "') ...");
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Publishing Homeassistant Discovery topics (Meter Type: '" + mqtt_controll_config.meterType + "', Value Unit: '" + mqtt_controll_config.valueUnit + "' , Rate Unit: '" + mqtt_controll_config.rateUnit + "') ...");
int aFreeInternalHeapSizeBefore = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
bool allSendsSuccessed = false;
int aFreeInternalHeapSizeBefore = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
// Group | Field | User Friendly Name | Icon | Unit | Device Class | State Class | Entity Category
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "uptime", "Uptime", "progress-clock", "s", "duration", "measurement", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "MAC", "MAC Address", "network-outline", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "fwVersion", "Firmware Version", "application-outline", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "hostname", "Hostname", "network-outline", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "freeMem", "Free Memory", "memory", "B", "", "measurement", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "wifiRSSI", "Wi-Fi RSSI", "wifi", "dBm", "signal_strength", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "CPUtemp", "CPU Temperature", "thermometer", "°C", "temperature", "measurement", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "interval", "Interval", "clock-time-eight-outline", "min", "", "measurement", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "IP", "IP", "network-outline", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "status", "Status", "list-status", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "flowstart", "Manual Flow Start", "timer-play-outline", "", "", "", "", qos);
// Group | Field | User Friendly Name | Icon | Unit | Device Class | State Class | Entity Category
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "uptime", "Uptime", "progress-clock", "s", "duration", "measurement", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "MAC", "MAC Address", "network-outline", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "fwVersion", "Firmware Version", "application-outline", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "hostname", "Hostname", "network-outline", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "freeMem", "Free Memory", "memory", "B", "", "measurement", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "wifiRSSI", "Wi-Fi RSSI", "wifi", "dBm", "signal_strength", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "CPUtemp", "CPU Temperature", "thermometer", "°C", "temperature", "measurement", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "interval", "Interval", "clock-time-eight-outline", "min", "", "measurement", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "IP", "IP", "network-outline", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "status", "Status", "list-status", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic("", "flowstart", "Manual Flow Start", "timer-play-outline", "", "", "", "", qos);
for (int i = 0; i < (*NUMBERS).size(); ++i) {
for (int i = 0; i < (*NUMBERS).size(); ++i)
{
std::string group = (*NUMBERS)[i]->name;
if (group == "default") {
if (group == "default")
{
group = "";
}
/* If "Allow neg. rate" is true, use "measurement" instead of "total_increasing" for the State Class, see https://github.com/jomjol/AI-on-the-edge-device/issues/3331 */
std::string value_state_class = "total_increasing";
if (meterType == "temperature") {
if (mqtt_controll_config.meterType == "temperature")
{
value_state_class = "measurement";
}
else if ((*NUMBERS)[i]->AllowNegativeRates) {
else if ((*NUMBERS)[i]->AllowNegativeRates)
{
value_state_class = "total";
}
/* Energy meters need a different Device Class, see https://github.com/jomjol/AI-on-the-edge-device/issues/3333 */
std::string rate_device_class = "volume_flow_rate";
if (meterType == "energy") {
if (mqtt_controll_config.meterType == "energy")
{
rate_device_class = "power";
}
// Group | Field | User Friendly Name | Icon | Unit | Device Class | State Class | Entity Category | QoS
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "value", "Value", "gauge", valueUnit, meterType, value_state_class, "", qos); // State Class = "total_increasing" if <NUMBERS>.AllowNegativeRates = false, "measurement" in case of a thermometer, else use "total".
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "raw", "Raw Value", "raw", valueUnit, meterType, value_state_class, "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "error", "Error", "alert-circle-outline", "", "", "", "diagnostic", qos);
// Group | Field | User Friendly Name | Icon | Unit | Device Class | State Class | Entity Category | QoS
// State Class = "total_increasing" if <NUMBERS>.AllowNegativeRates = false, "measurement" in case of a thermometer, else use "total".
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "value", "Value", "gauge", mqtt_controll_config.valueUnit, mqtt_controll_config.meterType, value_state_class, "", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "raw", "Raw Value", "raw", mqtt_controll_config.valueUnit, mqtt_controll_config.meterType, value_state_class, "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "error", "Error", "alert-circle-outline", "", "", "", "diagnostic", qos);
/* Not announcing "rate" as it is better to use rate_per_time_unit resp. rate_per_digitization_round */
// allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "rate", "Rate (Unit/Minute)", "swap-vertical", "", "", "", "", qos); // Legacy, always Unit per Minute
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "rate_per_time_unit", "Rate (" + rateUnit + ")", "swap-vertical", rateUnit, rate_device_class, "measurement", "", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "rate_per_digitization_round","Change since last Digitization round", "arrow-expand-vertical", valueUnit, "", "measurement", "", qos); // correctly the Unit is Unit/Interval!
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "timestamp", "Timestamp", "clock-time-eight-outline", "", "timestamp", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "json", "JSON", "code-json", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "problem", "Problem", "alert-outline", "", "problem", "", "", qos); // Special binary sensor which is based on error topic
// allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "rate", "Rate (Unit/Minute)", "swap-vertical", "", "", "", "", qos); // Legacy, always Unit per Minute
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "rate_per_time_unit", "Rate (" + mqtt_controll_config.rateUnit + ")", "swap-vertical", mqtt_controll_config.rateUnit, rate_device_class, "measurement", "", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "rate_per_digitization_round", "Change since last Digitization round", "arrow-expand-vertical", mqtt_controll_config.valueUnit, "", "measurement", "", qos); // correctly the Unit is Unit/Interval!
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "timestamp", "Timestamp", "clock-time-eight-outline", "", "timestamp", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "json", "JSON", "code-json", "", "", "", "diagnostic", qos);
allSendsSuccessed |= sendHomeAssistantDiscoveryTopic(group, "problem", "Problem", "alert-outline", "", "problem", "", "", qos); // Special binary sensor which is based on error topic
}
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Successfully published all Homeassistant Discovery MQTT topics");
int aFreeInternalHeapSizeAfter = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
int aMinFreeInternalHeapSize = heap_caps_get_minimum_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
int aMinFreeInternalHeapSize = heap_caps_get_minimum_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Int. Heap Usage before Publishing Homeassistand Discovery Topics: " +
to_string(aFreeInternalHeapSizeBefore) + ", after: " + to_string(aFreeInternalHeapSizeAfter) + ", delta: " +
to_string(aFreeInternalHeapSizeBefore - aFreeInternalHeapSizeAfter) + ", lowest free: " + to_string(aMinFreeInternalHeapSize));
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Int. Heap Usage before Publishing Homeassistand Discovery Topics: " + to_string(aFreeInternalHeapSizeBefore) + ", after: " + to_string(aFreeInternalHeapSizeAfter) + ", delta: " + to_string(aFreeInternalHeapSizeBefore - aFreeInternalHeapSizeAfter) + ", lowest free: " + to_string(aMinFreeInternalHeapSize));
return allSendsSuccessed;
}
bool publishSystemData(int qos) {
bool allSendsSuccessed = false;
if (!getMQTTisConnected()) {
bool publishSystemData(int qos)
{
if (!getMQTTisConnected())
{
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Unable to send System Topics, we are not connected to the MQTT broker!");
return false;
}
char tmp_char[50];
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Publishing System MQTT topics...");
int aFreeInternalHeapSizeBefore = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
char tmp_char[50];
bool allSendsSuccessed = false;
int aFreeInternalHeapSizeBefore = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
allSendsSuccessed |= MQTTPublish(maintopic + "/" + std::string(LWT_TOPIC), LWT_CONNECTED, qos, retainFlag); // Publish "connected" to maintopic/connection
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + std::string(LWT_TOPIC), LWT_CONNECTED, qos, mqtt_controll_config.retainFlag); // Publish "connected" to maintopic/connection
sprintf(tmp_char, "%ld", (long)getUpTime());
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "uptime", std::string(tmp_char), qos, retainFlag);
sprintf(tmp_char, "%lu", (long) getESPHeapSize());
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "freeMem", std::string(tmp_char), qos, retainFlag);
sprintf(tmp_char, "%ld", (long)get_uptime());
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "uptime", std::string(tmp_char), qos, mqtt_controll_config.retainFlag);
sprintf(tmp_char, "%d", get_WIFI_RSSI());
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "wifiRSSI", std::string(tmp_char), qos, retainFlag);
sprintf(tmp_char, "%lu", (long)get_heapsize());
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "freeMem", std::string(tmp_char), qos, mqtt_controll_config.retainFlag);
sprintf(tmp_char, "%d", (int)temperatureRead());
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "CPUtemp", std::string(tmp_char), qos, retainFlag);
sprintf(tmp_char, "%d", get_wifi_rssi());
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "wifiRSSI", std::string(tmp_char), qos, mqtt_controll_config.retainFlag);
sprintf(tmp_char, "%d", (int)read_tempsensor());
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "CPUtemp", std::string(tmp_char), qos, mqtt_controll_config.retainFlag);
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Successfully published all System MQTT topics");
int aFreeInternalHeapSizeAfter = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
int aMinFreeInternalHeapSize = heap_caps_get_minimum_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
int aFreeInternalHeapSizeAfter = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
int aMinFreeInternalHeapSize = heap_caps_get_minimum_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Int. Heap Usage before publishing System Topics: " +
to_string(aFreeInternalHeapSizeBefore) + ", after: " + to_string(aFreeInternalHeapSizeAfter) + ", delta: " +
to_string(aFreeInternalHeapSizeBefore - aFreeInternalHeapSizeAfter) + ", lowest free: " + to_string(aMinFreeInternalHeapSize));
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Int. Heap Usage before publishing System Topics: " + to_string(aFreeInternalHeapSizeBefore) + ", after: " + to_string(aFreeInternalHeapSizeAfter) + ", delta: " + to_string(aFreeInternalHeapSizeBefore - aFreeInternalHeapSizeAfter) + ", lowest free: " + to_string(aMinFreeInternalHeapSize));
return allSendsSuccessed;
}
bool publishStaticData(int qos) {
bool allSendsSuccessed = false;
if (!getMQTTisConnected()) {
bool publishStaticData(int qos)
{
if (!getMQTTisConnected())
{
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "Unable to send Static Topics, we are not connected to the MQTT broker!");
return false;
}
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Publishing static MQTT topics...");
int aFreeInternalHeapSizeBefore = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
bool allSendsSuccessed = false;
int aFreeInternalHeapSizeBefore = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "fwVersion", getFwVersion().c_str(), qos, retainFlag);
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "MAC", getMac(), qos, retainFlag);
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "IP", *getIPAddress(), qos, retainFlag);
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "hostname", wlan_config.hostname, qos, retainFlag);
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "fwVersion", getFwVersion().c_str(), qos, mqtt_controll_config.retainFlag);
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "MAC", get_mac(), qos, mqtt_controll_config.retainFlag);
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "IP", network_config.ipaddress, qos, mqtt_controll_config.retainFlag);
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "hostname", network_config.hostname, qos, mqtt_controll_config.retainFlag);
std::stringstream stream;
stream << std::fixed << std::setprecision(1) << roundInterval; // minutes
allSendsSuccessed |= MQTTPublish(maintopic + "/" + "interval", stream.str(), qos, retainFlag);
stream << std::fixed << std::setprecision(1) << mqtt_controll_config.roundInterval; // minutes
allSendsSuccessed |= MQTTPublish(mqtt_controll_config.maintopic + "/" + "interval", stream.str(), qos, mqtt_controll_config.retainFlag);
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Successfully published all Static MQTT topics");
int aFreeInternalHeapSizeAfter = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
int aMinFreeInternalHeapSize = heap_caps_get_minimum_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
int aFreeInternalHeapSizeAfter = heap_caps_get_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
int aMinFreeInternalHeapSize = heap_caps_get_minimum_free_size(MALLOC_CAP_8BIT | MALLOC_CAP_INTERNAL);
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Int. Heap Usage before Publishing Static Topics: " +
to_string(aFreeInternalHeapSizeBefore) + ", after: " + to_string(aFreeInternalHeapSizeAfter) + ", delta: " +
to_string(aFreeInternalHeapSizeBefore - aFreeInternalHeapSizeAfter) + ", lowest free: " + to_string(aMinFreeInternalHeapSize));
LogFile.WriteToFile(ESP_LOG_DEBUG, TAG, "Int. Heap Usage before Publishing Static Topics: " + to_string(aFreeInternalHeapSizeBefore) + ", after: " + to_string(aFreeInternalHeapSizeAfter) + ", delta: " + to_string(aFreeInternalHeapSizeBefore - aFreeInternalHeapSizeAfter) + ", lowest free: " + to_string(aMinFreeInternalHeapSize));
return allSendsSuccessed;
}
esp_err_t scheduleSendingDiscovery_and_static_Topics(httpd_req_t *req) {
esp_err_t scheduleSendingDiscovery_and_static_Topics(httpd_req_t *req)
{
sendingOf_DiscoveryAndStaticTopics_scheduled = true;
char msg[] = "MQTT Homeassistant Discovery and Static Topics scheduled";
httpd_resp_send(req, msg, strlen(msg));
httpd_resp_send(req, msg, strlen(msg));
return ESP_OK;
}
esp_err_t sendDiscovery_and_static_Topics(void) {
bool success = false;
if (!sendingOf_DiscoveryAndStaticTopics_scheduled) {
esp_err_t sendDiscovery_and_static_Topics(void)
{
if (!sendingOf_DiscoveryAndStaticTopics_scheduled)
{
// Flag not set, nothing to do
return ESP_OK;
}
if (HomeassistantDiscovery) {
bool success = false;
if (mqtt_controll_config.HomeAssistantDiscovery)
{
success = MQTThomeassistantDiscovery(1);
}
success |= publishStaticData(1);
if (success) { // Success, clear the flag
if (success)
{
// Success, clear the flag
sendingOf_DiscoveryAndStaticTopics_scheduled = false;
return ESP_OK;
}
else {
else
{
LogFile.WriteToFile(ESP_LOG_WARN, TAG, "One or more MQTT topics failed to be published, will try sending them in the next round!");
/* Keep sendingOf_DiscoveryAndStaticTopics_scheduled set so we can retry after the next round */
return ESP_FAIL;
}
}
void GotConnected(std::string maintopic, bool retainFlag) {
void GotConnected(std::string maintopic, bool retainFlag)
{
// Nothing to do
}
void register_server_mqtt_uri(httpd_handle_t server) {
httpd_uri_t uri = { };
uri.method = HTTP_GET;
std::string mqttServer_getMainTopic()
{
return mqtt_controll_config.maintopic;
}
uri.uri = "/mqtt_publish_discovery";
void mqtt_register_uri(httpd_handle_t server)
{
httpd_uri_t uri = {};
uri.method = HTTP_GET;
uri.uri = "/mqtt_publish_discovery";
uri.handler = APPLY_BASIC_AUTH_FILTER(scheduleSendingDiscovery_and_static_Topics);
uri.user_ctx = (void*) "";
httpd_register_uri_handler(server, &uri);
uri.user_ctx = (void *)"";
httpd_register_uri_handler(server, &uri);
}
std::string getTimeUnit(void) {
return timeUnit;
}
void SetHomeassistantDiscoveryEnabled(bool enabled) {
HomeassistantDiscovery = enabled;
}
void setMqtt_Server_Retain(bool _retainFlag) {
retainFlag = _retainFlag;
}
void mqttServer_setMainTopic( std::string _maintopic) {
maintopic = _maintopic;
}
std::string mqttServer_getMainTopic() {
return maintopic;
}
void mqttServer_setDmoticzInTopic( std::string _domoticzintopic) {
domoticzintopic = _domoticzintopic;
}
#endif //ENABLE_MQTT

24
code/components/jomjol_mqtt/server_mqtt.h
View File

@@ -1,31 +1,19 @@
#ifdef ENABLE_MQTT
#pragma once
#ifndef SERVERMQTT_H
#define SERVERMQTT_H
#ifndef SERVER_MQTT_H
#define SERVER_MQTT_H
#include "ClassFlowDefineTypes.h"
void SetHomeassistantDiscoveryEnabled(bool enabled);
void mqttServer_setParameter(std::vector<NumberPost*>* _NUMBERS, int interval, float roundInterval);
void mqttServer_setMeterType(std::string meterType, std::string valueUnit, std::string timeUnit,std::string rateUnit);
void setMqtt_Server_Retain(bool SetRetainFlag);
void mqttServer_setMainTopic( std::string maintopic);
void mqttServer_setDmoticzInTopic( std::string domoticzintopic);
void mqttServer_setParameter(std::vector<NumberPost *> *_NUMBERS);
std::string mqttServer_getMainTopic();
void register_server_mqtt_uri(httpd_handle_t server);
bool publishSystemData(int qos);
std::string getTimeUnit(void);
void GotConnected(std::string maintopic, bool SetRetainFlag);
esp_err_t sendDiscovery_and_static_Topics(void);
std::string createNodeId(std::string &topic);
#endif //SERVERMQTT_H
#endif //ENABLE_MQTT
void mqtt_register_uri(httpd_handle_t server);
#endif // SERVER_MQTT_H