From 015a9cec4af2c5d374655e17a8c5fb4e56ed5e54 Mon Sep 17 00:00:00 2001 From: "Michael A. Cojocari" Date: Sun, 30 Aug 2026 15:05:09 -0400 Subject: [PATCH] mqtt: add connection status command --- CONSOLE_MODE.md | 2 +- src/MyMesh.cpp | 116 ++++++++++++++++++ src/helpers/esp32/MqttBridge.cpp | 92 ++++++++++++-- src/helpers/esp32/MqttBridge.h | 34 +++++ .../esp32/WebSocketCompanionServer.cpp | 4 +- src/ui-touch/ConsoleUI.cpp | 3 +- src/ui-touch/UITask.cpp | 5 +- 7 files changed, 243 insertions(+), 13 deletions(-) diff --git a/CONSOLE_MODE.md b/CONSOLE_MODE.md index 8e5e0f6..931469c 100644 --- a/CONSOLE_MODE.md +++ b/CONSOLE_MODE.md @@ -366,7 +366,7 @@ which is why the console needed far fewer new commands than the UI has screens. | Quick replies | none | gap | | Backups | none | gap | | Language | none | gap | -| MQTT bridge | none | gap | +| MQTT bridge | `mqtt status` | **done** (diagnostics only) | | App permissions | none | with Phase 4 | | General | `console`/`ui` toggle | partial | diff --git a/src/MyMesh.cpp b/src/MyMesh.cpp index b769e62..79172e3 100644 --- a/src/MyMesh.cpp +++ b/src/MyMesh.cpp @@ -280,12 +280,101 @@ static const char* kMeshcomodHelpMsg = "set tx set TX power\n" "wifi status|on|off|scan\n" "wifi use | set ssid | set pwd | apply | clear\n" + "mqtt status\n" "tcp status|on|off\n" "ble status|on|off\n" "ota status|start|netdiag | ota url \n" "reboot restart the device\n" "bootloader / dfu reboot to download mode"; +#if defined(ESP32) && defined(MULTI_TRANSPORT_COMPANION) +static const char* mqttResultText(int result) { + switch (result) { + case -4: return "connection timeout"; + case -3: return "connection lost"; + case -2: return "TCP connect failed"; + case -1: return "disconnected"; + case 0: return "connected"; + case 1: return "bad protocol"; + case 2: return "rejected client ID"; + case 3: return "broker unavailable"; + case 4: return "bad credentials"; + case 5: return "unauthorized"; + default: return "unknown"; + } +} + +static void mqttFormatAge(char* out, size_t cap, uint32_t timestamp, uint32_t now) { + if (!timestamp) { snprintf(out, cap, "never"); return; } + const uint32_t seconds = (uint32_t)(now - timestamp) / 1000u; + if (seconds < 60u) snprintf(out, cap, "%lus ago", (unsigned long)seconds); + else if (seconds < 3600u) snprintf(out, cap, "%lum %lus ago", + (unsigned long)(seconds / 60u), + (unsigned long)(seconds % 60u)); + else snprintf(out, cap, "%luh %lum ago", (unsigned long)(seconds / 3600u), + (unsigned long)((seconds / 60u) % 60u)); +} + +static void mqttFormatStatus(char* out, size_t cap) { + MqttBridgeStatus status; + mqtt_bridge.getStatus(status); + if (!status.available) { + snprintf(out, cap, "mqtt: unavailable on this build"); + return; + } + if (status.phase == MqttBridgePhase::Disabled) { + snprintf(out, cap, "mqtt: disabled"); + return; + } + if (status.phase == MqttBridgePhase::Misconfigured) { + snprintf(out, cap, "mqtt: misconfigured (broker host is empty)"); + return; + } + + const uint32_t now = millis(); + char state[64]; + char timing[96] = {}; + switch (status.phase) { + case MqttBridgePhase::WaitingForWifi: + snprintf(state, sizeof(state), "mqtt: waiting for Wi-Fi"); + break; + case MqttBridgePhase::Connecting: { + char age[32]; + mqttFormatAge(age, sizeof(age), status.lastAttemptMs, now); + snprintf(state, sizeof(state), "mqtt: connecting"); + snprintf(timing, sizeof(timing), "attempt: %s", age); + break; + } + case MqttBridgePhase::Connected: { + char age[32]; + mqttFormatAge(age, sizeof(age), status.lastConnectedMs, now); + snprintf(state, sizeof(state), "mqtt: connected"); + snprintf(timing, sizeof(timing), "connected: %s", age); + break; + } + case MqttBridgePhase::RetryWait: { + const int32_t remaining = (int32_t)(status.retryAtMs - now); + const uint32_t seconds = remaining > 0 ? ((uint32_t)remaining + 999u) / 1000u : 0u; + snprintf(state, sizeof(state), "mqtt: retrying (%lus)", (unsigned long)seconds); + if (status.lastResultMs) { + char age[32]; + mqttFormatAge(age, sizeof(age), status.lastResultMs, now); + snprintf(timing, sizeof(timing), "last error: %d (%s), %s", + (int)status.lastResult, mqttResultText(status.lastResult), age); + } + break; + } + default: + snprintf(state, sizeof(state), "mqtt: unavailable on this build"); + break; + } + snprintf(out, cap, "%s\nbroker: %s:%u\npublish: channels=%s dm=%s encrypted=%s%s%s", + state, status.host, (unsigned)status.port, + status.publishChannel ? "on" : "off", status.publishDm ? "on" : "off", + status.encrypted ? "yes" : "no", timing[0] ? "\n" : "", timing); +} +#endif + #define MESHCOMOD_CMD_CACHE_SIZE 6 struct MeshcomodCmdCacheEntry { uint32_t ts; @@ -698,6 +787,23 @@ bool MyMesh::handleMeshcomodCommand(const char* text, int text_len) { return true; } + if (isCmd(p, "mqtt")) { + const char* q = p + 4; + while (*q == ' ' || *q == '\t') q++; + if (isCmd(q, "status")) { +#if defined(ESP32) && defined(MULTI_TRANSPORT_COMPANION) + char reply[320]; + mqttFormatStatus(reply, sizeof(reply)); + pushMeshcomodReply(reply); +#else + pushMeshcomodReply("mqtt: unavailable on this build"); +#endif + return true; + } + pushMeshcomodReply("usage:\n- mqtt status"); + return true; + } + if (strncasecmp(p, "tcp", 3) == 0 && (p[3] == '\0' || p[3] == ' ' || p[3] == '\t')) { p += 3; while (*p == ' ' || *p == '\t') p++; @@ -5219,6 +5325,16 @@ void MyMesh::checkCLIRescueCmd() { } else { Serial.printf(" Error: unknown config: %s\n", config); } + } else if (strcmp(cli_command, "mqtt.status") == 0 || + strcmp(cli_command, "get mqtt.status") == 0 || + strcmp(cli_command, "mqtt status") == 0) { +#if defined(ESP32) && defined(MULTI_TRANSPORT_COMPANION) + char reply[320]; + mqttFormatStatus(reply, sizeof(reply)); + Serial.printf(" > %s\n", reply); +#else + Serial.println(" > mqtt: unavailable on this build"); +#endif } else if (strcmp(cli_command, "get wifi.ssid") == 0) { #ifdef ESP32 #if defined(WIFI_SSID) || defined(MULTI_TRANSPORT_COMPANION) diff --git a/src/helpers/esp32/MqttBridge.cpp b/src/helpers/esp32/MqttBridge.cpp index 0ed2093..0422844 100644 --- a/src/helpers/esp32/MqttBridge.cpp +++ b/src/helpers/esp32/MqttBridge.cpp @@ -36,17 +36,52 @@ size_t MqttNbClient::write(const uint8_t* buf, size_t size) { return WiFiClient::write(buf, size); } +void MqttBridge::getStatus(MqttBridgeStatus& out) const { + portENTER_CRITICAL(&_statusMux); + out = _status; + portEXIT_CRITICAL(&_statusMux); +} + +void MqttBridge::syncStatusConfig(MqttBridgePhase phase) { + MqttBridgeStatus next; + next.phase = phase; + next.available = true; + next.requestedEnabled = _requestedEnabled; + next.publishDm = _pubDm; + next.publishChannel = _pubChannel; + next.encrypted = _encOn; + strncpy(next.host, _host, sizeof(next.host) - 1); + next.port = _port; + portENTER_CRITICAL(&_statusMux); + _status = next; + portEXIT_CRITICAL(&_statusMux); +} + +void MqttBridge::setStatusPhase(MqttBridgePhase phase, int result, uint32_t now, + bool recordResult, uint32_t retryAt) { + portENTER_CRITICAL(&_statusMux); + _status.phase = phase; + _status.retryAtMs = retryAt; + if (phase == MqttBridgePhase::Connecting) _status.lastAttemptMs = now; + if (phase == MqttBridgePhase::Connected) _status.lastConnectedMs = now; + if (recordResult) { + _status.lastResult = (int8_t)result; + _status.lastResultMs = now; + } + portEXIT_CRITICAL(&_statusMux); +} + // Read every field from the "mqtt" NVS namespace into the members. isKey() guards // avoid the [E] NOT_FOUND log spam that getString emits for absent keys on the // USB-CDC companion stream (see TouchPrefsStore prefsGetStr note). void MqttBridge::loadConfig() { - _enabled = false; _pubDm = false; _pubChannel = true; + _requestedEnabled = false; _enabled = false; _pubDm = false; _pubChannel = true; _host[0] = _user[0] = _pwd[0] = _psk[0] = '\0'; _port = 1883; SdNvsPrefs p; if (p.begin("mqtt", true)) { - _enabled = p.getBool("en", false); + _requestedEnabled = p.getBool("en", false); _port = (uint16_t)p.getUInt("port", 1883); _pubDm = p.getBool("dm", false); _pubChannel = p.getBool("ch", true); @@ -57,7 +92,13 @@ void MqttBridge::loadConfig() { p.end(); } deriveKey(); - if (!_enabled || _host[0] == '\0') _enabled = false; + _enabled = _requestedEnabled && _host[0] != '\0'; + MqttBridgePhase phase = MqttBridgePhase::Disabled; + if (_requestedEnabled && _host[0] == '\0') phase = MqttBridgePhase::Misconfigured; + else if (_enabled) phase = WiFi.status() == WL_CONNECTED + ? MqttBridgePhase::RetryWait + : MqttBridgePhase::WaitingForWifi; + syncStatusConfig(phase); } // Derive the AES key from the passphrase. mbedtls_md() (generic digest) is used @@ -88,7 +129,12 @@ void MqttBridge::begin(const char* nodeHex) { } bool MqttBridge::reconnect() { - if (!_enabled || WiFi.status() != WL_CONNECTED) return false; + if (!_enabled) return false; + if (WiFi.status() != WL_CONNECTED) { + setStatusPhase(MqttBridgePhase::WaitingForWifi, MQTT_DISCONNECTED, + millis(), false); + return false; + } char clientId[32], lwtTopic[80]; snprintf(clientId, sizeof(clientId), "wadamesh-%s", _nodeHex); @@ -98,11 +144,21 @@ bool MqttBridge::reconnect() { ? _mqtt.connect(clientId, _user, _pwd, lwtTopic, 0, true, "offline") : _mqtt.connect(clientId, nullptr, nullptr, lwtTopic, 0, true, "offline"); + const uint32_t now = millis(); + const int result = _mqtt.state(); if (ok) { + setStatusPhase(MqttBridgePhase::Connected, result, now, true); _mqtt.publish(lwtTopic, "online", true); Serial.printf("[MQTT] connected as %s\n", clientId); } else { - Serial.printf("[MQTT] connect failed, rc=%d\n", _mqtt.state()); + MqttBridgeStatus current; + getStatus(current); + const uint32_t retryAt = current.lastAttemptMs + ? current.lastAttemptMs + RECONNECT_INTERVAL_MS + : now + RECONNECT_INTERVAL_MS; + setStatusPhase(MqttBridgePhase::RetryWait, result, now, true, + retryAt); + Serial.printf("[MQTT] connect failed, rc=%d\n", result); } return ok; } @@ -119,19 +175,41 @@ void MqttBridge::reconnectTask(void* arg) { void MqttBridge::loop() { if (!_enabled || _connecting) return; + uint32_t now = millis(); + if (WiFi.status() != WL_CONNECTED) { + if (_mqtt.connected()) _mqtt.loop(); + setStatusPhase(MqttBridgePhase::WaitingForWifi, MQTT_DISCONNECTED, + now, false); + return; + } if (!_mqtt.connected()) { - uint32_t now = millis(); + MqttBridgeStatus current; + getStatus(current); + if (current.phase == MqttBridgePhase::Connected) { + setStatusPhase(MqttBridgePhase::RetryWait, _mqtt.state(), now, true, + now); + } if ((uint32_t)(now - _lastReconnectMs) >= RECONNECT_INTERVAL_MS) { _lastReconnectMs = now; _connecting = true; + setStatusPhase(MqttBridgePhase::Connecting, MQTT_DISCONNECTED, + now, false); if (xTaskCreatePinnedToCore(reconnectTask, "mqtt_conn", 6144, this, 1, nullptr, 0) != pdPASS) { _connecting = false; // task OOM — try again next interval + setStatusPhase(MqttBridgePhase::RetryWait, MQTT_DISCONNECTED, + now, false, now + RECONNECT_INTERVAL_MS); } + } else { + setStatusPhase(MqttBridgePhase::RetryWait, MQTT_DISCONNECTED, + now, false, _lastReconnectMs + RECONNECT_INTERVAL_MS); } return; } - _mqtt.loop(); + if (!_mqtt.loop()) { + setStatusPhase(MqttBridgePhase::RetryWait, _mqtt.state(), now, true, + now); + } } // AES-256-GCM seal: out = base64( nonce[12] || ciphertext || tag[16] ). Single-task diff --git a/src/helpers/esp32/MqttBridge.h b/src/helpers/esp32/MqttBridge.h index 751eed7..18997ac 100644 --- a/src/helpers/esp32/MqttBridge.h +++ b/src/helpers/esp32/MqttBridge.h @@ -3,6 +3,32 @@ #include +enum class MqttBridgePhase : uint8_t { + Unavailable, + Disabled, + Misconfigured, + WaitingForWifi, + Connecting, + Connected, + RetryWait, +}; + +struct MqttBridgeStatus { + MqttBridgePhase phase = MqttBridgePhase::Unavailable; + bool available = false; + bool requestedEnabled = false; + bool publishDm = false; + bool publishChannel = false; + bool encrypted = false; + char host[64] = {}; + uint16_t port = 1883; + int8_t lastResult = -1; + uint32_t lastAttemptMs = 0; + uint32_t lastConnectedMs = 0; + uint32_t lastResultMs = 0; + uint32_t retryAtMs = 0; +}; + #if defined(HAS_TANMATSU) || defined(HAS_TDISPLAY_P4) // ---- MQTT DISABLED on the ESP32-P4 boards (Tanmatsu + T-Display P4) ---- // The P4's Wi-Fi runs through the esp-hosted C6 co-processor; the Arduino WiFiClient / PubSubClient @@ -16,6 +42,7 @@ public: void publishDM(const char*, const uint8_t*, float, uint8_t, uint32_t, const char*) {} void publishChannel(int, const char*, float, uint8_t, uint32_t, const char*) {} bool enabled() const { return false; } + void getStatus(MqttBridgeStatus& out) const { out = MqttBridgeStatus{}; } static void saveConfig(const char*, uint16_t, const char*, const char*, bool, bool, const char*, bool) {} void reloadConfig() {} @@ -75,6 +102,7 @@ public: float snr, uint8_t hops, uint32_t ts, const char* text); bool enabled() const { return _enabled; } + void getStatus(MqttBridgeStatus& out) const; // Persist config (called from Settings UI save). static void saveConfig(const char* host, uint16_t port, @@ -87,6 +115,7 @@ private: MqttNbClient _wc; PubSubClient _mqtt{_wc}; char _nodeHex[13] = {}; // 6-byte key → 12 hex chars + '\0' + bool _requestedEnabled = false; bool _enabled = false; bool _pubDm = false; // DMs off by default (private 1:1 traffic) bool _pubChannel = true; // channel messages on by default @@ -101,11 +130,16 @@ private: // True while the one-shot connect task owns _mqtt/_wc. The loop thread must // not touch either until it clears (PubSubClient is not thread-safe). volatile bool _connecting = false; + mutable portMUX_TYPE _statusMux = portMUX_INITIALIZER_UNLOCKED; + MqttBridgeStatus _status; static const uint32_t RECONNECT_INTERVAL_MS = 15000; void loadConfig(); // shared by begin() / reloadConfig() void deriveKey(); // _key = SHA-256(_psk); sets _encOn + void setStatusPhase(MqttBridgePhase phase, int result, uint32_t now, + bool recordResult, uint32_t retryAt = 0); + void syncStatusConfig(MqttBridgePhase phase); bool reconnect(); static void reconnectTask(void* arg); // one-shot task body wrapping reconnect() void pub(const char* subtopic, const char* json); // seals if _encOn diff --git a/src/helpers/esp32/WebSocketCompanionServer.cpp b/src/helpers/esp32/WebSocketCompanionServer.cpp index 2692bc5..5160325 100644 --- a/src/helpers/esp32/WebSocketCompanionServer.cpp +++ b/src/helpers/esp32/WebSocketCompanionServer.cpp @@ -326,12 +326,12 @@ static const char WS_HTML_TERMINAL_PAGE[] = "var QCMDS=[{h:'CHAT'},{l:'list - contacts',c:'list'},{l:'channels',c:'channels'},{l:'to ',c:'to '},{l:'send ',c:'send '},{l:'public ',c:'public '},{l:'exit',c:'exit'},\n" " {h:'INFO'},{l:'help',c:'help'},{l:'ver - firmware version',c:'ver'},{l:'clock - RTC time',c:'clock'},{l:'status',c:'status'},{l:'get - radio params',c:'get'},\n" " {h:'RADIO / MESH'},{l:'advert',c:'advert'},{l:'advert.zerohop',c:'advert.zerohop'},{l:'set name ',c:'set name '},{l:'set freq ',c:'set freq '},{l:'set bw ',c:'set bw '},{l:'set sf <7-12>',c:'set sf '},{l:'set cr <5-8>',c:'set cr '},{l:'set tx ',c:'set tx '},\n" - " {h:'CONNECTIVITY'},{l:'wifi status',c:'wifi status'},{l:'wifi on',c:'wifi on'},{l:'wifi off',c:'wifi off'},{l:'wifi scan',c:'wifi scan'},{l:'wifi apply',c:'wifi apply'},{l:'wifi clear',c:'wifi clear'},{l:'tcp status',c:'tcp status'},{l:'tcp on',c:'tcp on'},{l:'tcp off',c:'tcp off'},{l:'ble status',c:'ble status'},{l:'ble on',c:'ble on'},{l:'ble off',c:'ble off'},{l:'ota status',c:'ota status'},{l:'ota start',c:'ota start'},\n" + " {h:'CONNECTIVITY'},{l:'wifi status',c:'wifi status'},{l:'wifi on',c:'wifi on'},{l:'wifi off',c:'wifi off'},{l:'wifi scan',c:'wifi scan'},{l:'wifi apply',c:'wifi apply'},{l:'wifi clear',c:'wifi clear'},{l:'mqtt status',c:'mqtt status'},{l:'tcp status',c:'tcp status'},{l:'tcp on',c:'tcp on'},{l:'tcp off',c:'tcp off'},{l:'ble status',c:'ble status'},{l:'ble on',c:'ble on'},{l:'ble off',c:'ble off'},{l:'ota status',c:'ota status'},{l:'ota start',c:'ota start'},\n" " {h:'SYSTEM'},{l:'reboot',c:'reboot'},{l:'bootloader - download mode',c:'bootloader'}];\n" "function cmdPicker(){var h='

Commands

';QCMDS.forEach(function(x,i){if(x.h)h+='
'+esc(x.h)+'
';else h+=''});\n" " E('sheet').innerHTML=h;each(E('sheet').querySelectorAll('.si'),function(b){b.onclick=function(){var x=QCMDS[+b.getAttribute('data-i')];closeOv();showTab('term');E('i').value=x.c;E('i').focus()}});\n" " E('sheet').classList.add('on');E('scrim').classList.add('on')}\n" - "var hist=[],hi=0,CMDS=['help','?','status','ver','clock','get','advert','advert.zerohop','reboot','bootloader','set ','tcp ','ble ','ota ','wifi ','list','contacts','channels','to ','send ','public ','exit','leave','ok','cancel','clear'];\n" + "var hist=[],hi=0,CMDS=['help','?','status','ver','clock','get','advert','advert.zerohop','reboot','bootloader','set ','mqtt ','tcp ','ble ','ota ','wifi ','list','contacts','channels','to ','send ','public ','exit','leave','ok','cancel','clear'];\n" "function tadd(t,cls){var s=document.createElement('span');if(cls)s.className=cls;s.textContent=t;E('o').appendChild(s);E('o').scrollTop=E('o').scrollHeight}\n" "function tsend(){var v=E('i').value;E('i').value='';if(!v.trim())return;if(v.trim()=='clear'){E('o').innerHTML='';return}\n" " tadd('\\n> '+v+'\\n','c');hist.push(v);hi=hist.length;if(!wsSend(v))tadd('(not connected)\\n','err')}\n" diff --git a/src/ui-touch/ConsoleUI.cpp b/src/ui-touch/ConsoleUI.cpp index e4a078f..1ced2d3 100644 --- a/src/ui-touch/ConsoleUI.cpp +++ b/src/ui-touch/ConsoleUI.cpp @@ -548,11 +548,12 @@ void submit() { consoleWriteLine("mesh contacts, chans, unread"); consoleWriteLine(" chat show a thread"); consoleWriteLine(" discover / discovered"); - consoleWriteLine("radio stat, batt, wifi"); consoleWriteLine(" read clear that thread's unread"); consoleWriteLine(" monitor on|off show arriving messages"); consoleWriteLine(" to pick a contact or channel"); consoleWriteLine(" msg send to it"); + consoleWriteLine("network mqtt status, tcp status, ble status"); + consoleWriteLine("radio stat, batt, wifi"); consoleWriteLine("node anything the CLI answers:"); consoleWriteLine(" advert, get name, set name , time, ver"); return; diff --git a/src/ui-touch/UITask.cpp b/src/ui-touch/UITask.cpp index b1f40fa..8eee99a 100644 --- a/src/ui-touch/UITask.cpp +++ b/src/ui-touch/UITask.cpp @@ -19181,7 +19181,7 @@ static const char* k_term_banner = "Config: ver clock status get advert reboot\n" " set \n" "MeshCom: wifi status, tcp status, ble status,\n" - " ota status, tcp on/off, ble on/off.\n"; + " mqtt status, ota status, tcp/ble on/off.\n"; // Quick-pick catalogue (mirrors the repeater admin picker). Commands here are // the ones MyMesh::handleMeshcomodCommand understands natively when run @@ -19219,6 +19219,7 @@ static const AdminCmdEntry k_term_cmds[] = { { "wifi set pwd ", "wifi set pwd " }, { "wifi apply - connect now", "wifi apply" }, { "wifi clear - forget creds", "wifi clear" }, + { "mqtt status", "mqtt status" }, { "tcp status", "tcp status" }, { "tcp on", "tcp on" }, { "tcp off", "tcp off" }, @@ -19542,7 +19543,7 @@ static void termCmdExit() { static bool termIsCliKeyword(const char* word, size_t len) { static const char* kw[] = { "help", "ver", "version", "clock", "time", "status", "get", "set", - "advert", "advert.zerohop", "reboot", "wifi", "tcp", "ble", "ota", + "advert", "advert.zerohop", "reboot", "wifi", "mqtt", "tcp", "ble", "ota", "ok", "cancel", }; for (const char* k : kw) {