MQTT messages working

This commit is contained in:
pgdalmeida 2026-08-26 17:56:56 +02:00
parent f1342965de
commit ef96513b58
Signed by: pedro.almeida
GPG Key ID: D4A6C394DF13F1D7
9 changed files with 118 additions and 12 deletions

View File

@ -18,6 +18,10 @@ debug = 0
[Network] [Network]
; MQTT broker address (the ground-station / ZKMS server). ; MQTT broker address (the ground-station / ZKMS server).
zkms_server_ip = 127.0.0.1 zkms_server_ip = 127.0.0.1
; Broker port. 1883 is the MQTT default (RabbitMQ's MQTT plugin included); 8883 for
; TLS. Ignored if zkms_server_ip already carries a port ("host:1884") or is a full
; URI ("tcp://host:1884") - in both cases that value wins.
mqtt_port = 1883
; Prefer the FGC_MQTT_USER / FGC_MQTT_PW environment variables over these. ; Prefer the FGC_MQTT_USER / FGC_MQTT_PW environment variables over these.
mqtt_user = CHANGE_ME mqtt_user = CHANGE_ME
mqtt_pw = CHANGE_ME mqtt_pw = CHANGE_ME

View File

@ -29,7 +29,8 @@ Parsed and validated by `ConfigLoader` ([src/core/Config.cpp](../src/core/Config
| `General` | `tower_name` | string | `Unnamed` | Tower identity; used in all MQTT topics/payloads | | `General` | `tower_name` | string | `Unnamed` | Tower identity; used in all MQTT topics/payloads |
| `General` | `image_interval` | int > 0 | `5` | Seconds between captures (→ `image_rate = 1/interval`) | | `General` | `image_interval` | int > 0 | `5` | Seconds between captures (→ `image_rate = 1/interval`) |
| `General` | `debug` | bool | `false` | Start with debug-level logging | | `General` | `debug` | bool | `false` | Start with debug-level logging |
| `Network` | `zkms_server_ip` | string | `127.0.0.1` | MQTT broker address | | `Network` | `zkms_server_ip` | string | `127.0.0.1` | MQTT broker host/IP, or a full `tcp://…` URI |
| `Network` | `mqtt_port` | int 1..65535 | `1883` | Broker port (`8883` for TLS). Ignored when `zkms_server_ip` carries its own port or a scheme |
| `Network` | `mqtt_user` / `mqtt_pw` | string | — | MQTT credentials (see secrets below) | | `Network` | `mqtt_user` / `mqtt_pw` | string | — | MQTT credentials (see secrets below) |
| `Serial` | `device` | string | `/dev/ttyACM0` | Motor-controller serial device | | `Serial` | `device` | string | `/dev/ttyACM0` | Motor-controller serial device |
| `Serial` | `baud` | int | `115200` | Serial baud rate | | `Serial` | `baud` | int | `115200` | Serial baud rate |

View File

@ -15,7 +15,11 @@ are **retained**.
## Connection ## Connection
- Broker URI = `Network.zkms_server_ip` from `config.ini`; client ID = the tower name. - Broker URI is assembled from `Network.zkms_server_ip` + `Network.mqtt_port` as `tcp://<host>:<port>`
(`NetworkConfig::brokerUri`), and logged on every connect and connect failure. Two escape hatches, so
existing configs keep working: a `zkms_server_ip` containing `://` is passed through verbatim, and one
already of the form `host:port` keeps that port. In both cases `mqtt_port` is ignored. Client ID = the
tower name.
- Auth: `mqtt_user` / `mqtt_pw` (preferring `$FGC_MQTT_USER`/`$FGC_MQTT_PW`); `clean_session = true`, - Auth: `mqtt_user` / `mqtt_pw` (preferring `$FGC_MQTT_USER`/`$FGC_MQTT_PW`); `clean_session = true`,
keep-alive 20 s, `set_automatic_reconnect(true)`. keep-alive 20 s, `set_automatic_reconnect(true)`.
- Connect timeout 5 s. On connection loss Paho auto-reconnects; the channel re-subscribes on reconnect. - Connect timeout 5 s. On connection loss Paho auto-reconnects; the channel re-subscribes on reconnect.

View File

@ -19,9 +19,22 @@ struct GeneralConfig {
}; };
struct NetworkConfig { struct NetworkConfig {
std::string broker_ip = "127.0.0.1"; // MQTT broker / ZKMS server std::string broker_ip = "127.0.0.1"; // MQTT broker host/IP, or a full tcp://... URI
int mqtt_port = 1883; // ignored when broker_ip carries its own port
std::string mqtt_user; // see secrets note below std::string mqtt_user; // see secrets note below
std::string mqtt_pw; std::string mqtt_pw;
// Paho serverURI for broker_ip + mqtt_port, e.g. "tcp://10.0.0.5:1883".
//
// Built here rather than handing Paho a bare address and relying on its
// default port, so the port is explicit, configurable and visible in the log.
// Two precedence rules, both to keep existing configs working:
// - broker_ip containing "://" is a URI the operator owns: returned verbatim,
// mqtt_port ignored (it may already have a port, or a ws:// path to break).
// - broker_ip already carrying "host:port" keeps that port; mqtt_port ignored.
// A bare IPv6 address is bracketed, or Paho's rightmost-colon parse would read
// the last hextet as the port.
std::string brokerUri() const;
}; };
struct SerialConfig { struct SerialConfig {

View File

@ -22,7 +22,9 @@ class MqttControlChannel : public IControlChannel,
public virtual mqtt::callback, public virtual mqtt::callback,
public virtual mqtt::iaction_listener { public virtual mqtt::iaction_listener {
public: public:
MqttControlChannel(const std::string& broker_ip, const std::string& tower, // broker_uri: a full Paho serverURI, e.g. "tcp://10.0.0.5:1883"
// (see NetworkConfig::brokerUri).
MqttControlChannel(const std::string& broker_uri, const std::string& tower,
const std::string& user, const std::string& password); const std::string& user, const std::string& password);
~MqttControlChannel() override; ~MqttControlChannel() override;
@ -46,6 +48,7 @@ private:
void publish(const std::string& topic, const std::string& payload); void publish(const std::string& topic, const std::string& payload);
std::string broker_uri_;
std::string tower_; std::string tower_;
std::string topic_target_hdg_; std::string topic_target_hdg_;
std::string topic_control_code_; std::string topic_control_code_;

View File

@ -244,7 +244,7 @@ struct Application::Impl {
bool want = opts.use_mqtt.value_or(cfg.features.enable_mqtt); bool want = opts.use_mqtt.value_or(cfg.features.enable_mqtt);
#if FGC_WITH_MQTT #if FGC_WITH_MQTT
if (want) { if (want) {
return std::make_unique<MqttControlChannel>(cfg.network.broker_ip, return std::make_unique<MqttControlChannel>(cfg.network.brokerUri(),
cfg.general.tower_name, cfg.general.tower_name,
cfg.network.mqtt_user, cfg.network.mqtt_user,
cfg.network.mqtt_pw); cfg.network.mqtt_pw);
@ -572,7 +572,7 @@ struct Application::Impl {
// --- Connectivity --- // --- Connectivity ---
s.conn.mqtt_enabled = opts.use_mqtt.value_or(cfg.features.enable_mqtt); s.conn.mqtt_enabled = opts.use_mqtt.value_or(cfg.features.enable_mqtt);
s.conn.mqtt_connected = channel && channel->connected(); s.conn.mqtt_connected = channel && channel->connected();
s.conn.broker = cfg.network.broker_ip; s.conn.broker = cfg.network.brokerUri();
s.conn.tower = cfg.general.tower_name; s.conn.tower = cfg.general.tower_name;
s.conn.control_code = scheduler ? scheduler->controlCode() : 0; s.conn.control_code = scheduler ? scheduler->controlCode() : 0;
s.conn.target_heading = scheduler ? scheduler->targetHeading() : "0"; s.conn.target_heading = scheduler ? scheduler->targetHeading() : "0";

View File

@ -1,6 +1,7 @@
#include "fgc/Config.h" #include "fgc/Config.h"
#include "fgc/Paths.h" #include "fgc/Paths.h"
#include <algorithm>
#include <cctype> #include <cctype>
#include <cstdio> #include <cstdio>
#include <cstdlib> #include <cstdlib>
@ -27,7 +28,13 @@ int getInt(const std::map<std::string, std::string>& kv, const std::string& key,
auto it = kv.find(key); auto it = kv.find(key);
if (it == kv.end() || it->second.empty()) return fallback; if (it == kv.end() || it->second.empty()) return fallback;
try { try {
return std::stoi(it->second); // Reject trailing characters: bare std::stoi stops at the first non-digit,
// so "1883x" would quietly become 1883 and a typo'd value would configure
// something the operator never wrote. Same reasoning as getIntAutoBase below.
size_t consumed = 0;
int out = std::stoi(it->second, &consumed);
if (consumed != it->second.size()) throw std::invalid_argument("trailing characters");
return out;
} catch (const std::exception&) { } catch (const std::exception&) {
throw std::runtime_error("Config key '" + key + "' must be an integer, got '" + throw std::runtime_error("Config key '" + key + "' must be an integer, got '" +
it->second + "'"); it->second + "'");
@ -101,6 +108,29 @@ int iniHandler(void* user, const char* section, const char* name, const char* va
} // namespace } // namespace
std::string NetworkConfig::brokerUri() const {
// A scheme means the operator wrote the whole URI: hands off. Appending a port
// could corrupt one that already has it, or a ws:// path.
if (broker_ip.find("://") != std::string::npos) return broker_ip;
std::string host = broker_ip;
bool has_port = false;
if (!host.empty() && host.front() == '[') { // bracketed IPv6
const size_t close = host.rfind(']');
has_port = close != std::string::npos && host.find(':', close) != std::string::npos;
} else {
const size_t colons = static_cast<size_t>(std::count(host.begin(), host.end(), ':'));
if (colons == 1) {
has_port = true;
} else if (colons > 1) {
host = "[" + host + "]"; // bare IPv6; bracket it so the port is unambiguous
}
}
if (has_port) return "tcp://" + host;
return "tcp://" + host + ":" + std::to_string(mqtt_port);
}
double AppConfig::image_rate() const { double AppConfig::image_rate() const {
return general.image_interval > 0 ? 1.0 / general.image_interval : 0.0; return general.image_interval > 0 ? 1.0 / general.image_interval : 0.0;
} }
@ -152,6 +182,9 @@ AppConfig ConfigLoader::fromMap(const std::map<std::string, std::string>& kv) {
cfg.general.debug = getBool(kv, "General.debug", cfg.general.debug); cfg.general.debug = getBool(kv, "General.debug", cfg.general.debug);
cfg.network.broker_ip = get(kv, "Network.zkms_server_ip", cfg.network.broker_ip); cfg.network.broker_ip = get(kv, "Network.zkms_server_ip", cfg.network.broker_ip);
cfg.network.mqtt_port = getInt(kv, "Network.mqtt_port", cfg.network.mqtt_port);
if (cfg.network.mqtt_port < 1 || cfg.network.mqtt_port > 65535)
throw std::runtime_error("Network.mqtt_port must be in 1..65535");
// Secrets: environment variables win over the file. // Secrets: environment variables win over the file.
cfg.network.mqtt_user = envOr("FGC_MQTT_USER", get(kv, "Network.mqtt_user")); cfg.network.mqtt_user = envOr("FGC_MQTT_USER", get(kv, "Network.mqtt_user"));
cfg.network.mqtt_pw = envOr("FGC_MQTT_PW", get(kv, "Network.mqtt_pw")); cfg.network.mqtt_pw = envOr("FGC_MQTT_PW", get(kv, "Network.mqtt_pw"));

View File

@ -12,15 +12,16 @@ const auto kConnectTimeout = std::chrono::seconds(5);
const auto kPublishTimeout = std::chrono::seconds(3); const auto kPublishTimeout = std::chrono::seconds(3);
} // namespace } // namespace
MqttControlChannel::MqttControlChannel(const std::string& broker_ip, const std::string& tower, MqttControlChannel::MqttControlChannel(const std::string& broker_uri, const std::string& tower,
const std::string& user, const std::string& password) const std::string& user, const std::string& password)
: tower_(tower), : broker_uri_(broker_uri),
tower_(tower),
topic_target_hdg_("GGS/FWT/" + tower + "/target_HDG"), topic_target_hdg_("GGS/FWT/" + tower + "/target_HDG"),
topic_control_code_("GGS/FWT/" + tower + "/ControlCode"), topic_control_code_("GGS/FWT/" + tower + "/ControlCode"),
topic_status_("GGS/FWT/" + tower + "/StatusCode"), topic_status_("GGS/FWT/" + tower + "/StatusCode"),
topic_cam_event_("GGS/FWT/" + tower + "/CamEvent"), topic_cam_event_("GGS/FWT/" + tower + "/CamEvent"),
topic_env_("GGS/FWT/" + tower + "/Env"), topic_env_("GGS/FWT/" + tower + "/Env"),
client_(broker_ip, tower) { client_(broker_uri, tower) {
conn_opts_.set_keep_alive_interval(20); conn_opts_.set_keep_alive_interval(20);
conn_opts_.set_clean_session(true); conn_opts_.set_clean_session(true);
conn_opts_.set_automatic_reconnect(true); // replaces the old manual reconnect loop conn_opts_.set_automatic_reconnect(true); // replaces the old manual reconnect loop
@ -36,10 +37,11 @@ bool MqttControlChannel::connect() {
client_.connect(conn_opts_, nullptr, *this)->wait_for(kConnectTimeout); client_.connect(conn_opts_, nullptr, *this)->wait_for(kConnectTimeout);
connected_ = client_.is_connected(); connected_ = client_.is_connected();
} catch (const mqtt::exception& e) { } catch (const mqtt::exception& e) {
LOG_ERROR << "MQTT connect failed: " << e.what(); LOG_ERROR << "MQTT connect failed (" << broker_uri_ << "): " << e.what();
connected_ = false; connected_ = false;
} }
if (connected_) LOG_INFO << "MQTT connected to broker as '" << tower_ << "'"; if (connected_)
LOG_INFO << "MQTT connected to " << broker_uri_ << " as '" << tower_ << "'";
return connected_; return connected_;
} }

View File

@ -84,6 +84,52 @@ TEST_CASE("ConfigLoader validates input") {
CHECK_THROWS(ConfigLoader::fromMap({{"General.debug", "maybe"}})); CHECK_THROWS(ConfigLoader::fromMap({{"General.debug", "maybe"}}));
} }
TEST_CASE("brokerUri builds a Paho serverURI from host + port") {
// The port used to be whatever Paho defaulted to when handed a bare address.
// It is now explicit, so these cases pin the whole resolution table.
auto uri = [](const std::string& host, int port) {
NetworkConfig n;
n.broker_ip = host;
n.mqtt_port = port;
return n.brokerUri();
};
CHECK(uri("10.11.12.20", 1883) == "tcp://10.11.12.20:1883");
CHECK(uri("broker.local", 8883) == "tcp://broker.local:8883");
// A port already in the host wins - existing configs that smuggled one in
// there must keep working.
CHECK(uri("10.11.12.20:1884", 1883) == "tcp://10.11.12.20:1884");
// A scheme means the operator owns the whole string: verbatim, port ignored.
// Appending to a ws:// path would corrupt it.
CHECK(uri("tcp://host:1883", 9999) == "tcp://host:1883");
CHECK(uri("ws://host/mqtt", 9999) == "ws://host/mqtt");
CHECK(uri("ssl://host:8883", 1883) == "ssl://host:8883");
// IPv6: bracketed passes through, bare gets bracketed - otherwise Paho's
// rightmost-colon parse reads the last hextet as the port.
CHECK(uri("[::1]", 1883) == "tcp://[::1]:1883");
CHECK(uri("[::1]:1884", 1883) == "tcp://[::1]:1884");
CHECK(uri("::1", 1883) == "tcp://[::1]:1883");
CHECK(uri("fe80::1234:5678", 1883) == "tcp://[fe80::1234:5678]:1883");
// The default config resolves to the MQTT default port.
CHECK(ConfigLoader::fromMap({}).network.brokerUri() == "tcp://127.0.0.1:1883");
CHECK(ConfigLoader::fromMap({{"Network.zkms_server_ip", "10.0.0.5"},
{"Network.mqtt_port", "8883"}})
.network.brokerUri() == "tcp://10.0.0.5:8883");
}
TEST_CASE("Network.mqtt_port is range-checked") {
CHECK(ConfigLoader::fromMap({}).network.mqtt_port == 1883);
CHECK(ConfigLoader::fromMap({{"Network.mqtt_port", "8883"}}).network.mqtt_port == 8883);
CHECK_THROWS(ConfigLoader::fromMap({{"Network.mqtt_port", "0"}}));
CHECK_THROWS(ConfigLoader::fromMap({{"Network.mqtt_port", "65536"}}));
CHECK_THROWS(ConfigLoader::fromMap({{"Network.mqtt_port", "-1"}}));
CHECK_THROWS(ConfigLoader::fromMap({{"Network.mqtt_port", "1883x"}}));
}
TEST_CASE("Camera mode selection + encoding config: defaults and overrides") { TEST_CASE("Camera mode selection + encoding config: defaults and overrides") {
AppConfig d = ConfigLoader::fromMap({}); AppConfig d = ConfigLoader::fromMap({});
CHECK(d.camera.mode == "standard"); CHECK(d.camera.mode == "standard");