Initial Commit

.gitignore

@@ -0,0 +1,6 @@
+*.o.*
+*.d.*
+*.a.*
+.*.swp
+.AppleDouble
+lib

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "Makefiles"]
+	path = Makefiles
+	url = https://gogs.dierkse.nl/JDierkse/Makefiles

Libraries/Logging

@@ -0,0 +1 @@
+../../Logging

Libraries/SimpleSignal

@@ -0,0 +1 @@
+../../SimpleSignal

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

MQTT/MQTTImpl.cpp

@@ -0,0 +1,164 @@
+#include "MQTTImpl.h"
+#include <Logging.h>
+#include <mosquitto.h>
+#include <sstream>
+#include <unistd.h>
+
+
+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<MQTTImpl*>(pObject);
+	pMQTTImpl->ConnectCallback(result);
+}
+
+void MQTTImpl::MessageCallback(mosquitto* pMosquitto, void* pObject, const mosquitto_message* pMessage)
+{
+	auto pMQTTImpl = static_cast<MQTTImpl*>(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<std::mutex> lock(m_mutex);
+	m_loop = true;
+	m_thread = std::thread([&] { Loop(); });
+}
+
+bool MQTTImpl::RunLoop() const
+{
+	std::unique_lock<std::mutex> 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<std::mutex> 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<char*>(pMessage->payload), pMessage->payloadlen);
+
+	m_signal.emit(message);
+}
+
+} // namespace MQTT

MQTT/MQTTImpl.h

@@ -0,0 +1,64 @@
+#ifndef MQTT_MQTTIMPL_H
+#define MQTT_MQTTIMPL_H
+
+#include "Message.h"
+#include "MQTT.h"
+#include <SimpleSignal.h>
+#include <mutex>
+#include <string>
+#include <thread>
+
+
+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<void(const Message&)> 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

MQTT/Makefile

@@ -0,0 +1 @@
+../Makefile

Makefile

@@ -0,0 +1 @@
+Makefiles/Makefile

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

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

Makefiles

@@ -0,0 +1 @@
+Subproject commit cd9ef1a1b0f7b88f36b0c51b4209d1859aca83aa

include/MQTT.h

@@ -0,0 +1,36 @@
+#ifndef MQTT_H
+#define MQTT_H
+
+#include "Message.h"
+#include <functional>
+#include <memory>
+#include <string>
+
+
+namespace MQTT {
+
+class MQTTImpl;
+
+class MQTT
+{
+public:
+	MQTT(const std::string& hostname, int port);
+	~MQTT();
+
+public:
+	typedef std::function<void(const Message&)> 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<MQTTImpl> m_pMQTTImpl;
+};
+
+} // namespace MQTT
+
+#endif // MQTT_H

include/Message.h

@@ -0,0 +1,17 @@
+#ifndef MESSAGE_H
+#define MESSAGE_H
+
+#include <string>
+
+
+namespace MQTT {
+
+struct Message
+{
+	std::string topic;
+	std::string payload;
+};
+
+} // namespace MQTT
+
+#endif // MESSAGE_H