first version of MQTT support

configure.ac

@@ -26,9 +26,10 @@ AC_CHECK_HEADERS([arpa/inet.h \
 AC_CHECK_LIB([pthread], [pthread_setname_np],
 	AC_DEFINE([HAVE_PTHREAD_SETNAME_NP], [1], [Define to 1 if pthread has pthread_setname_np.]),
 	AC_MSG_RESULT([Could not find pthread_setname_np in pthread.]))
-RT_LIB=
+EXTRA_LIBS=
-AC_CHECK_LIB([rt], [clock_gettime], [RT_LIB="-lrt"])
+AC_CHECK_LIB([rt], [clock_gettime], [EXTRA_LIBS+="-lrt"])
-AC_SUBST(RT_LIB)
+AC_SUBST(EXTRA_LIBS)
+
 AC_CHECK_FUNC([pselect], [AC_DEFINE(HAVE_PSELECT, [1], [Define to 1 if pselect() is available.])])
 AC_CHECK_FUNC([ppoll], [AC_DEFINE(HAVE_PPOLL, [1], [Define to 1 if ppoll() is available.])])
 
@@ -45,6 +46,15 @@ if test "x$have_argp" = "xyes"; then
 else
 	AC_MSG_ERROR([argp library not found, specify argp-standalone location in --with-argp-lib= and --with-argp-include= options.])
 fi
+AC_ARG_WITH(mqtt, AS_HELP_STRING([--without-mqtt], [disable support for MQTT handling]), [], [with_mqtt=yes])
+if test "x$with_mqtt" != "xno"; then
+	AC_CHECK_LIB([mosquitto], [mosquitto_lib_init],
+		[AC_DEFINE_UNQUOTED(HAVE_MQTT, [1], [Define to 1 for enabled MQTT handling.])
+		EXTRA_LIBS+=" -lmosquitto"],
+		[AC_MSG_RESULT([Could not find mosquitto_lib_init in libmosquitto.])
+		with_mqtt="no"])
+fi
+AM_CONDITIONAL([MQTT], [test "x$with_mqtt" != "xno"])
 
 AC_MSG_CHECKING([for direct float format conversion])
 AC_TRY_RUN(

src/ebusd/Makefile.am

@@ -8,6 +8,8 @@ bin_PROGRAMS = ebusd
 
 ebusd_SOURCES = bushandler.cpp \
 		bushandler.h \
+		datahandler.cpp \
+		datahandler.h \
 		network.cpp \
 		network.h \
 		mainloop.cpp \
@@ -15,10 +17,14 @@ ebusd_SOURCES = bushandler.cpp \
 		main.h \
 		main.cpp
 
+if MQTT
+ebusd_SOURCES += mqtthandler.cpp mqtthandler.h
+endif
+
 ebusd_LDADD = ../lib/utils/libutils.a \
               ../lib/ebus/libebus.a \
 	      -lpthread \
-	      @RT_LIB@
+	      @EXTRA_LIBS@
 
 distclean-local:
 	-rm -f Makefile.in

src/ebusd/mqtthandler.cpp

@@ -0,0 +1,308 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2016-2017 John Baier <ebusd@ebusd.eu>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "mqtthandler.h"
+#include "log.h"
+#ifdef HAVE_CONFIG_H
+#	include <config.h>
+#endif
+
+using namespace std;
+
+/** the definition of the MQTT arguments. */
+static const struct argp_option g_mqtt_argp_options[] = {
+	{NULL,             0, NULL,    0, "MQTT options:", 1 },
+	{"mqttport",       1, "PORT",  0, "Connect to MQTT broker on PORT, 0 to disable [0]", 0 },
+	{"mqtttopic",      2, "TOPIC", 0, "Use MQTT TOPIC (prefix if without wildcards) [ebusd]", 0 },
+
+	{NULL,             0,        NULL,    0, NULL, 0 },
+};
+
+static uint16_t g_port = 0; //!< optional port of MQTT broker, 0 to disable [0]
+static const char* g_topic = PACKAGE; //!< MQTT topic to use (prefix if without wildcards) [ebusd/$circuit/$name]
+
+
+/**
+ * The MQTT argument parsing function.
+ * @param key the key from @a mqtt_argp_options.
+ * @param arg the option argument, or NULL.
+ * @param state the parsing state.
+ */
+static error_t mqtt_parse_opt(int key, char *arg, struct argp_state *state)
+{
+	result_t result = RESULT_OK;
+
+	switch (key) {
+	case 1: // --mqttport=1883
+		g_port = (uint16_t)parseInt(arg, 10, 1, 65535, result);
+		if (result != RESULT_OK) {
+			argp_error(state, "invalid mqttport");
+			return EINVAL;
+		}
+		break;
+
+	case 2: // --mqtttopic=ebusd
+		if (arg == NULL || arg[0] == 0 || arg[0]=='/' || arg[strlen(arg)-1]=='/') {
+			argp_error(state, "invalid mqtttopic");
+			return EINVAL;
+		}
+		g_topic = arg;
+		break;
+	default:
+		return ARGP_ERR_UNKNOWN;
+	}
+	return 0;
+}
+
+static struct argp g_mqtt_argp = { g_mqtt_argp_options, mqtt_parse_opt, NULL, NULL, NULL, NULL, NULL };
+static const struct argp_child g_mqtt_argp_child = {&g_mqtt_argp, 0, "", 1};
+
+const struct argp_child* mqtthandler_getargs()
+{
+	return &g_mqtt_argp_child;
+}
+
+DataHandler* mqtthandler_register(BusHandler* busHandler)
+{
+	return new MqttHandler(busHandler);
+}
+
+/** the known topic column names. */
+static const char* columnNames[] = {
+	"circuit",
+	"name",
+	"field",
+};
+
+/** the known topic column IDs. */
+static const size_t columnIds[] = {
+	COLUMN_CIRCUIT,
+	COLUMN_NAME,
+	COLUMN_FIELDS,
+};
+
+/** the number of known column names. */
+static const size_t columnCount = sizeof(columnNames) / sizeof(char*);
+
+
+/**
+ * Parse the topic template.
+ * @param topic the topic template.
+ * @param strs the @a vector to which the string parts shall be added.
+ * @param cols the @a vector to which the column parts shall be added.
+ * @return true on success, false on malformed topic template.
+ */
+bool parseTopic(const string topic, vector<string> &strs, vector<size_t> &cols) {
+	size_t lastpos = 0;
+	size_t end = topic.length();
+	vector<string> columns;
+	for (size_t pos=topic.find('$', lastpos); pos!=string::npos; ) {
+		size_t col = columnCount;
+		size_t len = 0;
+		for (size_t i = 0; i < columnCount; i++) {
+			len = strlen(columnNames[i]);
+			if (topic.substr(pos+1, len)==columnNames[i]) {
+				col = columnIds[i];
+				break;
+			}
+		}
+		if (col==columnCount) {
+			return false;
+		}
+		strs.push_back(topic.substr(lastpos, pos));
+		cols.push_back(col);
+		lastpos = pos+1+len;
+	}
+	if (lastpos<end) {
+		strs.push_back(topic.substr(lastpos, end));
+	}
+	if (strs.empty()) {
+		strs.push_back("");
+	}
+	return true;
+}
+
+
+MqttHandler::MqttHandler(BusHandler* busHandler)
+	: DataSink(), DataSource(busHandler), Thread()
+{
+	bool enabled = g_port!=0;
+	m_publishByField = false;
+	m_mosquitto = NULL;
+	if (enabled && !parseTopic(g_topic, m_topicStrs, m_topicCols)) {
+		logOtherError("mqtt", "malformed topic %s", g_topic);
+		enabled = false;
+	}
+	if (enabled) {
+		if (m_topicCols.empty()) {
+			if (m_topicStrs.empty()) {
+				m_topicStrs.push_back("");
+			} else {
+				string str = m_topicStrs[0];
+				if (str.empty() || str[str.length()-1]!='/') {
+					m_topicStrs[0] = str+"/";
+				}
+			}
+			m_topicCols.push_back(COLUMN_CIRCUIT); // circuit
+			m_topicStrs.push_back("/");
+			m_topicCols.push_back(COLUMN_NAME); // name
+		} else {
+			for (size_t i=0; i<m_topicCols.size(); i++) {
+				if (m_topicCols[i]==COLUMN_FIELDS) { // fields
+					m_publishByField = true;
+					break;
+				}
+			}
+		}
+		m_globalTopic = getTopic(NULL)+"global/";
+		m_mosquitto = NULL;
+		if (mosquitto_lib_init()!=MOSQ_ERR_SUCCESS) {
+			logOtherError("mqtt", "unable to initialize");
+		} else {
+			string clientId = PACKAGE_STRING;
+			clientId += " "+static_cast<unsigned>(getpid());
+			m_mosquitto = mosquitto_new(clientId.c_str(), NULL);
+			if (!m_mosquitto) {
+				logOtherError("mqtt", "unable to instantiate");
+			}
+		}
+		if (m_mosquitto) {
+			/*mosquitto_log_init(m_mosquitto, MOSQ_LOG_DEBUG | MOSQ_LOG_ERR | MOSQ_LOG_WARNING
+								  | MOSQ_LOG_NOTICE | MOSQ_LOG_INFO, MOSQ_LOG_STDERR);*/
+			string willTopic = m_globalTopic+"running";
+			string willData = "false";
+			size_t len = willData.length();
+			mosquitto_will_set(m_mosquitto, true, willTopic.c_str(), (uint32_t)len, (uint8_t*)(willData.c_str()), 0, true);
+			if (mosquitto_connect(m_mosquitto, "localhost", g_port, 60, true)!=MOSQ_ERR_SUCCESS) {
+				logOtherError("mqtt", "unable to connect");
+				mosquitto_destroy(m_mosquitto);
+				m_mosquitto = NULL;
+			} else {
+				logOtherNotice("mqtt", "connection established");
+			}
+		}
+	}
+}
+
+MqttHandler::~MqttHandler()
+{
+	join();
+	if (m_mosquitto) {
+		mosquitto_destroy(m_mosquitto);
+		m_mosquitto = NULL;
+	}
+	mosquitto_lib_cleanup();
+}
+
+void MqttHandler::start()
+{
+	if (m_mosquitto) {
+		Thread::start("MQTT");
+	}
+}
+
+void MqttHandler::run()
+{
+	time_t lastTaskRun, now, start, lastSignal = 0;
+	bool signal = false;
+	string signalTopic = m_globalTopic+"signal";
+	string uptimeTopic = m_globalTopic+"uptime";
+	ostringstream updates;
+
+	time(&now);
+	start = lastTaskRun = now;
+	publishTopic(m_globalTopic+"version", PACKAGE_STRING "." REVISION);
+	publishTopic(m_globalTopic+"running", "true");
+	publishTopic(signalTopic, "false");
+	while (isRunning()) {
+		handleTraffic();
+		time(&now);
+		if (now<start) {
+			// clock skew
+			if (now<lastSignal) {
+				lastSignal -= lastTaskRun-now;
+			}
+			lastTaskRun = now;
+		} else if (now > lastTaskRun+15) {
+			if (m_busHandler->hasSignal()) {
+				lastSignal = now;
+				if (!signal) {
+					signal = true;
+					publishTopic(signalTopic, "true");
+				}
+			} else {
+				if (signal) {
+					signal = false;
+					publishTopic(signalTopic, "false");
+				}
+			}
+			time_t uptime = now-start;
+			updates.str("");
+			updates.clear();
+			updates << dec << static_cast<unsigned>(uptime);
+			publishTopic(uptimeTopic, updates.str());
+			time(&lastTaskRun);
+		}
+		if (!m_updatedMessages.empty()) {
+			for (map<Message*, int>::iterator it = m_updatedMessages.begin(); it != m_updatedMessages.end(); it++) {
+				Message* message = it->first;
+				updates.str("");
+				updates.clear();
+				updates << dec;
+				if (message->decodeLastData(updates)==RESULT_OK) {
+					if (m_publishByField) {
+						//message->hasField()
+					} else {
+						string topic = getTopic(message);
+						string data = updates.str();
+						publishTopic(topic, data);
+					}
+				}
+			}
+			m_updatedMessages.clear();
+		}
+	}
+}
+
+void MqttHandler::handleTraffic()
+{
+	if (m_mosquitto) {
+		mosquitto_loop(m_mosquitto, 0);
+	}
+}
+
+string MqttHandler::getTopic(Message* message)
+{
+	ostringstream ret;
+	for (size_t i=0; i<m_topicStrs.size(); i++) {
+		ret << m_topicStrs[i];
+		if (!message) {
+			break;
+		}
+		if (i<m_topicCols.size()) {
+			 message->dumpColumn(ret, m_topicCols[i], false);
+		}
+	}
+	return ret.str();
+}
+
+void MqttHandler::publishTopic(string topic, string data, bool retain)
+{
+	logOtherDebug("mqtt", "publish %s %s", topic.c_str(), data.c_str());
+	mosquitto_publish(m_mosquitto, NULL, topic.c_str(), (uint32_t)data.size(), (uint8_t*)(data.c_str()), 0, retain);
+}

src/ebusd/mqtthandler.h

@@ -0,0 +1,111 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2016-2017 John Baier <ebusd@ebusd.eu>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef MQTTHANDLER_H_
+#define MQTTHANDLER_H_
+
+#include "datahandler.h"
+#include "bushandler.h"
+#include "message.h"
+#include <mosquitto.h>
+
+/** @file mqtthandler.h
+ * A data handler enabling MQTT support via mosquitto.
+ */
+
+using namespace std;
+
+/**
+ * Helper function for getting the argp definition for MQTT.
+ * @return a pointer to the argp_child structure.
+ */
+const struct argp_child* mqtthandler_getargs();
+
+/**
+ * Registration function that is called once during initialization.
+ * @param busHandler the @a BusHandler instance.
+ * @return the create @a DataHandler, or NULL on error.
+ */
+DataHandler* mqtthandler_register(BusHandler* busHandler);
+
+/**
+ * The main class supporting MQTT data handling.
+ */
+class MqttHandler : public DataSink, DataSource, Thread
+{
+public:
+
+	/**
+	 * Constructor.
+	 * @param busHandler the @a BusHandler instance.
+	 */
+	MqttHandler(BusHandler* busHandler);
+
+	/**
+	 * Destructor.
+	 */
+	virtual ~MqttHandler();
+
+	// @copydoc
+	virtual void start();
+
+protected:
+
+	// @copydoc
+	virtual void run();
+
+private:
+
+	/**
+	 * Called regularly to handle MQTT traffic.
+	 */
+	void handleTraffic();
+
+	/**
+	 * Build the MQTT topic string for the @a Message.
+	 * @param message the @a Message to build the topic string for.
+	 * @return the topic string.
+	 */
+	string getTopic(Message* message);
+
+	/**
+	 * Publish a topic update to MQTT.
+	 * @param topic the topic string.
+	 * @param data the data string.
+	 * @param retain whether the topic shall be retained.
+	 */
+	void publishTopic(string topic, string data, bool retain=true);
+
+	/** the MQTT topic string parts. */
+	vector<string> m_topicStrs;
+
+	/** the MQTT topic column parts. */
+	vector<size_t> m_topicCols;
+
+	/** the global topic prefix. */
+	string m_globalTopic;
+
+	/** whether to publish a separate topic for each message field. */
+	bool m_publishByField;
+
+	/** the mosquitto structure if initialized, or NULL. */
+	struct mosquitto* m_mosquitto;
+
+};
+
+#endif // DATAHANDLER_H_