From e3873bf6ecb434336275e359e9576406f57422fa Mon Sep 17 00:00:00 2001
From: JDierkse <j.dierkse@madeo.nl>
Date: Thu, 16 Jan 2025 14:57:27 +0100
Subject: [PATCH] Add multi-threaded behavior

---
 Libraries/CTPL    | 1 +
 MQTT/MQTT.cpp     | 4 ++--
 MQTT/MQTTImpl.cpp | 8 ++++++--
 MQTT/MQTTImpl.h   | 4 +++-
 Makefile.conf     | 1 +
 include/MQTT.h    | 2 +-
 6 files changed, 14 insertions(+), 6 deletions(-)
 create mode 120000 Libraries/CTPL

diff --git a/Libraries/CTPL b/Libraries/CTPL
new file mode 120000
index 0000000..3408abd
--- /dev/null
+++ b/Libraries/CTPL
@@ -0,0 +1 @@
+../../CTPL
\ No newline at end of file
diff --git a/MQTT/MQTT.cpp b/MQTT/MQTT.cpp
index 6b0fd05..26ed134 100644
--- a/MQTT/MQTT.cpp
+++ b/MQTT/MQTT.cpp
@@ -4,8 +4,8 @@
 
 namespace MQTT {
 
-MQTT::MQTT(const std::string& hostname, int port) :
-	m_pMQTTImpl(new MQTTImpl(hostname, port))
+MQTT::MQTT(const std::string& hostname, int port, int threads) :
+	m_pMQTTImpl(new MQTTImpl(hostname, port, threads))
 {
 }
 
diff --git a/MQTT/MQTTImpl.cpp b/MQTT/MQTTImpl.cpp
index 0929ca7..ba95a70 100644
--- a/MQTT/MQTTImpl.cpp
+++ b/MQTT/MQTTImpl.cpp
@@ -7,7 +7,8 @@
 
 namespace MQTT {
 
-MQTTImpl::MQTTImpl(const std::string& hostname, int port) :
+MQTTImpl::MQTTImpl(const std::string& hostname, int port, int threads) :
+	m_threadPool(threads),
 	m_hostname(hostname),
 	m_port(port),
 	m_connected(false),
@@ -26,6 +27,7 @@ MQTTImpl::MQTTImpl(const std::string& hostname, int port) :
 
 MQTTImpl::~MQTTImpl()
 {
+	m_threadPool.stop();
 	StopLoop();
 	Disconnect();
 	mosquitto_destroy(m_pMosquitto);
@@ -160,7 +162,9 @@ void MQTTImpl::MessageCallback(const mosquitto_message* pMessage)
 	message.topic = std::string(pMessage->topic);
 	message.payload = std::string(static_cast<char*>(pMessage->payload), pMessage->payloadlen);
 
-	m_signal.emit(message);
+	m_threadPool.push([this, message](int) {
+		m_signal.emit(message);
+	});
 }
 
 bool MQTTImpl::InternalSubscribe(const std::string& topic)
diff --git a/MQTT/MQTTImpl.h b/MQTT/MQTTImpl.h
index c47370e..372de11 100644
--- a/MQTT/MQTTImpl.h
+++ b/MQTT/MQTTImpl.h
@@ -3,6 +3,7 @@
 
 #include "MQTT.h"
 #include "MQTTMessage.h"
+#include <ctpl_stl.h>
 #include <SimpleSignal.h>
 #include <mutex>
 #include <string>
@@ -18,7 +19,7 @@ namespace MQTT {
 class MQTTImpl
 {
 public:
-	MQTTImpl(const std::string& hostname, int port);
+	MQTTImpl(const std::string& hostname, int port, int threads);
 	~MQTTImpl();
 
 	bool Send(const MQTTMessage& message);
@@ -52,6 +53,7 @@ private:
 private:
 	std::thread m_thread;
 	mutable std::mutex m_mutex;
+	ctpl::thread_pool m_threadPool;
 
 	std::string m_hostname;
 	int m_port;
diff --git a/Makefile.conf b/Makefile.conf
index d1c04d9..87529aa 100644
--- a/Makefile.conf
+++ b/Makefile.conf
@@ -4,6 +4,7 @@
 
 LIBRARIES += Logging
 
+CFLAGS += -I$(ROOTPATH)/Libraries/CTPL
 CFLAGS += -I$(ROOTPATH)/Libraries/SimpleSignal
 
 DEBUGDIR := .debug
diff --git a/include/MQTT.h b/include/MQTT.h
index ecdc42d..8ffd463 100644
--- a/include/MQTT.h
+++ b/include/MQTT.h
@@ -14,7 +14,7 @@ class MQTTImpl;
 class MQTT
 {
 public:
-	MQTT(const std::string& hostname, int port);
+	MQTT(const std::string& hostname, int port, int threads = 3);
 	~MQTT();
 
 	MQTT(const MQTT&) = delete;