Resubscribe upon reconnect

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

MQTT/MQTTImpl.h

@@ -7,6 +7,7 @@
 #include <mutex>
 #include <string>
 #include <thread>
+#include <vector>
 
 
 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<std::string> m_topics;
 
 	MessageSignal m_signal;
 };

include/MQTT.h

@@ -17,6 +17,8 @@ public:
 	MQTT(const std::string& hostname, int port);
 	~MQTT();
 
+	MQTT(const MQTT&) = delete;
+
 public:
 	typedef std::function<void(const MQTTMessage&)> CallbackMethod;