MQTT client on libmosquitto with its own thread (iteration 2, step 2)

MqttClient is the interface the HA layer will use; MosquittoClient implements it with one std::thread for DNS, connect, network loop and reconnect with backoff (D45, amends D42: mosquitto_loop_start() blocks the caller on DNS and stops silently on TLS/protocol errors). tst_mqtt tests it against a local mosquitto: messages, retained messages, first connect and reconnect, last will, login. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-06 16:09:53 +02:00
parent e0229d6647
commit 96df60e6bd
8 changed files with 805 additions and 29 deletions
@@ -52,6 +52,8 @@ message(STATUS "Vulkan backend: ${_vulkan_ok}")
 # The app's own logic (no SteamVR, no rendering): linked by the app and its tests.
 qt_add_library(frameassistant_lib STATIC
    config/Config.h config/Config.cpp
+    mqtt/MqttClient.h
+    mqtt/MosquittoClient.h mqtt/MosquittoClient.cpp
 )
 target_include_directories(frameassistant_lib PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
 target_link_libraries(frameassistant_lib PUBLIC Qt6::Core PkgConfig::MOSQUITTO)
@@ -0,0 +1,248 @@
+#include "mqtt/MosquittoClient.h"
+
+#include <mosquitto.h>
+
+#include <QDebug>
+#include <algorithm>
+#include <cerrno>
+#include <chrono>
+#include <cstring>
+#include <utility>
+
+namespace {
+
+using Clock = std::chrono::steady_clock;
+
+// How long the network loop may still run after a stop request, to send the
+// DISCONNECT packet.
+constexpr auto kStopGrace = std::chrono::seconds(2);
+
+void initLibrary() {
+  static std::once_flag once;
+  std::call_once(once, [] { mosquitto_lib_init(); });
+}
+
+// A libmosquitto result as text; `err` is errno right after the call. In
+// English regardless of the locale, like all other log and UI messages.
+QString errorText(int rc, int err) {
+  if (rc == MOSQ_ERR_ERRNO) {
+    const char* text = strerrordesc_np(err);
+    return text ? QString::fromUtf8(text)
+                : QStringLiteral("System error %1").arg(err);
+  }
+  // libmosquitto loses getaddrinfo()'s code (errno is overwritten before it
+  // returns), so there is no reason to give.
+  if (rc == MOSQ_ERR_EAI)
+    return QStringLiteral("Cannot resolve the broker's host name");
+  return QString::fromUtf8(mosquitto_strerror(rc));
+}
+
+}  // namespace
+
+MosquittoClient::MosquittoClient(Settings settings, QObject* parent)
+    : MqttClient(parent), m_settings(std::move(settings)) {
+  initLibrary();
+  m_mosq = mosquitto_new(m_settings.clientId.toUtf8().constData(), true, this);
+  if (!m_mosq) {
+    m_initError = QStringLiteral("Cannot create the MQTT client: %1")
+                      .arg(QString::fromLocal8Bit(strerror(errno)));
+    return;
+  }
+  // Our own thread drives the loop (not mosquitto_loop_start()).
+  mosquitto_threaded_set(m_mosq, true);
+  mosquitto_connect_callback_set(m_mosq, &MosquittoClient::onConnect);
+  mosquitto_disconnect_callback_set(m_mosq, &MosquittoClient::onDisconnect);
+  mosquitto_publish_callback_set(m_mosq, &MosquittoClient::onPublish);
+  mosquitto_message_callback_set(m_mosq, &MosquittoClient::onMessage);
+
+  const MqttConfig& broker = m_settings.broker;
+  int rc = MOSQ_ERR_SUCCESS;
+  if (!broker.username.isEmpty()) {
+    rc = mosquitto_username_pw_set(m_mosq, broker.username.toUtf8().constData(),
+                                   broker.password.isEmpty()
+                                       ? nullptr
+                                       : broker.password.toUtf8().constData());
+  }
+  if (rc == MOSQ_ERR_SUCCESS && !m_settings.will.topic.isEmpty()) {
+    const Will& will = m_settings.will;
+    rc = mosquitto_will_set(m_mosq, will.topic.toUtf8().constData(),
+                            int(will.payload.size()), will.payload.constData(),
+                            1, will.retain);
+  }
+  if (rc == MOSQ_ERR_SUCCESS && broker.tls.enabled)
+    m_initError = QStringLiteral("TLS is not supported yet");
+  if (rc != MOSQ_ERR_SUCCESS) m_initError = errorText(rc, errno);
+}
+
+MosquittoClient::~MosquittoClient() {
+  stopThread();
+  if (m_mosq) mosquitto_destroy(m_mosq);
+}
+
+void MosquittoClient::connectToBroker() {
+  if (m_thread.joinable()) return;
+  if (!m_initError.isEmpty()) {
+    setState(State::Disconnected, m_initError);
+    return;
+  }
+  qInfo().noquote() << "MQTT: connecting to"
+                    << QStringLiteral("%1:%2")
+                           .arg(m_settings.broker.host)
+                           .arg(m_settings.broker.port)
+                    << "as" << m_settings.clientId;
+  m_stop = false;
+  setState(State::Connecting, {});
+  m_thread = std::thread(&MosquittoClient::run, this);
+}
+
+bool MosquittoClient::disconnectFromBroker(int timeoutMs) {
+  bool acknowledged = true;
+  if (m_state == State::Connected) {
+    const Will& will = m_settings.will;
+    if (!will.topic.isEmpty()) publish(will.topic, will.payload, will.retain);
+    acknowledged = waitForPublished(timeoutMs);
+  }
+  stopThread();
+  setState(State::Disconnected, {});
+  return acknowledged;
+}
+
+bool MosquittoClient::publish(const QString& topic, const QByteArray& payload,
+                              bool retain) {
+  if (m_state != State::Connected) return false;
+  int mid = 0;
+  const int rc =
+      mosquitto_publish(m_mosq, &mid, topic.toUtf8().constData(),
+                        int(payload.size()), payload.constData(), 1, retain);
+  if (rc != MOSQ_ERR_SUCCESS) {
+    qWarning().noquote() << "MQTT: cannot publish to" << topic << "-"
+                         << errorText(rc, errno);
+    return false;
+  }
+  std::lock_guard lock(m_mutex);
+  if (m_earlyAcks.erase(mid) == 0) m_unacked.insert(mid);
+  return true;
+}
+
+bool MosquittoClient::subscribe(const QString& topic) {
+  if (m_state != State::Connected) return false;
+  const int rc =
+      mosquitto_subscribe(m_mosq, nullptr, topic.toUtf8().constData(), 1);
+  if (rc != MOSQ_ERR_SUCCESS) {
+    qWarning().noquote() << "MQTT: cannot subscribe to" << topic << "-"
+                         << errorText(rc, errno);
+    return false;
+  }
+  return true;
+}
+
+bool MosquittoClient::waitForPublished(int timeoutMs) {
+  std::unique_lock lock(m_mutex);
+  return m_wake.wait_for(lock, std::chrono::milliseconds(timeoutMs),
+                         [this] { return m_unacked.empty(); });
+}
+
+void MosquittoClient::run() {
+  const QByteArray host = m_settings.broker.host.toUtf8();
+  int delayMs = m_settings.reconnectMinMs;
+  while (!m_stop) {
+    m_lastError.clear();
+    // Blocks for the DNS lookup; the TCP connect itself completes in the
+    // loop below.
+    int rc =
+        mosquitto_connect_async(m_mosq, host.constData(),
+                                m_settings.broker.port, m_settings.keepaliveS);
+    int err = errno;
+    if (rc == MOSQ_ERR_SUCCESS) {
+      // Runs until the connection ends; after a stop request only as long
+      // as it takes to send the DISCONNECT.
+      Clock::time_point stopDeadline = Clock::time_point::max();
+      do {
+        rc = mosquitto_loop(m_mosq, 1000, 1);
+        err = errno;
+        if (m_stop && stopDeadline == Clock::time_point::max())
+          stopDeadline = Clock::now() + kStopGrace;
+      } while (rc == MOSQ_ERR_SUCCESS && Clock::now() < stopDeadline);
+    }
+    if (m_stop) break;
+    // A connection that came up starts the backoff again.
+    if (m_state == State::Connected) delayMs = m_settings.reconnectMinMs;
+    setState(State::Disconnected,
+             m_lastError.isEmpty() ? errorText(rc, err) : m_lastError);
+    if (!sleepUnlessStopped(delayMs)) break;
+    delayMs = std::min(delayMs * 2, m_settings.reconnectMaxMs);
+  }
+}
+
+bool MosquittoClient::sleepUnlessStopped(int ms) {
+  std::unique_lock lock(m_mutex);
+  return !m_wake.wait_for(lock, std::chrono::milliseconds(ms),
+                          [this] { return m_stop.load(); });
+}
+
+void MosquittoClient::stopThread() {
+  if (!m_thread.joinable()) return;
+  {
+    std::lock_guard lock(m_mutex);
+    m_stop = true;
+  }
+  m_wake.notify_all();
+  // Sends DISCONNECT and wakes the network loop.
+  mosquitto_disconnect(m_mosq);
+  m_thread.join();
+}
+
+void MosquittoClient::setState(State state, const QString& error) {
+  const State previous = m_state.exchange(state);
+  // Repeated failures with the same reason: report them once.
+  if (previous == state && error == m_reportedError) return;
+  m_reportedError = error;
+  switch (state) {
+    case State::Connected:
+      qInfo().noquote() << "MQTT: connected";
+      break;
+    case State::Disconnected:
+      if (error.isEmpty())
+        qInfo().noquote() << "MQTT: disconnected";
+      else
+        qWarning().noquote() << "MQTT: not connected -" << error;
+      break;
+    case State::Connecting:
+      break;
+  }
+  emit connectionChanged(state, error);
+}
+
+void MosquittoClient::onConnect(mosquitto*, void* self, int rc) {
+  auto* client = static_cast<MosquittoClient*>(self);
+  if (rc == 0) {
+    client->setState(State::Connected, {});
+  } else {
+    // The broker refused; the loop ends with MOSQ_ERR_CONN_REFUSED.
+    client->m_lastError = QString::fromUtf8(mosquitto_connack_string(rc));
+  }
+}
+
+void MosquittoClient::onDisconnect(mosquitto*, void* self, int rc) {
+  auto* client = static_cast<MosquittoClient*>(self);
+  if (rc != 0 && client->m_lastError.isEmpty())
+    client->m_lastError = errorText(rc, errno);
+}
+
+void MosquittoClient::onPublish(mosquitto*, void* self, int mid) {
+  auto* client = static_cast<MosquittoClient*>(self);
+  {
+    std::lock_guard lock(client->m_mutex);
+    if (client->m_unacked.erase(mid) == 0) client->m_earlyAcks.insert(mid);
+  }
+  client->m_wake.notify_all();
+}
+
+void MosquittoClient::onMessage(mosquitto*, void* self,
+                                const mosquitto_message* message) {
+  auto* client = static_cast<MosquittoClient*>(self);
+  emit client->messageReceived(
+      QString::fromUtf8(message->topic),
+      QByteArray(static_cast<const char*>(message->payload),
+                 message->payloadlen));
+}
@@ -0,0 +1,80 @@
+#pragma once
+
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+#include <set>
+#include <thread>
+
+#include "config/Config.h"
+#include "mqtt/MqttClient.h"
+
+struct mosquitto;
+struct mosquitto_message;
+
+// MqttClient on libmosquitto (docs/DESIGN.md, D42, D45).
+//
+// One thread of our own runs everything that can block: the DNS lookup, the
+// connect, libmosquitto's network loop, and the reconnect with backoff. The
+// libmosquitto callbacks run there and only emit signals with copied values.
+// publish()/subscribe() from the main thread are safe: libmosquitto guards
+// its state with mutexes (mosquitto_threaded_set()).
+class MosquittoClient : public MqttClient {
+  Q_OBJECT
+ public:
+  struct Settings {
+    MqttConfig broker;
+    QString clientId;
+    Will will;
+    // Reconnect backoff: doubles from min to max after each failed attempt.
+    int reconnectMinMs = 1000;
+    int reconnectMaxMs = 60000;
+    int keepaliveS = 30;
+  };
+
+  explicit MosquittoClient(Settings settings, QObject* parent = nullptr);
+  ~MosquittoClient() override;
+
+  void connectToBroker() override;
+  bool disconnectFromBroker(int timeoutMs) override;
+  bool publish(const QString& topic, const QByteArray& payload,
+               bool retain) override;
+  bool subscribe(const QString& topic) override;
+  bool waitForPublished(int timeoutMs) override;
+  State state() const override { return m_state; }
+
+ private:
+  // Thread body: connect, loop, back off, until m_stop.
+  void run();
+  // Sleeps up to `ms`; returns early (false) when stopping.
+  bool sleepUnlessStopped(int ms);
+  // Ends the thread and joins it.
+  void stopThread();
+  void setState(State state, const QString& error);
+
+  // libmosquitto callbacks (our thread).
+  static void onConnect(mosquitto* mosq, void* self, int rc);
+  static void onDisconnect(mosquitto* mosq, void* self, int rc);
+  static void onPublish(mosquitto* mosq, void* self, int mid);
+  static void onMessage(mosquitto* mosq, void* self,
+                        const mosquitto_message* message);
+
+  const Settings m_settings;
+  mosquitto* m_mosq = nullptr;
+  QString m_initError;  // set: the client can't work at all
+  std::thread m_thread;
+  std::atomic<State> m_state{State::Disconnected};
+  std::atomic<bool> m_stop{false};
+  // Why the last attempt failed (our thread only).
+  QString m_lastError;
+  // The error last reported by setState(), to report repeats only once.
+  QString m_reportedError;
+
+  // Guards m_stop's wakeup and the acknowledgement bookkeeping.
+  std::mutex m_mutex;
+  std::condition_variable m_wake;
+  // QoS 1 messages not yet acknowledged, and acknowledgements that arrived
+  // before publish() could record their message id.
+  std::set<int> m_unacked;
+  std::set<int> m_earlyAcks;
+};
@@ -0,0 +1,54 @@
+#pragma once
+
+#include <QByteArray>
+#include <QObject>
+#include <QString>
+
+// A connection to an MQTT broker (docs/DESIGN.md, "Iteration 2"). The HA layer
+// only knows this interface; MosquittoClient implements it on libmosquitto,
+// tests use a fake.
+//
+// Call it from the thread it lives in (the main thread). Its signals may be
+// emitted from another thread: connect receivers that live in the main
+// thread, so Qt queues them.
+class MqttClient : public QObject {
+  Q_OBJECT
+ public:
+  enum class State { Disconnected, Connecting, Connected };
+  Q_ENUM(State)
+
+  // Published by the broker when the connection breaks, and by
+  // disconnectFromBroker() before it disconnects.
+  struct Will {
+    QString topic;
+    QByteArray payload;
+    bool retain = false;
+  };
+
+  using QObject::QObject;
+
+  // Starts connecting. Retries on its own, with backoff, until
+  // disconnectFromBroker(); connectionChanged() reports progress.
+  virtual void connectToBroker() = 0;
+  // Publishes the last will, waits up to `timeoutMs` for everything published
+  // to be acknowledged, and disconnects. Returns false on timeout.
+  virtual bool disconnectFromBroker(int timeoutMs) = 0;
+
+  // QoS 1. Only while connected (nothing is buffered); returns false
+  // otherwise.
+  virtual bool publish(const QString& topic, const QByteArray& payload,
+                       bool retain) = 0;
+  // Subscriptions don't survive a reconnect: subscribe again after each
+  // connect.
+  virtual bool subscribe(const QString& topic) = 0;
+  // Waits up to `timeoutMs` until the broker has acknowledged everything
+  // published so far. Returns false on timeout.
+  virtual bool waitForPublished(int timeoutMs) = 0;
+
+  virtual State state() const = 0;
+
+ signals:
+  // `error` says why the connection failed or broke; empty otherwise.
+  void connectionChanged(MqttClient::State state, const QString& error);
+  void messageReceived(const QString& topic, const QByteArray& payload);
+};