exodus/Leo-ElixirLights
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature: Including MQTT client

Browse Source
This commit is contained in:
Jakob Gegeniger
2025-01-16 22:44:55 +01:00
parent 968d078c07
commit 5fcc0dda1e
8 changed files with 383 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/sockets/CMakeLists.txt
+1 -1
View File
@@ -1,5 +1,5 @@
idf_component_register(
SRC_DIRS "src"
INCLUDE_DIRS "include"
REQUIRES esp_netif lwip project_bases freeRtosApi
REQUIRES esp_netif lwip project_bases freeRtosApi mqtt
)
components/sockets/include/mqttClient.hpp
+65
View File
@@ -0,0 +1,65 @@
/**
* @file mqttClient.hpp
* @author Gustice
* @brief Wrapper class for TCP/IP based MQTT APIs
*
* @copyright Copyright (c) 2024
*/
#pragma once
#include "esp_event.h"
#include "esp_system.h"
#include "mqtt_client.h"
#include <functional>
#include <string>
#include <string_view>
/**
* @brief MqttClient wrapper (Singleton)
*/
class MqttClient {
public:
/// @brief Init and mount file system
/// @param broker URI to web MQTT broker server
static MqttClient &init(std::string broker, std::string prefix);
void addSubscription(std::string topic, std::function<void(std::string)> call);
void publish(std::string topic, std::string data);
void start();
bool isRunning() { return _state != State::Init; }
/// @brief Get instance, ensure initialization
/// @param broker URI to web MQTT broker server
/// @returnb
static MqttClient &getInstance();
// Disable move and assignment
MqttClient(MqttClient &other) = delete;
void operator=(const MqttClient &) = delete;
// protected:
MqttClient(std::string broker, std::string pefix);
~MqttClient();
private:
enum class State {
Init,
Started,
Disconnected,
Connected
};
struct Subscription {
std::string topic;
std::function<void(std::string)> callback;
bool active{};
int id{};
};
std::string _broker;
std::string _prefix;
State _state{};
static MqttClient *_instance;
esp_mqtt_client_handle_t _client;
std::vector<Subscription> _subscriptions;
static void eventHandler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data);
};
components/sockets/include/udpSockets.hpp
+5
View File
@@ -269,6 +269,9 @@ class Syslog {
if (_Socket == nullptr) {
return;
}
// if (severity > _severityThresh) {
// return;
// }
char buffer[512];
// V T H A P M D Msg
@@ -286,6 +289,7 @@ class Syslog {
if (_Socket) {
return;
}
ESP_LOGI("UDP", "Starting syslog session to %s", target.getIp().c_str());
_Facility = Facility::Syslog;
_HostName = "ESP";
_AppName = "App";
@@ -301,4 +305,5 @@ class Syslog {
static std::string _AppName;
static UdpSocket *_Socket;
static IpAdr _Target;
static Severity _severityThresh;
};
components/sockets/src/mqttClient.cpp
+150
View File
@@ -0,0 +1,150 @@
/**
* @file mqttClient.cpp
* @author Gustice
* @brief Wrapper class for TCP/IP based MQTT APIs
*
* @copyright Copyright (c) 2024
*/
#include "mqttClient.hpp"
#include "esp_log.h"
#include "mqtt_client.h"
#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstring>
MqttClient *MqttClient::_instance = nullptr;
const char *TAG = "Mqtt";
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);
}
}
void MqttClient::eventHandler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data) {
ESP_LOGI(TAG, "Event dispatched from event loop base=%s, event_id=%" PRIi32 "", base, event_id);
esp_mqtt_event_handle_t event = (esp_mqtt_event_handle_t)event_data;
esp_mqtt_client_handle_t client = event->client;
MqttClient &instance = *((MqttClient *)handler_args);
int msg_id;
switch ((esp_mqtt_event_id_t)event_id) {
case MQTT_EVENT_CONNECTED:
instance._state = State::Connected;
ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");
for (auto &s : instance._subscriptions) {
msg_id = esp_mqtt_client_subscribe(client, s.topic.c_str(), 1);
ESP_LOGI(TAG, "sent subscribe successful: topic='%s', msg_id=%d", s.topic.c_str(), msg_id);
}
break;
case MQTT_EVENT_DISCONNECTED:
instance._state = State::Disconnected;
ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
break;
case MQTT_EVENT_SUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_PUBLISHED:
ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_DATA: {
bool found = false;
std::string topic(event->topic, event->topic_len);
for (auto &s : instance._subscriptions) {
if (s.topic == topic) {
found = true;
s.callback(std::string(event->data, event->data_len));
}
}
if (!found) {
ESP_LOGW(TAG, "Unrecognized Topic");
printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
printf("DATA=%.*s\r\n", event->data_len, event->data);
}
} break;
case MQTT_EVENT_ERROR:
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));
}
break;
default:
ESP_LOGI(TAG, "Other event id:%d", event->event_id);
break;
}
}
MqttClient::MqttClient(std::string broker, std::string prefix)
: _broker("mqtt://" + broker), _prefix(prefix) {
// esp_log_level_set("*", ESP_LOG_INFO);
// esp_log_level_set(TAG, ESP_LOG_VERBOSE);
// esp_log_level_set("mqtt_client", ESP_LOG_VERBOSE);
// esp_log_level_set("transport_base", ESP_LOG_VERBOSE);
// esp_log_level_set("esp-tls", ESP_LOG_VERBOSE);
// esp_log_level_set("transport", ESP_LOG_VERBOSE);
// esp_log_level_set("outbox", ESP_LOG_VERBOSE);
esp_mqtt_client_config_t mqtt_cfg{};
mqtt_cfg.broker.address.uri = _broker.c_str();
_client = esp_mqtt_client_init(&mqtt_cfg);
/* The last argument may be used to pass data to the event handler, in this example mqtt_event_handler */
esp_mqtt_client_register_event(_client, (esp_mqtt_event_id_t)ESP_EVENT_ANY_ID, eventHandler, this);
}
MqttClient::~MqttClient() {
for (auto &s : _subscriptions) {
auto msg_id = esp_mqtt_client_unsubscribe(_client, s.topic.c_str());
ESP_LOGI(TAG, "sent unsubscribe successful, msg_id=%d", msg_id);
}
esp_mqtt_client_disconnect(_client);
}
MqttClient &MqttClient::getInstance() {
if (_instance == nullptr) {
throw "No instance created";
}
return *_instance;
}
MqttClient &MqttClient::init(std::string broker, std::string prefix) {
if (_instance != nullptr) {
return *_instance;
}
_instance = new MqttClient(broker, prefix);
return *_instance;
}
void MqttClient::addSubscription(std::string topic, std::function<void(std::string)> call) {
std::string fullTopic = _prefix + "/" + topic;
_subscriptions.emplace_back(fullTopic, call, false, 0);
// ToDo ensure each topic is only present once
}
void MqttClient::start() {
ESP_LOGI(TAG, "Starting MQTT client");
_state = State::Started;
esp_mqtt_client_start(_client);
}
void MqttClient::publish(std::string topic, std::string data) {
if (_state != State::Connected) {
ESP_LOGW(TAG, "not connected, cannot publish s=%i", (int)_state);
return;
}
std::string fullTopic = _prefix + "/" + topic;
int msg_id = esp_mqtt_client_publish(_client, fullTopic.c_str(), data.c_str(), data.size(), 1, 0);
ESP_LOGI(TAG, "sent publish of topic '%s' successful, msg_id=%d", fullTopic.c_str(), msg_id);
}
components/sockets/src/udpSocekts.cpp
+1
View File
@@ -13,6 +13,7 @@ Syslog::Facility Syslog::_Facility;
std::string Syslog::_HostName;
std::string Syslog::_AppName;
IpAdr Syslog::_Target(INADDR_ANY, 0);
Syslog::Severity Syslog::_severityThresh = Syslog::Severity::Informational;
void Syslog::SyslogTask() {
ESP_LOGI("SysLog", "Starting Syslog Task");