From 859e336eca5176a8c7bd9c505721f0dcd190b241 Mon Sep 17 00:00:00 2001 From: JDierkse Date: Fri, 12 Jun 2020 11:31:09 +0200 Subject: [PATCH] Initial Commit --- .gitignore | 6 ++ .gitmodules | 3 + Libraries/Logging | 1 + Libraries/SimpleSignal | 1 + MQTT/MQTT.cpp | 41 +++++++++++ MQTT/MQTTImpl.cpp | 164 +++++++++++++++++++++++++++++++++++++++++ MQTT/MQTTImpl.h | 64 ++++++++++++++++ MQTT/Makefile | 1 + Makefile | 1 + Makefile.conf | 17 +++++ Makefile.target | 12 +++ Makefiles | 1 + include/MQTT.h | 36 +++++++++ include/Message.h | 17 +++++ 14 files changed, 365 insertions(+) create mode 100644 .gitignore create mode 100644 .gitmodules create mode 120000 Libraries/Logging create mode 120000 Libraries/SimpleSignal create mode 100644 MQTT/MQTT.cpp create mode 100644 MQTT/MQTTImpl.cpp create mode 100644 MQTT/MQTTImpl.h create mode 120000 MQTT/Makefile create mode 120000 Makefile create mode 100644 Makefile.conf create mode 100644 Makefile.target create mode 160000 Makefiles create mode 100644 include/MQTT.h create mode 100644 include/Message.h diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..851ef09 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +*.o.* +*.d.* +*.a.* +.*.swp +.AppleDouble +lib diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..876e403 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "Makefiles"] + path = Makefiles + url = https://gogs.dierkse.nl/JDierkse/Makefiles diff --git a/Libraries/Logging b/Libraries/Logging new file mode 120000 index 0000000..e18bda1 --- /dev/null +++ b/Libraries/Logging @@ -0,0 +1 @@ +../../Logging \ No newline at end of file diff --git a/Libraries/SimpleSignal b/Libraries/SimpleSignal new file mode 120000 index 0000000..86aa44f --- /dev/null +++ b/Libraries/SimpleSignal @@ -0,0 +1 @@ +../../SimpleSignal \ No newline at end of file diff --git a/MQTT/MQTT.cpp b/MQTT/MQTT.cpp new file mode 100644 index 0000000..f4c94d9 --- /dev/null +++ b/MQTT/MQTT.cpp @@ -0,0 +1,41 @@ +#include "MQTT.h" +#include "MQTTImpl.h" + + +namespace MQTT { + +MQTT::MQTT(const std::string& hostname, int port) : + m_pMQTTImpl(new MQTTImpl(hostname, port)) +{ +} + +MQTT::~MQTT() +{ +} + +bool MQTT::Send(const Message& message) +{ + return m_pMQTTImpl->Send(message); +} + +bool MQTT::Subscribe(const std::string& topic) +{ + return m_pMQTTImpl->Subscribe(topic); +} + +void MQTT::Unsubscribe(const std::string& topic) +{ + m_pMQTTImpl->Unsubscribe(topic); +} + +size_t MQTT::Connect(CallbackMethod function) +{ + return m_pMQTTImpl->Connect(function); +} + +void MQTT::Disconnect(size_t connection) +{ + m_pMQTTImpl->Disconnect(connection); +} + +} // namespace MQTT diff --git a/MQTT/MQTTImpl.cpp b/MQTT/MQTTImpl.cpp new file mode 100644 index 0000000..7ffb9e3 --- /dev/null +++ b/MQTT/MQTTImpl.cpp @@ -0,0 +1,164 @@ +#include "MQTTImpl.h" +#include +#include +#include +#include + + +namespace MQTT { + +MQTTImpl::MQTTImpl(const std::string& hostname, int port) : + m_hostname(hostname), + m_port(port), + m_connected(false), + m_loop(false) +{ + std::stringstream clientId; + clientId << "MQTT_" << getpid(); + m_clientId = clientId.str(); + + mosquitto_lib_init(); + m_pMosquitto = mosquitto_new(m_clientId.c_str(), true, this); + + m_connected = Connect(); + StartLoop(); +} + +MQTTImpl::~MQTTImpl() +{ + StopLoop(); + Disconnect(); + mosquitto_destroy(m_pMosquitto); + mosquitto_lib_cleanup(); +} + +bool MQTTImpl::Send(const Message& message) +{ + if (m_connected) + return (MOSQ_ERR_SUCCESS == mosquitto_publish(m_pMosquitto, NULL, message.topic.c_str(), message.payload.length(), message.payload.c_str(), 0, 0)); + return false; +} + +size_t MQTTImpl::Connect(MQTT::CallbackMethod function) +{ + return m_signal.connect(function); +} + +void MQTTImpl::Disconnect(size_t connection) +{ + m_signal.disconnect(connection); +} + +bool MQTTImpl::Subscribe(const std::string& topic) +{ + if (m_pMosquitto && m_connected) + { + if (MOSQ_ERR_SUCCESS != mosquitto_subscribe(m_pMosquitto, NULL, topic.c_str(), 0)) + return false; + return true; + } + + return false; +} + +void MQTTImpl::Unsubscribe(const std::string& topic) +{ + if (m_pMosquitto && m_connected) + mosquitto_unsubscribe(m_pMosquitto, NULL, topic.c_str()); +} + +void MQTTImpl::ConnectCallback(mosquitto* pMosquitto, void* pObject, int result) +{ + auto pMQTTImpl = static_cast(pObject); + pMQTTImpl->ConnectCallback(result); +} + +void MQTTImpl::MessageCallback(mosquitto* pMosquitto, void* pObject, const mosquitto_message* pMessage) +{ + auto pMQTTImpl = static_cast(pObject); + pMQTTImpl->MessageCallback(pMessage); +} + +bool MQTTImpl::Connect() +{ + if (m_pMosquitto) + { + mosquitto_connect_callback_set(m_pMosquitto, MQTTImpl::ConnectCallback); + mosquitto_message_callback_set(m_pMosquitto, MQTTImpl::MessageCallback); + + if (MOSQ_ERR_SUCCESS != mosquitto_connect(m_pMosquitto, m_hostname.c_str(), m_port, 60)) + return false; + + Logging::Log(Logging::Severity::Info, "MQTT::Connect() - Connected"); + + return true; + } + + return false; +} + +void MQTTImpl::Disconnect() +{ + if (m_pMosquitto && m_connected) + { + if (MOSQ_ERR_SUCCESS != mosquitto_disconnect(m_pMosquitto)) + return; + m_connected = false; + } +} + +void MQTTImpl::StartLoop() +{ + std::unique_lock lock(m_mutex); + m_loop = true; + m_thread = std::thread([&] { Loop(); }); +} + +bool MQTTImpl::RunLoop() const +{ + std::unique_lock lock(m_mutex); + return m_loop; +} + +void MQTTImpl::Loop() +{ + while (RunLoop()) + { + bool success = (MOSQ_ERR_SUCCESS == mosquitto_loop(m_pMosquitto, -1, 1)); + + if (m_loop && !success) + { + Logging::Log(Logging::Severity::Info, "MQTT::Loop() - Disconnected"); + Disconnect(); + sleep(3); + while (RunLoop() && !success) + success = Connect(); + } + } + + Logging::Log(Logging::Severity::Info, "MQTT::Loop() - End"); +} + +void MQTTImpl::StopLoop() +{ + { + std::unique_lock lock(m_mutex); + m_loop = false; + } + m_thread.join(); +} + +void MQTTImpl::ConnectCallback(int result) +{ +} + +void MQTTImpl::MessageCallback(const mosquitto_message* pMessage) +{ + Message message; + message.topic = std::string(pMessage->topic); + message.payload = std::string(static_cast(pMessage->payload), pMessage->payloadlen); + + m_signal.emit(message); +} + +} // namespace MQTT diff --git a/MQTT/MQTTImpl.h b/MQTT/MQTTImpl.h new file mode 100644 index 0000000..85956a7 --- /dev/null +++ b/MQTT/MQTTImpl.h @@ -0,0 +1,64 @@ +#ifndef MQTT_MQTTIMPL_H +#define MQTT_MQTTIMPL_H + +#include "Message.h" +#include "MQTT.h" +#include +#include +#include +#include + + +struct mosquitto; +struct mosquitto_message; + +namespace MQTT { + +class MQTTImpl +{ +public: + MQTTImpl(const std::string& hostname, int port); + ~MQTTImpl(); + + bool Send(const Message& message); + bool Subscribe(const std::string& topic); + void Unsubscribe(const std::string& topic); + size_t Connect(MQTT::CallbackMethod function); + void Disconnect(size_t connection); + +private: + typedef Simple::Signal MessageSignal; + +private: + static void ConnectCallback(mosquitto* pMosquitto, void* pObject, int result); + static void MessageCallback(mosquitto* pMosquitto, void* pObject, const mosquitto_message* pMessage); + +private: + bool Connect(); + void Disconnect(); + + void StartLoop(); + bool RunLoop() const; + void Loop(); + void StopLoop(); + + void ConnectCallback(int result); + void MessageCallback(const mosquitto_message* pMessage); + +private: + std::thread m_thread; + mutable std::mutex m_mutex; + + std::string m_hostname; + int m_port; + bool m_connected; + bool m_loop; + std::string m_clientId; + mosquitto* m_pMosquitto; + + MessageSignal m_signal; +}; + +} // namespace MQTT + +#endif // MQTTIMPL_H diff --git a/MQTT/Makefile b/MQTT/Makefile new file mode 120000 index 0000000..d0b0e8e --- /dev/null +++ b/MQTT/Makefile @@ -0,0 +1 @@ +../Makefile \ No newline at end of file diff --git a/Makefile b/Makefile new file mode 120000 index 0000000..6aa5187 --- /dev/null +++ b/Makefile @@ -0,0 +1 @@ +Makefiles/Makefile \ No newline at end of file diff --git a/Makefile.conf b/Makefile.conf new file mode 100644 index 0000000..2163e0b --- /dev/null +++ b/Makefile.conf @@ -0,0 +1,17 @@ +# +# Makefile.conf +# + +LFLAGS += -lLogging +LFLAGS += -L$(ROOTPATH)/Libraries/Logging/lib/$(ARCH) +CFLAGS += -I$(ROOTPATH)/Libraries/Logging/include + +CFLAGS += -I$(ROOTPATH)/Libraries/SimpleSignal + +LFLAGS += -lmosquitto + +LFLAGS += -lMQTT +LFLAGS += -L$(ROOTPATH)/lib/$(ARCH) +CFLAGS += -I$(ROOTPATH) -I$(ROOTPATH)/include + +DEBUGDIR := .debug diff --git a/Makefile.target b/Makefile.target new file mode 100644 index 0000000..efad89f --- /dev/null +++ b/Makefile.target @@ -0,0 +1,12 @@ +# +# Makefile.target +# + +MQTT.a.$(ARCH) : $(OBJECTS) + $(call build_target_library_arch,$@,$^) +MQTT.a: + $(call build_target,$@) + +.DEFAULT_GOAL := MQTT.a + +TARGETS += MQTT.a diff --git a/Makefiles b/Makefiles new file mode 160000 index 0000000..cd9ef1a --- /dev/null +++ b/Makefiles @@ -0,0 +1 @@ +Subproject commit cd9ef1a1b0f7b88f36b0c51b4209d1859aca83aa diff --git a/include/MQTT.h b/include/MQTT.h new file mode 100644 index 0000000..6fd142f --- /dev/null +++ b/include/MQTT.h @@ -0,0 +1,36 @@ +#ifndef MQTT_H +#define MQTT_H + +#include "Message.h" +#include +#include +#include + + +namespace MQTT { + +class MQTTImpl; + +class MQTT +{ +public: + MQTT(const std::string& hostname, int port); + ~MQTT(); + +public: + typedef std::function CallbackMethod; + +public: + bool Send(const Message& message); + bool Subscribe(const std::string& topic); + void Unsubscribe(const std::string& topic); + size_t Connect(CallbackMethod function); + void Disconnect(size_t connection); + +private: + std::unique_ptr m_pMQTTImpl; +}; + +} // namespace MQTT + +#endif // MQTT_H diff --git a/include/Message.h b/include/Message.h new file mode 100644 index 0000000..03b05e9 --- /dev/null +++ b/include/Message.h @@ -0,0 +1,17 @@ +#ifndef MESSAGE_H +#define MESSAGE_H + +#include + + +namespace MQTT { + +struct Message +{ + std::string topic; + std::string payload; +}; + +} // namespace MQTT + +#endif // MESSAGE_H