mgillespie/TinyMqtt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MqttClient::client renamed to tcp_client

Browse Source
This commit is contained in:
Francois BIOT
2022-12-29 12:52:29 +01:00
committed by hsaturn
parent 040568b478
commit efa94cc4a4
2 changed files with 35 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
src/TinyMqtt.cpp
View File

@@ -38,12 +38,12 @@ MqttClient::MqttClient(MqttBroker* local_broker, TcpClient* new_client)
connect(local_broker); connect(local_broker);
debug("MqttClient private with broker"); debug("MqttClient private with broker");
#ifdef TINY_MQTT_ASYNC #ifdef TINY_MQTT_ASYNC
client = new_client; tcp_client = new_client;
client->onData(onData, this); tcp_client->onData(onData, this);
// client->onConnect() TODO // client->onConnect() TODO
// client->onDisconnect() TODO // client->onDisconnect() TODO
#else #else
client = new WiFiClient(*new_client); tcp_client = new WiFiClient(*new_client);
#endif #endif
alive = millis()+5000; alive = millis()+5000;
} }
@@ -52,7 +52,6 @@ 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;
client = nullptr;
if (local_broker) local_broker->addClient(this); if (local_broker) local_broker->addClient(this);
} }
@@ -60,22 +59,23 @@ MqttClient::MqttClient(MqttBroker* local_broker, const std::string& id)
MqttClient::~MqttClient() MqttClient::~MqttClient()
{ {
close(); close();
delete client; delete tcp_client;
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_flags &= ~FlagConnected;
if (client) // connected to a remote broker if (tcp_client) // connected to a remote broker
{ {
if (bSendDisconnect and client->connected()) if (bSendDisconnect and tcp_client->connected())
{ {
message.create(MqttMessage::Type::Disconnect); message.create(MqttMessage::Type::Disconnect);
message.hexdump("close"); message.hexdump("close");
message.sendTo(this); message.sendTo(this);
} }
client->stop(); tcp_client->stop();
} }
if (local_broker) if (local_broker)
@@ -98,18 +98,18 @@ 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();
if (client) delete client; if (tcp_client) delete tcp_client;
client = new TcpClient; tcp_client = new TcpClient;
#ifdef TINY_MQTT_ASYNC #ifdef TINY_MQTT_ASYNC
client->onData(onData, this); tcp_client->onData(onData, this);
client->onConnect(onConnect, this); tcp_client->onConnect(onConnect, this);
client->connect(broker.c_str(), port, ka); tcp_client->connect(broker.c_str(), port, ka);
#else #else
if (client->connect(broker.c_str(), port)) if (tcp_client->connect(broker.c_str(), port))
{ {
debug("link established"); debug("link established");
onConnect(this, client); onConnect(this, tcp_client);
} }
else else
{ {
@@ -182,7 +182,7 @@ void MqttBroker::loop()
for(size_t i=0; i<clients.size(); i++) for(size_t i=0; i<clients.size(); i++)
{ {
auto client = clients[i]; MqttClient* client = clients[i];
if (client->connected()) if (client->connected())
{ {
client->loop(); client->loop();
@@ -284,11 +284,12 @@ void MqttClient::loop()
close(); close();
debug(red << "closed"); debug(red << "closed");
} }
else if (client && client->connected()) else if (tcp_client && tcp_client->connected())
{ {
debug("pingreq"); debug("pingreq");
uint16_t pingreq = MqttMessage::Type::PingReq; uint16_t pingreq = MqttMessage::Type::PingReq;
client->write((const char*)(&pingreq), 2);
tcp_client->write((const char*)(&pingreq), 2);
clientAlive(); clientAlive();
// TODO when many MqttClient passes through a local broker // TODO when many MqttClient passes through a local broker
@@ -296,9 +297,9 @@ void MqttClient::loop()
} }
} }
#ifndef TINY_MQTT_ASYNC #ifndef TINY_MQTT_ASYNC
while(client && client->available()>0) while(tcp_client && tcp_client->available()>0)
{ {
message.incoming(client->read()); message.incoming(tcp_client->read());
if (message.type()) if (message.type())
{ {
processMessage(&message); processMessage(&message);
@@ -512,11 +513,11 @@ void MqttClient::processMessage(MqttMessage* mesg)
case MqttMessage::Type::PingReq: case MqttMessage::Type::PingReq:
if (not (mqtt_flags & FlagConnected)) break; if (not (mqtt_flags & FlagConnected)) break;
if (client) if (tcp_client)
{ {
uint16_t pingreq = MqttMessage::Type::PingResp; uint16_t pingreq = MqttMessage::Type::PingResp;
debug(cyan << "Ping response to client "); debug(cyan << "Ping response to client ");
client->write((const char*)(&pingreq), 2); tcp_client->write((const char*)(&pingreq), 2);
bclose = false; bclose = false;
} }
else else
@@ -578,9 +579,9 @@ void MqttClient::processMessage(MqttMessage* mesg)
case MqttMessage::Type::Publish: case MqttMessage::Type::Publish:
#if TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
Console << "publish " << (mqtt_flags & FlagConnected) << '/' << (long) client << endl; Console << "publish " << (mqtt_flags & FlagConnected) << '/' << (long) tcp_client << endl;
#endif #endif
if ((mqtt_flags & FlagConnected) or client == nullptr) if ((mqtt_flags & FlagConnected) or tcp_client == nullptr)
{ {
uint8_t qos = mesg->flags(); uint8_t qos = mesg->flags();
payload = header; payload = header;
@@ -596,7 +597,7 @@ void MqttClient::processMessage(MqttMessage* mesg)
// TODO reset DUP // TODO reset DUP
// TODO reset RETAIN // TODO reset RETAIN
if (local_broker==nullptr or client==nullptr) // internal MqttClient receives publish if (local_broker==nullptr or tcp_client==nullptr) // internal MqttClient receives publish
{ {
#if TINY_MQTT_DEBUG #if TINY_MQTT_DEBUG
if (TinyMqtt::debug >= 2) if (TinyMqtt::debug >= 2)
@@ -722,7 +723,7 @@ MqttError MqttClient::publish(const Topic& topic, const char* payload, size_t pa
{ {
return local_broker->publish(this, topic, msg); return local_broker->publish(this, topic, msg);
} }
else if (client) else if (tcp_client)
return msg.sendTo(this); return msg.sendTo(this);
else else
return MqttNowhereToSend; return MqttNowhereToSend;
@@ -736,7 +737,7 @@ MqttError MqttClient::publishIfSubscribed(const Topic& topic, MqttMessage& msg)
debug("mqttclient publishIfSubscribed " << topic.c_str() << ' ' << subscriptions.size()); debug("mqttclient publishIfSubscribed " << topic.c_str() << ' ' << subscriptions.size());
if (isSubscribedTo(topic)) if (isSubscribedTo(topic))
{ {
if (client) if (tcp_client)
retval = msg.sendTo(this); retval = msg.sendTo(this);
else else
{ {

13
src/TinyMqtt.h
View File

@@ -199,12 +199,13 @@ class MqttClient
// no negociation occurred // no negociation occurred
bool connected() bool connected()
{ {
return (local_broker!=nullptr and client==nullptr) or (client and client->connected()); return (local_broker!=nullptr and tcp_client==nullptr)
or (tcp_client and tcp_client->connected());
} }
void write(const char* buf, size_t length) void write(const char* buf, size_t length)
{ {
if (client) client->write(buf, length); if (tcp_client) tcp_client->write(buf, length);
} }
const std::string& id() const { return clientId; } const std::string& id() const { return clientId; }
@@ -235,7 +236,7 @@ class MqttClient
// connected to local broker // connected to local broker
// TODO seems to be useless // TODO seems to be useless
bool isLocal() const { return client == nullptr; } bool isLocal() const { return tcp_client == nullptr; }
void dump(std::string indent="") void dump(std::string indent="")
{ {
@@ -244,9 +245,9 @@ class MqttClient
uint32_t ms=millis(); uint32_t ms=millis();
Console << indent << "+-- " << '\'' << clientId.c_str() << "' " << (connected() ? " ON " : " OFF"); Console << indent << "+-- " << '\'' << clientId.c_str() << "' " << (connected() ? " ON " : " OFF");
Console << ", alive=" << alive << '/' << ms << ", ka=" << keep_alive << ' '; Console << ", alive=" << alive << '/' << ms << ", ka=" << keep_alive << ' ';
if (client) if (tcp_client)
{ {
if (client->connected()) if (tcp_client->connected())
Console << TinyConsole::green << "connected"; Console << TinyConsole::green << "connected";
else else
Console << TinyConsole::red << "disconnected"; Console << TinyConsole::red << "disconnected";
@@ -302,7 +303,7 @@ class MqttClient
// when MqttBroker uses MqttClient for each external connexion // when MqttBroker uses MqttClient for each external connexion
MqttBroker* local_broker=nullptr; MqttBroker* local_broker=nullptr;
TcpClient* client=nullptr; // 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;