diff --git a/.github/workflows/build_examples_pio.yml b/.github/workflows/build_examples_pio.yml new file mode 100644 index 0000000..a8ad0ae --- /dev/null +++ b/.github/workflows/build_examples_pio.yml @@ -0,0 +1,27 @@ +name: Build with Platformio + +on: [push, pull_request] + +jobs: + build: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v1 + - name: Set up Python + uses: actions/setup-python@v1 + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install platformio + - name: Add libraries + run: | + platformio lib -g install AsyncTCP + platformio lib -g install ESPAsyncTCP + - name: Getting ready + run: | + chmod +x ./scripts/CI/build_examples_pio.sh + - name: Build examples + run: | + ./scripts/CI/build_examples_pio.sh diff --git a/.github/workflows/cpplint.yml b/.github/workflows/cpplint.yml new file mode 100644 index 0000000..3dc4670 --- /dev/null +++ b/.github/workflows/cpplint.yml @@ -0,0 +1,20 @@ +name: cpplint + +on: [push, pull_request] + +jobs: + build: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v1 + - name: Set up Python + uses: actions/setup-python@v1 + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install cpplint + - name: Linting + run: | + cpplint --repository=. --recursive --filter=-whitespace/line_length,-legal/copyright,-runtime/printf,-build/include,-build/namespace ./src diff --git a/.gitignore b/.gitignore index 2af98ca..f9526b3 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ /config.json +.vscode/ \ No newline at end of file diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index f5d3766..0000000 --- a/.travis.yml +++ /dev/null @@ -1,20 +0,0 @@ -language: python -python: - - "2.7" - -cache: - directories: - - "~/.platformio" - -env: - - PLATFORMIO_CI_SRC=examples/FullyFeatured-ESP8266 PLATFORMIO_CI_EXTRA_ARGS="--board=esp01 --board=nodemcuv2" - - PLATFORMIO_CI_SRC=examples/FullyFeatured-ESP32 PLATFORMIO_CI_EXTRA_ARGS="--board=lolin32" - - CPPLINT=true - -install: - - pip install -U https://github.com/platformio/platformio-core/archive/develop.zip - - pip install -U cpplint - - platformio lib -g install file://. - -script: - - if [[ "$CPPLINT" ]]; then make cpplint; else platformio ci $PLATFORMIO_CI_EXTRA_ARGS; fi diff --git a/LICENSE b/LICENSE index a6183c6..a3ee217 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ The MIT License (MIT) -Copyright (c) 2015 Marvin Roger +Copyright (c) 2015-2021 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 diff --git a/README.md b/README.md index af66bee..beae892 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,10 @@ -Async MQTT client for ESP8266 and ESP32 -============================= +# Async MQTT client for ESP8266 and ESP32 -[![Build Status](https://img.shields.io/travis/marvinroger/async-mqtt-client/master.svg?style=flat-square)](https://travis-ci.org/marvinroger/async-mqtt-client) +![Build with PlatformIO](https://github.com/marvinroger/async-mqtt-client/workflows/Build%20with%20Platformio/badge.svg) +![cpplint](https://github.com/marvinroger/async-mqtt-client/workflows/cpplint/badge.svg) 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 diff --git a/docs/1.-Getting-started.md b/docs/1.-Getting-started.md index 2319b59..2af3fda 100644 --- a/docs/1.-Getting-started.md +++ b/docs/1.-Getting-started.md @@ -2,9 +2,9 @@ To use AsyncMqttClient, you need: -* An ESP8266 -* The Arduino IDE for ESP8266 (version 2.2.0 minimum) -* Basic knowledge of the Arduino environment (upload a sketch, import libraries, ...) +* An ESP8266 or ESP32 +* The Arduino IDE or equivalent IDE for ESP8266/32 +* Basic knowledge of the Arduino environment (use the IDE, upload a sketch, import libraries, ...) ## Installing AsyncMqttClient @@ -15,7 +15,9 @@ There are two ways to install AsyncMqttClient. 1. Download the [corresponding release](https://github.com/marvinroger/async-mqtt-client/releases/latest) 2. Load the `.zip` with **Sketch → Include Library → Add .ZIP Library** -AsyncMqttClient has 1 dependency: [ESPAsyncTCP](https://github.com/me-no-dev/ESPAsyncTCP). Download the [.zip](https://github.com/me-no-dev/ESPAsyncTCP/archive/master.zip) and install it with the same method as above. +AsyncMqttClient has 1 dependency: +* For ESP8266: [ESPAsyncTCP](https://github.com/me-no-dev/ESPAsyncTCP). Download the [.zip](https://github.com/me-no-dev/ESPAsyncTCP/archive/master.zip) and install it with the same method as above. +* Fors ESP32: [AsyncTCP](https://github.com/me-no-dev/AsyncTCP). Download the [.zip](https://github.com/me-no-dev/AsyncTCP/archive/master.zip) and install it with the same method as above. ## Fully-featured sketch diff --git a/docs/2.-API-reference.md b/docs/2.-API-reference.md index d992184..c3da35e 100644 --- a/docs/2.-API-reference.md +++ b/docs/2.-API-reference.md @@ -156,5 +156,11 @@ Return the packet ID (or 1 if QoS 0) or 0 if failed. * **`retain`**: Retain flag * **`payload`**: Payload. If unset, the payload will be empty * **`length`**: Payload length. If unset or set to 0, the payload will be considered as a string and its size will be calculated using `strlen(payload)` -* **`dup`**: Duplicate flag. If set or set to 1, the payload will be flagged as a duplicate -* **`message_id`**: The message ID. If unset or set to 0, the message ID will be automtaically assigned. Use this with the DUP flag to identify which message is being duplicated +* **`dup`**: ~~Duplicate flag. If set or set to 1, the payload will be flagged as a duplicate~~ Setting is not used anymore +* **`message_id`**: ~~The message ID. If unset or set to 0, the message ID will be automtaically assigned. Use this with the DUP flag to identify which message is being duplicated~~ Setting is not used anymore + +#### bool clearQueue() + +When disconnected, clears all queued messages + +Returns true on succes, false on failure (client is no disconnected) diff --git a/docs/3.-Memory-management.md b/docs/3.-Memory-management.md index ad27d19..3282005 100644 --- a/docs/3.-Memory-management.md +++ b/docs/3.-Memory-management.md @@ -1,7 +1,12 @@ # Memory management -AsyncMqttClient does not use an internal buffer, it uses the raw TCP buffer. +AsyncMqttClient buffers outgoing messages in a queue. On sending data is copied to a raw TCP buffer. Received data is passed directly to the API. -The max receive size is about 1460 bytes per call to your onMessage callback. But the amount of data you can receive is unlimited, as if you receive, say, a 300kB payload (such as an OTA payload), then your `onMessage` callback will be called about 200 times, with the according len, index and total parameters. Keep in mind the library will call your `onMessage` callbacks with the same topic buffer, so if you change the buffer on one call, the buffer will remain changed on subsequent calls. +## Outgoing messages -You can send data as long as you stay below the available TCP window (which is about 3-4kB on the ESP8266). The data is indeed held in memory by the async TCP code until ACK is received. If the TCP window was sufficient to send your packet, the `publish` method will return a packet ID indicating the packet was sent. Otherwise, a `0` will be returned, and it's your responsability to resend the packet with `publish`. +You can send data as long as memory permits. A minimum amount of free memory is set at 4096 bytes. You can lower (or raise) this value by setting `MQTT_MIN_FREE_MEMORY` to your desired value. +If the free memory was sufficient to send your packet, the `publish` method will return a packet ID indicating the packet was queued. Otherwise, a `0` will be returned, and it's your responsability to resend the packet with `publish`. + +## Incoming messages + +No incoming data is buffered by this library. Messages received by the TCP library is passed directly to the API. The max receive size is about 1460 bytes per call to your onMessage callback but the amount of data you can receive is unlimited. If you receive, say, a 300kB payload (such as an OTA payload), then your `onMessage` callback will be called about 200 times, with the according len, index and total parameters. Keep in mind the library will call your `onMessage` callbacks with the same topic buffer, so if you change the buffer on one call, the buffer will remain changed on subsequent calls. diff --git a/docs/4.-Limitations-and-known-issues.md b/docs/4.-Limitations-and-known-issues.md index c85b0d3..69215ab 100644 --- a/docs/4.-Limitations-and-known-issues.md +++ b/docs/4.-Limitations-and-known-issues.md @@ -1,19 +1,19 @@ # Limitations and known issues -* When the CleanSession is set to `false`, the implementation is not spec compliant. The following is not honored: +* The library is spec compliant with one limitation. In case of power loss the following is not honored: > Must be kept in memory: * All messages in a QoS 1 or 2 flow, which are not confirmed by the broker * All received QoS 2 messages, which are not yet confirmed to the broker -This means retransmission is not honored in case of a failure. +This means retransmission is not honored in case of a power failure. This behaviour is like explained in point 4.1.1 of the MQTT specification v3.1.1 * You cannot send payload larger that what can fit on RAM. ## SSL limitations -* SSL requires use of esp8266/Arduino 2.4.0, which is not yet released (platform = espressif8266_stage in PlatformIO). * SSL requires the build flag -DASYNC_TCP_SSL_ENABLED=1 * SSL only supports fingerprints for server validation. * If you do not specify one or more acceptable server fingerprints, the SSL connection will be vulnerable to man-in-the-middle attacks. -* Some server certificate signature algorithms do not work. SHA1, SHA224, SHA256, and MD5 are working. SHA384, and SHA512 will cause a crash. +* Some server certificate signature algorithms do not work. SHA1, SHA224, SHA256, and MD5 are working. SHA384, and SHA512 will cause a crash. +* TLS1.2 is not supported. diff --git a/docs/5.-Troubleshooting.md b/docs/5.-Troubleshooting.md index a59672b..ead28f9 100644 --- a/docs/5.-Troubleshooting.md +++ b/docs/5.-Troubleshooting.md @@ -1,3 +1,11 @@ # Troubleshooting -To be completed when issues arise. +* The payload of incoming messages contains **raw data**. You cannot just print out the data without formatting. This is because Arduino's `print` functions expect a C-string as input and a MQTT payload is not. A simple solution is to print each character of the payload: + +```cpp +for (size_t i = 0; i < len; ++i) { + Serial.print(payload[i]); +} +``` + +Further reading: https://en.wikipedia.org/wiki/C_string_handling diff --git a/examples/FullyFeatured-ESP8266/FullyFeatured-ESP8266.ino b/examples/FullyFeatured-ESP8266/FullyFeatured-ESP8266.ino index 82f981d..3018b6e 100644 --- a/examples/FullyFeatured-ESP8266/FullyFeatured-ESP8266.ino +++ b/examples/FullyFeatured-ESP8266/FullyFeatured-ESP8266.ino @@ -20,6 +20,11 @@ void connectToWifi() { WiFi.begin(WIFI_SSID, WIFI_PASSWORD); } +void connectToMqtt() { + Serial.println("Connecting to MQTT..."); + mqttClient.connect(); +} + void onWifiConnect(const WiFiEventStationModeGotIP& event) { Serial.println("Connected to Wi-Fi."); connectToMqtt(); @@ -31,11 +36,6 @@ void onWifiDisconnect(const WiFiEventStationModeDisconnected& event) { wifiReconnectTimer.once(2, connectToWifi); } -void connectToMqtt() { - Serial.println("Connecting to MQTT..."); - mqttClient.connect(); -} - void onMqttConnect(bool sessionPresent) { Serial.println("Connected to MQTT."); Serial.print("Session present: "); diff --git a/keywords.txt b/keywords.txt index 98c4bf3..ef43170 100644 --- a/keywords.txt +++ b/keywords.txt @@ -33,6 +33,7 @@ disconnect KEYWORD2 subscribe KEYWORD2 unsubscribe KEYWORD2 publish KEYWORD2 +clearQueue KEYWORD2 ####################################### # Constants (LITERAL1) diff --git a/library.json b/library.json index b7f0fd3..9074278 100644 --- a/library.json +++ b/library.json @@ -12,18 +12,18 @@ "type": "git", "url": "https://github.com/marvinroger/async-mqtt-client.git" }, - "version": "0.8.2", + "version": "0.9.0", "frameworks": "arduino", "platforms": ["espressif8266", "espressif32"], "dependencies": [ { "name": "ESPAsyncTCP", - "version": "1.2.0", + "version": ">=1.2.2", "platforms": "espressif8266" }, { "name": "AsyncTCP", - "version": "^1.0.0", + "version": ">=1.1.1", "platforms": "espressif32" } ] diff --git a/library.properties b/library.properties index e2e6037..33136fe 100644 --- a/library.properties +++ b/library.properties @@ -1,5 +1,5 @@ name=AsyncMqttClient -version=0.8.2 +version=0.9.0 author=Marvin ROGER maintainer=Marvin ROGER sentence=An Arduino for ESP8266 and ESP32 asynchronous MQTT client implementation diff --git a/scripts/CI/build_examples_pio.sh b/scripts/CI/build_examples_pio.sh new file mode 100755 index 0000000..8bfd7ce --- /dev/null +++ b/scripts/CI/build_examples_pio.sh @@ -0,0 +1,49 @@ +#!/bin/bash + +#pip install -U platformio +#platformio update +platformio lib -g install AsyncTCP +platformio lib -g install ESPAsyncTCP + +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[0;33m' +NC='\033[0m' + +lines=$(find ./examples/ -maxdepth 1 -mindepth 1 -type d) +retval=0 +while read line; do + if [[ "$line" != *ESP8266 && "$line" != *ESP32 ]] + then + echo -e "========================== BUILDING $line ==========================" + echo -e "${YELLOW}SKIPPING${NC}" + continue + fi + echo -e "========================== BUILDING $line ==========================" + if [[ -e "$line/platformio.ini" ]] + then + # skipping + #output=$(platformio ci --lib="." --project-conf="$line/platformio.ini" $line 2>&1) + : + else + if [[ "$line" == *ESP8266 ]] + then + output=$(platformio ci --lib="." --project-conf="scripts/CI/platformio_esp8266.ini" $line 2>&1) + else + output=$(platformio ci --lib="." --project-conf="scripts/CI/platformio_esp32.ini" $line 2>&1) + fi + fi + if [ $? -ne 0 ]; then + echo "$output" + echo -e "Building $line ${RED}FAILED${NC}" + retval=1 + else + echo -e "${GREEN}SUCCESS${NC}" + fi +done <<< "$lines" + +# cleanup +platformio lib -g uninstall AsyncTCP +platformio lib -g uninstall ESPAsyncTCP + +exit "$retval" diff --git a/scripts/CI/platformio_esp32.ini b/scripts/CI/platformio_esp32.ini new file mode 100644 index 0000000..9e6ed40 --- /dev/null +++ b/scripts/CI/platformio_esp32.ini @@ -0,0 +1,16 @@ +; PlatformIO Project Configuration File +; +; Build options: build flags, source filter +; Upload options: custom upload port, speed and extra flags +; Library options: dependencies, extra library storages +; Advanced options: extra scripting +; +; Please visit documentation for the other options and examples +; https://docs.platformio.org/page/projectconf.html + +[env:esp32] +platform = espressif32 +board = esp32dev +framework = arduino +build_flags = + -Wall \ No newline at end of file diff --git a/scripts/CI/platformio_esp8266.ini b/scripts/CI/platformio_esp8266.ini new file mode 100644 index 0000000..e0b87d9 --- /dev/null +++ b/scripts/CI/platformio_esp8266.ini @@ -0,0 +1,16 @@ +; PlatformIO Project Configuration File +; +; Build options: build flags, source filter +; Upload options: custom upload port, speed and extra flags +; Library options: dependencies, extra library storages +; Advanced options: extra scripting +; +; Please visit documentation for the other options and examples +; https://docs.platformio.org/page/projectconf.html + +[env:esp8266] +platform = espressif8266 +board = esp01_1m +framework = arduino +build_flags = + -Wall diff --git a/src/AsyncMqttClient.cpp b/src/AsyncMqttClient.cpp index fb25c54..b4375cf 100644 --- a/src/AsyncMqttClient.cpp +++ b/src/AsyncMqttClient.cpp @@ -1,12 +1,17 @@ #include "AsyncMqttClient.hpp" AsyncMqttClient::AsyncMqttClient() -: _connected(false) -, _disconnectOnPoll(false) +: _client() +, _head(nullptr) +, _tail(nullptr) +, _sent(0) +, _state(DISCONNECTED) , _disconnectReason(AsyncMqttClientDisconnectReason::TCP_DISCONNECTED) , _lastClientActivity(0) , _lastServerActivity(0) , _lastPingRequestTime(0) +, _generatedClientId{0} +, _ip() , _host(nullptr) , _useIp(false) #if ASYNC_TCP_SSL_ENABLED @@ -23,18 +28,28 @@ AsyncMqttClient::AsyncMqttClient() , _willPayloadLength(0) , _willQos(0) , _willRetain(false) +#if ASYNC_TCP_SSL_ENABLED +, _secureServerFingerprints() +#endif +, _onConnectUserCallbacks() +, _onDisconnectUserCallbacks() +, _onSubscribeUserCallbacks() +, _onUnsubscribeUserCallbacks() +, _onMessageUserCallbacks() +, _onPublishUserCallbacks() , _parsingInformation { .bufferState = AsyncMqttClientInternals::BufferState::NONE } , _currentParsedPacket(nullptr) , _remainingLengthBufferPosition(0) -, _nextPacketId(1) { - _client.onConnect([](void* obj, AsyncClient* c) { (static_cast(obj))->_onConnect(c); }, this); - _client.onDisconnect([](void* obj, AsyncClient* c) { (static_cast(obj))->_onDisconnect(c); }, this); - _client.onError([](void* obj, AsyncClient* c, int8_t error) { (static_cast(obj))->_onError(c, error); }, this); - _client.onTimeout([](void* obj, AsyncClient* c, uint32_t time) { (static_cast(obj))->_onTimeout(c, time); }, this); - _client.onAck([](void* obj, AsyncClient* c, size_t len, uint32_t time) { (static_cast(obj))->_onAck(c, len, time); }, this); - _client.onData([](void* obj, AsyncClient* c, void* data, size_t len) { (static_cast(obj))->_onData(c, static_cast(data), len); }, this); - _client.onPoll([](void* obj, AsyncClient* c) { (static_cast(obj))->_onPoll(c); }, this); - +, _remainingLengthBuffer{0} +, _pendingPubRels() { + _client.onConnect([](void* obj, AsyncClient* c) { (static_cast(obj))->_onConnect(); }, this); + _client.onDisconnect([](void* obj, AsyncClient* c) { (static_cast(obj))->_onDisconnect(); }, this); + // _client.onError([](void* obj, AsyncClient* c, int8_t error) { (static_cast(obj))->_onError(error); }, this); + // _client.onTimeout([](void* obj, AsyncClient* c, uint32_t time) { (static_cast(obj))->_onTimeout(); }, this); + _client.onAck([](void* obj, AsyncClient* c, size_t len, uint32_t time) { (static_cast(obj))->_onAck(len); }, this); + _client.onData([](void* obj, AsyncClient* c, void* data, size_t len) { (static_cast(obj))->_onData(static_cast(data), len); }, this); + _client.onPoll([](void* obj, AsyncClient* c) { (static_cast(obj))->_onPoll(); }, this); + _client.setNoDelay(true); // send small packets immediately (PINGREQ/DISCONN are only 2 bytes) #ifdef ESP32 sprintf(_generatedClientId, "esp32-%06llx", ESP.getEfuseMac()); _xSemaphore = xSemaphoreCreateMutex(); @@ -49,6 +64,10 @@ AsyncMqttClient::AsyncMqttClient() AsyncMqttClient::~AsyncMqttClient() { delete _currentParsedPacket; delete[] _parsingInformation.topicBuffer; + _clear(); + _pendingPubRels.clear(); + _pendingPubRels.shrink_to_fit(); + _clearQueue(false); // _clear() doesn't clear session data #ifdef ESP32 vSemaphoreDelete(_xSemaphore); #endif @@ -156,25 +175,17 @@ void AsyncMqttClient::_freeCurrentParsedPacket() { void AsyncMqttClient::_clear() { _lastPingRequestTime = 0; - _connected = false; - _disconnectOnPoll = false; - _disconnectReason = AsyncMqttClientDisconnectReason::TCP_DISCONNECTED; // reset any previous _freeCurrentParsedPacket(); + _clearQueue(true); // keep session data for now - _pendingPubRels.clear(); - _pendingPubRels.shrink_to_fit(); - - _toSendAcks.clear(); - _toSendAcks.shrink_to_fit(); - - _nextPacketId = 1; _parsingInformation.bufferState = AsyncMqttClientInternals::BufferState::NONE; + + _client.setRxTimeout(0); } /* TCP */ -void AsyncMqttClient::_onConnect(AsyncClient* client) { - (void)client; - +void AsyncMqttClient::_onConnect() { + log_i("TCP conn, MQTT CONNECT"); #if ASYNC_TCP_SSL_ENABLED if (_secure && _secureServerFingerprints.size() > 0) { SSL* clientSsl = _client.getSSL(); @@ -194,192 +205,51 @@ void AsyncMqttClient::_onConnect(AsyncClient* client) { } } #endif + AsyncMqttClientInternals::OutPacket* msg = + new AsyncMqttClientInternals::ConnectOutPacket(_cleanSession, + _username, + _password, + _willTopic, + _willRetain, + _willQos, + _willPayload, + _willPayloadLength, + _keepAlive, + _clientId); + _addFront(msg); + _handleQueue(); +} + +void AsyncMqttClient::_onDisconnect() { + log_i("TCP disconn"); + _state = DISCONNECTED; - 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) { - _disconnectReason = AsyncMqttClientDisconnectReason::ESP8266_NOT_ENOUGH_SPACE; - _client.close(true); - SEMAPHORE_GIVE(); - return; - } - - _client.add(fixedHeader, 1 + remainingLengthLength); - - // Using a sendbuffer to fix bug setwill on SSL not working - char sendbuffer[12]; - sendbuffer[0] = protocolNameLengthBytes[0]; - sendbuffer[1] = protocolNameLengthBytes[1]; - - sendbuffer[2] = 'M'; - sendbuffer[3] = 'Q'; - sendbuffer[4] = 'T'; - sendbuffer[5] = 'T'; - - sendbuffer[6] = protocolLevel[0]; - sendbuffer[7] = connectFlags[0]; - sendbuffer[8] = keepAliveBytes[0]; - sendbuffer[9] = keepAliveBytes[1]; - sendbuffer[10] = clientIdLengthBytes[0]; - sendbuffer[11] = clientIdLengthBytes[1]; - - _client.add(sendbuffer, 12); - - _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; - AsyncMqttClientDisconnectReason reason = _disconnectReason; _clear(); - for (auto callback : _onDisconnectUserCallbacks) callback(reason); + for (auto callback : _onDisconnectUserCallbacks) callback(_disconnectReason); } -void AsyncMqttClient::_onError(AsyncClient* client, int8_t error) { - (void)client; +/* +void AsyncMqttClient::_onError(int8_t error) { (void)error; // _onDisconnect called anyway } -void AsyncMqttClient::_onTimeout(AsyncClient* client, uint32_t time) { - (void)client; - (void)time; +void AsyncMqttClient::_onTimeout() { // 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::_onAck(size_t len) { + log_i("ack %u", len); + _handleQueue(); } -void AsyncMqttClient::_onData(AsyncClient* client, char* data, size_t len) { - (void)client; +void AsyncMqttClient::_onData(char* data, size_t len) { + log_i("data rcv (%u)", len); size_t currentBytePosition = 0; char currentByte; + _lastServerActivity = millis(); do { switch (_parsingInformation.bufferState) { case AsyncMqttClientInternals::BufferState::NONE: @@ -387,36 +257,47 @@ void AsyncMqttClient::_onData(AsyncClient* client, char* data, size_t len) { _parsingInformation.packetType = currentByte >> 4; _parsingInformation.packetFlags = (currentByte << 4) >> 4; _parsingInformation.bufferState = AsyncMqttClientInternals::BufferState::REMAINING_LENGTH; - _lastServerActivity = millis(); switch (_parsingInformation.packetType) { case AsyncMqttClientInternals::PacketType.CONNACK: + log_i("rcv CONNACK"); _currentParsedPacket = new AsyncMqttClientInternals::ConnAckPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onConnAck, this, std::placeholders::_1, std::placeholders::_2)); + _client.setRxTimeout(0); break; case AsyncMqttClientInternals::PacketType.PINGRESP: + log_i("rcv PINGRESP"); _currentParsedPacket = new AsyncMqttClientInternals::PingRespPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPingResp, this)); break; case AsyncMqttClientInternals::PacketType.SUBACK: + log_i("rcv SUBACK"); _currentParsedPacket = new AsyncMqttClientInternals::SubAckPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onSubAck, this, std::placeholders::_1, std::placeholders::_2)); break; case AsyncMqttClientInternals::PacketType.UNSUBACK: + log_i("rcv UNSUBACK"); _currentParsedPacket = new AsyncMqttClientInternals::UnsubAckPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onUnsubAck, this, std::placeholders::_1)); break; case AsyncMqttClientInternals::PacketType.PUBLISH: + log_i("rcv 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: + log_i("rcv PUBREL"); _currentParsedPacket = new AsyncMqttClientInternals::PubRelPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPubRel, this, std::placeholders::_1)); break; case AsyncMqttClientInternals::PacketType.PUBACK: + log_i("rcv PUBACK"); _currentParsedPacket = new AsyncMqttClientInternals::PubAckPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPubAck, this, std::placeholders::_1)); break; case AsyncMqttClientInternals::PacketType.PUBREC: + log_i("rcv PUBREC"); _currentParsedPacket = new AsyncMqttClientInternals::PubRecPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPubRec, this, std::placeholders::_1)); break; case AsyncMqttClientInternals::PacketType.PUBCOMP: + log_i("rcv PUBCOMP"); _currentParsedPacket = new AsyncMqttClientInternals::PubCompPacket(&_parsingInformation, std::bind(&AsyncMqttClient::_onPubComp, this, std::placeholders::_1)); break; default: + log_i("rcv PROTOCOL VIOLATION"); + disconnect(true); break; } break; @@ -447,63 +328,226 @@ void AsyncMqttClient::_onData(AsyncClient* client, char* data, size_t len) { } while (currentBytePosition != len); } -void AsyncMqttClient::_onPoll(AsyncClient* client) { - if (!_connected) return; - +void AsyncMqttClient::_onPoll() { // 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(); + log_w("PING t/o, disconnecting"); + disconnect(true); 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)) { + if (_state == CONNECTED && _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)) { + } else if (_state == CONNECTED && _lastPingRequestTime == 0 && (millis() - _lastServerActivity) >= (_keepAlive * 1000 * 0.7)) { _sendPing(); } + _handleQueue(); +} +/* QUEUE */ - // handle to send ack packets +void AsyncMqttClient::_insert(AsyncMqttClientInternals::OutPacket* packet) { + // We only use this for QoS2 PUBREL so there must be a PUBLISH packet present. + // The queue therefore cannot be empty and _head points to this PUBLISH packet. + SEMAPHORE_TAKE(); + log_i("new insert #%u", packet->packetType()); + packet->next = _head->next; + _head->next = packet; + if (_head == _tail) { // PUB packet is the only one in the queue + _tail = packet; + } + SEMAPHORE_GIVE(); + _handleQueue(); +} - _sendAcks(); +void AsyncMqttClient::_addFront(AsyncMqttClientInternals::OutPacket* packet) { + // This is only used for the CONNECT packet, to be able to establish a connection + // before anything else. The queue can be empty or has packets from the continued session. + // In both cases, _head should always point to the CONNECT packet afterwards. + SEMAPHORE_TAKE(); + log_i("new front #%u", packet->packetType()); + if (_head == nullptr) { + _tail = packet; + } else { + packet->next = _head; + } + _head = packet; + SEMAPHORE_GIVE(); + _handleQueue(); +} - // handle disconnect +void AsyncMqttClient::_addBack(AsyncMqttClientInternals::OutPacket* packet) { + SEMAPHORE_TAKE(); + log_i("new back #%u", packet->packetType()); + if (!_tail) { + _head = packet; + } else { + _tail->next = packet; + } + _tail = packet; + _tail->next = nullptr; + SEMAPHORE_GIVE(); + _handleQueue(); +} - if (_disconnectOnPoll) { - _sendDisconnect(); +void AsyncMqttClient::_handleQueue() { + SEMAPHORE_TAKE(); + // On ESP32, onDisconnect is called within the close()-call. So we need to make sure we don't lock + bool disconnect = false; + + while (_head && _client.space() > 10) { // safe but arbitrary value, send at least 10 bytes + // 1. try to send + if (_head->size() > _sent) { + // On SSL the TCP library returns the total amount of bytes, not just the unencrypted payload length. + // So we calculate the amount to be written ourselves. + size_t willSend = std::min(_head->size() - _sent, _client.space()); + size_t realSent = _client.add(reinterpret_cast(_head->data(_sent)), willSend, ASYNC_WRITE_FLAG_COPY); // flag is set by LWIP anyway, added for clarity + _sent += willSend; + (void)realSent; + _client.send(); + _lastClientActivity = millis(); + _lastPingRequestTime = 0; + #if ASYNC_TCP_SSL_ENABLED + log_i("snd #%u: (tls: %u) %u/%u", _head->packetType(), realSent, _sent, _head->size()); + #else + log_i("snd #%u: %u/%u", _head->packetType(), _sent, _head->size()); + #endif + if (_head->packetType() == AsyncMqttClientInternals::PacketType.DISCONNECT) { + disconnect = true; + } + } + + // 2. stop processing when we have to wait for an MQTT acknowledgment + if (_head->size() == _sent) { + if (_head->released()) { + log_i("p #%d rel", _head->packetType()); + AsyncMqttClientInternals::OutPacket* tmp = _head; + _head = _head->next; + if (!_head) _tail = nullptr; + delete tmp; + _sent = 0; + } else { + break; // sending is complete however send next only after mqtt confirmation + } + } + } + + SEMAPHORE_GIVE(); + if (disconnect) { + log_i("snd DISCONN, disconnecting"); + _client.close(); } } +void AsyncMqttClient::_clearQueue(bool keepSessionData) { + SEMAPHORE_TAKE(); + AsyncMqttClientInternals::OutPacket* packet = _head; + _head = nullptr; + _tail = nullptr; + + while (packet) { + /* MQTT spec 3.1.2.4 Clean Session: + * - QoS 1 and QoS 2 messages which have been sent to the Server, but have not been completely acknowledged. + * - QoS 2 messages which have been received from the Server, but have not been completely acknowledged. + * + (unsent PUB messages with QoS > 0) + * + * To be kept: + * - possibly first message (sent to server but not acked) + * - PUBREC messages (QoS 2 PUB received but not acked) + * - PUBCOMP messages (QoS 2 PUBREL received but not acked) + */ + if (keepSessionData) { + if (packet->qos() > 0 && packet->size() <= _sent) { // check for qos includes check for PUB-packet type + reinterpret_cast(packet)->setDup(); + AsyncMqttClientInternals::OutPacket* next = packet->next; + log_i("keep #%u", packet->packetType()); + SEMAPHORE_GIVE(); + _addBack(packet); + SEMAPHORE_TAKE(); + packet = next; + } else if (packet->qos() > 0 || + packet->packetType() == AsyncMqttClientInternals::PacketType.PUBREC || + packet->packetType() == AsyncMqttClientInternals::PacketType.PUBCOMP) { + AsyncMqttClientInternals::OutPacket* next = packet->next; + log_i("keep #%u", packet->packetType()); + SEMAPHORE_GIVE(); + _addBack(packet); + SEMAPHORE_TAKE(); + packet = next; + } else { + AsyncMqttClientInternals::OutPacket* next = packet->next; + delete packet; + packet = next; + } + /* Delete everything when not keeping session data + */ + } else { + AsyncMqttClientInternals::OutPacket* next = packet->next; + delete packet; + packet = next; + } + } + _sent = 0; + SEMAPHORE_GIVE(); +} + /* MQTT */ void AsyncMqttClient::_onPingResp() { + log_i("PINGRESP"); _freeCurrentParsedPacket(); _lastPingRequestTime = 0; } void AsyncMqttClient::_onConnAck(bool sessionPresent, uint8_t connectReturnCode) { - (void)sessionPresent; + log_i("CONNACK"); _freeCurrentParsedPacket(); + if (!sessionPresent) { + _pendingPubRels.clear(); + _pendingPubRels.shrink_to_fit(); + _clearQueue(false); // remove session data + } + if (connectReturnCode == 0) { - _connected = true; + _state = CONNECTED; for (auto callback : _onConnectUserCallbacks) callback(sessionPresent); } else { - // Callbacks are handled by the ondisconnect function which is called from the AsyncTcp lib + // Callbacks are handled by the onDisconnect function which is called from the AsyncTcp lib _disconnectReason = static_cast(connectReturnCode); + return; } + _handleQueue(); // send any remaining data from continued session } void AsyncMqttClient::_onSubAck(uint16_t packetId, char status) { + log_i("SUBACK"); _freeCurrentParsedPacket(); + SEMAPHORE_TAKE(); + if (_head && _head->packetId() == packetId) { + _head->release(); + log_i("SUB released"); + } + SEMAPHORE_GIVE(); for (auto callback : _onSubscribeUserCallbacks) callback(packetId, status); + + _handleQueue(); // subscribe confirmed, ready to send next queued item } void AsyncMqttClient::_onUnsubAck(uint16_t packetId) { + log_i("UNSUBACK"); _freeCurrentParsedPacket(); + SEMAPHORE_TAKE(); + if (_head && _head->packetId() == packetId) { + _head->release(); + log_i("UNSUB released"); + } + SEMAPHORE_GIVE(); for (auto callback : _onUnsubscribeUserCallbacks) callback(packetId); + + _handleQueue(); // unsubscribe confirmed, ready to send next queued item } 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) { @@ -535,12 +579,14 @@ void AsyncMqttClient::_onPublish(uint16_t packetId, uint8_t qos) { pendingAck.packetType = AsyncMqttClientInternals::PacketType.PUBACK; pendingAck.headerFlag = AsyncMqttClientInternals::HeaderFlag.PUBACK_RESERVED; pendingAck.packetId = packetId; - _toSendAcks.push_back(pendingAck); + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::PubAckOutPacket(pendingAck); + _addBack(msg); } else if (qos == 2) { pendingAck.packetType = AsyncMqttClientInternals::PacketType.PUBREC; pendingAck.headerFlag = AsyncMqttClientInternals::HeaderFlag.PUBREC_RESERVED; pendingAck.packetId = packetId; - _toSendAcks.push_back(pendingAck); + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::PubAckOutPacket(pendingAck); + _addBack(msg); bool pubRelAwaiting = false; for (AsyncMqttClientInternals::PendingPubRel pendingPubRel : _pendingPubRels) { @@ -555,8 +601,6 @@ void AsyncMqttClient::_onPublish(uint16_t packetId, uint8_t qos) { pendingPubRel.packetId = packetId; _pendingPubRels.push_back(pendingPubRel); } - - _sendAcks(); } _freeCurrentParsedPacket(); @@ -569,7 +613,12 @@ void AsyncMqttClient::_onPubRel(uint16_t packetId) { pendingAck.packetType = AsyncMqttClientInternals::PacketType.PUBCOMP; pendingAck.headerFlag = AsyncMqttClientInternals::HeaderFlag.PUBCOMP_RESERVED; pendingAck.packetId = packetId; - _toSendAcks.push_back(pendingAck); + if (_head && _head->packetId() == packetId) { + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::PubAckOutPacket(pendingAck); + _head->release(); + _insert(msg); + log_i("PUBREC released"); + } for (size_t i = 0; i < _pendingPubRels.size(); i++) { if (_pendingPubRels[i].packetId == packetId) { @@ -577,12 +626,14 @@ void AsyncMqttClient::_onPubRel(uint16_t packetId) { _pendingPubRels.shrink_to_fit(); } } - - _sendAcks(); } void AsyncMqttClient::_onPubAck(uint16_t packetId) { _freeCurrentParsedPacket(); + if (_head && _head->packetId() == packetId) { + _head->release(); + log_i("PUB released"); + } for (auto callback : _onPublishUserCallbacks) callback(packetId); } @@ -590,113 +641,53 @@ void AsyncMqttClient::_onPubAck(uint16_t packetId) { void AsyncMqttClient::_onPubRec(uint16_t packetId) { _freeCurrentParsedPacket(); + // We will only be sending 1 QoS>0 PUB message at a time (to honor message + // ordering). So no need to store ACKS in a separate container as it will + // be stored in the outgoing queue until a PUBCOMP comes in. AsyncMqttClientInternals::PendingAck pendingAck; pendingAck.packetType = AsyncMqttClientInternals::PacketType.PUBREL; pendingAck.headerFlag = AsyncMqttClientInternals::HeaderFlag.PUBREL_RESERVED; pendingAck.packetId = packetId; - _toSendAcks.push_back(pendingAck); + log_i("snd PUBREL"); - _sendAcks(); + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::PubAckOutPacket(pendingAck); + if (_head && _head->packetId() == packetId) { + _head->release(); + log_i("PUB released"); + } + _insert(msg); } 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(); + // _head points to the PUBREL package + if (_head && _head->packetId() == packetId) { + _head->release(); + log_i("PUBREL released"); } - 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); - - _disconnectOnPoll = false; - - SEMAPHORE_GIVE(); - return true; + for (auto callback : _onPublishUserCallbacks) callback(packetId); } -uint16_t AsyncMqttClient::_getNextPacketId() { - uint16_t nextPacketId = _nextPacketId; - - if (_nextPacketId == 65535) _nextPacketId = 0; // 0 is forbidden - _nextPacketId++; - - return nextPacketId; +void AsyncMqttClient::_sendPing() { + log_i("PING"); + _lastPingRequestTime = millis(); + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::PingReqOutPacket; + _addBack(msg); } bool AsyncMqttClient::connected() const { - return _connected; + return _state == CONNECTED; } void AsyncMqttClient::connect() { - if (_connected) return; + if (_state != DISCONNECTED) return; + log_i("CONNECTING"); + _state = CONNECTING; + _disconnectReason = AsyncMqttClientDisconnectReason::TCP_DISCONNECTED; // reset any previous + + _client.setRxTimeout(_keepAlive); #if ASYNC_TCP_SSL_ENABLED if (_useIp) { @@ -714,172 +705,51 @@ void AsyncMqttClient::connect() { } void AsyncMqttClient::disconnect(bool force) { - if (!_connected) return; - + if (_state == DISCONNECTED) return; + log_i("DISCONNECT (f:%d)", force); if (force) { + _state = DISCONNECTED; _client.close(true); - } else { - _sendDisconnect(); - _disconnectOnPoll = false; + } else if (_state != DISCONNECTING) { + _state = DISCONNECTING; + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::DisconnOutPacket; + _addBack(msg); } } 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(); + if (_state != CONNECTED) return 0; + log_i("SUBSCRIBE"); - SEMAPHORE_GIVE(); - return packetId; + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::SubscribeOutPacket(topic, qos); + _addBack(msg); + return msg->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; + if (_state != CONNECTED) return 0; + log_i("UNSUBSCRIBE"); - 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; + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::UnsubscribeOutPacket(topic); + _addBack(msg); + return msg->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; - } + if (_state != CONNECTED || GET_FREE_MEMORY() < MQTT_MIN_FREE_MEMORY) return 0; + log_i("PUBLISH"); - _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(); + AsyncMqttClientInternals::OutPacket* msg = new AsyncMqttClientInternals::PublishOutPacket(topic, qos, retain, payload, length); + _addBack(msg); + return msg->packetId(); +} - SEMAPHORE_GIVE(); - if (qos != 0) { - return packetId; - } else { - return 1; - } +bool AsyncMqttClient::clearQueue() { + if (_state != DISCONNECTED) return false; + _clearQueue(false); + return true; } -const char* AsyncMqttClient::getClientId() { +const char* AsyncMqttClient::getClientId() const { return _clientId; -} \ No newline at end of file +} diff --git a/src/AsyncMqttClient.hpp b/src/AsyncMqttClient.hpp index 4c97745..1e81103 100644 --- a/src/AsyncMqttClient.hpp +++ b/src/AsyncMqttClient.hpp @@ -5,6 +5,10 @@ #include "Arduino.h" +#ifndef MQTT_MIN_FREE_MEMORY +#define MQTT_MIN_FREE_MEMORY 4096 +#endif + #ifdef ESP32 #include #include @@ -38,13 +42,13 @@ #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 +#include "AsyncMqttClient/Packets/Out/Connect.hpp" +#include "AsyncMqttClient/Packets/Out/PingReq.hpp" +#include "AsyncMqttClient/Packets/Out/PubAck.hpp" +#include "AsyncMqttClient/Packets/Out/Disconn.hpp" +#include "AsyncMqttClient/Packets/Out/Subscribe.hpp" +#include "AsyncMqttClient/Packets/Out/Unsubscribe.hpp" +#include "AsyncMqttClient/Packets/Out/Publish.hpp" class AsyncMqttClient { public: @@ -77,20 +81,27 @@ class AsyncMqttClient { 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); + bool clearQueue(); // Not MQTT compliant! - const char* getClientId(); + const char* getClientId() const; private: AsyncClient _client; - - bool _connected; - bool _disconnectOnPoll; + AsyncMqttClientInternals::OutPacket* _head; + AsyncMqttClientInternals::OutPacket* _tail; + size_t _sent; + enum { + CONNECTING, + CONNECTED, + DISCONNECTING, + DISCONNECTED + } _state; AsyncMqttClientDisconnectReason _disconnectReason; uint32_t _lastClientActivity; uint32_t _lastServerActivity; uint32_t _lastPingRequestTime; - char _generatedClientId[18 + 1]; // esp8266-abc123 and esp32-abcdef123456 + char _generatedClientId[18 + 1]; // esp8266-abc123 and esp32-abcdef123456 IPAddress _ip; const char* _host; bool _useIp; @@ -125,27 +136,32 @@ class AsyncMqttClient { uint8_t _remainingLengthBufferPosition; char _remainingLengthBuffer[4]; - uint16_t _nextPacketId; - std::vector _pendingPubRels; - std::vector _toSendAcks; - -#ifdef ESP32 +#if defined(ESP32) SemaphoreHandle_t _xSemaphore = nullptr; +#elif defined(ESP8266) + bool _xSemaphore = false; #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); + void _onConnect(); + void _onDisconnect(); + // void _onError(int8_t error); + // void _onTimeout(); + void _onAck(size_t len); + void _onData(char* data, size_t len); + void _onPoll(); + + // QUEUE + void _insert(AsyncMqttClientInternals::OutPacket* packet); // for PUBREL + void _addFront(AsyncMqttClientInternals::OutPacket* packet); // for CONNECT + void _addBack(AsyncMqttClientInternals::OutPacket* packet); // all the rest + void _handleQueue(); + void _clearQueue(bool keepSessionData); // MQTT void _onPingResp(); @@ -159,9 +175,5 @@ class AsyncMqttClient { void _onPubRec(uint16_t packetId); void _onPubComp(uint16_t packetId); - bool _sendPing(); - void _sendAcks(); - bool _sendDisconnect(); - - uint16_t _getNextPacketId(); + void _sendPing(); }; diff --git a/src/AsyncMqttClient/Callbacks.hpp b/src/AsyncMqttClient/Callbacks.hpp index 2a4a09f..414034d 100644 --- a/src/AsyncMqttClient/Callbacks.hpp +++ b/src/AsyncMqttClient/Callbacks.hpp @@ -4,6 +4,7 @@ #include "DisconnectReasons.hpp" #include "MessageProperties.hpp" +#include "Errors.hpp" namespace AsyncMqttClientInternals { // user callbacks @@ -13,6 +14,7 @@ typedef std::function OnSubscribeUserCallb typedef std::function OnUnsubscribeUserCallback; typedef std::function OnMessageUserCallback; typedef std::function OnPublishUserCallback; +typedef std::function OnErrorUserCallback; // internal callbacks typedef std::function OnConnAckInternalCallback; diff --git a/src/AsyncMqttClient/Errors.hpp b/src/AsyncMqttClient/Errors.hpp new file mode 100644 index 0000000..f93e80e --- /dev/null +++ b/src/AsyncMqttClient/Errors.hpp @@ -0,0 +1,6 @@ +#pragma once + +enum class AsyncMqttClientError : uint8_t { + MAX_RETRIES = 0, + OUT_OF_MEMORY = 1 +}; diff --git a/src/AsyncMqttClient/Helpers.hpp b/src/AsyncMqttClient/Helpers.hpp index 7113033..ecb620f 100644 --- a/src/AsyncMqttClient/Helpers.hpp +++ b/src/AsyncMqttClient/Helpers.hpp @@ -35,4 +35,27 @@ class Helpers { return bytesNeeded; } }; + +#if defined(ARDUINO_ARCH_ESP32) + #define SEMAPHORE_TAKE() xSemaphoreTake(_xSemaphore, portMAX_DELAY) + #define SEMAPHORE_GIVE() xSemaphoreGive(_xSemaphore) + #define GET_FREE_MEMORY() ESP.getMaxAllocHeap() + #include +#elif defined(ARDUINO_ARCH_ESP8266) + #define SEMAPHORE_TAKE(X) while (_xSemaphore) { /*ESP.wdtFeed();*/ } _xSemaphore = true + #define SEMAPHORE_GIVE() _xSemaphore = false + #define GET_FREE_MEMORY() ESP.getMaxFreeBlockSize() + #if defined(DEBUG_ESP_PORT) && defined(DEBUG_ASYNC_MQTT_CLIENT) + #define log_i(...) DEBUG_ESP_PORT.printf(__VA_ARGS__); DEBUG_ESP_PORT.print("\n") + #define log_e(...) DEBUG_ESP_PORT.printf(__VA_ARGS__); DEBUG_ESP_PORT.print("\n") + #define log_w(...) DEBUG_ESP_PORT.printf(__VA_ARGS__); DEBUG_ESP_PORT.print("\n") + #else + #define log_i(...) + #define log_e(...) + #define log_w(...) + #endif +#else + #pragma error "No valid architecture" +#endif + } // namespace AsyncMqttClientInternals diff --git a/src/AsyncMqttClient/Packets/Out/Connect.cpp b/src/AsyncMqttClient/Packets/Out/Connect.cpp new file mode 100644 index 0000000..a9a86e4 --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Connect.cpp @@ -0,0 +1,162 @@ +#include "Connect.hpp" + +using AsyncMqttClientInternals::ConnectOutPacket; + +ConnectOutPacket::ConnectOutPacket(bool cleanSession, + const char* username, + const char* password, + const char* willTopic, + bool willRetain, + uint8_t willQos, + const char* willPayload, + uint16_t willPayloadLength, + uint16_t keepAlive, + const char* clientId) { + 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]; + 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; + } + + _data.reserve(neededSpace); + + _data.insert(_data.end(), fixedHeader, fixedHeader + 1 + remainingLengthLength); + + _data.push_back(protocolNameLengthBytes[0]); + _data.push_back(protocolNameLengthBytes[1]); + + _data.push_back('M'); + _data.push_back('Q'); + _data.push_back('T'); + _data.push_back('T'); + + _data.push_back(protocolLevel[0]); + _data.push_back(connectFlags[0]); + _data.push_back(keepAliveBytes[0]); + _data.push_back(keepAliveBytes[1]); + _data.push_back(clientIdLengthBytes[0]); + _data.push_back(clientIdLengthBytes[1]); + + _data.insert(_data.end(), clientId, clientId + clientIdLength); + if (willTopic != nullptr) { + _data.insert(_data.end(), willTopicLengthBytes, willTopicLengthBytes + 2); + _data.insert(_data.end(), willTopic, willTopic + willTopicLength); + + _data.insert(_data.end(), willPayloadLengthBytes, willPayloadLengthBytes + 2); + if (willPayload != nullptr) _data.insert(_data.end(), willPayload, willPayload + willPayloadLength); + } + if (username != nullptr) { + _data.insert(_data.end(), usernameLengthBytes, usernameLengthBytes + 2); + _data.insert(_data.end(), username, username + usernameLength); + } + if (password != nullptr) { + _data.insert(_data.end(), passwordLengthBytes, passwordLengthBytes + 2); + _data.insert(_data.end(), password, password + passwordLength); + } +} + +const uint8_t* ConnectOutPacket::data(size_t index) const { + return &_data.data()[index]; +} + +size_t ConnectOutPacket::size() const { + return _data.size(); +} diff --git a/src/AsyncMqttClient/Packets/Out/Connect.hpp b/src/AsyncMqttClient/Packets/Out/Connect.hpp new file mode 100644 index 0000000..5b17632 --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Connect.hpp @@ -0,0 +1,29 @@ +#pragma once + +#include +#include // strlen + +#include "OutPacket.hpp" +#include "../../Flags.hpp" +#include "../../Helpers.hpp" + +namespace AsyncMqttClientInternals { +class ConnectOutPacket : public OutPacket { + public: + ConnectOutPacket(bool cleanSession, + const char* username, + const char* password, + const char* willTopic, + bool willRetain, + uint8_t willQos, + const char* willPayload, + uint16_t willPayloadLength, + uint16_t keepAlive, + const char* clientId); + const uint8_t* data(size_t index = 0) const; + size_t size() const; + + private: + std::vector _data; +}; +} // namespace AsyncMqttClientInternals diff --git a/src/AsyncMqttClient/Packets/Out/Disconn.cpp b/src/AsyncMqttClient/Packets/Out/Disconn.cpp new file mode 100644 index 0000000..3e2890d --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Disconn.cpp @@ -0,0 +1,18 @@ +#include "Disconn.hpp" + +using AsyncMqttClientInternals::DisconnOutPacket; + +DisconnOutPacket::DisconnOutPacket() { + _data[0] = AsyncMqttClientInternals::PacketType.DISCONNECT; + _data[0] = _data[0] << 4; + _data[0] = _data[0] | AsyncMqttClientInternals::HeaderFlag.DISCONNECT_RESERVED; + _data[1] = 0; +} + +const uint8_t* DisconnOutPacket::data(size_t index) const { + return &_data[index]; +} + +size_t DisconnOutPacket::size() const { + return 2; +} diff --git a/src/AsyncMqttClient/Packets/Out/Disconn.hpp b/src/AsyncMqttClient/Packets/Out/Disconn.hpp new file mode 100644 index 0000000..38dc915 --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Disconn.hpp @@ -0,0 +1,17 @@ +#pragma once + +#include "OutPacket.hpp" +#include "../../Flags.hpp" +#include "../../Helpers.hpp" + +namespace AsyncMqttClientInternals { +class DisconnOutPacket : public OutPacket { + public: + DisconnOutPacket(); + const uint8_t* data(size_t index = 0) const; + size_t size() const; + + private: + uint8_t _data[2]; +}; +} // namespace AsyncMqttClientInternals diff --git a/src/AsyncMqttClient/Packets/Out/OutPacket.cpp b/src/AsyncMqttClient/Packets/Out/OutPacket.cpp new file mode 100644 index 0000000..e69a87f --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/OutPacket.cpp @@ -0,0 +1,44 @@ +#include "OutPacket.hpp" + +using AsyncMqttClientInternals::OutPacket; + +OutPacket::OutPacket() +: next(nullptr) +, timeout(0) +, noTries(0) +, _released(true) +, _packetId(0) {} + +OutPacket::~OutPacket() {} + +bool OutPacket::released() const { + return _released; +} + +uint8_t OutPacket::packetType() const { + return data(0)[0] >> 4; +} + +uint16_t OutPacket::packetId() const { + return _packetId; +} + +uint8_t OutPacket::qos() const { + if (packetType() == AsyncMqttClientInternals::PacketType.PUBLISH) { + return (data()[1] & 0x06) >> 1; + } + return 0; +} + +void OutPacket::release() { + _released = true; +} + +uint16_t OutPacket::_nextPacketId = 0; + +uint16_t OutPacket::_getNextPacketId() { + if (++_nextPacketId == 0) { + ++_nextPacketId; + } + return _nextPacketId; +} diff --git a/src/AsyncMqttClient/Packets/Out/OutPacket.hpp b/src/AsyncMqttClient/Packets/Out/OutPacket.hpp new file mode 100644 index 0000000..52c37de --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/OutPacket.hpp @@ -0,0 +1,35 @@ +#pragma once + +#include // uint*_t +#include // size_t +#include // std::min + +#include "../../Flags.hpp" + +namespace AsyncMqttClientInternals { +class OutPacket { + public: + OutPacket(); + virtual ~OutPacket(); + virtual const uint8_t* data(size_t index = 0) const = 0; + virtual size_t size() const = 0; + bool released() const; + uint8_t packetType() const; + uint16_t packetId() const; + uint8_t qos() const; + void release(); + + public: + OutPacket* next; + uint32_t timeout; + uint8_t noTries; + + protected: + static uint16_t _getNextPacketId(); + bool _released; + uint16_t _packetId; + + private: + static uint16_t _nextPacketId; +}; +} // namespace AsyncMqttClientInternals diff --git a/src/AsyncMqttClient/Packets/Out/PingReq.cpp b/src/AsyncMqttClient/Packets/Out/PingReq.cpp new file mode 100644 index 0000000..d59cf3d --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/PingReq.cpp @@ -0,0 +1,18 @@ +#include "PingReq.hpp" + +using AsyncMqttClientInternals::PingReqOutPacket; + +PingReqOutPacket::PingReqOutPacket() { + _data[0] = AsyncMqttClientInternals::PacketType.PINGREQ; + _data[0] = _data[0] << 4; + _data[0] = _data[0] | AsyncMqttClientInternals::HeaderFlag.PINGREQ_RESERVED; + _data[1] = 0; +} + +const uint8_t* PingReqOutPacket::data(size_t index) const { + return &_data[index];; +} + +size_t PingReqOutPacket::size() const { + return 2; +} diff --git a/src/AsyncMqttClient/Packets/Out/PingReq.hpp b/src/AsyncMqttClient/Packets/Out/PingReq.hpp new file mode 100644 index 0000000..1cb19a3 --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/PingReq.hpp @@ -0,0 +1,17 @@ +#pragma once + +#include "OutPacket.hpp" +#include "../../Flags.hpp" +#include "../../Helpers.hpp" + +namespace AsyncMqttClientInternals { +class PingReqOutPacket : public OutPacket { + public: + PingReqOutPacket(); + const uint8_t* data(size_t index = 0) const; + size_t size() const; + + private: + uint8_t _data[2]; +}; +} // namespace AsyncMqttClientInternals diff --git a/src/AsyncMqttClient/Packets/Out/PubAck.cpp b/src/AsyncMqttClient/Packets/Out/PubAck.cpp new file mode 100644 index 0000000..634607b --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/PubAck.cpp @@ -0,0 +1,25 @@ +#include "PubAck.hpp" + +using AsyncMqttClientInternals::PubAckOutPacket; + +PubAckOutPacket::PubAckOutPacket(PendingAck pendingAck) { + _data[0] = pendingAck.packetType; + _data[0] = _data[0] << 4; + _data[0] = _data[0] | pendingAck.headerFlag; + _data[1] = 2; + _packetId = pendingAck.packetId; + _data[2] = pendingAck.packetId >> 8; + _data[3] = pendingAck.packetId & 0xFF; + if (packetType() == AsyncMqttClientInternals::PacketType.PUBREL || + packetType() == AsyncMqttClientInternals::PacketType.PUBREC) { + _released = false; + } +} + +const uint8_t* PubAckOutPacket::data(size_t index) const { + return &_data[index]; +} + +size_t PubAckOutPacket::size() const { + return 4; +} diff --git a/src/AsyncMqttClient/Packets/Out/PubAck.hpp b/src/AsyncMqttClient/Packets/Out/PubAck.hpp new file mode 100644 index 0000000..9cd830e --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/PubAck.hpp @@ -0,0 +1,18 @@ +#pragma once + +#include "OutPacket.hpp" +#include "../../Flags.hpp" +#include "../../Helpers.hpp" +#include "../../Storage.hpp" + +namespace AsyncMqttClientInternals { +class PubAckOutPacket : public OutPacket { + public: + explicit PubAckOutPacket(PendingAck pendingAck); + const uint8_t* data(size_t index = 0) const; + size_t size() const; + + private: + uint8_t _data[4]; +}; +} // namespace AsyncMqttClientInternals diff --git a/src/AsyncMqttClient/Packets/Out/Publish.cpp b/src/AsyncMqttClient/Packets/Out/Publish.cpp new file mode 100644 index 0000000..3f4365b --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Publish.cpp @@ -0,0 +1,69 @@ +#include "Publish.hpp" + +using AsyncMqttClientInternals::PublishOutPacket; + +PublishOutPacket::PublishOutPacket(const char* topic, uint8_t qos, bool retain, const char* payload, size_t length) { + 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; + + _data.reserve(neededSpace); + + _packetId = (qos !=0) ? _getNextPacketId() : 1; + char packetIdBytes[2]; + packetIdBytes[0] = _packetId >> 8; + packetIdBytes[1] = _packetId & 0xFF; + + _data.insert(_data.end(), fixedHeader, fixedHeader + 1 + remainingLengthLength); + _data.insert(_data.end(), topicLengthBytes, topicLengthBytes + 2); + _data.insert(_data.end(), topic, topic + topicLength); + if (qos != 0) { + _data.insert(_data.end(), packetIdBytes, packetIdBytes + 2); + _released = false; + } + if (payload != nullptr) _data.insert(_data.end(), payload, payload + payloadLength); +} + +const uint8_t* PublishOutPacket::data(size_t index) const { + return &_data.data()[index]; +} + +size_t PublishOutPacket::size() const { + return _data.size(); +} + +void PublishOutPacket::setDup() { + _data[0] |= AsyncMqttClientInternals::HeaderFlag.PUBLISH_DUP; +} diff --git a/src/AsyncMqttClient/Packets/Out/Publish.hpp b/src/AsyncMqttClient/Packets/Out/Publish.hpp new file mode 100644 index 0000000..6b8272e --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Publish.hpp @@ -0,0 +1,23 @@ +#pragma once + +#include // strlen +#include + +#include "OutPacket.hpp" +#include "../../Flags.hpp" +#include "../../Helpers.hpp" +#include "../../Storage.hpp" + +namespace AsyncMqttClientInternals { +class PublishOutPacket : public OutPacket { + public: + PublishOutPacket(const char* topic, uint8_t qos, bool retain, const char* payload, size_t length); + const uint8_t* data(size_t index = 0) const; + size_t size() const; + + void setDup(); // you cannot unset dup + + private: + std::vector _data; +}; +} // namespace AsyncMqttClientInternals diff --git a/src/AsyncMqttClient/Packets/Out/Subscribe.cpp b/src/AsyncMqttClient/Packets/Out/Subscribe.cpp new file mode 100644 index 0000000..85c10db --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Subscribe.cpp @@ -0,0 +1,49 @@ +#include "Subscribe.hpp" + +using AsyncMqttClientInternals::SubscribeOutPacket; + +SubscribeOutPacket::SubscribeOutPacket(const char* topic, uint8_t qos) { + 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; + + _data.reserve(neededSpace); + + _packetId = _getNextPacketId(); + char packetIdBytes[2]; + packetIdBytes[0] = _packetId >> 8; + packetIdBytes[1] = _packetId & 0xFF; + + _data.insert(_data.end(), fixedHeader, fixedHeader + 1 + remainingLengthLength); + _data.insert(_data.end(), packetIdBytes, packetIdBytes + 2); + _data.insert(_data.end(), topicLengthBytes, topicLengthBytes + 2); + _data.insert(_data.end(), topic, topic + topicLength); + _data.push_back(qosByte[0]); + _released = false; +} + +const uint8_t* SubscribeOutPacket::data(size_t index) const { + return &_data.data()[index]; +} + +size_t SubscribeOutPacket::size() const { + return _data.size(); +} diff --git a/src/AsyncMqttClient/Packets/Out/Subscribe.hpp b/src/AsyncMqttClient/Packets/Out/Subscribe.hpp new file mode 100644 index 0000000..1f85f59 --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Subscribe.hpp @@ -0,0 +1,21 @@ +#pragma once + +#include // strlen +#include + +#include "OutPacket.hpp" +#include "../../Flags.hpp" +#include "../../Helpers.hpp" +#include "../../Storage.hpp" + +namespace AsyncMqttClientInternals { +class SubscribeOutPacket : public OutPacket { + public: + SubscribeOutPacket(const char* topic, uint8_t qos); + const uint8_t* data(size_t index = 0) const; + size_t size() const; + + private: + std::vector _data; +}; +} // namespace AsyncMqttClientInternals diff --git a/src/AsyncMqttClient/Packets/Out/Unsubscribe.cpp b/src/AsyncMqttClient/Packets/Out/Unsubscribe.cpp new file mode 100644 index 0000000..4d859c9 --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Unsubscribe.cpp @@ -0,0 +1,42 @@ +#include "Unsubscribe.hpp" + +using AsyncMqttClientInternals::UnsubscribeOutPacket; + +UnsubscribeOutPacket::UnsubscribeOutPacket(const char* topic) { + 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; + + _packetId = _getNextPacketId(); + char packetIdBytes[2]; + packetIdBytes[0] = _packetId >> 8; + packetIdBytes[1] = _packetId & 0xFF; + + _data.insert(_data.end(), fixedHeader, fixedHeader + 1 + remainingLengthLength); + _data.insert(_data.end(), packetIdBytes, packetIdBytes + 2); + _data.insert(_data.end(), topicLengthBytes, topicLengthBytes + 2); + _data.insert(_data.end(), topic, topic + topicLength); + _released = false; +} + +const uint8_t* UnsubscribeOutPacket::data(size_t index) const { + return &_data.data()[index]; +} + +size_t UnsubscribeOutPacket::size() const { + return _data.size(); +} diff --git a/src/AsyncMqttClient/Packets/Out/Unsubscribe.hpp b/src/AsyncMqttClient/Packets/Out/Unsubscribe.hpp new file mode 100644 index 0000000..621802f --- /dev/null +++ b/src/AsyncMqttClient/Packets/Out/Unsubscribe.hpp @@ -0,0 +1,21 @@ +#pragma once + +#include // strlen +#include + +#include "OutPacket.hpp" +#include "../../Flags.hpp" +#include "../../Helpers.hpp" +#include "../../Storage.hpp" + +namespace AsyncMqttClientInternals { +class UnsubscribeOutPacket : public OutPacket { + public: + explicit UnsubscribeOutPacket(const char* topic); + const uint8_t* data(size_t index = 0) const; + size_t size() const; + + private: + std::vector _data; +}; +} // namespace AsyncMqttClientInternals