Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a0435b2cfb | ||
|
|
bda041417d | ||
|
|
baffda8a6d | ||
|
|
09e3a3e45f | ||
|
|
f17ece3376 | ||
|
|
0db07df27b | ||
|
|
292592c3dd | ||
|
|
1f267c135b |
2
.github/workflows/aunit.yml
vendored
2
.github/workflows/aunit.yml
vendored
@@ -18,7 +18,7 @@ jobs:
|
|||||||
run: |
|
run: |
|
||||||
cd ..
|
cd ..
|
||||||
git clone https://github.com/hsaturn/TinyConsole
|
git clone https://github.com/hsaturn/TinyConsole
|
||||||
git clone https://github.com/bxparks/EpoxyDuino
|
git clone https://github.com/hsaturn/EpoxyDuino
|
||||||
git clone https://github.com/bxparks/AceRoutine
|
git clone https://github.com/bxparks/AceRoutine
|
||||||
git clone https://github.com/bxparks/AUnit
|
git clone https://github.com/bxparks/AUnit
|
||||||
git clone https://github.com/bxparks/AceCommon
|
git clone https://github.com/bxparks/AceCommon
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
name=TinyMqtt
|
name=TinyMqtt
|
||||||
version=0.9.11
|
version=0.9.12
|
||||||
author=Francois BIOT, HSaturn, <hsaturn@gmail.com>
|
author=Francois BIOT, HSaturn, <hsaturn@gmail.com>
|
||||||
maintainer=Francois BIOT, HSaturn, <hsaturn@gmail.com>
|
maintainer=Francois BIOT, HSaturn, <hsaturn@gmail.com>
|
||||||
sentence=A tiny broker and client library for MQTT messaging.
|
sentence=A tiny broker and client library for MQTT messaging.
|
||||||
|
|||||||
154
src/TinyMqtt.cpp
154
src/TinyMqtt.cpp
@@ -17,16 +17,25 @@ int TinyMqtt::debug=2;
|
|||||||
|
|
||||||
MqttBroker::MqttBroker(uint16_t port)
|
MqttBroker::MqttBroker(uint16_t port)
|
||||||
{
|
{
|
||||||
server = std::unique_ptr<TcpServer>(new TcpServer(port));
|
server = new TcpServer(port);
|
||||||
#ifdef TINY_MQTT_ASYNC
|
#ifdef TINY_MQTT_ASYNC
|
||||||
server->onClient(onClient, this);
|
server->onClient(onClient, this);
|
||||||
#endif
|
#endif
|
||||||
}
|
}
|
||||||
|
|
||||||
|
MqttBroker::~MqttBroker()
|
||||||
|
{
|
||||||
|
while(clients.size())
|
||||||
|
{
|
||||||
|
delete clients[0];
|
||||||
|
}
|
||||||
|
delete server;
|
||||||
|
}
|
||||||
|
|
||||||
// private constructor used by broker only
|
// private constructor used by broker only
|
||||||
MqttClient::MqttClient(MqttBroker* local_broker, TcpClient* new_client)
|
MqttClient::MqttClient(MqttBroker* local_broker, TcpClient* new_client)
|
||||||
|
: local_broker(local_broker)
|
||||||
{
|
{
|
||||||
connect(local_broker);
|
|
||||||
debug("MqttClient private with broker");
|
debug("MqttClient private with broker");
|
||||||
#ifdef TINY_MQTT_ASYNC
|
#ifdef TINY_MQTT_ASYNC
|
||||||
tcp_client = new_client;
|
tcp_client = new_client;
|
||||||
@@ -34,15 +43,20 @@ MqttClient::MqttClient(MqttBroker* local_broker, TcpClient* new_client)
|
|||||||
// client->onConnect() TODO
|
// client->onConnect() TODO
|
||||||
// client->onDisconnect() TODO
|
// client->onDisconnect() TODO
|
||||||
#else
|
#else
|
||||||
tcp_client.reset(new WiFiClient(*new_client));
|
tcp_client = new WiFiClient(*new_client);
|
||||||
|
#endif
|
||||||
|
#ifdef EPOXY_DUINO
|
||||||
|
alive = millis()+500000;
|
||||||
|
#else
|
||||||
|
alive = millis()+5000; // TODO MAGIC client expires after 5s if no CONNECT msg
|
||||||
#endif
|
#endif
|
||||||
alive = millis()+5000;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
MqttClient::MqttClient(MqttBroker* local_broker, const std::string& id)
|
MqttClient::MqttClient(MqttBroker* local_broker, const std::string& id)
|
||||||
: local_broker(local_broker), clientId(id)
|
: local_broker(local_broker), clientId(id)
|
||||||
{
|
{
|
||||||
alive = 0;
|
alive = 0;
|
||||||
|
keep_alive = 0;
|
||||||
|
|
||||||
if (local_broker) local_broker->addClient(this);
|
if (local_broker) local_broker->addClient(this);
|
||||||
}
|
}
|
||||||
@@ -50,13 +64,14 @@ MqttClient::MqttClient(MqttBroker* local_broker, const std::string& id)
|
|||||||
MqttClient::~MqttClient()
|
MqttClient::~MqttClient()
|
||||||
{
|
{
|
||||||
close();
|
close();
|
||||||
|
delete tcp_client;
|
||||||
debug("*** MqttClient delete()");
|
debug("*** MqttClient delete()");
|
||||||
}
|
}
|
||||||
|
|
||||||
void MqttClient::close(bool bSendDisconnect)
|
void MqttClient::close(bool bSendDisconnect)
|
||||||
{
|
{
|
||||||
debug("close " << id().c_str());
|
debug("close " << id().c_str());
|
||||||
mqtt_flags &= ~FlagConnected;
|
mqtt_connected = false;
|
||||||
if (tcp_client) // connected to a remote broker
|
if (tcp_client) // connected to a remote broker
|
||||||
{
|
{
|
||||||
if (bSendDisconnect and tcp_client->connected())
|
if (bSendDisconnect and tcp_client->connected())
|
||||||
@@ -78,10 +93,8 @@ void MqttClient::close(bool bSendDisconnect)
|
|||||||
void MqttClient::connect(MqttBroker* local)
|
void MqttClient::connect(MqttBroker* local)
|
||||||
{
|
{
|
||||||
debug("MqttClient::connect_local");
|
debug("MqttClient::connect_local");
|
||||||
alive = 0;
|
|
||||||
close();
|
close();
|
||||||
local_broker = local;
|
local_broker = local;
|
||||||
clientAlive();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
|
void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
|
||||||
@@ -89,7 +102,8 @@ void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
|
|||||||
debug("MqttClient::connect_to_host " << broker << ':' << port);
|
debug("MqttClient::connect_to_host " << broker << ':' << port);
|
||||||
keep_alive = ka;
|
keep_alive = ka;
|
||||||
close();
|
close();
|
||||||
tcp_client.reset(new TcpClient);
|
if (tcp_client) delete tcp_client;
|
||||||
|
tcp_client = new TcpClient;
|
||||||
|
|
||||||
#ifdef TINY_MQTT_ASYNC
|
#ifdef TINY_MQTT_ASYNC
|
||||||
tcp_client->onData(onData, this);
|
tcp_client->onData(onData, this);
|
||||||
@@ -99,7 +113,7 @@ void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
|
|||||||
if (tcp_client->connect(broker.c_str(), port))
|
if (tcp_client->connect(broker.c_str(), port))
|
||||||
{
|
{
|
||||||
debug("link established");
|
debug("link established");
|
||||||
onConnect(this, tcp_client.get());
|
onConnect(this, tcp_client);
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
@@ -108,10 +122,10 @@ void MqttClient::connect(std::string broker, uint16_t port, uint16_t ka)
|
|||||||
#endif
|
#endif
|
||||||
}
|
}
|
||||||
|
|
||||||
void MqttBroker::addClient(TcpClient* client)
|
void MqttBroker::addClient(MqttClient* client)
|
||||||
{
|
{
|
||||||
debug("MqttBroker::addClient");
|
debug("MqttBroker::addClient");
|
||||||
clients.insert(std::unique_ptr<MqttClient>(new MqttClient(this, client)));
|
clients.push_back(client);
|
||||||
}
|
}
|
||||||
|
|
||||||
void MqttBroker::connect(const std::string& host, uint16_t port)
|
void MqttBroker::connect(const std::string& host, uint16_t port)
|
||||||
@@ -124,7 +138,24 @@ void MqttBroker::connect(const std::string& host, uint16_t port)
|
|||||||
|
|
||||||
void MqttBroker::removeClient(MqttClient* remove)
|
void MqttBroker::removeClient(MqttClient* remove)
|
||||||
{
|
{
|
||||||
local_clients.erase(remove);
|
debug("removeClient");
|
||||||
|
for(auto it=clients.begin(); it!=clients.end(); it++)
|
||||||
|
{
|
||||||
|
auto client=*it;
|
||||||
|
if (client==remove)
|
||||||
|
{
|
||||||
|
// TODO if this broker is connected to an external broker
|
||||||
|
// we have to unsubscribe remove's topics.
|
||||||
|
// (but doing this, check that other clients are not subscribed...)
|
||||||
|
// Unless -> we could receive useless messages
|
||||||
|
// -> we are using (memory) one IndexedString plus its string for nothing.
|
||||||
|
debug("Remove " << clients.size());
|
||||||
|
clients.erase(it);
|
||||||
|
debug("Client removed " << clients.size());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
debug(red << "Error cannot remove client"); // TODO should not occur
|
||||||
}
|
}
|
||||||
|
|
||||||
void MqttBroker::onClient(void* broker_ptr, TcpClient* client)
|
void MqttBroker::onClient(void* broker_ptr, TcpClient* client)
|
||||||
@@ -132,7 +163,7 @@ void MqttBroker::onClient(void* broker_ptr, TcpClient* client)
|
|||||||
debug("MqttBroker::onClient");
|
debug("MqttBroker::onClient");
|
||||||
MqttBroker* broker = static_cast<MqttBroker*>(broker_ptr);
|
MqttBroker* broker = static_cast<MqttBroker*>(broker_ptr);
|
||||||
|
|
||||||
broker->addClient(client);
|
broker->addClient(new MqttClient(broker, client));
|
||||||
debug("New client");
|
debug("New client");
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -153,18 +184,21 @@ void MqttBroker::loop()
|
|||||||
remote_broker->loop();
|
remote_broker->loop();
|
||||||
}
|
}
|
||||||
|
|
||||||
// 200 bytes shorter than for(auto& client: clients) !
|
for(size_t i=0; i<clients.size(); i++)
|
||||||
for(auto it=clients.begin(); it!=clients.end(); it++)
|
|
||||||
{
|
{
|
||||||
it->get()->loop();
|
MqttClient* client = clients[i];
|
||||||
if (not it->get()->connected())
|
if (client->connected())
|
||||||
{
|
{
|
||||||
clients.erase(it);
|
client->loop();
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
debug("Client " << client->id().c_str() << " Disconnected, local_broker=" << (dbg_ptr)client->local_broker);
|
||||||
|
// Note: deleting a client not added by the broker itself will probably crash later.
|
||||||
|
delete client;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for(const auto& client: local_clients)
|
|
||||||
client->loop();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
|
MqttError MqttBroker::subscribe(const Topic& topic, uint8_t qos)
|
||||||
@@ -182,13 +216,14 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, Mqtt
|
|||||||
MqttError retval = MqttOk;
|
MqttError retval = MqttOk;
|
||||||
|
|
||||||
debug("MqttBroker::publish");
|
debug("MqttBroker::publish");
|
||||||
int clt_num = 0;
|
int i=0;
|
||||||
for(auto& client: clients)
|
for(auto client: clients)
|
||||||
{
|
{
|
||||||
debug (" broker:" << (remote_broker && remote_broker->connected() ? "linked" : "alone")
|
i++;
|
||||||
<< " srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << ++clt_num
|
#if TINY_MQTT_DEBUG
|
||||||
<< ", local=" << client->isLocal() << ", con=" << client->connected());
|
Console << __LINE__ << " broker:" << (remote_broker && remote_broker->connected() ? "linked" : "alone") <<
|
||||||
|
" srce=" << (source->isLocal() ? "loc" : "rem") << " clt#" << i << ", local=" << client->isLocal() << ", con=" << client->connected() << endl;
|
||||||
|
#endif
|
||||||
bool doit = false;
|
bool doit = false;
|
||||||
if (remote_broker && remote_broker->connected()) // this (MqttBroker) is connected (to a external broker)
|
if (remote_broker && remote_broker->connected()) // this (MqttBroker) is connected (to a external broker)
|
||||||
{
|
{
|
||||||
@@ -206,8 +241,9 @@ MqttError MqttBroker::publish(const MqttClient* source, const Topic& topic, Mqtt
|
|||||||
{
|
{
|
||||||
doit = true;
|
doit = true;
|
||||||
}
|
}
|
||||||
|
#if TINY_MQTT_DEBUG
|
||||||
debug(" doit=" << doit << ' ');
|
Console << ", doit=" << doit << ' ';
|
||||||
|
#endif
|
||||||
|
|
||||||
if (doit) retval = client->publishIfSubscribed(topic, msg);
|
if (doit) retval = client->publishIfSubscribed(topic, msg);
|
||||||
debug("");
|
debug("");
|
||||||
@@ -231,12 +267,16 @@ void MqttMessage::getString(const char* &buff, uint16_t& len)
|
|||||||
buff+=2;
|
buff+=2;
|
||||||
}
|
}
|
||||||
|
|
||||||
void MqttClient::clientAlive()
|
void MqttClient::clientAlive(uint32_t more_seconds)
|
||||||
{
|
{
|
||||||
debug("MqttClient::clientAlive");
|
debug("MqttClient::clientAlive");
|
||||||
if (keep_alive)
|
if (keep_alive)
|
||||||
{
|
{
|
||||||
alive=millis()+1000*(keep_alive+(local_broker ? TINY_MQTT_CLIENT_ALIVE_TOLERANCE : 0));
|
#ifdef EPOXY_DUINO
|
||||||
|
alive=millis()+500000+0*more_seconds;
|
||||||
|
#else
|
||||||
|
alive=millis()+1000*(keep_alive+more_seconds);
|
||||||
|
#endif
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
alive=0;
|
alive=0;
|
||||||
@@ -244,11 +284,11 @@ void MqttClient::clientAlive()
|
|||||||
|
|
||||||
void MqttClient::loop()
|
void MqttClient::loop()
|
||||||
{
|
{
|
||||||
if (alive && (millis() >= alive))
|
if (keep_alive && (millis() >= alive))
|
||||||
{
|
{
|
||||||
if (local_broker)
|
if (local_broker)
|
||||||
{
|
{
|
||||||
Serial << "timeout client " << clientId << endl;
|
debug(red << "timeout client");
|
||||||
close();
|
close();
|
||||||
debug(red << "closed");
|
debug(red << "closed");
|
||||||
}
|
}
|
||||||
@@ -256,9 +296,8 @@ void MqttClient::loop()
|
|||||||
{
|
{
|
||||||
debug("pingreq");
|
debug("pingreq");
|
||||||
uint16_t pingreq = MqttMessage::Type::PingReq;
|
uint16_t pingreq = MqttMessage::Type::PingReq;
|
||||||
|
|
||||||
tcp_client->write((const char*)(&pingreq), 2);
|
tcp_client->write((const char*)(&pingreq), 2);
|
||||||
clientAlive();
|
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.
|
||||||
@@ -294,7 +333,7 @@ void MqttClient::onConnect(void *mqttclient_ptr, TcpClient*)
|
|||||||
msg.reset();
|
msg.reset();
|
||||||
debug("cnx: mqtt sent " << (dbg_ptr)mqtt->local_broker);
|
debug("cnx: mqtt sent " << (dbg_ptr)mqtt->local_broker);
|
||||||
|
|
||||||
mqtt->clientAlive();
|
mqtt->clientAlive(0);
|
||||||
}
|
}
|
||||||
|
|
||||||
#ifdef TINY_MQTT_ASYNC
|
#ifdef TINY_MQTT_ASYNC
|
||||||
@@ -326,7 +365,7 @@ void MqttClient::resubscribe()
|
|||||||
msg.add(0);
|
msg.add(0);
|
||||||
msg.add(0);
|
msg.add(0);
|
||||||
|
|
||||||
for(const auto& topic: subscriptions)
|
for(auto topic: subscriptions)
|
||||||
{
|
{
|
||||||
msg.add(topic);
|
msg.add(topic);
|
||||||
msg.add(0); // TODO qos
|
msg.add(0); // TODO qos
|
||||||
@@ -401,14 +440,13 @@ void MqttClient::processMessage(MqttMessage* mesg)
|
|||||||
switch(mesg->type())
|
switch(mesg->type())
|
||||||
{
|
{
|
||||||
case MqttMessage::Type::Connect:
|
case MqttMessage::Type::Connect:
|
||||||
if (mqtt_flags & FlagConnected)
|
if (mqtt_connected)
|
||||||
{
|
{
|
||||||
debug("already connected");
|
debug("already connected");
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
payload = header+10;
|
payload = header+10;
|
||||||
// Todo should check that reserved == 0 (spec)
|
mqtt_flags = header[7];
|
||||||
mqtt_flags = header[7] & ~FlagConnected;
|
|
||||||
keep_alive = MqttMessage::getSize(header+8);
|
keep_alive = MqttMessage::getSize(header+8);
|
||||||
if (strncmp("MQTT", header+2,4))
|
if (strncmp("MQTT", header+2,4))
|
||||||
{
|
{
|
||||||
@@ -448,10 +486,11 @@ void MqttClient::processMessage(MqttMessage* mesg)
|
|||||||
payload += len;
|
payload += len;
|
||||||
}
|
}
|
||||||
|
|
||||||
debug(yellow << "Client " << clientId << " connected : keep alive=" << keep_alive << '.' << white);
|
#if TINY_MQTT_DEBUG
|
||||||
|
Console << yellow << "Client " << clientId << " connected : keep alive=" << keep_alive << '.' << white << endl;
|
||||||
|
#endif
|
||||||
bclose = false;
|
bclose = false;
|
||||||
mqtt_flags |= FlagConnected;
|
mqtt_connected=true;
|
||||||
{
|
{
|
||||||
MqttMessage msg(MqttMessage::Type::ConnAck);
|
MqttMessage msg(MqttMessage::Type::ConnAck);
|
||||||
msg.add(0); // Session present (not implemented)
|
msg.add(0); // Session present (not implemented)
|
||||||
@@ -461,14 +500,14 @@ void MqttClient::processMessage(MqttMessage* mesg)
|
|||||||
break;
|
break;
|
||||||
|
|
||||||
case MqttMessage::Type::ConnAck:
|
case MqttMessage::Type::ConnAck:
|
||||||
mqtt_flags |= FlagConnected;
|
mqtt_connected = true;
|
||||||
bclose = false;
|
bclose = false;
|
||||||
resubscribe();
|
resubscribe();
|
||||||
break;
|
break;
|
||||||
|
|
||||||
case MqttMessage::Type::SubAck:
|
case MqttMessage::Type::SubAck:
|
||||||
case MqttMessage::Type::PubAck:
|
case MqttMessage::Type::PubAck:
|
||||||
if (not (mqtt_flags & FlagConnected)) break;
|
if (!mqtt_connected) break;
|
||||||
// Ignore acks
|
// Ignore acks
|
||||||
bclose = false;
|
bclose = false;
|
||||||
break;
|
break;
|
||||||
@@ -479,7 +518,7 @@ void MqttClient::processMessage(MqttMessage* mesg)
|
|||||||
break;
|
break;
|
||||||
|
|
||||||
case MqttMessage::Type::PingReq:
|
case MqttMessage::Type::PingReq:
|
||||||
if (not (mqtt_flags & FlagConnected)) break;
|
if (!mqtt_connected) break;
|
||||||
if (tcp_client)
|
if (tcp_client)
|
||||||
{
|
{
|
||||||
uint16_t pingreq = MqttMessage::Type::PingResp;
|
uint16_t pingreq = MqttMessage::Type::PingResp;
|
||||||
@@ -496,7 +535,7 @@ void MqttClient::processMessage(MqttMessage* mesg)
|
|||||||
case MqttMessage::Type::Subscribe:
|
case MqttMessage::Type::Subscribe:
|
||||||
case MqttMessage::Type::UnSubscribe:
|
case MqttMessage::Type::UnSubscribe:
|
||||||
{
|
{
|
||||||
if (not (mqtt_flags & FlagConnected)) break;
|
if (!mqtt_connected) break;
|
||||||
payload = header+2;
|
payload = header+2;
|
||||||
|
|
||||||
debug("un/subscribe loop");
|
debug("un/subscribe loop");
|
||||||
@@ -540,20 +579,24 @@ void MqttClient::processMessage(MqttMessage* mesg)
|
|||||||
break;
|
break;
|
||||||
|
|
||||||
case MqttMessage::Type::UnSuback:
|
case MqttMessage::Type::UnSuback:
|
||||||
if (not (mqtt_flags & FlagConnected)) break;
|
if (!mqtt_connected) break;
|
||||||
bclose = false;
|
bclose = false;
|
||||||
break;
|
break;
|
||||||
|
|
||||||
case MqttMessage::Type::Publish:
|
case MqttMessage::Type::Publish:
|
||||||
debug("publish " << (mqtt_flags & FlagConnected) << '/' << (long) tcp_client.get());
|
#if TINY_MQTT_DEBUG
|
||||||
if ((mqtt_flags & FlagConnected) or tcp_client == nullptr)
|
Console << "publish " << mqtt_connected << '/' << (long) tcp_client << endl;
|
||||||
|
#endif
|
||||||
|
if (mqtt_connected or tcp_client == nullptr)
|
||||||
{
|
{
|
||||||
uint8_t qos = mesg->flags();
|
uint8_t qos = mesg->flags();
|
||||||
payload = header;
|
payload = header;
|
||||||
mesg->getString(payload, len);
|
mesg->getString(payload, len);
|
||||||
Topic published(payload, len);
|
Topic published(payload, len);
|
||||||
payload += len;
|
payload += len;
|
||||||
debug("Received Publish (" << published.str().c_str() << ") size=" << (int)len);
|
#if TINY_MQTT_DEBUG
|
||||||
|
Console << "Received Publish (" << published.str().c_str() << ") size=" << (int)len << endl;
|
||||||
|
#endif
|
||||||
// << '(' << std::string(payload, len).c_str() << ')' << " msglen=" << mesg->length() << endl;
|
// << '(' << std::string(payload, len).c_str() << ')' << " msglen=" << mesg->length() << endl;
|
||||||
if (qos) payload+=2; // ignore packet identifier if any
|
if (qos) payload+=2; // ignore packet identifier if any
|
||||||
len=mesg->end()-payload;
|
len=mesg->end()-payload;
|
||||||
@@ -585,8 +628,8 @@ void MqttClient::processMessage(MqttMessage* mesg)
|
|||||||
|
|
||||||
case MqttMessage::Type::Disconnect:
|
case MqttMessage::Type::Disconnect:
|
||||||
// TODO should discard any will msg
|
// TODO should discard any will msg
|
||||||
if (not (mqtt_flags & FlagConnected)) break;
|
if (!mqtt_connected) break;
|
||||||
mqtt_flags &= ~FlagConnected;
|
mqtt_connected = false;
|
||||||
close(false);
|
close(false);
|
||||||
bclose=false;
|
bclose=false;
|
||||||
break;
|
break;
|
||||||
@@ -607,7 +650,7 @@ void MqttClient::processMessage(MqttMessage* mesg)
|
|||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
clientAlive();
|
clientAlive(local_broker ? 5 : 0);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -705,7 +748,10 @@ MqttError MqttClient::publishIfSubscribed(const Topic& topic, MqttMessage& msg)
|
|||||||
else
|
else
|
||||||
{
|
{
|
||||||
processMessage(&msg);
|
processMessage(&msg);
|
||||||
debug("Should call the callback ?");
|
|
||||||
|
#if TINY_MQTT_DEBUG
|
||||||
|
Console << "Should call the callback ?\n";
|
||||||
|
#endif
|
||||||
// callback(this, topic, nullptr, 0); // TODO Payload
|
// callback(this, topic, nullptr, 0); // TODO Payload
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,10 +4,6 @@
|
|||||||
#ifndef TINY_MQTT_DEBUG
|
#ifndef TINY_MQTT_DEBUG
|
||||||
#define TINY_MQTT_DEBUG 0
|
#define TINY_MQTT_DEBUG 0
|
||||||
#endif
|
#endif
|
||||||
#ifndef TINY_MQTT_DEFAULT_ALIVE
|
|
||||||
#define TINY_MQTT_DEFAULT_ALIVE 10
|
|
||||||
#endif
|
|
||||||
#define TINY_MQTT_CLIENT_ALIVE_TOLERANCE 5
|
|
||||||
|
|
||||||
// TODO Should add a AUnit with both TINY_MQTT_ASYNC and not TINY_MQTT_ASYNC
|
// TODO Should add a AUnit with both TINY_MQTT_ASYNC and not TINY_MQTT_ASYNC
|
||||||
// #define TINY_MQTT_ASYNC // Uncomment this to use ESPAsyncTCP instead of normal cnx
|
// #define TINY_MQTT_ASYNC // Uncomment this to use ESPAsyncTCP instead of normal cnx
|
||||||
@@ -37,7 +33,7 @@
|
|||||||
#include <rpcWiFi.h>
|
#include <rpcWiFi.h>
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
#include <memory>
|
#include <vector>
|
||||||
#include <set>
|
#include <set>
|
||||||
#include <string>
|
#include <string>
|
||||||
#include "StringIndexer.h"
|
#include "StringIndexer.h"
|
||||||
@@ -52,7 +48,7 @@
|
|||||||
static int debug;
|
static int debug;
|
||||||
};
|
};
|
||||||
|
|
||||||
#define debug(what) { if (TinyMqtt::debug>=1) Console << (int)__LINE__ << ' ' << what << TinyConsole::white << endl; delay(10); }
|
#define debug(what) { if (TinyMqtt::debug>=1) Console << (int)__LINE__ << ' ' << what << TinyConsole::white << endl; delay(100); }
|
||||||
#else
|
#else
|
||||||
#define debug(what) {}
|
#define debug(what) {}
|
||||||
#endif
|
#endif
|
||||||
@@ -177,9 +173,7 @@ class MqttClient
|
|||||||
FlagWillQos = 16 | 8, // unsupported
|
FlagWillQos = 16 | 8, // unsupported
|
||||||
FlagWill = 4, // unsupported
|
FlagWill = 4, // unsupported
|
||||||
FlagCleanSession = 2, // unsupported
|
FlagCleanSession = 2, // unsupported
|
||||||
|
FlagReserved = 1
|
||||||
FlagReserved = 1, // use reserved as connected (save 1 byte)
|
|
||||||
FlagConnected = 1
|
|
||||||
};
|
};
|
||||||
public:
|
public:
|
||||||
|
|
||||||
@@ -193,7 +187,7 @@ class MqttClient
|
|||||||
~MqttClient();
|
~MqttClient();
|
||||||
|
|
||||||
void connect(MqttBroker* local_broker);
|
void connect(MqttBroker* local_broker);
|
||||||
void connect(std::string broker, uint16_t port, uint16_t keep_alive = TINY_MQTT_DEFAULT_ALIVE);
|
void connect(std::string broker, uint16_t port, uint16_t keep_alive = 10);
|
||||||
|
|
||||||
// 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
|
||||||
@@ -289,21 +283,20 @@ class MqttClient
|
|||||||
// republish a received publish if topic matches any in subscriptions
|
// republish a received publish if topic matches any in subscriptions
|
||||||
MqttError publishIfSubscribed(const Topic& topic, MqttMessage& msg);
|
MqttError publishIfSubscribed(const Topic& topic, MqttMessage& msg);
|
||||||
|
|
||||||
void clientAlive();
|
void clientAlive(uint32_t more_seconds);
|
||||||
void processMessage(MqttMessage* message);
|
void processMessage(MqttMessage* message);
|
||||||
|
|
||||||
char mqtt_flags = 0;
|
bool mqtt_connected = false;
|
||||||
uint16_t keep_alive = 30;
|
char mqtt_flags;
|
||||||
// for client connected to remote broker, PingReq is sent when millis() >= alive
|
uint32_t keep_alive = 30;
|
||||||
// for a client managed by a broker, disconnect it if millis() >= alive
|
uint32_t alive;
|
||||||
uint32_t alive; // PingReq if millis() > alive,
|
|
||||||
MqttMessage message;
|
MqttMessage message;
|
||||||
|
|
||||||
// connection to local broker, or link to the parent
|
// connection to local broker, or link to the parent
|
||||||
// when MqttBroker uses MqttClient for each external connexion
|
// when MqttBroker uses MqttClient for each external connexion
|
||||||
MqttBroker* local_broker=nullptr;
|
MqttBroker* local_broker=nullptr;
|
||||||
|
|
||||||
std::unique_ptr<TcpClient> tcp_client; // connection to remote broker
|
TcpClient* tcp_client=nullptr; // connection to remote broker
|
||||||
std::set<Topic> subscriptions;
|
std::set<Topic> subscriptions;
|
||||||
std::string clientId;
|
std::string clientId;
|
||||||
CallBack callback = nullptr;
|
CallBack callback = nullptr;
|
||||||
@@ -320,6 +313,7 @@ class MqttBroker
|
|||||||
public:
|
public:
|
||||||
// TODO limit max number of clients
|
// TODO limit max number of clients
|
||||||
MqttBroker(uint16_t port);
|
MqttBroker(uint16_t port);
|
||||||
|
~MqttBroker();
|
||||||
|
|
||||||
void begin() { server->begin(); }
|
void begin() { server->begin(); }
|
||||||
void loop();
|
void loop();
|
||||||
@@ -327,21 +321,15 @@ class MqttBroker
|
|||||||
void connect(const std::string& host, uint16_t port=1883);
|
void connect(const std::string& host, uint16_t port=1883);
|
||||||
bool connected() const { return state == Connected; }
|
bool connected() const { return state == Connected; }
|
||||||
|
|
||||||
|
size_t clientsCount() const { return clients.size(); }
|
||||||
|
|
||||||
void dump(std::string indent="")
|
void dump(std::string indent="")
|
||||||
{
|
{
|
||||||
for(const auto& client: clients)
|
for(auto client: clients)
|
||||||
client->dump(indent);
|
client->dump(indent);
|
||||||
}
|
}
|
||||||
|
|
||||||
using Clients = std::set<std::unique_ptr<MqttClient>>;
|
const std::vector<MqttClient*> getClients() const { return clients; }
|
||||||
using LocalClients = std::set<MqttClient*>;
|
|
||||||
|
|
||||||
const Clients& getClients() const { return clients; }
|
|
||||||
const LocalClients& getLocalClients() const { return local_clients; }
|
|
||||||
|
|
||||||
size_t clientsCount() const { return clients.size(); }
|
|
||||||
size_t localClientsCount() const { return local_clients.size(); }
|
|
||||||
|
|
||||||
private:
|
private:
|
||||||
friend class MqttClient;
|
friend class MqttClient;
|
||||||
@@ -358,17 +346,15 @@ class MqttBroker
|
|||||||
|
|
||||||
MqttError subscribe(const Topic& topic, uint8_t qos);
|
MqttError subscribe(const Topic& topic, uint8_t qos);
|
||||||
|
|
||||||
void addClient(MqttClient* local) { local_clients.insert(local); }
|
// For clients that are added not by the broker itself (local clients)
|
||||||
void addClient(TcpClient* client);
|
void addClient(MqttClient* client);
|
||||||
|
void removeClient(MqttClient* client);
|
||||||
void removeClient(MqttClient* local);
|
|
||||||
|
|
||||||
bool compareString(const char* good, const char* str, uint8_t str_len) const;
|
bool compareString(const char* good, const char* str, uint8_t str_len) const;
|
||||||
Clients clients;
|
std::vector<MqttClient*> clients;
|
||||||
LocalClients local_clients;
|
|
||||||
|
|
||||||
private:
|
private:
|
||||||
std::unique_ptr<TcpServer> server;
|
TcpServer* server = nullptr;
|
||||||
|
|
||||||
const char* auth_user = "guest";
|
const char* auth_user = "guest";
|
||||||
const char* auth_password = "guest";
|
const char* auth_password = "guest";
|
||||||
|
|||||||
@@ -1,142 +0,0 @@
|
|||||||
// Implementation of C++14's make_unique for C++11 compilers.
|
|
||||||
//
|
|
||||||
// This has been tested with:
|
|
||||||
// - MSVC 11.0 (Visual Studio 2012)
|
|
||||||
// - gcc 4.6.3
|
|
||||||
// - Xcode 4.4 (with clang "4.0")
|
|
||||||
//
|
|
||||||
// It is based off an implementation proposed by Stephan T. Lavavej for
|
|
||||||
// inclusion in the C++14 standard:
|
|
||||||
// http://isocpp.org/files/papers/N3656.txt
|
|
||||||
// Where appropriate, it borrows the use of MSVC's _VARIADIC_EXPAND_0X macro
|
|
||||||
// machinery to compensate for lack of variadic templates.
|
|
||||||
//
|
|
||||||
// This file injects make_unique into the std namespace, which I acknowledge is
|
|
||||||
// technically forbidden ([C++11: 17.6.4.2.2.1/1]), but is necessary in order
|
|
||||||
// to have syntax compatibility with C++14.
|
|
||||||
//
|
|
||||||
// I perform compiler version checking for MSVC, gcc, and clang to ensure that
|
|
||||||
// we don't add make_unique if it is already there (instead, we include
|
|
||||||
// <memory> to get the compiler-provided one). You can override the compiler
|
|
||||||
// version checking by defining the symbol COMPILER_SUPPORTS_MAKE_UNIQUE.
|
|
||||||
//
|
|
||||||
//
|
|
||||||
// ===============================================================================
|
|
||||||
// This file is released into the public domain. See LICENCE for more information.
|
|
||||||
// ===============================================================================
|
|
||||||
|
|
||||||
#pragma once
|
|
||||||
|
|
||||||
// If user hasn't specified COMPILER_SUPPORTS_MAKE_UNIQUE then try to figure out
|
|
||||||
// based on compiler version if std::make_unique is provided.
|
|
||||||
#if !defined(COMPILER_SUPPORTS_MAKE_UNIQUE)
|
|
||||||
#if defined(_MSC_VER)
|
|
||||||
// std::make_unique was added in MSVC 12.0
|
|
||||||
#if _MSC_VER >= 1800 // MSVC 12.0 (Visual Studio 2013)
|
|
||||||
#define COMPILER_SUPPORTS_MAKE_UNIQUE
|
|
||||||
#endif
|
|
||||||
#elif defined(__clang__)
|
|
||||||
// std::make_unique was added in clang 3.4, but not until Xcode 6.
|
|
||||||
// Annoyingly, Apple makes the clang version defines match the version
|
|
||||||
// of Xcode, not the version of clang.
|
|
||||||
#define CLANG_VERSION (__clang_major__ * 10000 + __clang_minor__ * 100 + __clang_patchlevel__)
|
|
||||||
#if defined(__APPLE__) && CLANG_VERSION >= 60000
|
|
||||||
#define COMPILER_SUPPORTS_MAKE_UNIQUE
|
|
||||||
#elif !defined(__APPLE__) && CLANG_VERSION >= 30400
|
|
||||||
#define COMPILER_SUPPORTS_MAKE_UNIQUE
|
|
||||||
#endif
|
|
||||||
#elif defined(__GNUC__)
|
|
||||||
// std::make_unique was added in gcc 4.9, for standards versions greater
|
|
||||||
// than -std=c++11.
|
|
||||||
#define GCC_VERSION (__GNUC__ * 10000 + __GNUC_MINOR__ * 100 + __GNUC_PATCHLEVEL__)
|
|
||||||
#if GCC_VERSION >= 40900 && __cplusplus > 201103L
|
|
||||||
#define COMPILER_SUPPORTS_MAKE_UNIQUE
|
|
||||||
#endif
|
|
||||||
#endif
|
|
||||||
#endif
|
|
||||||
|
|
||||||
#if defined(COMPILER_SUPPORTS_MAKE_UNIQUE)
|
|
||||||
|
|
||||||
// If the compiler supports std::make_unique, then pull in <memory> to get it.
|
|
||||||
#include <memory>
|
|
||||||
|
|
||||||
#else
|
|
||||||
|
|
||||||
// Otherwise, the compiler doesn't provide it, so implement it ourselves.
|
|
||||||
|
|
||||||
#include <cstddef>
|
|
||||||
#include <memory>
|
|
||||||
#include <type_traits>
|
|
||||||
#include <utility>
|
|
||||||
|
|
||||||
namespace std {
|
|
||||||
|
|
||||||
template<class _Ty> struct _Unique_if {
|
|
||||||
typedef unique_ptr<_Ty> _Single_object;
|
|
||||||
};
|
|
||||||
|
|
||||||
template<class _Ty> struct _Unique_if<_Ty[]> {
|
|
||||||
typedef unique_ptr<_Ty[]> _Unknown_bound;
|
|
||||||
};
|
|
||||||
|
|
||||||
template<class _Ty, size_t N> struct _Unique_if<_Ty[N]> {
|
|
||||||
typedef void _Known_bound;
|
|
||||||
};
|
|
||||||
|
|
||||||
//
|
|
||||||
// template< class T, class... Args >
|
|
||||||
// unique_ptr<T> make_unique( Args&&... args);
|
|
||||||
//
|
|
||||||
|
|
||||||
#if defined(_MSC_VER) && (_MSC_VER < 1800)
|
|
||||||
|
|
||||||
// Macro machinery because MSVC 11.0 doesn't support variadic templates.
|
|
||||||
// The _VARIADIC_EXPAND_0X stuff is defined in <xstddef>
|
|
||||||
#define _MAKE_UNIQUE( \
|
|
||||||
TEMPLATE_LIST, PADDING_LIST, LIST, COMMA, X1, X2, X3, X4) \
|
|
||||||
template<class _Ty COMMA LIST(_CLASS_TYPE)> inline \
|
|
||||||
typename _Unique_if<_Ty>::_Single_object make_unique(LIST(_TYPE_REFREF_ARG)) \
|
|
||||||
{ \
|
|
||||||
return unique_ptr<_Ty>(new _Ty(LIST(_FORWARD_ARG))); \
|
|
||||||
} \
|
|
||||||
|
|
||||||
_VARIADIC_EXPAND_0X(_MAKE_UNIQUE, , , , )
|
|
||||||
#undef _MAKE_UNIQUE
|
|
||||||
|
|
||||||
#else // not MSVC 11.0 or earlier
|
|
||||||
|
|
||||||
template<class _Ty, class... Args>
|
|
||||||
typename _Unique_if<_Ty>::_Single_object
|
|
||||||
make_unique(Args&&... args) {
|
|
||||||
return unique_ptr<_Ty>(new _Ty(std::forward<Args>(args)...));
|
|
||||||
}
|
|
||||||
|
|
||||||
#endif
|
|
||||||
|
|
||||||
// template< class T >
|
|
||||||
// unique_ptr<T> make_unique( std::size_t size );
|
|
||||||
|
|
||||||
template<class _Ty>
|
|
||||||
typename _Unique_if<_Ty>::_Unknown_bound
|
|
||||||
make_unique(size_t n) {
|
|
||||||
typedef typename remove_extent<_Ty>::type U;
|
|
||||||
return unique_ptr<_Ty>(new U[n]());
|
|
||||||
}
|
|
||||||
|
|
||||||
// template< class T, class... Args >
|
|
||||||
// /* unspecified */ make_unique( Args&&... args ) = delete;
|
|
||||||
|
|
||||||
// MSVC 11.0 doesn't support deleted functions, so the best we can do
|
|
||||||
// is simply not define the function.
|
|
||||||
#if !(defined(_MSC_VER) && (_MSC_VER < 1800))
|
|
||||||
|
|
||||||
template<class T, class... Args>
|
|
||||||
typename _Unique_if<T>::_Known_bound
|
|
||||||
make_unique(Args&&...) = delete;
|
|
||||||
|
|
||||||
#endif
|
|
||||||
|
|
||||||
} // namespace std
|
|
||||||
|
|
||||||
#endif // !COMPILER_SUPPORTS_MAKE_UNIQUE
|
|
||||||
|
|
||||||
@@ -1,8 +1,6 @@
|
|||||||
SUB=n
|
|
||||||
|
|
||||||
tests:
|
tests:
|
||||||
set -e; \
|
set -e; \
|
||||||
for i in ${SUB}*-tests/Makefile; do \
|
for i in *-tests/Makefile; do \
|
||||||
echo '==== Making:' $$(dirname $$i); \
|
echo '==== Making:' $$(dirname $$i); \
|
||||||
$(MAKE) -C $$(dirname $$i) -j; \
|
$(MAKE) -C $$(dirname $$i) -j; \
|
||||||
done
|
done
|
||||||
@@ -17,14 +15,14 @@ runtests: debugtest
|
|||||||
$(MAKE) clean
|
$(MAKE) clean
|
||||||
$(MAKE) tests
|
$(MAKE) tests
|
||||||
set -e; \
|
set -e; \
|
||||||
for i in ${SUB}*-tests/Makefile; do \
|
for i in *-tests/Makefile; do \
|
||||||
echo '==== Running:' $$(dirname $$i); \
|
echo '==== Running:' $$(dirname $$i); \
|
||||||
$$(dirname $$i)/$$(dirname $$i).out; \
|
$$(dirname $$i)/$$(dirname $$i).out; \
|
||||||
done
|
done
|
||||||
|
|
||||||
clean:
|
clean:
|
||||||
set -e; \
|
set -e; \
|
||||||
for i in ${SUB}*-tests/Makefile; do \
|
for i in *-tests/Makefile; do \
|
||||||
echo '==== Cleaning:' $$(dirname $$i); \
|
echo '==== Cleaning:' $$(dirname $$i); \
|
||||||
$(MAKE) -C $$(dirname $$i) clean; \
|
$(MAKE) -C $$(dirname $$i) clean; \
|
||||||
done
|
done
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
# See https://github.com/bxparks/EpoxyDuino for documentation about this
|
# See https://github.com/bxparks/EpoxyDuino for documentation about this
|
||||||
# Makefile to compile and run Arduino programs natively on Linux or MacOS.
|
# Makefile to compile and run Arduino programs natively on Linux or MacOS.
|
||||||
|
|
||||||
EXTRA_CXXFLAGS=-g3 -O0 -DTINY_MQTT_DEFAULT_ALIVE=1
|
EXTRA_CXXFLAGS=-g3 -O0
|
||||||
|
|
||||||
# Remove flto flag from EpoxyDuino (too many <optimized out>)
|
# Remove flto flag from EpoxyDuino (too many <optimized out>)
|
||||||
CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics
|
CXXFLAGS = -Wextra -Wall -std=gnu++11 -fno-exceptions -fno-threadsafe-statics
|
||||||
|
|||||||
@@ -14,6 +14,7 @@
|
|||||||
|
|
||||||
using namespace std;
|
using namespace std;
|
||||||
|
|
||||||
|
MqttBroker broker(1883);
|
||||||
|
|
||||||
std::map<std::string, std::map<Topic, int>> published; // map[client_id] => map[topic] = count
|
std::map<std::string, std::map<Topic, int>> published; // map[client_id] => map[topic] = count
|
||||||
|
|
||||||
@@ -30,48 +31,57 @@ void onPublish(const MqttClient* srce, const Topic& topic, const char* payload,
|
|||||||
|
|
||||||
test(local_client_should_unregister_when_destroyed)
|
test(local_client_should_unregister_when_destroyed)
|
||||||
{
|
{
|
||||||
MqttBroker broker(1883);
|
assertEqual(broker.clientsCount(), (size_t)0);
|
||||||
assertEqual(broker.localClientsCount(), (size_t)0);
|
|
||||||
{
|
{
|
||||||
assertEqual(broker.localClientsCount(), (size_t)0); // Ensure client is not yet connected
|
assertEqual(broker.clientsCount(), (size_t)0); // Ensure client is not yet connected
|
||||||
MqttClient client(&broker);
|
MqttClient client(&broker);
|
||||||
assertEqual(broker.localClientsCount(), (size_t)1); // Ensure client is now connected
|
assertEqual(broker.clientsCount(), (size_t)1); // Ensure client is now connected
|
||||||
}
|
}
|
||||||
assertEqual(broker.localClientsCount(), (size_t)0);
|
assertEqual(broker.clientsCount(), (size_t)0);
|
||||||
}
|
}
|
||||||
|
|
||||||
test(local_client_alive)
|
test(local_client_do_not_disconnect_after_publishing)
|
||||||
{
|
{
|
||||||
set_millis(0);
|
set_millis(0);
|
||||||
MqttBroker broker(1883);
|
MqttBroker broker(1883);
|
||||||
MqttClient client(&broker);
|
MqttClient client(&broker, "client");
|
||||||
|
MqttClient sender(&broker, "sender");
|
||||||
broker.loop();
|
broker.loop();
|
||||||
assertEqual(broker.localClientsCount(), (size_t)1); // Ensure client is now connected
|
|
||||||
|
|
||||||
add_millis(TINY_MQTT_DEFAULT_ALIVE*1000/2);
|
client.subscribe("#");
|
||||||
broker.loop();
|
client.subscribe("test");
|
||||||
assertEqual(broker.localClientsCount(), (size_t)1); // Ensure client is still connected
|
client.setCallback(onPublish);
|
||||||
|
assertEqual(broker.clientsCount(), (size_t)2);
|
||||||
|
|
||||||
add_seconds(TINY_MQTT_DEFAULT_ALIVE*5);
|
sender.publish("test", "value");
|
||||||
broker.loop();
|
broker.loop();
|
||||||
assertEqual(broker.localClientsCount(), (size_t)1); // Ensure client is still connected
|
|
||||||
|
add_seconds(60);
|
||||||
|
client.loop();
|
||||||
|
sender.loop();
|
||||||
|
broker.loop();
|
||||||
|
|
||||||
|
assertEqual(broker.clientsCount(), (size_t)2);
|
||||||
|
assertEqual(sender.connected(), true);
|
||||||
|
assertEqual(client.connected(), true);
|
||||||
|
|
||||||
|
assertEqual(published.size(), (size_t)1); // client has received something
|
||||||
}
|
}
|
||||||
|
|
||||||
#if 0
|
#if 0
|
||||||
test(local_connect)
|
test(local_connect)
|
||||||
{
|
{
|
||||||
assertEqual(broker.localClientsCount(), (size_t)0);
|
assertEqual(broker.clientsCount(), (size_t)0);
|
||||||
|
|
||||||
MqttClient client;
|
MqttClient client;
|
||||||
assertTrue(client.connected());
|
assertTrue(client.connected());
|
||||||
assertEqual(broker.localClientsCount(), (size_t)1);
|
assertEqual(broker.clientsCount(), (size_t)1);
|
||||||
}
|
}
|
||||||
|
|
||||||
test(local_publish_should_be_dispatched)
|
test(local_publish_should_be_dispatched)
|
||||||
{
|
{
|
||||||
published.clear();
|
published.clear();
|
||||||
assertEqual(broker.localClientsCount(), (size_t)0);
|
assertEqual(broker.clientsCount(), (size_t)0);
|
||||||
|
|
||||||
MqttClient subscriber;
|
MqttClient subscriber;
|
||||||
subscriber.subscribe("a/b");
|
subscriber.subscribe("a/b");
|
||||||
@@ -91,7 +101,7 @@ test(local_publish_should_be_dispatched)
|
|||||||
test(local_publish_should_be_dispatched_to_local_clients)
|
test(local_publish_should_be_dispatched_to_local_clients)
|
||||||
{
|
{
|
||||||
published.clear();
|
published.clear();
|
||||||
assertEqual(broker.localClientsCount(), (size_t)0);
|
assertEqual(broker.clientsCount(), (size_t)0);
|
||||||
|
|
||||||
MqttClient subscriber_a("A");
|
MqttClient subscriber_a("A");
|
||||||
subscriber_a.setCallback(onPublish);
|
subscriber_a.setCallback(onPublish);
|
||||||
@@ -116,7 +126,7 @@ test(local_publish_should_be_dispatched_to_local_clients)
|
|||||||
test(local_unsubscribe)
|
test(local_unsubscribe)
|
||||||
{
|
{
|
||||||
published.clear();
|
published.clear();
|
||||||
assertEqual(broker.localClientsCount(), (size_t)0);
|
assertEqual(broker.clientsCount(), (size_t)0);
|
||||||
|
|
||||||
MqttClient subscriber;
|
MqttClient subscriber;
|
||||||
subscriber.setCallback(onPublish);
|
subscriber.setCallback(onPublish);
|
||||||
@@ -136,7 +146,7 @@ test(local_unsubscribe)
|
|||||||
test(local_nocallback_when_destroyed)
|
test(local_nocallback_when_destroyed)
|
||||||
{
|
{
|
||||||
published.clear();
|
published.clear();
|
||||||
assertEqual(broker.localClientsCount(), (size_t)0);
|
assertEqual(broker.clientsCount(), (size_t)0);
|
||||||
|
|
||||||
MqttClient publisher;
|
MqttClient publisher;
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -142,55 +142,6 @@ test(suback)
|
|||||||
assertEqual(MqttClient::counters[MqttMessage::Type::SubAck], 1);
|
assertEqual(MqttClient::counters[MqttMessage::Type::SubAck], 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
uint32_t getClientKeepAlive(MqttBroker& broker)
|
|
||||||
{
|
|
||||||
if (broker.getClients().size() == 1)
|
|
||||||
for (auto& it : broker.getClients())
|
|
||||||
return it->keepAlive();
|
|
||||||
|
|
||||||
return 9999;
|
|
||||||
}
|
|
||||||
|
|
||||||
test(network_client_alive)
|
|
||||||
{
|
|
||||||
const uint32_t keep_alive=1;
|
|
||||||
start_servers(2, true);
|
|
||||||
assertEqual(WiFi.status(), WL_CONNECTED);
|
|
||||||
set_millis(0); // Enter simulated time
|
|
||||||
|
|
||||||
MqttBroker broker(1883);
|
|
||||||
broker.begin();
|
|
||||||
IPAddress broker_ip = WiFi.localIP();
|
|
||||||
|
|
||||||
ESP8266WiFiClass::selectInstance(2);
|
|
||||||
MqttClient client;
|
|
||||||
client.connect(broker_ip.toString().c_str(), 1883, keep_alive);
|
|
||||||
broker.loop();
|
|
||||||
client.loop();
|
|
||||||
|
|
||||||
assertTrue(broker.clientsCount() == 1);
|
|
||||||
assertTrue(client.connected());
|
|
||||||
|
|
||||||
uint32_t ka = getClientKeepAlive(broker);
|
|
||||||
assertEqual(ka, keep_alive);
|
|
||||||
assertEqual(broker.clientsCount(), (size_t)1);
|
|
||||||
|
|
||||||
// All is going well if we call client.loop()
|
|
||||||
// The client is able to send PingReq to the broker
|
|
||||||
add_seconds(keep_alive);
|
|
||||||
client.loop();
|
|
||||||
broker.loop();
|
|
||||||
assertEqual(broker.clientsCount(), (size_t)1);
|
|
||||||
|
|
||||||
// Now simulate that the client is frozen for
|
|
||||||
// a too long time
|
|
||||||
add_seconds(TINY_MQTT_CLIENT_ALIVE_TOLERANCE*2);
|
|
||||||
broker.loop();
|
|
||||||
assertEqual(broker.clientsCount(), (size_t)0);
|
|
||||||
|
|
||||||
set_real_time();
|
|
||||||
}
|
|
||||||
|
|
||||||
test(network_client_keep_alive_high)
|
test(network_client_keep_alive_high)
|
||||||
{
|
{
|
||||||
const uint32_t keep_alive=1000;
|
const uint32_t keep_alive=1000;
|
||||||
@@ -221,8 +172,9 @@ test(network_client_keep_alive_high)
|
|||||||
uint32_t sz = broker.getClients().size();
|
uint32_t sz = broker.getClients().size();
|
||||||
assertEqual(sz , (uint32_t)1);
|
assertEqual(sz , (uint32_t)1);
|
||||||
|
|
||||||
uint32_t ka = getClientKeepAlive(broker);
|
uint32_t ka = broker.getClients()[0]->keepAlive();
|
||||||
assertEqual(ka, keep_alive);
|
assertEqual(ka, keep_alive);
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
test(network_client_to_broker_connexion)
|
test(network_client_to_broker_connexion)
|
||||||
|
|||||||
Reference in New Issue
Block a user