From 96df60e6bded8c32feb3a74b4e3d9b9c60914b32 Mon Sep 17 00:00:00 2001 From: Matzman Date: Tue, 6 Oct 2026 16:09:53 +0200 Subject: [PATCH] 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 --- docs/DESIGN.md | 63 +++--- docs/PROGRESS.md | 17 +- src/CMakeLists.txt | 2 + src/mqtt/MosquittoClient.cpp | 248 ++++++++++++++++++++++++ src/mqtt/MosquittoClient.h | 80 ++++++++ src/mqtt/MqttClient.h | 54 ++++++ tests/CMakeLists.txt | 5 + tests/tst_mqtt.cpp | 365 +++++++++++++++++++++++++++++++++++ 8 files changed, 805 insertions(+), 29 deletions(-) create mode 100644 src/mqtt/MosquittoClient.cpp create mode 100644 src/mqtt/MosquittoClient.h create mode 100644 src/mqtt/MqttClient.h create mode 100644 tests/tst_mqtt.cpp diff --git a/docs/DESIGN.md b/docs/DESIGN.md index fa7296b..d7948c2 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -15,7 +15,7 @@ MQTT and HA device"), phase 3 (implementation). Requirements and their decisions | Thread | Runs | |---|---| | main | Qt event loop, QML, off-screen rendering of all surfaces, SteamVR event polling and texture submission, **all OpenVR calls** (I/O wrappers), HA device and entities, managers | - | MQTT | libmosquitto's network loop (`mosquitto_loop_start()`, a thread libmosquitto creates and manages, D42): DNS lookup, TCP and TLS connect block for seconds when the broker is unreachable | + | MQTT | libmosquitto's connect and network loop, in one `std::thread` of `MosquittoClient` (D42, D45): DNS lookup, TCP and TLS connect block for seconds when the broker is unreachable | - OpenVR stays in the main thread: it isn't documented as thread-safe, and all events come through one queue. @@ -28,9 +28,9 @@ MQTT and HA device"), phase 3 (implementation). Requirements and their decisions HA device ◀── connectionChanged(state, error) ── MQTT client ``` - - The MQTT client object lives in the main thread; libmosquitto calls its callbacks in its - own thread, and they only emit these signals with copied values (D42, details in - "Iteration 2: MQTT and HA device"). + - The MQTT client object lives in the main thread; libmosquitto calls its callbacks in the + client's own thread, and they only emit these signals with copied values (D42, D45; + details in "Iteration 2: MQTT and HA device"). - Not needed as own threads: WebEngine (own processes), Qt Multimedia and QtNetwork (asynchronous), later D-Bus and `QProcess` for network/audio (asynchronous). Rendering stays in the main thread as long as NFR-3 holds. @@ -269,7 +269,7 @@ Implements the iteration 2 requirements ([PROGRESS.md](PROGRESS.md), "Next steps ### Components ``` -main thread libmosquitto thread +main thread MosquittoClient's thread Dashboard (app) ├─ ConnectionStatus (QML singleton, UI-1) ◀─ signals ─┐ ├─ HaDevice ── publish/subscribe ──▶ MqttClient ◀─────┼── callbacks (connect, message, disconnect) @@ -282,39 +282,49 @@ MQTT (D17). | Class | File | Role | |---|---|---| -| `MqttClient` | `src/mqtt/MqttClient.h` | **Interface** (QObject): `connectToBroker()`, `publish(topic, payload, retain)`, `subscribe(topic)`; signals `connectionChanged(state, error)`, `messageReceived(topic, payload)`. Tests use a `FakeMqttClient` | +| `MqttClient` | `src/mqtt/MqttClient.h` | **Interface** (QObject): `connectToBroker()`, `disconnectFromBroker(timeout)` (publishes the will first), `publish(topic, payload, retain)`, `subscribe(topic)`, `waitForPublished(timeout)`, `state()`; signals `connectionChanged(state, error)`, `messageReceived(topic, payload)`. Tests use a `FakeMqttClient` | | `MosquittoClient` | `src/mqtt/MosquittoClient.*` | Implementation on libmosquitto: login, TLS, last will, reconnect with backoff | | `HaDevice` | `src/ha/HaDevice.*` | One HA device (D18): discovery, availability, republish on reconnect and HA birth, `removeFromHa()`. Owns its entities, uses an `MqttClient*` | | `HaEntity`, `HaSensor` | `src/ha/HaEntity.*`, `HaSensor.*` | Base class (D23): description (id, name, device class, unit, category, throttle interval), discovery component, state and availability topics, throttling (SEN-14). `HaSensor::setValue()` / `clearValue()` (unavailable, SEN-15) | | `ConnectionStatus` | `src/ConnectionStatus.*` (app, QML module) | QML singleton for UI-1: device name, broker, state (`Disconnected`/`Connecting`/`Connected`), last error. Member of `Dashboard` | -### MQTT client (D42) +### MQTT client (D42, D45) -- **Thread:** `mosquitto_connect_async()` plus `mosquitto_loop_start()`. libmosquitto creates - and manages its own thread (a pthread, not a `QThread`, no Qt event loop) for connect, - network traffic, keep-alive and reconnect, and calls our callbacks there. +- **Thread (D45):** `MosquittoClient` starts one `std::thread` that loops: `mosquitto_connect_async()` + (blocks for the DNS lookup), then `mosquitto_loop()` until the connection ends, then waits + with backoff (1 s doubling to 60 s, reset after a connection that came up) and starts over, + until stopped. `mosquitto_threaded_set()` tells libmosquitto that we drive the loop. + - Not `mosquitto_loop_start()` (D42 as first written), checked in libmosquitto 2.0.21's + source: `mosquitto_connect_async()` resolves the host with a blocking `getaddrinfo()` in + the *calling* thread, and `mosquitto_loop_forever()` returns on "fatal" errors (TLS, + protocol), after which libmosquitto's thread ends silently and never reconnects. + - Our own loop also reports the reason of every failed attempt; libmosquitto's internal + reconnect doesn't report its failures. - **Callbacks only emit signals** with copied values. The `MosquittoClient` object lives in the main thread, so Qt delivers them queued (D29's rules hold). -- **Calls from the main thread** (`mosquitto_publish()`, `mosquitto_subscribe()`) are safe - while the loop runs: libmosquitto guards its state with mutexes. Needs a libmosquitto with - thread support (Debian, Arch); otherwise `mosquitto_loop_start()` returns - `MOSQ_ERR_NOT_SUPPORTED`, which is logged as an error. +- **Calls from the main thread** (`mosquitto_publish()`, `mosquitto_subscribe()`, + `mosquitto_disconnect()`) are safe while the loop runs: libmosquitto guards its state with + mutexes and wakes the loop through its socket pair. - **Client id** `frameassistant_`: two instances with the same id knock each other - off the broker (expected, D22). MQTT 3.1.1, QoS 1 for everything. + off the broker (expected, D22). MQTT 3.1.1, QoS 1 for everything, keep-alive 30 s. +- **Nothing is buffered:** `publish()` and `subscribe()` return false unless connected. + Subscriptions are lost on reconnect (clean session); `HaDevice` subscribes after each connect. - **Last will:** `offline`, retained, on `frameassistant//availability`. A killed process (also `pkill` from `install.sh`) closes its TCP connection, so the broker sends the will at once. -- **Normal exit:** publish `offline`, wait up to 2 s for its acknowledgement, - `mosquitto_disconnect()`, `mosquitto_loop_stop()` (joins the thread), `mosquitto_destroy()`. - After that no callback can arrive. -- **Reconnect (CON-4):** `mosquitto_reconnect_delay_set(1 s, 60 s, exponential)`. Nothing is - buffered. +- **Normal exit:** `disconnectFromBroker(timeout)` publishes the will payload itself, waits up to + the timeout for all acknowledgements (`waitForPublished()`), sends DISCONNECT and joins the + thread. After that no callback can arrive. - **TLS:** `mosquitto_tls_set()` with `ca_file`; with an empty `ca_file`, `MOSQ_OPT_TLS_USE_OS_CERTS` (libmosquitto ≥ 2.0). Client certificate if configured. -- **Logging:** every connection state change and error with its reason (`qInfo`/`qWarning`, so - also in the log file). -- To verify when implementing: whether the loop also retries when the *first* - `connect_async` fails (e.g. DNS error); if not, `MosquittoClient` retries that case itself. + (Step 5; until then a TLS setting is reported as an error.) +- **Errors** are reported in English regardless of the locale (`strerrordesc_np()`). A DNS + failure has no reason: libmosquitto overwrites `getaddrinfo()`'s code in `errno` before it + returns. The log line "connecting to host:port" names the host. +- **Logging:** every connection state change and error (`qInfo`/`qWarning`, so also in the log + file); repeated failures with the same reason are reported once. Retries after a failure + stay "not connected" (no "connecting" flicker). +- Checked: a failed *first* connect (refused, DNS error) is retried (`tst_mqtt`). ### HA device @@ -391,7 +401,7 @@ delete). |---|---| | `tst_haentity` | discovery component, topics, throttling (several values within an interval → one publish with the last value), unavailable without a value | | `tst_hadevice` (fake client) | discovery payload, `online` and states after connect, republish after reconnect and birth, will configuration, `removeFromHa()` | -| `tst_mqtt` (integration; real `mosquitto` on a free port, QSKIP without it) | connect; login right and wrong; TLS with a test CA; reconnect after a broker restart; last will after `kill -9` of a small helper process | +| `tst_mqtt` (integration; real `mosquitto` on a free port, QSKIP without it) | connect and messages; retained message for a later subscriber; nothing published while disconnected; first connect retried until the broker runs; reconnect after a broker restart; unresolvable host; last will after `kill -9` of a helper process (the test binary itself, `--will-helper`); `disconnectFromBroker()` publishes the will; login right and wrong; TLS with a test CA (step 5) | | `tst_surfaces` | `--desktop-novr`; app command: hook called, no surface created, exit code returned | | Manual | from the PC against the user's HA: device appears, online/offline, broker restart, two devices, `--ha-remove`, UI-1 in `--desktop`. On the Frame: the package with libmosquitto runs | @@ -595,6 +605,7 @@ Collected 2026-10-05 for the notification design. | D37 | 2026-10-05 | The MVP is implemented in four iterations: 1 DSH-1 + configuration, 2 MQTT/HA device/UI-1, 3 sensors, 4 notifications | Small iterations (working rules); DSH-1 first because the user wants to see it early and it doesn't depend on MQTT | | D38 | 2026-10-05 | App-specific command-line options: the platform's base constructor takes an optional list of extra options, parses them with its own and exposes their values | The command line is parsed in the base constructor, before the derived app can contribute (`--config`); backward compatible, `--help` stays complete | | D39 | 2026-10-05 | *Superseded by D40 (REQUIREMENTS.md): tab and WebEngine removed.* DSH-1: `QtWebEngineQuick::initialize()` in `main.cpp` (no platform change); a named, persistent WebEngine profile under `~/.local/share/frameassistant/`; tab name "HA Dashboard" | WebEngine must be initialised before the `QGuiApplication`; the HA login must survive restarts | -| D42 | 2026-10-06 | The MQTT client's blocking work runs in **libmosquitto's own thread** (`mosquitto_connect_async()` + `mosquitto_loop_start()`), incl. its reconnect with backoff; callbacks only emit Qt signals with values to the main thread. Changes how D29 is carried out, not its rules | Same effect as an own `QThread` with much less code; libmosquitto's publish/subscribe are thread-safe while its loop runs | +| D42 | 2026-10-06 | *Carried out as amended by D45.* The MQTT client's blocking work runs in **libmosquitto's own thread** (`mosquitto_connect_async()` + `mosquitto_loop_start()`), incl. its reconnect with backoff; callbacks only emit Qt signals with values to the main thread. Changes how D29 is carried out, not its rules | Same effect as an own `QThread` with much less code; libmosquitto's publish/subscribe are thread-safe while its loop runs | | D43 | 2026-10-06 | **App commands** in the platform: the app passes command options to the base constructor; they get a `QCoreApplication` like the manifest commands, and `exec()` runs `prepare()` and then the hook `runCommand(name)` | `--ha-remove` must work without a display (`uninstall.sh` over SSH); reuses the existing manifest-command path; `--config`/`--log-file`/`--help` stay consistent | | D44 | 2026-10-06 | TLS test certificates are committed to `tests/data/` (100-year validity) | Deterministic tests; no openssl needed in the containers | +| D45 | 2026-10-06 | *Amends D42.* `MosquittoClient` runs connect, network loop and reconnect in **one `std::thread` of its own** instead of `mosquitto_loop_start()` | libmosquitto 2.0.21's source: `mosquitto_connect_async()` resolves DNS blocking in the calling thread, and the `loop_start` thread ends silently on TLS/protocol errors; our own loop also reports each failed attempt | diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md index 4bf3669..734d926 100644 --- a/docs/PROGRESS.md +++ b/docs/PROGRESS.md @@ -22,9 +22,9 @@ CLAUDE.md ("Working rules"). The error itself is covered off the device by `tst_config`. Checks 3 and 4 run in the next session on the Frame (iteration 2 testing needs it too). -- **Implementation** follows the order in DESIGN.md (steps 1–7). Step 1 (libmosquitto in - CMake and the containers) done; next: step 2 (`MqttClient`, `MosquittoClient`, - integration test). +- **Implementation** follows the order in DESIGN.md (steps 1–7). Steps 1 (libmosquitto in + CMake and the containers) and 2 (`MqttClient`, `MosquittoClient`, `tst_mqtt`) done; next: + step 3 (`HaEntity`, `HaSensor`, `HaDevice`, unit tests). ## Next steps @@ -133,6 +133,17 @@ Known limitation: the bundled-Qt package does not support WebEngine in this iter 2.0.21 (aarch64) and the copyright file. The binary doesn't need the library yet (nothing calls it; `--as-needed`), which changes in step 2. On the host, the GUI tests can't run in this session (no display). +- User: no need to ask for permission to continue between the steps of this phase. +- Step 2: `MqttClient` interface, `MosquittoClient`, `tst_mqtt` (12 cases against a local + mosquitto). Reading libmosquitto 2.0.21's source showed that `mosquitto_loop_start()` + doesn't fit (blocking DNS in the caller, silent thread end on TLS/protocol errors); the + user chose an own `std::thread` (D45, amends D42). Findings: a failed first connect is + retried; libmosquitto loses the DNS error code; `strerror()` is localised (German on the + dev host), so errors use `strerrordesc_np()`. Test pitfall: `QTRY_*` evaluates its + expression once more at the end, so it must not consume data (`readAll…()`). Checked: + host 3 × 12/12; `container-test.sh` 9/9 (Debian's libmosquitto 2.0.21), no Vulkan + validation messages; cross build: private Qt symbols only `QVulkan*`, at most + `GLIBC_2.34`; the app binary links libmosquitto only from step 4 on (first use). ### 2026-10-05: requirements approved diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index a87fcaa..b5d19a0 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -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) diff --git a/src/mqtt/MosquittoClient.cpp b/src/mqtt/MosquittoClient.cpp new file mode 100644 index 0000000..fb5298e --- /dev/null +++ b/src/mqtt/MosquittoClient.cpp @@ -0,0 +1,248 @@ +#include "mqtt/MosquittoClient.h" + +#include + +#include +#include +#include +#include +#include +#include + +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(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(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(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(self); + emit client->messageReceived( + QString::fromUtf8(message->topic), + QByteArray(static_cast(message->payload), + message->payloadlen)); +} diff --git a/src/mqtt/MosquittoClient.h b/src/mqtt/MosquittoClient.h new file mode 100644 index 0000000..2eb09e1 --- /dev/null +++ b/src/mqtt/MosquittoClient.h @@ -0,0 +1,80 @@ +#pragma once + +#include +#include +#include +#include +#include + +#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 m_state{State::Disconnected}; + std::atomic 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 m_unacked; + std::set m_earlyAcks; +}; diff --git a/src/mqtt/MqttClient.h b/src/mqtt/MqttClient.h new file mode 100644 index 0000000..5899871 --- /dev/null +++ b/src/mqtt/MqttClient.h @@ -0,0 +1,54 @@ +#pragma once + +#include +#include +#include + +// 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); +}; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 23e5d8d..5c97362 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -40,3 +40,8 @@ set_tests_properties(surfaces_opengl surfaces_vulkan surfaces_cli_key qt_add_executable(tst_config tst_config.cpp) target_link_libraries(tst_config PRIVATE frameassistant_lib Qt6::Test) add_test(NAME config COMMAND tst_config) + +# MosquittoClient against a local mosquitto broker (D30 level 2); skips without one. +qt_add_executable(tst_mqtt tst_mqtt.cpp) +target_link_libraries(tst_mqtt PRIVATE frameassistant_lib Qt6::Test) +add_test(NAME mqtt COMMAND tst_mqtt) diff --git a/tests/tst_mqtt.cpp b/tests/tst_mqtt.cpp new file mode 100644 index 0000000..5000164 --- /dev/null +++ b/tests/tst_mqtt.cpp @@ -0,0 +1,365 @@ +// MosquittoClient against a real broker (docs/DESIGN.md, D30 level 2): a +// local mosquitto on a free port. Connect, messages, retained messages, +// reconnect, last will, login. Skips when mosquitto isn't installed. +// +// The binary is also its own helper process for the last-will test: +// `tst_mqtt --will-helper ` connects with a will and waits to be +// killed. + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "mqtt/MosquittoClient.h" + +namespace { + +constexpr int kTimeoutMs = 10000; + +QString findTool(const QString& name) { + QString path = QStandardPaths::findExecutable(name); + if (path.isEmpty()) + path = QStandardPaths::findExecutable(name, {QStringLiteral("/usr/sbin")}); + return path; +} + +// A TCP port nobody listens on right now. +int freePort() { + const int fd = socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + socklen_t length = sizeof(address); + bind(fd, reinterpret_cast(&address), length); + getsockname(fd, reinterpret_cast(&address), &length); + close(fd); + return ntohs(address.sin_port); +} + +bool portOpen(int port) { + const int fd = socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + address.sin_port = htons(port); + const bool open = ::connect(fd, reinterpret_cast(&address), + sizeof(address)) == 0; + close(fd); + return open; +} + +// A mosquitto process with its own configuration. +class Broker { + public: + explicit Broker(const QString& dir) : m_dir(dir), m_port(freePort()) {} + ~Broker() { stop(); } + + int port() const { return m_port; } + + // `extra`: further configuration lines; without them anonymous clients may + // connect. + bool start(const QString& extra = QStringLiteral("allow_anonymous true\n")) { + const QString config = m_dir + QStringLiteral("/mosquitto.conf"); + QFile file(config); + if (!file.open(QIODevice::WriteOnly)) return false; + file.write(QStringLiteral("listener %1 127.0.0.1\npersistence false\n%2") + .arg(m_port) + .arg(extra) + .toUtf8()); + file.close(); + m_process = std::make_unique(); + m_process->setProcessChannelMode(QProcess::ForwardedErrorChannel); + m_process->start(findTool(QStringLiteral("mosquitto")), + {QStringLiteral("-c"), config}); + if (!m_process->waitForStarted()) return false; + QDeadlineTimer deadline(5000); + while (!portOpen(m_port)) { + if (deadline.hasExpired() || m_process->state() != QProcess::Running) + return false; + QThread::msleep(20); + } + return true; + } + + void stop() { + if (!m_process) return; + m_process->terminate(); + if (!m_process->waitForFinished(5000)) m_process->kill(); + m_process->waitForFinished(); + m_process.reset(); + } + + private: + QString m_dir; + int m_port; + std::unique_ptr m_process; +}; + +MosquittoClient::Settings settings(int port, const QString& clientId) { + MosquittoClient::Settings s; + s.broker.host = QStringLiteral("127.0.0.1"); + s.broker.port = port; + s.clientId = clientId; + s.reconnectMinMs = 100; + s.reconnectMaxMs = 500; + return s; +} + +// Collects a client's signals in the main thread (queued connections). +class Recorder : public QObject { + public: + explicit Recorder(MqttClient* client) { + connect(client, &MqttClient::connectionChanged, this, + [this](MqttClient::State state, const QString& error) { + states.append(state); + if (!error.isEmpty()) errors.append(error); + }); + connect(client, &MqttClient::messageReceived, this, + [this](const QString& topic, const QByteArray& payload) { + messages.append({topic, payload}); + }); + } + + QList states; + QStringList errors; + QList> messages; +}; + +// Connects and waits for it; then subscribes to `topic` if given. +bool connectClient(MosquittoClient& client, Recorder& recorder, + const QString& topic = {}) { + client.connectToBroker(); + if (!QTest::qWaitFor( + [&] { + return recorder.states.contains(MqttClient::State::Connected); + }, + kTimeoutMs)) + return false; + if (topic.isEmpty()) return true; + if (!client.subscribe(topic)) return false; + // No SUBACK signal: give the broker a moment to register it. + QTest::qWait(200); + return true; +} + +int runWillHelper(int argc, char** argv) { + QCoreApplication app(argc, argv); + MosquittoClient::Settings s = + settings(QByteArray(argv[2]).toInt(), QStringLiteral("will-helper")); + s.will = {QString::fromUtf8(argv[3]), "offline", true}; + MosquittoClient client(s); + QObject::connect(&client, &MqttClient::connectionChanged, &app, + [](MqttClient::State state) { + if (state == MqttClient::State::Connected) { + fputs("connected\n", stdout); + fflush(stdout); + } + }); + client.connectToBroker(); + return app.exec(); +} + +} // namespace + +class TestMqtt : public QObject { + Q_OBJECT + + private slots: + void initTestCase() { + if (findTool(QStringLiteral("mosquitto")).isEmpty()) + QSKIP("mosquitto (the broker) is not installed"); + QVERIFY(m_dir.isValid()); + } + + void connectsAndExchangesMessages() { + Broker broker(m_dir.path()); + QVERIFY(broker.start()); + MosquittoClient a(settings(broker.port(), QStringLiteral("a"))); + MosquittoClient b(settings(broker.port(), QStringLiteral("b"))); + Recorder ra(&a), rb(&b); + QVERIFY(connectClient(a, ra)); + QCOMPARE(ra.states.first(), MqttClient::State::Connecting); + QCOMPARE(a.state(), MqttClient::State::Connected); + QVERIFY(connectClient(b, rb, QStringLiteral("test/#"))); + + QVERIFY(a.publish(QStringLiteral("test/x"), "hello", false)); + QVERIFY(a.waitForPublished(kTimeoutMs)); + QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs); + QCOMPARE(rb.messages.first().first, QStringLiteral("test/x")); + QCOMPARE(rb.messages.first().second, QByteArray("hello")); + QVERIFY(ra.errors.isEmpty()); + } + + void retainedMessageReachesLaterSubscriber() { + Broker broker(m_dir.path()); + QVERIFY(broker.start()); + MosquittoClient a(settings(broker.port(), QStringLiteral("a"))); + Recorder ra(&a); + QVERIFY(connectClient(a, ra)); + QVERIFY(a.publish(QStringLiteral("test/retained"), "kept", true)); + QVERIFY(a.waitForPublished(kTimeoutMs)); + + MosquittoClient b(settings(broker.port(), QStringLiteral("b"))); + Recorder rb(&b); + QVERIFY(connectClient(b, rb, QStringLiteral("test/retained"))); + QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs); + QCOMPARE(rb.messages.first().second, QByteArray("kept")); + } + + void publishFailsWhileDisconnected() { + MosquittoClient a(settings(freePort(), QStringLiteral("a"))); + QVERIFY(!a.publish(QStringLiteral("test/x"), "lost", false)); + QVERIFY(!a.subscribe(QStringLiteral("test/x"))); + QVERIFY(a.waitForPublished(0)); + } + + // The broker isn't running yet: the error is reported, and the client + // connects once it runs. + void firstConnectRetriesUntilBrokerRuns() { + Broker broker(m_dir.path()); + MosquittoClient a(settings(broker.port(), QStringLiteral("a"))); + Recorder ra(&a); + a.connectToBroker(); + QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs); + QCOMPARE(a.state(), MqttClient::State::Disconnected); + QVERIFY2(ra.errors.first().contains(QStringLiteral("refused")), + qPrintable(ra.errors.first())); + QVERIFY(broker.start()); + QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Connected, + kTimeoutMs); + } + + void reconnectsAfterBrokerRestart() { + Broker broker(m_dir.path()); + QVERIFY(broker.start()); + MosquittoClient a(settings(broker.port(), QStringLiteral("a"))); + Recorder ra(&a); + QVERIFY(connectClient(a, ra)); + broker.stop(); + QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Disconnected, + kTimeoutMs); + QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs); + ra.states.clear(); + QVERIFY(broker.start()); + QTRY_VERIFY_WITH_TIMEOUT(ra.states.contains(MqttClient::State::Connected), + kTimeoutMs); + } + + void unresolvableHostIsReported() { + MosquittoClient::Settings s = settings(1883, QStringLiteral("a")); + s.broker.host = QStringLiteral("no-such-host.invalid"); + MosquittoClient a(s); + Recorder ra(&a); + a.connectToBroker(); + QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs); + QVERIFY2(ra.errors.first().startsWith(QStringLiteral("Cannot resolve")), + qPrintable(ra.errors.first())); + } + + // CON-3: a killed process (no DISCONNECT) → the broker publishes the will. + void lastWillWhenKilled() { + Broker broker(m_dir.path()); + QVERIFY(broker.start()); + MosquittoClient watcher(settings(broker.port(), QStringLiteral("w"))); + Recorder rw(&watcher); + QVERIFY(connectClient(watcher, rw, QStringLiteral("test/availability"))); + + QProcess helper; + helper.start( + QCoreApplication::applicationFilePath(), + {QStringLiteral("--will-helper"), QString::number(broker.port()), + QStringLiteral("test/availability")}); + QVERIFY(helper.waitForStarted()); + // QTRY_* evaluates its expression again at the end: collect the output. + QByteArray output; + QTRY_VERIFY_WITH_TIMEOUT( + (output += helper.readAllStandardOutput()).contains("connected"), + kTimeoutMs); + QVERIFY(rw.messages.isEmpty()); + helper.kill(); + helper.waitForFinished(); + QTRY_COMPARE_WITH_TIMEOUT(rw.messages.size(), 1, kTimeoutMs); + QCOMPARE(rw.messages.first().second, QByteArray("offline")); + } + + // A normal exit publishes the will itself (retained), then disconnects. + void disconnectPublishesWill() { + Broker broker(m_dir.path()); + QVERIFY(broker.start()); + MosquittoClient::Settings s = settings(broker.port(), QStringLiteral("a")); + s.will = {QStringLiteral("test/availability"), "offline", true}; + MosquittoClient a(s); + Recorder ra(&a); + QVERIFY(connectClient(a, ra)); + QVERIFY(a.disconnectFromBroker(kTimeoutMs)); + QCOMPARE(a.state(), MqttClient::State::Disconnected); + QTRY_COMPARE_WITH_TIMEOUT(ra.states.last(), MqttClient::State::Disconnected, + kTimeoutMs); + + MosquittoClient b(settings(broker.port(), QStringLiteral("b"))); + Recorder rb(&b); + QVERIFY(connectClient(b, rb, QStringLiteral("test/availability"))); + QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs); + QCOMPARE(rb.messages.first().second, QByteArray("offline")); + } + + void login_data() { + QTest::addColumn("password"); + QTest::addColumn("accepted"); + QTest::newRow("right") << QStringLiteral("secret") << true; + QTest::newRow("wrong") << QStringLiteral("wrong") << false; + } + + void login() { + QFETCH(QString, password); + QFETCH(bool, accepted); + const QString passwd = findTool(QStringLiteral("mosquitto_passwd")); + if (passwd.isEmpty()) QSKIP("mosquitto_passwd is not installed"); + const QString file = m_dir.path() + QStringLiteral("/passwd"); + QFile::remove(file); + QCOMPARE(QProcess::execute( + passwd, {QStringLiteral("-c"), QStringLiteral("-b"), file, + QStringLiteral("frame"), QStringLiteral("secret")}), + 0); + Broker broker(m_dir.path()); + QVERIFY(broker.start( + QStringLiteral("allow_anonymous false\npassword_file %1\n").arg(file))); + + MosquittoClient::Settings s = settings(broker.port(), QStringLiteral("a")); + s.broker.username = QStringLiteral("frame"); + s.broker.password = password; + MosquittoClient a(s); + Recorder ra(&a); + a.connectToBroker(); + if (accepted) { + QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Connected, + kTimeoutMs); + } else { + QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs); + QVERIFY2(ra.errors.first().contains(QStringLiteral("not authori")), + qPrintable(ra.errors.first())); + QVERIFY(!ra.states.contains(MqttClient::State::Connected)); + } + } + + private: + QTemporaryDir m_dir; +}; + +int main(int argc, char** argv) { + if (argc == 4 && qstrcmp(argv[1], "--will-helper") == 0) + return runWillHelper(argc, argv); + QCoreApplication app(argc, argv); + TestMqtt test; + return QTest::qExec(&test, argc, argv); +} + +#include "tst_mqtt.moc"