Update AsyncMqttClient
This commit is contained in:
cschwinne
parent
5f606bb0b7
commit
47d2218efd
14
.vscode/tasks.json
vendored
14
.vscode/tasks.json
vendored
@ -9,8 +9,8 @@
|
||||
],
|
||||
"dependsOrder": "sequence",
|
||||
"problemMatcher": [
|
||||
"$platformio",
|
||||
],
|
||||
"$platformio"
|
||||
]
|
||||
},
|
||||
{
|
||||
"type": "PlatformIO",
|
||||
@ -18,7 +18,7 @@
|
||||
"task": "Build",
|
||||
"group": {
|
||||
"kind": "build",
|
||||
"isDefault": true,
|
||||
"isDefault": true
|
||||
},
|
||||
"problemMatcher": [
|
||||
"$platformio"
|
||||
@ -37,6 +37,14 @@
|
||||
"presentation": {
|
||||
"panel": "shared"
|
||||
}
|
||||
},
|
||||
{
|
||||
"type": "PlatformIO",
|
||||
"task": "Verbose Build",
|
||||
"problemMatcher": [
|
||||
"$platformio"
|
||||
],
|
||||
"label": "PlatformIO: Verbose Build"
|
||||
}
|
||||
]
|
||||
}
|
||||
@ -164,6 +164,7 @@ lib_deps =
|
||||
fastled/FastLED @ 3.5.0
|
||||
IRremoteESP8266 @ 2.8.2
|
||||
https://github.com/Aircoookie/ESPAsyncWebServer.git @ ~2.0.5
|
||||
marvinroger/AsyncMqttClient@^0.9.0
|
||||
#For use of the TTGO T-Display ESP32 Module with integrated TFT display uncomment the following line
|
||||
#TFT_eSPI
|
||||
#For use SSD1306 OLED display uncomment following
|
||||
|
||||
@ -1,877 +0,0 @@
|
||||
#include "AsyncMqttClient.hpp"
|
||||
|
||||
AsyncMqttClient::AsyncMqttClient()
|
||||
: _connected(false)
|
||||
, _connectPacketNotEnoughSpace(false)
|
||||
, _disconnectFlagged(false)
|
||||
, _tlsBadFingerprint(false)
|
||||
, _lastClientActivity(0)
|
||||
, _lastServerActivity(0)
|
||||
, _lastPingRequestTime(0)
|
||||
, _host(nullptr)
|
||||
, _useIp(false)
|
||||
#if ASYNC_TCP_SSL_ENABLED
|
||||
, _secure(false)
|
||||
#endif
|
||||
, _port(0)
|
||||
, _keepAlive(15)
|
||||
, _cleanSession(true)
|
||||
, _clientId(nullptr)
|
||||
, _username(nullptr)
|
||||
, _password(nullptr)
|
||||
, _willTopic(nullptr)
|
||||
, _willPayload(nullptr)
|
||||
, _willPayloadLength(0)
|
||||
, _willQos(0)
|
||||
, _willRetain(false)
|
||||
, _parsingInformation { .bufferState = AsyncMqttClientInternals::BufferState::NONE }
|
||||
, _currentParsedPacket(nullptr)
|
||||
, _remainingLengthBufferPosition(0)
|
||||
, _nextPacketId(1) {
|
||||
_client.onConnect([](void* obj, AsyncClient* c) { (static_cast<AsyncMqttClient*>(obj))->_onConnect(c); }, this);
|
||||
_client.onDisconnect([](void* obj, AsyncClient* c) { (static_cast<AsyncMqttClient*>(obj))->_onDisconnect(c); }, this);
|
||||
_client.onError([](void* obj, AsyncClient* c, int8_t error) { (static_cast<AsyncMqttClient*>(obj))->_onError(c, error); }, this);
|
||||
_client.onTimeout([](void* obj, AsyncClient* c, uint32_t time) { (static_cast<AsyncMqttClient*>(obj))->_onTimeout(c, time); }, this);
|
||||
_client.onAck([](void* obj, AsyncClient* c, size_t len, uint32_t time) { (static_cast<AsyncMqttClient*>(obj))->_onAck(c, len, time); }, this);
|
||||
_client.onData([](void* obj, AsyncClient* c, void* data, size_t len) { (static_cast<AsyncMqttClient*>(obj))->_onData(c, static_cast<char*>(data), len); }, this);
|
||||
_client.onPoll([](void* obj, AsyncClient* c) { (static_cast<AsyncMqttClient*>(obj))->_onPoll(c); }, this);
|
||||
|
||||
#ifdef ESP32
|
||||
sprintf(_generatedClientId, "esp32%06x", (uint32_t)ESP.getEfuseMac());
|
||||
_xSemaphore = xSemaphoreCreateMutex();
|
||||
#elif defined(ESP8266)
|
||||
sprintf(_generatedClientId, "esp8266%06x", (uint32_t)ESP.getChipId());
|
||||
#endif
|
||||
_clientId = _generatedClientId;
|
||||
|
||||
setMaxTopicLength(128);
|
||||
}
|
||||
|
||||
AsyncMqttClient::~AsyncMqttClient() {
|
||||
delete _currentParsedPacket;
|
||||
delete[] _parsingInformation.topicBuffer;
|
||||
#ifdef ESP32
|
||||
vSemaphoreDelete(_xSemaphore);
|
||||
#endif
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::setKeepAlive(uint16_t keepAlive) {
|
||||
_keepAlive = keepAlive;
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::setClientId(const char* clientId) {
|
||||
_clientId = clientId;
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::setCleanSession(bool cleanSession) {
|
||||
_cleanSession = cleanSession;
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::setMaxTopicLength(uint16_t maxTopicLength) {
|
||||
_parsingInformation.maxTopicLength = maxTopicLength;
|
||||
delete[] _parsingInformation.topicBuffer;
|
||||
_parsingInformation.topicBuffer = new char[maxTopicLength + 1];
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::setCredentials(const char* username, const char* password) {
|
||||
_username = username;
|
||||
_password = password;
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::setWill(const char* topic, uint8_t qos, bool retain, const char* payload, size_t length) {
|
||||
_willTopic = topic;
|
||||
_willQos = qos;
|
||||
_willRetain = retain;
|
||||
_willPayload = payload;
|
||||
_willPayloadLength = length;
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::setServer(IPAddress ip, uint16_t port) {
|
||||
_useIp = true;
|
||||
_ip = ip;
|
||||
_port = port;
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::setServer(const char* host, uint16_t port) {
|
||||
_useIp = false;
|
||||
_host = host;
|
||||
_port = port;
|
||||
return *this;
|
||||
}
|
||||
|
||||
#if ASYNC_TCP_SSL_ENABLED
|
||||
AsyncMqttClient& AsyncMqttClient::setSecure(bool secure) {
|
||||
_secure = secure;
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::addServerFingerprint(const uint8_t* fingerprint) {
|
||||
std::array<uint8_t, SHA1_SIZE> newFingerprint;
|
||||
memcpy(newFingerprint.data(), fingerprint, SHA1_SIZE);
|
||||
_secureServerFingerprints.push_back(newFingerprint);
|
||||
return *this;
|
||||
}
|
||||
#endif
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::onConnect(AsyncMqttClientInternals::OnConnectUserCallback callback) {
|
||||
_onConnectUserCallbacks.push_back(callback);
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::onDisconnect(AsyncMqttClientInternals::OnDisconnectUserCallback callback) {
|
||||
_onDisconnectUserCallbacks.push_back(callback);
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::onSubscribe(AsyncMqttClientInternals::OnSubscribeUserCallback callback) {
|
||||
_onSubscribeUserCallbacks.push_back(callback);
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::onUnsubscribe(AsyncMqttClientInternals::OnUnsubscribeUserCallback callback) {
|
||||
_onUnsubscribeUserCallbacks.push_back(callback);
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::onMessage(AsyncMqttClientInternals::OnMessageUserCallback callback) {
|
||||
_onMessageUserCallbacks.push_back(callback);
|
||||
return *this;
|
||||
}
|
||||
|
||||
AsyncMqttClient& AsyncMqttClient::onPublish(AsyncMqttClientInternals::OnPublishUserCallback callback) {
|
||||
_onPublishUserCallbacks.push_back(callback);
|
||||
return *this;
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_freeCurrentParsedPacket() {
|
||||
delete _currentParsedPacket;
|
||||
_currentParsedPacket = nullptr;
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_clear() {
|
||||
_lastPingRequestTime = 0;
|
||||
_connected = false;
|
||||
_disconnectFlagged = false;
|
||||
_connectPacketNotEnoughSpace = false;
|
||||
_tlsBadFingerprint = false;
|
||||
_freeCurrentParsedPacket();
|
||||
|
||||
_pendingPubRels.clear();
|
||||
_pendingPubRels.shrink_to_fit();
|
||||
|
||||
_toSendAcks.clear();
|
||||
_toSendAcks.shrink_to_fit();
|
||||
|
||||
_nextPacketId = 1;
|
||||
_parsingInformation.bufferState = AsyncMqttClientInternals::BufferState::NONE;
|
||||
}
|
||||
|
||||
/* TCP */
|
||||
void AsyncMqttClient::_onConnect(AsyncClient* client) {
|
||||
(void)client;
|
||||
|
||||
#if ASYNC_TCP_SSL_ENABLED
|
||||
if (_secure && _secureServerFingerprints.size() > 0) {
|
||||
SSL* clientSsl = _client.getSSL();
|
||||
|
||||
bool sslFoundFingerprint = false;
|
||||
for (std::array<uint8_t, SHA1_SIZE> fingerprint : _secureServerFingerprints) {
|
||||
if (ssl_match_fingerprint(clientSsl, fingerprint.data()) == SSL_OK) {
|
||||
sslFoundFingerprint = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (!sslFoundFingerprint) {
|
||||
_tlsBadFingerprint = true;
|
||||
_client.close(true);
|
||||
return;
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
char fixedHeader[5];
|
||||
fixedHeader[0] = AsyncMqttClientInternals::PacketType.CONNECT;
|
||||
fixedHeader[0] = fixedHeader[0] << 4;
|
||||
fixedHeader[0] = fixedHeader[0] | AsyncMqttClientInternals::HeaderFlag.CONNECT_RESERVED;
|
||||
|
||||
uint16_t protocolNameLength = 4;
|
||||
char protocolNameLengthBytes[2];
|
||||
protocolNameLengthBytes[0] = protocolNameLength >> 8;
|
||||
protocolNameLengthBytes[1] = protocolNameLength & 0xFF;
|
||||
|
||||
char protocolLevel[1];
|
||||
protocolLevel[0] = 0x04;
|
||||
|
||||
char connectFlags[1];
|
||||
connectFlags[0] = 0;
|
||||
if (_cleanSession) connectFlags[0] |= AsyncMqttClientInternals::ConnectFlag.CLEAN_SESSION;
|
||||
if (_username != nullptr) connectFlags[0] |= AsyncMqttClientInternals::ConnectFlag.USERNAME;
|
||||
if (_password != nullptr) connectFlags[0] |= AsyncMqttClientInternals::ConnectFlag.PASSWORD;
|
||||
if (_willTopic != nullptr) {
|
||||
connectFlags[0] |= AsyncMqttClientInternals::ConnectFlag.WILL;
|
||||
if (_willRetain) connectFlags[0] |= AsyncMqttClientInternals::ConnectFlag.WILL_RETAIN;
|
||||
switch (_willQos) {
|
||||
case 0:
|
||||
connectFlags[0] |= AsyncMqttClientInternals::ConnectFlag.WILL_QOS0;
|
||||
break;
|
||||
case 1:
|
||||
connectFlags[0] |= AsyncMqttClientInternals::ConnectFlag.WILL_QOS1;
|
||||
break;
|
||||
case 2:
|
||||
connectFlags[0] |= AsyncMqttClientInternals::ConnectFlag.WILL_QOS2;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
char keepAliveBytes[2];
|
||||
keepAliveBytes[0] = _keepAlive >> 8;
|
||||
keepAliveBytes[1] = _keepAlive & 0xFF;
|
||||
|
||||
uint16_t clientIdLength = strlen(_clientId);
|
||||
char clientIdLengthBytes[2];
|
||||
clientIdLengthBytes[0] = clientIdLength >> 8;
|
||||
clientIdLengthBytes[1] = clientIdLength & 0xFF;
|
||||
|
||||
// Optional fields
|
||||
uint16_t willTopicLength = 0;
|
||||
char willTopicLengthBytes[2];
|
||||
uint16_t willPayloadLength = _willPayloadLength;
|
||||
char willPayloadLengthBytes[2];
|
||||
if (_willTopic != nullptr) {
|
||||
willTopicLength = strlen(_willTopic);
|
||||
willTopicLengthBytes[0] = willTopicLength >> 8;
|
||||
willTopicLengthBytes[1] = willTopicLength & 0xFF;
|
||||
|
||||
if (_willPayload != nullptr && willPayloadLength == 0) willPayloadLength = strlen(_willPayload);
|
||||
|
||||
willPayloadLengthBytes[0] = willPayloadLength >> 8;
|
||||
willPayloadLengthBytes[1] = willPayloadLength & 0xFF;
|
||||
}
|
||||
|
||||
uint16_t usernameLength = 0;
|
||||
char usernameLengthBytes[2];
|
||||
if (_username != nullptr) {
|
||||
usernameLength = strlen(_username);
|
||||
usernameLengthBytes[0] = usernameLength >> 8;
|
||||
usernameLengthBytes[1] = usernameLength & 0xFF;
|
||||
}
|
||||
|
||||
uint16_t passwordLength = 0;
|
||||
char passwordLengthBytes[2];
|
||||
if (_password != nullptr) {
|
||||
passwordLength = strlen(_password);
|
||||
passwordLengthBytes[0] = passwordLength >> 8;
|
||||
passwordLengthBytes[1] = passwordLength & 0xFF;
|
||||
}
|
||||
|
||||
uint32_t remainingLength = 2 + protocolNameLength + 1 + 1 + 2 + 2 + clientIdLength; // always present
|
||||
if (_willTopic != nullptr) remainingLength += 2 + willTopicLength + 2 + willPayloadLength;
|
||||
if (_username != nullptr) remainingLength += 2 + usernameLength;
|
||||
if (_password != nullptr) remainingLength += 2 + passwordLength;
|
||||
uint8_t remainingLengthLength = AsyncMqttClientInternals::Helpers::encodeRemainingLength(remainingLength, fixedHeader + 1);
|
||||
|
||||
uint32_t neededSpace = 1 + remainingLengthLength;
|
||||
neededSpace += 2;
|
||||
neededSpace += protocolNameLength;
|
||||
neededSpace += 1;
|
||||
neededSpace += 1;
|
||||
neededSpace += 2;
|
||||
neededSpace += 2;
|
||||
neededSpace += clientIdLength;
|
||||
if (_willTopic != nullptr) {
|
||||
neededSpace += 2;
|
||||
neededSpace += willTopicLength;
|
||||
|
||||
neededSpace += 2;
|
||||
if (_willPayload != nullptr) neededSpace += willPayloadLength;
|
||||
}
|
||||
if (_username != nullptr) {
|
||||
neededSpace += 2;
|
||||
neededSpace += usernameLength;
|
||||
}
|
||||
if (_password != nullptr) {
|
||||
neededSpace += 2;
|
||||
neededSpace += passwordLength;
|
||||
}
|
||||
|
||||
SEMAPHORE_TAKE();
|
||||
if (_client.space() < neededSpace) {
|
||||
_connectPacketNotEnoughSpace = true;
|
||||
_client.close(true);
|
||||
SEMAPHORE_GIVE();
|
||||
return;
|
||||
}
|
||||
|
||||
_client.add(fixedHeader, 1 + remainingLengthLength);
|
||||
_client.add(protocolNameLengthBytes, 2);
|
||||
_client.add("MQTT", protocolNameLength);
|
||||
_client.add(protocolLevel, 1);
|
||||
_client.add(connectFlags, 1);
|
||||
_client.add(keepAliveBytes, 2);
|
||||
_client.add(clientIdLengthBytes, 2);
|
||||
_client.add(_clientId, clientIdLength);
|
||||
if (_willTopic != nullptr) {
|
||||
_client.add(willTopicLengthBytes, 2);
|
||||
_client.add(_willTopic, willTopicLength);
|
||||
|
||||
_client.add(willPayloadLengthBytes, 2);
|
||||
if (_willPayload != nullptr) _client.add(_willPayload, willPayloadLength);
|
||||
}
|
||||
if (_username != nullptr) {
|
||||
_client.add(usernameLengthBytes, 2);
|
||||
_client.add(_username, usernameLength);
|
||||
}
|
||||
if (_password != nullptr) {
|
||||
_client.add(passwordLengthBytes, 2);
|
||||
_client.add(_password, passwordLength);
|
||||
}
|
||||
_client.send();
|
||||
_lastClientActivity = millis();
|
||||
SEMAPHORE_GIVE();
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onDisconnect(AsyncClient* client) {
|
||||
(void)client;
|
||||
if (!_disconnectFlagged) {
|
||||
AsyncMqttClientDisconnectReason reason;
|
||||
|
||||
if (_connectPacketNotEnoughSpace) {
|
||||
reason = AsyncMqttClientDisconnectReason::ESP8266_NOT_ENOUGH_SPACE;
|
||||
} else if (_tlsBadFingerprint) {
|
||||
reason = AsyncMqttClientDisconnectReason::TLS_BAD_FINGERPRINT;
|
||||
} else {
|
||||
reason = AsyncMqttClientDisconnectReason::TCP_DISCONNECTED;
|
||||
}
|
||||
for (auto callback : _onDisconnectUserCallbacks) callback(reason);
|
||||
}
|
||||
_clear();
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onError(AsyncClient* client, int8_t error) {
|
||||
(void)client;
|
||||
(void)error;
|
||||
// _onDisconnect called anyway
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onTimeout(AsyncClient* client, uint32_t time) {
|
||||
(void)client;
|
||||
(void)time;
|
||||
// disconnection will be handled by ping/pong management
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onAck(AsyncClient* client, size_t len, uint32_t time) {
|
||||
(void)client;
|
||||
(void)len;
|
||||
(void)time;
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onData(AsyncClient* client, char* data, size_t len) {
|
||||
(void)client;
|
||||
size_t currentBytePosition = 0;
|
||||
char currentByte;
|
||||
do {
|
||||
switch (_parsingInformation.bufferState) {
|
||||
case AsyncMqttClientInternals::BufferState::NONE:
|
||||
currentByte = data[currentBytePosition++];
|
||||
_parsingInformation.packetType = currentByte >> 4;
|
||||
_parsingInformation.packetFlags = (currentByte << 4) >> 4;
|
||||
_parsingInformation.bufferState = AsyncMqttClientInternals::BufferState::REMAINING_LENGTH;
|
||||
_lastServerActivity = millis();
|
||||
switch (_parsingInformation.packetType) {
|
||||
case AsyncMqttClientInternals::PacketType.CONNACK:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::ConnAckPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onConnAck, this, std::placeholders::_1, std::placeholders::_2));
|
||||
break;
|
||||
case AsyncMqttClientInternals::PacketType.PINGRESP:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::PingRespPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPingResp, this));
|
||||
break;
|
||||
case AsyncMqttClientInternals::PacketType.SUBACK:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::SubAckPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onSubAck, this, std::placeholders::_1, std::placeholders::_2));
|
||||
break;
|
||||
case AsyncMqttClientInternals::PacketType.UNSUBACK:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::UnsubAckPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onUnsubAck, this, std::placeholders::_1));
|
||||
break;
|
||||
case AsyncMqttClientInternals::PacketType.PUBLISH:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::PublishPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5, std::placeholders::_6, std::placeholders::_7, std::placeholders::_8, std::placeholders::_9), std::bind(&AsyncMqttClient::_onPublish, this, std::placeholders::_1, std::placeholders::_2));
|
||||
break;
|
||||
case AsyncMqttClientInternals::PacketType.PUBREL:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::PubRelPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPubRel, this, std::placeholders::_1));
|
||||
break;
|
||||
case AsyncMqttClientInternals::PacketType.PUBACK:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::PubAckPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPubAck, this, std::placeholders::_1));
|
||||
break;
|
||||
case AsyncMqttClientInternals::PacketType.PUBREC:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::PubRecPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPubRec, this, std::placeholders::_1));
|
||||
break;
|
||||
case AsyncMqttClientInternals::PacketType.PUBCOMP:
|
||||
_currentParsedPacket = new AsyncMqttClientInternals::PubCompPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPubComp, this, std::placeholders::_1));
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
break;
|
||||
case AsyncMqttClientInternals::BufferState::REMAINING_LENGTH:
|
||||
currentByte = data[currentBytePosition++];
|
||||
_remainingLengthBuffer[_remainingLengthBufferPosition++] = currentByte;
|
||||
if (currentByte >> 7 == 0) {
|
||||
_parsingInformation.remainingLength = AsyncMqttClientInternals::Helpers::decodeRemainingLength(_remainingLengthBuffer);
|
||||
_remainingLengthBufferPosition = 0;
|
||||
if (_parsingInformation.remainingLength > 0) {
|
||||
_parsingInformation.bufferState = AsyncMqttClientInternals::BufferState::VARIABLE_HEADER;
|
||||
} else {
|
||||
// PINGRESP is a special case where it has no variable header, so the packet ends right here
|
||||
_parsingInformation.bufferState = AsyncMqttClientInternals::BufferState::NONE;
|
||||
_onPingResp();
|
||||
}
|
||||
}
|
||||
break;
|
||||
case AsyncMqttClientInternals::BufferState::VARIABLE_HEADER:
|
||||
_currentParsedPacket->parseVariableHeader(data, len, ¤tBytePosition);
|
||||
break;
|
||||
case AsyncMqttClientInternals::BufferState::PAYLOAD:
|
||||
_currentParsedPacket->parsePayload(data, len, ¤tBytePosition);
|
||||
break;
|
||||
default:
|
||||
currentBytePosition = len;
|
||||
}
|
||||
} while (currentBytePosition != len);
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onPoll(AsyncClient* client) {
|
||||
if (!_connected) return;
|
||||
|
||||
// if there is too much time the client has sent a ping request without a response, disconnect client to avoid half open connections
|
||||
if (_lastPingRequestTime != 0 && (millis() - _lastPingRequestTime) >= (_keepAlive * 1000 * 2)) {
|
||||
disconnect();
|
||||
return;
|
||||
// send ping to ensure the server will receive at least one message inside keepalive window
|
||||
} else if (_lastPingRequestTime == 0 && (millis() - _lastClientActivity) >= (_keepAlive * 1000 * 0.7)) {
|
||||
_sendPing();
|
||||
|
||||
// send ping to verify if the server is still there (ensure this is not a half connection)
|
||||
} else if (_connected && _lastPingRequestTime == 0 && (millis() - _lastServerActivity) >= (_keepAlive * 1000 * 0.7)) {
|
||||
_sendPing();
|
||||
}
|
||||
|
||||
// handle to send ack packets
|
||||
|
||||
_sendAcks();
|
||||
|
||||
// handle disconnect
|
||||
|
||||
if (_disconnectFlagged) {
|
||||
_sendDisconnect();
|
||||
}
|
||||
}
|
||||
|
||||
/* MQTT */
|
||||
void AsyncMqttClient::_onPingResp() {
|
||||
_freeCurrentParsedPacket();
|
||||
_lastPingRequestTime = 0;
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onConnAck(bool sessionPresent, uint8_t connectReturnCode) {
|
||||
(void)sessionPresent;
|
||||
_freeCurrentParsedPacket();
|
||||
|
||||
if (connectReturnCode == 0) {
|
||||
_connected = true;
|
||||
for (auto callback : _onConnectUserCallbacks) callback(sessionPresent);
|
||||
} else {
|
||||
for (auto callback : _onDisconnectUserCallbacks) callback(static_cast<AsyncMqttClientDisconnectReason>(connectReturnCode));
|
||||
_disconnectFlagged = true;
|
||||
}
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onSubAck(uint16_t packetId, char status) {
|
||||
_freeCurrentParsedPacket();
|
||||
|
||||
for (auto callback : _onSubscribeUserCallbacks) callback(packetId, status);
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onUnsubAck(uint16_t packetId) {
|
||||
_freeCurrentParsedPacket();
|
||||
|
||||
for (auto callback : _onUnsubscribeUserCallbacks) callback(packetId);
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onMessage(char* topic, char* payload, uint8_t qos, bool dup, bool retain, size_t len, size_t index, size_t total, uint16_t packetId) {
|
||||
bool notifyPublish = true;
|
||||
|
||||
if (qos == 2) {
|
||||
for (AsyncMqttClientInternals::PendingPubRel pendingPubRel : _pendingPubRels) {
|
||||
if (pendingPubRel.packetId == packetId) {
|
||||
notifyPublish = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (notifyPublish) {
|
||||
AsyncMqttClientMessageProperties properties;
|
||||
properties.qos = qos;
|
||||
properties.dup = dup;
|
||||
properties.retain = retain;
|
||||
|
||||
for (auto callback : _onMessageUserCallbacks) callback(topic, payload, properties, len, index, total);
|
||||
}
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onPublish(uint16_t packetId, uint8_t qos) {
|
||||
AsyncMqttClientInternals::PendingAck pendingAck;
|
||||
|
||||
if (qos == 1) {
|
||||
pendingAck.packetType = AsyncMqttClientInternals::PacketType.PUBACK;
|
||||
pendingAck.headerFlag = AsyncMqttClientInternals::HeaderFlag.PUBACK_RESERVED;
|
||||
pendingAck.packetId = packetId;
|
||||
_toSendAcks.push_back(pendingAck);
|
||||
} else if (qos == 2) {
|
||||
pendingAck.packetType = AsyncMqttClientInternals::PacketType.PUBREC;
|
||||
pendingAck.headerFlag = AsyncMqttClientInternals::HeaderFlag.PUBREC_RESERVED;
|
||||
pendingAck.packetId = packetId;
|
||||
_toSendAcks.push_back(pendingAck);
|
||||
|
||||
bool pubRelAwaiting = false;
|
||||
for (AsyncMqttClientInternals::PendingPubRel pendingPubRel : _pendingPubRels) {
|
||||
if (pendingPubRel.packetId == packetId) {
|
||||
pubRelAwaiting = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (!pubRelAwaiting) {
|
||||
AsyncMqttClientInternals::PendingPubRel pendingPubRel;
|
||||
pendingPubRel.packetId = packetId;
|
||||
_pendingPubRels.push_back(pendingPubRel);
|
||||
}
|
||||
|
||||
_sendAcks();
|
||||
}
|
||||
|
||||
_freeCurrentParsedPacket();
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onPubRel(uint16_t packetId) {
|
||||
_freeCurrentParsedPacket();
|
||||
|
||||
AsyncMqttClientInternals::PendingAck pendingAck;
|
||||
pendingAck.packetType = AsyncMqttClientInternals::PacketType.PUBCOMP;
|
||||
pendingAck.headerFlag = AsyncMqttClientInternals::HeaderFlag.PUBCOMP_RESERVED;
|
||||
pendingAck.packetId = packetId;
|
||||
_toSendAcks.push_back(pendingAck);
|
||||
|
||||
for (size_t i = 0; i < _pendingPubRels.size(); i++) {
|
||||
if (_pendingPubRels[i].packetId == packetId) {
|
||||
_pendingPubRels.erase(_pendingPubRels.begin() + i);
|
||||
_pendingPubRels.shrink_to_fit();
|
||||
}
|
||||
}
|
||||
|
||||
_sendAcks();
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onPubAck(uint16_t packetId) {
|
||||
_freeCurrentParsedPacket();
|
||||
|
||||
for (auto callback : _onPublishUserCallbacks) callback(packetId);
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onPubRec(uint16_t packetId) {
|
||||
_freeCurrentParsedPacket();
|
||||
|
||||
AsyncMqttClientInternals::PendingAck pendingAck;
|
||||
pendingAck.packetType = AsyncMqttClientInternals::PacketType.PUBREL;
|
||||
pendingAck.headerFlag = AsyncMqttClientInternals::HeaderFlag.PUBREL_RESERVED;
|
||||
pendingAck.packetId = packetId;
|
||||
_toSendAcks.push_back(pendingAck);
|
||||
|
||||
_sendAcks();
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_onPubComp(uint16_t packetId) {
|
||||
_freeCurrentParsedPacket();
|
||||
|
||||
for (auto callback : _onPublishUserCallbacks) callback(packetId);
|
||||
}
|
||||
|
||||
bool AsyncMqttClient::_sendPing() {
|
||||
char fixedHeader[2];
|
||||
fixedHeader[0] = AsyncMqttClientInternals::PacketType.PINGREQ;
|
||||
fixedHeader[0] = fixedHeader[0] << 4;
|
||||
fixedHeader[0] = fixedHeader[0] | AsyncMqttClientInternals::HeaderFlag.PINGREQ_RESERVED;
|
||||
fixedHeader[1] = 0;
|
||||
|
||||
size_t neededSpace = 2;
|
||||
|
||||
SEMAPHORE_TAKE(false);
|
||||
if (_client.space() < neededSpace) { SEMAPHORE_GIVE(); return false; }
|
||||
|
||||
_client.add(fixedHeader, 2);
|
||||
_client.send();
|
||||
_lastClientActivity = millis();
|
||||
_lastPingRequestTime = millis();
|
||||
|
||||
SEMAPHORE_GIVE();
|
||||
return true;
|
||||
}
|
||||
|
||||
void AsyncMqttClient::_sendAcks() {
|
||||
uint8_t neededAckSpace = 2 + 2;
|
||||
|
||||
SEMAPHORE_TAKE();
|
||||
for (size_t i = 0; i < _toSendAcks.size(); i++) {
|
||||
if (_client.space() < neededAckSpace) break;
|
||||
|
||||
AsyncMqttClientInternals::PendingAck pendingAck = _toSendAcks[i];
|
||||
|
||||
char fixedHeader[2];
|
||||
fixedHeader[0] = pendingAck.packetType;
|
||||
fixedHeader[0] = fixedHeader[0] << 4;
|
||||
fixedHeader[0] = fixedHeader[0] | pendingAck.headerFlag;
|
||||
fixedHeader[1] = 2;
|
||||
|
||||
char packetIdBytes[2];
|
||||
packetIdBytes[0] = pendingAck.packetId >> 8;
|
||||
packetIdBytes[1] = pendingAck.packetId & 0xFF;
|
||||
|
||||
_client.add(fixedHeader, 2);
|
||||
_client.add(packetIdBytes, 2);
|
||||
_client.send();
|
||||
|
||||
_toSendAcks.erase(_toSendAcks.begin() + i);
|
||||
_toSendAcks.shrink_to_fit();
|
||||
|
||||
_lastClientActivity = millis();
|
||||
}
|
||||
SEMAPHORE_GIVE();
|
||||
}
|
||||
|
||||
bool AsyncMqttClient::_sendDisconnect() {
|
||||
if (!_connected) return true;
|
||||
|
||||
const uint8_t neededSpace = 2;
|
||||
|
||||
SEMAPHORE_TAKE(false);
|
||||
|
||||
if (_client.space() < neededSpace) { SEMAPHORE_GIVE(); return false; }
|
||||
|
||||
char fixedHeader[2];
|
||||
fixedHeader[0] = AsyncMqttClientInternals::PacketType.DISCONNECT;
|
||||
fixedHeader[0] = fixedHeader[0] << 4;
|
||||
fixedHeader[0] = fixedHeader[0] | AsyncMqttClientInternals::HeaderFlag.DISCONNECT_RESERVED;
|
||||
fixedHeader[1] = 0;
|
||||
|
||||
_client.add(fixedHeader, 2);
|
||||
_client.send();
|
||||
_client.close(true);
|
||||
|
||||
_disconnectFlagged = false;
|
||||
|
||||
SEMAPHORE_GIVE();
|
||||
return true;
|
||||
}
|
||||
|
||||
uint16_t AsyncMqttClient::_getNextPacketId() {
|
||||
uint16_t nextPacketId = _nextPacketId;
|
||||
|
||||
if (_nextPacketId == 65535) _nextPacketId = 0; // 0 is forbidden
|
||||
_nextPacketId++;
|
||||
|
||||
return nextPacketId;
|
||||
}
|
||||
|
||||
bool AsyncMqttClient::connected() const {
|
||||
return _connected;
|
||||
}
|
||||
|
||||
void AsyncMqttClient::connect() {
|
||||
if (_connected) return;
|
||||
|
||||
#if ASYNC_TCP_SSL_ENABLED
|
||||
if (_useIp) {
|
||||
_client.connect(_ip, _port, _secure);
|
||||
} else {
|
||||
_client.connect(_host, _port, _secure);
|
||||
}
|
||||
#else
|
||||
if (_useIp) {
|
||||
_client.connect(_ip, _port);
|
||||
} else {
|
||||
_client.connect(_host, _port);
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
void AsyncMqttClient::disconnect(bool force) {
|
||||
if (!_connected) return;
|
||||
|
||||
if (force) {
|
||||
_client.close(true);
|
||||
} else {
|
||||
_disconnectFlagged = true;
|
||||
_sendDisconnect();
|
||||
}
|
||||
}
|
||||
|
||||
uint16_t AsyncMqttClient::subscribe(const char* topic, uint8_t qos) {
|
||||
if (!_connected) return 0;
|
||||
|
||||
char fixedHeader[5];
|
||||
fixedHeader[0] = AsyncMqttClientInternals::PacketType.SUBSCRIBE;
|
||||
fixedHeader[0] = fixedHeader[0] << 4;
|
||||
fixedHeader[0] = fixedHeader[0] | AsyncMqttClientInternals::HeaderFlag.SUBSCRIBE_RESERVED;
|
||||
|
||||
uint16_t topicLength = strlen(topic);
|
||||
char topicLengthBytes[2];
|
||||
topicLengthBytes[0] = topicLength >> 8;
|
||||
topicLengthBytes[1] = topicLength & 0xFF;
|
||||
|
||||
char qosByte[1];
|
||||
qosByte[0] = qos;
|
||||
|
||||
uint8_t remainingLengthLength = AsyncMqttClientInternals::Helpers::encodeRemainingLength(2 + 2 + topicLength + 1, fixedHeader + 1);
|
||||
|
||||
size_t neededSpace = 0;
|
||||
neededSpace += 1 + remainingLengthLength;
|
||||
neededSpace += 2;
|
||||
neededSpace += 2;
|
||||
neededSpace += topicLength;
|
||||
neededSpace += 1;
|
||||
|
||||
SEMAPHORE_TAKE(0);
|
||||
if (_client.space() < neededSpace) { SEMAPHORE_GIVE(); return 0; }
|
||||
|
||||
uint16_t packetId = _getNextPacketId();
|
||||
char packetIdBytes[2];
|
||||
packetIdBytes[0] = packetId >> 8;
|
||||
packetIdBytes[1] = packetId & 0xFF;
|
||||
|
||||
_client.add(fixedHeader, 1 + remainingLengthLength);
|
||||
_client.add(packetIdBytes, 2);
|
||||
_client.add(topicLengthBytes, 2);
|
||||
_client.add(topic, topicLength);
|
||||
_client.add(qosByte, 1);
|
||||
_client.send();
|
||||
_lastClientActivity = millis();
|
||||
|
||||
SEMAPHORE_GIVE();
|
||||
return packetId;
|
||||
}
|
||||
|
||||
uint16_t AsyncMqttClient::unsubscribe(const char* topic) {
|
||||
if (!_connected) return 0;
|
||||
|
||||
char fixedHeader[5];
|
||||
fixedHeader[0] = AsyncMqttClientInternals::PacketType.UNSUBSCRIBE;
|
||||
fixedHeader[0] = fixedHeader[0] << 4;
|
||||
fixedHeader[0] = fixedHeader[0] | AsyncMqttClientInternals::HeaderFlag.UNSUBSCRIBE_RESERVED;
|
||||
|
||||
uint16_t topicLength = strlen(topic);
|
||||
char topicLengthBytes[2];
|
||||
topicLengthBytes[0] = topicLength >> 8;
|
||||
topicLengthBytes[1] = topicLength & 0xFF;
|
||||
|
||||
uint8_t remainingLengthLength = AsyncMqttClientInternals::Helpers::encodeRemainingLength(2 + 2 + topicLength, fixedHeader + 1);
|
||||
|
||||
size_t neededSpace = 0;
|
||||
neededSpace += 1 + remainingLengthLength;
|
||||
neededSpace += 2;
|
||||
neededSpace += 2;
|
||||
neededSpace += topicLength;
|
||||
|
||||
SEMAPHORE_TAKE(0);
|
||||
if (_client.space() < neededSpace) { SEMAPHORE_GIVE(); return 0; }
|
||||
|
||||
uint16_t packetId = _getNextPacketId();
|
||||
char packetIdBytes[2];
|
||||
packetIdBytes[0] = packetId >> 8;
|
||||
packetIdBytes[1] = packetId & 0xFF;
|
||||
|
||||
_client.add(fixedHeader, 1 + remainingLengthLength);
|
||||
_client.add(packetIdBytes, 2);
|
||||
_client.add(topicLengthBytes, 2);
|
||||
_client.add(topic, topicLength);
|
||||
_client.send();
|
||||
_lastClientActivity = millis();
|
||||
|
||||
SEMAPHORE_GIVE();
|
||||
return packetId;
|
||||
}
|
||||
|
||||
uint16_t AsyncMqttClient::publish(const char* topic, uint8_t qos, bool retain, const char* payload, size_t length, bool dup, uint16_t message_id) {
|
||||
if (!_connected) return 0;
|
||||
|
||||
char fixedHeader[5];
|
||||
fixedHeader[0] = AsyncMqttClientInternals::PacketType.PUBLISH;
|
||||
fixedHeader[0] = fixedHeader[0] << 4;
|
||||
if (dup) fixedHeader[0] |= AsyncMqttClientInternals::HeaderFlag.PUBLISH_DUP;
|
||||
if (retain) fixedHeader[0] |= AsyncMqttClientInternals::HeaderFlag.PUBLISH_RETAIN;
|
||||
switch (qos) {
|
||||
case 0:
|
||||
fixedHeader[0] |= AsyncMqttClientInternals::HeaderFlag.PUBLISH_QOS0;
|
||||
break;
|
||||
case 1:
|
||||
fixedHeader[0] |= AsyncMqttClientInternals::HeaderFlag.PUBLISH_QOS1;
|
||||
break;
|
||||
case 2:
|
||||
fixedHeader[0] |= AsyncMqttClientInternals::HeaderFlag.PUBLISH_QOS2;
|
||||
break;
|
||||
}
|
||||
|
||||
uint16_t topicLength = strlen(topic);
|
||||
char topicLengthBytes[2];
|
||||
topicLengthBytes[0] = topicLength >> 8;
|
||||
topicLengthBytes[1] = topicLength & 0xFF;
|
||||
|
||||
uint32_t payloadLength = length;
|
||||
if (payload != nullptr && payloadLength == 0) payloadLength = strlen(payload);
|
||||
|
||||
uint32_t remainingLength = 2 + topicLength + payloadLength;
|
||||
if (qos != 0) remainingLength += 2;
|
||||
uint8_t remainingLengthLength = AsyncMqttClientInternals::Helpers::encodeRemainingLength(remainingLength, fixedHeader + 1);
|
||||
|
||||
size_t neededSpace = 0;
|
||||
neededSpace += 1 + remainingLengthLength;
|
||||
neededSpace += 2;
|
||||
neededSpace += topicLength;
|
||||
if (qos != 0) neededSpace += 2;
|
||||
if (payload != nullptr) neededSpace += payloadLength;
|
||||
|
||||
SEMAPHORE_TAKE(0);
|
||||
if (_client.space() < neededSpace) { SEMAPHORE_GIVE(); return 0; }
|
||||
|
||||
uint16_t packetId = 0;
|
||||
char packetIdBytes[2];
|
||||
if (qos != 0) {
|
||||
if (dup && message_id > 0) {
|
||||
packetId = message_id;
|
||||
} else {
|
||||
packetId = _getNextPacketId();
|
||||
}
|
||||
|
||||
packetIdBytes[0] = packetId >> 8;
|
||||
packetIdBytes[1] = packetId & 0xFF;
|
||||
}
|
||||
|
||||
_client.add(fixedHeader, 1 + remainingLengthLength);
|
||||
_client.add(topicLengthBytes, 2);
|
||||
_client.add(topic, topicLength);
|
||||
if (qos != 0) _client.add(packetIdBytes, 2);
|
||||
if (payload != nullptr) _client.add(payload, payloadLength);
|
||||
_client.send();
|
||||
_lastClientActivity = millis();
|
||||
|
||||
SEMAPHORE_GIVE();
|
||||
if (qos != 0) {
|
||||
return packetId;
|
||||
} else {
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
@ -1,6 +0,0 @@
|
||||
#ifndef SRC_ASYNCMQTTCLIENT_H_
|
||||
#define SRC_ASYNCMQTTCLIENT_H_
|
||||
|
||||
#include "AsyncMqttClient.hpp"
|
||||
|
||||
#endif // SRC_ASYNCMQTTCLIENT_H_
|
||||
@ -1,166 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include <functional>
|
||||
#include <vector>
|
||||
|
||||
#include "Arduino.h"
|
||||
|
||||
#ifdef ESP32
|
||||
#include <AsyncTCP.h>
|
||||
#include <freertos/semphr.h>
|
||||
#elif defined(ESP8266)
|
||||
#include <ESPAsyncTCP.h>
|
||||
#else
|
||||
#error Platform not supported
|
||||
#endif
|
||||
|
||||
#if ASYNC_TCP_SSL_ENABLED
|
||||
#include <tcp_axtls.h>
|
||||
#define SHA1_SIZE 20
|
||||
#endif
|
||||
|
||||
#include "AsyncMqttClient/Flags.hpp"
|
||||
#include "AsyncMqttClient/ParsingInformation.hpp"
|
||||
#include "AsyncMqttClient/MessageProperties.hpp"
|
||||
#include "AsyncMqttClient/Helpers.hpp"
|
||||
#include "AsyncMqttClient/Callbacks.hpp"
|
||||
#include "AsyncMqttClient/DisconnectReasons.hpp"
|
||||
#include "AsyncMqttClient/Storage.hpp"
|
||||
|
||||
#include "AsyncMqttClient/Packets/Packet.hpp"
|
||||
#include "AsyncMqttClient/Packets/ConnAckPacket.hpp"
|
||||
#include "AsyncMqttClient/Packets/PingRespPacket.hpp"
|
||||
#include "AsyncMqttClient/Packets/SubAckPacket.hpp"
|
||||
#include "AsyncMqttClient/Packets/UnsubAckPacket.hpp"
|
||||
#include "AsyncMqttClient/Packets/PublishPacket.hpp"
|
||||
#include "AsyncMqttClient/Packets/PubRelPacket.hpp"
|
||||
#include "AsyncMqttClient/Packets/PubAckPacket.hpp"
|
||||
#include "AsyncMqttClient/Packets/PubRecPacket.hpp"
|
||||
#include "AsyncMqttClient/Packets/PubCompPacket.hpp"
|
||||
|
||||
#if ESP32
|
||||
#define SEMAPHORE_TAKE(X) if (xSemaphoreTake(_xSemaphore, 1000 / portTICK_PERIOD_MS) != pdTRUE) { return X; } // Waits max 1000ms
|
||||
#define SEMAPHORE_GIVE() xSemaphoreGive(_xSemaphore);
|
||||
#elif defined(ESP8266)
|
||||
#define SEMAPHORE_TAKE(X) void()
|
||||
#define SEMAPHORE_GIVE() void()
|
||||
#endif
|
||||
|
||||
class AsyncMqttClient {
|
||||
public:
|
||||
AsyncMqttClient();
|
||||
~AsyncMqttClient();
|
||||
|
||||
AsyncMqttClient& setKeepAlive(uint16_t keepAlive);
|
||||
AsyncMqttClient& setClientId(const char* clientId);
|
||||
AsyncMqttClient& setCleanSession(bool cleanSession);
|
||||
AsyncMqttClient& setMaxTopicLength(uint16_t maxTopicLength);
|
||||
AsyncMqttClient& setCredentials(const char* username, const char* password = nullptr);
|
||||
AsyncMqttClient& setWill(const char* topic, uint8_t qos, bool retain, const char* payload = nullptr, size_t length = 0);
|
||||
AsyncMqttClient& setServer(IPAddress ip, uint16_t port);
|
||||
AsyncMqttClient& setServer(const char* host, uint16_t port);
|
||||
#if ASYNC_TCP_SSL_ENABLED
|
||||
AsyncMqttClient& setSecure(bool secure);
|
||||
AsyncMqttClient& addServerFingerprint(const uint8_t* fingerprint);
|
||||
#endif
|
||||
|
||||
AsyncMqttClient& onConnect(AsyncMqttClientInternals::OnConnectUserCallback callback);
|
||||
AsyncMqttClient& onDisconnect(AsyncMqttClientInternals::OnDisconnectUserCallback callback);
|
||||
AsyncMqttClient& onSubscribe(AsyncMqttClientInternals::OnSubscribeUserCallback callback);
|
||||
AsyncMqttClient& onUnsubscribe(AsyncMqttClientInternals::OnUnsubscribeUserCallback callback);
|
||||
AsyncMqttClient& onMessage(AsyncMqttClientInternals::OnMessageUserCallback callback);
|
||||
AsyncMqttClient& onPublish(AsyncMqttClientInternals::OnPublishUserCallback callback);
|
||||
|
||||
bool connected() const;
|
||||
void connect();
|
||||
void disconnect(bool force = false);
|
||||
uint16_t subscribe(const char* topic, uint8_t qos);
|
||||
uint16_t unsubscribe(const char* topic);
|
||||
uint16_t publish(const char* topic, uint8_t qos, bool retain, const char* payload = nullptr, size_t length = 0, bool dup = false, uint16_t message_id = 0);
|
||||
|
||||
private:
|
||||
AsyncClient _client;
|
||||
|
||||
bool _connected;
|
||||
bool _connectPacketNotEnoughSpace;
|
||||
bool _disconnectFlagged;
|
||||
bool _tlsBadFingerprint;
|
||||
uint32_t _lastClientActivity;
|
||||
uint32_t _lastServerActivity;
|
||||
uint32_t _lastPingRequestTime;
|
||||
|
||||
char _generatedClientId[13 + 1]; // esp8266abc123
|
||||
IPAddress _ip;
|
||||
const char* _host;
|
||||
bool _useIp;
|
||||
#if ASYNC_TCP_SSL_ENABLED
|
||||
bool _secure;
|
||||
#endif
|
||||
uint16_t _port;
|
||||
uint16_t _keepAlive;
|
||||
bool _cleanSession;
|
||||
const char* _clientId;
|
||||
const char* _username;
|
||||
const char* _password;
|
||||
const char* _willTopic;
|
||||
const char* _willPayload;
|
||||
uint16_t _willPayloadLength;
|
||||
uint8_t _willQos;
|
||||
bool _willRetain;
|
||||
|
||||
#if ASYNC_TCP_SSL_ENABLED
|
||||
std::vector<std::array<uint8_t, SHA1_SIZE>> _secureServerFingerprints;
|
||||
#endif
|
||||
|
||||
std::vector<AsyncMqttClientInternals::OnConnectUserCallback> _onConnectUserCallbacks;
|
||||
std::vector<AsyncMqttClientInternals::OnDisconnectUserCallback> _onDisconnectUserCallbacks;
|
||||
std::vector<AsyncMqttClientInternals::OnSubscribeUserCallback> _onSubscribeUserCallbacks;
|
||||
std::vector<AsyncMqttClientInternals::OnUnsubscribeUserCallback> _onUnsubscribeUserCallbacks;
|
||||
std::vector<AsyncMqttClientInternals::OnMessageUserCallback> _onMessageUserCallbacks;
|
||||
std::vector<AsyncMqttClientInternals::OnPublishUserCallback> _onPublishUserCallbacks;
|
||||
|
||||
AsyncMqttClientInternals::ParsingInformation _parsingInformation;
|
||||
AsyncMqttClientInternals::Packet* _currentParsedPacket;
|
||||
uint8_t _remainingLengthBufferPosition;
|
||||
char _remainingLengthBuffer[4];
|
||||
|
||||
uint16_t _nextPacketId;
|
||||
|
||||
std::vector<AsyncMqttClientInternals::PendingPubRel> _pendingPubRels;
|
||||
|
||||
std::vector<AsyncMqttClientInternals::PendingAck> _toSendAcks;
|
||||
|
||||
#ifdef ESP32
|
||||
SemaphoreHandle_t _xSemaphore = nullptr;
|
||||
#endif
|
||||
|
||||
void _clear();
|
||||
void _freeCurrentParsedPacket();
|
||||
|
||||
// TCP
|
||||
void _onConnect(AsyncClient* client);
|
||||
void _onDisconnect(AsyncClient* client);
|
||||
static void _onError(AsyncClient* client, int8_t error);
|
||||
void _onTimeout(AsyncClient* client, uint32_t time);
|
||||
static void _onAck(AsyncClient* client, size_t len, uint32_t time);
|
||||
void _onData(AsyncClient* client, char* data, size_t len);
|
||||
void _onPoll(AsyncClient* client);
|
||||
|
||||
// MQTT
|
||||
void _onPingResp();
|
||||
void _onConnAck(bool sessionPresent, uint8_t connectReturnCode);
|
||||
void _onSubAck(uint16_t packetId, char status);
|
||||
void _onUnsubAck(uint16_t packetId);
|
||||
void _onMessage(char* topic, char* payload, uint8_t qos, bool dup, bool retain, size_t len, size_t index, size_t total, uint16_t packetId);
|
||||
void _onPublish(uint16_t packetId, uint8_t qos);
|
||||
void _onPubRel(uint16_t packetId);
|
||||
void _onPubAck(uint16_t packetId);
|
||||
void _onPubRec(uint16_t packetId);
|
||||
void _onPubComp(uint16_t packetId);
|
||||
|
||||
bool _sendPing();
|
||||
void _sendAcks();
|
||||
bool _sendDisconnect();
|
||||
|
||||
uint16_t _getNextPacketId();
|
||||
};
|
||||
@ -1,28 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include <functional>
|
||||
|
||||
#include "DisconnectReasons.hpp"
|
||||
#include "MessageProperties.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
// user callbacks
|
||||
typedef std::function<void(bool sessionPresent)> OnConnectUserCallback;
|
||||
typedef std::function<void(AsyncMqttClientDisconnectReason reason)> OnDisconnectUserCallback;
|
||||
typedef std::function<void(uint16_t packetId, uint8_t qos)> OnSubscribeUserCallback;
|
||||
typedef std::function<void(uint16_t packetId)> OnUnsubscribeUserCallback;
|
||||
typedef std::function<void(char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total)> OnMessageUserCallback;
|
||||
typedef std::function<void(uint16_t packetId)> OnPublishUserCallback;
|
||||
|
||||
// internal callbacks
|
||||
typedef std::function<void(bool sessionPresent, uint8_t connectReturnCode)> OnConnAckInternalCallback;
|
||||
typedef std::function<void()> OnPingRespInternalCallback;
|
||||
typedef std::function<void(uint16_t packetId, char status)> OnSubAckInternalCallback;
|
||||
typedef std::function<void(uint16_t packetId)> OnUnsubAckInternalCallback;
|
||||
typedef std::function<void(char* topic, char* payload, uint8_t qos, bool dup, bool retain, size_t len, size_t index, size_t total, uint16_t packetId)> OnMessageInternalCallback;
|
||||
typedef std::function<void(uint16_t packetId, uint8_t qos)> OnPublishInternalCallback;
|
||||
typedef std::function<void(uint16_t packetId)> OnPubRelInternalCallback;
|
||||
typedef std::function<void(uint16_t packetId)> OnPubAckInternalCallback;
|
||||
typedef std::function<void(uint16_t packetId)> OnPubRecInternalCallback;
|
||||
typedef std::function<void(uint16_t packetId)> OnPubCompInternalCallback;
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,15 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
enum class AsyncMqttClientDisconnectReason : int8_t {
|
||||
TCP_DISCONNECTED = 0,
|
||||
|
||||
MQTT_UNACCEPTABLE_PROTOCOL_VERSION = 1,
|
||||
MQTT_IDENTIFIER_REJECTED = 2,
|
||||
MQTT_SERVER_UNAVAILABLE = 3,
|
||||
MQTT_MALFORMED_CREDENTIALS = 4,
|
||||
MQTT_NOT_AUTHORIZED = 5,
|
||||
|
||||
ESP8266_NOT_ENOUGH_SPACE = 6,
|
||||
|
||||
TLS_BAD_FINGERPRINT = 7
|
||||
};
|
||||
@ -1,57 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
constexpr struct {
|
||||
const uint8_t RESERVED = 0;
|
||||
const uint8_t CONNECT = 1;
|
||||
const uint8_t CONNACK = 2;
|
||||
const uint8_t PUBLISH = 3;
|
||||
const uint8_t PUBACK = 4;
|
||||
const uint8_t PUBREC = 5;
|
||||
const uint8_t PUBREL = 6;
|
||||
const uint8_t PUBCOMP = 7;
|
||||
const uint8_t SUBSCRIBE = 8;
|
||||
const uint8_t SUBACK = 9;
|
||||
const uint8_t UNSUBSCRIBE = 10;
|
||||
const uint8_t UNSUBACK = 11;
|
||||
const uint8_t PINGREQ = 12;
|
||||
const uint8_t PINGRESP = 13;
|
||||
const uint8_t DISCONNECT = 14;
|
||||
const uint8_t RESERVED2 = 1;
|
||||
} PacketType;
|
||||
|
||||
constexpr struct {
|
||||
const uint8_t CONNECT_RESERVED = 0x00;
|
||||
const uint8_t CONNACK_RESERVED = 0x00;
|
||||
const uint8_t PUBLISH_DUP = 0x08;
|
||||
const uint8_t PUBLISH_QOS0 = 0x00;
|
||||
const uint8_t PUBLISH_QOS1 = 0x02;
|
||||
const uint8_t PUBLISH_QOS2 = 0x04;
|
||||
const uint8_t PUBLISH_QOSRESERVED = 0x06;
|
||||
const uint8_t PUBLISH_RETAIN = 0x01;
|
||||
const uint8_t PUBACK_RESERVED = 0x00;
|
||||
const uint8_t PUBREC_RESERVED = 0x00;
|
||||
const uint8_t PUBREL_RESERVED = 0x02;
|
||||
const uint8_t PUBCOMP_RESERVED = 0x00;
|
||||
const uint8_t SUBSCRIBE_RESERVED = 0x02;
|
||||
const uint8_t SUBACK_RESERVED = 0x00;
|
||||
const uint8_t UNSUBSCRIBE_RESERVED = 0x02;
|
||||
const uint8_t UNSUBACK_RESERVED = 0x00;
|
||||
const uint8_t PINGREQ_RESERVED = 0x00;
|
||||
const uint8_t PINGRESP_RESERVED = 0x00;
|
||||
const uint8_t DISCONNECT_RESERVED = 0x00;
|
||||
const uint8_t RESERVED2_RESERVED = 0x00;
|
||||
} HeaderFlag;
|
||||
|
||||
constexpr struct {
|
||||
const uint8_t USERNAME = 0x80;
|
||||
const uint8_t PASSWORD = 0x40;
|
||||
const uint8_t WILL_RETAIN = 0x20;
|
||||
const uint8_t WILL_QOS0 = 0x00;
|
||||
const uint8_t WILL_QOS1 = 0x08;
|
||||
const uint8_t WILL_QOS2 = 0x10;
|
||||
const uint8_t WILL = 0x04;
|
||||
const uint8_t CLEAN_SESSION = 0x02;
|
||||
const uint8_t RESERVED = 0x00;
|
||||
} ConnectFlag;
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,38 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class Helpers {
|
||||
public:
|
||||
static uint32_t decodeRemainingLength(char* bytes) {
|
||||
uint32_t multiplier = 1;
|
||||
uint32_t value = 0;
|
||||
uint8_t currentByte = 0;
|
||||
uint8_t encodedByte;
|
||||
do {
|
||||
encodedByte = bytes[currentByte++];
|
||||
value += (encodedByte & 127) * multiplier;
|
||||
multiplier *= 128;
|
||||
} while ((encodedByte & 128) != 0);
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
static uint8_t encodeRemainingLength(uint32_t remainingLength, char* destination) {
|
||||
uint8_t currentByte = 0;
|
||||
uint8_t bytesNeeded = 0;
|
||||
|
||||
do {
|
||||
uint8_t encodedByte = remainingLength % 128;
|
||||
remainingLength /= 128;
|
||||
if (remainingLength > 0) {
|
||||
encodedByte = encodedByte | 128;
|
||||
}
|
||||
|
||||
destination[currentByte++] = encodedByte;
|
||||
bytesNeeded++;
|
||||
} while (remainingLength > 0);
|
||||
|
||||
return bytesNeeded;
|
||||
}
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,7 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
struct AsyncMqttClientMessageProperties {
|
||||
uint8_t qos;
|
||||
bool dup;
|
||||
bool retain;
|
||||
};
|
||||
@ -1,30 +0,0 @@
|
||||
#include "ConnAckPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::ConnAckPacket;
|
||||
|
||||
ConnAckPacket::ConnAckPacket(ParsingInformation* parsingInformation, OnConnAckInternalCallback callback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _callback(callback)
|
||||
, _bytePosition(0)
|
||||
, _sessionPresent(false)
|
||||
, _connectReturnCode(0) {
|
||||
}
|
||||
|
||||
ConnAckPacket::~ConnAckPacket() {
|
||||
}
|
||||
|
||||
void ConnAckPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char currentByte = data[(*currentBytePosition)++];
|
||||
if (_bytePosition++ == 0) {
|
||||
_sessionPresent = (currentByte << 7) >> 7;
|
||||
} else {
|
||||
_connectReturnCode = currentByte;
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
_callback(_sessionPresent, _connectReturnCode);
|
||||
}
|
||||
}
|
||||
|
||||
void ConnAckPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
(void)data;
|
||||
(void)currentBytePosition;
|
||||
}
|
||||
@ -1,25 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class ConnAckPacket : public Packet {
|
||||
public:
|
||||
explicit ConnAckPacket(ParsingInformation* parsingInformation, OnConnAckInternalCallback callback);
|
||||
~ConnAckPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnConnAckInternalCallback _callback;
|
||||
|
||||
uint8_t _bytePosition;
|
||||
bool _sessionPresent;
|
||||
uint8_t _connectReturnCode;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,11 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class Packet {
|
||||
public:
|
||||
virtual ~Packet() {}
|
||||
|
||||
virtual void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) = 0;
|
||||
virtual void parsePayload(char* data, size_t len, size_t* currentBytePosition) = 0;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,21 +0,0 @@
|
||||
#include "PingRespPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::PingRespPacket;
|
||||
|
||||
PingRespPacket::PingRespPacket(ParsingInformation* parsingInformation, OnPingRespInternalCallback callback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _callback(callback) {
|
||||
}
|
||||
|
||||
PingRespPacket::~PingRespPacket() {
|
||||
}
|
||||
|
||||
void PingRespPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
(void)data;
|
||||
(void)currentBytePosition;
|
||||
}
|
||||
|
||||
void PingRespPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
(void)data;
|
||||
(void)currentBytePosition;
|
||||
}
|
||||
@ -1,21 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class PingRespPacket : public Packet {
|
||||
public:
|
||||
explicit PingRespPacket(ParsingInformation* parsingInformation, OnPingRespInternalCallback callback);
|
||||
~PingRespPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnPingRespInternalCallback _callback;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,30 +0,0 @@
|
||||
#include "PubAckPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::PubAckPacket;
|
||||
|
||||
PubAckPacket::PubAckPacket(ParsingInformation* parsingInformation, OnPubAckInternalCallback callback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _callback(callback)
|
||||
, _bytePosition(0)
|
||||
, _packetIdMsb(0)
|
||||
, _packetId(0) {
|
||||
}
|
||||
|
||||
PubAckPacket::~PubAckPacket() {
|
||||
}
|
||||
|
||||
void PubAckPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char currentByte = data[(*currentBytePosition)++];
|
||||
if (_bytePosition++ == 0) {
|
||||
_packetIdMsb = currentByte;
|
||||
} else {
|
||||
_packetId = currentByte | _packetIdMsb << 8;
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
_callback(_packetId);
|
||||
}
|
||||
}
|
||||
|
||||
void PubAckPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
(void)data;
|
||||
(void)currentBytePosition;
|
||||
}
|
||||
@ -1,25 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class PubAckPacket : public Packet {
|
||||
public:
|
||||
explicit PubAckPacket(ParsingInformation* parsingInformation, OnPubAckInternalCallback callback);
|
||||
~PubAckPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnPubAckInternalCallback _callback;
|
||||
|
||||
uint8_t _bytePosition;
|
||||
char _packetIdMsb;
|
||||
uint16_t _packetId;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,30 +0,0 @@
|
||||
#include "PubCompPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::PubCompPacket;
|
||||
|
||||
PubCompPacket::PubCompPacket(ParsingInformation* parsingInformation, OnPubCompInternalCallback callback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _callback(callback)
|
||||
, _bytePosition(0)
|
||||
, _packetIdMsb(0)
|
||||
, _packetId(0) {
|
||||
}
|
||||
|
||||
PubCompPacket::~PubCompPacket() {
|
||||
}
|
||||
|
||||
void PubCompPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char currentByte = data[(*currentBytePosition)++];
|
||||
if (_bytePosition++ == 0) {
|
||||
_packetIdMsb = currentByte;
|
||||
} else {
|
||||
_packetId = currentByte | _packetIdMsb << 8;
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
_callback(_packetId);
|
||||
}
|
||||
}
|
||||
|
||||
void PubCompPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
(void)data;
|
||||
(void)currentBytePosition;
|
||||
}
|
||||
@ -1,25 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class PubCompPacket : public Packet {
|
||||
public:
|
||||
explicit PubCompPacket(ParsingInformation* parsingInformation, OnPubCompInternalCallback callback);
|
||||
~PubCompPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnPubCompInternalCallback _callback;
|
||||
|
||||
uint8_t _bytePosition;
|
||||
char _packetIdMsb;
|
||||
uint16_t _packetId;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,30 +0,0 @@
|
||||
#include "PubRecPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::PubRecPacket;
|
||||
|
||||
PubRecPacket::PubRecPacket(ParsingInformation* parsingInformation, OnPubRecInternalCallback callback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _callback(callback)
|
||||
, _bytePosition(0)
|
||||
, _packetIdMsb(0)
|
||||
, _packetId(0) {
|
||||
}
|
||||
|
||||
PubRecPacket::~PubRecPacket() {
|
||||
}
|
||||
|
||||
void PubRecPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char currentByte = data[(*currentBytePosition)++];
|
||||
if (_bytePosition++ == 0) {
|
||||
_packetIdMsb = currentByte;
|
||||
} else {
|
||||
_packetId = currentByte | _packetIdMsb << 8;
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
_callback(_packetId);
|
||||
}
|
||||
}
|
||||
|
||||
void PubRecPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
(void)data;
|
||||
(void)currentBytePosition;
|
||||
}
|
||||
@ -1,25 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class PubRecPacket : public Packet {
|
||||
public:
|
||||
explicit PubRecPacket(ParsingInformation* parsingInformation, OnPubRecInternalCallback callback);
|
||||
~PubRecPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnPubRecInternalCallback _callback;
|
||||
|
||||
uint8_t _bytePosition;
|
||||
char _packetIdMsb;
|
||||
uint16_t _packetId;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,30 +0,0 @@
|
||||
#include "PubRelPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::PubRelPacket;
|
||||
|
||||
PubRelPacket::PubRelPacket(ParsingInformation* parsingInformation, OnPubRelInternalCallback callback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _callback(callback)
|
||||
, _bytePosition(0)
|
||||
, _packetIdMsb(0)
|
||||
, _packetId(0) {
|
||||
}
|
||||
|
||||
PubRelPacket::~PubRelPacket() {
|
||||
}
|
||||
|
||||
void PubRelPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char currentByte = data[(*currentBytePosition)++];
|
||||
if (_bytePosition++ == 0) {
|
||||
_packetIdMsb = currentByte;
|
||||
} else {
|
||||
_packetId = currentByte | _packetIdMsb << 8;
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
_callback(_packetId);
|
||||
}
|
||||
}
|
||||
|
||||
void PubRelPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
(void)data;
|
||||
(void)currentBytePosition;
|
||||
}
|
||||
@ -1,25 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class PubRelPacket : public Packet {
|
||||
public:
|
||||
explicit PubRelPacket(ParsingInformation* parsingInformation, OnPubRelInternalCallback callback);
|
||||
~PubRelPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnPubRelInternalCallback _callback;
|
||||
|
||||
uint8_t _bytePosition;
|
||||
char _packetIdMsb;
|
||||
uint16_t _packetId;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,91 +0,0 @@
|
||||
#include "PublishPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::PublishPacket;
|
||||
|
||||
PublishPacket::PublishPacket(ParsingInformation* parsingInformation, OnMessageInternalCallback dataCallback, OnPublishInternalCallback completeCallback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _dataCallback(dataCallback)
|
||||
, _completeCallback(completeCallback)
|
||||
, _dup(false)
|
||||
, _qos(0)
|
||||
, _retain(0)
|
||||
, _bytePosition(0)
|
||||
, _topicLengthMsb(0)
|
||||
, _topicLength(0)
|
||||
, _ignore(false)
|
||||
, _packetIdMsb(0)
|
||||
, _packetId(0)
|
||||
, _payloadLength(0)
|
||||
, _payloadBytesRead(0) {
|
||||
_dup = _parsingInformation->packetFlags & HeaderFlag.PUBLISH_DUP;
|
||||
_retain = _parsingInformation->packetFlags & HeaderFlag.PUBLISH_RETAIN;
|
||||
char qosMasked = _parsingInformation->packetFlags & 0x06;
|
||||
switch (qosMasked) {
|
||||
case HeaderFlag.PUBLISH_QOS0:
|
||||
_qos = 0;
|
||||
break;
|
||||
case HeaderFlag.PUBLISH_QOS1:
|
||||
_qos = 1;
|
||||
break;
|
||||
case HeaderFlag.PUBLISH_QOS2:
|
||||
_qos = 2;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
PublishPacket::~PublishPacket() {
|
||||
}
|
||||
|
||||
void PublishPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char currentByte = data[(*currentBytePosition)++];
|
||||
if (_bytePosition == 0) {
|
||||
_topicLengthMsb = currentByte;
|
||||
} else if (_bytePosition == 1) {
|
||||
_topicLength = currentByte | _topicLengthMsb << 8;
|
||||
if (_topicLength > _parsingInformation->maxTopicLength) {
|
||||
_ignore = true;
|
||||
} else {
|
||||
_parsingInformation->topicBuffer[_topicLength] = '\0';
|
||||
}
|
||||
} else if (_bytePosition >= 2 && _bytePosition < 2 + _topicLength) {
|
||||
// Starting from here, _ignore might be true
|
||||
if (!_ignore) _parsingInformation->topicBuffer[_bytePosition - 2] = currentByte;
|
||||
if (_bytePosition == 2 + _topicLength - 1 && _qos == 0) {
|
||||
_preparePayloadHandling(_parsingInformation->remainingLength - (_bytePosition + 1));
|
||||
return;
|
||||
}
|
||||
} else if (_bytePosition == 2 + _topicLength) {
|
||||
_packetIdMsb = currentByte;
|
||||
} else {
|
||||
_packetId = currentByte | _packetIdMsb << 8;
|
||||
_preparePayloadHandling(_parsingInformation->remainingLength - (_bytePosition + 1));
|
||||
}
|
||||
_bytePosition++;
|
||||
}
|
||||
|
||||
void PublishPacket::_preparePayloadHandling(uint32_t payloadLength) {
|
||||
_payloadLength = payloadLength;
|
||||
if (payloadLength == 0) {
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
if (!_ignore) {
|
||||
_dataCallback(_parsingInformation->topicBuffer, nullptr, _qos, _dup, _retain, 0, 0, 0, _packetId);
|
||||
_completeCallback(_packetId, _qos);
|
||||
}
|
||||
} else {
|
||||
_parsingInformation->bufferState = BufferState::PAYLOAD;
|
||||
}
|
||||
}
|
||||
|
||||
void PublishPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
size_t remainToRead = len - (*currentBytePosition);
|
||||
if (_payloadBytesRead + remainToRead > _payloadLength) remainToRead = _payloadLength - _payloadBytesRead;
|
||||
|
||||
if (!_ignore) _dataCallback(_parsingInformation->topicBuffer, data + (*currentBytePosition), _qos, _dup, _retain, remainToRead, _payloadBytesRead, _payloadLength, _packetId);
|
||||
_payloadBytesRead += remainToRead;
|
||||
(*currentBytePosition) += remainToRead;
|
||||
|
||||
if (_payloadBytesRead == _payloadLength) {
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
if (!_ignore) _completeCallback(_packetId, _qos);
|
||||
}
|
||||
}
|
||||
@ -1,38 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../Flags.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class PublishPacket : public Packet {
|
||||
public:
|
||||
explicit PublishPacket(ParsingInformation* parsingInformation, OnMessageInternalCallback dataCallback, OnPublishInternalCallback completeCallback);
|
||||
~PublishPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnMessageInternalCallback _dataCallback;
|
||||
OnPublishInternalCallback _completeCallback;
|
||||
|
||||
void _preparePayloadHandling(uint32_t payloadLength);
|
||||
|
||||
bool _dup;
|
||||
uint8_t _qos;
|
||||
bool _retain;
|
||||
|
||||
uint8_t _bytePosition;
|
||||
char _topicLengthMsb;
|
||||
uint16_t _topicLength;
|
||||
bool _ignore;
|
||||
char _packetIdMsb;
|
||||
uint16_t _packetId;
|
||||
uint32_t _payloadLength;
|
||||
uint32_t _payloadBytesRead;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,46 +0,0 @@
|
||||
#include "SubAckPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::SubAckPacket;
|
||||
|
||||
SubAckPacket::SubAckPacket(ParsingInformation* parsingInformation, OnSubAckInternalCallback callback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _callback(callback)
|
||||
, _bytePosition(0)
|
||||
, _packetIdMsb(0)
|
||||
, _packetId(0) {
|
||||
}
|
||||
|
||||
SubAckPacket::~SubAckPacket() {
|
||||
}
|
||||
|
||||
void SubAckPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char currentByte = data[(*currentBytePosition)++];
|
||||
if (_bytePosition++ == 0) {
|
||||
_packetIdMsb = currentByte;
|
||||
} else {
|
||||
_packetId = currentByte | _packetIdMsb << 8;
|
||||
_parsingInformation->bufferState = BufferState::PAYLOAD;
|
||||
}
|
||||
}
|
||||
|
||||
void SubAckPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char status = data[(*currentBytePosition)++];
|
||||
|
||||
/* switch (status) {
|
||||
case 0:
|
||||
Serial.println("Success QoS 0");
|
||||
break;
|
||||
case 1:
|
||||
Serial.println("Success QoS 1");
|
||||
break;
|
||||
case 2:
|
||||
Serial.println("Success QoS 2");
|
||||
break;
|
||||
case 0x80:
|
||||
Serial.println("Failure");
|
||||
break;
|
||||
} */
|
||||
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
_callback(_packetId, status);
|
||||
}
|
||||
@ -1,25 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class SubAckPacket : public Packet {
|
||||
public:
|
||||
explicit SubAckPacket(ParsingInformation* parsingInformation, OnSubAckInternalCallback callback);
|
||||
~SubAckPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnSubAckInternalCallback _callback;
|
||||
|
||||
uint8_t _bytePosition;
|
||||
char _packetIdMsb;
|
||||
uint16_t _packetId;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,30 +0,0 @@
|
||||
#include "UnsubAckPacket.hpp"
|
||||
|
||||
using AsyncMqttClientInternals::UnsubAckPacket;
|
||||
|
||||
UnsubAckPacket::UnsubAckPacket(ParsingInformation* parsingInformation, OnUnsubAckInternalCallback callback)
|
||||
: _parsingInformation(parsingInformation)
|
||||
, _callback(callback)
|
||||
, _bytePosition(0)
|
||||
, _packetIdMsb(0)
|
||||
, _packetId(0) {
|
||||
}
|
||||
|
||||
UnsubAckPacket::~UnsubAckPacket() {
|
||||
}
|
||||
|
||||
void UnsubAckPacket::parseVariableHeader(char* data, size_t len, size_t* currentBytePosition) {
|
||||
char currentByte = data[(*currentBytePosition)++];
|
||||
if (_bytePosition++ == 0) {
|
||||
_packetIdMsb = currentByte;
|
||||
} else {
|
||||
_packetId = currentByte | _packetIdMsb << 8;
|
||||
_parsingInformation->bufferState = BufferState::NONE;
|
||||
_callback(_packetId);
|
||||
}
|
||||
}
|
||||
|
||||
void UnsubAckPacket::parsePayload(char* data, size_t len, size_t* currentBytePosition) {
|
||||
(void)data;
|
||||
(void)currentBytePosition;
|
||||
}
|
||||
@ -1,25 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#include "Arduino.h"
|
||||
#include "Packet.hpp"
|
||||
#include "../ParsingInformation.hpp"
|
||||
#include "../Callbacks.hpp"
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
class UnsubAckPacket : public Packet {
|
||||
public:
|
||||
explicit UnsubAckPacket(ParsingInformation* parsingInformation, OnUnsubAckInternalCallback callback);
|
||||
~UnsubAckPacket();
|
||||
|
||||
void parseVariableHeader(char* data, size_t len, size_t* currentBytePosition);
|
||||
void parsePayload(char* data, size_t len, size_t* currentBytePosition);
|
||||
|
||||
private:
|
||||
ParsingInformation* _parsingInformation;
|
||||
OnUnsubAckInternalCallback _callback;
|
||||
|
||||
uint8_t _bytePosition;
|
||||
char _packetIdMsb;
|
||||
uint16_t _packetId;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,21 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
enum class BufferState : uint8_t {
|
||||
NONE = 0,
|
||||
REMAINING_LENGTH = 2,
|
||||
VARIABLE_HEADER = 3,
|
||||
PAYLOAD = 4
|
||||
};
|
||||
|
||||
struct ParsingInformation {
|
||||
BufferState bufferState;
|
||||
|
||||
uint16_t maxTopicLength;
|
||||
char* topicBuffer;
|
||||
|
||||
uint8_t packetType;
|
||||
uint16_t packetFlags;
|
||||
uint32_t remainingLength;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,13 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
namespace AsyncMqttClientInternals {
|
||||
struct PendingPubRel {
|
||||
uint16_t packetId;
|
||||
};
|
||||
|
||||
struct PendingAck {
|
||||
uint8_t packetType;
|
||||
uint8_t headerFlag;
|
||||
uint16_t packetId;
|
||||
};
|
||||
} // namespace AsyncMqttClientInternals
|
||||
@ -1,21 +0,0 @@
|
||||
The MIT License (MIT)
|
||||
|
||||
Copyright (c) 2015 Marvin Roger
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@ -1,18 +0,0 @@
|
||||
Async MQTT client for ESP8266 and ESP32 (Github: https://github.com/marvinroger/async-mqtt-client)
|
||||
=============================
|
||||
|
||||
[](https://travis-ci.org/marvinroger/async-mqtt-client)
|
||||
|
||||
An Arduino for ESP8266 and ESP32 asynchronous [MQTT](http://mqtt.org/) client implementation, built on [me-no-dev/ESPAsyncTCP (ESP8266)](https://github.com/me-no-dev/ESPAsyncTCP) | [me-no-dev/AsyncTCP (ESP32)](https://github.com/me-no-dev/AsyncTCP) .
|
||||
## Features
|
||||
|
||||
* Compliant with the 3.1.1 version of the protocol
|
||||
* Fully asynchronous
|
||||
* Subscribe at QoS 0, 1 and 2
|
||||
* Publish at QoS 0, 1 and 2
|
||||
* SSL/TLS support
|
||||
* Available in the [PlatformIO registry](http://platformio.org/lib/show/346/AsyncMqttClient)
|
||||
|
||||
## Requirements, installation and usage
|
||||
|
||||
The project is documented in the [/docs folder](docs).
|
||||
@ -130,7 +130,7 @@
|
||||
#endif
|
||||
|
||||
#include "src/dependencies/e131/ESPAsyncE131.h"
|
||||
#include "src/dependencies/async-mqtt-client/AsyncMqttClient.h"
|
||||
#include <AsyncMqttClient.h>
|
||||
|
||||
#define ARDUINOJSON_DECODE_UNICODE 0
|
||||
#include "src/dependencies/json/AsyncJson-v6.h"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user