diff --git a/MQTT/MQTTImpl.cpp b/MQTT/MQTTImpl.cpp index 0077b67..0929ca7 100644 --- a/MQTT/MQTTImpl.cpp +++ b/MQTT/MQTTImpl.cpp @@ -51,20 +51,21 @@ void MQTTImpl::Disconnect(size_t connection) bool MQTTImpl::Subscribe(const std::string& topic) { + if (std::find(m_topics.begin(), m_topics.end(), topic) == m_topics.end()) + m_topics.push_back(topic); + if (m_pMosquitto && m_connected) - { - if (MOSQ_ERR_SUCCESS != mosquitto_subscribe(m_pMosquitto, NULL, topic.c_str(), 0)) - return false; - return true; - } + return InternalSubscribe(topic); return false; } void MQTTImpl::Unsubscribe(const std::string& topic) { + std::remove(m_topics.begin(), m_topics.end(), topic); + if (m_pMosquitto && m_connected) - mosquitto_unsubscribe(m_pMosquitto, NULL, topic.c_str()); + InternalUnsubscribe(topic); } void MQTTImpl::ConnectCallback(mosquitto* pMosquitto, void* pObject, int result) @@ -130,7 +131,6 @@ void MQTTImpl::Loop() { Logging::Log(Logging::Severity::Info, "MQTT::Loop() - Disconnected"); Disconnect(); - sleep(3); while (RunLoop() && !success) success = Connect(); } @@ -150,6 +150,8 @@ void MQTTImpl::StopLoop() void MQTTImpl::ConnectCallback(int result) { + for (const auto &topic : m_topics) + InternalSubscribe(topic); } void MQTTImpl::MessageCallback(const mosquitto_message* pMessage) @@ -161,4 +163,16 @@ void MQTTImpl::MessageCallback(const mosquitto_message* pMessage) m_signal.emit(message); } +bool MQTTImpl::InternalSubscribe(const std::string& topic) +{ + if (MOSQ_ERR_SUCCESS != mosquitto_subscribe(m_pMosquitto, NULL, topic.c_str(), 0)) + return false; + return true; +} + +void MQTTImpl::InternalUnsubscribe(const std::string& topic) +{ + mosquitto_unsubscribe(m_pMosquitto, NULL, topic.c_str()); +} + } // namespace MQTT diff --git a/MQTT/MQTTImpl.h b/MQTT/MQTTImpl.h index c6b64bd..c47370e 100644 --- a/MQTT/MQTTImpl.h +++ b/MQTT/MQTTImpl.h @@ -7,6 +7,7 @@ #include #include #include +#include struct mosquitto; @@ -45,6 +46,9 @@ private: void ConnectCallback(int result); void MessageCallback(const mosquitto_message* pMessage); + bool InternalSubscribe(const std::string& topic); + void InternalUnsubscribe(const std::string& topic); + private: std::thread m_thread; mutable std::mutex m_mutex; @@ -55,6 +59,7 @@ private: bool m_loop; std::string m_clientId; mosquitto* m_pMosquitto; + std::vector m_topics; MessageSignal m_signal; }; diff --git a/include/MQTT.h b/include/MQTT.h index 1635ace..ecdc42d 100644 --- a/include/MQTT.h +++ b/include/MQTT.h @@ -17,6 +17,8 @@ public: MQTT(const std::string& hostname, int port); ~MQTT(); + MQTT(const MQTT&) = delete; + public: typedef std::function CallbackMethod;