mirror of
https://github.com/emsesp/EMS-ESP32.git
synced 2025-12-06 07:49:52 +03:00
Merge branch 'dev2' of https://github.com/emsesp/EMS-ESP32 into dev2
This commit is contained in:
MichaelDvP
@@ -24,167 +24,171 @@ the LICENSE file.
|
||||
#include "Transport/Transport.h"
|
||||
|
||||
class MqttClient {
|
||||
public:
|
||||
virtual ~MqttClient();
|
||||
bool connected() const;
|
||||
bool disconnected() const;
|
||||
bool connect();
|
||||
bool disconnect(bool force = false);
|
||||
template <typename... Args>
|
||||
uint16_t subscribe(const char* topic, uint8_t qos, Args&&... args) {
|
||||
uint16_t packetId = _getNextPacketId();
|
||||
if (_state != State::connected) {
|
||||
packetId = 0;
|
||||
} else {
|
||||
EMC_SEMAPHORE_TAKE();
|
||||
if (!_addPacket(packetId, topic, qos, std::forward<Args>(args) ...)) {
|
||||
emc_log_e("Could not create SUBSCRIBE packet");
|
||||
packetId = 0;
|
||||
}
|
||||
EMC_SEMAPHORE_GIVE();
|
||||
public:
|
||||
virtual ~MqttClient();
|
||||
bool connected() const;
|
||||
bool disconnected() const;
|
||||
bool connect();
|
||||
bool disconnect(bool force = false);
|
||||
template <typename... Args>
|
||||
uint16_t subscribe(const char * topic, uint8_t qos, Args &&... args) {
|
||||
uint16_t packetId = _getNextPacketId();
|
||||
if (_state != State::connected) {
|
||||
packetId = 0;
|
||||
} else {
|
||||
EMC_SEMAPHORE_TAKE();
|
||||
if (!_addPacket(packetId, topic, qos, std::forward<Args>(args)...)) {
|
||||
emc_log_e("Could not create SUBSCRIBE packet");
|
||||
packetId = 0;
|
||||
}
|
||||
EMC_SEMAPHORE_GIVE();
|
||||
}
|
||||
return packetId;
|
||||
}
|
||||
return packetId;
|
||||
}
|
||||
template <typename... Args>
|
||||
uint16_t unsubscribe(const char* topic, Args&&... args) {
|
||||
uint16_t packetId = _getNextPacketId();
|
||||
if (_state != State::connected) {
|
||||
packetId = 0;
|
||||
} else {
|
||||
EMC_SEMAPHORE_TAKE();
|
||||
if (!_addPacket(packetId, topic, std::forward<Args>(args) ...)) {
|
||||
emc_log_e("Could not create UNSUBSCRIBE packet");
|
||||
packetId = 0;
|
||||
}
|
||||
EMC_SEMAPHORE_GIVE();
|
||||
template <typename... Args>
|
||||
uint16_t unsubscribe(const char * topic, Args &&... args) {
|
||||
uint16_t packetId = _getNextPacketId();
|
||||
if (_state != State::connected) {
|
||||
packetId = 0;
|
||||
} else {
|
||||
EMC_SEMAPHORE_TAKE();
|
||||
if (!_addPacket(packetId, topic, std::forward<Args>(args)...)) {
|
||||
emc_log_e("Could not create UNSUBSCRIBE packet");
|
||||
packetId = 0;
|
||||
}
|
||||
EMC_SEMAPHORE_GIVE();
|
||||
}
|
||||
return packetId;
|
||||
}
|
||||
return packetId;
|
||||
}
|
||||
uint16_t publish(const char* topic, uint8_t qos, bool retain, const uint8_t* payload, size_t length);
|
||||
uint16_t publish(const char* topic, uint8_t qos, bool retain, const char* payload);
|
||||
uint16_t publish(const char* topic, uint8_t qos, bool retain, espMqttClientTypes::PayloadCallback callback, size_t length);
|
||||
void clearQueue(bool deleteSessionData = false); // Not MQTT compliant and may cause unpredictable results when `deleteSessionData` = true!
|
||||
const char* getClientId() const;
|
||||
uint16_t getQueue() const;
|
||||
void loop();
|
||||
uint16_t publish(const char * topic, uint8_t qos, bool retain, const uint8_t * payload, size_t length);
|
||||
uint16_t publish(const char * topic, uint8_t qos, bool retain, const char * payload);
|
||||
uint16_t publish(const char * topic, uint8_t qos, bool retain, espMqttClientTypes::PayloadCallback callback, size_t length);
|
||||
void clearQueue(bool deleteSessionData = false); // Not MQTT compliant and may cause unpredictable results when `deleteSessionData` = true!
|
||||
const char * getClientId() const;
|
||||
uint16_t getQueue() const;
|
||||
void loop();
|
||||
|
||||
protected:
|
||||
explicit MqttClient(espMqttClientTypes::UseInternalTask useInternalTask, uint8_t priority = 1, uint8_t core = 1);
|
||||
espMqttClientTypes::UseInternalTask _useInternalTask;
|
||||
espMqttClientInternals::Transport* _transport;
|
||||
protected:
|
||||
explicit MqttClient(espMqttClientTypes::UseInternalTask useInternalTask, uint8_t priority = 1, uint8_t core = 1);
|
||||
espMqttClientTypes::UseInternalTask _useInternalTask;
|
||||
espMqttClientInternals::Transport * _transport;
|
||||
|
||||
espMqttClientTypes::OnConnectCallback _onConnectCallback;
|
||||
espMqttClientTypes::OnDisconnectCallback _onDisconnectCallback;
|
||||
espMqttClientTypes::OnSubscribeCallback _onSubscribeCallback;
|
||||
espMqttClientTypes::OnUnsubscribeCallback _onUnsubscribeCallback;
|
||||
espMqttClientTypes::OnMessageCallback _onMessageCallback;
|
||||
espMqttClientTypes::OnPublishCallback _onPublishCallback;
|
||||
espMqttClientTypes::OnErrorCallback _onErrorCallback;
|
||||
typedef void(*mqttClientHook)(void*);
|
||||
const char* _clientId;
|
||||
IPAddress _ip;
|
||||
const char* _host;
|
||||
uint16_t _port;
|
||||
bool _useIp;
|
||||
uint32_t _keepAlive;
|
||||
bool _cleanSession;
|
||||
const char* _username;
|
||||
const char* _password;
|
||||
const char* _willTopic;
|
||||
const uint8_t* _willPayload;
|
||||
uint16_t _willPayloadLength;
|
||||
uint8_t _willQos;
|
||||
bool _willRetain;
|
||||
uint32_t _timeout;
|
||||
espMqttClientTypes::OnConnectCallback _onConnectCallback;
|
||||
espMqttClientTypes::OnDisconnectCallback _onDisconnectCallback;
|
||||
espMqttClientTypes::OnSubscribeCallback _onSubscribeCallback;
|
||||
espMqttClientTypes::OnUnsubscribeCallback _onUnsubscribeCallback;
|
||||
espMqttClientTypes::OnMessageCallback _onMessageCallback;
|
||||
espMqttClientTypes::OnPublishCallback _onPublishCallback;
|
||||
espMqttClientTypes::OnErrorCallback _onErrorCallback;
|
||||
typedef void (*mqttClientHook)(void *);
|
||||
const char * _clientId;
|
||||
IPAddress _ip;
|
||||
const char * _host;
|
||||
uint16_t _port;
|
||||
bool _useIp;
|
||||
uint32_t _keepAlive;
|
||||
bool _cleanSession;
|
||||
const char * _username;
|
||||
const char * _password;
|
||||
const char * _willTopic;
|
||||
const uint8_t * _willPayload;
|
||||
uint16_t _willPayloadLength;
|
||||
uint8_t _willQos;
|
||||
bool _willRetain;
|
||||
uint32_t _timeout;
|
||||
|
||||
// state is protected to allow state changes by the transport system, defined in child classes
|
||||
// eg. to allow AsyncTCP
|
||||
enum class State {
|
||||
disconnected = 0,
|
||||
connectingTcp1 = 1,
|
||||
connectingTcp2 = 2,
|
||||
connectingMqtt = 3,
|
||||
connected = 4,
|
||||
disconnectingMqtt1 = 5,
|
||||
disconnectingMqtt2 = 6,
|
||||
disconnectingTcp1 = 7,
|
||||
disconnectingTcp2 = 8
|
||||
};
|
||||
std::atomic<State> _state;
|
||||
// state is protected to allow state changes by the transport system, defined in child classes
|
||||
// eg. to allow AsyncTCP
|
||||
enum class State {
|
||||
disconnected = 0,
|
||||
connectingTcp1 = 1,
|
||||
connectingTcp2 = 2,
|
||||
connectingMqtt = 3,
|
||||
connected = 4,
|
||||
disconnectingMqtt1 = 5,
|
||||
disconnectingMqtt2 = 6,
|
||||
disconnectingTcp1 = 7,
|
||||
disconnectingTcp2 = 8
|
||||
};
|
||||
std::atomic<State> _state;
|
||||
|
||||
private:
|
||||
char _generatedClientId[EMC_CLIENTID_LENGTH];
|
||||
uint16_t _packetId;
|
||||
private:
|
||||
char _generatedClientId[EMC_CLIENTID_LENGTH];
|
||||
uint16_t _packetId;
|
||||
|
||||
#if defined(ARDUINO_ARCH_ESP32)
|
||||
SemaphoreHandle_t _xSemaphore;
|
||||
TaskHandle_t _taskHandle;
|
||||
static void _loop(MqttClient* c);
|
||||
SemaphoreHandle_t _xSemaphore;
|
||||
TaskHandle_t _taskHandle;
|
||||
static void _loop(MqttClient * c);
|
||||
#elif defined(ARDUINO_ARCH_ESP8266) && EMC_ESP8266_MULTITHREADING
|
||||
std::atomic<bool> _xSemaphore = false;
|
||||
std::atomic<bool> _xSemaphore = false;
|
||||
#elif defined(__linux__)
|
||||
std::mutex mtx;
|
||||
// added mutable to compile EMS-ESP standalone
|
||||
mutable std::mutex mtx;
|
||||
#endif
|
||||
|
||||
uint8_t _rxBuffer[EMC_RX_BUFFER_SIZE];
|
||||
struct OutgoingPacket {
|
||||
uint32_t timeSent;
|
||||
espMqttClientInternals::Packet packet;
|
||||
uint8_t _rxBuffer[EMC_RX_BUFFER_SIZE];
|
||||
struct OutgoingPacket {
|
||||
uint32_t timeSent;
|
||||
espMqttClientInternals::Packet packet;
|
||||
template <typename... Args>
|
||||
OutgoingPacket(uint32_t t, espMqttClientTypes::Error error, Args &&... args)
|
||||
: timeSent(t)
|
||||
, packet(error, std::forward<Args>(args)...) {
|
||||
}
|
||||
};
|
||||
espMqttClientInternals::Outbox<OutgoingPacket> _outbox;
|
||||
size_t _bytesSent;
|
||||
espMqttClientInternals::Parser _parser;
|
||||
uint32_t _lastClientActivity;
|
||||
uint32_t _lastServerActivity;
|
||||
bool _pingSent;
|
||||
espMqttClientTypes::DisconnectReason _disconnectReason;
|
||||
|
||||
uint16_t _getNextPacketId();
|
||||
|
||||
template <typename... Args>
|
||||
OutgoingPacket(uint32_t t, espMqttClientTypes::Error error, Args&&... args) :
|
||||
timeSent(t),
|
||||
packet(error, std::forward<Args>(args) ...) {}
|
||||
};
|
||||
espMqttClientInternals::Outbox<OutgoingPacket> _outbox;
|
||||
size_t _bytesSent;
|
||||
espMqttClientInternals::Parser _parser;
|
||||
uint32_t _lastClientActivity;
|
||||
uint32_t _lastServerActivity;
|
||||
bool _pingSent;
|
||||
espMqttClientTypes::DisconnectReason _disconnectReason;
|
||||
bool _addPacket(Args &&... args) {
|
||||
espMqttClientTypes::Error error(espMqttClientTypes::Error::SUCCESS);
|
||||
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.emplace(0, error, std::forward<Args>(args)...);
|
||||
if (it && error == espMqttClientTypes::Error::SUCCESS)
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
uint16_t _getNextPacketId();
|
||||
template <typename... Args>
|
||||
bool _addPacketFront(Args &&... args) {
|
||||
espMqttClientTypes::Error error(espMqttClientTypes::Error::SUCCESS);
|
||||
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.emplaceFront(0, error, std::forward<Args>(args)...);
|
||||
if (it && error == espMqttClientTypes::Error::SUCCESS)
|
||||
return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
template <typename... Args>
|
||||
bool _addPacket(Args&&... args) {
|
||||
espMqttClientTypes::Error error(espMqttClientTypes::Error::SUCCESS);
|
||||
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.emplace(0, error, std::forward<Args>(args) ...);
|
||||
if (it && error == espMqttClientTypes::Error::SUCCESS) return true;
|
||||
return false;
|
||||
}
|
||||
void _checkOutbox();
|
||||
int _sendPacket();
|
||||
bool _advanceOutbox();
|
||||
void _checkIncoming();
|
||||
void _checkPing();
|
||||
void _checkTimeout();
|
||||
|
||||
template <typename... Args>
|
||||
bool _addPacketFront(Args&&... args) {
|
||||
espMqttClientTypes::Error error(espMqttClientTypes::Error::SUCCESS);
|
||||
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.emplaceFront(0, error, std::forward<Args>(args) ...);
|
||||
if (it && error == espMqttClientTypes::Error::SUCCESS) return true;
|
||||
return false;
|
||||
}
|
||||
void _onConnack();
|
||||
void _onPublish();
|
||||
void _onPuback();
|
||||
void _onPubrec();
|
||||
void _onPubrel();
|
||||
void _onPubcomp();
|
||||
void _onSuback();
|
||||
void _onUnsuback();
|
||||
|
||||
void _checkOutbox();
|
||||
int _sendPacket();
|
||||
bool _advanceOutbox();
|
||||
void _checkIncoming();
|
||||
void _checkPing();
|
||||
void _checkTimeout();
|
||||
void _clearQueue(int clearData); // 0: keep session,
|
||||
// 1: keep only PUBLISH qos > 0
|
||||
// 2: delete all
|
||||
void _onError(uint16_t packetId, espMqttClientTypes::Error error);
|
||||
|
||||
void _onConnack();
|
||||
void _onPublish();
|
||||
void _onPuback();
|
||||
void _onPubrec();
|
||||
void _onPubrel();
|
||||
void _onPubcomp();
|
||||
void _onSuback();
|
||||
void _onUnsuback();
|
||||
|
||||
void _clearQueue(int clearData); // 0: keep session,
|
||||
// 1: keep only PUBLISH qos > 0
|
||||
// 2: delete all
|
||||
void _onError(uint16_t packetId, espMqttClientTypes::Error error);
|
||||
|
||||
#if defined(ARDUINO_ARCH_ESP32)
|
||||
#if ARDUHAL_LOG_LEVEL >= ARDUHAL_LOG_LEVEL_INFO
|
||||
size_t _highWaterMark;
|
||||
#endif
|
||||
#endif
|
||||
#if defined(ARDUINO_ARCH_ESP32)
|
||||
#if ARDUHAL_LOG_LEVEL >= ARDUHAL_LOG_LEVEL_INFO
|
||||
size_t _highWaterMark;
|
||||
#endif
|
||||
#endif
|
||||
};
|
||||
|
||||
@@ -172,14 +172,14 @@ void MqttSettingsService::WiFiEvent(WiFiEvent_t event, WiFiEventInfo_t info) {
|
||||
}
|
||||
|
||||
bool MqttSettingsService::configureMqtt() {
|
||||
// disconnect if connected
|
||||
// disconnect if already connected
|
||||
if (_mqttClient->connected()) {
|
||||
emsesp::EMSESP::logger().info("Disconneting to configure");
|
||||
emsesp::EMSESP::logger().info("Disconnecting to configure");
|
||||
_mqttClient->disconnect(true);
|
||||
}
|
||||
|
||||
// only connect if WiFi is connected and MQTT is enabled
|
||||
if (_state.enabled && emsesp::EMSESP::system_.network_connected() && !_state.host.isEmpty()) {
|
||||
// if (_state.enabled && !_state.host.isEmpty()) {
|
||||
_reconfigureMqtt = false;
|
||||
#if CONFIG_IDF_TARGET_ESP32S3
|
||||
if (_state.rootCA.length() > 0) {
|
||||
@@ -212,6 +212,7 @@ bool MqttSettingsService::configureMqtt() {
|
||||
static_cast<espMqttClient *>(_mqttClient)->setCleanSession(_state.cleanSession);
|
||||
return _mqttClient->connect();
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user