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

update to 1.7.0

Browse Source
This commit is contained in:
proddy
2024-06-04 21:19:48 +02:00
parent 26ac0057a5
commit 7f1dbbcb94
23 changed files with 2025 additions and 1428 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
lib/espMqttClient/README.md
View File

@@ -3,50 +3,16 @@
MQTT client library for the Espressif devices ESP8266 and ESP32 on the Arduino framework. MQTT client library for the Espressif devices ESP8266 and ESP32 on the Arduino framework.
Aims to be a non-blocking, fully compliant MQTT 3.1.1 client. Aims to be a non-blocking, fully compliant MQTT 3.1.1 client.
![platformio](https://github.com/bertmelis/espMqttClient/actions/workflows/build_platformio.yml/badge.svg) Copy of <https://github.com/bertmelis/espMqttClient>
![cpplint](https://github.com/bertmelis/espMqttClient/actions/workflows/cpplint.yml/badge.svg)
![cppcheck](https://github.com/bertmelis/espMqttClient/actions/workflows/cppcheck.yml/badge.svg)
[![PlatformIO Registry](https://badges.registry.platformio.org/packages/bertmelis/library/espMqttClient.svg)](https://registry.platformio.org/libraries/bertmelis/espMqttClient)
# Features Based on Version 1.7.0 - <https://github.com/bertmelis/espMqttClient/tree/v1.7.0>
- MQTT 3.1.1 compliant library with additional changes to support EMS-ESP such as compiling with Tasmota and not using `SecureWifiClient` in these two files:
- Sending and receiving at all QoS levels
- TCP and TCP/TLS using standard WiFiClient and WiFiClientSecure connections
- Virtually unlimited incoming and outgoing payload sizes
- Readable and understandable code
- Fully async clients available via [AsyncTCP](https://github.com/me-no-dev/AsyncTCP) or [ESPAsnycTCP](https://github.com/me-no-dev/ESPAsyncTCP) (no TLS supported)
- Supported platforms:
- Espressif ESP8266 and ESP32 using the Arduino framework
- Basic Linux compatibility*. This includes WSL on Windows
> Linux compatibility is mainly for automatic testing. It relies on a quick and dirty Arduino-style `Client` with a POSIX TCP client underneath and Arduino-style `IPAddress` class. These are lacking many features needed for proper Linux support. ```
src/espMqttClient.cpp
# Documentation src/Transport/ClientSecureSync.h
```
See [documentation](https://www.emelis.net/espMqttClient/) and the [examples](examples/).
## Limitations
### MQTT 3.1.1 Compliancy
Outgoing messages and session data are not stored in non-volatile memory. Any events like loss of power or sudden resets result in loss of data. Despite this limitation, one could still consider this library as fully complaint based on the non normative remark in point 4.1.1 of the specification.
### Non-blocking
This library aims to be fully non-blocking. It is however limited by the underlying `WiFiClient` library which is part of the Arduino framework and has a blocking `connect` method. This is not an issue on ESP32 because the call is offloaded to a separate task. On ESP8266 however, connecting will block until succesful or until the connection timeouts.
If you need a fully asynchronous MQTT client, you can use `espMqttClientAsync` which uses AsyncTCP/ESPAsyncTCP under the hood. These underlying libraries do not support TLS (anymore). I will not provide support TLS for the async client.
# Bugs and feature requests
Please use Github's facilities to get in touch.
# About this library
This client wouldn't exist without [Async-mqtt-client](https://github.com/marvinroger/async-mqtt-client). It has been my go-to MQTT client for many years. It was fast, reliable and had features that were non-existing in alternative libraries. However, the underlying async TCP libraries are lacking updates, especially updates related to secure connections. Adapting this library to use up-to-date TCP clients would not be trivial. I eventually decided to write my own MQTT library, from scratch.
The result is an almost non-blocking library with no external dependencies. The library is almost a drop-in replacement for the async-mqtt-client except a few parameter type changes (eg. `uint8_t*` instead of `char*` for payloads).
# License # License

19
lib/espMqttClient/src/Config.h
View File

@@ -8,12 +8,8 @@ the LICENSE file.
#pragma once #pragma once
#ifndef TASMOTA_SDK
#define EMC_CLIENT_SECURE
#endif
#ifndef EMC_TX_TIMEOUT #ifndef EMC_TX_TIMEOUT
#define EMC_TX_TIMEOUT 2000 #define EMC_TX_TIMEOUT 10000
#endif #endif
#ifndef EMC_RX_BUFFER_SIZE #ifndef EMC_RX_BUFFER_SIZE
@@ -64,3 +60,16 @@ the LICENSE file.
#ifndef EMC_USE_WATCHDOG #ifndef EMC_USE_WATCHDOG
#define EMC_USE_WATCHDOG 0 #define EMC_USE_WATCHDOG 0
#endif #endif
#ifndef EMC_USE_MEMPOOL
#define EMC_USE_MEMPOOL 0
#endif
#if EMC_USE_MEMPOOL
#ifndef EMC_NUM_POOL_ELEMENTS
#define EMC_NUM_POOL_ELEMENTS 32
#endif
#ifndef EMC_SIZE_POOL_ELEMENTS
#define EMC_SIZE_POOL_ELEMENTS 128
#endif
#endif

21
lib/espMqttClient/src/MemoryPool/LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2024 Bert Melis
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

105
lib/espMqttClient/src/MemoryPool/README.md Normal file
View File

@@ -0,0 +1,105 @@
# Memory Pool
EARLY VERSION. USE AT OWN RISK.
### Description
This is a simple memory pool that doesn't solve the fragmentation problem but contains it. Inside the pool you will still suffer memory fragmentation. The upside is that you're not restricted on memory size. As long as it fits in the pool, you can request any size!
For applications where the (maximum) size to allocate is known, a simple fixed block size memory pool is available. There is no memory fragmentation happening in this case. The downside is wastage of memory if you need less then the specified blocksize.
#### Features
- pool memory is statically allocated
- pool size adjusts on architecture
- no size calculation required: input number of blocks and size of block
- header-only library
- Variable size pool: no restriction on allocated size
- Variable size pool: malloc and free are O(n); The number of allocated blocks affects lookup.
- Fixed size pool: malloc and free are O(1).
[![Test with Platformio](https://github.com/bertmelis/MemoryPool/actions/workflows/test-platformio.yml/badge.svg)](https://github.com/bertmelis/MemoryPool/actions/workflows/test-platformio.yml)
[![cpplint](https://github.com/bertmelis/MemoryPool/actions/workflows/cpplint.yml/badge.svg)](https://github.com/bertmelis/MemoryPool/actions/workflows/cpplint.yml)
<!---[![cppcheck](https://github.com/bertmelis/MemoryPool/actions/workflows/cppcheck.yml/badge.svg)](https://github.com/bertmelis/MemoryPool/actions/workflows/cppcheck.yml)--->
### Usage
#### Variable size pool
```cpp
#include <MemoryPool.h>
Struct MyStruct {
unsigned int id;
std::size_t size;
unsigned char data[256];
};
// pool will be able to hold 10 blocks the size of MyStruct
MemoryPool::Variable<10, sizeof(MyStruct)> pool;
// you can allocate the specified blocksize
// allocation is done in number of 'unsigned char'
MyStruct* s = reinterpret_cast<MyStruct*>(pool.malloc(sizeof(MyStruct)));
// you can allocate less than the specified blocksize
int* i = reinterpret_cast<int*>(pool.malloc(sizeof(int)));
// you can allocate more than the specified blocksize
unsigned char* m = reinterpret_cast<unsigned char*>(pool.malloc(400));
pool.free(s);
pool.free(i);
pool.free(m);
```
#### Fixed size pool
```cpp
#include <MemoryPool.h>
Struct MyStruct {
unsigned int id;
std::size_t size;
unsigned char data[256];
};
// pool will be able to hold 10 blocks the size of MyStruct
MemoryPool::Fixed<10, sizeof(MyStruct)> pool;
// there is no size argument in the malloc function!
MyStruct* s = reinterpret_cast<MyStruct*>(pool.malloc());
// you can allocate less than the specified blocksize
int* i = reinterpret_cast<int*>(pool.malloc());
pool.free(s);
pool.free(i);
```
#### How it works
##### Variable size pool
Free blocks are organized as a linked list with their header (contains pointer to next and size). An allocated block also has this header with it's pointer set to `nullptr`. Therefore, each allocation wastes memory the size of the header (`sizeof(void*) + sizeof(std::size_t)`). On creation, the pool calculations the needed space to store the number of blocks wich each their header.
However, memory allocation isn't restricted the the specified blocksize. So in reality, you can allocate more if you allocate larger chunks because less memory blocks means less headers. After all, memory needs to be contiguous.
If you inspect the pool you'll see that a free pool only has one big block.
Allocation is linear: the pool is iterated until a suitable spot is found.
Freeing is also linear as the pool is traversed to insert the chunk in the linked list of free blocks
When freeing, free blocks which are adjacent are combined into one.
##### Fixed size pool
The fixed size pool is implemented as an array. Free blocks are saved as a linked list in this array.
### Bugs and feature requests
Please use Github's facilities to get in touch.
### License
This library is released under the MIT Licence. A copy is included in the repo.

16
lib/espMqttClient/src/MemoryPool/keywords.txt Normal file
View File

@@ -0,0 +1,16 @@
# Datatypes (KEYWORD1)
Fixed KEYWORD1
Variable KEYWORD1
# Methods and Functions (KEYWORD2)
malloc KEYWORD2
free KEYWORD2
freeMemory KEYWORD2
maxBlockSize KEYWORD2
print KEYWORD2
# Structures (KEYWORD3)
# structure KEYWORD3
# Constants (LITERAL1)
MemoryPool LITERAL1

21
lib/espMqttClient/src/MemoryPool/library.json Normal file
View File

@@ -0,0 +1,21 @@
{
"name": "MemoryPool",
"keywords": "memory",
"description": "A simple memory pool for fixed and variable sizes",
"authors":
{
"name": "Bert Melis",
"url": "https://github.com/bertmelis"
},
"license": "MIT",
"homepage": "https://github.com/bertmelis/MemoryPool",
"repository":
{
"type": "git",
"url": "https://github.com/bertmelis/MemoryPool.git"
},
"version": "0.1.0",
"frameworks": "*",
"platforms": "*",
"headers": ["MemoryPool.h"]
}

10
lib/espMqttClient/src/MemoryPool/library.properties Normal file
View File

@@ -0,0 +1,10 @@
name=MemoryPool
version=0.1.0
author=Bert Melis
maintainer=Bert Melis
sentence=A simple memory pool for fixed and variable sizes
paragraph=
category=Other
url=https://github.com/bertmelis/MemoryPool
architectures=*
includes=MemoryPool.h

119
lib/espMqttClient/src/MemoryPool/src/Fixed.h Normal file
View File

@@ -0,0 +1,119 @@
/*
Copyright (c) 2024 Bert Melis. All rights reserved.
This work is licensed under the terms of the MIT license.
For a copy, see <https://opensource.org/licenses/MIT> or
the LICENSE file.
*/
#pragma once
#include <cstddef> // std::size_t
#include <cassert> // assert
#if _GLIBCXX_HAS_GTHREADS
#include <mutex> // NOLINT [build/c++11] std::mutex, std::lock_guard
#else
#warning "The memory pool is not thread safe"
#endif
#ifdef MEMPOL_DEBUG
#include <iostream>
#endif
namespace MemoryPool {
template <std::size_t nrBlocks, std::size_t blocksize>
class Fixed {
public:
Fixed() // cppcheck-suppress uninitMemberVar
: _buffer{0}
, _head(_buffer) {
unsigned char* b = _head;
std::size_t adjustedBlocksize = sizeof(std::size_t) > blocksize ? sizeof(std::size_t) : blocksize;
for (std::size_t i = 0; i < nrBlocks - 1; ++i) {
*reinterpret_cast<unsigned char**>(b) = b + adjustedBlocksize;
b += adjustedBlocksize;
}
*reinterpret_cast<unsigned char**>(b) = nullptr;
}
// no copy nor move
Fixed (const Fixed&) = delete;
Fixed& operator= (const Fixed&) = delete;
void* malloc() {
#if _GLIBCXX_HAS_GTHREADS
const std::lock_guard<std::mutex> lockGuard(_mutex);
#endif
if (_head) {
void* retVal = _head;
_head = *reinterpret_cast<unsigned char**>(_head);
return retVal;
}
return nullptr;
}
void free(void* ptr) {
if (!ptr) return;
#if _GLIBCXX_HAS_GTHREADS
const std::lock_guard<std::mutex> lockGuard(_mutex);
#endif
*reinterpret_cast<unsigned char**>(ptr) = _head;
_head = reinterpret_cast<unsigned char*>(ptr);
}
std::size_t freeMemory() {
#if _GLIBCXX_HAS_GTHREADS
const std::lock_guard<std::mutex> lockGuard(_mutex);
#endif
unsigned char* i = _head;
std::size_t retVal = 0;
while (i) {
retVal += blocksize;
i = reinterpret_cast<unsigned char**>(i)[0];
}
return retVal;
}
#ifdef MEMPOL_DEBUG
void print() {
std::size_t adjustedBlocksize = sizeof(std::size_t) > blocksize ? sizeof(std::size_t) : blocksize;
std::cout << "+--------------------" << std::endl;
std::cout << "|start:" << reinterpret_cast<void*>(_buffer) << std::endl;
std::cout << "|blocks:" << nrBlocks << std::endl;
std::cout << "|blocksize:" << adjustedBlocksize << std::endl;
std::cout << "|head: " << reinterpret_cast<void*>(_head) << std::endl;
unsigned char* currentBlock = _buffer;
for (std::size_t i = 0; i < nrBlocks; ++i) {
std::cout << "|" << i + 1 << ": " << reinterpret_cast<void*>(currentBlock) << std::endl;
if (_isFree(currentBlock)) {
std::cout << "| free" << std::endl;
std::cout << "| next: " << reinterpret_cast<void*>(*reinterpret_cast<unsigned char**>(currentBlock)) << std::endl;
} else {
std::cout << "| allocated" << std::endl;
}
currentBlock += adjustedBlocksize;
}
std::cout << "+--------------------" << std::endl;
}
bool _isFree(const unsigned char* ptr) {
unsigned char* b = _head;
while (b) {
if (b == ptr) return true;
b = *reinterpret_cast<unsigned char**>(b);
}
return false;
}
#endif
private:
unsigned char _buffer[nrBlocks * (sizeof(std::size_t) > blocksize ? sizeof(std::size_t) : blocksize)];
unsigned char* _head;
#if _GLIBCXX_HAS_GTHREADS
std::mutex _mutex;
#endif
};
} // end namespace MemoryPool

12
lib/espMqttClient/src/MemoryPool/src/MemoryPool.h Normal file
View File

@@ -0,0 +1,12 @@
/*
Copyright (c) 2024 Bert Melis. All rights reserved.
This work is licensed under the terms of the MIT license.
For a copy, see <https://opensource.org/licenses/MIT> or
the LICENSE file.
*/
#pragma once
#include "Variable.h"
#include "Fixed.h"

242
lib/espMqttClient/src/MemoryPool/src/Variable.h Normal file
View File

@@ -0,0 +1,242 @@
/*
Copyright (c) 2024 Bert Melis. All rights reserved.
This work is licensed under the terms of the MIT license.
For a copy, see <https://opensource.org/licenses/MIT> or
the LICENSE file.
*/
#pragma once
#include <cstddef> // std::size_t
#include <cassert> // assert
#if _GLIBCXX_HAS_GTHREADS
#include <mutex> // NOLINT [build/c++11] std::mutex, std::lock_guard
#else
#warning "The memory pool is not thread safe"
#endif
#ifdef MEMPOL_DEBUG
#include <iostream>
#endif
namespace MemoryPool {
template <std::size_t nrBlocks, std::size_t blocksize>
class Variable {
public:
Variable()
: _buffer{0}
, _head(nullptr)
#ifdef MEMPOL_DEBUG
, _bufferSize(0)
#endif
{
std::size_t _normBlocksize = blocksize / sizeof(BlockHeader) + ((blocksize % sizeof(BlockHeader)) ? 1 : 0);
size_t nrBlocksToAlloc = nrBlocks * (_normBlocksize + 1);
BlockHeader* h = reinterpret_cast<BlockHeader*>(_buffer);
h->next = nullptr;
h->size = nrBlocksToAlloc;
_head = h;
#ifdef MEMPOL_DEBUG
_bufferSize = nrBlocksToAlloc;
#endif
}
// no copy nor move
Variable (const Variable&) = delete;
Variable& operator= (const Variable&) = delete;
void* malloc(size_t size) {
#if _GLIBCXX_HAS_GTHREADS
const std::lock_guard<std::mutex> lockGuard(_mutex);
#endif
if (size == 0) return nullptr;
size = (size / sizeof(BlockHeader) + (size % sizeof(BlockHeader) != 0)) + 1; // count by BlockHeader size, add 1 for header
#ifdef MEMPOL_DEBUG
std::cout << "malloc (raw) " << size << std::endl;
std::cout << "malloc (adj) " << size << " - ";
#endif
BlockHeader* currentBlock = _head;
BlockHeader* previousBlock = nullptr;
void* retVal = nullptr;
// iterate through linked free blocks
while (currentBlock) {
// consume whole block is size equals required size
if (currentBlock->size == size) {
if (previousBlock) previousBlock->next = currentBlock->next;
break;
// split block if size is larger and add second part to list of free blocks
} else if (currentBlock->size > size) {
BlockHeader* newBlock = currentBlock + size;
if (previousBlock) previousBlock->next = newBlock;
newBlock->next = currentBlock->next;
newBlock->size = currentBlock->size - size;
currentBlock->next = newBlock;
break;
}
previousBlock = currentBlock;
currentBlock = currentBlock->next;
}
if (currentBlock) {
if (currentBlock == _head) {
_head = currentBlock->next;
}
currentBlock->size = size;
currentBlock->next = nullptr; // used when freeing memory
retVal = currentBlock + 1;
#ifdef MEMPOL_DEBUG
std::cout << "ok" << std::endl;
#endif
} else {
#ifdef MEMPOL_DEBUG
std::cout << "nok" << std::endl;
#endif
(void)0;
}
return retVal;
}
void free(void* ptr) {
if (!ptr) return;
// check if ptr points to region in _buffer
#ifdef MEMPOL_DEBUG
std::cout << "free " << static_cast<void*>(reinterpret_cast<BlockHeader*>(ptr) - 1) << std::endl;
#endif
#if _GLIBCXX_HAS_GTHREADS
const std::lock_guard<std::mutex> lockGuard(_mutex);
#endif
BlockHeader* toFree = reinterpret_cast<BlockHeader*>(ptr) - 1;
BlockHeader* previous = reinterpret_cast<BlockHeader*>(_buffer);
BlockHeader* next = _head;
// toFree is the only free block
if (!next) {
_head = toFree;
return;
}
while (previous) {
if (!next || toFree < next) {
// 1. add block to linked list of free blocks
if (toFree < _head) {
toFree->next = _head;
_head = toFree;
} else {
previous->next = toFree;
toFree->next = next;
}
// 2. merge with previous if adjacent
if (toFree > _head && toFree == previous + previous->size) {
previous->size += toFree->size;
previous->next = toFree->next;
toFree = previous; // used in next check
}
// 3. merge with next if adjacent
if (toFree + toFree->size == next) {
toFree->size += next->size;
toFree->next = next->next;
}
// 4. done
return;
}
previous = next;
next = next->next;
}
}
std::size_t freeMemory() {
#if _GLIBCXX_HAS_GTHREADS
const std::lock_guard<std::mutex> lockGuard(_mutex);
#endif
size_t retVal = 0;
BlockHeader* currentBlock = reinterpret_cast<BlockHeader*>(_head);
while (currentBlock) {
retVal += currentBlock->size - 1;
currentBlock = currentBlock->next;
}
return retVal * sizeof(BlockHeader);
}
std::size_t maxBlockSize() {
#if _GLIBCXX_HAS_GTHREADS
const std::lock_guard<std::mutex> lockGuard(_mutex);
#endif
size_t retVal = 0;
BlockHeader* currentBlock = reinterpret_cast<BlockHeader*>(_head);
while (currentBlock) {
retVal = (currentBlock->size - 1 > retVal) ? currentBlock->size - 1 : retVal;
currentBlock = currentBlock->next;
}
return retVal * sizeof(BlockHeader);
}
#ifdef MEMPOL_DEBUG
void print() {
std::cout << "+--------------------" << std::endl;
std::cout << "|start:" << static_cast<void*>(_buffer) << std::endl;
std::cout << "|size:" << _bufferSize << std::endl;
std::cout << "|headersize:" << sizeof(BlockHeader) << std::endl;
std::cout << "|head: " << static_cast<void*>(_head) << std::endl;
BlockHeader* nextFreeBlock = _head;
BlockHeader* currentBlock = reinterpret_cast<BlockHeader*>(_buffer);
size_t blockNumber = 1;
while (currentBlock < reinterpret_cast<BlockHeader*>(_buffer) + _bufferSize) {
std::cout << "|" << blockNumber << ": " << static_cast<void*>(currentBlock) << std::endl;
std::cout << "| " << static_cast<void*>(currentBlock->next) << std::endl;
std::cout << "| " << currentBlock->size << std::endl;
if (currentBlock == nextFreeBlock) {
std::cout << "| free" << std::endl;
nextFreeBlock = nextFreeBlock->next;
} else {
std::cout << "| allocated" << std::endl;
}
++blockNumber;
currentBlock += currentBlock->size;
}
std::cout << "+--------------------" << std::endl;
}
#endif
private:
struct BlockHeader {
BlockHeader* next;
std::size_t size;
};
/*
pool size is aligned to sizeof(BlockHeader).
requested blocksize is therefore multiple of blockheader (rounded up)
total size = nr requested blocks * multiplier * blockheadersize
see constructor for calculation
*/
unsigned char _buffer[(nrBlocks * ((blocksize / sizeof(BlockHeader) + ((blocksize % sizeof(BlockHeader)) ? 1 : 0)) + 1)) * sizeof(BlockHeader)];
BlockHeader* _head;
#if _GLIBCXX_HAS_GTHREADS
std::mutex _mutex;
#endif
#ifdef MEMPOL_DEBUG
std::size_t _bufferSize;
#endif
};
} // end namespace MemoryPool

322
lib/espMqttClient/src/MqttClient.cpp
View File

@@ -14,49 +14,49 @@ using espMqttClientTypes::DisconnectReason;
using espMqttClientTypes::Error; using espMqttClientTypes::Error;
MqttClient::MqttClient(espMqttClientTypes::UseInternalTask useInternalTask, uint8_t priority, uint8_t core) MqttClient::MqttClient(espMqttClientTypes::UseInternalTask useInternalTask, uint8_t priority, uint8_t core)
: _useInternalTask(useInternalTask) : _useInternalTask(useInternalTask)
, _transport(nullptr) , _transport(nullptr)
, _onConnectCallback(nullptr) , _onConnectCallback(nullptr)
, _onDisconnectCallback(nullptr) , _onDisconnectCallback(nullptr)
, _onSubscribeCallback(nullptr) , _onSubscribeCallback(nullptr)
, _onUnsubscribeCallback(nullptr) , _onUnsubscribeCallback(nullptr)
, _onMessageCallback(nullptr) , _onMessageCallback(nullptr)
, _onPublishCallback(nullptr) , _onPublishCallback(nullptr)
, _onErrorCallback(nullptr) , _onErrorCallback(nullptr)
, _clientId(nullptr) , _clientId(nullptr)
, _ip() , _ip()
, _host(nullptr) , _host(nullptr)
, _port(1883) , _port(1883)
, _useIp(false) , _useIp(false)
, _keepAlive(15000) , _keepAlive(15000)
, _cleanSession(true) , _cleanSession(true)
, _username(nullptr) , _username(nullptr)
, _password(nullptr) , _password(nullptr)
, _willTopic(nullptr) , _willTopic(nullptr)
, _willPayload(nullptr) , _willPayload(nullptr)
, _willPayloadLength(0) , _willPayloadLength(0)
, _willQos(0) , _willQos(0)
, _willRetain(false) , _willRetain(false)
, _timeout(EMC_TX_TIMEOUT) , _timeout(EMC_TX_TIMEOUT)
, _state(State::disconnected) , _state(State::disconnected)
, _generatedClientId{0} , _generatedClientId{0}
, _packetId(0) , _packetId(0)
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
, _xSemaphore(nullptr) , _xSemaphore(nullptr)
, _taskHandle(nullptr) , _taskHandle(nullptr)
#endif #endif
, _rxBuffer{0} , _rxBuffer{0}
, _outbox() , _outbox()
, _bytesSent(0) , _bytesSent(0)
, _parser() , _parser()
, _lastClientActivity(0) , _lastClientActivity(0)
, _lastServerActivity(0) , _lastServerActivity(0)
, _pingSent(false) , _pingSent(false)
, _disconnectReason(DisconnectReason::TCP_DISCONNECTED) , _disconnectReason(DisconnectReason::TCP_DISCONNECTED)
#if defined(ARDUINO_ARCH_ESP32) && ARDUHAL_LOG_LEVEL >= ARDUHAL_LOG_LEVEL_INFO #if defined(ARDUINO_ARCH_ESP32) && ARDUHAL_LOG_LEVEL >= ARDUHAL_LOG_LEVEL_INFO
, _highWaterMark(4294967295) , _highWaterMark(4294967295)
#endif #endif
{ {
EMC_GENERATE_CLIENTID(_generatedClientId); EMC_GENERATE_CLIENTID(_generatedClientId);
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
_xSemaphore = xSemaphoreCreateMutex(); _xSemaphore = xSemaphoreCreateMutex();
@@ -65,9 +65,9 @@ MqttClient::MqttClient(espMqttClientTypes::UseInternalTask useInternalTask, uint
xTaskCreatePinnedToCore((TaskFunction_t)_loop, "mqttclient", EMC_TASK_STACK_SIZE, this, priority, &_taskHandle, core); xTaskCreatePinnedToCore((TaskFunction_t)_loop, "mqttclient", EMC_TASK_STACK_SIZE, this, priority, &_taskHandle, core);
} }
#else #else
(void)useInternalTask; (void) useInternalTask;
(void)priority; (void) priority;
(void)core; (void) core;
#endif #endif
_clientId = _generatedClientId; _clientId = _generatedClientId;
} }
@@ -78,23 +78,21 @@ MqttClient::~MqttClient() {
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
vSemaphoreDelete(_xSemaphore); vSemaphoreDelete(_xSemaphore);
if (_useInternalTask == espMqttClientTypes::UseInternalTask::YES) { if (_useInternalTask == espMqttClientTypes::UseInternalTask::YES) {
#if EMC_USE_WATCHDOG #if EMC_USE_WATCHDOG
esp_task_wdt_delete(_taskHandle); // not sure if this is really needed esp_task_wdt_delete(_taskHandle); // not sure if this is really needed
#endif #endif
vTaskDelete(_taskHandle); vTaskDelete(_taskHandle);
} }
#endif #endif
} }
bool MqttClient::connected() const { bool MqttClient::connected() const {
if (_state == State::connected) if (_state == State::connected) return true;
return true;
return false; return false;
} }
bool MqttClient::disconnected() const { bool MqttClient::disconnected() const {
if (_state == State::disconnected) if (_state == State::disconnected) return true;
return true;
return false; return false;
} }
@@ -113,16 +111,17 @@ bool MqttClient::connect() {
(uint16_t)(_keepAlive / 1000), // 32b to 16b doesn't overflow because it comes from 16b orignally (uint16_t)(_keepAlive / 1000), // 32b to 16b doesn't overflow because it comes from 16b orignally
_clientId)) { _clientId)) {
result = true; result = true;
_state = State::connectingTcp1; _setState(State::connectingTcp1);
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
if (_useInternalTask == espMqttClientTypes::UseInternalTask::YES) { if (_useInternalTask == espMqttClientTypes::UseInternalTask::YES) {
vTaskResume(_taskHandle); vTaskResume(_taskHandle);
} }
#endif #endif
} else { } else {
EMC_SEMAPHORE_GIVE();
emc_log_e("Could not create CONNECT packet"); emc_log_e("Could not create CONNECT packet");
EMC_SEMAPHORE_GIVE();
_onError(0, Error::OUT_OF_MEMORY); _onError(0, Error::OUT_OF_MEMORY);
EMC_SEMAPHORE_TAKE();
} }
EMC_SEMAPHORE_GIVE(); EMC_SEMAPHORE_GIVE();
} }
@@ -131,53 +130,57 @@ bool MqttClient::connect() {
bool MqttClient::disconnect(bool force) { bool MqttClient::disconnect(bool force) {
if (force && _state != State::disconnected && _state != State::disconnectingTcp1 && _state != State::disconnectingTcp2) { if (force && _state != State::disconnected && _state != State::disconnectingTcp1 && _state != State::disconnectingTcp2) {
_state = State::disconnectingTcp1; _setState(State::disconnectingTcp1);
return true; return true;
} }
if (!force && _state == State::connected) { if (!force && _state == State::connected) {
_state = State::disconnectingMqtt1; _setState(State::disconnectingMqtt1);
return true; return true;
} }
return false; return false;
} }
uint16_t MqttClient::publish(const char * topic, uint8_t qos, bool retain, const uint8_t * payload, size_t length) { uint16_t MqttClient::publish(const char* topic, uint8_t qos, bool retain, const uint8_t* payload, size_t length) {
#if !EMC_ALLOW_NOT_CONNECTED_PUBLISH #if !EMC_ALLOW_NOT_CONNECTED_PUBLISH
if (_state != State::connected) { if (_state != State::connected) {
#else #else
if (_state > State::connected) { if (_state > State::connected) {
#endif #endif
return 0; return 0;
} }
EMC_SEMAPHORE_TAKE(); EMC_SEMAPHORE_TAKE();
uint16_t packetId = (qos > 0) ? _getNextPacketId() : 1; uint16_t packetId = (qos > 0) ? _getNextPacketId() : 1;
if (!_addPacket(packetId, topic, payload, length, qos, retain)) { if (!_addPacket(packetId, topic, payload, length, qos, retain)) {
emc_log_e("Could not create PUBLISH packet"); emc_log_e("Could not create PUBLISH packet");
EMC_SEMAPHORE_GIVE();
_onError(packetId, Error::OUT_OF_MEMORY); _onError(packetId, Error::OUT_OF_MEMORY);
EMC_SEMAPHORE_TAKE();
packetId = 0; packetId = 0;
} }
EMC_SEMAPHORE_GIVE(); EMC_SEMAPHORE_GIVE();
return packetId; return packetId;
} }
uint16_t MqttClient::publish(const char * topic, uint8_t qos, bool retain, const char * payload) { uint16_t MqttClient::publish(const char* topic, uint8_t qos, bool retain, const char* payload) {
size_t len = strlen(payload); size_t len = strlen(payload);
return publish(topic, qos, retain, reinterpret_cast<const uint8_t *>(payload), len); return publish(topic, qos, retain, reinterpret_cast<const uint8_t*>(payload), len);
} }
uint16_t MqttClient::publish(const char * topic, uint8_t qos, bool retain, espMqttClientTypes::PayloadCallback callback, size_t length) { uint16_t MqttClient::publish(const char* topic, uint8_t qos, bool retain, espMqttClientTypes::PayloadCallback callback, size_t length) {
#if !EMC_ALLOW_NOT_CONNECTED_PUBLISH #if !EMC_ALLOW_NOT_CONNECTED_PUBLISH
if (_state != State::connected) { if (_state != State::connected) {
#else #else
if (_state > State::connected) { if (_state > State::connected) {
#endif #endif
return 0; return 0;
} }
EMC_SEMAPHORE_TAKE(); EMC_SEMAPHORE_TAKE();
uint16_t packetId = (qos > 0) ? _getNextPacketId() : 1; uint16_t packetId = (qos > 0) ? _getNextPacketId() : 1;
if (!_addPacket(packetId, topic, callback, length, qos, retain)) { if (!_addPacket(packetId, topic, callback, length, qos, retain)) {
emc_log_e("Could not create PUBLISH packet"); emc_log_e("Could not create PUBLISH packet");
EMC_SEMAPHORE_GIVE();
_onError(packetId, Error::OUT_OF_MEMORY); _onError(packetId, Error::OUT_OF_MEMORY);
EMC_SEMAPHORE_TAKE();
packetId = 0; packetId = 0;
} }
EMC_SEMAPHORE_GIVE(); EMC_SEMAPHORE_GIVE();
@@ -185,10 +188,12 @@ uint16_t MqttClient::publish(const char * topic, uint8_t qos, bool retain, espMq
} }
void MqttClient::clearQueue(bool deleteSessionData) { void MqttClient::clearQueue(bool deleteSessionData) {
EMC_SEMAPHORE_TAKE();
_clearQueue(deleteSessionData ? 2 : 0); _clearQueue(deleteSessionData ? 2 : 0);
EMC_SEMAPHORE_GIVE();
} }
const char * MqttClient::getClientId() const { const char* MqttClient::getClientId() const {
return _clientId; return _clientId;
} }
@@ -201,19 +206,19 @@ size_t MqttClient::queueSize() {
} }
void MqttClient::loop() { void MqttClient::loop() {
switch ((State)_state) { // modified by proddy for EMS-ESP compiling standalone switch (_state) {
case State::disconnected: case State::disconnected:
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
if (_useInternalTask == espMqttClientTypes::UseInternalTask::YES) { if (_useInternalTask == espMqttClientTypes::UseInternalTask::YES) {
vTaskSuspend(_taskHandle); vTaskSuspend(_taskHandle);
} }
#endif #endif
break; break;
case State::connectingTcp1: case State::connectingTcp1:
if (_useIp ? _transport->connect(_ip, _port) : _transport->connect(_host, _port)) { if (_useIp ? _transport->connect(_ip, _port) : _transport->connect(_host, _port)) {
_state = State::connectingTcp2; _setState(State::connectingTcp2);
} else { } else {
_state = State::disconnectingTcp1; _setState(State::disconnectingTcp1);
_disconnectReason = DisconnectReason::TCP_DISCONNECTED; _disconnectReason = DisconnectReason::TCP_DISCONNECTED;
break; break;
} }
@@ -223,37 +228,44 @@ void MqttClient::loop() {
if (_transport->connected()) { if (_transport->connected()) {
_parser.reset(); _parser.reset();
_lastClientActivity = _lastServerActivity = millis(); _lastClientActivity = _lastServerActivity = millis();
_state = State::connectingMqtt; _setState(State::connectingMqtt);
} } else if (_transport->disconnected()) { // sync: implemented as "not connected"; async: depending on state of pcb in underlying lib
break; _setState(State::disconnectingTcp1);
case State::connectingMqtt:
#if EMC_WAIT_FOR_CONNACK
if (_transport->connected()) {
_sendPacket();
_checkIncoming();
_checkPing();
} else {
_state = State::disconnectingTcp1;
_disconnectReason = DisconnectReason::TCP_DISCONNECTED; _disconnectReason = DisconnectReason::TCP_DISCONNECTED;
} }
break; break;
#else case State::connectingMqtt:
#if EMC_WAIT_FOR_CONNACK
if (_transport->connected()) {
EMC_SEMAPHORE_TAKE();
_sendPacket();
_checkIncoming();
_checkPing();
EMC_SEMAPHORE_GIVE();
} else {
_setState(State::disconnectingTcp1);
_disconnectReason = DisconnectReason::TCP_DISCONNECTED;
}
break;
#else
// receipt of CONNACK packet will set state to CONNECTED // receipt of CONNACK packet will set state to CONNECTED
// client however is allowed to send packets before CONNACK is received // client however is allowed to send packets before CONNACK is received
// so we fall through to 'connected' // so we fall through to 'connected'
[[fallthrough]]; [[fallthrough]];
#endif #endif
case State::connected: case State::connected:
[[fallthrough]]; [[fallthrough]];
case State::disconnectingMqtt2: case State::disconnectingMqtt2:
if (_transport->connected()) { if (_transport->connected()) {
// CONNECT packet is first in the queue // CONNECT packet is first in the queue
EMC_SEMAPHORE_TAKE();
_checkOutbox(); _checkOutbox();
_checkIncoming(); _checkIncoming();
_checkPing(); _checkPing();
_checkTimeout(); _checkTimeout();
EMC_SEMAPHORE_GIVE();
} else { } else {
_state = State::disconnectingTcp1; _setState(State::disconnectingTcp1);
_disconnectReason = DisconnectReason::TCP_DISCONNECTED; _disconnectReason = DisconnectReason::TCP_DISCONNECTED;
} }
break; break;
@@ -264,61 +276,69 @@ void MqttClient::loop() {
EMC_SEMAPHORE_GIVE(); EMC_SEMAPHORE_GIVE();
emc_log_e("Could not create DISCONNECT packet"); emc_log_e("Could not create DISCONNECT packet");
_onError(0, Error::OUT_OF_MEMORY); _onError(0, Error::OUT_OF_MEMORY);
EMC_SEMAPHORE_TAKE();
} else { } else {
_state = State::disconnectingMqtt2; _setState(State::disconnectingMqtt2);
} }
} }
EMC_SEMAPHORE_GIVE();
_checkOutbox(); _checkOutbox();
_checkIncoming(); _checkIncoming();
_checkPing(); _checkPing();
_checkTimeout(); _checkTimeout();
EMC_SEMAPHORE_GIVE();
break; break;
case State::disconnectingTcp1: case State::disconnectingTcp1:
_transport->stop(); _transport->stop();
_state = State::disconnectingTcp2; _setState(State::disconnectingTcp2);
break; // keep break to accomodate async clients break; // keep break to accomodate async clients
case State::disconnectingTcp2: case State::disconnectingTcp2:
if (_transport->disconnected()) { if (_transport->disconnected()) {
EMC_SEMAPHORE_TAKE();
_clearQueue(0); _clearQueue(0);
EMC_SEMAPHORE_GIVE();
_bytesSent = 0; _bytesSent = 0;
_state = State::disconnected; _setState(State::disconnected);
if (_onDisconnectCallback) if (_onDisconnectCallback) {
_onDisconnectCallback(_disconnectReason); _onDisconnectCallback(_disconnectReason);
} }
}
break; break;
// all cases covered, no default case // all cases covered, no default case
} }
EMC_YIELD(); EMC_YIELD();
#if defined(ARDUINO_ARCH_ESP32) && ARDUHAL_LOG_LEVEL >= ARDUHAL_LOG_LEVEL_INFO #if defined(ARDUINO_ARCH_ESP32) && ARDUHAL_LOG_LEVEL >= ARDUHAL_LOG_LEVEL_INFO
size_t waterMark = uxTaskGetStackHighWaterMark(NULL); size_t waterMark = uxTaskGetStackHighWaterMark(NULL);
if (waterMark < _highWaterMark) { if (waterMark < _highWaterMark) {
_highWaterMark = waterMark; _highWaterMark = waterMark;
emc_log_i("Stack usage: %zu/%i", EMC_TASK_STACK_SIZE - _highWaterMark, EMC_TASK_STACK_SIZE); emc_log_i("Stack usage: %zu/%i", EMC_TASK_STACK_SIZE - _highWaterMark, EMC_TASK_STACK_SIZE);
} }
#endif #endif
} }
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
void MqttClient::_loop(MqttClient * c) { void MqttClient::_loop(MqttClient* c) {
#if EMC_USE_WATCHDOG #if EMC_USE_WATCHDOG
if (esp_task_wdt_add(NULL) != ESP_OK) { if (esp_task_wdt_add(NULL) != ESP_OK) {
emc_log_e("Failed to add async task to WDT"); emc_log_e("Failed to add async task to WDT");
} }
#endif #endif
for (;;) { for (;;) {
c->loop(); c->loop();
#if EMC_USE_WATCHDOG #if EMC_USE_WATCHDOG
esp_task_wdt_reset(); esp_task_wdt_reset();
#endif #endif
} }
} }
#endif #endif
inline void MqttClient::_setState(State newState) {
emc_log_i("state %i --> %i", static_cast<std::underlying_type<State>::type>(_state.load()), static_cast<std::underlying_type<State>::type>(newState));
_state = newState;
}
uint16_t MqttClient::_getNextPacketId() { uint16_t MqttClient::_getNextPacketId() {
++_packetId; ++_packetId;
if (_packetId == 0) if (_packetId == 0) ++_packetId;
++_packetId;
return _packetId; return _packetId;
} }
@@ -331,14 +351,12 @@ void MqttClient::_checkOutbox() {
} }
int MqttClient::_sendPacket() { int MqttClient::_sendPacket() {
EMC_SEMAPHORE_TAKE(); OutgoingPacket* packet = _outbox.getCurrent();
OutgoingPacket * packet = _outbox.getCurrent();
size_t written = 0; size_t written = 0;
if (packet) { if (packet) {
size_t wantToWrite = packet->packet.available(_bytesSent); size_t wantToWrite = packet->packet.available(_bytesSent);
if (wantToWrite == 0) { if (wantToWrite == 0) {
EMC_SEMAPHORE_GIVE();
return 0; return 0;
} }
written = _transport->write(packet->packet.data(_bytesSent), wantToWrite); written = _transport->write(packet->packet.data(_bytesSent), wantToWrite);
@@ -347,30 +365,26 @@ int MqttClient::_sendPacket() {
_bytesSent += written; _bytesSent += written;
emc_log_i("tx %zu/%zu (%02x)", _bytesSent, packet->packet.size(), packet->packet.packetType()); emc_log_i("tx %zu/%zu (%02x)", _bytesSent, packet->packet.size(), packet->packet.packetType());
} }
EMC_SEMAPHORE_GIVE();
return written; return written;
} }
bool MqttClient::_advanceOutbox() { bool MqttClient::_advanceOutbox() {
EMC_SEMAPHORE_TAKE(); OutgoingPacket* packet = _outbox.getCurrent();
OutgoingPacket * packet = _outbox.getCurrent();
if (packet && _bytesSent == packet->packet.size()) { if (packet && _bytesSent == packet->packet.size()) {
if ((packet->packet.packetType()) == PacketType.DISCONNECT) { if ((packet->packet.packetType()) == PacketType.DISCONNECT) {
_state = State::disconnectingTcp1; _setState(State::disconnectingTcp1);
_disconnectReason = DisconnectReason::USER_OK; _disconnectReason = DisconnectReason::USER_OK;
} }
if (packet->packet.removable()) { if (packet->packet.removable()) {
_outbox.removeCurrent(); _outbox.removeCurrent();
} else { } else {
// we already set 'dup' here, in case we have to retry // we already set 'dup' here, in case we have to retry
if ((packet->packet.packetType()) == PacketType.PUBLISH) if ((packet->packet.packetType()) == PacketType.PUBLISH) packet->packet.setDup();
packet->packet.setDup();
_outbox.next(); _outbox.next();
} }
packet = _outbox.getCurrent(); packet = _outbox.getCurrent();
_bytesSent = 0; _bytesSent = 0;
} }
EMC_SEMAPHORE_GIVE();
return packet; return packet;
} }
@@ -387,10 +401,10 @@ void MqttClient::_checkIncoming() {
espMqttClientInternals::MQTTPacketType packetType = _parser.getPacket().fixedHeader.packetType & 0xF0; espMqttClientInternals::MQTTPacketType packetType = _parser.getPacket().fixedHeader.packetType & 0xF0;
if (_state == State::connectingMqtt && packetType != PacketType.CONNACK) { if (_state == State::connectingMqtt && packetType != PacketType.CONNACK) {
emc_log_w("Disconnecting, expected CONNACK - protocol error"); emc_log_w("Disconnecting, expected CONNACK - protocol error");
_state = State::disconnectingTcp1; _setState(State::disconnectingTcp1);
return; return;
} }
switch (packetType & 0xF0) { switch (packetType) {
case PacketType.CONNACK: case PacketType.CONNACK:
_onConnack(); _onConnack();
if (_state != State::connected) { if (_state != State::connected) {
@@ -398,8 +412,7 @@ void MqttClient::_checkIncoming() {
} }
break; break;
case PacketType.PUBLISH: case PacketType.PUBLISH:
if (_state >= State::disconnectingMqtt1) if (_state >= State::disconnectingMqtt1) break; // stop processing incoming once user has called disconnect
break; // stop processing incoming once user has called disconnect
_onPublish(); _onPublish();
break; break;
case PacketType.PUBACK: case PacketType.PUBACK:
@@ -426,7 +439,7 @@ void MqttClient::_checkIncoming() {
} }
} else if (result == espMqttClientInternals::ParserResult::protocolError) { } else if (result == espMqttClientInternals::ParserResult::protocolError) {
emc_log_w("Disconnecting, protocol error"); emc_log_w("Disconnecting, protocol error");
_state = State::disconnectingTcp1; _setState(State::disconnectingTcp1);
_disconnectReason = DisconnectReason::TCP_DISCONNECTED; _disconnectReason = DisconnectReason::TCP_DISCONNECTED;
return; return;
} }
@@ -439,35 +452,32 @@ void MqttClient::_checkIncoming() {
} }
void MqttClient::_checkPing() { void MqttClient::_checkPing() {
if (_keepAlive == 0) if (_keepAlive == 0) return; // keepalive is disabled
return; // keepalive is disabled
uint32_t currentMillis = millis(); uint32_t currentMillis = millis();
// disconnect when server was inactive for twice the keepalive time // disconnect when server was inactive for twice the keepalive time
if (currentMillis - _lastServerActivity > 2 * _keepAlive) { if (currentMillis - _lastServerActivity > 2 * _keepAlive) {
emc_log_w("Disconnecting, server exceeded keepalive"); emc_log_w("Disconnecting, server exceeded keepalive");
_state = State::disconnectingTcp1; _setState(State::disconnectingTcp1);
_disconnectReason = DisconnectReason::TCP_DISCONNECTED; _disconnectReason = DisconnectReason::TCP_DISCONNECTED;
return; return;
} }
// send ping when client was inactive during the keepalive time // send ping when client was inactive during the keepalive time
// or when server hasn't responded within keepalive time (typically due to QOS 0) // or when server hasn't responded within keepalive time (typically due to QOS 0)
if (!_pingSent && ((currentMillis - _lastClientActivity > _keepAlive) || (currentMillis - _lastServerActivity > _keepAlive))) { if (!_pingSent &&
EMC_SEMAPHORE_TAKE(); ((currentMillis - _lastClientActivity > _keepAlive) ||
(currentMillis - _lastServerActivity > _keepAlive))) {
if (!_addPacket(PacketType.PINGREQ)) { if (!_addPacket(PacketType.PINGREQ)) {
EMC_SEMAPHORE_GIVE();
emc_log_e("Could not create PING packet"); emc_log_e("Could not create PING packet");
return; return;
} }
EMC_SEMAPHORE_GIVE();
_pingSent = true; _pingSent = true;
} }
} }
void MqttClient::_checkTimeout() { void MqttClient::_checkTimeout() {
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
// check that we're not busy sending // check that we're not busy sending
// don't check when first item hasn't been sent yet // don't check when first item hasn't been sent yet
@@ -477,29 +487,30 @@ void MqttClient::_checkTimeout() {
_outbox.resetCurrent(); _outbox.resetCurrent();
} }
} }
EMC_SEMAPHORE_GIVE();
} }
void MqttClient::_onConnack() { void MqttClient::_onConnack() {
if (_parser.getPacket().variableHeader.fixed.connackVarHeader.returnCode == 0x00) { if (_parser.getPacket().variableHeader.fixed.connackVarHeader.returnCode == 0x00) {
_pingSent = false; // reset after keepalive timeout disconnect _pingSent = false; // reset after keepalive timeout disconnect
_state = State::connected; _setState(State::connected);
_advanceOutbox(); _advanceOutbox();
if (_parser.getPacket().variableHeader.fixed.connackVarHeader.sessionPresent == 0) { if (_parser.getPacket().variableHeader.fixed.connackVarHeader.sessionPresent == 0) {
_clearQueue(1); _clearQueue(1);
} }
if (_onConnectCallback) { if (_onConnectCallback) {
EMC_SEMAPHORE_GIVE();
_onConnectCallback(_parser.getPacket().variableHeader.fixed.connackVarHeader.sessionPresent); _onConnectCallback(_parser.getPacket().variableHeader.fixed.connackVarHeader.sessionPresent);
EMC_SEMAPHORE_TAKE();
} }
} else { } else {
_state = State::disconnectingTcp1; _setState(State::disconnectingTcp1);
// cast is safe because the parser already checked for a valid return code // cast is safe because the parser already checked for a valid return code
_disconnectReason = static_cast<DisconnectReason>(_parser.getPacket().variableHeader.fixed.connackVarHeader.returnCode); _disconnectReason = static_cast<DisconnectReason>(_parser.getPacket().variableHeader.fixed.connackVarHeader.returnCode);
} }
} }
void MqttClient::_onPublish() { void MqttClient::_onPublish() {
const espMqttClientInternals::IncomingPacket & p = _parser.getPacket(); const espMqttClientInternals::IncomingPacket& p = _parser.getPacket();
uint8_t qos = p.qos(); uint8_t qos = p.qos();
bool retain = p.retain(); bool retain = p.retain();
bool dup = p.dup(); bool dup = p.dup();
@@ -507,19 +518,15 @@ void MqttClient::_onPublish() {
bool callback = true; bool callback = true;
if (qos == 1) { if (qos == 1) {
if (p.payload.index + p.payload.length == p.payload.total) { if (p.payload.index + p.payload.length == p.payload.total) {
EMC_SEMAPHORE_TAKE();
if (!_addPacket(PacketType.PUBACK, packetId)) { if (!_addPacket(PacketType.PUBACK, packetId)) {
emc_log_e("Could not create PUBACK packet"); emc_log_e("Could not create PUBACK packet");
} }
EMC_SEMAPHORE_GIVE();
} }
} else if (qos == 2) { } else if (qos == 2) {
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
while (it) { while (it) {
if ((it.get()->packet.packetType()) == PacketType.PUBREC && it.get()->packet.packetId() == packetId) { if ((it.get()->packet.packetType()) == PacketType.PUBREC && it.get()->packet.packetId() == packetId) {
callback = false; callback = false;
_outbox.remove(it);
emc_log_e("QoS2 packet previously delivered"); emc_log_e("QoS2 packet previously delivered");
break; break;
} }
@@ -530,16 +537,22 @@ void MqttClient::_onPublish() {
emc_log_e("Could not create PUBREC packet"); emc_log_e("Could not create PUBREC packet");
} }
} }
EMC_SEMAPHORE_GIVE();
} }
if (callback && _onMessageCallback) if (callback && _onMessageCallback) {
_onMessageCallback({qos, dup, retain, packetId}, p.variableHeader.topic, p.payload.data, p.payload.length, p.payload.index, p.payload.total); EMC_SEMAPHORE_GIVE();
_onMessageCallback({qos, dup, retain, packetId},
p.variableHeader.topic,
p.payload.data,
p.payload.length,
p.payload.index,
p.payload.total);
EMC_SEMAPHORE_TAKE();
}
} }
void MqttClient::_onPuback() { void MqttClient::_onPuback() {
bool callback = false; bool callback = false;
uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId; uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId;
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
while (it) { while (it) {
// PUBACKs come in the order PUBs are sent. So we only check the first PUB packet in outbox // PUBACKs come in the order PUBs are sent. So we only check the first PUB packet in outbox
@@ -555,10 +568,12 @@ void MqttClient::_onPuback() {
} }
++it; ++it;
} }
EMC_SEMAPHORE_GIVE();
if (callback) { if (callback) {
if (_onPublishCallback) if (_onPublishCallback) {
EMC_SEMAPHORE_GIVE();
_onPublishCallback(idToMatch); _onPublishCallback(idToMatch);
EMC_SEMAPHORE_TAKE();
}
} else { } else {
emc_log_w("No matching PUBLISH packet found"); emc_log_w("No matching PUBLISH packet found");
} }
@@ -567,12 +582,11 @@ void MqttClient::_onPuback() {
void MqttClient::_onPubrec() { void MqttClient::_onPubrec() {
bool success = false; bool success = false;
uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId; uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId;
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
while (it) { while (it) {
// PUBRECs come in the order PUBs are sent. So we only check the first PUB packet in outbox // PUBRECs come in the order PUBs are sent. So we only check the first PUB packet in outbox
// if it doesn't match the ID, return // if it doesn't match the ID, return
if ((it.get()->packet.packetType()) == PacketType.PUBLISH || (it.get()->packet.packetType()) == PacketType.PUBREL) { if ((it.get()->packet.packetType()) == PacketType.PUBLISH) {
if (it.get()->packet.packetId() == idToMatch) { if (it.get()->packet.packetId() == idToMatch) {
if (!_addPacket(PacketType.PUBREL, idToMatch)) { if (!_addPacket(PacketType.PUBREL, idToMatch)) {
emc_log_e("Could not create PUBREL packet"); emc_log_e("Could not create PUBREL packet");
@@ -589,13 +603,11 @@ void MqttClient::_onPubrec() {
if (!success) { if (!success) {
emc_log_w("No matching PUBLISH packet found"); emc_log_w("No matching PUBLISH packet found");
} }
EMC_SEMAPHORE_GIVE();
} }
void MqttClient::_onPubrel() { void MqttClient::_onPubrel() {
bool success = false; bool success = false;
uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId; uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId;
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
while (it) { while (it) {
// PUBRELs come in the order PUBRECs are sent. So we only check the first PUBREC packet in outbox // PUBRELs come in the order PUBRECs are sent. So we only check the first PUBREC packet in outbox
@@ -617,12 +629,10 @@ void MqttClient::_onPubrel() {
if (!success) { if (!success) {
emc_log_w("No matching PUBREC packet found"); emc_log_w("No matching PUBREC packet found");
} }
EMC_SEMAPHORE_GIVE();
} }
void MqttClient::_onPubcomp() { void MqttClient::_onPubcomp() {
bool callback = false; bool callback = false;
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId; uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId;
while (it) { while (it) {
@@ -639,10 +649,12 @@ void MqttClient::_onPubcomp() {
} }
++it; ++it;
} }
EMC_SEMAPHORE_GIVE();
if (callback) { if (callback) {
if (_onPublishCallback) if (_onPublishCallback) {
EMC_SEMAPHORE_GIVE();
_onPublishCallback(idToMatch); _onPublishCallback(idToMatch);
EMC_SEMAPHORE_TAKE();
}
} else { } else {
emc_log_w("No matching PUBREL packet found"); emc_log_w("No matching PUBREL packet found");
} }
@@ -651,7 +663,6 @@ void MqttClient::_onPubcomp() {
void MqttClient::_onSuback() { void MqttClient::_onSuback() {
bool callback = false; bool callback = false;
uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId; uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId;
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
while (it) { while (it) {
if (((it.get()->packet.packetType()) == PacketType.SUBSCRIBE) && it.get()->packet.packetId() == idToMatch) { if (((it.get()->packet.packetType()) == PacketType.SUBSCRIBE) && it.get()->packet.packetId() == idToMatch) {
@@ -661,12 +672,12 @@ void MqttClient::_onSuback() {
} }
++it; ++it;
} }
EMC_SEMAPHORE_GIVE();
if (callback) { if (callback) {
if (_onSubscribeCallback) if (_onSubscribeCallback) {
_onSubscribeCallback(idToMatch, EMC_SEMAPHORE_GIVE();
reinterpret_cast<const espMqttClientTypes::SubscribeReturncode *>(_parser.getPacket().payload.data), _onSubscribeCallback(idToMatch, reinterpret_cast<const espMqttClientTypes::SubscribeReturncode*>(_parser.getPacket().payload.data), _parser.getPacket().payload.total);
_parser.getPacket().payload.total); EMC_SEMAPHORE_TAKE();
}
} else { } else {
emc_log_w("received SUBACK without SUB"); emc_log_w("received SUBACK without SUB");
} }
@@ -674,7 +685,6 @@ void MqttClient::_onSuback() {
void MqttClient::_onUnsuback() { void MqttClient::_onUnsuback() {
bool callback = false; bool callback = false;
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId; uint16_t idToMatch = _parser.getPacket().variableHeader.fixed.packetId;
while (it) { while (it) {
@@ -685,10 +695,12 @@ void MqttClient::_onUnsuback() {
} }
++it; ++it;
} }
EMC_SEMAPHORE_GIVE();
if (callback) { if (callback) {
if (_onUnsubscribeCallback) if (_onUnsubscribeCallback) {
EMC_SEMAPHORE_GIVE();
_onUnsubscribeCallback(idToMatch); _onUnsubscribeCallback(idToMatch);
EMC_SEMAPHORE_TAKE();
}
} else { } else {
emc_log_w("received UNSUBACK without UNSUB"); emc_log_w("received UNSUBACK without UNSUB");
} }
@@ -696,7 +708,6 @@ void MqttClient::_onUnsuback() {
void MqttClient::_clearQueue(int clearData) { void MqttClient::_clearQueue(int clearData) {
emc_log_i("clearing queue (clear session: %d)", clearData); emc_log_i("clearing queue (clear session: %d)", clearData);
EMC_SEMAPHORE_TAKE();
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front(); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.front();
if (clearData == 0) { if (clearData == 0) {
// keep PUB (qos > 0, aka packetID != 0), PUBREC and PUBREL // keep PUB (qos > 0, aka packetID != 0), PUBREC and PUBREL
@@ -704,7 +715,9 @@ void MqttClient::_clearQueue(int clearData) {
// and stores the packet id in the PUBREC packet. So we also must keep PUBREC. // and stores the packet id in the PUBREC packet. So we also must keep PUBREC.
while (it) { while (it) {
espMqttClientInternals::MQTTPacketType type = it.get()->packet.packetType(); espMqttClientInternals::MQTTPacketType type = it.get()->packet.packetType();
if (type == PacketType.PUBREC || type == PacketType.PUBREL || (type == PacketType.PUBLISH && it.get()->packet.packetId() != 0)) { if (type == PacketType.PUBREC ||
type == PacketType.PUBREL ||
(type == PacketType.PUBLISH && it.get()->packet.packetId() != 0)) {
++it; ++it;
} else { } else {
_outbox.remove(it); _outbox.remove(it);
@@ -724,7 +737,6 @@ void MqttClient::_clearQueue(int clearData) {
_outbox.remove(it); _outbox.remove(it);
} }
} }
EMC_SEMAPHORE_GIVE();
} }
void MqttClient::_onError(uint16_t packetId, espMqttClientTypes::Error error) { void MqttClient::_onError(uint16_t packetId, espMqttClientTypes::Error error) {

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

@@ -31,13 +31,14 @@ class MqttClient {
bool connect(); bool connect();
bool disconnect(bool force = false); bool disconnect(bool force = false);
template <typename... Args> template <typename... Args>
uint16_t subscribe(const char * topic, uint8_t qos, Args &&... args) { uint16_t subscribe(const char* topic, uint8_t qos, Args&&... args) {
uint16_t packetId = _getNextPacketId(); uint16_t packetId = 0;
if (_state != State::connected) { if (_state != State::connected) {
packetId = 0; return packetId;
} else { } else {
EMC_SEMAPHORE_TAKE(); EMC_SEMAPHORE_TAKE();
if (!_addPacket(packetId, topic, qos, std::forward<Args>(args)...)) { packetId = _getNextPacketId();
if (!_addPacket(packetId, topic, qos, std::forward<Args>(args) ...)) {
emc_log_e("Could not create SUBSCRIBE packet"); emc_log_e("Could not create SUBSCRIBE packet");
packetId = 0; packetId = 0;
} }
@@ -46,13 +47,14 @@ class MqttClient {
return packetId; return packetId;
} }
template <typename... Args> template <typename... Args>
uint16_t unsubscribe(const char * topic, Args &&... args) { uint16_t unsubscribe(const char* topic, Args&&... args) {
uint16_t packetId = _getNextPacketId(); uint16_t packetId = 0;
if (_state != State::connected) { if (_state != State::connected) {
packetId = 0; return packetId;
} else { } else {
EMC_SEMAPHORE_TAKE(); EMC_SEMAPHORE_TAKE();
if (!_addPacket(packetId, topic, std::forward<Args>(args)...)) { packetId = _getNextPacketId();
if (!_addPacket(packetId, topic, std::forward<Args>(args) ...)) {
emc_log_e("Could not create UNSUBSCRIBE packet"); emc_log_e("Could not create UNSUBSCRIBE packet");
packetId = 0; packetId = 0;
} }
@@ -60,18 +62,18 @@ class MqttClient {
} }
return packetId; return packetId;
} }
uint16_t publish(const char * topic, uint8_t qos, bool retain, const uint8_t * payload, size_t length); uint16_t publish(const char* topic, uint8_t qos, bool retain, const uint8_t* payload, size_t length);
uint16_t publish(const char * topic, uint8_t qos, bool retain, const char * payload); uint16_t publish(const char* topic, uint8_t qos, bool retain, const char* payload);
uint16_t publish(const char * topic, uint8_t qos, bool retain, espMqttClientTypes::PayloadCallback callback, size_t length); uint16_t publish(const char* topic, uint8_t qos, bool retain, espMqttClientTypes::PayloadCallback callback, size_t length);
void clearQueue(bool deleteSessionData = false); // Not MQTT compliant and may cause unpredictable results when `deleteSessionData` = true! void clearQueue(bool deleteSessionData = false); // Not MQTT compliant and may cause unpredictable results when `deleteSessionData` = true!
const char * getClientId() const; const char* getClientId() const;
size_t queueSize(); // No const because of mutex size_t queueSize(); // No const because of mutex
void loop(); void loop();
protected: protected:
explicit MqttClient(espMqttClientTypes::UseInternalTask useInternalTask, uint8_t priority = 1, uint8_t core = 1); explicit MqttClient(espMqttClientTypes::UseInternalTask useInternalTask, uint8_t priority = 1, uint8_t core = 1);
espMqttClientTypes::UseInternalTask _useInternalTask; espMqttClientTypes::UseInternalTask _useInternalTask;
espMqttClientInternals::Transport * _transport; espMqttClientInternals::Transport* _transport;
espMqttClientTypes::OnConnectCallback _onConnectCallback; espMqttClientTypes::OnConnectCallback _onConnectCallback;
espMqttClientTypes::OnDisconnectCallback _onDisconnectCallback; espMqttClientTypes::OnDisconnectCallback _onDisconnectCallback;
@@ -80,18 +82,18 @@ class MqttClient {
espMqttClientTypes::OnMessageCallback _onMessageCallback; espMqttClientTypes::OnMessageCallback _onMessageCallback;
espMqttClientTypes::OnPublishCallback _onPublishCallback; espMqttClientTypes::OnPublishCallback _onPublishCallback;
espMqttClientTypes::OnErrorCallback _onErrorCallback; espMqttClientTypes::OnErrorCallback _onErrorCallback;
typedef void (*mqttClientHook)(void *); typedef void(*mqttClientHook)(void*);
const char * _clientId; const char* _clientId;
IPAddress _ip; IPAddress _ip;
const char * _host; const char* _host;
uint16_t _port; uint16_t _port;
bool _useIp; bool _useIp;
uint32_t _keepAlive; uint32_t _keepAlive;
bool _cleanSession; bool _cleanSession;
const char * _username; const char* _username;
const char * _password; const char* _password;
const char * _willTopic; const char* _willTopic;
const uint8_t * _willPayload; const uint8_t* _willPayload;
uint16_t _willPayloadLength; uint16_t _willPayloadLength;
uint8_t _willQos; uint8_t _willQos;
bool _willRetain; bool _willRetain;
@@ -111,6 +113,7 @@ class MqttClient {
disconnectingTcp2 = 8 disconnectingTcp2 = 8
}; };
std::atomic<State> _state; std::atomic<State> _state;
inline void _setState(State newState);
private: private:
char _generatedClientId[EMC_CLIENTID_LENGTH]; char _generatedClientId[EMC_CLIENTID_LENGTH];
@@ -119,11 +122,11 @@ class MqttClient {
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
SemaphoreHandle_t _xSemaphore; SemaphoreHandle_t _xSemaphore;
TaskHandle_t _taskHandle; TaskHandle_t _taskHandle;
static void _loop(MqttClient * c); static void _loop(MqttClient* c);
#elif defined(ARDUINO_ARCH_ESP8266) && EMC_ESP8266_MULTITHREADING #elif defined(ARDUINO_ARCH_ESP8266) && EMC_ESP8266_MULTITHREADING
std::atomic<bool> _xSemaphore = false; std::atomic<bool> _xSemaphore = false;
#elif defined(__linux__) #elif defined(__linux__)
mutable std::mutex mtx; // modified by proddy for EMS-ESP compiling standalone std::mutex mtx;
#endif #endif
uint8_t _rxBuffer[EMC_RX_BUFFER_SIZE]; uint8_t _rxBuffer[EMC_RX_BUFFER_SIZE];
@@ -131,11 +134,9 @@ class MqttClient {
uint32_t timeSent; uint32_t timeSent;
espMqttClientInternals::Packet packet; espMqttClientInternals::Packet packet;
template <typename... Args> template <typename... Args>
OutgoingPacket(uint32_t t, espMqttClientTypes::Error & error, Args &&... args) OutgoingPacket(uint32_t t, espMqttClientTypes::Error& error, Args&&... args) : // NOLINT(runtime/references)
: // NOLINT(runtime/references) timeSent(t),
timeSent(t) packet(error, std::forward<Args>(args) ...) {}
, packet(error, std::forward<Args>(args)...) {
}
}; };
espMqttClientInternals::Outbox<OutgoingPacket> _outbox; espMqttClientInternals::Outbox<OutgoingPacket> _outbox;
size_t _bytesSent; size_t _bytesSent;
@@ -148,27 +149,25 @@ class MqttClient {
uint16_t _getNextPacketId(); uint16_t _getNextPacketId();
template <typename... Args> template <typename... Args>
bool _addPacket(Args &&... args) { bool _addPacket(Args&&... args) {
espMqttClientTypes::Error error(espMqttClientTypes::Error::SUCCESS); espMqttClientTypes::Error error(espMqttClientTypes::Error::SUCCESS);
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.emplace(0, error, std::forward<Args>(args)...); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.emplace(0, error, std::forward<Args>(args) ...);
if (it && error == espMqttClientTypes::Error::SUCCESS) { if (it && error == espMqttClientTypes::Error::SUCCESS) {
return true; return true;
} else { } else {
if (it) if (it) _outbox.remove(it);
_outbox.remove(it);
return false; return false;
} }
} }
template <typename... Args> template <typename... Args>
bool _addPacketFront(Args &&... args) { bool _addPacketFront(Args&&... args) {
espMqttClientTypes::Error error(espMqttClientTypes::Error::SUCCESS); espMqttClientTypes::Error error(espMqttClientTypes::Error::SUCCESS);
espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.emplaceFront(0, error, std::forward<Args>(args)...); espMqttClientInternals::Outbox<OutgoingPacket>::Iterator it = _outbox.emplaceFront(0, error, std::forward<Args>(args) ...);
if (it && error == espMqttClientTypes::Error::SUCCESS) { if (it && error == espMqttClientTypes::Error::SUCCESS) {
return true; return true;
} else { } else {
if (it) if (it) _outbox.remove(it);
_outbox.remove(it);
return false; return false;
} }
} }
@@ -194,9 +193,9 @@ class MqttClient {
// 2: delete all // 2: delete all
void _onError(uint16_t packetId, espMqttClientTypes::Error error); void _onError(uint16_t packetId, espMqttClientTypes::Error error);
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
#if ARDUHAL_LOG_LEVEL >= ARDUHAL_LOG_LEVEL_INFO #if ARDUHAL_LOG_LEVEL >= ARDUHAL_LOG_LEVEL_INFO
size_t _highWaterMark; size_t _highWaterMark;
#endif #endif
#endif #endif
}; };

46
lib/espMqttClient/src/Outbox.h
View File

@@ -9,7 +9,12 @@ the LICENSE file.
#pragma once #pragma once
#include <new> // new (std::nothrow) #if EMC_USE_MEMPOOL
#include "MemoryPool/src/MemoryPool.h"
#include "Config.h"
#else
#include <new> // new (std::nothrow)
#endif
#include <utility> // std::forward #include <utility> // std::forward
namespace espMqttClientInternals { namespace espMqttClientInternals {
@@ -28,11 +33,20 @@ class Outbox {
: _first(nullptr) : _first(nullptr)
, _last(nullptr) , _last(nullptr)
, _current(nullptr) , _current(nullptr)
, _prev(nullptr) {} , _prev(nullptr)
#if EMC_USE_MEMPOOL
, _memPool()
#endif
{}
~Outbox() { ~Outbox() {
while (_first) { while (_first) {
Node* n = _first->next; Node* n = _first->next;
#if EMC_USE_MEMPOOL
_first->~Node();
_memPool.free(_first);
#else
delete _first; delete _first;
#endif
_first = n; _first = n;
} }
} }
@@ -79,7 +93,15 @@ class Outbox {
template <class... Args> template <class... Args>
Iterator emplace(Args&&... args) { Iterator emplace(Args&&... args) {
Iterator it; Iterator it;
Node* node = new (std::nothrow) Node(std::forward<Args>(args) ...); #if EMC_USE_MEMPOOL
void* buf = _memPool.malloc();
Node* node = nullptr;
if (buf) {
node = new(buf) Node(std::forward<Args>(args) ...);
}
#else
Node* node = new(std::nothrow) Node(std::forward<Args>(args) ...);
#endif
if (node != nullptr) { if (node != nullptr) {
if (!_first) { if (!_first) {
// queue is empty // queue is empty
@@ -103,7 +125,15 @@ class Outbox {
template <class... Args> template <class... Args>
Iterator emplaceFront(Args&&... args) { Iterator emplaceFront(Args&&... args) {
Iterator it; Iterator it;
Node* node = new (std::nothrow) Node(std::forward<Args>(args) ...); #if EMC_USE_MEMPOOL
void* buf = _memPool.malloc();
Node* node = nullptr;
if (buf) {
node = new(buf) Node(std::forward<Args>(args) ...);
}
#else
Node* node = new(std::nothrow) Node(std::forward<Args>(args) ...);
#endif
if (node != nullptr) { if (node != nullptr) {
if (!_first) { if (!_first) {
// queue is empty // queue is empty
@@ -178,6 +208,9 @@ class Outbox {
Node* _last; Node* _last;
Node* _current; Node* _current;
Node* _prev; // element just before _current Node* _prev; // element just before _current
#if EMC_USE_MEMPOOL
MemoryPool::Fixed<EMC_NUM_POOL_ELEMENTS, sizeof(Node)> _memPool;
#endif
void _remove(Node* prev, Node* node) { void _remove(Node* prev, Node* node) {
if (!node) return; if (!node) return;
@@ -210,7 +243,12 @@ class Outbox {
} }
// finally, delete the node // finally, delete the node
#if EMC_USE_MEMPOOL
node->~Node();
_memPool.free(node);
#else
delete node; delete node;
#endif
} }
}; };

250
lib/espMqttClient/src/Packets/Packet.cpp
View File

@@ -10,24 +10,28 @@ the LICENSE file.
namespace espMqttClientInternals { namespace espMqttClientInternals {
#if EMC_USE_MEMPOOL
MemoryPool::Variable<EMC_NUM_POOL_ELEMENTS, EMC_SIZE_POOL_ELEMENTS> Packet::_memPool;
#endif
Packet::~Packet() { Packet::~Packet() {
#if EMC_USE_MEMPOOL
_memPool.free(_data);
#else
free(_data); free(_data);
#endif
} }
size_t Packet::available(size_t index) { size_t Packet::available(size_t index) {
if (index >= _size) if (index >= _size) return 0;
return 0; if (!_getPayload) return _size - index;
if (!_getPayload)
return _size - index;
return _chunkedAvailable(index); return _chunkedAvailable(index);
} }
const uint8_t * Packet::data(size_t index) const { const uint8_t* Packet::data(size_t index) const {
if (!_getPayload) { if (!_getPayload) {
if (!_data) if (!_data) return nullptr;
return nullptr; if (index >= _size) return nullptr;
if (index >= _size)
return nullptr;
return &_data[index]; return &_data[index];
} }
return _chunkedData(index); return _chunkedData(index);
@@ -38,12 +42,9 @@ size_t Packet::size() const {
} }
void Packet::setDup() { void Packet::setDup() {
if (!_data) if (!_data) return;
return; if (packetType() != PacketType.PUBLISH) return;
if (packetType() != PacketType.PUBLISH) if (_packetId == 0) return;
return;
if (_packetId == 0)
return;
_data[0] |= 0x08; _data[0] |= 0x08;
} }
@@ -52,39 +53,36 @@ uint16_t Packet::packetId() const {
} }
MQTTPacketType Packet::packetType() const { MQTTPacketType Packet::packetType() const {
if (_data) if (_data) return static_cast<MQTTPacketType>(_data[0] & 0xF0);
return static_cast<MQTTPacketType>(_data[0] & 0xF0);
return static_cast<MQTTPacketType>(0); return static_cast<MQTTPacketType>(0);
} }
bool Packet::removable() const { bool Packet::removable() const {
if (_packetId == 0) if (_packetId == 0) return true;
return true; if ((packetType() == PacketType.PUBACK) || (packetType() == PacketType.PUBCOMP)) return true;
if ((packetType() == PacketType.PUBACK) || (packetType() == PacketType.PUBCOMP))
return true;
return false; return false;
} }
Packet::Packet(espMqttClientTypes::Error & error, Packet::Packet(espMqttClientTypes::Error& error,
bool cleanSession, bool cleanSession,
const char * username, const char* username,
const char * password, const char* password,
const char * willTopic, const char* willTopic,
bool willRetain, bool willRetain,
uint8_t willQos, uint8_t willQos,
const uint8_t * willPayload, const uint8_t* willPayload,
uint16_t willPayloadLength, uint16_t willPayloadLength,
uint16_t keepAlive, uint16_t keepAlive,
const char * clientId) const char* clientId)
: _packetId(0) : _packetId(0)
, _data(nullptr) , _data(nullptr)
, _size(0) , _size(0)
, _payloadIndex(0) , _payloadIndex(0)
, _payloadStartIndex(0) , _payloadStartIndex(0)
, _payloadEndIndex(0) , _payloadEndIndex(0)
, _getPayload(nullptr) { , _getPayload(nullptr) {
if (willPayload && willPayloadLength == 0) { if (willPayload && willPayloadLength == 0) {
size_t length = strlen(reinterpret_cast<const char *>(willPayload)); size_t length = strlen(reinterpret_cast<const char*>(willPayload));
if (length > UINT16_MAX) { if (length > UINT16_MAX) {
emc_log_w("Payload length truncated (l:%zu)", length); emc_log_w("Payload length truncated (l:%zu)", length);
willPayloadLength = UINT16_MAX; willPayloadLength = UINT16_MAX;
@@ -99,12 +97,15 @@ Packet::Packet(espMqttClientTypes::Error & error,
} }
// Calculate size // Calculate size
size_t remainingLength = 6 + // protocol size_t remainingLength =
6 + // protocol
1 + // protocol level 1 + // protocol level
1 + // connect flags 1 + // connect flags
2 + // keepalive 2 + // keepalive
2 + strlen(clientId) + (willTopic ? 2 + strlen(willTopic) + 2 + willPayloadLength : 0) + (username ? 2 + strlen(username) : 0) 2 + strlen(clientId) +
+ (password ? 2 + strlen(password) : 0); (willTopic ? 2 + strlen(willTopic) + 2 + willPayloadLength : 0) +
(username ? 2 + strlen(username) : 0) +
(password ? 2 + strlen(password) : 0);
// allocate memory // allocate memory
if (!_allocate(remainingLength, false)) { if (!_allocate(remainingLength, false)) {
@@ -121,16 +122,12 @@ Packet::Packet(espMqttClientTypes::Error & error,
pos += encodeString(PROTOCOL, &_data[pos]); pos += encodeString(PROTOCOL, &_data[pos]);
_data[pos++] = PROTOCOL_LEVEL; _data[pos++] = PROTOCOL_LEVEL;
uint8_t connectFlags = 0; uint8_t connectFlags = 0;
if (cleanSession) if (cleanSession) connectFlags |= espMqttClientInternals::ConnectFlag.CLEAN_SESSION;
connectFlags |= espMqttClientInternals::ConnectFlag.CLEAN_SESSION; if (username != nullptr) connectFlags |= espMqttClientInternals::ConnectFlag.USERNAME;
if (username != nullptr) if (password != nullptr) connectFlags |= espMqttClientInternals::ConnectFlag.PASSWORD;
connectFlags |= espMqttClientInternals::ConnectFlag.USERNAME;
if (password != nullptr)
connectFlags |= espMqttClientInternals::ConnectFlag.PASSWORD;
if (willTopic != nullptr) { if (willTopic != nullptr) {
connectFlags |= espMqttClientInternals::ConnectFlag.WILL; connectFlags |= espMqttClientInternals::ConnectFlag.WILL;
if (willRetain) if (willRetain) connectFlags |= espMqttClientInternals::ConnectFlag.WILL_RETAIN;
connectFlags |= espMqttClientInternals::ConnectFlag.WILL_RETAIN;
switch (willQos) { switch (willQos) {
case 0: case 0:
connectFlags |= espMqttClientInternals::ConnectFlag.WILL_QOS0; connectFlags |= espMqttClientInternals::ConnectFlag.WILL_QOS0;
@@ -159,23 +156,28 @@ Packet::Packet(espMqttClientTypes::Error & error,
pos += willPayloadLength; pos += willPayloadLength;
} }
// credentials // credentials
if (username != nullptr) if (username != nullptr) pos += encodeString(username, &_data[pos]);
pos += encodeString(username, &_data[pos]); if (password != nullptr) encodeString(password, &_data[pos]);
if (password != nullptr)
encodeString(password, &_data[pos]);
error = espMqttClientTypes::Error::SUCCESS; error = espMqttClientTypes::Error::SUCCESS;
} }
Packet::Packet(espMqttClientTypes::Error & error, uint16_t packetId, const char * topic, const uint8_t * payload, size_t payloadLength, uint8_t qos, bool retain) Packet::Packet(espMqttClientTypes::Error& error,
: _packetId(packetId) uint16_t packetId,
, _data(nullptr) const char* topic,
, _size(0) const uint8_t* payload,
, _payloadIndex(0) size_t payloadLength,
, _payloadStartIndex(0) uint8_t qos,
, _payloadEndIndex(0) bool retain)
, _getPayload(nullptr) { : _packetId(packetId)
size_t remainingLength = 2 + strlen(topic) + // topic length + topic , _data(nullptr)
, _size(0)
, _payloadIndex(0)
, _payloadStartIndex(0)
, _payloadEndIndex(0)
, _getPayload(nullptr) {
size_t remainingLength =
2 + strlen(topic) + // topic length + topic
2 + // packet ID 2 + // packet ID
payloadLength; payloadLength;
@@ -184,7 +186,7 @@ Packet::Packet(espMqttClientTypes::Error & error, uint16_t packetId, const char
_packetId = 0; _packetId = 0;
} }
if (!_allocate(remainingLength)) { if (!_allocate(remainingLength, true)) {
error = espMqttClientTypes::Error::OUT_OF_MEMORY; error = espMqttClientTypes::Error::OUT_OF_MEMORY;
return; return;
} }
@@ -197,21 +199,22 @@ Packet::Packet(espMqttClientTypes::Error & error, uint16_t packetId, const char
error = espMqttClientTypes::Error::SUCCESS; error = espMqttClientTypes::Error::SUCCESS;
} }
Packet::Packet(espMqttClientTypes::Error & error, Packet::Packet(espMqttClientTypes::Error& error,
uint16_t packetId, uint16_t packetId,
const char * topic, const char* topic,
espMqttClientTypes::PayloadCallback payloadCallback, espMqttClientTypes::PayloadCallback payloadCallback,
size_t payloadLength, size_t payloadLength,
uint8_t qos, uint8_t qos,
bool retain) bool retain)
: _packetId(packetId) : _packetId(packetId)
, _data(nullptr) , _data(nullptr)
, _size(0) , _size(0)
, _payloadIndex(0) , _payloadIndex(0)
, _payloadStartIndex(0) , _payloadStartIndex(0)
, _payloadEndIndex(0) , _payloadEndIndex(0)
, _getPayload(payloadCallback) { , _getPayload(payloadCallback) {
size_t remainingLength = 2 + strlen(topic) + // topic length + topic size_t remainingLength =
2 + strlen(topic) + // topic length + topic
2 + // packet ID 2 + // packet ID
payloadLength; payloadLength;
@@ -220,7 +223,7 @@ Packet::Packet(espMqttClientTypes::Error & error,
_packetId = 0; _packetId = 0;
} }
if (!_allocate(remainingLength - payloadLength + std::min(payloadLength, static_cast<size_t>(EMC_RX_BUFFER_SIZE)))) { if (!_allocate(remainingLength - payloadLength + std::min(payloadLength, static_cast<size_t>(EMC_RX_BUFFER_SIZE)), true)) {
error = espMqttClientTypes::Error::OUT_OF_MEMORY; error = espMqttClientTypes::Error::OUT_OF_MEMORY;
return; return;
} }
@@ -236,27 +239,27 @@ Packet::Packet(espMqttClientTypes::Error & error,
error = espMqttClientTypes::Error::SUCCESS; error = espMqttClientTypes::Error::SUCCESS;
} }
Packet::Packet(espMqttClientTypes::Error & error, uint16_t packetId, const char * topic, uint8_t qos) Packet::Packet(espMqttClientTypes::Error& error, uint16_t packetId, const char* topic, uint8_t qos)
: _packetId(packetId) : _packetId(packetId)
, _data(nullptr) , _data(nullptr)
, _size(0) , _size(0)
, _payloadIndex(0) , _payloadIndex(0)
, _payloadStartIndex(0) , _payloadStartIndex(0)
, _payloadEndIndex(0) , _payloadEndIndex(0)
, _getPayload(nullptr) { , _getPayload(nullptr) {
SubscribeItem list[1] = {{topic, qos}}; SubscribeItem list[1] = {topic, qos};
_createSubscribe(error, list, 1); _createSubscribe(error, list, 1);
} }
Packet::Packet(espMqttClientTypes::Error & error, MQTTPacketType type, uint16_t packetId) Packet::Packet(espMqttClientTypes::Error& error, MQTTPacketType type, uint16_t packetId)
: _packetId(packetId) : _packetId(packetId)
, _data(nullptr) , _data(nullptr)
, _size(0) , _size(0)
, _payloadIndex(0) , _payloadIndex(0)
, _payloadStartIndex(0) , _payloadStartIndex(0)
, _payloadEndIndex(0) , _payloadEndIndex(0)
, _getPayload(nullptr) { , _getPayload(nullptr) {
if (!_allocate(2)) { if (!_allocate(2, true)) {
error = espMqttClientTypes::Error::OUT_OF_MEMORY; error = espMqttClientTypes::Error::OUT_OF_MEMORY;
return; return;
} }
@@ -275,27 +278,27 @@ Packet::Packet(espMqttClientTypes::Error & error, MQTTPacketType type, uint16_t
error = espMqttClientTypes::Error::SUCCESS; error = espMqttClientTypes::Error::SUCCESS;
} }
Packet::Packet(espMqttClientTypes::Error & error, uint16_t packetId, const char * topic) Packet::Packet(espMqttClientTypes::Error& error, uint16_t packetId, const char* topic)
: _packetId(packetId) : _packetId(packetId)
, _data(nullptr) , _data(nullptr)
, _size(0) , _size(0)
, _payloadIndex(0) , _payloadIndex(0)
, _payloadStartIndex(0) , _payloadStartIndex(0)
, _payloadEndIndex(0) , _payloadEndIndex(0)
, _getPayload(nullptr) { , _getPayload(nullptr) {
const char * list[1] = {topic}; const char* list[1] = {topic};
_createUnsubscribe(error, list, 1); _createUnsubscribe(error, list, 1);
} }
Packet::Packet(espMqttClientTypes::Error & error, MQTTPacketType type) Packet::Packet(espMqttClientTypes::Error& error, MQTTPacketType type)
: _packetId(0) : _packetId(0)
, _data(nullptr) , _data(nullptr)
, _size(0) , _size(0)
, _payloadIndex(0) , _payloadIndex(0)
, _payloadStartIndex(0) , _payloadStartIndex(0)
, _payloadEndIndex(0) , _payloadEndIndex(0)
, _getPayload(nullptr) { , _getPayload(nullptr) {
if (!_allocate(0)) { if (!_allocate(0, true)) {
error = espMqttClientTypes::Error::OUT_OF_MEMORY; error = espMqttClientTypes::Error::OUT_OF_MEMORY;
return; return;
} }
@@ -306,12 +309,20 @@ Packet::Packet(espMqttClientTypes::Error & error, MQTTPacketType type)
bool Packet::_allocate(size_t remainingLength, bool check) { bool Packet::_allocate(size_t remainingLength, bool check) {
#if EMC_USE_MEMPOOL
(void) check;
#else
if (check && EMC_GET_FREE_MEMORY() < EMC_MIN_FREE_MEMORY) { if (check && EMC_GET_FREE_MEMORY() < EMC_MIN_FREE_MEMORY) {
emc_log_w("Packet buffer not allocated: low memory"); emc_log_w("Packet buffer not allocated: low memory");
return false; return false;
} }
#endif
_size = 1 + remainingLengthLength(remainingLength) + remainingLength; _size = 1 + remainingLengthLength(remainingLength) + remainingLength;
_data = reinterpret_cast<uint8_t *>(malloc(_size)); #if EMC_USE_MEMPOOL
_data = reinterpret_cast<uint8_t*>(_memPool.malloc(_size));
#else
_data = reinterpret_cast<uint8_t*>(malloc(_size));
#endif
if (!_data) { if (!_data) {
_size = 0; _size = 0;
emc_log_w("Alloc failed (l:%zu)", _size); emc_log_w("Alloc failed (l:%zu)", _size);
@@ -322,13 +333,16 @@ bool Packet::_allocate(size_t remainingLength, bool check) {
return true; return true;
} }
size_t Packet::_fillPublishHeader(uint16_t packetId, const char * topic, size_t remainingLength, uint8_t qos, bool retain) { size_t Packet::_fillPublishHeader(uint16_t packetId,
const char* topic,
size_t remainingLength,
uint8_t qos,
bool retain) {
size_t index = 0; size_t index = 0;
// FIXED HEADER // FIXED HEADER
_data[index] = PacketType.PUBLISH; _data[index] = PacketType.PUBLISH;
if (retain) if (retain) _data[index] |= HeaderFlag.PUBLISH_RETAIN;
_data[index] |= HeaderFlag.PUBLISH_RETAIN;
if (qos == 0) { if (qos == 0) {
_data[index++] |= HeaderFlag.PUBLISH_QOS0; _data[index++] |= HeaderFlag.PUBLISH_QOS0;
} else if (qos == 1) { } else if (qos == 1) {
@@ -348,7 +362,9 @@ size_t Packet::_fillPublishHeader(uint16_t packetId, const char * topic, size_t
return index; return index;
} }
void Packet::_createSubscribe(espMqttClientTypes::Error & error, SubscribeItem * list, size_t numberTopics) { void Packet::_createSubscribe(espMqttClientTypes::Error& error,
SubscribeItem* list,
size_t numberTopics) {
// Calculate size // Calculate size
size_t payload = 0; size_t payload = 0;
for (size_t i = 0; i < numberTopics; ++i) { for (size_t i = 0; i < numberTopics; ++i) {
@@ -357,7 +373,7 @@ void Packet::_createSubscribe(espMqttClientTypes::Error & error, SubscribeItem *
size_t remainingLength = 2 + payload; // packetId + payload size_t remainingLength = 2 + payload; // packetId + payload
// allocate memory // allocate memory
if (!_allocate(remainingLength)) { if (!_allocate(remainingLength, true)) {
error = espMqttClientTypes::Error::OUT_OF_MEMORY; error = espMqttClientTypes::Error::OUT_OF_MEMORY;
return; return;
} }
@@ -376,7 +392,9 @@ void Packet::_createSubscribe(espMqttClientTypes::Error & error, SubscribeItem *
error = espMqttClientTypes::Error::SUCCESS; error = espMqttClientTypes::Error::SUCCESS;
} }
void Packet::_createUnsubscribe(espMqttClientTypes::Error & error, const char ** list, size_t numberTopics) { void Packet::_createUnsubscribe(espMqttClientTypes::Error& error,
const char** list,
size_t numberTopics) {
// Calculate size // Calculate size
size_t payload = 0; size_t payload = 0;
for (size_t i = 0; i < numberTopics; ++i) { for (size_t i = 0; i < numberTopics; ++i) {
@@ -385,7 +403,7 @@ void Packet::_createUnsubscribe(espMqttClientTypes::Error & error, const char **
size_t remainingLength = 2 + payload; // packetId + payload size_t remainingLength = 2 + payload; // packetId + payload
// allocate memory // allocate memory
if (!_allocate(remainingLength)) { if (!_allocate(remainingLength, true)) {
error = espMqttClientTypes::Error::OUT_OF_MEMORY; error = espMqttClientTypes::Error::OUT_OF_MEMORY;
return; return;
} }
@@ -425,7 +443,7 @@ size_t Packet::_chunkedAvailable(size_t index) {
return _payloadEndIndex - index + 1; return _payloadEndIndex - index + 1;
} }
const uint8_t * Packet::_chunkedData(size_t index) const { const uint8_t* Packet::_chunkedData(size_t index) const {
// CAUTION!! available(index) has to be called first to check available data and possibly fill payloadbuffer // CAUTION!! available(index) has to be called first to check available data and possibly fill payloadbuffer
if (index < _payloadIndex) { if (index < _payloadIndex) {
return &_data[index]; return &_data[index];

12
lib/espMqttClient/src/Packets/Packet.h
View File

@@ -17,7 +17,11 @@ the LICENSE file.
#include "../Helpers.h" #include "../Helpers.h"
#include "../Logging.h" #include "../Logging.h"
#include "RemainingLength.h" #include "RemainingLength.h"
#include "String.h" #include "StringUtil.h"
#if EMC_USE_MEMPOOL
#include "MemoryPool/src/MemoryPool.h"
#endif
namespace espMqttClientInternals { namespace espMqttClientInternals {
@@ -133,7 +137,7 @@ class Packet {
private: private:
// pass remainingLength = total size - header - remainingLengthLength! // pass remainingLength = total size - header - remainingLengthLength!
bool _allocate(size_t remainingLength, bool check = true); bool _allocate(size_t remainingLength, bool check);
// fills header and returns index of next available byte in buffer // fills header and returns index of next available byte in buffer
size_t _fillPublishHeader(uint16_t packetId, size_t _fillPublishHeader(uint16_t packetId,
@@ -150,6 +154,10 @@ class Packet {
size_t _chunkedAvailable(size_t index); size_t _chunkedAvailable(size_t index);
const uint8_t* _chunkedData(size_t index) const; const uint8_t* _chunkedData(size_t index) const;
#if EMC_USE_MEMPOOL
static MemoryPool::Variable<EMC_NUM_POOL_ELEMENTS, EMC_SIZE_POOL_ELEMENTS> _memPool;
#endif
}; };
} // end namespace espMqttClientInternals } // end namespace espMqttClientInternals

2
lib/espMqttClient/src/Packets/String.cpp → lib/espMqttClient/src/Packets/StringUtil.cpp
View File

@@ -6,7 +6,7 @@ For a copy, see <https://opensource.org/licenses/MIT> or
the LICENSE file. the LICENSE file.
*/ */
#include "String.h" #include "StringUtil.h"
namespace espMqttClientInternals { namespace espMqttClientInternals {

0
lib/espMqttClient/src/Packets/String.h → lib/espMqttClient/src/Packets/StringUtil.h
View File

3
lib/espMqttClient/src/Transport/ClientPosix.cpp
View File

@@ -38,9 +38,10 @@ bool ClientPosix::connect(IPAddress ip, uint16_t port) {
memset(&_host, 0, sizeof(_host)); memset(&_host, 0, sizeof(_host));
_host.sin_family = AF_INET; _host.sin_family = AF_INET;
_host.sin_addr.s_addr = htonl(uint32_t(ip)); _host.sin_addr.s_addr = htonl(uint32_t(ip));
_host.sin_port = htons(port); // modified by proddy for EMS-ESP compiling standalone _host.sin_port = ::htons(port);
int ret = ::connect(_sockfd, reinterpret_cast<sockaddr*>(&_host), sizeof(_host)); int ret = ::connect(_sockfd, reinterpret_cast<sockaddr*>(&_host), sizeof(_host));
if (ret < 0) { if (ret < 0) {
emc_log_e("Error connecting: %d - (%d) %s", ret, errno, strerror(errno)); emc_log_e("Error connecting: %d - (%d) %s", ret, errno, strerror(errno));
return false; return false;

2
lib/espMqttClient/src/Transport/IPAddress.cpp → lib/espMqttClient/src/Transport/ClientPosixIPAddress.cpp
View File

@@ -8,7 +8,7 @@ the LICENSE file.
#if defined(__linux__) #if defined(__linux__)
#include "IPAddress.h" #include "ClientPosixIPAddress.h"
IPAddress::IPAddress() IPAddress::IPAddress()
: _address(0) { : _address(0) {

0
lib/espMqttClient/src/Transport/IPAddress.h → lib/espMqttClient/src/Transport/ClientPosixIPAddress.h
View File

2
lib/espMqttClient/src/Transport/ClientSecureSync.h
View File

@@ -17,7 +17,6 @@ the LICENSE file.
#else #else
#include <WiFiClient.h> #include <WiFiClient.h>
#endif #endif
#include "Transport.h" #include "Transport.h"
namespace espMqttClientInternals { namespace espMqttClientInternals {
@@ -32,6 +31,7 @@ class ClientSecureSync : public Transport {
void stop() override; void stop() override;
bool connected() override; bool connected() override;
bool disconnected() override; bool disconnected() override;
// added for EMS-ESP
#if defined(EMC_CLIENT_SECURE) #if defined(EMC_CLIENT_SECURE)
WiFiClientSecure client; WiFiClientSecure client;
#else #else

2
lib/espMqttClient/src/Transport/Transport.h
View File

@@ -10,7 +10,7 @@ the LICENSE file.
#include <stddef.h> // size_t #include <stddef.h> // size_t
#include "IPAddress.h" #include "ClientPosixIPAddress.h"
namespace espMqttClientInternals { namespace espMqttClientInternals {

50
lib/espMqttClient/src/espMqttClient.cpp
View File

@@ -10,43 +10,43 @@ the LICENSE file.
#if defined(ARDUINO_ARCH_ESP8266) #if defined(ARDUINO_ARCH_ESP8266)
espMqttClient::espMqttClient() espMqttClient::espMqttClient()
: MqttClientSetup(espMqttClientTypes::UseInternalTask::NO) : MqttClientSetup(espMqttClientTypes::UseInternalTask::NO)
, _client() { , _client() {
_transport = &_client; _transport = &_client;
} }
espMqttClientSecure::espMqttClientSecure() espMqttClientSecure::espMqttClientSecure()
: MqttClientSetup(espMqttClientTypes::UseInternalTask::NO) : MqttClientSetup(espMqttClientTypes::UseInternalTask::NO)
, _client() { , _client() {
_transport = &_client; _transport = &_client;
} }
espMqttClientSecure& espMqttClientSecure::setInsecure() { espMqttClientSecure & espMqttClientSecure::setInsecure() {
_client.client.setInsecure(); _client.client.setInsecure();
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setFingerprint(const uint8_t fingerprint[20]) { espMqttClientSecure & espMqttClientSecure::setFingerprint(const uint8_t fingerprint[20]) {
_client.client.setFingerprint(fingerprint); _client.client.setFingerprint(fingerprint);
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setTrustAnchors(const X509List *ta) { espMqttClientSecure & espMqttClientSecure::setTrustAnchors(const X509List * ta) {
_client.client.setTrustAnchors(ta); _client.client.setTrustAnchors(ta);
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setClientRSACert(const X509List *cert, const PrivateKey *sk) { espMqttClientSecure & espMqttClientSecure::setClientRSACert(const X509List * cert, const PrivateKey * sk) {
_client.client.setClientRSACert(cert, sk); _client.client.setClientRSACert(cert, sk);
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setClientECCert(const X509List *cert, const PrivateKey *sk, unsigned allowed_usages, unsigned cert_issuer_key_type) { espMqttClientSecure & espMqttClientSecure::setClientECCert(const X509List * cert, const PrivateKey * sk, unsigned allowed_usages, unsigned cert_issuer_key_type) {
_client.client.setClientECCert(cert, sk, allowed_usages, cert_issuer_key_type); _client.client.setClientECCert(cert, sk, allowed_usages, cert_issuer_key_type);
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setCertStore(CertStoreBase *certStore) { espMqttClientSecure & espMqttClientSecure::setCertStore(CertStoreBase * certStore) {
_client.client.setCertStore(certStore); _client.client.setCertStore(certStore);
return *this; return *this;
} }
@@ -54,58 +54,58 @@ espMqttClientSecure& espMqttClientSecure::setCertStore(CertStoreBase *certStore)
#if defined(ARDUINO_ARCH_ESP32) #if defined(ARDUINO_ARCH_ESP32)
espMqttClient::espMqttClient(espMqttClientTypes::UseInternalTask useInternalTask) espMqttClient::espMqttClient(espMqttClientTypes::UseInternalTask useInternalTask)
: MqttClientSetup(useInternalTask) : MqttClientSetup(useInternalTask)
, _client() { , _client() {
_transport = &_client; _transport = &_client;
} }
espMqttClient::espMqttClient(uint8_t priority, uint8_t core) espMqttClient::espMqttClient(uint8_t priority, uint8_t core)
: MqttClientSetup(espMqttClientTypes::UseInternalTask::YES, priority, core) : MqttClientSetup(espMqttClientTypes::UseInternalTask::YES, priority, core)
, _client() { , _client() {
_transport = &_client; _transport = &_client;
} }
espMqttClientSecure::espMqttClientSecure(espMqttClientTypes::UseInternalTask useInternalTask) espMqttClientSecure::espMqttClientSecure(espMqttClientTypes::UseInternalTask useInternalTask)
: MqttClientSetup(useInternalTask) : MqttClientSetup(useInternalTask)
, _client() { , _client() {
_transport = &_client; _transport = &_client;
} }
espMqttClientSecure::espMqttClientSecure(uint8_t priority, uint8_t core) espMqttClientSecure::espMqttClientSecure(uint8_t priority, uint8_t core)
: MqttClientSetup(espMqttClientTypes::UseInternalTask::YES, priority, core) : MqttClientSetup(espMqttClientTypes::UseInternalTask::YES, priority, core)
, _client() { , _client() {
_transport = &_client; _transport = &_client;
} }
espMqttClientSecure& espMqttClientSecure::setInsecure() { espMqttClientSecure & espMqttClientSecure::setInsecure() {
#if defined(EMC_CLIENT_SECURE) #if defined(EMC_CLIENT_SECURE)
_client.client.setInsecure(); _client.client.setInsecure();
#endif #endif
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setCACert(const char* rootCA) { espMqttClientSecure & espMqttClientSecure::setCACert(const char * rootCA) {
#if defined(EMC_CLIENT_SECURE) #if defined(EMC_CLIENT_SECURE)
_client.client.setCACert(rootCA); _client.client.setCACert(rootCA);
#endif #endif
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setCertificate(const char* clientCa) { espMqttClientSecure & espMqttClientSecure::setCertificate(const char * clientCa) {
#if defined(EMC_CLIENT_SECURE) #if defined(EMC_CLIENT_SECURE)
_client.client.setCertificate(clientCa); _client.client.setCertificate(clientCa);
#endif #endif
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setPrivateKey(const char* privateKey) { espMqttClientSecure & espMqttClientSecure::setPrivateKey(const char * privateKey) {
#if defined(EMC_CLIENT_SECURE) #if defined(EMC_CLIENT_SECURE)
_client.client.setPrivateKey(privateKey); _client.client.setPrivateKey(privateKey);
#endif #endif
return *this; return *this;
} }
espMqttClientSecure& espMqttClientSecure::setPreSharedKey(const char* pskIdent, const char* psKey) { espMqttClientSecure & espMqttClientSecure::setPreSharedKey(const char * pskIdent, const char * psKey) {
#if defined(EMC_CLIENT_SECURE) #if defined(EMC_CLIENT_SECURE)
_client.client.setPreSharedKey(pskIdent, psKey); _client.client.setPreSharedKey(pskIdent, psKey);
#endif #endif
@@ -116,8 +116,8 @@ espMqttClientSecure& espMqttClientSecure::setPreSharedKey(const char* pskIdent,
#if defined(__linux__) #if defined(__linux__)
espMqttClient::espMqttClient() espMqttClient::espMqttClient()
: MqttClientSetup(espMqttClientTypes::UseInternalTask::NO) : MqttClientSetup(espMqttClientTypes::UseInternalTask::NO)
, _client() { , _client() {
_transport = &_client; _transport = &_client;
} }
#endif #endif