dandri/MeshCore-mqtt-observer
1
0
Fork 0
mirror of https://github.com/agessaman/MeshCore.git synced 2026-07-12 13:08:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update MQTT presets to include 'cascadiamesh' and enhance error messaging in CLI. Introduce keepalive configuration for MQTT presets and improve slot management in MQTTBridge for better token handling and connection stability.

Browse Source
This commit is contained in:
agessaman
2026-03-24 16:58:26 -07:00
parent 1c98d1c605
commit eaf5fcbc9b
3 changed files with 44 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/helpers/CommonCLI.cpp
+1 -1
View File
@@ -1222,7 +1222,7 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch
sprintf(reply, "OK - slot %d preset: %s", slot + 1, preset_name);
}
} else {
strcpy(reply, "Error: valid presets are: analyzer-us, analyzer-eu, meshmapper, meshrank, waev, custom, none");
strcpy(reply, "Error: valid presets are: analyzer-us, analyzer-eu, meshmapper, meshrank, waev, cascadiamesh, custom, none");
}
} else if (memcmp(subcmd, "server ", 7) == 0) {
StrHelper::strncpy(_prefs->mqtt_slot_host[slot], &subcmd[7], sizeof(_prefs->mqtt_slot_host[slot]));
src/helpers/MQTTPresets.h
+8 -6
View File
@@ -22,6 +22,7 @@ struct MQTTPresetDef {
MQTTTopicStyle topic_style;
unsigned long token_lifetime; // JWT token lifetime in seconds (0 = use default 86400)
bool allow_retain; // Whether the broker allows the MQTT retain flag
uint16_t keepalive; // MQTT keepalive in seconds (0 = library default 120s)
};
// Google Trust Services - GTS Root R4 (used by LetsMesh Analyzer)
@@ -83,15 +84,16 @@ static const char ISRG_ROOT_X1[] PROGMEM =
"-----END CERTIFICATE-----\n";
// Number of built-in presets
static const int MQTT_PRESET_COUNT = 5;
static const int MQTT_PRESET_COUNT = 6;
// Built-in preset definitions (stored in flash)
static const MQTTPresetDef MQTT_PRESETS[MQTT_PRESET_COUNT] = {
{ "analyzer-us", "wss://mqtt-us-v1.letsmesh.net:443/mqtt", "mqtt-us-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true },
{ "analyzer-eu", "wss://mqtt-eu-v1.letsmesh.net:443/mqtt", "mqtt-eu-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true },
{ "meshmapper", "wss://mqtt.meshmapper.cc:443/mqtt", "mqtt.meshmapper.cc", ISRG_ROOT_X1, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true },
{ "meshrank", "mqtts://meshrank.net:8883", nullptr, ISRG_ROOT_X1, MQTT_AUTH_NONE, MQTT_TOPIC_MESHRANK, 0, false },
{ "waev", "wss://mqtt.waev.app:443/mqtt", "mqtt.waev.app", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 3300, false },
{ "analyzer-us", "wss://mqtt-us-v1.letsmesh.net:443/mqtt", "mqtt-us-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55 },
{ "analyzer-eu", "wss://mqtt-eu-v1.letsmesh.net:443/mqtt", "mqtt-eu-v1.letsmesh.net", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55 },
{ "meshmapper", "wss://mqtt.meshmapper.cc:443/mqtt", "mqtt.meshmapper.cc", ISRG_ROOT_X1, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 0, true, 55 },
{ "meshrank", "mqtts://meshrank.net:8883", nullptr, ISRG_ROOT_X1, MQTT_AUTH_NONE, MQTT_TOPIC_MESHRANK, 0, false, 0 },
{ "waev", "wss://mqtt.waev.app:443/mqtt", "mqtt.waev.app", GTS_ROOT_R4, MQTT_AUTH_JWT, MQTT_TOPIC_MESHCORE, 3300, false, 55 },
{ "cascadiamesh", "mqtt://cascadiamesh.ddns.net:1883", nullptr, nullptr, MQTT_AUTH_NONE, MQTT_TOPIC_MESHCORE, 0, true, 0 },
};
// Find a preset by name, returns nullptr if not found
src/helpers/bridges/MQTTBridge.cpp
+35 -11
View File
@@ -602,6 +602,15 @@ void MQTTBridge::mqttTaskLoop() {
syncTimeWithNTP();
}
// Retry NTP every 30s if initial sync failed (slots can't start without valid time)
if (!_ntp_synced && WiFi.status() == WL_CONNECTED) {
static unsigned long last_ntp_retry = 0;
if (now - last_ntp_retry >= 30000) {
last_ntp_retry = now;
syncTimeWithNTP();
}
}
// Deferred slot setup: wait until NTP is synced so JWT tokens get valid timestamps.
// This avoids wasted TLS handshakes that get rejected due to bad token times.
if (_ntp_synced && !_slots_setup_done) {
@@ -772,6 +781,9 @@ void MQTTBridge::setupSlot(int index) {
slot.client = new PsychicMqttClient();
slot.client->setAutoReconnect(false); // We handle reconnect with our own backoff logic
if (slot.preset && slot.preset->keepalive > 0) {
slot.client->setKeepAlive(slot.preset->keepalive);
}
optimizeMqttClientConfig(slot.client, slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT);
// Callbacks (capture index by value)
@@ -930,7 +942,6 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
if (createSlotAuthToken(index)) {
MQTT_DEBUG_PRINTLN("Slot %d token renewed", index);
slot.client->setCredentials(_jwt_username, slot.auth_token);
const unsigned long DISCONNECT_THRESHOLD = 60;
bool old_token_expired_or_imminent = !time_synced ||
@@ -939,16 +950,15 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
(time_synced && old_token_expires_at >= 1000000000 &&
current_time >= (old_token_expires_at - DISCONNECT_THRESHOLD));
if (old_token_expired_or_imminent && slot.client->connected()) {
slot.client->disconnect();
#ifdef ESP_PLATFORM
vTaskDelay(pdMS_TO_TICKS(1000 + index * 2000)); // stagger reconnects after token renewal
#endif
slot.last_reconnect_attempt = millis();
slot.client->connect();
} else if (!slot.client->connected()) {
slot.last_reconnect_attempt = now_millis;
slot.client->connect();
if (old_token_expired_or_imminent || !slot.client->connected()) {
// Clean teardown+setup ensures fresh TLS state and credentials
MQTT_DEBUG_PRINTLN("Slot %d token renewal: reconnecting with fresh client", index);
teardownSlot(index);
setupSlot(index);
reconnect_attempted = true;
} else {
// Token renewed but old one still valid — just update credentials for next reconnect
slot.client->setCredentials(_jwt_username, slot.auth_token);
}
} else {
MQTT_DEBUG_PRINTLN("Slot %d token renewal failed", index);
@@ -969,6 +979,13 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
slot.last_reconnect_attempt = now_millis;
reconnect_attempted = true;
MQTT_DEBUG_PRINTLN("Slot %d circuit breaker probe (attempting single reconnect after %lu ms)", index, probe_elapsed);
// Refresh JWT token before reconnecting (it may have expired while tripped)
if (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) {
if (createSlotAuthToken(index)) {
MQTT_DEBUG_PRINTLN("Slot %d token refreshed before circuit breaker probe", index);
slot.client->setCredentials(_jwt_username, slot.auth_token);
}
}
slot.client->connect();
// If the connect callback fires and sets slot.connected = true,
// the next maintenance cycle will clear the circuit breaker via the
@@ -1000,6 +1017,13 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
}
MQTT_DEBUG_PRINTLN("Slot %d reconnecting (backoff level %d, failures at max: %d)", index, slot.reconnect_backoff, slot.max_backoff_failures);
reconnect_attempted = true;
// Refresh JWT token before reconnecting (it may have expired during backoff)
if (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) {
if (createSlotAuthToken(index)) {
MQTT_DEBUG_PRINTLN("Slot %d token refreshed before reconnect", index);
slot.client->setCredentials(_jwt_username, slot.auth_token);
}
}
slot.client->connect();
}
}