diff --git a/src/modules/AdminModule.cpp b/src/modules/AdminModule.cpp index 9be681b42f..4fae5b1432 100644 --- a/src/modules/AdminModule.cpp +++ b/src/modules/AdminModule.cpp @@ -1112,7 +1112,7 @@ void AdminModule::handleSetConfig(const meshtastic_Config &c, bool fromOthers) // Default root is in use, so subscribe to the appropriate MQTT topic for this region snprintf(moduleConfig.mqtt.root, sizeof(moduleConfig.mqtt.root), "%s/%s", default_mqtt_root, myRegion->name); } - changes = SEGMENT_CONFIG | SEGMENT_MODULECONFIG; + changes |= SEGMENT_CONFIG | SEGMENT_MODULECONFIG; } // use_preset and bandwidth are coerced into valid values by the check. } diff --git a/src/mqtt/MQTT.cpp b/src/mqtt/MQTT.cpp index 8108ba050c..e8b9010f6a 100644 --- a/src/mqtt/MQTT.cpp +++ b/src/mqtt/MQTT.cpp @@ -374,6 +374,23 @@ void mqttInit() new MQTT(); } +void MQTT::reinitTopics() +{ + topicRoot = moduleConfig.mqtt.root; + const std::string root = *moduleConfig.mqtt.root ? moduleConfig.mqtt.root : default_mqtt_root; + cryptTopic = root + "/2/e/"; + mapTopic = root + "/2/map/"; + isConfiguredForDefaultRootTopic = isDefaultRootTopic(moduleConfig.mqtt.root); + +#if HAS_NETWORKING + // Force a broker reconnect so subscriptions are refreshed with the new topic prefix + if (pubSub.connected()) { + pubSub.disconnect(); + } + isConnected = false; +#endif +} + #if HAS_NETWORKING MQTT::MQTT() : MQTT(std::unique_ptr(new MQTTClient())) {} MQTT::MQTT(std::unique_ptr _mqttClient) @@ -388,15 +405,7 @@ MQTT::MQTT() : concurrency::OSThread("mqtt"), mqttQueue(MAX_MQTT_QUEUE) assert(!mqtt); mqtt = this; - if (*moduleConfig.mqtt.root) { - cryptTopic = moduleConfig.mqtt.root + cryptTopic; - mapTopic = moduleConfig.mqtt.root + mapTopic; - isConfiguredForDefaultRootTopic = isDefaultRootTopic(moduleConfig.mqtt.root); - } else { - cryptTopic = "msh" + cryptTopic; - mapTopic = "msh" + mapTopic; - isConfiguredForDefaultRootTopic = true; - } + reinitTopics(); if (moduleConfig.mqtt.map_reporting_enabled && moduleConfig.mqtt.has_map_report_settings) { map_position_precision = Default::getConfiguredOrDefault(moduleConfig.mqtt.map_report_settings.position_precision, @@ -572,6 +581,9 @@ int32_t MQTT::runOnce() { if (!moduleConfig.mqtt.enabled || !(moduleConfig.mqtt.map_reporting_enabled || channels.anyMqttEnabled())) return disable(); + // A region change rewrites the root at runtime, from several call sites + if (topicRoot != moduleConfig.mqtt.root) + reinitTopics(); bool wantConnection = wantsLink(); perhapsReportToMap(); @@ -762,6 +774,8 @@ void MQTT::onSend(const meshtastic_MeshPacket &mp_encrypted, const meshtastic_Me .channel_id = const_cast(channelId), .gateway_id = const_cast(nodeId.c_str())}; size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &meshtastic_ServiceEnvelope_msg, &env); + if (topicRoot != moduleConfig.mqtt.root) + reinitTopics(); // root changed before runOnce() noticed std::string topic = cryptTopic + channelId + "/" + nodeId; if (moduleConfig.mqtt.proxy_to_client_enabled || this->isConnectedDirectly()) { diff --git a/src/mqtt/MQTT.h b/src/mqtt/MQTT.h index 2851efc6dd..5a905433c8 100644 --- a/src/mqtt/MQTT.h +++ b/src/mqtt/MQTT.h @@ -98,6 +98,7 @@ class MQTT : private concurrency::OSThread explicit MQTT(std::unique_ptr mqttClient); #endif + std::string topicRoot; // moduleConfig.mqtt.root the topics below were built from std::string cryptTopic = "/2/e/"; // msh/2/e/CHANNELID/NODEID std::string mapTopic = "/2/map/"; // For protobuf-encoded MapReport messages @@ -111,6 +112,9 @@ class MQTT : private concurrency::OSThread */ void reconnect(); + /// Rebuild topics from moduleConfig.mqtt.root and force a resubscribe. + void reinitTopics(); + /** Tell the server what subscriptions we want (based on channels.downlink_enabled) */ void sendSubscriptions(); diff --git a/test/test_mqtt/MQTT.cpp b/test/test_mqtt/MQTT.cpp index 4d2ea6c268..0229c0fe59 100644 --- a/test/test_mqtt/MQTT.cpp +++ b/test/test_mqtt/MQTT.cpp @@ -1360,6 +1360,35 @@ void test_customMqttRoot(void) [] { return pubsub->subscriptions_.count("custom/2/e/test/+") && pubsub->subscriptions_.count("custom/2/e/PKI/+"); })); } +// A LoRa region change rewrites moduleConfig.mqtt.root without telling MQTT (AdminModule, MenuHandler, +// InkHUD). MQTT must pick it up, rebuild the topics and resubscribe; otherwise the node keeps +// publishing and subscribing under the old region's root until reboot. An uplink sent before +// runOnce() runs must already use the new root. +void test_rootChange_rebuildsTopics(void) +{ + // Start MQTT with a US region root. + strcpy(moduleConfig.mqtt.root, "msh/US"); + MQTTUnitTest::restart(); + + TEST_ASSERT_TRUE(loopUntil( + [] { return pubsub->subscriptions_.count("msh/US/2/e/test/+") && pubsub->subscriptions_.count("msh/US/2/e/PKI/+"); })); + + // Simulate region change: only the root changes, nobody notifies MQTT. + strcpy(moduleConfig.mqtt.root, "msh/EU_868"); + pubsub->subscriptions_.clear(); + pubsub->published_.clear(); + mqtt->onSend(encrypted, decoded, 0); + + // Subscriptions are refreshed, and the uplink is published under the new root after the reconnect. + TEST_ASSERT_TRUE(loopUntil([] { + return pubsub->subscriptions_.count("msh/EU_868/2/e/test/+") && pubsub->subscriptions_.count("msh/EU_868/2/e/PKI/+") && + !pubsub->published_.empty(); + })); + TEST_ASSERT_EQUAL(1, pubsub->published_.size()); + const auto &[topic, payload] = pubsub->published_.front(); + TEST_ASSERT_EQUAL_STRING("msh/EU_868/2/e/test/!12345678", topic.c_str()); +} + // Empty configuration is valid. void test_configEmptyIsValid(void) { @@ -1555,6 +1584,7 @@ void setup() RUN_TEST(test_disabled); RUN_TEST(test_mqttInitSkipsAllocationWhenDisabled); RUN_TEST(test_customMqttRoot); + RUN_TEST(test_rootChange_rebuildsTopics); RUN_TEST(test_configEmptyIsValid); RUN_TEST(test_configEnabledEmptyIsValid); RUN_TEST(test_configWithDefaultServer);