diff --git a/configure.ac b/configure.ac index 999f6ba9..ecfc6b0a 100755 --- a/configure.ac +++ b/configure.ac @@ -52,8 +52,7 @@ AM_CONDITIONAL([WITH_EBUSFEED], [test "x$with_ebusfeed" == "xyes"]) 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], [Defined if MQTT handling is enabled.]) - EXTRA_LIBS+=" -lmosquitto"], + [AC_DEFINE_UNQUOTED(HAVE_MQTT, [1], [Defined if MQTT handling is enabled.])], [AC_MSG_RESULT([Could not find mosquitto_lib_init in libmosquitto.]) with_mqtt="no"]) fi diff --git a/src/ebusd/CMakeLists.txt b/src/ebusd/CMakeLists.txt index db96bcb4..ae34c07f 100644 --- a/src/ebusd/CMakeLists.txt +++ b/src/ebusd/CMakeLists.txt @@ -11,7 +11,8 @@ set(ebusd_SOURCES ) if(HAVE_MQTT) - set(ebusd_SOURCES ${ebusd_SOURCES} mqtthandler.cpp mqtthandler.h) + set(ebusd_SOURCES ${ebusd_SOURCES} mqtthandler.cpp mqtthandler.h mqttclient.cpp mqttclient.h) + set(ebusd_SOURCES ${ebusd_SOURCES} mqttclient_mosquitto.cpp mqttclient_mosquitto.h) set(ebusd_LIBS ${ebusd_LIBS} mosquitto) endif(HAVE_MQTT) diff --git a/src/ebusd/Makefile.am b/src/ebusd/Makefile.am index ee169cf6..174750a5 100644 --- a/src/ebusd/Makefile.am +++ b/src/ebusd/Makefile.am @@ -12,15 +12,19 @@ ebusd_SOURCES = \ mainloop.h mainloop.cpp \ scan.h scan.cpp \ main.h main.cpp main_args.cpp -if MQTT -ebusd_SOURCES += mqtthandler.cpp mqtthandler.h -endif + ebusd_LDADD = ../lib/utils/libutils.a \ ../lib/ebus/libebus.a \ -lpthread \ @EXTRA_LIBS@ +if MQTT +ebusd_SOURCES += mqtthandler.cpp mqtthandler.h mqttclient.cpp mqttclient.h +ebusd_SOURCES += mqttclient_mosquitto.cpp mqttclient_mosquitto.h +ebusd_LDADD += -lmosquitto +endif + if KNX ebusd_SOURCES += knxhandler.cpp knxhandler.h ebusd_LDADD += ../lib/knx/libknx.a diff --git a/src/ebusd/mqttclient.cpp b/src/ebusd/mqttclient.cpp new file mode 100644 index 00000000..d752c3fb --- /dev/null +++ b/src/ebusd/mqttclient.cpp @@ -0,0 +1,34 @@ +/* + * ebusd - daemon for communication with eBUS heating systems. + * Copyright (C) 2023 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 . + */ + +#ifdef HAVE_CONFIG_H +# include +#endif + +#include +#include "ebusd/mqttclient.h" +#include "ebusd/mqttclient_mosquitto.h" + +namespace ebusd { + +// copydoc +MqttClient* MqttClient::create(mqtt_client_config_t config, MqttClientListener *listener) { + return new MqttClientMosquitto(config, listener); +} + +} // namespace ebusd diff --git a/src/ebusd/mqttclient.h b/src/ebusd/mqttclient.h new file mode 100755 index 00000000..febaa048 --- /dev/null +++ b/src/ebusd/mqttclient.h @@ -0,0 +1,157 @@ +/* + * ebusd - daemon for communication with eBUS heating systems. + * Copyright (C) 2023 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 EBUSD_MQTTCLIENT_H_ +#define EBUSD_MQTTCLIENT_H_ + +#include +#include +#include +#include +#include + +namespace ebusd { + +/** \file ebusd/mqttclient.h + * An abstraction for an MQTT client. + */ + +using std::map; +using std::pair; +using std::string; +using std::vector; + +/** settings for the connection to an MQTT broker. */ +typedef struct mqtt_client_config { + const char* host; //!< host name or IP address of MQTT broker + uint16_t port; //!< optional port of MQTT broker + const char* clientId; //!< optional clientid override for MQTT broker + const char* username; //!< optional user name for MQTT broker + const char* password; //!< optional password for MQTT broker + bool logEvents; //!< whether to log library events + bool version311; //!< true to use protocol version 3.1.1 + bool ignoreInvalidParams; //!< ignore invalid parameters during init + const char* cafile; //!< optional CA file for TLS + const char* capath; //!< optional CA path for TLS + const char* certfile; //!< optional client certificate file for TLS + const char* keyfile; //!< optional client key file for TLS + const char* keypass; //!< optional client key file password for TLS + bool insecure; //!< whether to allow insecure TLS connection + const char* lastWillTopic; //!< optional last will topic. + const char* lastWillData; //!< optional last will data. +} mqtt_client_config_t; + + +/** + * Interface for listening to MQTT client events. + */ +class MqttClientListener { + public: + /** + * Destructor. + */ + virtual ~MqttClientListener() {} + + /** + * Notification of status of connection to the broker. + */ + virtual void notifyMqttStatus(bool connected) = 0; // abstract + + /** + * Notification of a received MQTT message. + * @param topic the topic string. + * @param data the data string. + */ + virtual void notifyMqttTopic(const string& topic, const string& data) = 0; // abstract +}; + + +/** + * An abstract MQTT client. + */ +class MqttClient { + public: + /** + * Constructor. + * @param config the client configuration to use. + * @param listener the client listener to use. + */ + MqttClient(mqtt_client_config_t config, MqttClientListener *listener) + : m_config(config), m_listener(listener) {} + + /** + * Destructor. + */ + virtual ~MqttClient() {} + + /** + * Create a new instance. + * @param config the client configuration to use. + * @param listener the client listener to use. + * @return the new MqttClient, or @a nullptr on error. + */ + static MqttClient* create(mqtt_client_config_t config, MqttClientListener *listener); + + /** + * Connect to the broker and start handling MQTT traffic. + * @param isAsync set to true if the asynchronous client was started and @a run() does not + * have to be called at all, false if the client is synchronous and does + * it's work in @a run() only. + * @param connected set to true if the connection was already established. + * @return true on success, false if connection failed and the client is no longer usable (i.e. should be destroyed). + */ + virtual bool connect(bool &isAsync, bool &connected) = 0; // abstract + + /** + * Called regularly to handle MQTT traffic. + * @param allowReconnect true when reconnecting to the broker is allowed. + * @return true on error for waiting a bit until next call, or false otherwise. + */ + virtual bool run(bool allowReconnect, bool &connected) = 0; // abstract + + /** + * Publish a topic update. + * @param topic the topic string. + * @param data the data string. + * @param retain whether the topic shall be retained. + */ + virtual void publishTopic(const string& topic, const string& data, int qos, bool retain = false) = 0; // abstract + + /** + * Publish a topic update without any data. + * @param topic the topic string. + */ + virtual void publishEmptyTopic(const string& topic, int qos, bool retain = false) = 0; // abstract + + /** + * Subscribe to the specified topic pattern. + * @param topic the topic pattern string to subscribe to. + */ + virtual void subscribeTopic(const string& topic) = 0; // abstract + + public: + /** the client configuration to use. */ + mqtt_client_config_t m_config; + + /** the @a MqttClientListener instance. */ + MqttClientListener* m_listener; +}; + +} // namespace ebusd + +#endif // EBUSD_MQTTCLIENT_H_ diff --git a/src/ebusd/mqttclient_mosquitto.cpp b/src/ebusd/mqttclient_mosquitto.cpp new file mode 100755 index 00000000..93d80ce4 --- /dev/null +++ b/src/ebusd/mqttclient_mosquitto.cpp @@ -0,0 +1,321 @@ +/* + * ebusd - daemon for communication with eBUS heating systems. + * Copyright (C) 2023 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 . + */ + +#ifdef HAVE_CONFIG_H +# include +#endif + +#include "ebusd/mqttclient.h" +#include "ebusd/mqttclient_mosquitto.h" +#include +#include +#include +#include +#include "lib/utils/log.h" +#include "lib/ebus/symbol.h" + +namespace ebusd { + +using std::dec; + + +bool check(int code, const char* method) { + if (code == MOSQ_ERR_SUCCESS) { + return true; + } + if (code == MOSQ_ERR_ERRNO) { + char* error = strerror(errno); + logOtherError("mqtt", "%s: errno %d=%s", method, errno, error); + return false; + } +#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) + const char* msg = mosquitto_strerror(code); + logOtherError("mqtt", "%s: %s", method, msg); +#else + logOtherError("mqtt", "%s: error code %d", method, code); +#endif + return false; +} + + + +#if (LIBMOSQUITTO_MAJOR >= 1) +int on_keypassword(char *buf, int size, int rwflag, void *userdata) { + MqttClientMosquitto* client = reinterpret_cast(userdata); + if (!client || !client->m_config.keypass) { + return 0; + } + int len = static_cast(strlen(client->m_config.keypass)); + if (len > size) { + len = size; + } + memcpy(buf, client->m_config.keypass, len); + return len; +} +#endif + +void on_connect( +#if (LIBMOSQUITTO_MAJOR >= 1) + struct mosquitto *mosq, +#endif + void *obj, int rc) { + if (rc == 0) { + logOtherNotice("mqtt", "connection established"); + MqttClientMosquitto* client = reinterpret_cast(obj); + if (client) { + client->m_listener->notifyMqttStatus(true); + } + } else { + if (rc >= 1 && rc <= 3) { + logOtherError("mqtt", "connection refused: %s", + rc == 1 ? "wrong protocol" : (rc == 2 ? "wrong username/password" : "broker down")); + } else { + logOtherError("mqtt", "connection refused: %d", rc); + } + } +} + +#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) +void on_log(struct mosquitto *mosq, void *obj, int level, const char* msg) { + switch (level) { + case MOSQ_LOG_DEBUG: + logOtherDebug("mqtt", "log %s", msg); + break; + case MOSQ_LOG_INFO: + logOtherInfo("mqtt", "log %s", msg); + break; + case MOSQ_LOG_NOTICE: + logOtherNotice("mqtt", "log %s", msg); + break; + case MOSQ_LOG_WARNING: + logOtherNotice("mqtt", "log warning %s", msg); + break; + case MOSQ_LOG_ERR: + logOtherError("mqtt", "log %s", msg); + break; + default: + logOtherError("mqtt", "log other %s", msg); + break; + } +} +#endif + +void on_message( +#if (LIBMOSQUITTO_MAJOR >= 1) + struct mosquitto *mosq, +#endif + void *obj, const struct mosquitto_message *message) { + MqttClientMosquitto* client = reinterpret_cast(obj); + if (!client || !message) { + return; + } + string topic(message->topic); + string data(message->payloadlen > 0 ? reinterpret_cast(message->payload) : ""); + client->m_listener->notifyMqttTopic(topic, data); +} + +MqttClientMosquitto::MqttClientMosquitto(mqtt_client_config_t config, MqttClientListener *listener) + : MqttClient(config, listener), + m_mosquitto(nullptr), + m_initialConnectFailed(false), + m_lastErrorLogTime(0) { + int major = -1; + int minor = -1; + int revision = -1; + mosquitto_lib_version(&major, &minor, &revision); + if (major < LIBMOSQUITTO_MAJOR) { + logOtherError("mqtt", "invalid mosquitto version %d instead of %d, will try connecting anyway", major, + LIBMOSQUITTO_MAJOR); + } + logOtherInfo("mqtt", "mosquitto version %d.%d.%d (compiled with %d.%d.%d)", major, minor, revision, + LIBMOSQUITTO_MAJOR, LIBMOSQUITTO_MINOR, LIBMOSQUITTO_REVISION); + if (check(mosquitto_lib_init(), "unable to initialize")) { + signal(SIGPIPE, SIG_IGN); // needed before libmosquitto v. 1.1.3 +#if (LIBMOSQUITTO_MAJOR >= 1) + m_mosquitto = mosquitto_new(config.clientId, true, this); +#else + m_mosquitto = mosquitto_new(config.clientId, this); +#endif + if (!m_mosquitto) { + logOtherError("mqtt", "unable to instantiate"); + } + } + if (m_mosquitto) { +#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001) + check(mosquitto_threaded_set(m_mosquitto, true), "threaded_set"); + int version = config.version311 ? MQTT_PROTOCOL_V311 : MQTT_PROTOCOL_V31; + check(mosquitto_opts_set(m_mosquitto, MOSQ_OPT_PROTOCOL_VERSION, reinterpret_cast(&version)), + "opts_set protocol version"); +#else + if (config.version311) { + logOtherError("mqtt", "version 3.1.1 not supported"); + } +#endif + if (config.username || config.password) { + if (mosquitto_username_pw_set(m_mosquitto, config.username, config.password) != MOSQ_ERR_SUCCESS) { + logOtherError("mqtt", "unable to set username/password, trying without"); + } + } + if (config.lastWillTopic) { + size_t len = config.lastWillData ? strlen(config.lastWillData) : 0; +#if (LIBMOSQUITTO_MAJOR >= 1) + mosquitto_will_set(m_mosquitto, config.lastWillTopic, (uint32_t)len, + reinterpret_cast(config.lastWillData), 0, true); +#else + mosquitto_will_set(m_mosquitto, true, config.lastWillTopic, (uint32_t)len, + reinterpret_cast(config.lastWillData), 0, true); +#endif + } + + if (config.cafile || config.capath) { +#if (LIBMOSQUITTO_MAJOR >= 1) + mosquitto_user_data_set(m_mosquitto, this); + int ret; + ret = mosquitto_tls_set(m_mosquitto, config.cafile, config.capath, config.certfile, config.keyfile, + on_keypassword); + if (ret != MOSQ_ERR_SUCCESS) { + logOtherError("mqtt", "unable to set TLS: %d", ret); + } else if (config.insecure) { + ret = mosquitto_tls_insecure_set(m_mosquitto, true); + if (ret != MOSQ_ERR_SUCCESS) { + logOtherError("mqtt", "unable to set TLS insecure: %d", ret); + } + } +#else + logOtherError("mqtt", "use of TLS not supported"); +#endif + } + if (config.logEvents) { +#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) + mosquitto_log_callback_set(m_mosquitto, on_log); +#else + logOtherError("mqtt", "logging of library events not supported"); +#endif + } + mosquitto_connect_callback_set(m_mosquitto, on_connect); + // mosquitto_disconnect_callback_set(m_mosquitto, on_disconnect); + mosquitto_message_callback_set(m_mosquitto, on_message); + } +} + +bool MqttClientMosquitto::connect(bool &isAsync, bool &connected) { + isAsync = false; + if (!m_mosquitto) { + connected = false; + return false; + } + int ret; +#if (LIBMOSQUITTO_MAJOR >= 1) + ret = mosquitto_connect(m_mosquitto, m_config.host, m_config.port, 60); +#else + ret = mosquitto_connect(m_mosquitto, config.host, config.port, 60, true); +#endif + if (ret == MOSQ_ERR_INVAL && !m_config.ignoreInvalidParams) { + logOtherError("mqtt", "unable to connect (invalid parameters)"); + mosquitto_destroy(m_mosquitto); + m_mosquitto = nullptr; + connected = false; + return false; // never try again + } + if (!check(ret, "unable to connect, retrying")) { + connected = false; + m_initialConnectFailed = m_config.ignoreInvalidParams; + return true; + } + connected = true; // assume success until connect_callback says otherwise + logOtherDebug("mqtt", "connection requested"); + return true; +} + +MqttClientMosquitto::~MqttClientMosquitto() { + if (m_mosquitto) { + mosquitto_destroy(m_mosquitto); + m_mosquitto = nullptr; + } + mosquitto_lib_cleanup(); +} + +bool MqttClientMosquitto::run(bool allowReconnect, bool &connected) { + if (!m_mosquitto) { + return false; + } + int ret; +#if (LIBMOSQUITTO_MAJOR >= 1) + ret = mosquitto_loop(m_mosquitto, -1, 1); // waits up to 1 second for network traffic +#else + ret = mosquitto_loop(m_mosquitto, -1); // waits up to 1 second for network traffic +#endif + if (!connected && (ret == MOSQ_ERR_NO_CONN || ret == MOSQ_ERR_CONN_LOST) && allowReconnect) { + if (m_initialConnectFailed) { +#if (LIBMOSQUITTO_MAJOR >= 1) + ret = mosquitto_connect(m_mosquitto, m_config.host, m_config.port, 60); +#else + ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60, true); +#endif + if (ret == MOSQ_ERR_INVAL) { + logOtherError("mqtt", "unable to connect (invalid parameters), retrying"); + } + if (ret == MOSQ_ERR_SUCCESS) { + m_initialConnectFailed = false; + } + } else { + ret = mosquitto_reconnect(m_mosquitto); + } + } + if (!connected && ret == MOSQ_ERR_SUCCESS) { + connected = true; + logOtherNotice("mqtt", "connection re-established"); + } + if (!connected || ret == MOSQ_ERR_SUCCESS) { + return false; + } + if (ret == MOSQ_ERR_NO_CONN || ret == MOSQ_ERR_CONN_LOST || ret == MOSQ_ERR_CONN_REFUSED) { + logOtherError("mqtt", "communication error: %s", ret == MOSQ_ERR_NO_CONN ? "not connected" + : (ret == MOSQ_ERR_CONN_LOST ? "connection lost" : "connection refused")); + connected = false; + } else { + time_t now; + time(&now); + if (now > m_lastErrorLogTime + 10) { // log at most every 10 seconds + m_lastErrorLogTime = now; + check(ret, "communication error"); + } + } + return true; +} + +void MqttClientMosquitto::publishTopic(const string& topic, const string& data, int qos, bool retain) { + const char* topicStr = topic.c_str(); + const char* dataStr = data.c_str(); + const size_t len = strlen(dataStr); + logOtherDebug("mqtt", "publish %s %s", topicStr, dataStr); + check(mosquitto_publish(m_mosquitto, nullptr, topicStr, (uint32_t)len, + reinterpret_cast(dataStr), qos, retain), "publish"); +} + +void MqttClientMosquitto::publishEmptyTopic(const string& topic, int qos, bool retain) { + const char* topicStr = topic.c_str(); + logOtherDebug("mqtt", "publish empty %s", topicStr); + check(mosquitto_publish(m_mosquitto, nullptr, topicStr, 0, nullptr, qos, retain), "publish empty"); +} + +void MqttClientMosquitto::subscribeTopic(const string& topic) { + check(mosquitto_subscribe(m_mosquitto, nullptr, topic.c_str(), 0), "subscribe"); +} + +} // namespace ebusd diff --git a/src/ebusd/mqttclient_mosquitto.h b/src/ebusd/mqttclient_mosquitto.h new file mode 100755 index 00000000..d749b60b --- /dev/null +++ b/src/ebusd/mqttclient_mosquitto.h @@ -0,0 +1,80 @@ +/* + * ebusd - daemon for communication with eBUS heating systems. + * Copyright (C) 2023 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 EBUSD_MQTTCLIENT_MOSQUITTO_H_ +#define EBUSD_MQTTCLIENT_MOSQUITTO_H_ + +#include "ebusd/mqttclient.h" +#include +#include +#include +#include +#include +#include + +namespace ebusd { + +/** \file ebusd/mqttclient.h + * An abstraction for an MQTT client. + */ + +using std::map; +using std::pair; +using std::string; +using std::vector; + +class MqttClientMosquitto : public MqttClient { + public: + /** + * Constructor. + * @param config the client configuration to use. + * @param listener the client listener to use. + */ + MqttClientMosquitto(mqtt_client_config_t config, MqttClientListener *listener); + + virtual ~MqttClientMosquitto(); + + // @copydoc + bool connect(bool &isAsync, bool &connected) override; + + // @copydoc + bool run(bool allowReconnect, bool &connected) override; + + // @copydoc + void publishTopic(const string& topic, const string& data, int qos, bool retain = false) override; + + // @copydoc + void publishEmptyTopic(const string& topic, int qos, bool retain = false) override; + + // @copydoc + void subscribeTopic(const string& topic) override; + + private: + /** the mosquitto structure if initialized, or nullptr. */ + struct mosquitto* m_mosquitto; + + /** whether the initial connect failed. */ + bool m_initialConnectFailed; + + /** the last system time when a communication error was logged. */ + time_t m_lastErrorLogTime; +}; + +} // namespace ebusd + +#endif // EBUSD_MQTTCLIENT_MOSQUITTO_H_ diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index f2e75b7b..54b4a760 100755 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -75,32 +75,40 @@ static const argDef g_mqtt_argDefs[] = { {"mqttjson", O_JSON, "short", af_optional, "Publish in JSON format instead of strings, optionally in short (value directly below field key)"}, {"mqttverbose", O_VERB, nullptr, 0, "Publish all available attributes"}, -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) {"mqttlog", O_LOGL, nullptr, 0, "Log library events"}, -#endif -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001) {"mqttversion", O_VERS, "VERSION", 0, "Use protocol VERSION [3.1]"}, -#endif {"mqttignoreinvalid", O_IGIN, nullptr, 0, "Ignore invalid parameters during init (e.g. for DNS not resolvable yet)"}, {"mqttchanges", O_CHGS, nullptr, 0, "Whether to only publish changed messages instead of all received"}, -#if (LIBMOSQUITTO_MAJOR >= 1) {"mqttca", O_CAFI, "CA", 0, "Use CA file or dir (ending with '/') for MQTT TLS (no default)"}, {"mqttcert", O_CERT, "CERTFILE", 0, "Use CERTFILE for MQTT TLS client certificate (no default)"}, {"mqttkey", O_KEYF, "KEYFILE", 0, "Use KEYFILE for MQTT TLS client certificate (no default)"}, {"mqttkeypass", O_KEPA, "PASSWORD", 0, "Use PASSWORD for the encrypted KEYFILE (no default)"}, {"mqttinsecure", O_INSE, nullptr, 0, "Allow insecure TLS connection (e.g. using a self signed certificate)"}, -#endif {nullptr, 0, nullptr, 0, nullptr}, }; -static const char* g_host = "localhost"; //!< host name of MQTT broker [localhost] -static uint16_t g_port = 0; //!< optional port of MQTT broker, 0 to disable [0] -static const char* g_clientId = nullptr; //!< optional clientid override for MQTT broker -static const char* g_username = nullptr; //!< optional user name for MQTT broker (no default) -static const char* g_password = nullptr; //!< optional password for MQTT broker (no default) +// options for the MQTT client +static mqtt_client_config_t g_opt = { + .host = "localhost", + .port = 0, + .clientId = nullptr, + .username = nullptr, + .password = nullptr, + .logEvents = false, + .version311 = false, + .ignoreInvalidParams = false, + .cafile = nullptr, + .capath = nullptr, + .certfile = nullptr, + .keyfile = nullptr, + .keypass = nullptr, + .insecure = false, + .lastWillTopic = nullptr, + .lastWillData = nullptr, +}; static const char* g_topic = nullptr; //!< optional topic template static const char* g_globalTopic = nullptr; //!< optional global topic static const char* g_integrationFile = nullptr; //!< the integration settings file @@ -108,24 +116,8 @@ static vector* g_integrationVars = nullptr; //!< the integration settin static bool g_retain = false; //!< whether to retail all topics static int g_qos = 0; //!< the qos value for all topics static OutputFormat g_publishFormat = OF_NONE; //!< the OutputFormat for publishing messages -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) -static bool g_logFromLib = false; //!< log library events -#endif -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001) -static int g_version = MQTT_PROTOCOL_V31; //!< protocol version to use -#endif -static bool g_ignoreInvalidParams = false; //!< ignore invalid parameters during init static bool g_onlyChanges = false; //!< whether to only publish changed messages instead of all received -#if (LIBMOSQUITTO_MAJOR >= 1) -static const char* g_cafile = nullptr; //!< CA file for TLS -static const char* g_capath = nullptr; //!< CA path for TLS -static const char* g_certfile = nullptr; //!< client certificate file for TLS -static const char* g_keyfile = nullptr; //!< client key file for TLS -static const char* g_keypass = nullptr; //!< client key file password for TLS -static bool g_insecure = false; //!< whether to allow insecure TLS connection -#endif - /** * Replace all characters in the string with a space and return a copy of the original string. * @param arg the string to replace. @@ -157,7 +149,7 @@ static int mqtt_parse_opt(int key, char *arg, const argParseOpt *parseOpt) { argParseError(parseOpt, "invalid mqtthost"); return EINVAL; } - g_host = arg; + g_opt.host = arg; break; case O_PORT: // --mqttport=1883 @@ -166,7 +158,7 @@ static int mqtt_parse_opt(int key, char *arg, const argParseOpt *parseOpt) { argParseError(parseOpt, "invalid mqttport"); return EINVAL; } - g_port = (uint16_t)value; + g_opt.port = (uint16_t)value; break; case O_CLID: // --mqttclientid=clientid @@ -174,7 +166,7 @@ static int mqtt_parse_opt(int key, char *arg, const argParseOpt *parseOpt) { argParseError(parseOpt, "invalid mqttclientid"); return EINVAL; } - g_clientId = arg; + g_opt.clientId = arg; break; case O_USER: // --mqttuser=username @@ -182,7 +174,7 @@ static int mqtt_parse_opt(int key, char *arg, const argParseOpt *parseOpt) { argParseError(parseOpt, "invalid mqttuser"); return EINVAL; } - g_username = arg; + g_opt.username = arg; break; case O_PASS: // --mqttpass=password @@ -190,7 +182,7 @@ static int mqtt_parse_opt(int key, char *arg, const argParseOpt *parseOpt) { argParseError(parseOpt, "invalid mqttpass"); return EINVAL; } - g_password = replaceSecret(arg); + g_opt.password = replaceSecret(arg); break; case O_TOPI: // --mqtttopic=ebusd @@ -268,72 +260,66 @@ static int mqtt_parse_opt(int key, char *arg, const argParseOpt *parseOpt) { g_publishFormat = (g_publishFormat & ~OF_SHORT) | OF_NAMES|OF_UNITS|OF_COMMENTS|OF_ALL_ATTRS; break; -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) case O_LOGL: - g_logFromLib = true; + g_opt.logEvents = true; break; -#endif -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001) case O_VERS: // --mqttversion=3.1.1 if (arg == nullptr || arg[0] == 0 || (strcmp(arg, "3.1") != 0 && strcmp(arg, "3.1.1") != 0)) { argParseError(parseOpt, "invalid mqttversion"); return EINVAL; } - g_version = strcmp(arg, "3.1.1") == 0 ? MQTT_PROTOCOL_V311 : MQTT_PROTOCOL_V31; + g_opt.version311 = strcmp(arg, "3.1.1") == 0; break; -#endif case O_IGIN: - g_ignoreInvalidParams = true; + g_opt.ignoreInvalidParams = true; break; case O_CHGS: g_onlyChanges = true; break; -#if (LIBMOSQUITTO_MAJOR >= 1) - case O_CAFI: // --mqttca=file or --mqttca=dir/ - if (arg == nullptr || arg[0] == 0) { - argParseError(parseOpt, "invalid mqttca"); - return EINVAL; - } - if (arg[strlen(arg)-1] == '/') { - g_cafile = nullptr; - g_capath = arg; - } else { - g_cafile = arg; - g_capath = nullptr; - } - break; + case O_CAFI: // --mqttca=file or --mqttca=dir/ + if (arg == nullptr || arg[0] == 0) { + argParseError(parseOpt, "invalid mqttca"); + return EINVAL; + } + if (arg[strlen(arg)-1] == '/') { + g_opt.cafile = nullptr; + g_opt.capath = arg; + } else { + g_opt.cafile = arg; + g_opt.capath = nullptr; + } + break; - case O_CERT: // --mqttcert=CERTFILE - if (arg == nullptr || arg[0] == 0) { - argParseError(parseOpt, "invalid mqttcert"); - return EINVAL; - } - g_certfile = arg; - break; + case O_CERT: // --mqttcert=CERTFILE + if (arg == nullptr || arg[0] == 0) { + argParseError(parseOpt, "invalid mqttcert"); + return EINVAL; + } + g_opt.certfile = arg; + break; - case O_KEYF: // --mqttkey=KEYFILE - if (arg == nullptr || arg[0] == 0) { - argParseError(parseOpt, "invalid mqttkey"); - return EINVAL; - } - g_keyfile = arg; - break; + case O_KEYF: // --mqttkey=KEYFILE + if (arg == nullptr || arg[0] == 0) { + argParseError(parseOpt, "invalid mqttkey"); + return EINVAL; + } + g_opt.keyfile = arg; + break; - case O_KEPA: // --mqttkeypass=PASSWORD - if (arg == nullptr) { - argParseError(parseOpt, "invalid mqttkeypass"); - return EINVAL; - } - g_keypass = replaceSecret(arg); - break; - case O_INSE: // --mqttinsecure - g_insecure = true; - break; -#endif + case O_KEPA: // --mqttkeypass=PASSWORD + if (arg == nullptr) { + argParseError(parseOpt, "invalid mqttkeypass"); + return EINVAL; + } + g_opt.keypass = replaceSecret(arg); + break; + case O_INSE: // --mqttinsecure + g_opt.insecure = true; + break; default: return EINVAL; @@ -350,117 +336,15 @@ const argParseChildOpt* mqtthandler_getargs() { return &g_mqtt_arg_child; } -bool check(int code, const char* method) { - if (code == MOSQ_ERR_SUCCESS) { - return true; - } - if (code == MOSQ_ERR_ERRNO) { - char* error = strerror(errno); - logOtherError("mqtt", "%s: errno %d=%s", method, errno, error); - return false; - } -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) - const char* msg = mosquitto_strerror(code); - logOtherError("mqtt", "%s: %s", method, msg); -#else - logOtherError("mqtt", "%s: error code %d", method, code); -#endif - return false; -} - bool mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages, list* handlers) { - if (g_port > 0) { - int major = -1; - int minor = -1; - int revision = -1; - mosquitto_lib_version(&major, &minor, &revision); - if (major < LIBMOSQUITTO_MAJOR) { - logOtherError("mqtt", "invalid mosquitto version %d instead of %d, will try connecting anyway", major, - LIBMOSQUITTO_MAJOR); - } - logOtherInfo("mqtt", "mosquitto version %d.%d.%d (compiled with %d.%d.%d)", major, minor, revision, - LIBMOSQUITTO_MAJOR, LIBMOSQUITTO_MINOR, LIBMOSQUITTO_REVISION); + if (g_opt.port > 0) { handlers->push_back(new MqttHandler(userInfo, busHandler, messages)); } return true; } -#if (LIBMOSQUITTO_MAJOR >= 1) -int on_keypassword(char *buf, int size, int rwflag, void *userdata) { - if (!g_keypass) { - return 0; - } - int len = static_cast(strlen(g_keypass)); - if (len > size) { - len = size; - } - memcpy(buf, g_keypass, len); - return len; -} -#endif - -void on_connect( -#if (LIBMOSQUITTO_MAJOR >= 1) - struct mosquitto *mosq, -#endif - void *obj, int rc) { - if (rc == 0) { - logOtherNotice("mqtt", "connection established"); - MqttHandler* handler = reinterpret_cast(obj); - if (handler) { - handler->notifyConnected(); - } - } else { - if (rc >= 1 && rc <= 3) { - logOtherError("mqtt", "connection refused: %s", - rc == 1 ? "wrong protocol" : (rc == 2 ? "wrong username/password" : "broker down")); - } else { - logOtherError("mqtt", "connection refused: %d", rc); - } - } -} - -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) -void on_log(struct mosquitto *mosq, void *obj, int level, const char* msg) { - switch (level) { - case MOSQ_LOG_DEBUG: - logOtherDebug("mqtt", "log %s", msg); - break; - case MOSQ_LOG_INFO: - logOtherInfo("mqtt", "log %s", msg); - break; - case MOSQ_LOG_NOTICE: - logOtherNotice("mqtt", "log %s", msg); - break; - case MOSQ_LOG_WARNING: - logOtherNotice("mqtt", "log warning %s", msg); - break; - case MOSQ_LOG_ERR: - logOtherError("mqtt", "log %s", msg); - break; - default: - logOtherError("mqtt", "log other %s", msg); - break; - } -} -#endif - -void on_message( -#if (LIBMOSQUITTO_MAJOR >= 1) - struct mosquitto *mosq, -#endif - void *obj, const struct mosquitto_message *message) { - MqttHandler* handler = reinterpret_cast(obj); - if (!handler || !message || !handler->isRunning()) { - return; - } - string topic(message->topic); - string data(message->payloadlen > 0 ? reinterpret_cast(message->payload) : ""); - handler->notifyTopic(topic, data); -} - /** * possible data type names. */ @@ -485,10 +369,9 @@ string removeTrailingNonTopicPart(const string& str) { MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages) : DataSink(userInfo, "mqtt"), DataSource(busHandler), WaitThread(), m_messages(messages), m_connected(false), - m_initialConnectFailed(false), m_lastUpdateCheckResult("."), m_lastScanStatus(SCAN_STATUS_NONE), - m_lastErrorLogTime(0) { + m_lastUpdateCheckResult("."), m_lastScanStatus(SCAN_STATUS_NONE) { m_definitionsSince = 0; - m_mosquitto = nullptr; + m_client = nullptr; bool hasIntegration = false; if (g_integrationFile != nullptr) { if (!m_replacers.parseFile(g_integrationFile)) { @@ -603,121 +486,60 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* m_globalTopic.compress(values); } m_subscribeTopic = getTopic(nullptr, "#"); - if (check(mosquitto_lib_init(), "unable to initialize")) { - signal(SIGPIPE, SIG_IGN); // needed before libmosquitto v. 1.1.3 + if (!g_opt.clientId) { ostringstream clientId; - if (g_clientId) { - clientId << g_clientId; - } else { - clientId << PACKAGE_NAME << '_' << PACKAGE_VERSION << '_' << static_cast(getpid()); - } -#if (LIBMOSQUITTO_MAJOR >= 1) - m_mosquitto = mosquitto_new(clientId.str().c_str(), true, this); -#else - m_mosquitto = mosquitto_new(clientId.str().c_str(), this); -#endif - if (!m_mosquitto) { - logOtherError("mqtt", "unable to instantiate"); - } + clientId << PACKAGE_NAME << '_' << PACKAGE_VERSION << '_' << static_cast(getpid()); + g_opt.clientId = strdup(clientId.str().c_str()); } - if (m_mosquitto) { -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001) - check(mosquitto_threaded_set(m_mosquitto, true), "threaded_set"); - check(mosquitto_opts_set(m_mosquitto, MOSQ_OPT_PROTOCOL_VERSION, reinterpret_cast(&g_version)), - "opts_set protocol version"); -#endif - if (g_username || g_password) { - if (!g_username) { - g_username = PACKAGE; - } - if (mosquitto_username_pw_set(m_mosquitto, g_username, g_password) != MOSQ_ERR_SUCCESS) { - logOtherError("mqtt", "unable to set username/password, trying without"); - } - } - string willTopic = m_globalTopic.get("", "running"); - string willData = "false"; - size_t len = willData.length(); -#if (LIBMOSQUITTO_MAJOR >= 1) - mosquitto_will_set(m_mosquitto, willTopic.c_str(), (uint32_t)len, - reinterpret_cast(willData.c_str()), 0, true); -#else - mosquitto_will_set(m_mosquitto, true, willTopic.c_str(), (uint32_t)len, - reinterpret_cast(willData.c_str()), 0, true); -#endif - -#if (LIBMOSQUITTO_MAJOR >= 1) - if (g_cafile || g_capath) { - int ret; - ret = mosquitto_tls_set(m_mosquitto, g_cafile, g_capath, g_certfile, g_keyfile, on_keypassword); - if (ret != MOSQ_ERR_SUCCESS) { - logOtherError("mqtt", "unable to set TLS: %d", ret); - } else if (g_insecure) { - ret = mosquitto_tls_insecure_set(m_mosquitto, true); - if (ret != MOSQ_ERR_SUCCESS) { - logOtherError("mqtt", "unable to set TLS insecure: %d", ret); - } - } - } -#endif -#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001) - if (g_logFromLib) { - mosquitto_log_callback_set(m_mosquitto, on_log); - } -#endif - mosquitto_connect_callback_set(m_mosquitto, on_connect); - mosquitto_message_callback_set(m_mosquitto, on_message); - int ret; -#if (LIBMOSQUITTO_MAJOR >= 1) - ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60); -#else - ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60, true); -#endif - if (ret == MOSQ_ERR_INVAL && !g_ignoreInvalidParams) { - logOtherError("mqtt", "unable to connect (invalid parameters)"); - mosquitto_destroy(m_mosquitto); - m_mosquitto = nullptr; - } else if (!check(ret, "unable to connect, retrying")) { - m_connected = false; - m_initialConnectFailed = g_ignoreInvalidParams; - } else { - m_connected = true; // assume success until connect_callback says otherwise - logOtherDebug("mqtt", "connection requested"); - } + if (g_opt.password && !g_opt.username) { + g_opt.username = PACKAGE; + } + string willTopic = m_globalTopic.get("", "running"); + if (!willTopic.empty()) { + g_opt.lastWillTopic = strdup(willTopic.c_str()); + g_opt.lastWillData = "false"; + } + m_client = MqttClient::create(g_opt, this); + m_isAsync = false; + bool ret = m_client->connect(m_isAsync, m_connected); + if (!ret) { + logOtherError("mqtt", "unable to connect (invalid parameters)"); + delete m_client; + m_client = nullptr; } } MqttHandler::~MqttHandler() { join(); - if (m_mosquitto) { - mosquitto_destroy(m_mosquitto); - m_mosquitto = nullptr; + if (m_client) { + delete m_client; + m_client = nullptr; } - mosquitto_lib_cleanup(); } void MqttHandler::startHandler() { - if (m_mosquitto) { + if (m_client) { WaitThread::start("MQTT"); } } -void MqttHandler::notifyConnected() { - if (m_mosquitto && isRunning()) { +void MqttHandler::notifyMqttStatus(bool connected) { + if (connected && m_client && isRunning()) { const string sep = (g_publishFormat & OF_JSON) ? "\"" : ""; if (m_globalTopic.has("name")) { - publishTopic(m_globalTopic.get("", "version"), sep + (PACKAGE_STRING "." REVISION) + sep, true); + m_client->publishTopic(m_globalTopic.get("", "version"), sep + (PACKAGE_STRING "." REVISION) + sep, true); } publishTopic(m_globalTopic.get("", "running"), "true", true); if (!m_staticTopic) { - check(mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0), "subscribe"); + m_client->subscribeTopic(m_subscribeTopic); if (!m_subscribeConfigRestartTopic.empty()) { - check(mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeConfigRestartTopic.c_str(), 0), "subscribe def."); + m_client->subscribeTopic(m_subscribeConfigRestartTopic); } } } } -void MqttHandler::notifyTopic(const string& topic, const string& data) { +void MqttHandler::notifyMqttTopic(const string& topic, const string& data) { size_t pos = topic.rfind('/'); if (pos == string::npos) { return; @@ -919,7 +741,7 @@ void MqttHandler::run() { bool allowReconnect = false; while (isRunning()) { bool wasConnected = m_connected; - bool needsWait = handleTraffic(allowReconnect); + bool needsWait = m_isAsync || handleTraffic(allowReconnect); bool reconnected = !wasConnected && m_connected; allowReconnect = false; time(&now); @@ -1307,52 +1129,10 @@ void MqttHandler::publishDefinition(const StringReplacers& values) { } bool MqttHandler::handleTraffic(bool allowReconnect) { - if (!m_mosquitto) { + if (!m_client) { return false; } - int ret; -#if (LIBMOSQUITTO_MAJOR >= 1) - ret = mosquitto_loop(m_mosquitto, -1, 1); // waits up to 1 second for network traffic -#else - ret = mosquitto_loop(m_mosquitto, -1); // waits up to 1 second for network traffic -#endif - if (!m_connected && (ret == MOSQ_ERR_NO_CONN || ret == MOSQ_ERR_CONN_LOST) && allowReconnect) { - if (m_initialConnectFailed) { -#if (LIBMOSQUITTO_MAJOR >= 1) - ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60); -#else - ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60, true); -#endif - if (ret == MOSQ_ERR_INVAL) { - logOtherError("mqtt", "unable to connect (invalid parameters), retrying"); - } - if (ret == MOSQ_ERR_SUCCESS) { - m_initialConnectFailed = false; - } - } else { - ret = mosquitto_reconnect(m_mosquitto); - } - } - if (!m_connected && ret == MOSQ_ERR_SUCCESS) { - m_connected = true; - logOtherNotice("mqtt", "connection re-established"); - } - if (!m_connected || ret == MOSQ_ERR_SUCCESS) { - return false; - } - if (ret == MOSQ_ERR_NO_CONN || ret == MOSQ_ERR_CONN_LOST || ret == MOSQ_ERR_CONN_REFUSED) { - logOtherError("mqtt", "communication error: %s", ret == MOSQ_ERR_NO_CONN ? "not connected" - : (ret == MOSQ_ERR_CONN_LOST ? "connection lost" : "connection refused")); - m_connected = false; - } else { - time_t now; - time(&now); - if (now > m_lastErrorLogTime + 10) { // log at most every 10 seconds - m_lastErrorLogTime = now; - check(ret, "communication error"); - } - } - return true; + return m_client->run(allowReconnect, m_connected); } string MqttHandler::getTopic(const Message* message, const string& suffix, const string& fieldName) { @@ -1423,16 +1203,14 @@ void MqttHandler::publishMessage(const Message* message, ostringstream* updates, void MqttHandler::publishTopic(const string& topic, const string& data, bool retain) { const char* topicStr = topic.c_str(); const char* dataStr = data.c_str(); - const size_t len = strlen(dataStr); logOtherDebug("mqtt", "publish %s %s", topicStr, dataStr); - check(mosquitto_publish(m_mosquitto, nullptr, topicStr, (uint32_t)len, - reinterpret_cast(dataStr), g_qos, g_retain || retain), "publish"); + m_client->publishTopic(topicStr, data, g_qos, g_retain || retain); } void MqttHandler::publishEmptyTopic(const string& topic) { const char* topicStr = topic.c_str(); logOtherDebug("mqtt", "publish empty %s", topicStr); - check(mosquitto_publish(m_mosquitto, nullptr, topicStr, 0, nullptr, 0, g_retain), "publish empty"); + m_client->publishEmptyTopic(topic, 0, g_retain); } } // namespace ebusd diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h old mode 100644 new mode 100755 index acef8dab..db20f6db --- a/src/ebusd/mqtthandler.h +++ b/src/ebusd/mqtthandler.h @@ -19,7 +19,6 @@ #ifndef EBUSD_MQTTHANDLER_H_ #define EBUSD_MQTTHANDLER_H_ -#include #include #include #include @@ -27,6 +26,7 @@ #include #include "ebusd/datahandler.h" #include "ebusd/bushandler.h" +#include "ebusd/mqttclient.h" #include "lib/ebus/message.h" #include "lib/ebus/stringhelper.h" #include "lib/utils/arg.h" @@ -63,7 +63,7 @@ bool mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap /** * The main class supporting MQTT data handling. */ -class MqttHandler : public DataSink, public DataSource, public WaitThread { +class MqttHandler : public DataSink, public DataSource, public WaitThread, public MqttClientListener { public: /** * Constructor. @@ -82,17 +82,11 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread { // @copydoc void startHandler() override; - /** - * Notify the handler of a (re-)established connection to the broker. - */ - void notifyConnected(); + // @copydoc + void notifyMqttStatus(bool connected) override; - /** - * Notify the handler of a received MQTT message. - * @param topic the topic string. - * @param data the data string. - */ - void notifyTopic(const string& topic, const string& data); + // @copydoc + void notifyMqttTopic(const string& topic, const string& data) override; // @copydoc void notifyUpdateCheckResult(const string& checkResult) override; @@ -199,23 +193,24 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread { /** the last system time when the message definitions were published. */ time_t m_definitionsSince; - /** the mosquitto structure if initialized, or nullptr. */ - struct mosquitto* m_mosquitto; + /** the @a MqttClient instance. */ + MqttClient* m_client; + + /** + * true if the client is asynchronous and its @a run() method does not + * have to be called at all, false if the client is synchronous and does + * it's work in its @a run() method only. + */ + bool m_isAsync; /** whether the connection to the broker is established. */ bool m_connected; - /** whether the initial connect failed. */ - bool m_initialConnectFailed; - /** the last update check result. */ string m_lastUpdateCheckResult; /** the last scan status. */ scanStatus_t m_lastScanStatus; - - /** the last system time when a communication error was logged. */ - time_t m_lastErrorLogTime; }; } // namespace ebusd