diff --git a/configure.ac b/configure.ac index 876eb262..5763716a 100644 --- a/configure.ac +++ b/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= -AC_CHECK_LIB([rt], [clock_gettime], [RT_LIB="-lrt"]) -AC_SUBST(RT_LIB) +EXTRA_LIBS= +AC_CHECK_LIB([rt], [clock_gettime], [EXTRA_LIBS+="-lrt"]) +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( diff --git a/src/ebusd/Makefile.am b/src/ebusd/Makefile.am index d6b4395a..c2a21d74 100644 --- a/src/ebusd/Makefile.am +++ b/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 diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp new file mode 100644 index 00000000..8a0258b9 --- /dev/null +++ b/src/ebusd/mqtthandler.cpp @@ -0,0 +1,308 @@ +/* + * ebusd - daemon for communication with eBUS heating systems. + * Copyright (C) 2016-2017 John Baier + * + * 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 . + */ + +#include "mqtthandler.h" +#include "log.h" +#ifdef HAVE_CONFIG_H +# include +#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 &strs, vector &cols) { + size_t lastpos = 0; + size_t end = topic.length(); + vector 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(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 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(uptime); + publishTopic(uptimeTopic, updates.str()); + time(&lastTaskRun); + } + if (!m_updatedMessages.empty()) { + for (map::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; idumpColumn(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); +} diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h new file mode 100644 index 00000000..96ff19c1 --- /dev/null +++ b/src/ebusd/mqtthandler.h @@ -0,0 +1,111 @@ +/* + * ebusd - daemon for communication with eBUS heating systems. + * Copyright (C) 2016-2017 John Baier + * + * 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 . + */ + +#ifndef MQTTHANDLER_H_ +#define MQTTHANDLER_H_ + +#include "datahandler.h" +#include "bushandler.h" +#include "message.h" +#include + +/** @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 m_topicStrs; + + /** the MQTT topic column parts. */ + vector 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_