Compare commits

..

12 Commits

Author SHA1 Message Date
hsaturn
5d07294fde Merge branch 'main' into timeout 2023-03-23 13:46:02 +01:00
hsaturn
088071d17f [readme] Words about retain 2023-03-23 13:44:48 +01:00
Francois BIOT
e41452edf0 [fixes] Timeout and broker to broker modifications 2023-03-23 13:42:09 +01:00
hsaturn
f8a2e35dd9 [readme] Words about retain 2023-03-22 20:37:28 +01:00
hsaturn
43dbea1f17 [test] Fix payload bug in test + moved huge paylod from network to here 2023-03-22 20:31:41 +01:00
hsaturn
02496bef73 [test] Test added for retain 2023-03-22 20:31:05 +01:00
hsaturn
e4ad27c805 [MqttClient] Added connect(IPAddress) 2023-03-22 20:29:52 +01:00
hsaturn
37fb46ec3b [TinyMqtt.cpp] Removed obsolete comment 2023-03-22 20:29:25 +01:00
hsaturn
245e74666e [examples] Little modification for RETAIN 2023-03-22 20:28:58 +01:00
hsaturn
294657f2ca [bump_version] Added a commit message for bump version 2023-03-22 09:43:04 +01:00
Francois BIOT
bf84e29831 retain is coming git status! 2023-03-22 00:29:55 +01:00
Francois BIOT
0c7c830a26 Release 0.9.19 2023-03-11 18:51:08 +01:00
12 changed files with 289 additions and 75 deletions

View File

@@ -24,6 +24,7 @@ TinyMqtt is a small, fast and capable Mqtt Broker and Client for Esp8266 / Esp32
## Features ## Features
- Supports retained messages (not activated by default)
- Async Wifi compatible (me-no-dev/ESPAsyncTCP@^1.2.2) - Async Wifi compatible (me-no-dev/ESPAsyncTCP@^1.2.2)
- 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. clients that had subscribed (payload ~15 bytes ESP8266). No topic lost.
@@ -64,6 +65,13 @@ TinyMqtt is a small, fast and capable Mqtt Broker and Client for Esp8266 / Esp32
- tinymqtt-test : This is a complex sketch with a terminal console - tinymqtt-test : This is a complex sketch with a terminal console
that allows to add clients publish, connect etc with interpreted commands. that allows to add clients publish, connect etc with interpreted commands.
## Retained messages
Qos 1 is not supported, but retained messages are. So a new subscription is able to send old messages.
This feature is disabled by default.
The default retain parameter of MqttBroker::MqttBroker takes an optional (0 by default) number of retained messages.
MqttBroker::retain(n) will also make the broker store n messages at max.
## Standalone mode (zeroconf) ## Standalone mode (zeroconf)
-> The zeroconf mode is not yet implemented -> The zeroconf mode is not yet implemented
zeroconf clients to connect to broker on local network. zeroconf clients to connect to broker on local network.

View File

@@ -8,7 +8,7 @@ else
fi fi
if [ "$1" == "" ]; then if [ "$1" == "" ]; then
echo echo
echo "Syntax: $0 [-d] {new_version}" echo "Syntax: $0 [-d] {new_version} [commit message]"
echo echo
echo " -d : dry run, generate json and update properties but do not run git commands" echo " -d : dry run, generate json and update properties but do not run git commands"
echo "" echo ""
@@ -34,7 +34,6 @@ else
depends=$(echo "$value" | sed "s/,/ /g") depends=$(echo "$value" | sed "s/,/ /g")
echo " Depends=$depends" echo " Depends=$depends"
fi fi
echo " " sed -i "s@#$name@$value@g" library.json
sed -i "s@#$name@$value@g" library.json sed -i "s@#$name@$value@g" library.json
done < library.properties done < library.properties
deps="" deps=""
@@ -47,10 +46,11 @@ else
sed -i "s@#dependencies@$deps@g" library.json sed -i "s@#dependencies@$deps@g" library.json
sed -i "s/'/\"/g" library.json sed -i "s/'/\"/g" library.json
if [ "$do" == "1" ]; then if [ "$do" == "1" ]; then
echo "Pushing all"
git tag $1 git tag $1
git add library.properties git add library.properties
git add library.json git add library.json
git commit -m "Release $1" git commit -m "Release $1 $2"
git push git push
git push --tags git push --tags
fi fi

View File

@@ -1,7 +1,9 @@
#include "TinyMqtt.h" // https://github.com/hsaturn/TinyMqtt #include "TinyMqtt.h" // https://github.com/hsaturn/TinyMqtt
#define PORT 1883 const uint16_t PORT 1883;
MqttBroker broker(PORT); const uint8_t RETAIN = 10; // Max retained messages
MqttBroker broker(PORT, RETAIN);
/** Basic Mqtt Broker /** Basic Mqtt Broker
* *
@@ -16,6 +18,8 @@ MqttBroker broker(PORT);
* Your ESP will become a MqttBroker. * Your ESP will become a MqttBroker.
* You can test it with any client such as mqtt-spy for example * You can test it with any client such as mqtt-spy for example
* *
* Messages are retained *only* if retain > 0
*
*/ */
const char* ssid = ""; const char* ssid = "";

View File

@@ -1,7 +1,7 @@
#include "TinyMqtt.h" // https://github.com/hsaturn/TinyMqtt #include "TinyMqtt.h" // https://github.com/hsaturn/TinyMqtt
#include "TinyStreaming.h" // https://github.com/hsaturn/TinyConsole #include "TinyStreaming.h" // https://github.com/hsaturn/TinyConsole
/** Simple Client (The simplest configuration) /** Simple Client (The simplest configuration, client only sends topics)
* *
* *
* +--------+ * +--------+

View File

@@ -8,7 +8,7 @@
}, },
"dependencies": "dependencies":
{ "hsaturn/TinyConsole" : "*" }, { "hsaturn/TinyConsole" : "*" },
"version": "0.9.18", "version": "0.9.19",
"exclude": "", "exclude": "",
"examples": "examples/*/*.ino", "examples": "examples/*/*.ino",
"frameworks": "arduino", "frameworks": "arduino",

View File

@@ -1,5 +1,5 @@
name=TinyMqtt name=TinyMqtt
version=0.9.18 version=0.9.19
author=Francois BIOT, HSaturn, <hsaturn@gmail.com> author=Francois BIOT, HSaturn, <hsaturn@gmail.com>
maintainer=Francois BIOT <hsaturn@gmail.com> maintainer=Francois BIOT <hsaturn@gmail.com>
sentence=A tiny broker and client library for MQTT messaging. sentence=A tiny broker and client library for MQTT messaging.

View File

@@ -107,6 +107,8 @@ class IndexedString
index = source.index; index = source.index;
} }
IndexedString(IndexedString&& i) : index(i.index) {}
IndexedString(const char* str, uint8_t len) IndexedString(const char* str, uint8_t len)
{ {
index=StringIndexer::strToIndex(str, len); index=StringIndexer::strToIndex(str, len);

View File

@@ -15,8 +15,10 @@ int TinyMqtt::debug=2;
std::map<MqttMessage::Type, int> MqttClient::counters; std::map<MqttMessage::Type, int> MqttClient::counters;
#endif #endif
MqttBroker::MqttBroker(uint16_t port) MqttBroker::MqttBroker(uint16_t port, uint8_t max_retain_size)
{ {
debug("New broker" << port);
retain_size = max_retain_size;
server = new TcpServer(port); server = new TcpServer(port);
#ifdef TINY_MQTT_ASYNC #ifdef TINY_MQTT_ASYNC
server->onClient(onClient, this); server->onClient(onClient, this);
@@ -142,6 +144,7 @@ void MqttBroker::connect(const string& host, uint16_t port)
if (remote_broker == nullptr) remote_broker = new MqttClient; if (remote_broker == nullptr) remote_broker = new MqttClient;
remote_broker->connect(host, port); remote_broker->connect(host, port);
remote_broker->local_broker = this; // Because connect removed the link remote_broker->local_broker = this; // Because connect removed the link
// TODO shouldn't we resubscribe to all client subscriptions ?
} }
void MqttBroker::removeClient(MqttClient* remove) void MqttBroker::removeClient(MqttClient* remove)
@@ -211,9 +214,19 @@ void MqttBroker::loop()
} }
} }
MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos) // Obvioulsy called when the broker is connected to another broker.
MqttError MqttBroker::subscribe(MqttClient* client, const Topic& topic, uint8_t qos)
{ {
debug("MqttBroker::subscribe"); debug("MqttBroker::subscribe to " << topic.str() << ", retained=" << retained.size() );
for(auto& [retained_topic, retain]: retained)
{
debug(" retained: " << retained_topic.str());
if (topic.matches(retained_topic))
{
debug(" -> sending");
client->publishIfSubscribed(retained_topic, retain.msg);
}
}
if (remote_broker && remote_broker->connected()) if (remote_broker && remote_broker->connected())
{ {
return remote_broker->subscribe(topic, qos); return remote_broker->subscribe(topic, qos);
@@ -221,10 +234,12 @@ MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
return MqttNowhereToSend; return MqttNowhereToSend;
} }
MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, MqttMessage& msg) const MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, MqttMessage& msg)
{ {
MqttError retval = MqttOk; MqttError retval = MqttOk;
retain(topic, msg);
debug("MqttBroker::publish"); debug("MqttBroker::publish");
int i=0; int i=0;
for(auto client: clients) for(auto client: clients)
@@ -294,25 +309,26 @@ void MqttClient::clientAlive(uint32_t more_seconds)
void MqttClient::loop() void MqttClient::loop()
{ {
if (keep_alive && (millis() >= alive)) if (keep_alive && (millis() >= alive - 5000))
{ {
if (local_broker) if (tcp_client && tcp_client->connected())
{
debug(red << "timeout client");
close();
debug(red << "closed");
}
else if (tcp_client && tcp_client->connected())
{ {
debug("pingreq"); debug("pingreq");
uint16_t pingreq = MqttMessage::Type::PingReq; static MqttMessage pingreq(MqttMessage::Type::PingReq);
tcp_client->write((const char*)(&pingreq), 2); pingreq.sendTo(this);
clientAlive(0); clientAlive(0);
// TODO when many MqttClient passes through a local broker // TODO when many MqttClient passes through a local broker
// there is no need to send one PingReq per instance. // there is no need to send one PingReq per instance.
} }
else if (local_broker)
{
debug(red << "timeout client");
close();
debug(red << "closed");
}
} }
#ifndef TINY_MQTT_ASYNC #ifndef TINY_MQTT_ASYNC
while(tcp_client && tcp_client->available()>0) while(tcp_client && tcp_client->available()>0)
{ {
@@ -389,15 +405,15 @@ MqttError MqttClient::subscribe(Topic topic, uint8_t qos)
debug("MqttClient::subsribe(" << topic.c_str() << ")"); debug("MqttClient::subsribe(" << topic.c_str() << ")");
MqttError ret = MqttOk; MqttError ret = MqttOk;
subscriptions.insert(topic); subscriptions.insert(topic);
if (local_broker==nullptr) // remote broker if (local_broker==nullptr) // connected to a remote broker
{ {
return sendTopic(topic, MqttMessage::Type::Subscribe, qos); return sendTopic(topic, MqttMessage::Type::Subscribe, qos);
} }
else else
{ {
return local_broker->subscribe(topic, qos); return local_broker->subscribe(this, topic, qos);
} }
return ret; return ret;
} }
@@ -568,7 +584,7 @@ void MqttClient::processMessage(MqttMessage* mesg)
} }
else else
qoss.push_back(qos); qoss.push_back(qos);
subscriptions.insert(topic); subscribe(topic);
} }
else else
{ {
@@ -618,13 +634,13 @@ void MqttClient::processMessage(MqttMessage* mesg)
#if TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
if (TinyMqtt::debug >= 2) if (TinyMqtt::debug >= 2)
{ {
Console << (isSubscribedTo(published) ? "not" : "") << " subscribed.\n"; Console << (isSubscribedTo(published) ? "not" : "") << " subscribed.\r\n";
Console << "has " << (callback ? "" : "no ") << " callback.\n"; Console << "has " << (callback ? "" : "no ") << " callback.\r\n";
} }
#endif #endif
if (callback and isSubscribedTo(published)) if (callback and isSubscribedTo(published))
{ {
callback(this, published, payload, len); // TODO send the real payload callback(this, published, payload, len);
} }
} }
else if (local_broker) // from outside to inside else if (local_broker) // from outside to inside
@@ -728,9 +744,9 @@ bool Topic::matches(const Topic& topic) const
// publish from local client // publish from local client
MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pay_length) MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pay_length, bool retain)
{ {
MqttMessage msg(MqttMessage::Publish); MqttMessage msg(MqttMessage::Publish, retain ? 1 : 0);
msg.add(topic); msg.add(topic);
msg.add(payload, pay_length, false); msg.add(payload, pay_length, false);
msg.complete(); msg.complete();
@@ -894,6 +910,35 @@ MqttError MqttMessage::sendTo(MqttClient* client)
return MqttOk; return MqttOk;
} }
void MqttBroker::retainDrop()
{
if (retained.size() >= retain_size)
{
std::map<Topic, Retain>::iterator oldest = retained.begin();
auto it = oldest;
while(++it != retained.end())
{
if (oldest->second.timestamp > it->second.timestamp)
oldest = it;
}
retained.erase(oldest);
}
}
void MqttBroker::retain(const Topic& topic, const MqttMessage& msg)
{
debug("MqttBroker::retain msg_type=" << _HEX(msg.type()));
if (retain_size==0 or msg.type() != MqttMessage::Publish) return;
if (msg.flags() & 1) // flag RETAIN
{
debug(" retaining " << topic.str());
if (retained.find(topic) == retained.end()) retainDrop();
// FIXME if payload size == 0 remove message from retained
Retain r(micros(), msg);
retained.insert({ topic, std::move(r)});
}
}
void MqttMessage::hexdump(const char* prefix) const void MqttMessage::hexdump(const char* prefix) const
{ {
(void)prefix; (void)prefix;

View File

@@ -121,7 +121,10 @@ class MqttMessage
return (*bun << 8) | bun[1]; } return (*bun << 8) | bun[1]; }
MqttMessage() { reset(); } MqttMessage() { reset(); }
MqttMessage(Type t, uint8_t bits_d3_d0=0) { create(t); buffer[0] |= bits_d3_d0; } MqttMessage(Type t, uint8_t bits_d3_d0=0) { create(t); buffer[0] |= (bits_d3_d0 & 0xF); }
MqttMessage(const MqttMessage& m)
: buffer(m.buffer), vheader(m.vheader), size(m.size), state(m.state) {}
void incoming(char byte); void incoming(char byte);
void add(char byte) { incoming(byte); } void add(char byte) { incoming(byte); }
void add(const char* p, size_t len, bool addLength=true ); void add(const char* p, size_t len, bool addLength=true );
@@ -156,6 +159,15 @@ class MqttMessage
MqttError sendTo(MqttClient*); MqttError sendTo(MqttClient*);
void hexdump(const char* prefix=nullptr) const; void hexdump(const char* prefix=nullptr) const;
MqttMessage& operator = (MqttMessage&& m)
{
buffer = std::move(m.buffer);
vheader = m.vheader;
size = m.size;
state = m.state;
return *this;
}
private: private:
void encodeLength(); void encodeLength();
@@ -198,6 +210,8 @@ class MqttClient
void connect(MqttBroker* local_broker); void connect(MqttBroker* local_broker);
void connect(string broker, uint16_t port, uint16_t keep_alive = 10); void connect(string broker, uint16_t port, uint16_t keep_alive = 10);
void connect(const IPAddress& ip, uint16_t port, uint16_t keep_alive = 10)
{ connect(ip.toString().c_str(), port, keep_alive); }
// TODO it seems that connected returns true in tcp mode even if // TODO it seems that connected returns true in tcp mode even if
// no negociation occurred // no negociation occurred
@@ -228,11 +242,11 @@ class MqttClient
}; };
// Publish from client to the world // Publish from client to the world
MqttError publish(const Topic&, const char* payload, size_t pay_length); MqttError publish(const Topic&, const char* payload, size_t pay_length, bool retain=false);
MqttError publish(const Topic& t, const char* payload) { return publish(t, payload, strlen(payload)); } MqttError publish(const Topic& t, const char* payload, bool retain=false) { return publish(t, payload, strlen(payload), retain); }
MqttError publish(const Topic& t, const String& s) { return publish(t, s.c_str(), s.length()); } MqttError publish(const Topic& t, const String& s, bool retain=false) { return publish(t, s.c_str(), s.length(), retain); }
MqttError publish(const Topic& t, const string& s) { return publish(t,s.c_str(),s.length());} MqttError publish(const Topic& t, const string& s, bool retain=false) { return publish(t,s.c_str(),s.length(), retain);}
MqttError publish(const Topic& t) { return publish(t, nullptr, 0);}; MqttError publish(const Topic& t, bool retain=false) { return publish(t, nullptr, 0, retain);};
MqttError subscribe(Topic topic, uint8_t qos=0); MqttError subscribe(Topic topic, uint8_t qos=0);
MqttError unsubscribe(Topic topic); MqttError unsubscribe(Topic topic);
@@ -317,15 +331,9 @@ class MqttClient
class MqttBroker class MqttBroker
{ {
enum __attribute__((packed)) State
{
Disconnected, // Also the initial state
Connecting, // connect and sends a fake publish to avoid circular cnx
Connected, // this->broker is connected and circular cnx avoided
};
public: public:
// TODO limit max number of clients // TODO limit max number of clients
MqttBroker(uint16_t port); MqttBroker(uint16_t port, uint8_t retain_size=0);
~MqttBroker(); ~MqttBroker();
void begin() { server->begin(); } void begin() { server->begin(); }
@@ -334,9 +342,10 @@ class MqttBroker
/** Connect the broker to a parent broker */ /** Connect the broker to a parent broker */
void connect(const string& host, uint16_t port=1883); void connect(const string& host, uint16_t port=1883);
/** returns true if connected to another broker */ /** returns true if connected to another broker */
bool connected() const { return state == Connected; } bool connected() const { return remote_broker ? remote_broker->connected() : false; }
size_t clientsCount() const { return clients.size(); } size_t clientsCount() const { return clients.size(); }
void retain(uint8_t size) { retain_size = size; }
void dump(string indent="") void dump(string indent="")
{ {
@@ -356,10 +365,9 @@ class MqttBroker
bool checkPassword(const char* password, uint8_t len) const bool checkPassword(const char* password, uint8_t len) const
{ return compareString(auth_password, password, len); } { return compareString(auth_password, password, len); }
MqttError publish(const MqttClient* source, const Topic& topic, MqttMessage& msg);
MqttError publish(const MqttClient* source, const Topic& topic, MqttMessage& msg) const; MqttError subscribe(MqttClient*, const Topic& topic, uint8_t qos);
MqttError subscribe(const Topic& topic, uint8_t qos);
// For clients that are added not by the broker itself (local clients) // For clients that are added not by the broker itself (local clients)
void addClient(MqttClient* client); void addClient(MqttClient* client);
@@ -375,5 +383,25 @@ class MqttBroker
const char* auth_password = "guest"; const char* auth_password = "guest";
MqttClient* remote_broker = nullptr; MqttClient* remote_broker = nullptr;
State state = Disconnected; void retain(const Topic& topic, const MqttMessage& msg);
void retainDrop();
struct Retain
{
Retain(unsigned long ts, const MqttMessage& m) : timestamp(ts), msg(m) {}
Retain(Retain&& r) : timestamp(r.timestamp), msg(std::move(r.msg)) {}
Retain& operator=(Retain&& r)
{
timestamp = r.timestamp;
msg = std::move(r.msg);
return *this;
}
unsigned long timestamp;
MqttMessage msg;
};
std::map<Topic, Retain> retained;
uint8_t retain_size;
}; };

View File

@@ -1,7 +1,7 @@
# GCC # GCC
# CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics # CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics
EXTRA_CXXFLAGS=-g3 -O0 EXTRA_CXXFLAGS=-g3 -O0 -std=c++17
CXXFLAGS=-D_GNU_SOURCE -Werror=return-type -std=gnu++17 -Wall -g3 -O0 CXXFLAGS=-D_GNU_SOURCE -Werror=return-type -std=gnu++17 -Wall -g3 -O0

View File

@@ -18,14 +18,14 @@ MqttBroker broker(1883);
std::map<string, std::map<Topic, int>> published; // map[client_id] => map[topic] = count std::map<string, std::map<Topic, int>> published; // map[client_id] => map[topic] = count
const char* lastPayload; std::string lastPayload;
size_t lastLength; size_t lastLength;
void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length) void onPublish(const MqttClient* srce, const Topic& topic, const char* payload, size_t length)
{ {
if (srce) if (srce)
published[srce->id()][topic]++; published[srce->id()][topic]++;
lastPayload = payload; lastPayload = std::string(payload, length);
lastLength = length; lastLength = length;
} }
@@ -51,6 +51,7 @@ test(local_client_should_unregister_when_destroyed)
test(local_client_do_not_disconnect_after_publishing_and_long_inactivity) test(local_client_do_not_disconnect_after_publishing_and_long_inactivity)
{ {
published.clear();
EpoxyTest::set_millis(0); EpoxyTest::set_millis(0);
MqttBroker broker(1883); MqttBroker broker(1883);
MqttClient client(&broker, "client"); MqttClient client(&broker, "client");
@@ -121,6 +122,25 @@ test(local_publish_should_be_dispatched)
assertEqual(published[""]["a/c"], 2); assertEqual(published[""]["a/c"], 2);
} }
test(hudge_payload)
{
published.clear();
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);
assertEqual(broker.clientsCount(), (size_t)1);
subscriber.setCallback(onPublish);
subscriber.subscribe("a/b"); // Note -> this does not send any byte .... (nowhere to send)
MqttClient publisher(&broker);
publisher.publish("a/b", payload); // This publish is received
// onPublish should have filled lastPayload and lastLength
assertEqual(payload, lastPayload.c_str());
assertEqual(lastLength, strlen(payload));
assertEqual(strncmp(payload, lastPayload.c_str(), lastLength), 0);
}
test(local_publish_should_be_dispatched_to_local_clients) test(local_publish_should_be_dispatched_to_local_clients)
{ {
published.clear(); published.clear();

View File

@@ -85,7 +85,7 @@ std::map<string, std::map<Topic, int>> published; // map[client_id] => map[t
char* lastPayload = nullptr; char* lastPayload = nullptr;
size_t lastLength; size_t lastLength;
void start_servers(int n, bool early_accept = true) void start_many_wifi_esp(int n, bool early_accept = true)
{ {
ESP8266WiFiClass::resetInstances(); ESP8266WiFiClass::resetInstances();
ESP8266WiFiClass::earlyAccept = early_accept; ESP8266WiFiClass::earlyAccept = early_accept;
@@ -107,7 +107,7 @@ void onPublish(const MqttClient* srce, const Topic& topic, const char* payload,
lastLength = length; lastLength = length;
} }
test(network_single_broker_begin) test(single_broker_begin)
{ {
assertEqual(WiFi.status(), WL_CONNECTED); assertEqual(WiFi.status(), WL_CONNECTED);
@@ -119,7 +119,7 @@ test(network_single_broker_begin)
test(suback) test(suback)
{ {
start_servers(2, true); start_many_wifi_esp(2, true);
assertEqual(WiFi.status(), WL_CONNECTED); assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883); MqttBroker broker(1883);
@@ -144,10 +144,10 @@ test(suback)
assertEqual(MqttClient::counters[MqttMessage::Type::SubAck], 1); assertEqual(MqttClient::counters[MqttMessage::Type::SubAck], 1);
} }
test(network_client_keep_alive_high) test(client_keep_alive_high)
{ {
const uint32_t keep_alive=1000; const uint32_t keep_alive=1000;
start_servers(2, true); start_many_wifi_esp(2, true);
assertEqual(WiFi.status(), WL_CONNECTED); assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883); MqttBroker broker(1883);
@@ -179,9 +179,116 @@ test(network_client_keep_alive_high)
} }
test(network_client_to_broker_connexion) test(retained_message)
{ {
start_servers(2, true); published.clear();
start_many_wifi_esp(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
broker.retain(10);
IPAddress broker_ip = WiFi.localIP();
MqttClient local_client(&broker);
// Send a retained message
// No remote client connected
local_client.publish("topic", "retained", true);
for(int i=0; i<2; i++)
{
broker.loop();
local_client.loop();
};
// No connect a client from 2nd Esp
ESP8266WiFiClass::selectInstance(2);
MqttClient remote_client;
remote_client.connect(broker_ip, 1883);
remote_client.setCallback(onPublish);
assertTrue(remote_client.connected());
for(int i=0; i<4; i++) { broker.loop(); local_client.loop(); remote_client.loop(); };
assertEqual(broker.clientsCount(), (size_t) 2);
// Should not have received anything yet
assertEqual(published.size(), (size_t)0);
// Now, remote client subscribes to topic
remote_client.subscribe("topic");
for(int i=0; i<4; i++) { broker.loop(); local_client.loop(); remote_client.loop(); };
// Check that the retained message is published
assertEqual(published.size(), (size_t)1);
// FIXME we should check that
// 1 - Retained message has the retain flag set
// 2 - Published retained messages that are send normally have their retain flag off
// The next part of this test does not pass yet (due to remote_client.close()
// that does not work well.
return;
// Now remove the retained message
remote_client.close();
for(int i=0; i<4; i++) { broker.loop(); local_client.loop(); remote_client.loop(); };
assertFalse(remote_client.connected());
assertEqual(broker.clientsCount(), (size_t) 1);
// Disconnect / reconnect the remote clien that should receive again the message
remote_client.connect(broker_ip, 1883);
remote_client.subscribe("topic");
assertTrue(remote_client.connected());
for(int i=0; i<4; i++) { broker.loop(); local_client.loop(); remote_client.loop(); };
assertEqual(broker.clientsCount(), (size_t) 2);
assertEqual(published.size(), (size_t)2);
// Remove the retained message now
local_client.publish("topic", "", true);
assertEqual(published.size(), (size_t)1);
// And reconnect the remote client
remote_client.connect(broker_ip, 1883);
for(int i=0; i<4; i++) { broker.loop(); local_client.loop(); remote_client.loop(); };
assertEqual(broker.clientsCount(), (size_t) 2);
// check that the message was received
assertEqual(published.size(), (size_t)2);
}
test(remote_client_disconnect_reconnect)
{
published.clear();
start_many_wifi_esp(2, true);
assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883);
broker.begin();
IPAddress broker_ip = WiFi.localIP();
ESP8266WiFiClass::selectInstance(2);
MqttClient client;
client.connect(broker_ip, 1883);
for(int i=0; i<4; i++) { broker.loop(); client.loop(); };
assertEqual(broker.clientsCount(), (size_t) 1);
// Disconnect the client
client.close();
for(int i=0; i<4; i++) { broker.loop(); client.loop(); };
assertEqual(broker.clientsCount(), (size_t) 0);
// Reconnect the client
client.connect(broker_ip, 1883);
for(int i=0; i<4; i++) { broker.loop(); client.loop(); };
assertEqual(broker.clientsCount(), (size_t) 1);
}
test(client_to_broker_connexion)
{
start_many_wifi_esp(2, true);
assertEqual(WiFi.status(), WL_CONNECTED); assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883); MqttBroker broker(1883);
@@ -197,9 +304,9 @@ test(network_client_to_broker_connexion)
assertTrue(client.connected()); assertTrue(client.connected());
} }
test(network_one_client_one_broker_publish_and_subscribe_through_network) test(one_client_one_broker_publish_and_subscribe)
{ {
start_servers(2, true); start_many_wifi_esp(2, true);
published.clear(); published.clear();
assertEqual(WiFi.status(), WL_CONNECTED); assertEqual(WiFi.status(), WL_CONNECTED);
@@ -228,9 +335,9 @@ test(network_one_client_one_broker_publish_and_subscribe_through_network)
assertEqual((int)lastLength, (int)2); // sizeof(ab) assertEqual((int)lastLength, (int)2); // sizeof(ab)
} }
test(network_one_client_one_broker_hudge_publish_and_subscribe_through_network) test(one_client_one_broker_hudge_payload)
{ {
start_servers(2, true); start_many_wifi_esp(2, true);
published.clear(); published.clear();
assertEqual(WiFi.status(), WL_CONNECTED); assertEqual(WiFi.status(), WL_CONNECTED);
@@ -247,8 +354,8 @@ test(network_one_client_one_broker_hudge_publish_and_subscribe_through_network)
std::string sent; std::string sent;
for(int i=0; i<200; i++) for(int i=0; i<400; i++)
sent += char('0'+i%10); sent += char('a'+i%26);
client.setCallback(onPublish); client.setCallback(onPublish);
client.subscribe("a/b"); client.subscribe("a/b");
@@ -264,7 +371,7 @@ test(network_one_client_one_broker_hudge_publish_and_subscribe_through_network)
assertEqual((unsigned int)lastLength, (unsigned int)sent.size()); assertEqual((unsigned int)lastLength, (unsigned int)sent.size());
} }
test(network_client_should_unregister_when_destroyed) test(client_should_unregister_when_destroyed)
{ {
assertEqual(broker.clientsCount(), (size_t)0); assertEqual(broker.clientsCount(), (size_t)0);
{ {
@@ -278,7 +385,7 @@ test(network_client_should_unregister_when_destroyed)
// THESE TESTS ARE IN LOCAL MODE // THESE TESTS ARE IN LOCAL MODE
// WE HAVE TO CONVERT THEM TO WIFI MODE (pass through virtual TCP link) // WE HAVE TO CONVERT THEM TO WIFI MODE (pass through virtual TCP link)
test(network_connect) test(connect)
{ {
assertEqual(broker.clientsCount(), (size_t)0); assertEqual(broker.clientsCount(), (size_t)0);
@@ -287,7 +394,7 @@ test(network_connect)
assertEqual(broker.clientsCount(), (size_t)1); assertEqual(broker.clientsCount(), (size_t)1);
} }
test(network_publish_should_be_dispatched) test(publish_should_be_dispatched)
{ {
published.clear(); published.clear();
assertEqual(broker.clientsCount(), (size_t)0); assertEqual(broker.clientsCount(), (size_t)0);
@@ -307,7 +414,7 @@ test(network_publish_should_be_dispatched)
assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/c"], 2); assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/c"], 2);
} }
test(network_publish_should_be_dispatched_to_clients) test(publish_should_be_dispatched_to_clients)
{ {
published.clear(); published.clear();
assertEqual(broker.clientsCount(), (size_t)0); assertEqual(broker.clientsCount(), (size_t)0);
@@ -332,7 +439,7 @@ test(network_publish_should_be_dispatched_to_clients)
assertEqual(published["B"]["a/c"], 0); assertEqual(published["B"]["a/c"], 0);
} }
test(network_unsubscribe) test(unsubscribe)
{ {
published.clear(); published.clear();
assertEqual(broker.clientsCount(), (size_t)0); assertEqual(broker.clientsCount(), (size_t)0);
@@ -352,7 +459,7 @@ test(network_unsubscribe)
assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/b"], 1); // Only one publish has been received assertEqual(published[TINY_MQTT_DEFAULT_CLIENT_ID]["a/b"], 1); // Only one publish has been received
} }
test(network_nocallback_when_destroyed) test(nocallback_when_destroyed)
{ {
published.clear(); published.clear();
assertEqual(broker.clientsCount(), (size_t)0); assertEqual(broker.clientsCount(), (size_t)0);
@@ -371,7 +478,7 @@ test(network_nocallback_when_destroyed)
assertEqual(published.size(), (size_t)1); // Only one publish has been received assertEqual(published.size(), (size_t)1); // Only one publish has been received
} }
test(network_small_payload) test(small_payload)
{ {
published.clear(); published.clear();
@@ -389,7 +496,7 @@ test(network_small_payload)
assertEqual(lastLength, (size_t)4); assertEqual(lastLength, (size_t)4);
} }
test(network_hudge_payload) test(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 !"; 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 !";
@@ -436,7 +543,7 @@ test(connack)
} }
); );
start_servers(2, true); start_many_wifi_esp(2, true);
assertEqual(WiFi.status(), WL_CONNECTED); assertEqual(WiFi.status(), WL_CONNECTED);
MqttBroker broker(1883); MqttBroker broker(1883);
@@ -469,7 +576,7 @@ void setup() {
while(!Serial); while(!Serial);
*/ */
Serial.println("=============[ FAKE NETWORK TinyMqtt TESTS ]========================"); Serial.println("=============[ NETWORK TinyMqtt TESTS ]========================");
WiFi.mode(WIFI_STA); WiFi.mode(WIFI_STA);
WiFi.begin("network", "password"); WiFi.begin("network", "password");