Replaced mosquitto with paho mqtt.

shajen · shajen · commit bdd10d99f8a3 · 2025-10-26T19:30:38.000+01:00
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -22,11 +22,7 @@ find_package(Gnuradio COMPONENTS
     soapy
 )
 find_package(nlohmann_json REQUIRED)
-
-find_library(MOSQUITTO_LIB mosquitto)
-if(NOT MOSQUITTO_LIB)
-  message(FATAL_ERROR "mosquitto library not found")
-endif()
+find_package(PahoMqttCpp REQUIRED)
 
 file(GLOB_RECURSE SOURCES
     "${PROJECT_SOURCE_DIR}/sources/*.h"
@@ -50,9 +46,9 @@ target_link_libraries(auto_sdr
     gnuradio::gnuradio-fft
     gnuradio::gnuradio-filter
     gnuradio::gnuradio-soapy
-    mosquitto
     nlohmann_json::nlohmann_json
     spdlog::spdlog
+    PahoMqttCpp::paho-mqttpp3
 )
 
 add_executable(auto_sdr_test ${TEST_SOURCES} "tests/test_main.cpp")
diff --git a/Dockerfile b/Dockerfile
@@ -1,7 +1,7 @@
 FROM ubuntu:24.04 AS build
 ENV DEBIAN_FRONTEND=noninteractive
 RUN apt-get update && \
-    apt-get install -y --no-install-recommends ca-certificates curl git zip build-essential cmake ccache tzdata libspdlog-dev libliquid-dev nlohmann-json3-dev libmosquitto-dev libgtest-dev libgmock-dev libusb-1.0-0-dev libfftw3-dev libboost-all-dev libsoapysdr-dev gnuradio gnuradio-dev libsndfile1-dev
+    apt-get install -y --no-install-recommends ca-certificates curl git zip build-essential cmake ccache tzdata libspdlog-dev libliquid-dev nlohmann-json3-dev libgtest-dev libgmock-dev libusb-1.0-0-dev libfftw3-dev libboost-all-dev libsoapysdr-dev gnuradio gnuradio-dev libsndfile1-dev libssl-dev libpaho-mqtt-dev libpaho-mqttpp-dev
 
 COPY sdrplay/SDRplay_RSP_API-Linux-3.15.2.run /
 RUN mkdir -p /sdrplay && \
@@ -43,7 +43,7 @@ RUN --mount=type=cache,target=/root/.cache/ccache,id=ccache \
 FROM ubuntu:24.04 AS run
 ENV DEBIAN_FRONTEND=noninteractive
 RUN apt-get update && \
-    apt-get install -y tzdata libspdlog1.12 libliquid1 nlohmann-json3-dev libmosquitto1 libusb-1.0-0 libfftw3-bin && \
+    apt-get install -y tzdata ca-certificates libspdlog1.12 libliquid1 nlohmann-json3-dev libusb-1.0-0 libfftw3-bin libssl3t64 libpaho-mqtt1.3 libpaho-mqttpp3-1 && \
     apt-get install -y --no-install-recommends gnuradio libsoapysdr0.8 soapysdr0.8-module-all && \
     apt-get purge -y soapysdr0.8-module-audio soapysdr0.8-module-uhd && \
     apt-get clean all && \
diff --git a/sources/config.cpp b/sources/config.cpp
@@ -81,8 +81,7 @@ Config::Config(const nlohmann::json& json)
       m_recordingTimeout(std::chrono::milliseconds(readKey<int>(json, {"recording", "max_noise_time_ms"}))),
       m_recordingTuningStep(readKey<Frequency>(json, {"recording", "step"})),
       m_workers(readKey<int>(json, {"workers"})),
-      m_mqttHostname(getEnv("MQTT_HOST")),
-      m_mqttPort(stoi(getEnv("MQTT_PORT_TCP"))),
+      m_mqttUrl(getEnv("MQTT_URL")),
       m_mqttUsername(getEnv("MQTT_USER")),
       m_mqttPassword(getEnv("MQTT_PASSWORD")) {}
 
@@ -124,7 +123,7 @@ void Config::saveToFile(const std::string& path, const nlohmann::json& json) {
 }
 
 nlohmann::json Config::json() const { return m_json; }
-std::string Config::mqtt() const { return fmt::format("{}@{}:{}", m_mqttUsername, m_mqttHostname, m_mqttPort); };
+std::string Config::mqtt() const { return fmt::format("{}@{}", m_mqttUsername, m_mqttUrl); };
 
 std::vector<Device> Config::devices() const { return m_devices; }
 
@@ -143,7 +142,6 @@ std::chrono::milliseconds Config::recordingMinTime() const { return m_recordingM
 std::chrono::milliseconds Config::recordingTimeout() const { return m_recordingTimeout; }
 Frequency Config::recordingTuningStep() const { return m_recordingTuningStep; }
 
-std::string Config::mqttHostname() const { return m_mqttHostname; }
-int Config::mqttPort() const { return m_mqttPort; }
+std::string Config::mqttUrl() const { return m_mqttUrl; }
 std::string Config::mqttUsername() const { return m_mqttUsername; }
 std::string Config::mqttPassword() const { return m_mqttPassword; }
diff --git a/sources/config.h b/sources/config.h
@@ -57,8 +57,7 @@ class Config {
   std::chrono::milliseconds recordingTimeout() const;
   Frequency recordingTuningStep() const;
 
-  std::string mqttHostname() const;
-  int mqttPort() const;
+  std::string mqttUrl() const;
   std::string mqttUsername() const;
   std::string mqttPassword() const;
 
@@ -80,8 +79,7 @@ class Config {
   const Frequency m_recordingTuningStep;
   const int m_workers;
 
-  const std::string m_mqttHostname;
-  const int m_mqttPort;
+  const std::string m_mqttUrl;
   const std::string m_mqttUsername;
   const std::string m_mqttPassword;
 };
diff --git a/sources/network/mqtt.cpp b/sources/network/mqtt.cpp
@@ -4,27 +4,38 @@
 #include <utils/utils.h>
 
 constexpr auto LABEL = "mqtt";
-constexpr auto KEEP_ALIVE = 60;
-constexpr auto LOOP_TIMEOUT_MS = 100;
 constexpr auto QOS_SUB = 2;
-constexpr auto RECONNECT_INTERVAL = std::chrono::seconds(1);
 constexpr auto QUEUE_MAX_SIZE = 1000;
+constexpr auto LOOP_TIMEOUT = std::chrono::milliseconds(10);
+constexpr auto RECONNECT_INTERVAL = std::chrono::seconds(5);
+constexpr auto CONNECT_TIMEOUT = std::chrono::seconds(5);
+constexpr auto KEEP_ALIVE = std::chrono::seconds(60);
 
-Mqtt::Mqtt(const Config &config)
-    : m_client(mosquitto_new(nullptr, true, this)), m_isRunning(true), m_thread([this, config]() {
+Mqtt::Mqtt(const Config& config)
+    : m_config(config), m_client(config.mqttUrl(), "sdr-scanner"), m_isRunning(true), m_thread([this, config]() {
         Logger::info(LABEL, "started");
-        mosquitto_username_pw_set(m_client, config.mqttUsername().c_str(), config.mqttPassword().c_str());
-        mosquitto_connect_callback_set(m_client, [](mosquitto *, void *p, int) { reinterpret_cast<Mqtt *>(p)->onConnect(); });
-        mosquitto_disconnect_callback_set(m_client, [](mosquitto *, void *p, int) { reinterpret_cast<Mqtt *>(p)->onDisconnect(); });
-        mosquitto_message_callback_set(m_client, [](mosquitto *, void *p, const struct mosquitto_message *m) { reinterpret_cast<Mqtt *>(p)->onMessage(m); });
-        mosquitto_connect(m_client, config.mqttHostname().c_str(), config.mqttPort(), KEEP_ALIVE);
+        connect();
         while (m_isRunning) {
-          mosquitto_loop(m_client, LOOP_TIMEOUT_MS, 1);
-          while (m_isRunning && !m_messages.empty()) {
-            const auto &[topic, data, qos] = m_messages.front();
-            mosquitto_publish(m_client, nullptr, topic.c_str(), data.size(), data.data(), qos, false);
-            std::unique_lock lock(m_mutex);
-            m_messages.pop();
+          if (m_client.is_connected()) {
+            std::shared_ptr<const mqtt::message> message;
+            while (m_client.try_consume_message_for(&message, LOOP_TIMEOUT) && message) {
+              onMessage(message->get_topic(), message->get_payload());
+            }
+            while (m_isRunning && !m_messages.empty()) {
+              const auto& [topic, data, qos] = m_messages.front();
+              m_client.publish(topic, data.data(), data.size(), qos, false);
+              std::unique_lock lock(m_mutex);
+              m_messages.pop();
+            }
+          } else {
+            onDisconnected();
+            while (m_isRunning && !m_client.is_connected()) {
+              Logger::info(LABEL, "reconnecting...");
+              connect();
+              if (!m_client.is_connected()) {
+                std::this_thread::sleep_for(RECONNECT_INTERVAL);
+              }
+            }
           }
         }
         Logger::info(LABEL, "stopped");
@@ -33,75 +44,106 @@ Mqtt::Mqtt(const Config &config)
 Mqtt::~Mqtt() {
   m_isRunning = false;
   m_thread.join();
-  mosquitto_disconnect(m_client);
-  mosquitto_destroy(m_client);
+  if (m_client.is_connected()) {
+    m_client.disconnect();
+  }
 }
 
-void Mqtt::publish(const std::string &topic, const std::string &data, int qos) {
+void Mqtt::publish(const std::string& topic, const std::string& data, int qos) {
   std::unique_lock lock(m_mutex);
   if (m_messages.size() < QUEUE_MAX_SIZE) {
     m_messages.emplace(topic, std::vector<uint8_t>{data.begin(), data.end()}, qos);
     Logger::trace(LABEL, "queue size: {}", m_messages.size());
   }
 }
 
-void Mqtt::publish(const std::string &topic, const std::vector<uint8_t> &data, int qos) {
+void Mqtt::publish(const std::string& topic, const std::vector<uint8_t>& data, int qos) {
   std::unique_lock lock(m_mutex);
   if (m_messages.size() < QUEUE_MAX_SIZE) {
     m_messages.emplace(topic, data, qos);
     Logger::trace(LABEL, "queue size: {}", m_messages.size());
   }
 }
 
-void Mqtt::publish(const std::string &topic, const std::vector<uint8_t> &&data, int qos) {
+void Mqtt::publish(const std::string& topic, const std::vector<uint8_t>&& data, int qos) {
   std::unique_lock lock(m_mutex);
   if (m_messages.size() < QUEUE_MAX_SIZE) {
     m_messages.emplace(topic, std::move(data), qos);
     Logger::trace(LABEL, "queue size: {}", m_messages.size());
   }
 }
 
-void Mqtt::setMessageCallback(const std::string &topic, std::function<void(const std::string &)> callback) {
+void Mqtt::setMessageCallback(const std::string& topic, std::function<void(const std::string&)> callback) {
   subscribe(topic);
   m_callbacks.emplace_back(topic, callback);
 }
 
-void Mqtt::subscribe(const std::string &topic) {
-  if (m_topics.count(topic) == 0) {
-    mosquitto_subscribe(m_client, nullptr, topic.c_str(), QOS_SUB);
-    m_topics.insert(topic);
+void Mqtt::connect() {
+  mqtt::ssl_options ssl_options;
+  ssl_options.ca_path("/etc/ssl/certs");
+
+  const auto options = mqtt::connect_options_builder()
+                           .mqtt_version(MQTTVERSION_3_1_1)
+                           .ssl(ssl_options)
+                           .user_name(m_config.mqttUsername())
+                           .password(m_config.mqttPassword())
+                           .keep_alive_interval(KEEP_ALIVE)
+                           .connect_timeout(CONNECT_TIMEOUT)
+                           .automatic_reconnect(false)
+                           .clean_session(true)
+                           .finalize();
+
+  try {
+    const auto response = m_client.connect(options);
+    if (response.is_session_present()) {
+      Logger::info(LABEL, "session already present");
+    } else {
+      Logger::info(LABEL, "new session created");
+    }
+  } catch (const std::runtime_error& exception) {
+    Logger::warn(LABEL, "exception: {}", exception.what());
+  }
+  if (m_client.is_connected()) {
+    onConnected();
   }
 }
 
-void Mqtt::onConnect() {
+void Mqtt::onConnected() {
   Logger::info(LABEL, "connected");
-  for (const auto &topic : m_topics) {
+  std::unique_lock lock(m_mutex);
+
+  for (const auto& topic : m_topics) {
     Logger::info(LABEL, "subscribe: {}", colored(GREEN, "{}", topic));
-    mosquitto_subscribe(m_client, nullptr, topic.c_str(), QOS_SUB);
+    m_client.subscribe(topic, QOS_SUB);
   }
-}
 
-void Mqtt::onDisconnect() {
-  if (!m_isRunning) {
-    return;
-  }
-  Logger::warn(LABEL, "disconnected");
-  while (m_isRunning && mosquitto_reconnect(m_client) != MOSQ_ERR_SUCCESS) {
-    Logger::info(LABEL, "reconnecting");
-    std::this_thread::sleep_for(RECONNECT_INTERVAL);
+  for (const auto& topic : m_waitingTopics) {
+    Logger::info(LABEL, "subscribe: {}", colored(GREEN, "{}", topic));
+    m_client.subscribe(topic, QOS_SUB);
+    m_topics.insert(topic);
   }
-  Logger::info(LABEL, "reconnecting success");
+  m_waitingTopics.clear();
+}
+
+void Mqtt::onDisconnected() { Logger::info(LABEL, "disconnected"); }
+
+void Mqtt::subscribe(const std::string& topic) {
   std::unique_lock lock(m_mutex);
-  while (!m_messages.empty()) {
-    m_messages.pop();
+  if (m_client.is_connected()) {
+    if (m_topics.count(topic) == 0) {
+      Logger::info(LABEL, "subscribe: {}", colored(GREEN, "{}", topic));
+      m_client.subscribe(topic, QOS_SUB);
+      m_topics.insert(topic);
+    }
+  } else {
+    m_waitingTopics.insert(topic);
   }
 }
 
-void Mqtt::onMessage(const mosquitto_message *message) {
-  Logger::debug(LABEL, "topic: {}, data: {}", message->topic, static_cast<char *>(message->payload));
-  const std::string data(static_cast<char *>(message->payload), message->payloadlen);
-  for (auto &[topic, callback] : m_callbacks) {
-    if (strcmp(message->topic, topic.c_str()) == 0) {
+void Mqtt::onMessage(const std::string& topic, const std::string& data) {
+  Logger::debug(LABEL, "topic: {}, data: {}", topic, data);
+  for (auto& [callbackTopic, callback] : m_callbacks) {
+    if (topic == callbackTopic) {
       callback(data);
     }
   }
diff --git a/sources/network/mqtt.h b/sources/network/mqtt.h
@@ -1,7 +1,7 @@
 #pragma once
 
 #include <config.h>
-#include <mosquitto.h>
+#include <mqtt/client.h>
 
 #include <atomic>
 #include <functional>
@@ -23,16 +23,19 @@ class Mqtt {
   void setMessageCallback(const std::string& topic, std::function<void(const std::string&)> callback);
 
  private:
-  void onConnect();
-  void onDisconnect();
-  void onMessage(const mosquitto_message* message);
+  void connect();
+  void onConnected();
+  void onDisconnected();
+  void onMessage(const std::string& topic, const std::string& data);
   void subscribe(const std::string& topic);
 
-  mosquitto* m_client;
+  const Config& m_config;
+  mqtt::client m_client;
   std::atomic_bool m_isRunning;
   std::thread m_thread;
   std::mutex m_mutex;
   std::queue<std::tuple<std::string, std::vector<uint8_t>, int>> m_messages;
   std::set<std::string> m_topics;
+  std::set<std::string> m_waitingTopics;
   std::vector<std::pair<std::string, std::function<void(const std::string&)>>> m_callbacks;
 };
