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 9e164a7..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) @@ -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"); @@ -184,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; @@ -391,18 +397,19 @@ 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 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 @@ -614,6 +628,7 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa msg.add(topic); msg.add(payload, pay_length, false); msg.complete(); + if (parent) { return parent->publish(this, topic, msg); @@ -625,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; @@ -637,6 +652,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 } } @@ -665,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); @@ -714,7 +727,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 +738,33 @@ 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() { - do + if (state != Complete) { - uint8_t encoded(length & 0x7F); - length >>=7; - if (length) encoded |= 0x80; - *msb++ = encoded; - } while (length); + 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; + } }; -void MqttMessage::complete() -{ - encodeLength(&buffer[1], buffer.size()-2); - state = Complete; -} - -MqttError MqttMessage::sendTo(MqttClient* client) const +MqttError MqttMessage::sendTo(MqttClient* client) { 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..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,8 +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]; } - uint16_t length() const { return buffer.size(); } - void complete(); + void complete() { encodeLength(); } void reset(); @@ -118,18 +117,19 @@ 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(char* msb, int length) const; + void encodeLength(); - mutable std::string buffer; // mutable -> sendTo() + std::string buffer; uint8_t vheader; uint16_t size; // bytes left to receive State state; @@ -172,7 +172,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); @@ -210,10 +217,11 @@ class MqttClient Serial << endl; } - /** Count the number of messages that have been sent **/ - static long counter; + static long counter; // Number of processed messages 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); @@ -224,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; @@ -240,7 +248,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; @@ -284,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() {