From a6b3540cb8b5ddd648712efb9e3ba1a91c625d8d Mon Sep 17 00:00:00 2001 From: hsaturn Date: Fri, 17 Sep 2021 22:32:00 +0200 Subject: [PATCH] Fix issue_2 : Broken large payloads --- README.md | 2 +- library.json | 2 +- library.properties | 2 +- src/TinyMqtt.cpp | 60 ++++++++++------------------- src/TinyMqtt.h | 25 ++++++------ tests/nowifi-tests/Makefile | 5 +++ tests/nowifi-tests/nowifi-tests.ino | 18 ++++++++- 7 files changed, 59 insertions(+), 55 deletions(-) diff --git a/README.md b/README.md index 4504255..030fb1f 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ TinyMqtt is a small, fast and capable Mqtt Broker and Client for Esp8266 / Esp32 ## Features -- Very (very !!) fast broker I saw it re-sent 1000 topics per second for two +- Very fast broker I saw it re-sent 1000 topics per second for two clients that had subscribed (payload ~15 bytes ESP8266). No topic lost. The max I've seen was 2k msg/s (1 client 1 subscription) - Act as as a mqtt broker and/or a mqtt client diff --git a/library.json b/library.json index 65584f9..96d3ac2 100644 --- a/library.json +++ b/library.json @@ -6,7 +6,7 @@ "type": "git", "url": "https://github.com/hsaturn/TinyMqtt.git" }, - "version": "0.7.9", + "version": "0.8.0", "exclude": "", "examples": "examples/*/*.ino", "frameworks": "arduino", diff --git a/library.properties b/library.properties index 6a23b7a..4581787 100644 --- a/library.properties +++ b/library.properties @@ -1,5 +1,5 @@ name=TinyMqtt -version=0.7.9 +version=0.8.0 author=Francois BIOT, HSaturn, maintainer=Francois BIOT, HSaturn, sentence=A tiny broker and client library for MQTT messaging. diff --git a/src/TinyMqtt.cpp b/src/TinyMqtt.cpp index 9077ba9..9bed544 100644 --- a/src/TinyMqtt.cpp +++ b/src/TinyMqtt.cpp @@ -38,7 +38,7 @@ MqttClient::MqttClient(MqttBroker* parent, TcpClient* new_client) #else client = new WiFiClient(*new_client); #endif - alive = millis()+5000; // client expires after 5s if no CONNECT msg + alive = millis()+5000; // TODO MAGIC client expires after 5s if no CONNECT msg } MqttClient::MqttClient(MqttBroker* parent, const std::string& id) @@ -190,7 +190,7 @@ MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos) return MqttNowhereToSend; } -MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, const MqttMessage& msg) const +MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, MqttMessage& msg) const { MqttError retval = MqttOk; @@ -397,7 +397,7 @@ MqttError MqttClient::sendTopic(const Topic& topic, MqttMessage::Type type, uint long MqttClient::counter=0; -void MqttClient::processMessage(const MqttMessage* mesg) +void MqttClient::processMessage(MqttMessage* mesg) { counter++; #ifdef TINY_MQTT_DEBUG @@ -627,6 +627,8 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa MqttMessage msg(MqttMessage::Publish); msg.add(topic); msg.add(payload, pay_length, false); + msg.complete(); + if (parent) { return parent->publish(this, topic, msg); @@ -638,7 +640,7 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa } // republish a received publish if it matches any in subscriptions -MqttError MqttClient::publishIfSubscribed(const Topic& topic, const MqttMessage& msg) +MqttError MqttClient::publishIfSubscribed(const Topic& topic, MqttMessage& msg) { MqttError retval=MqttOk; @@ -682,29 +684,23 @@ void MqttMessage::incoming(char in_byte) switch(state) { case FixedHeader: - size=0; + size=MaxBufferLength; state = Length; break; case Length: - if (size==0) + + if (size==MaxBufferLength) size = in_byte & 0x7F; - else if (size<128) - size += static_cast(in_byte & 0x7F)<<7; else - state = Error; // Really don't want to handle msg with length > 16k + size += static_cast(in_byte & 0x7F)<<7; + if (size > MaxBufferLength) - { state = Error; - } else if ((in_byte & 0x80) == 0) { vheader = buffer.length(); if (size==0) state = Complete; - else if (size > 500) // TODO magic - { - state = Error; - } else { buffer.reserve(size); @@ -749,35 +745,21 @@ void MqttMessage::add(const char* p, size_t len, bool addLength) while(len--) incoming(*p++); } -void MqttMessage::encodeLength() const +void MqttMessage::encodeLength() { if (state != Complete) { - int length = buffer.size()-2; // 1 byte for header, 1 byte for pre-reserved length field. - std::string::size_type ins=1; - do - { - uint8_t encoded(length & 0x7F); - length >>=7; - if (length) encoded |= 0x80; - - if (ins==1) - buffer[ins]=encoded; - else - buffer.insert(ins, 1, encoded); - // On pourrait optimiser, cet insert est couteux, il faudrait en fait non pas - // insérer, mais réserver 4 octets pour les remplir - // plus tard avec ke fixed header et la taille. - // Cela changerait en revanche le début du message qui ne serait plus - // buffer[0], mais buffer[0..3] selon la taille du message. - - ++ins; - } while (length); - state = Complete; - } + int length = buffer.size()-3; // 3 = 1 byte for header + 2 bytes for pre-reserved length field. + buffer[1] = 0x80 | (length & 0x7F); + buffer[2] = (length >> 7); + vheader = 3; + + // We could check that buffer[2] < 128 (end of length encoding) + state = Complete; + } }; -MqttError MqttMessage::sendTo(MqttClient* client) const +MqttError MqttMessage::sendTo(MqttClient* client) { if (buffer.size()) { diff --git a/src/TinyMqtt.h b/src/TinyMqtt.h index 2a8d01a..ecea277 100644 --- a/src/TinyMqtt.h +++ b/src/TinyMqtt.h @@ -64,7 +64,7 @@ class Topic : public IndexedString class MqttClient; class MqttMessage { - const uint16_t MaxBufferLength = 4096; //hard limit: 16k + const uint16_t MaxBufferLength = 4096; //hard limit: 16k due to size decoding public: enum Type { @@ -101,6 +101,7 @@ class MqttMessage void add(const Topic& t) { add(t.str()); } const char* end() const { return &buffer[0]+buffer.size(); } const char* getVHeader() const { return &buffer[vheader]; } + void complete() { encodeLength(); } void reset(); @@ -116,21 +117,22 @@ class MqttMessage void create(Type type) { buffer=(char)type; - buffer+='\0'; // reserved for msg length - vheader=2; + buffer+='\0'; // reserved for msg length byte 1/2 + buffer+='\0'; // reserved for msg length byte 2/2 (fixed) + vheader=3; // Should never change size=0; state=Create; } - MqttError sendTo(MqttClient*) const; + MqttError sendTo(MqttClient*); void hexdump(const char* prefix=nullptr) const; private: - void encodeLength() const; + void encodeLength(); - mutable std::string buffer; // mutable -> sendTo() + std::string buffer; uint8_t vheader; uint16_t size; // bytes left to receive - mutable State state; // mutable -> encodeLength() + State state; }; class MqttBroker; @@ -215,8 +217,7 @@ class MqttClient Serial << endl; } - /** Count the number of messages that have been sent **/ - static long counter; + static long counter; // Number of processed messages private: @@ -231,10 +232,10 @@ class MqttClient friend class MqttBroker; MqttClient(MqttBroker* parent, TcpClient* client); // republish a received publish if topic matches any in subscriptions - MqttError publishIfSubscribed(const Topic& topic, const MqttMessage& msg); + MqttError publishIfSubscribed(const Topic& topic, MqttMessage& msg); void clientAlive(uint32_t more_seconds); - void processMessage(const MqttMessage* message); + void processMessage(MqttMessage* message); bool mqtt_connected = false; char mqtt_flags; @@ -291,7 +292,7 @@ class MqttBroker { return compareString(auth_password, password, len); } - MqttError publish(const MqttClient* source, const Topic& topic, const MqttMessage& msg) const; + MqttError publish(const MqttClient* source, const Topic& topic, MqttMessage& msg) const; MqttError subscribe(const Topic& topic, uint8_t qos); diff --git a/tests/nowifi-tests/Makefile b/tests/nowifi-tests/Makefile index 5e4a14e..3d297d7 100644 --- a/tests/nowifi-tests/Makefile +++ b/tests/nowifi-tests/Makefile @@ -1,6 +1,11 @@ # See https://github.com/bxparks/EpoxyDuino for documentation about this # Makefile to compile and run Arduino programs natively on Linux or MacOS. +EXTRA_CXXFLAGS=-g3 -O0 + +# Remove flto flag from EpoxyDuino (too many ) +CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics + APP_NAME := nowifi-tests ARDUINO_LIBS := AUnit AceCommon AceTime TinyMqtt EspMock ESP8266WiFi ESPAsyncTCP ARDUINO_LIB_DIRS := ../../../EspMock/libraries diff --git a/tests/nowifi-tests/nowifi-tests.ino b/tests/nowifi-tests/nowifi-tests.ino index a0346cf..fc32904 100644 --- a/tests/nowifi-tests/nowifi-tests.ino +++ b/tests/nowifi-tests/nowifi-tests.ino @@ -132,7 +132,7 @@ test(nowifi_nocallback_when_destroyed) assertEqual(published.size(), (size_t)1); // Only one publish has been received } -test(nowifi_payload_nullptr) +test(nowifi_small_payload) { published.clear(); @@ -150,6 +150,22 @@ test(nowifi_payload_nullptr) assertEqual(lastLength, (size_t)4); } +test(nowifi_hudge_payload) +{ + const char* payload="This payload is hudge, just because its length exceeds 127. Thus when encoding length, we have to encode it on two bytes at min. This should not prevent the message from being encoded and decoded successfully !"; + + MqttClient subscriber(&broker); + subscriber.setCallback(onPublish); + subscriber.subscribe("a/b"); + + MqttClient publisher(&broker); + publisher.publish("a/b", payload); // This publish is received + + // onPublish should have filled lastPayload and lastLength + assertEqual(payload, lastPayload); + assertEqual(lastLength, strlen(payload)); +} + //---------------------------------------------------------------------------- // setup() and loop() void setup() {