From d39c58d8f52f7feb5af7910884f9f22f491b0a58 Mon Sep 17 00:00:00 2001 From: hsaturn Date: Fri, 17 Sep 2021 19:51:32 +0200 Subject: [PATCH 1/2] Fix issue_2 broken payload --- src/TinyMqtt.cpp | 76 +++++++++++++++++++++++++++++++++--------------- src/TinyMqtt.h | 19 ++++++++---- 2 files changed, 65 insertions(+), 30 deletions(-) diff --git a/src/TinyMqtt.cpp b/src/TinyMqtt.cpp index 9e164a7..9077ba9 100644 --- a/src/TinyMqtt.cpp +++ b/src/TinyMqtt.cpp @@ -59,7 +59,7 @@ void MqttClient::close(bool bSendDisconnect) { debug("close " << id().c_str()); mqtt_connected = false; - if (client) + if (client) // connected to a remote broker { if (bSendDisconnect and client->connected()) { @@ -72,10 +72,16 @@ void MqttClient::close(bool bSendDisconnect) if (parent) { parent->removeClient(this); - parent=nullptr; + parent = nullptr; } } +void MqttClient::connect(MqttBroker* parentBroker) +{ + close(); + parent = parentBroker; +} + void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka) { debug("cnx: closing"); @@ -397,12 +403,13 @@ void MqttClient::processMessage(const MqttMessage* mesg) #ifdef TINY_MQTT_DEBUG if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::Type::PingResp) { -#ifdef NOT_ESP_CORE - Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << " ESP.getFreeHeap() "<< endl; -#else - Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl; -#endif + #ifdef NOT_ESP_CORE + Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << " ESP.getFreeHeap() "<< endl; + #else + Serial << "---> INCOMING " << _HEX(mesg->type()) << " client(" << (dbg_ptr)client << ':' << clientId << ") mem=" << ESP.getFreeHeap() << endl; + #endif // mesg->hexdump("Incoming"); + mesg->hexdump("Incoming"); } #endif auto header = mesg->getVHeader(); @@ -544,6 +551,9 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T break; case MqttMessage::Type::Publish: + #ifdef TINY_MQTT_DEBUG + Serial << "publish " << mqtt_connected << '/' << (long) client << endl; + #endif if (mqtt_connected or client == nullptr) { uint8_t qos = mesg->type() & 0x6; @@ -558,8 +568,12 @@ if (mesg->type() != MqttMessage::Type::PingReq && mesg->type() != MqttMessage::T // TODO reset DUP // TODO reset RETAIN - if (client==nullptr) // internal MqttClient receives publish + if (parent==nullptr or client==nullptr) // internal MqttClient receives publish { + #ifdef TINY_MQTT_DEBUG + Serial << (isSubscribedTo(published) ? "not" : "") << " subscribed.\n"; + Serial << "has " << (callback ? "" : "no ") << " callback.\n"; + #endif if (callback and isSubscribedTo(published)) { callback(this, published, payload, len); // TODO send the real payload @@ -613,7 +627,6 @@ 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); @@ -637,6 +650,10 @@ MqttError MqttClient::publishIfSubscribed(const Topic& topic, const MqttMessage& else { processMessage(&msg); + + #ifdef TINY_MQTT_DEBUG + Serial << "Should call the callback ?\n"; + #endif // callback(this, topic, nullptr, 0); // TODO Payload } } @@ -714,7 +731,7 @@ void MqttMessage::incoming(char in_byte) reset(); break; } - if (buffer.length() > MaxBufferLength) // TODO magic 256 ? + if (buffer.length() > MaxBufferLength) { debug("Too long " << state); reset(); @@ -725,36 +742,47 @@ void MqttMessage::add(const char* p, size_t len, bool addLength) { if (addLength) { - buffer.reserve(buffer.length()+addLength+2); + buffer.reserve(buffer.length()+2); incoming(len>>8); incoming(len & 0xFF); } while(len--) incoming(*p++); } -void MqttMessage::encodeLength(char* msb, int length) const +void MqttMessage::encodeLength() const { - do + if (state != Complete) { - uint8_t encoded(length & 0x7F); - length >>=7; - if (length) encoded |= 0x80; - *msb++ = encoded; - } while (length); -}; + 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; -void MqttMessage::complete() -{ - encodeLength(&buffer[1], buffer.size()-2); + 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; -} + } +}; MqttError MqttMessage::sendTo(MqttClient* client) const { if (buffer.size()) { debug("sending " << buffer.size() << " bytes"); - encodeLength(&buffer[1], buffer.size()-2); + encodeLength(); // hexdump("snd"); client->write(&buffer[0], buffer.size()); } diff --git a/src/TinyMqtt.h b/src/TinyMqtt.h index 27487dd..2a8d01a 100644 --- a/src/TinyMqtt.h +++ b/src/TinyMqtt.h @@ -101,8 +101,6 @@ 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]; } - uint16_t length() const { return buffer.size(); } - void complete(); void reset(); @@ -127,12 +125,12 @@ class MqttMessage void hexdump(const char* prefix=nullptr) const; private: - void encodeLength(char* msb, int length) const; + void encodeLength() const; mutable std::string buffer; // mutable -> sendTo() uint8_t vheader; uint16_t size; // bytes left to receive - State state; + mutable State state; // mutable -> encodeLength() }; class MqttBroker; @@ -172,7 +170,14 @@ class MqttClient /** Should be called in main loop() */ void loop(); void close(bool bSendDisconnect=true); - void setCallback(CallBack fun) {callback=fun; }; + void setCallback(CallBack fun) + { + callback=fun; + #ifdef TINY_MQTT_DEBUG + Serial << "Callback set to " << (long)fun << endl; + if (callback) callback(this, "test/topic", "value", 5); + #endif + }; // Publish from client to the world MqttError publish(const Topic&, const char* payload, size_t pay_length); @@ -214,6 +219,8 @@ class MqttClient static long counter; private: + + // event when tcp/ip link established (real or fake) static void onConnect(void * client_ptr, TcpClient*); #ifdef TCP_ASYNC static void onData(void* client_ptr, TcpClient*, void* data, size_t len); @@ -240,7 +247,7 @@ class MqttClient // (this is the case when MqttBroker isn't used except here) MqttBroker* parent=nullptr; // connection to local broker - TcpClient* client=nullptr; // connection to mqtt client or to remote broker + TcpClient* client=nullptr; // connection to remote broker std::set subscriptions; std::string clientId; CallBack callback = nullptr; From a6b3540cb8b5ddd648712efb9e3ba1a91c625d8d Mon Sep 17 00:00:00 2001 From: hsaturn Date: Fri, 17 Sep 2021 22:32:00 +0200 Subject: [PATCH 2/2] 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() {