andry/EMS-ESP32
1
0
Fork 0
mirror of https://github.com/emsesp/EMS-ESP32.git synced 2025-12-06 07:49:52 +03:00
Code Issues Packages Projects Releases Wiki Activity

make standalone compile

Browse Source
This commit is contained in:
Proddy
2023-07-13 21:50:34 +02:00
parent 0afda87092
commit 9ff8d01e93
9 changed files with 189 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files

294
lib/espMqttClient/src/MqttClient.h
View File

@@ -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
};