abstract MQTT client

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

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)
 

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

src/ebusd/mqttclient.cpp

@@ -0,0 +1,34 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2023 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/>.
+ */
+
+#ifdef HAVE_CONFIG_H
+#  include <config.h>
+#endif
+
+#include <string.h>
+#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

src/ebusd/mqttclient.h

@@ -0,0 +1,157 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2023 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 EBUSD_MQTTCLIENT_H_
+#define EBUSD_MQTTCLIENT_H_
+
+#include <list>
+#include <map>
+#include <string>
+#include <utility>
+#include <vector>
+
+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_

src/ebusd/mqttclient_mosquitto.cpp

@@ -0,0 +1,321 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2023 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/>.
+ */
+
+#ifdef HAVE_CONFIG_H
+#  include <config.h>
+#endif
+
+#include "ebusd/mqttclient.h"
+#include "ebusd/mqttclient_mosquitto.h"
+#include <csignal>
+#include <deque>
+#include <algorithm>
+#include <utility>
+#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<MqttClientMosquitto*>(userdata);
+  if (!client || !client->m_config.keypass) {
+    return 0;
+  }
+  int len = static_cast<int>(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<MqttClientMosquitto*>(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<MqttClientMosquitto*>(obj);
+  if (!client || !message) {
+    return;
+  }
+  string topic(message->topic);
+  string data(message->payloadlen > 0 ? reinterpret_cast<char*>(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<void*>(&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<const uint8_t*>(config.lastWillData), 0, true);
+#else
+      mosquitto_will_set(m_mosquitto, true, config.lastWillTopic, (uint32_t)len,
+        reinterpret_cast<const uint8_t*>(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<const uint8_t*>(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

src/ebusd/mqttclient_mosquitto.h

@@ -0,0 +1,80 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2023 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 EBUSD_MQTTCLIENT_MOSQUITTO_H_
+#define EBUSD_MQTTCLIENT_MOSQUITTO_H_
+
+#include "ebusd/mqttclient.h"
+#include <mosquitto.h>
+#include <list>
+#include <map>
+#include <string>
+#include <utility>
+#include <vector>
+
+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_

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<string>* 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<DataHandler*>* 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<int>(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<MqttHandler*>(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<MqttHandler*>(obj);
-  if (!handler || !message || !handler->isRunning()) {
-    return;
-  }
-  string topic(message->topic);
-  string data(message->payloadlen > 0 ? reinterpret_cast<char*>(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<unsigned>(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<unsigned>(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<void*>(&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<const uint8_t*>(willData.c_str()), 0, true);
-#else
-    mosquitto_will_set(m_mosquitto, true, willTopic.c_str(), (uint32_t)len,
-        reinterpret_cast<const uint8_t*>(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<const uint8_t*>(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

src/ebusd/mqtthandler.h

@@ -19,7 +19,6 @@
 #ifndef EBUSD_MQTTHANDLER_H_
 #define EBUSD_MQTTHANDLER_H_
 
-#include <mosquitto.h>
 #include <list>
 #include <map>
 #include <string>
@@ -27,6 +26,7 @@
 #include <vector>
 #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