Добавлены библиотеки для работы с MQTT

libraries/async-mqtt-client/src/AsyncMqttClient.cpp

@@ -0,0 +1,877 @@
+#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", ESP.getEfuseMac());
+  _xSemaphore = xSemaphoreCreateMutex();
+#elif defined(ESP8266)
+  sprintf(_generatedClientId, "esp8266%06x", 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, &currentBytePosition);
+        break;
+      case AsyncMqttClientInternals::BufferState::PAYLOAD:
+        _currentParsedPacket->parsePayload(data, len, &currentBytePosition);
+        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;
+  }
+}

libraries/async-mqtt-client/src/AsyncMqttClient.h

@@ -0,0 +1,6 @@
+#ifndef SRC_ASYNCMQTTCLIENT_H_
+#define SRC_ASYNCMQTTCLIENT_H_
+
+#include "AsyncMqttClient.hpp"
+
+#endif  // SRC_ASYNCMQTTCLIENT_H_

libraries/async-mqtt-client/src/AsyncMqttClient.hpp

@@ -0,0 +1,166 @@
+#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();
+};

libraries/async-mqtt-client/src/AsyncMqttClient/Callbacks.hpp

@@ -0,0 +1,28 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/DisconnectReasons.hpp

@@ -0,0 +1,15 @@
+#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
+};

libraries/async-mqtt-client/src/AsyncMqttClient/Flags.hpp

@@ -0,0 +1,57 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Helpers.hpp

@@ -0,0 +1,38 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/MessageProperties.hpp

@@ -0,0 +1,7 @@
+#pragma once
+
+struct AsyncMqttClientMessageProperties {
+  uint8_t qos;
+  bool dup;
+  bool retain;
+};

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/ConnAckPacket.cpp

@@ -0,0 +1,30 @@
+#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;
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/ConnAckPacket.hpp

@@ -0,0 +1,25 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/Packet.hpp

@@ -0,0 +1,11 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PingRespPacket.cpp

@@ -0,0 +1,21 @@
+#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;
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PingRespPacket.hpp

@@ -0,0 +1,21 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PubAckPacket.cpp

@@ -0,0 +1,30 @@
+#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;
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PubAckPacket.hpp

@@ -0,0 +1,25 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PubCompPacket.cpp

@@ -0,0 +1,30 @@
+#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;
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PubCompPacket.hpp

@@ -0,0 +1,25 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PubRecPacket.cpp

@@ -0,0 +1,30 @@
+#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;
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PubRecPacket.hpp

@@ -0,0 +1,25 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PubRelPacket.cpp

@@ -0,0 +1,30 @@
+#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;
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PubRelPacket.hpp

@@ -0,0 +1,25 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PublishPacket.cpp

@@ -0,0 +1,91 @@
+#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);
+  }
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/PublishPacket.hpp

@@ -0,0 +1,38 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/SubAckPacket.cpp

@@ -0,0 +1,46 @@
+#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);
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/SubAckPacket.hpp

@@ -0,0 +1,25 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/UnsubAckPacket.cpp

@@ -0,0 +1,30 @@
+#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;
+}

libraries/async-mqtt-client/src/AsyncMqttClient/Packets/UnsubAckPacket.hpp

@@ -0,0 +1,25 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/ParsingInformation.hpp

@@ -0,0 +1,21 @@
+#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

libraries/async-mqtt-client/src/AsyncMqttClient/Storage.hpp

@@ -0,0 +1,13 @@
+#pragma once
+
+namespace AsyncMqttClientInternals {
+struct PendingPubRel {
+  uint16_t packetId;
+};
+
+struct PendingAck {
+  uint8_t packetType;
+  uint8_t headerFlag;
+  uint16_t packetId;
+};
+}  // namespace AsyncMqttClientInternals