Merge PR stack #20-#22: prefs migration, duty cycle + CAD, FEM RX gain

Merges three stacked PRs in one push so the rolling release only builds
the final state:

#20 — Complete observer settings migration and upstream-seam cleanup
- Implement the missing one-time /com_prefs -> /mqtt_prefs observer
  settings migration (silent data loss on upgrade before this).
- Harden /mqtt_prefs into a versioned format (magic + version +
  payload_len header); legacy headerless layouts detected by size and
  migrated once. Drop vestigial _legacy_* fields.
- Deployed flex layout pinned by static_assert (Legacy6SlotMQTTPrefs ==
  2904 bytes); verified with the host migration harness and on-device
  (Heltec V4: presets, WiFi, and origin survive upgrade and reboot).

#21 — Restore upstream duty-cycle enforcement and CAD
- Restore Dispatcher/StaticPoolPacketManager verbatim from upstream/dev
  (token-bucket duty cycle, CAD interface); re-apply the MQTT radio
  watchdog as pure additions behind WITH_MQTT_BRIDGE.
- Restore cad_enabled and radio_fem_rxgain persistence at upstream's
  /com_prefs offsets (byte-parity with upstream, tail 3 -> 5 bytes).
- Add RxReservePacketManager for observer builds: priority-aware shed +
  30s stale expiry so heavy throttling can't park the packet pool and
  make the node un-administrable.
- Device-confirmed on Heltec V4.2: throttling holds at set dutycycle 1
  on a busy mesh, CAD persists across boots, capture continues at full
  rate under throttle.

#22 — Drive LoRa FEM RX gain from radio_fem_rxgain
- MainBoard gains setLoRaFemLnaEnabled/canControlLoRaFemLna/
  isLoRaFemLnaEnabled; wired for heltec_v4, heltec_t096, and
  tracker_v2. radio.fem.rxgain CLI gated on board capability.
- Persisted pref applied at app startup; defaults ON (upstream parity),
  so LNA-capable boards gain RX boost on upgrade.
- Guard path device-confirmed on V4.2 (GC1109, PA-only: reports
  unsupported). LNA toggle on V4.3/KCT8103L pending hardware.

Multi-hour soak on the full stack: no memory issues or regressions.
This commit is contained in:
Adam Gessaman
2026-07-09 21:56:36 -07:00
committed by GitHub
37 changed files with 2235 additions and 1501 deletions
+4 -16
View File
@@ -134,22 +134,6 @@ The MQTT bridge uses a slot-based architecture with up to 6 concurrent connectio
- If more slots are configured than the device supports, excess slots show as `(inactive)` in `get mqtt.status`
- Slot configurations are preserved in preferences — moving firmware to a PSRAM device activates all slots
### Files
#### Core Implementation
- `src/helpers/bridges/MQTTBridge.h` - MQTT bridge class definition
- `src/helpers/bridges/MQTTBridge.cpp` - MQTT bridge implementation
- `src/helpers/MQTTPresets.h` - Preset definitions, CA certificates, and lookup functions
- `src/helpers/MQTTMessageBuilder.h` - JSON message formatting utilities
- `src/helpers/MQTTMessageBuilder.cpp` - JSON message formatting implementation
- `src/helpers/JWTHelper.h` - JWT token generation for Ed25519-based authentication
#### Integration
- Updated `examples/simple_repeater/MyMesh.h` - Added MQTT bridge support
- Updated `examples/simple_repeater/MyMesh.cpp` - Added MQTT bridge integration and raw radio data capture
- Updated `src/helpers/CommonCLI.h` - MQTT slot preferences, WiFi, and timezone fields
- Updated `src/helpers/CommonCLI.cpp` - MQTT slot CLI commands, migration logic
## Build Configuration
To build the MQTT bridge firmware:
@@ -790,3 +774,7 @@ set radio.watchdog 0 # disable watchdog
```
On very quiet meshes where no traffic is expected for long periods, increase the interval or set `0` to disable the watchdog and avoid unnecessary radio recoveries.
## Developer Documentation
For source layout, the seams that isolate the observer feature from upstream MeshCore code, and on-device settings migration across firmware versions, see [MQTT_INTERNALS.md](MQTT_INTERNALS.md).
+96
View File
@@ -0,0 +1,96 @@
# MQTT Bridge Internals
Developer-facing notes on how the MQTT observer feature is structured in the codebase: source files, the seams that keep it isolated from upstream MeshCore code, and how on-device settings are migrated across firmware versions. For user-facing setup and CLI reference, see [MQTT_IMPLEMENTATION.md](MQTT_IMPLEMENTATION.md).
## Files
### Core Implementation
- `src/helpers/bridges/MQTTBridge.h` - MQTT bridge class definition
- `src/helpers/bridges/MQTTBridge.cpp` - MQTT bridge implementation
- `src/helpers/MQTTPresets.h` - Preset definitions, CA certificates, and lookup functions
- `src/helpers/MQTTDefaults.h` - Compile-time defaults for fresh `/mqtt_prefs`
- `src/helpers/MQTTMessageBuilder.h` - JSON message formatting utilities
- `src/helpers/MQTTMessageBuilder.cpp` - JSON message formatting implementation
- `src/helpers/JWTHelper.h` - JWT token generation for Ed25519-based authentication
- `src/helpers/CommonCLI_Observer.cpp` - All observer CLI command handling (MQTT, WiFi,
timezone, NTP, OTA, SNMP, alerts)
### Integration seams with upstream code
The observer feature is kept out of upstream-tracked files through three mechanisms:
- **CLI hook methods** — upstream `CommonCLI.cpp` delegates to three `CommonCLI`
methods defined in the fork-owned `CommonCLI_Observer.cpp`: `handleObserverCommand()`,
`handleObserverSetCmd()`, and `handleObserverGetCmd()`. Each returns `true` if it
consumed the command, otherwise the upstream parser runs. Only these three call
sites touch upstream CLI code.
- **Callback virtuals** — observer behaviour needed from the application is exposed
as default-no-op virtuals on `CommonCLICallbacks` (e.g. `restartBridgeSlot`,
`isMqttBridgeRunning`, `syncMqttNtp`, `onAlertConfigChanged`, `sendAlertText`,
`resolveAlertScope`, `beginDeferredOtaUpdate`). The example apps override them
behind `#ifdef WITH_MQTT_BRIDGE`.
- **Separate settings file** — observer settings (MQTT slots, WiFi, timezone, SNMP,
radio watchdog, fault alerts) live in the `MQTTPrefs` struct persisted to
`/mqtt_prefs`, keeping `NodePrefs` / `/com_prefs` aligned with the upstream layout.
Remaining integration points in upstream files:
- `examples/simple_repeater/MyMesh.{h,cpp}`, `examples/simple_room_server/MyMesh.{h,cpp}` -
bridge/alerter/SNMP wiring and packet-feed hooks, guarded by `#ifdef WITH_MQTT_BRIDGE`;
plus the `createObserverPacketManager()` call in each constructor (see below)
- `src/helpers/CommonCLI.{h,cpp}` - the three CLI hooks, `MQTTPrefs` load/save/migration
- `src/Dispatcher.{h,cpp}` - radio watchdog block, guarded by `#ifdef WITH_MQTT_BRIDGE`
### Capture vs. duty-cycle throttling
RX processing needs a free packet from the static pool before `logRx()` (and thus the
MQTT uplink) can run — `Dispatcher::checkRecv()` silently discards received data when
the pool is empty. Because the outbound queue holds pool packets with no expiry,
duty-cycle throttling can park the entire pool waiting on TX budget, capping capture at
the TX rate — and the parked repeats absorb every budget refill, starving the node's
own CLI responses and making it un-administrable over the mesh. Observer builds
therefore use `RxReservePacketManager` (fork-owned,
`src/helpers/RxReservePacketManager.h`): below the RX reserve (a quarter of the pool)
it sheds only low-priority outbound (multi-hop flood repeats, adverts, trace), keeping
the node's own responses/ACKs queueable; below a smaller emergency floor it sheds
everything to keep capture alive. Queued packets still untransmitted 30 s past their
scheduled time are expired at dequeue, so under throttle the queue holds only fresh
traffic and admin responses reach the trickle of TX budget. Non-observer builds keep
the upstream pool behavior.
### `/mqtt_prefs` file format
`/mqtt_prefs` is written with an 8-byte `MQTTPrefsHeader` (`magic`, `version`,
`payload_len`) followed by the raw `MQTTPrefs` payload. The magic is
`{0xF5, 'M', 'Q', 'P'}` — its leading non-ASCII byte can never collide with the first
bytes of a legacy (headerless) file, whose payload begins with the `mqtt_origin`
string. Bump `MQTT_PREFS_VERSION` when the payload layout changes incompatibly; a file
whose version this firmware doesn't recognize is left untouched and the in-memory prefs
fall back to defaults (no downgrade, no misread). `saveMQTTPrefs()` also refuses to
write while such a file is present (`_mqtt_prefs_hold`), so a `set` command after a
firmware downgrade can't clobber the newer config — observer settings changed in that
state simply don't persist. The frozen legacy layouts are pinned with `static_assert`s
in `CommonCLI.h`, so every target build re-verifies the fleet's file offsets.
Adding a field to the current version stays backward compatible: append it to the end
of `MQTTPrefs`. An older, shorter payload still loads and the missing tail keeps its
default; a newer, longer one is truncated harmlessly.
### Settings upgrade / migration
`loadPrefs()` handles every historical on-device format one-time at boot:
- **`/mqtt_prefs`** — if the file has the version header it is read directly. Otherwise
it is a legacy headerless file and its layout is detected by size: pre-slot
(`OldMQTTPrefs`), 3-slot (`ThreeSlotMQTTPrefs`), or the 6-slot layout shipped on
`mqtt-bridge-implementation-flex` (`Legacy6SlotMQTTPrefs`). Each is field-copied into
the current compact `MQTTPrefs` and re-saved with the version header — which also
drops the vestigial `_legacy_*` fields the flex layout carried mid-struct. This is a
one-time rewrite; every deployed device performs it on its first boot of versioned
firmware, after which all reads take the header path.
- **`/com_prefs`** — a file written by fork firmware that predates the `MQTTPrefs` split
(a zero-filled MQTT gap plus a trailing observer block) is detected by size; the
trailing SNMP / radio-watchdog / fault-alert settings and the `rx_boosted_gain` /
`flood_max_*` fields are recovered, carried into `/mqtt_prefs`, and both files are
rewritten in the current formats.
- Settings the pre-split firmware stored *inside* the `/com_prefs` MQTT gap (the MQTT
slot/WiFi config itself) are **not** recovered — users upgrading from firmware that
old must re-enter their MQTT and WiFi configuration.
+94
View File
@@ -0,0 +1,94 @@
# Restoring accidentally-reverted upstream features
## Background
On 2026-03-20, commit `22eb9b87`*Revert "Merge remote-tracking branch 'origin/dev' into mqtt-bridge-implementation"* — reverted an entire upstream merge to escape a bad merge state, deleting 860 lines across 66 files. That was not intentional feature removal; it wholesale dropped a batch of upstream progress. When upstream was later re-merged, some collateral came back (MicroNMEA `claim()/release()`, the GAT562 board) but several upstream features were never reconciled and remained missing.
This branch restores them. They are **pure upstream code the fork accidentally dropped**, so restoring re-aligns the fork with upstream and *reduces* the future merge-conflict surface rather than adding to it.
## Phase 1 — duty-cycle enforcement (done on this branch)
Restored the token-bucket duty-cycle enforcement and its cluster:
- `src/Dispatcher.{h,cpp}` — restored upstream's `updateTxBudget()`, `tx_budget_ms`,
`duty_cycle_window_ms`, `getRemainingTxBudget()`, `getDutyCycleWindowMs()`, and the
windowed `checkSend()`/`loop()` budget logic. The fork's MQTT **radio watchdog** was
re-applied on top as pure additions (behind `#ifdef WITH_MQTT_BRIDGE`).
- `src/helpers/StaticPoolPacketManager.{h,cpp}` — restored `getOutboundTotal()` and the
`0xFFFFFFFF` count-all sentinel in `countBefore()`.
- `src/helpers/StatsFormatHelper.h` — restored the `getOutboundTotal()` call; kept the
fork's `formatRadioDiag` template.
Net effect: these files now diverge from upstream by **watchdog additions only (0 deletions)**
instead of rewriting upstream's TX logic.
### Why this is safe (no reinterpretation of stored settings)
`getAirtimeBudgetFactor()` changed *meaning* between fork and upstream, but the resulting
duty cycle is the **same formula**:
- Fork: `next_tx = t · factor` → duty ≈ `1/(1+factor)`
- Upstream: `duty = 1/(1+factor)` enforced over a rolling window
So a device's stored `airtime_factor` (a `NodePrefs` field, unchanged) keeps its meaning.
The only behavioral difference is upstream enforces it over a 1-hour window (allowing
short bursts) instead of rigid per-packet spacing — a strict improvement, and the
mechanism that keeps EU 868 MHz nodes under the legally-mandated duty cycle.
### Must be validated on-device before merge
This touches core TX timing. Confirm on hardware — **all device-confirmed 2026-07-09/10
on a Heltec V4.2 (busy live mesh + off-frequency bench):**
- ✅ Normal traffic still flows (repeater forwards, observer uplinks).
- ✅ Sustained TX is throttled to the configured duty cycle: at `set dutycycle 1` on a
busy mesh, TX pinned to ~1% while MQTT capture continued at full rate (after the
`RxReservePacketManager` fixes below). Remote admin remains usable under throttle
with the priority-shed + stale-expiry policy.
- ✅ CAD enabled under load: TX, capture, and CLI all normal.
- ✅ The observer radio watchdog fires and recovers non-destructively: with
`radio.watchdog 1` on a silent frequency, `err_flags` bit 8 set, radio remained in
RX state through repeated recovery cycles, and packets were still received after.
Note: firings are invisible on release builds (no MESH_DEBUG) — check
`stats-radio-diag` `err_flags`; the watchdog arms only after first radio activity
(`last_active > 0`), so a radio wedged from boot is deliberately not covered.
Remaining before undraft: passive soak at fleet-default `airtime_factor` (duty
convergence, flat heap, adverts still advertised).
### Known interaction: throttling starves MQTT capture (mitigated)
Found during the on-device duty-cycle load test (`set dutycycle 1`): queued
retransmissions hold static-pool packets with no expiry, so heavy throttling parks the
whole pool in the send queue. `Dispatcher::checkRecv()` then drops received packets
before `logRx()` runs, capping MQTT capture at the TX rate (each TX frees one packet
for one RX). Parked repeats also absorb every budget refill, starving the node's own
CLI responses — device-confirmed: a 1%-duty node on a busy mesh became
un-administrable within ~2 minutes. This is inherent upstream behavior — the old
fork's `next_tx` spacing had the same steady-state drain — but it defeats the
observer's purpose. Mitigated on observer builds by `RxReservePacketManager`
(`src/helpers/RxReservePacketManager.h`): priority-aware shedding below a pool
reserve (own responses/ACKs stay queueable) plus 30 s expiry of stale queued
outbound. See MQTT_INTERNALS.md "Capture vs. duty-cycle throttling".
## Phase 2 — CAD and FEM RX gain (NOT done; needs care + device testing)
Still missing at HEAD, also dropped by `22eb9b87`, still present upstream:
- **`cad_enabled`** — hardware Channel Activity Detection (listen-before-talk) before TX.
The `Dispatcher`/`Radio` interface (`setCADEnabled`/`getCADEnabled`) is restored by
Phase 1, so CAD currently stays **off by default** (unchanged behavior). To make it
configurable again, restore:
- `NodePrefs.cad_enabled` field, its CLI get/set, and its persistence in
`CommonCLI.cpp` (`loadPrefsInt`/`savePrefs`). **Offset care:** this branch's
`/com_prefs` layout is carefully managed — add `cad_enabled` following the same
append-and-size-guard pattern used for `rx_boosted_gain`/`flood_max_*`, and add a
host-side round-trip test (see `scratchpad/migtest`).
- The `MyMesh::getCADEnabled()` override returning `_prefs.cad_enabled`.
- `RadioLibWrappers::setCADEnabled()` override so the hardware CAD is actually driven
(the fork's wrapper currently doesn't override it).
- **`radio_fem_rxgain`** — LoRa front-end-module RX gain. Restore `NodePrefs.radio_fem_rxgain`
+ CLI + persistence (same offset care), plus the per-board FEM wiring reverted across
~20 `variants/*/target.cpp` and the `heltec_tracker_v2/LoRaFEMControl.{cpp,h}` files.
This is board-specific and only affects FEM-equipped hardware.
Phase 2 is lower urgency than duty-cycle enforcement (CAD/FEM are capability gaps, not a
compliance regression) and is best done as its own change with per-board hardware testing.
+6
View File
@@ -258,6 +258,10 @@ float MyMesh::getAirtimeBudgetFactor() const {
return _prefs.airtime_factor;
}
bool MyMesh::getCADEnabled() const {
return true; // hardware CAD before TX (no CLI toggle on companion; enabled by default)
}
int MyMesh::getInterferenceThreshold() const {
return 0; // disabled for now, until currentRSSI() problem is resolved
}
@@ -966,6 +970,8 @@ void MyMesh::begin(bool has_display) {
radio_driver.setRxBoostedGainMode(_prefs.rx_boosted_gain);
MESH_DEBUG_PRINTLN("RX Boosted Gain Mode: %s",
radio_driver.getRxBoostedGainMode() ? "Enabled" : "Disabled");
// NOTE: no FEM LNA wiring here — companion has its own NodePrefs without
// radio_fem_rxgain, matching upstream (which also doesn't wire companion).
}
const char *MyMesh::getNodeName() {
+1
View File
@@ -105,6 +105,7 @@ public:
protected:
float getAirtimeBudgetFactor() const override;
int getInterferenceThreshold() const override;
bool getCADEnabled() const override;
int calcRxDelay(float score, uint32_t air_time) const override;
uint32_t getRetransmitDelay(const mesh::Packet *packet) override;
uint32_t getDirectRetransmitDelay(const mesh::Packet *packet) override;
+19 -32
View File
@@ -1,6 +1,7 @@
#include "MyMesh.h"
#include <algorithm>
#include <stdlib.h> // for qsort()
#include <helpers/RxReservePacketManager.h>
/* ------------------------------ Config -------------------------------- */
@@ -901,7 +902,7 @@ void MyMesh::sendNodeDiscoverReq() {
MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondClock &ms, mesh::RNG &rng,
mesh::RTCClock &rtc, mesh::MeshTables &tables)
: mesh::Mesh(radio, ms, rng, rtc, *new StaticPoolPacketManager(32), tables),
: mesh::Mesh(radio, ms, rng, rtc, *createObserverPacketManager(32), tables),
region_map(key_store), temp_map(key_store),
_cli(board, rtc, sensors, region_map, acl, &_prefs, this),
telemetry(MAX_PACKET_PAYLOAD - 4),
@@ -951,21 +952,8 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
#ifdef WITH_MQTT_BRIDGE
_prefs.agc_reset_interval = 7; // 28 seconds (secs/4) — prevents AGC drift on long-running observers
#endif
_prefs.radio_watchdog_minutes = 5; // 5 minutes default
// Alert channel defaults — disabled by default, and the channel is left
// unconfigured so a freshly-flashed observer never broadcasts on the
// well-known Public hashtag. Operators must explicitly pick a private
// key (`set alert.psk`) or a hashtag (`set alert.hashtag`) before alerts
// can fire. The sender prefix on outgoing alert messages is always the
// node name (`set name ...`), so there's no separate `alert.name`.
_prefs.alert_enabled = 0;
_prefs.alert_psk_hex[0] = '\0';
_prefs.alert_hashtag[0] = '\0';
_prefs.alert_region[0] = '\0'; // empty = use default_scope
_prefs.alert_wifi_minutes = 30; // 30 minutes
_prefs.alert_mqtt_minutes = 240; // 4 hours
_prefs.alert_min_interval_min = 60; // re-arm window: 1 hour
// Observer defaults (radio_watchdog, alert.*, snmp.*) moved to applyMQTTDefaults()
// in MQTTDefaults.h — they live in /mqtt_prefs now, not NodePrefs.
// bridge defaults
_prefs.bridge_enabled = 1; // enabled
@@ -976,21 +964,12 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
StrHelper::strncpy(_prefs.bridge_secret, "LVSITANOS", sizeof(_prefs.bridge_secret));
// SNMP defaults
_prefs.snmp_enabled = 0;
StrHelper::strncpy(_prefs.snmp_community, "public", sizeof(_prefs.snmp_community));
// GPS defaults
_prefs.gps_enabled = 0;
_prefs.gps_interval = 0;
_prefs.advert_loc_policy = ADVERT_LOC_PREFS;
// MQTT slot/IATA/timezone defaults come from /mqtt_prefs via loadPrefs (see MQTTDefaults.h)
_prefs.mqtt_origin[0] = '\0';
// WiFi defaults (user-configured via CLI; placeholders until set)
StrHelper::strncpy(_prefs.wifi_ssid, "ssid_here", sizeof(_prefs.wifi_ssid));
StrHelper::strncpy(_prefs.wifi_password, "password_here", sizeof(_prefs.wifi_password));
// MQTT/WiFi/timezone/radio_watchdog defaults live in /mqtt_prefs now (see applyMQTTDefaults).
_prefs.adc_multiplier = 0.0f; // 0.0f means use default board multiplier
@@ -1001,6 +980,8 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
_prefs.rx_boosted_gain = 1; // enabled by default;
#endif
#endif
_prefs.radio_fem_rxgain = 1; // LoRa FEM RX gain on by default (FEM boards)
_prefs.cad_enabled = 0; // hardware CAD before TX (off by default; 'set cad on')
pending_discover_tag = 0;
pending_discover_until = 0;
@@ -1042,7 +1023,7 @@ void MyMesh::begin(FILESYSTEM *fs) {
if (_prefs.bridge_enabled) {
#ifdef WITH_MQTT_BRIDGE
// Defer construction to avoid static init crashes on ESP32 classic
bridge = new MQTTBridge(&_prefs, _mgr, getRTCClock(), &self_id);
bridge = new MQTTBridge(&_prefs, _cli.getObserverPrefs(), _mgr, getRTCClock(), &self_id);
#endif
if (bridge) {
// Set device public key for MQTT topics
@@ -1065,7 +1046,7 @@ void MyMesh::begin(FILESYSTEM *fs) {
// Set stats sources for automatic stats collection
bridge->setStatsSources(this, _radio, _cli.getBoard(), _ms);
#ifdef WITH_SNMP
if (_prefs.snmp_enabled) {
if (_cli.getObserverPrefs()->snmp_enabled) {
_snmp_agent.setNodeName(_prefs.node_name);
_snmp_agent.setFirmwareVersion(getFirmwareVer());
bridge->setSNMPAgent(&_snmp_agent);
@@ -1082,8 +1063,8 @@ void MyMesh::begin(FILESYSTEM *fs) {
// Passing `this` as the callbacks lets the reporter resolve a TransportKey
// scope (alert.region override, falling back to default_scope) so alert
// floods ride the same scope as adverts/channel messages.
_alerter.begin(&_prefs, this, this);
#if defined(WITH_MQTT_BRIDGE)
#ifdef WITH_MQTT_BRIDGE
_alerter.begin(&_prefs, _cli.getObserverPrefs(), this, this);
_alerter.setBridge(bridge);
#endif
@@ -1093,6 +1074,7 @@ void MyMesh::begin(FILESYSTEM *fs) {
radio_driver.setRxBoostedGainMode(_prefs.rx_boosted_gain);
MESH_DEBUG_PRINTLN("RX Boosted Gain Mode: %s",
radio_driver.getRxBoostedGainMode() ? "Enabled" : "Disabled");
board.setLoRaFemLnaEnabled(_prefs.radio_fem_rxgain); // LoRa FEM LNA (FEM boards only)
updateAdvertTimer();
updateFloodAdvertTimer();
@@ -1118,12 +1100,15 @@ bool MyMesh::resolveAlertScope(TransportKey& dest) {
// RegionMap so the operator can name a region that doesn't exist yet
// without polluting region_map state — we just silently fall through
// to default_scope on miss.
if (_prefs.alert_region[0]) {
auto r = region_map.findByNamePrefix(_prefs.alert_region);
#ifdef WITH_MQTT_BRIDGE
const char* alert_region = _cli.getObserverPrefs()->alert_region;
if (alert_region[0]) {
auto r = region_map.findByNamePrefix(alert_region);
if (r && region_map.getTransportKeysFor(*r, &dest, 1) > 0 && !dest.isNull()) {
return true;
}
}
#endif
if (!default_scope.isNull()) {
dest = default_scope;
return true;
@@ -1473,7 +1458,9 @@ void MyMesh::loop() {
uptime_millis += now - last_millis;
last_millis = now;
#ifdef WITH_MQTT_BRIDGE
_alerter.onLoop(now);
#endif
#ifdef WITH_SNMP
// Push radio stats to SNMP agent every 2 seconds
+17 -3
View File
@@ -132,7 +132,9 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
#ifdef WITH_SNMP
MeshSNMPAgent _snmp_agent;
#endif
#ifdef WITH_MQTT_BRIDGE
AlertReporter _alerter;
#endif
void putNeighbour(const mesh::Identity& id, uint32_t timestamp, float snr);
uint8_t handleLoginReq(const mesh::Identity& sender, const uint8_t* secret, uint32_t sender_timestamp, const uint8_t* data, bool is_flood);
@@ -150,6 +152,10 @@ protected:
return _prefs.airtime_factor;
}
bool getCADEnabled() const override {
return _prefs.cad_enabled;
}
bool allowPacketForward(const mesh::Packet* packet) override;
const char* getLogDateTime() override;
void logRxRaw(float snr, float rssi, const uint8_t raw[], int len) override;
@@ -168,9 +174,11 @@ protected:
int getAGCResetInterval() const override {
return ((int)_prefs.agc_reset_interval) * 4000; // milliseconds
}
#ifdef WITH_MQTT_BRIDGE
uint32_t getRadioWatchdogMillis() const override {
return ((uint32_t)_prefs.radio_watchdog_minutes) * 60000UL;
return ((uint32_t)_cli.getObserverPrefs()->radio_watchdog_minutes) * 60000UL;
}
#endif
uint8_t getExtraAckTransmitCount() const override {
return _prefs.multi_acks;
}
@@ -215,8 +223,10 @@ public:
// CommonCLICallbacks
void applyTempRadioParams(float freq, float bw, uint8_t sf, uint8_t cr, int timeout_mins) override;
#ifdef WITH_MQTT_BRIDGE
void onAlertConfigChanged() override { _alerter.onConfigChanged(); }
bool sendAlertText(const char* text) override { return _alerter.sendText(text); }
#endif
bool resolveAlertScope(TransportKey& dest) override;
bool formatFileSystem() override;
void sendSelfAdvertisement(int delay_millis, bool flood) override;
@@ -253,7 +263,7 @@ public:
void setBridgeState(bool enable) override {
if (!bridge) {
#ifdef WITH_MQTT_BRIDGE
bridge = new MQTTBridge(&_prefs, _mgr, getRTCClock(), &self_id);
bridge = new MQTTBridge(&_prefs, _cli.getObserverPrefs(), _mgr, getRTCClock(), &self_id);
#endif
if (!bridge) return;
}
@@ -303,8 +313,12 @@ public:
}
void restartBridgeSlot(int slot) override {
#ifdef WITH_MQTT_BRIDGE
if (!bridge || !bridge->isRunning()) return;
bridge->setSlotPreset(slot, _prefs.mqtt_slot_preset[slot]);
bridge->setSlotPreset(slot, _cli.getObserverPrefs()->mqtt_slot_preset[slot]);
#else
(void)slot;
#endif
}
// Schedule the pull-OTA flash to run from loop() in ~2.5 s, leaving time for the
+18 -21
View File
@@ -1,4 +1,5 @@
#include "MyMesh.h"
#include <helpers/RxReservePacketManager.h>
#define REPLY_DELAY_MILLIS 1500
#define PUSH_NOTIFY_DELAY_MILLIS 2000
@@ -634,7 +635,7 @@ void MyMesh::onAckRecv(mesh::Packet *packet, uint32_t ack_crc) {
MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondClock &ms, mesh::RNG &rng,
mesh::RTCClock &rtc, mesh::MeshTables &tables)
: mesh::Mesh(radio, ms, rng, rtc, *new StaticPoolPacketManager(32), tables),
: mesh::Mesh(radio, ms, rng, rtc, *createObserverPacketManager(32), tables),
region_map(key_store), temp_map(key_store),
_cli(board, rtc, sensors, region_map, acl, &_prefs, this),
telemetry(MAX_PACKET_PAYLOAD - 4)
@@ -672,6 +673,8 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
_prefs.flood_max_unscoped = 64;
_prefs.flood_max_advert = 8;
_prefs.interference_threshold = 0; // disabled
_prefs.radio_fem_rxgain = 1; // LoRa FEM RX gain on by default (FEM boards)
_prefs.cad_enabled = 0; // hardware CAD before TX (off by default; 'set cad on')
#ifdef ROOM_PASSWORD
StrHelper::strncpy(_prefs.guest_password, ROOM_PASSWORD, sizeof(_prefs.guest_password));
#endif
@@ -681,15 +684,8 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
_prefs.gps_interval = 0;
_prefs.advert_loc_policy = ADVERT_LOC_PREFS;
// Alert channel defaults (same as repeater; off by default and unconfigured).
// Operator must pick `set alert.psk` or `set alert.hashtag` before alerts fire.
_prefs.alert_enabled = 0;
_prefs.alert_psk_hex[0] = '\0';
_prefs.alert_hashtag[0] = '\0';
_prefs.alert_region[0] = '\0';
_prefs.alert_wifi_minutes = 30;
_prefs.alert_mqtt_minutes = 240;
_prefs.alert_min_interval_min = 60;
// Observer defaults (alert.*, etc.) moved to applyMQTTDefaults() — they live
// in /mqtt_prefs now, not NodePrefs.
// bridge defaults (same as repeater)
_prefs.bridge_enabled = 1; // enabled
@@ -698,12 +694,7 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
_prefs.bridge_baud = 115200; // baud rate
_prefs.bridge_channel = 1; // channel 1
// MQTT slot/IATA/timezone defaults come from /mqtt_prefs via loadPrefs (see MQTTDefaults.h)
_prefs.mqtt_origin[0] = '\0';
// WiFi defaults (user-configured via CLI; placeholders until set)
StrHelper::strncpy(_prefs.wifi_ssid, "ssid_here", sizeof(_prefs.wifi_ssid));
StrHelper::strncpy(_prefs.wifi_password, "password_here", sizeof(_prefs.wifi_password));
// MQTT/WiFi/timezone defaults live in /mqtt_prefs now (see applyMQTTDefaults).
next_post_idx = 0;
next_client_idx = 0;
@@ -745,6 +736,7 @@ void MyMesh::begin(FILESYSTEM *fs) {
radio_driver.setParams(_prefs.freq, _prefs.bw, _prefs.sf, _prefs.cr);
radio_driver.setTxPower(_prefs.tx_power_dbm);
board.setLoRaFemLnaEnabled(_prefs.radio_fem_rxgain); // LoRa FEM LNA (FEM boards only)
updateAdvertTimer();
updateFloodAdvertTimer();
@@ -757,7 +749,7 @@ void MyMesh::begin(FILESYSTEM *fs) {
#ifdef WITH_MQTT_BRIDGE
if (_prefs.bridge_enabled) {
// Defer construction to avoid static init crashes on ESP32 classic
bridge = new MQTTBridge(&_prefs, _mgr, getRTCClock(), &self_id);
bridge = new MQTTBridge(&_prefs, _cli.getObserverPrefs(), _mgr, getRTCClock(), &self_id);
if (bridge) {
// Set device public key for MQTT topics
char device_id[65];
@@ -787,8 +779,8 @@ void MyMesh::begin(FILESYSTEM *fs) {
// Passing `this` as the callbacks lets the reporter resolve a TransportKey
// scope (alert.region override, falling back to default_scope) so alert
// floods ride the same scope as adverts/channel messages.
_alerter.begin(&_prefs, this, this);
#if defined(WITH_MQTT_BRIDGE)
#ifdef WITH_MQTT_BRIDGE
_alerter.begin(&_prefs, _cli.getObserverPrefs(), this, this);
_alerter.setBridge(bridge);
#endif
}
@@ -806,12 +798,15 @@ void MyMesh::sendFloodScoped(const TransportKey& scope, mesh::Packet* pkt, uint3
bool MyMesh::resolveAlertScope(TransportKey& dest) {
// Same resolution policy as simple_repeater: alert.region > default_scope.
if (_prefs.alert_region[0]) {
auto r = region_map.findByNamePrefix(_prefs.alert_region);
#ifdef WITH_MQTT_BRIDGE
const char* alert_region = _cli.getObserverPrefs()->alert_region;
if (alert_region[0]) {
auto r = region_map.findByNamePrefix(alert_region);
if (r && region_map.getTransportKeysFor(*r, &dest, 1) > 0 && !dest.isNull()) {
return true;
}
}
#endif
if (!default_scope.isNull()) {
dest = default_scope;
return true;
@@ -1131,5 +1126,7 @@ void MyMesh::loop() {
uptime_millis += now - last_millis;
last_millis = now;
#ifdef WITH_MQTT_BRIDGE
_alerter.onLoop(now);
#endif
}
+14 -2
View File
@@ -126,7 +126,9 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
#ifdef WITH_MQTT_BRIDGE
MQTTBridge* bridge;
#endif
#ifdef WITH_MQTT_BRIDGE
AlertReporter _alerter;
#endif
void addPost(ClientInfo* client, const char* postData);
void pushPostToClient(ClientInfo* client, PostInfo& post);
@@ -141,6 +143,10 @@ protected:
return _prefs.airtime_factor;
}
bool getCADEnabled() const override {
return _prefs.cad_enabled;
}
void logRxRaw(float snr, float rssi, const uint8_t raw[], int len) override;
void logRx(mesh::Packet* pkt, int len, float score) override;
void logTx(mesh::Packet* pkt, int len) override;
@@ -201,8 +207,10 @@ public:
// CommonCLICallbacks
void applyTempRadioParams(float freq, float bw, uint8_t sf, uint8_t cr, int timeout_mins) override;
#ifdef WITH_MQTT_BRIDGE
void onAlertConfigChanged() override { _alerter.onConfigChanged(); }
bool sendAlertText(const char* text) override { return _alerter.sendText(text); }
#endif
bool resolveAlertScope(TransportKey& dest) override;
bool formatFileSystem() override;
void sendSelfAdvertisement(int delay_millis, bool flood) override;
@@ -241,7 +249,7 @@ public:
void setBridgeState(bool enable) override {
if (!bridge) {
#ifdef WITH_MQTT_BRIDGE
bridge = new MQTTBridge(&_prefs, _mgr, getRTCClock(), &self_id);
bridge = new MQTTBridge(&_prefs, _cli.getObserverPrefs(), _mgr, getRTCClock(), &self_id);
#endif
if (!bridge) return;
}
@@ -289,8 +297,12 @@ public:
}
void restartBridgeSlot(int slot) override {
#ifdef WITH_MQTT_BRIDGE
if (!bridge || !bridge->isRunning()) return;
bridge->setSlotPreset(slot, _prefs.mqtt_slot_preset[slot]);
bridge->setSlotPreset(slot, _cli.getObserverPrefs()->mqtt_slot_preset[slot]);
#else
(void)slot;
#endif
}
int getQueueSize() override {
+7
View File
@@ -301,6 +301,10 @@ float SensorMesh::getAirtimeBudgetFactor() const {
return _prefs.airtime_factor;
}
bool SensorMesh::getCADEnabled() const {
return _prefs.cad_enabled;
}
bool SensorMesh::allowPacketForward(const mesh::Packet* packet) {
if (_prefs.disable_fwd) return false;
if (packet->isRouteFlood() && packet->getPathHashCount() >= _prefs.flood_max) return false;
@@ -726,6 +730,8 @@ SensorMesh::SensorMesh(mesh::MainBoard& board, mesh::Radio& radio, mesh::Millise
_prefs.disable_fwd = true;
_prefs.flood_max = 64;
_prefs.interference_threshold = 0; // disabled
_prefs.radio_fem_rxgain = 1; // LoRa FEM RX gain on by default (FEM boards)
_prefs.cad_enabled = 0; // hardware CAD before TX (off by default; 'set cad on')
// GPS defaults
_prefs.gps_enabled = 0;
@@ -766,6 +772,7 @@ void SensorMesh::begin(FILESYSTEM* fs) {
radio_driver.setParams(_prefs.freq, _prefs.bw, _prefs.sf, _prefs.cr);
radio_driver.setTxPower(_prefs.tx_power_dbm);
board.setLoRaFemLnaEnabled(_prefs.radio_fem_rxgain); // LoRa FEM LNA (FEM boards only)
updateAdvertTimer();
updateFloodAdvertTimer();
+1
View File
@@ -120,6 +120,7 @@ protected:
uint32_t getRetransmitDelay(const mesh::Packet* packet) override;
uint32_t getDirectRetransmitDelay(const mesh::Packet* packet) override;
int getInterferenceThreshold() const override;
bool getCADEnabled() const override;
int getAGCResetInterval() const override;
void onAnonDataRecv(mesh::Packet* packet, const uint8_t* secret, const mesh::Identity& sender, uint8_t* data, size_t len) override;
int searchPeersByHash(const uint8_t* hash) override;
+63 -12
View File
@@ -8,7 +8,9 @@
namespace mesh {
#define MAX_RX_DELAY_MILLIS 32000 // 32 seconds
#define MAX_RX_DELAY_MILLIS 32000 // 32 seconds
#define MIN_TX_BUDGET_RESERVE_MS 100 // min budget (ms) required before allowing next TX
#define MIN_TX_BUDGET_AIRTIME_DIV 2 // require at least 1/N of estimated airtime as budget before TX
#ifndef NOISE_FLOOR_CALIB_INTERVAL
#define NOISE_FLOOR_CALIB_INTERVAL 2000 // 2 seconds
@@ -20,12 +22,34 @@ void Dispatcher::begin() {
_err_flags = 0;
radio_nonrx_start = _ms->getMillis();
duty_cycle_window_ms = getDutyCycleWindowMs();
float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
tx_budget_ms = (unsigned long)(duty_cycle_window_ms * duty_cycle);
last_budget_update = _ms->getMillis();
_radio->begin();
prev_isrecv_mode = _radio->isInRecvMode();
}
float Dispatcher::getAirtimeBudgetFactor() const {
return 2.0; // default, 33.3% (1/3rd)
return 1.0;
}
void Dispatcher::updateTxBudget() {
unsigned long now = _ms->getMillis();
unsigned long elapsed = now - last_budget_update;
float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
unsigned long max_budget = (unsigned long)(getDutyCycleWindowMs() * duty_cycle);
unsigned long refill = (unsigned long)(elapsed * duty_cycle);
if (refill > 0) {
tx_budget_ms += refill;
if (tx_budget_ms > max_budget) {
tx_budget_ms = max_budget;
}
last_budget_update = now;
}
}
int Dispatcher::calcRxDelay(float score, uint32_t air_time) const {
@@ -39,13 +63,16 @@ uint32_t Dispatcher::getCADFailMaxDuration() const {
return 4000; // 4 seconds
}
#ifdef WITH_MQTT_BRIDGE
uint32_t Dispatcher::getRadioWatchdogMillis() const {
return RADIO_WATCHDOG_MS;
}
#endif
void Dispatcher::loop() {
if (millisHasNowPassed(next_floor_calib_time)) {
_radio->triggerNoiseFloorCalibrate(getInterferenceThreshold());
_radio->setCADEnabled(getCADEnabled());
next_floor_calib_time = futureMillis(NOISE_FLOOR_CALIB_INTERVAL);
}
_radio->loop();
@@ -63,10 +90,9 @@ void Dispatcher::loop() {
}
// Radio watchdog: detect radio stuck in RX mode but not receiving any packets.
// Use a composite "last radio activity" timestamp: the most recent of any valid RX,
// any ISR event (even CRC errors), or any successful TX. This prevents false firings
// in quiet mesh environments where packets may be more than RADIO_WATCHDOG_MS apart,
// while still catching a truly stuck radio (PSRAM starvation → missed ISR → no activity).
// Observer-only feature (gated behind WITH_MQTT_BRIDGE); configured via the
// MQTTPrefs radio_watchdog_minutes setting.
#ifdef WITH_MQTT_BRIDGE
{
const uint32_t watchdog_ms = getRadioWatchdogMillis();
if (watchdog_ms > 0) {
@@ -87,15 +113,29 @@ void Dispatcher::loop() {
}
}
}
#endif // WITH_MQTT_BRIDGE (radio watchdog)
if (outbound) { // waiting for outbound send to be completed
if (_radio->isSendComplete()) {
long t = _ms->getMillis() - outbound_start;
total_air_time += t; // keep track of how much air time we are using
total_air_time += t;
//Serial.print(" airtime="); Serial.println(t);
// will need radio silence up to next_tx_time
next_tx_time = futureMillis(t * getAirtimeBudgetFactor());
updateTxBudget();
if (t > tx_budget_ms) {
tx_budget_ms = 0;
} else {
tx_budget_ms -= t;
}
if (tx_budget_ms < MIN_TX_BUDGET_RESERVE_MS) {
float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
unsigned long needed = MIN_TX_BUDGET_RESERVE_MS - tx_budget_ms;
next_tx_time = futureMillis((unsigned long)(needed / duty_cycle));
} else {
next_tx_time = _ms->getMillis();
}
_radio->onSendFinished();
last_radio_active_ms = _ms->getMillis(); // TX success → radio is alive
@@ -266,9 +306,20 @@ void Dispatcher::processRecvPacket(Packet* pkt) {
}
void Dispatcher::checkSend() {
if (_mgr->getOutboundCount(_ms->getMillis()) == 0) return; // nothing waiting to send
if (!millisHasNowPassed(next_tx_time)) return; // still in 'radio silence' phase (from airtime budget setting)
if (_radio->isReceiving()) { // LBT - check if radio is currently mid-receive, or if channel activity
if (_mgr->getOutboundCount(_ms->getMillis()) == 0) return;
updateTxBudget();
uint32_t est_airtime = _radio->getEstAirtimeFor(MAX_TRANS_UNIT);
if (tx_budget_ms < est_airtime / MIN_TX_BUDGET_AIRTIME_DIV) {
float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
unsigned long needed = est_airtime / MIN_TX_BUDGET_AIRTIME_DIV - tx_budget_ms;
next_tx_time = futureMillis((unsigned long)(needed / duty_cycle));
return;
}
if (!millisHasNowPassed(next_tx_time)) return;
if (_radio->isReceiving()) {
if (cad_busy_start == 0) {
cad_busy_start = _ms->getMillis(); // record when CAD busy state started
}
+18 -3
View File
@@ -68,6 +68,8 @@ public:
virtual void triggerNoiseFloorCalibrate(int threshold) { }
virtual void setCADEnabled(bool enable) { }
virtual void resetAGC() { }
virtual uint8_t getRadioState() const { return 0; }
@@ -102,6 +104,7 @@ public:
virtual void queueOutbound(Packet* packet, uint8_t priority, uint32_t scheduled_for) = 0;
virtual Packet* getNextOutbound(uint32_t now) = 0; // by priority
virtual int getOutboundCount(uint32_t now) const = 0;
virtual int getOutboundTotal() const = 0;
virtual int getFreeCount() const = 0;
virtual Packet* getOutboundByIdx(int i) = 0;
virtual Packet* removeOutboundByIdx(int i) = 0;
@@ -141,8 +144,12 @@ class Dispatcher {
bool prev_isrecv_mode;
uint32_t n_sent_flood, n_sent_direct;
uint32_t n_recv_flood, n_recv_direct;
unsigned long tx_budget_ms;
unsigned long last_budget_update;
unsigned long duty_cycle_window_ms;
void processRecvPacket(Packet* pkt);
void updateTxBudget();
protected:
PacketManager* _mgr;
@@ -155,12 +162,15 @@ protected:
{
outbound = NULL;
total_air_time = rx_air_time = 0;
next_tx_time = 0;
next_tx_time = ms.getMillis();
cad_busy_start = 0;
next_floor_calib_time = next_agc_reset_time = 0;
_err_flags = 0;
radio_nonrx_start = 0;
prev_isrecv_mode = true;
tx_budget_ms = 0;
last_budget_update = 0;
duty_cycle_window_ms = 3600000;
last_watchdog_recovery = 0;
last_radio_active_ms = 0;
}
@@ -179,8 +189,12 @@ protected:
virtual uint32_t getCADFailRetryDelay() const;
virtual uint32_t getCADFailMaxDuration() const;
virtual int getInterferenceThreshold() const { return 0; } // disabled by default
virtual bool getCADEnabled() const { return false; } // hardware CAD disabled by default
virtual int getAGCResetInterval() const { return 0; } // disabled by default
virtual uint32_t getRadioWatchdogMillis() const;
virtual unsigned long getDutyCycleWindowMs() const { return 3600000; }
#ifdef WITH_MQTT_BRIDGE
virtual uint32_t getRadioWatchdogMillis() const; // observer-only radio recovery
#endif
public:
void begin();
@@ -190,8 +204,9 @@ public:
void releasePacket(Packet* packet);
void sendPacket(Packet* packet, uint8_t priority, uint32_t delay_millis=0);
unsigned long getTotalAirTime() const { return total_air_time; } // in milliseconds
unsigned long getTotalAirTime() const { return total_air_time; }
unsigned long getReceiveAirTime() const {return rx_air_time; }
unsigned long getRemainingTxBudget() const { return tx_budget_ms; }
uint32_t getNumSentFlood() const { return n_sent_flood; }
uint32_t getNumSentDirect() const { return n_sent_direct; }
uint32_t getNumRecvFlood() const { return n_recv_flood; }
+6
View File
@@ -69,6 +69,12 @@ public:
// dry_run is true the build is only reported, not flashed. Observer (ESP32+WiFi) builds only.
virtual bool otaFromManifest(const char* current_ver, bool dry_run, char reply[]) { return false; }
// LoRa front-end-module LNA (RX gain) control. Only FEM-equipped boards override
// these; others report they can't control it. Driven by NodePrefs.radio_fem_rxgain.
virtual bool setLoRaFemLnaEnabled(bool enable) { return false; }
virtual bool canControlLoRaFemLna() const { return false; }
virtual bool isLoRaFemLnaEnabled() const { return false; }
// Power management interface (boards with power management override these)
virtual bool isExternalPowered() { return false; }
virtual uint16_t getBootVoltage() { return 0; }
+14 -11
View File
@@ -26,8 +26,9 @@
#define ALERT_DEBUG_PRINTLN(...) do {} while (0)
#endif
#ifdef WITH_MQTT_BRIDGE
AlertReporter::AlertReporter()
: _prefs(nullptr), _mesh(nullptr), _callbacks(nullptr),
: _prefs(nullptr), _obs(nullptr), _mesh(nullptr), _callbacks(nullptr),
#ifdef WITH_MQTT_BRIDGE
_bridge(nullptr),
#endif
@@ -38,18 +39,18 @@ AlertReporter::AlertReporter()
#endif
}
void AlertReporter::begin(NodePrefs* prefs, mesh::Mesh* mesh, CommonCLICallbacks* callbacks) {
void AlertReporter::begin(NodePrefs* prefs, MQTTPrefs* obs, mesh::Mesh* mesh, CommonCLICallbacks* callbacks) {
_prefs = prefs;
_obs = obs;
_mesh = mesh;
_callbacks = callbacks;
onConfigChanged();
}
#ifdef WITH_MQTT_BRIDGE
void AlertReporter::setBridge(MQTTBridge* bridge) {
_bridge = bridge;
}
#endif
#endif // WITH_MQTT_BRIDGE (AlertReporter methods, part 1)
// Channels banned as fault-alert destinations. Fault alerts are noisy
// operator-infrastructure messages; routing them to community channels would
@@ -97,6 +98,7 @@ const char* alertReporterBannedChannelMatchHex(const char* psk_hex) {
return alertReporterBannedChannelMatch(secret);
}
#ifdef WITH_MQTT_BRIDGE
bool AlertReporter::resolveChannel(mesh::GroupChannel& out) const {
if (!_prefs) return false;
@@ -105,7 +107,7 @@ bool AlertReporter::resolveChannel(mesh::GroupChannel& out) const {
// Only 16-byte secrets (32 hex chars) are supported; 32-byte channel keys
// are not used anywhere in MeshCore practice and not represented in the
// banned table either.
const char* psk = _prefs->alert_psk_hex;
const char* psk = _obs->alert_psk_hex;
if (strlen(psk) != 32) return false;
memset(out.secret, 0, sizeof(out.secret));
@@ -204,7 +206,7 @@ void AlertReporter::formatAge(unsigned long age_ms, char* out, size_t out_size)
}
void AlertReporter::onLoop(unsigned long now_ms) {
if (!_prefs || !_prefs->alert_enabled) return;
if (!_prefs || !_obs || !_obs->alert_enabled) return;
if (!_mesh) return;
// Throttle: ~5 s cadence. The thresholds are minutes-scale so this is fine.
@@ -216,17 +218,17 @@ void AlertReporter::onLoop(unsigned long now_ms) {
// already enforces this on set, but a stale prefs file or future field
// tweak shouldn't be able to drag the floor below 1 hour and let a
// flapping link spam the mesh.
uint16_t cfg_min = _prefs->alert_min_interval_min;
uint16_t cfg_min = _obs->alert_min_interval_min;
if (cfg_min < 60) cfg_min = 60;
unsigned long min_interval_ms = (unsigned long)cfg_min * 60000UL;
// -------- WiFi fault --------
if (_prefs->alert_wifi_minutes > 0) {
if (_obs->alert_wifi_minutes > 0) {
unsigned long wifi_disc_ms = MQTTBridge::getLastWifiDisconnectTime();
unsigned long wifi_conn_ms = MQTTBridge::getWifiConnectedAtMillis();
bool wifi_down = (wifi_disc_ms != 0 && wifi_conn_ms == 0);
unsigned long down_ms = wifi_down ? (now_ms - wifi_disc_ms) : 0;
unsigned long thresh_ms = (unsigned long)_prefs->alert_wifi_minutes * 60000UL;
unsigned long thresh_ms = (unsigned long)_obs->alert_wifi_minutes * 60000UL;
if (_wifi.state == OK) {
if (wifi_down && down_ms >= thresh_ms &&
@@ -263,10 +265,10 @@ void AlertReporter::onLoop(unsigned long now_ms) {
}
// -------- MQTT slot faults --------
if (_prefs->alert_mqtt_minutes > 0 && _bridge != nullptr) {
if (_obs->alert_mqtt_minutes > 0 && _bridge != nullptr) {
int n = MQTTBridge::getRuntimeSlotCount();
if (n > (int)(sizeof(_mqtt) / sizeof(_mqtt[0]))) n = (int)(sizeof(_mqtt) / sizeof(_mqtt[0]));
unsigned long thresh_ms = (unsigned long)_prefs->alert_mqtt_minutes * 60000UL;
unsigned long thresh_ms = (unsigned long)_obs->alert_mqtt_minutes * 60000UL;
for (int i = 0; i < n; i++) {
Fault& f = _mqtt[i];
@@ -311,3 +313,4 @@ void AlertReporter::onLoop(unsigned long now_ms) {
(void)now_ms;
#endif
}
#endif // WITH_MQTT_BRIDGE (AlertReporter methods, part 2)
+4 -1
View File
@@ -49,6 +49,7 @@ const char* alertReporterBannedChannelMatchHex(const char* psk_hex);
* - WiFi/MQTT polling is #ifdef WITH_MQTT_BRIDGE-gated; without it, the
* reporter still supports manual `alert test` sends.
*/
#ifdef WITH_MQTT_BRIDGE
class AlertReporter {
public:
AlertReporter();
@@ -59,7 +60,7 @@ public:
* it to resolve a TransportKey scope for outgoing alert floods (so the
* packet rides the repeater's default scope or an `alert.region` override).
*/
void begin(NodePrefs* prefs, mesh::Mesh* mesh, CommonCLICallbacks* callbacks = nullptr);
void begin(NodePrefs* prefs, MQTTPrefs* obs, mesh::Mesh* mesh, CommonCLICallbacks* callbacks = nullptr);
#ifdef WITH_MQTT_BRIDGE
/** Bridge can be (re)created lazily; pass nullptr to detach. */
@@ -98,6 +99,7 @@ private:
};
NodePrefs* _prefs;
MQTTPrefs* _obs;
mesh::Mesh* _mesh;
CommonCLICallbacks* _callbacks;
#ifdef WITH_MQTT_BRIDGE
@@ -107,3 +109,4 @@ private:
#endif
unsigned long _next_check_ms;
};
#endif // WITH_MQTT_BRIDGE
+436 -1237
View File
File diff suppressed because it is too large Load Diff
+126 -75
View File
@@ -63,71 +63,18 @@ struct NodePrefs { // persisted to file
uint32_t discovery_mod_timestamp;
float adc_multiplier;
char owner_info[120];
// MQTT settings (stored separately in /mqtt_prefs, but kept here for backward compatibility)
char mqtt_origin[32]; // Device name for MQTT topics
char mqtt_iata[8]; // IATA code for MQTT topics
uint8_t mqtt_status_enabled; // Enable status messages
uint8_t mqtt_packets_enabled; // Enable packet messages
uint8_t mqtt_raw_enabled; // Enable raw messages
uint8_t mqtt_tx_enabled; // TX packet uplinking: 0=off, 1=all, 2=advert (self-originated only)
uint32_t mqtt_status_interval; // Status publish interval (ms)
uint8_t mqtt_rx_enabled; // Enable RX packet uplinking (default: on)
// WiFi settings
char wifi_ssid[32]; // WiFi SSID
char wifi_password[64]; // WiFi password
uint8_t wifi_power_save; // WiFi power save mode: 0=min, 1=none, 2=max (default: 1=none)
// Timezone settings
char timezone_string[32]; // Timezone string (e.g., "America/Los_Angeles")
int8_t timezone_offset; // Timezone offset in hours (-12 to +14) - fallback
// MQTT slot presets (up to MAX_MQTT_SLOTS, each can be a preset name or "custom"/"none")
char mqtt_slot_preset[MAX_MQTT_SLOTS][24]; // e.g. "analyzer-us", "meshmapper", "custom", "none"
// Per-slot custom broker settings (only used when slot preset is "custom")
char mqtt_slot_host[MAX_MQTT_SLOTS][64];
uint16_t mqtt_slot_port[MAX_MQTT_SLOTS];
char mqtt_slot_username[MAX_MQTT_SLOTS][32];
char mqtt_slot_password[MAX_MQTT_SLOTS][64];
// Shared MQTT authentication
char mqtt_owner_public_key[65]; // Owner public key (hex string, same length as repeater public key)
char mqtt_email[64]; // Owner email address for matching nodes with owners
// Per-slot extended fields
char mqtt_slot_token[MAX_MQTT_SLOTS][48]; // Per-slot token (e.g., MeshRank account token)
char mqtt_slot_topic[MAX_MQTT_SLOTS][96]; // Per-slot custom topic template (custom preset only)
char mqtt_slot_audience[MAX_MQTT_SLOTS][64]; // JWT audience (non-empty enables JWT auth for custom slots)
uint8_t loop_detect;
// SNMP settings (optional, only used when WITH_SNMP is defined)
uint8_t snmp_enabled; // boolean: 0=off, 1=on
char snmp_community[24]; // community string (default "public")
uint8_t radio_watchdog_minutes; // 0=disabled, 1-120 minutes
// Restored from upstream (dropped by the 22eb9b87 revert). Persisted at the same
// /com_prefs offsets upstream uses (293, 294) so the file stays upstream-aligned.
uint8_t radio_fem_rxgain; // LoRa FEM RX-gain (LNA); default on. Hardware driving is
// wired per-board in the FEM-restore change; persisted here.
uint8_t cad_enabled; // hardware Channel Activity Detection before TX; default off
// Fault alert channel (LoRa group-channel "observer status" message on prolonged WiFi/MQTT outage).
// Sent over the radio (NOT over MQTT) so the alert still works while the MQTT path is broken.
// All fields are appended at the end of NodePrefs for binary-compatible upgrades.
uint8_t alert_enabled; // 0 = off (default), 1 = on
char alert_psk_hex[33]; // 32 lowercase hex chars (16-byte channel secret) + null; empty = alerts disabled. Banned keys (Public/#test/#bot) are rejected.
uint16_t alert_wifi_minutes; // WiFi-down threshold in minutes (0 = disabled), default 30
uint16_t alert_mqtt_minutes; // MQTT-down threshold in minutes (0 = disabled), default 240 (4 h)
uint16_t alert_min_interval_min; // min minutes between alerts for the same fault, default 60, floor 60
// When the operator configures via `set alert.hashtag <name>`, we derive
// alert_psk_hex from sha256("#name")[0..15] once and remember the hashtag
// text here purely for `get alert.hashtag` readback. A subsequent
// `set alert.psk` clears this field so it doesn't lie about provenance.
char alert_hashtag[24];
// Optional region name (e.g. "us", "eu"); empty = use the repeater's
// default_scope. Looked up lazily via RegionMap::findByNamePrefix at send
// time, so the operator can name a region that doesn't exist yet without
// polluting region_map state. Falls back to default_scope on miss.
char alert_region[31];
// Custom NTP server (MQTT observer); empty = built-in default primary (pool.ntp.org)
char mqtt_ntp_server[64];
// NOTE: observer settings (MQTT/WiFi/timezone/SNMP/alert) were moved out of
// NodePrefs into MQTTPrefs (persisted to /mqtt_prefs) so this struct stays
// aligned with upstream. See struct MQTTPrefs below.
};
#ifdef WITH_MQTT_BRIDGE
@@ -188,24 +135,40 @@ struct MQTTPrefs {
char mqtt_owner_public_key[65]; // Owner public key (hex string)
char mqtt_email[64]; // Owner email address
// --- Legacy fields (vestigial, kept for binary compatibility) ---
// Migration now uses OldMQTTPrefs/ThreeSlotMQTTPrefs structs. These fields are unused
// but must remain to preserve byte offsets for devices that already saved a new-format /mqtt_prefs file.
uint8_t _legacy_analyzer_us_enabled;
uint8_t _legacy_analyzer_eu_enabled;
char _legacy_mqtt_server[64];
uint16_t _legacy_mqtt_port;
char _legacy_mqtt_username[32];
char _legacy_mqtt_password[64];
// --- New fields (appended at end for migration safety) ---
// Per-slot extended fields
char mqtt_slot_token[MAX_MQTT_SLOTS][48]; // Per-slot token (e.g., MeshRank account token)
char mqtt_slot_topic[MAX_MQTT_SLOTS][96]; // Per-slot custom topic template (custom preset only)
char mqtt_slot_audience[MAX_MQTT_SLOTS][64]; // JWT audience (non-empty enables JWT auth for custom slots)
// --- Appended fields (added after initial 6-slot migration) ---
uint8_t mqtt_rx_enabled; // Enable RX packet uplinking (default: on)
char mqtt_ntp_server[64]; // Custom NTP server; empty = pool.ntp.org
// Observer non-MQTT settings (moved out of NodePrefs so this file stays aligned
// with upstream). New fields are appended here so a shorter /mqtt_prefs payload
// from an earlier v1 firmware still loads; the missing tail keeps its default.
uint8_t snmp_enabled; // boolean
char snmp_community[24]; // community string (default "public")
uint8_t radio_watchdog_minutes; // 0=disabled, 1-120 minutes (observer-only radio recovery)
uint8_t alert_enabled; // 0 = off (default)
char alert_psk_hex[33]; // 32 hex chars + null; empty = alerts disabled
uint16_t alert_wifi_minutes; // WiFi-down threshold (0 = disabled), default 30
uint16_t alert_mqtt_minutes; // MQTT-down threshold (0 = disabled), default 240
uint16_t alert_min_interval_min; // min minutes between same-fault alerts, default 60
char alert_hashtag[24]; // readback for `get alert.hashtag`
char alert_region[31]; // optional region override; empty = default_scope
};
// /mqtt_prefs is written with an 8-byte header so the format is self-describing.
// Files with no header are legacy (versionless) and detected by size in loadMQTTPrefs.
// The magic leads with a non-ASCII byte so it can never collide with the first
// bytes of a legacy file, whose payload starts with the mqtt_origin string.
static const uint8_t MQTT_PREFS_MAGIC[4] = {0xF5, 'M', 'Q', 'P'};
static const uint16_t MQTT_PREFS_VERSION = 1; // bump when the MQTTPrefs payload layout changes incompatibly
struct MQTTPrefsHeader {
uint8_t magic[4]; // MQTT_PREFS_MAGIC
uint16_t version; // MQTT_PREFS_VERSION
uint16_t payload_len; // sizeof(MQTTPrefs) at write time (sanity / forward-compat)
};
// 3-slot MQTTPrefs layout — used for migrating from 3-slot to 6-slot format.
@@ -240,6 +203,74 @@ struct ThreeSlotMQTTPrefs {
char mqtt_slot_token[3][48];
char mqtt_slot_topic[3][96];
};
// Versionless 6-slot layout as shipped on mqtt-bridge-implementation-flex (the
// several-thousand-device deployed fleet). This is the current MQTTPrefs minus the
// observer tail, and it still carries the now-removed `_legacy_*` fields mid-struct.
// loadMQTTPrefs reads a headerless file of this size into this struct, then
// field-copies (dropping `_legacy_*`) into the compact versioned MQTTPrefs.
struct Legacy6SlotMQTTPrefs {
char mqtt_origin[32];
char mqtt_iata[8];
uint8_t mqtt_status_enabled;
uint8_t mqtt_packets_enabled;
uint8_t mqtt_raw_enabled;
uint8_t mqtt_tx_enabled;
uint32_t mqtt_status_interval;
char wifi_ssid[32];
char wifi_password[64];
uint8_t wifi_power_save;
char timezone_string[32];
int8_t timezone_offset;
char mqtt_slot_preset[MAX_MQTT_SLOTS][24];
char mqtt_slot_host[MAX_MQTT_SLOTS][64];
uint16_t mqtt_slot_port[MAX_MQTT_SLOTS];
char mqtt_slot_username[MAX_MQTT_SLOTS][32];
char mqtt_slot_password[MAX_MQTT_SLOTS][64];
char mqtt_owner_public_key[65];
char mqtt_email[64];
uint8_t _legacy_analyzer_us_enabled;
uint8_t _legacy_analyzer_eu_enabled;
char _legacy_mqtt_server[64];
uint16_t _legacy_mqtt_port;
char _legacy_mqtt_username[32];
char _legacy_mqtt_password[64];
char mqtt_slot_token[MAX_MQTT_SLOTS][48];
char mqtt_slot_topic[MAX_MQTT_SLOTS][96];
char mqtt_slot_audience[MAX_MQTT_SLOTS][64];
uint8_t mqtt_rx_enabled;
char mqtt_ntp_server[64];
};
// The legacy layouts above describe files already written to the deployed fleet's
// flash, so their sizes are frozen forever — loadMQTTPrefs() tells the eras apart
// by file size and reads each file as a raw struct dump. These asserts pin the
// layouts on every target toolchain; if one fires, the compiler (or an edit to a
// legacy struct or MAX_MQTT_SLOTS) has changed a layout and fleet files would be
// read at wrong offsets.
static_assert(sizeof(MQTTPrefsHeader) == 8, "versioned /mqtt_prefs header must stay 8 bytes");
static_assert(sizeof(OldMQTTPrefs) == 472, "frozen pre-slot /mqtt_prefs layout changed");
static_assert(sizeof(ThreeSlotMQTTPrefs) == 1464, "frozen 3-slot /mqtt_prefs layout changed");
static_assert(sizeof(Legacy6SlotMQTTPrefs) == 2904, "frozen deployed-fleet /mqtt_prefs layout changed");
// Observer settings captured from the trailing block of an old-format /com_prefs
// (fork firmware that predates the NodePrefs -> MQTTPrefs split). loadPrefsInt()
// fills this in when it detects the old file layout; loadMQTTPrefs() then applies
// the values one-time if the loaded /mqtt_prefs predates the appended observer
// fields, so SNMP/watchdog/alert config survives the firmware upgrade.
struct LegacyObserverTail {
bool valid = false;
uint8_t snmp_enabled;
char snmp_community[24];
uint8_t radio_watchdog_minutes;
uint8_t alert_enabled;
char alert_psk_hex[33];
uint16_t alert_wifi_minutes;
uint16_t alert_mqtt_minutes;
uint16_t alert_min_interval_min;
char alert_hashtag[24];
char alert_region[31];
};
#endif
class CommonCLICallbacks {
@@ -352,7 +383,13 @@ class CommonCLI {
char tmp[PRV_KEY_SIZE*2 + 4];
#ifdef WITH_MQTT_BRIDGE
MQTTPrefs _mqtt_prefs;
LegacyObserverTail _legacy_tail;
// /mqtt_prefs carries a version newer than this firmware understands (a downgrade).
// The in-memory prefs run on defaults and saveMQTTPrefs() must not overwrite the
// file, or the first `set` command would destroy the newer config.
bool _mqtt_prefs_hold = false;
#endif
bool _com_prefs_needs_upgrade = false; // old-format /com_prefs detected; rewrite once after load
mesh::RTCClock* getRTCClock() { return _rtc; }
void savePrefs();
@@ -360,14 +397,22 @@ class CommonCLI {
#ifdef WITH_MQTT_BRIDGE
void loadMQTTPrefs(FILESYSTEM* fs);
void saveMQTTPrefs(FILESYSTEM* fs);
void syncMQTTPrefsToNodePrefs();
void syncNodePrefsToMQTTPrefs();
#endif
void handleRegionCmd(char* command, char* reply);
void handleGetCmd(uint32_t sender_timestamp, char* command, char* reply);
void handleSetCmd(uint32_t sender_timestamp, char* command, char* reply);
// Observer/MQTT/WiFi/timezone/alert/SNMP CLI handling lives in the fork-owned
// CommonCLI_Observer.cpp to keep these branches out of the upstream-tracked
// CommonCLI.cpp. Each returns true if it recognized (handled) the command, or
// false to fall through to the base get/set parsing.
bool handleObserverSetCmd(uint32_t sender_timestamp, const char* config, char* reply);
bool handleObserverGetCmd(uint32_t sender_timestamp, const char* config, char* reply);
// Observer-only top-level commands (ota check/update, tls.bundletest, alert test)
// also live in CommonCLI_Observer.cpp; returns true if it handled the command.
bool handleObserverCommand(uint32_t sender_timestamp, char* command, char* reply);
public:
CommonCLI(mesh::MainBoard& board, mesh::RTCClock& rtc, SensorManager& sensors, RegionMap& region_map, ClientACL& acl, NodePrefs* prefs, CommonCLICallbacks* callbacks)
: _board(&board), _rtc(&rtc), _sensors(&sensors), _region_map(&region_map), _acl(&acl), _prefs(prefs), _callbacks(callbacks) { }
@@ -377,4 +422,10 @@ public:
void handleCommand(uint32_t sender_timestamp, char* command, char* reply);
mesh::MainBoard* getBoard() { return _board; }
uint8_t buildAdvertData(uint8_t node_type, uint8_t* app_data);
#ifdef WITH_MQTT_BRIDGE
// Observer config (MQTT/WiFi/timezone/SNMP/alert), persisted to /mqtt_prefs.
// Exposed so the app can hand it to MQTTBridge/AlertReporter, which read these
// fields directly (they no longer live in NodePrefs).
MQTTPrefs* getObserverPrefs() const { return const_cast<MQTTPrefs*>(&_mqtt_prefs); }
#endif
};
+999
View File
@@ -0,0 +1,999 @@
// CommonCLI_Observer.cpp — fork-owned observer/MQTT/WiFi/timezone/alert/SNMP CLI
// command handling, split out of CommonCLI.cpp so the upstream-tracked file carries
// only two small delegation hooks. These are CommonCLI member functions, so they
// retain full access to _prefs/_callbacks/_board/savePrefs() with no re-plumbing.
//
// Behavior is intentionally identical to the previously-inlined branches: MQTT
// commands keep their WITH_MQTT_BRIDGE guard; alert/SNMP commands remain unguarded.
// Each handler returns true if it recognized the command, false to fall through to
// the base get/set parser in CommonCLI.cpp.
#include <Arduino.h>
#include "CommonCLI.h"
#include "TxtDataHelpers.h"
#include "AlertReporter.h" // for alertReporterBannedChannelMatch[Hex]()
#include <Utils.h>
#ifdef ESP_PLATFORM
#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <esp_wifi.h>
#endif
#ifdef WITH_MQTT_BRIDGE
#include "bridges/MQTTBridge.h"
#include "MQTTDefaults.h"
#endif
// Local copy of the busted-libc-safe atoi (the original in CommonCLI.cpp is static).
static uint32_t _atoi(const char* sp) {
uint32_t n = 0;
while (*sp && *sp >= '0' && *sp <= '9') {
n *= 10;
n += (*sp++ - '0');
}
return n;
}
#ifdef ESP_PLATFORM
// Optional embedded CA bundle symbols produced by board_build.embed_files.
// Weak linkage keeps non-bundle builds linkable.
extern const uint8_t rootca_crt_bundle_start[] asm("_binary_src_certs_x509_crt_bundle_bin_start") __attribute__((weak));
extern const uint8_t rootca_crt_bundle_end[] asm("_binary_src_certs_x509_crt_bundle_bin_end") __attribute__((weak));
static bool parseTlsBundleTarget(const char* input, char* host_out, size_t host_out_size, uint16_t* port_out) {
if (!input || !host_out || host_out_size == 0 || !port_out) return false;
while (*input == ' ') input++;
if (*input == '\0') return false;
const char* start = input;
const char* scheme = strstr(input, "://");
if (scheme) start = scheme + 3;
const char* end = start;
while (*end && *end != '/' && *end != '?' && *end != '#') end++;
if (end <= start) return false;
uint16_t port = 443;
const char* host_start = start;
const char* host_end = end;
if (*host_start == '[') {
const char* close = (const char*)memchr(host_start, ']', host_end - host_start);
if (!close) return false;
if ((close + 1) < host_end && *(close + 1) == ':') {
int p = atoi(close + 2);
if (p <= 0 || p > 65535) return false;
port = (uint16_t)p;
}
host_start++;
host_end = close;
} else {
const char* colon = (const char*)memchr(host_start, ':', host_end - host_start);
if (colon) {
int p = atoi(colon + 1);
if (p <= 0 || p > 65535) return false;
port = (uint16_t)p;
host_end = colon;
}
}
size_t host_len = (size_t)(host_end - host_start);
if (host_len == 0 || host_len >= host_out_size) return false;
memcpy(host_out, host_start, host_len);
host_out[host_len] = '\0';
*port_out = port;
return true;
}
#endif
#ifdef WITH_MQTT_BRIDGE
static int getMQTTPresetNameCount() {
// Include virtual presets accepted by CLI parser.
return MQTT_PRESET_COUNT + 2; // built-ins + custom + none
}
static bool isValidNtpHostname(const char* host) {
if (!host || host[0] == '\0') return false;
size_t len = strlen(host);
if (len > 63) return false;
if (host[0] == '.' || host[len - 1] == '.') return false;
for (size_t i = 0; i < len; i++) {
char c = host[i];
if (!((c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') ||
(c >= '0' && c <= '9') || c == '.' || c == '-')) {
return false;
}
}
return true;
}
static const char* getMQTTPresetNameByIndex(int index) {
if (index < MQTT_PRESET_COUNT) return MQTT_PRESETS[index].name;
if (index == MQTT_PRESET_COUNT) return MQTT_PRESET_CUSTOM;
if (index == MQTT_PRESET_COUNT + 1) return MQTT_PRESET_NONE;
return nullptr;
}
static void formatMQTTPresetListReply(char* reply, size_t reply_size, int start) {
if (!reply || reply_size == 0) return;
reply[0] = '\0';
const int total = getMQTTPresetNameCount();
if (start < 0 || start >= total) {
snprintf(reply, reply_size, "Error: preset list start must be 0-%d", total - 1);
return;
}
// Keep room for continuation marker and null terminator.
const size_t reserve_for_next = 18;
size_t used = 0;
bool wrote_any = false;
int index = start;
while (index < total) {
const char* name = getMQTTPresetNameByIndex(index);
if (!name) break;
size_t name_len = strlen(name);
size_t room = reply_size - used;
if (room <= reserve_for_next) break;
size_t needed = name_len + (wrote_any ? 1 : 0); // comma separator
if (needed >= room - reserve_for_next) break;
if (wrote_any) {
reply[used++] = ',';
}
memcpy(reply + used, name, name_len);
used += name_len;
reply[used] = '\0';
wrote_any = true;
index++;
}
if (!wrote_any) {
strcpy(reply, "Error: list page too small");
return;
}
if (index < total) {
snprintf(reply + used, reply_size - used, "... next:%d", index);
}
}
#endif
bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* config, char* reply) {
#ifdef WITH_MQTT_BRIDGE
bool handled = true;
if (memcmp(config, "snmp.community ", 15) == 0) {
StrHelper::strncpy(_mqtt_prefs.snmp_community, &config[15], sizeof(_mqtt_prefs.snmp_community));
savePrefs();
strcpy(reply, "OK - restart to apply");
} else if (memcmp(config, "snmp ", 5) == 0) {
_mqtt_prefs.snmp_enabled = memcmp(&config[5], "on", 2) == 0;
savePrefs();
strcpy(reply, "OK - restart to apply");
} else if (memcmp(config, "radio.watchdog ", 15) == 0) {
const char* val = &config[15];
bool all_digits = (*val != '\0');
for (const char* sp = val; *sp; sp++) {
if (*sp < '0' || *sp > '9') { all_digits = false; break; }
}
if (*val == '\0') {
strcpy(reply, "Error: missing radio.watchdog minutes");
} else if (!all_digits) {
strcpy(reply, "Error: radio.watchdog must be an integer 0-120");
} else {
int mins = atoi(val);
if (mins > 120) {
strcpy(reply, "Error: radio.watchdog must be 0-120 minutes");
} else {
_mqtt_prefs.radio_watchdog_minutes = (uint8_t)mins;
savePrefs();
if (mins == 0) {
strcpy(reply, "OK - radio watchdog disabled");
} else {
sprintf(reply, "OK - radio watchdog %d min", mins);
}
}
}
#ifdef WITH_MQTT_BRIDGE
} else if (strcmp(config, "mqtt.origin") == 0) {
_mqtt_prefs.mqtt_origin[0] = '\0';
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.origin ", 12) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_origin, &config[12], sizeof(_mqtt_prefs.mqtt_origin));
StrHelper::stripSurroundingQuotes(_mqtt_prefs.mqtt_origin, sizeof(_mqtt_prefs.mqtt_origin));
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.iata ", 10) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_iata, &config[10], sizeof(_mqtt_prefs.mqtt_iata));
for (int i = 0; _mqtt_prefs.mqtt_iata[i]; i++) {
_mqtt_prefs.mqtt_iata[i] = toupper(_mqtt_prefs.mqtt_iata[i]);
}
savePrefs();
_callbacks->restartBridge();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.status ", 12) == 0) {
_mqtt_prefs.mqtt_status_enabled = memcmp(&config[12], "on", 2) == 0;
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.packets ", 13) == 0) {
_mqtt_prefs.mqtt_packets_enabled = memcmp(&config[13], "on", 2) == 0;
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.raw ", 9) == 0) {
_mqtt_prefs.mqtt_raw_enabled = memcmp(&config[9], "on", 2) == 0;
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.tx ", 8) == 0) {
if (memcmp(&config[8], "advert", 6) == 0) {
_mqtt_prefs.mqtt_tx_enabled = 2;
} else {
_mqtt_prefs.mqtt_tx_enabled = memcmp(&config[8], "on", 2) == 0 ? 1 : 0;
}
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.rx ", 8) == 0) {
_mqtt_prefs.mqtt_rx_enabled = memcmp(&config[8], "on", 2) == 0 ? 1 : 0;
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.interval ", 14) == 0) {
uint32_t minutes = _atoi(&config[14]);
if (minutes >= 1 && minutes <= 60) {
_mqtt_prefs.mqtt_status_interval = minutes * 60000;
savePrefs();
_callbacks->restartBridge();
sprintf(reply, "OK - interval set to %u minutes (%lu ms), bridge restarted", minutes, (unsigned long)_mqtt_prefs.mqtt_status_interval);
} else {
strcpy(reply, "Error: interval must be between 1-60 minutes");
}
} else if (memcmp(config, "mqtt.ntp ", 9) == 0) {
const char* host = &config[9];
while (*host == ' ') host++;
bool clearing = strcmp(host, "none") == 0;
if (!clearing && !isValidNtpHostname(host)) {
strcpy(reply, "Error: invalid NTP hostname");
} else {
if (clearing) {
_mqtt_prefs.mqtt_ntp_server[0] = '\0';
} else {
StrHelper::strncpy(_mqtt_prefs.mqtt_ntp_server, host, sizeof(_mqtt_prefs.mqtt_ntp_server));
}
savePrefs();
#ifdef ESP_PLATFORM
// Validate by running an immediate sync. syncMqttNtp() marshals onto the MQTT
// task (Core 0) so no NTP I/O happens on this (Core 1) CLI thread.
if (WiFi.status() != WL_CONNECTED) {
strcpy(reply, "OK - saved (WiFi not connected; NTP sync pending)");
} else if (!_callbacks->isMqttBridgeRunning()) {
strcpy(reply, "OK - saved (MQTT bridge not running)");
} else if (_callbacks->syncMqttNtp()) {
strcpy(reply, "OK - time synced");
} else {
strcpy(reply, "Error: NTP sync failed");
}
#else
strcpy(reply, "OK - saved");
#endif
}
} else if (memcmp(config, "wifi.ssid ", 10) == 0) {
StrHelper::strncpy(_mqtt_prefs.wifi_ssid, &config[10], sizeof(_mqtt_prefs.wifi_ssid));
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "wifi.pwd ", 9) == 0) {
StrHelper::strncpy(_mqtt_prefs.wifi_password, &config[9], sizeof(_mqtt_prefs.wifi_password));
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "wifi.powersave ", 15) == 0) {
const char* value = &config[15];
uint8_t ps_value;
bool valid = false;
if (memcmp(value, "min", 3) == 0 && (value[3] == 0 || value[3] == ' ')) {
ps_value = 0;
valid = true;
} else if (memcmp(value, "none", 4) == 0 && (value[4] == 0 || value[4] == ' ')) {
ps_value = 1;
valid = true;
} else if (memcmp(value, "max", 3) == 0 && (value[3] == 0 || value[3] == ' ')) {
ps_value = 2;
valid = true;
}
if (!valid) {
strcpy(reply, "Error: must be none, min, or max");
} else {
_mqtt_prefs.wifi_power_save = ps_value;
savePrefs();
#ifdef ESP_PLATFORM
if (WiFi.status() == WL_CONNECTED) {
wifi_ps_type_t ps_mode = (ps_value == 1) ? WIFI_PS_NONE :
(ps_value == 2) ? WIFI_PS_MAX_MODEM : WIFI_PS_MIN_MODEM;
esp_err_t ps_result = esp_wifi_set_ps(ps_mode);
if (ps_result == ESP_OK) {
const char* ps_name = (ps_value == 1) ? "none" : (ps_value == 2) ? "max" : "min";
sprintf(reply, "OK - power save set to %s", ps_name);
} else {
sprintf(reply, "OK - saved, but failed to apply: %d", ps_result);
}
} else {
const char* ps_name = (ps_value == 1) ? "none" : (ps_value == 2) ? "max" : "min";
sprintf(reply, "OK - saved as %s (will apply on next WiFi connection)", ps_name);
}
#else
const char* ps_name = (ps_value == 1) ? "none" : (ps_value == 2) ? "max" : "min";
sprintf(reply, "OK - saved as %s", ps_name);
#endif
}
} else if (memcmp(config, "timezone ", 9) == 0) {
StrHelper::strncpy(_mqtt_prefs.timezone_string, &config[9], sizeof(_mqtt_prefs.timezone_string));
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(config, "timezone.offset ", 16) == 0) {
int8_t offset = _atoi(&config[16]);
if (offset >= -12 && offset <= 14) {
_mqtt_prefs.timezone_offset = offset;
savePrefs();
strcpy(reply, "OK");
} else {
strcpy(reply, "Error: timezone offset must be between -12 and +14");
}
} else if (config[0] == 'm' && config[1] == 'q' && config[2] == 't' && config[3] == 't' &&
config[4] >= '1' && config[4] <= ('0' + MAX_MQTT_SLOTS) && config[5] == '.') {
// Slot-based commands: set mqtt1.preset <name>, set mqtt1.server <host>, etc.
int slot = config[4] - '1'; // 0-5
const char* subcmd = &config[6];
if (memcmp(subcmd, "preset ", 7) == 0) {
const char* preset_name = &subcmd[7];
// Validate preset name
if (findMQTTPreset(preset_name) != nullptr ||
strcmp(preset_name, MQTT_PRESET_CUSTOM) == 0 ||
strcmp(preset_name, MQTT_PRESET_NONE) == 0) {
// Reject duplicate presets (except "none" and "custom")
int dup_slot = -1;
if (findMQTTPreset(preset_name) != nullptr) {
for (int s = 0; s < MAX_MQTT_SLOTS; s++) {
if (s != slot && strcmp(_mqtt_prefs.mqtt_slot_preset[s], preset_name) == 0) {
dup_slot = s;
break;
}
}
}
if (dup_slot >= 0) {
sprintf(reply, "Error: preset '%s' is already assigned to slot %d", preset_name, dup_slot + 1);
} else {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_preset[slot], preset_name, sizeof(_mqtt_prefs.mqtt_slot_preset[slot]));
savePrefs();
_callbacks->restartBridgeSlot(slot);
// Check if the slot has everything it needs to connect
const MQTTPresetDef* p = findMQTTPreset(preset_name);
if (p && p->topic_style == MQTT_TOPIC_MESHRANK && _mqtt_prefs.mqtt_slot_token[slot][0] == '\0') {
sprintf(reply, "OK - slot %d preset: %s (run 'set mqtt%d.token <your_token>' to connect)", slot + 1, preset_name, slot + 1);
} else if (p && p->topic_style == MQTT_TOPIC_MESHCORE &&
(strlen(_mqtt_prefs.mqtt_iata) == 0 || strcmp(_mqtt_prefs.mqtt_iata, "XXX") == 0)) {
sprintf(reply, "OK - slot %d preset: %s (run 'set mqtt.iata <airport_code>' to publish)", slot + 1, preset_name);
} else if (p && mqttPresetNeedsSlotCredentials(p) &&
(_mqtt_prefs.mqtt_slot_username[slot][0] == '\0' ||
_mqtt_prefs.mqtt_slot_password[slot][0] == '\0')) {
sprintf(reply,
"OK - slot %d preset: %s (run 'set mqtt%d.username <user>' and 'set mqtt%d.password <pass>' to connect)",
slot + 1, preset_name, slot + 1, slot + 1);
} else {
sprintf(reply, "OK - slot %d preset: %s", slot + 1, preset_name);
}
}
} else {
strcpy(reply, "Error: unknown preset. Use 'get mqtt.presets'");
}
} else if (memcmp(subcmd, "server ", 7) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_host[slot], &subcmd[7], sizeof(_mqtt_prefs.mqtt_slot_host[slot]));
savePrefs();
strcpy(reply, "OK");
} else if (memcmp(subcmd, "port ", 5) == 0) {
int port = atoi(&subcmd[5]);
if (port > 0 && port <= 65535) {
_mqtt_prefs.mqtt_slot_port[slot] = port;
savePrefs();
strcpy(reply, "OK");
} else {
strcpy(reply, "Error: port must be between 1 and 65535");
}
} else if (memcmp(subcmd, "username ", 9) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_username[slot], &subcmd[9], sizeof(_mqtt_prefs.mqtt_slot_username[slot]));
savePrefs();
_callbacks->restartBridgeSlot(slot);
strcpy(reply, "OK");
} else if (memcmp(subcmd, "password ", 9) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_password[slot], &subcmd[9], sizeof(_mqtt_prefs.mqtt_slot_password[slot]));
savePrefs();
_callbacks->restartBridgeSlot(slot);
strcpy(reply, "OK");
} else if (memcmp(subcmd, "token ", 6) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_token[slot], &subcmd[6], sizeof(_mqtt_prefs.mqtt_slot_token[slot]));
savePrefs();
_callbacks->restartBridgeSlot(slot);
sprintf(reply, "OK - slot %d token set", slot + 1);
} else if (memcmp(subcmd, "topic ", 6) == 0) {
if (strcmp(_mqtt_prefs.mqtt_slot_preset[slot], "custom") != 0) {
sprintf(reply, "Error: topic template only applies to custom preset slots");
} else {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_topic[slot], &subcmd[6], sizeof(_mqtt_prefs.mqtt_slot_topic[slot]));
savePrefs();
_callbacks->restartBridgeSlot(slot);
sprintf(reply, "OK - slot %d topic: %s", slot + 1, _mqtt_prefs.mqtt_slot_topic[slot]);
}
} else if (memcmp(subcmd, "audience ", 9) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_audience[slot], &subcmd[9], sizeof(_mqtt_prefs.mqtt_slot_audience[slot]));
savePrefs();
_callbacks->restartBridgeSlot(slot);
if (_mqtt_prefs.mqtt_slot_audience[slot][0] != '\0') {
sprintf(reply, "OK - slot %d JWT audience: %s", slot + 1, _mqtt_prefs.mqtt_slot_audience[slot]);
} else {
sprintf(reply, "OK - slot %d JWT audience cleared (using username/password auth)", slot + 1);
}
} else if (memcmp(subcmd, "audience", 8) == 0 && subcmd[8] == '\0') {
// "set mqttN.audience" with no value — clear the audience
_mqtt_prefs.mqtt_slot_audience[slot][0] = '\0';
savePrefs();
_callbacks->restartBridgeSlot(slot);
sprintf(reply, "OK - slot %d JWT audience cleared (using username/password auth)", slot + 1);
} else {
sprintf(reply, "unknown config: %s", config);
}
} else if (memcmp(config, "mqtt.analyzer.us ", 17) == 0) {
const int slot = 0;
if (memcmp(&config[17], "on", 2) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_preset[slot], "analyzer-us", sizeof(_mqtt_prefs.mqtt_slot_preset[slot]));
} else {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_preset[slot], MQTT_PRESET_NONE, sizeof(_mqtt_prefs.mqtt_slot_preset[slot]));
}
savePrefs();
_callbacks->restartBridgeSlot(slot);
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.analyzer.eu ", 17) == 0) {
const int slot = 1;
if (memcmp(&config[17], "on", 2) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_preset[slot], "analyzer-eu", sizeof(_mqtt_prefs.mqtt_slot_preset[slot]));
} else {
StrHelper::strncpy(_mqtt_prefs.mqtt_slot_preset[slot], MQTT_PRESET_NONE, sizeof(_mqtt_prefs.mqtt_slot_preset[slot]));
}
savePrefs();
_callbacks->restartBridgeSlot(slot);
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.owner ", 11) == 0) {
const char* owner_key = &config[11];
int key_len = strlen(owner_key);
if (key_len == 64) {
bool valid_key = true;
for (int i = 0; i < key_len; i++) {
if (!((owner_key[i] >= '0' && owner_key[i] <= '9') ||
(owner_key[i] >= 'A' && owner_key[i] <= 'F') ||
(owner_key[i] >= 'a' && owner_key[i] <= 'f'))) {
valid_key = false;
break;
}
}
if (valid_key) {
StrHelper::strncpy(_mqtt_prefs.mqtt_owner_public_key, owner_key, sizeof(_mqtt_prefs.mqtt_owner_public_key));
savePrefs();
strcpy(reply, "OK");
} else {
strcpy(reply, "Error: invalid hex characters in public key");
}
} else {
strcpy(reply, "Error: public key must be 64 hex characters (32 bytes)");
}
} else if (memcmp(config, "mqtt.email ", 11) == 0) {
StrHelper::strncpy(_mqtt_prefs.mqtt_email, &config[11], sizeof(_mqtt_prefs.mqtt_email));
savePrefs();
strcpy(reply, "OK");
#endif
} else if (memcmp(config, "alert ", 6) == 0) {
// set alert on|off
const char* val = &config[6];
if (memcmp(val, "on", 2) == 0 && (val[2] == 0 || val[2] == ' ')) {
_mqtt_prefs.alert_enabled = 1;
savePrefs();
_callbacks->onAlertConfigChanged();
strcpy(reply, "OK - alerts on");
} else if (memcmp(val, "off", 3) == 0 && (val[3] == 0 || val[3] == ' ')) {
_mqtt_prefs.alert_enabled = 0;
savePrefs();
_callbacks->onAlertConfigChanged();
strcpy(reply, "OK - alerts off");
} else {
strcpy(reply, "Error: usage set alert on|off");
}
} else if (memcmp(config, "alert.psk", 9) == 0 && (config[9] == 0 || config[9] == ' ')) {
// `set alert.psk` with no argument clears the field (alerts then disabled
// until a new psk/hashtag is configured).
const char* val = (config[9] == ' ') ? &config[10] : "";
while (*val == ' ') val++;
size_t len = strlen(val);
if (len == 0) {
_mqtt_prefs.alert_psk_hex[0] = '\0';
_mqtt_prefs.alert_hashtag[0] = '\0';
savePrefs();
_callbacks->onAlertConfigChanged();
strcpy(reply, "OK - alert.psk cleared (alerts disabled until configured)");
} else if (val[0] == '#') {
strcpy(reply, "Error: use 'set alert.hashtag' for hashtag channels");
} else if (len != 32) {
// 16-byte channel secret = 32 hex chars. This is what the mobile app's
// "Share Channel" emits, what `set alert.hashtag` derives, and what the
// BANNED_ALERT_CHANNELS table holds. 32-byte channels aren't used
// anywhere in MeshCore practice.
strcpy(reply, "Error: PSK must be 32 hex chars (16-byte channel secret)");
} else {
// Validate all-hex, then normalize via fromHex/toHex so the stored
// form is always lowercase regardless of input case.
uint8_t raw[16];
bool all_hex = true;
for (size_t i = 0; i < len; i++) {
if (!mesh::Utils::isHexChar(val[i])) { all_hex = false; break; }
}
if (!all_hex || !mesh::Utils::fromHex(raw, 16, val)) {
strcpy(reply, "Error: PSK must be 32 hex chars (16-byte channel secret)");
} else {
char normalized[33];
mesh::Utils::toHex(normalized, raw, 16);
if (const char* banned = alertReporterBannedChannelMatchHex(normalized)) {
// Refuse any key on the banned channel list (Public PSK, well-known
// auto-responder hashtags like #test/#bot, etc.). Fault alerts on
// those channels would spam every node in the area.
sprintf(reply, "Error: refusing banned channel '%s'; pick a private key or hashtag", banned);
} else {
StrHelper::strncpy(_mqtt_prefs.alert_psk_hex, normalized, sizeof(_mqtt_prefs.alert_psk_hex));
// The new PSK is operator-supplied, so any previously-derived
// hashtag name is no longer accurate provenance — drop it.
_mqtt_prefs.alert_hashtag[0] = '\0';
savePrefs();
_callbacks->onAlertConfigChanged();
strcpy(reply, "OK - alert.psk updated");
}
}
}
} else if (memcmp(config, "alert.hashtag", 13) == 0 && (config[13] == 0 || config[13] == ' ')) {
const char* val = (config[13] == ' ') ? &config[14] : "";
while (*val == ' ') val++;
size_t in_len = strlen(val);
if (in_len == 0) {
_mqtt_prefs.alert_psk_hex[0] = '\0';
_mqtt_prefs.alert_hashtag[0] = '\0';
savePrefs();
_callbacks->onAlertConfigChanged();
strcpy(reply, "OK - alert.hashtag cleared (alerts disabled until configured)");
} else {
// Canonical stored form is "#name" because the leading '#' is part of
// the sha256 input (matching the companion-app hashtag-channel
// derivation in docs/companion_protocol.md). Accept the user typing
// either "alerts" or "#alerts".
char hashtag[sizeof(_mqtt_prefs.alert_hashtag)];
size_t need = (val[0] == '#') ? in_len : in_len + 1;
if (need >= sizeof(hashtag)) {
strcpy(reply, "Error: hashtag too long");
} else {
if (val[0] == '#') {
StrHelper::strncpy(hashtag, val, sizeof(hashtag));
} else {
hashtag[0] = '#';
StrHelper::strncpy(&hashtag[1], val, sizeof(hashtag) - 1);
}
// Derive the channel key once: first 16 bytes of sha256("#name"),
// store hex-encoded in alert_psk_hex. We don't re-derive on every
// send — operators can later override with `set alert.psk` without
// leaving stale hashtag text behind.
uint8_t digest[32];
mesh::Utils::sha256(digest, sizeof(digest),
(const uint8_t*)hashtag, (int)strlen(hashtag));
if (const char* banned = alertReporterBannedChannelMatch(digest)) {
// Hashtag derives to a banned key (e.g. `set alert.hashtag test`
// hits the #test entry). Refuse before clobbering existing config.
sprintf(reply, "Error: refusing banned channel '%s'", banned);
} else {
char hex[33];
mesh::Utils::toHex(hex, digest, 16);
StrHelper::strncpy(_mqtt_prefs.alert_hashtag, hashtag, sizeof(_mqtt_prefs.alert_hashtag));
StrHelper::strncpy(_mqtt_prefs.alert_psk_hex, hex, sizeof(_mqtt_prefs.alert_psk_hex));
savePrefs();
_callbacks->onAlertConfigChanged();
sprintf(reply, "OK - alert.hashtag: %s", _mqtt_prefs.alert_hashtag);
}
}
}
} else if (memcmp(config, "alert.region", 12) == 0 && (config[12] == 0 || config[12] == ' ')) {
// `set alert.region <name>` overrides the repeater's default_scope for
// alert sends only. `set alert.region` (no arg) clears it. The name is
// looked up lazily via RegionMap at send time; we deliberately don't
// mutate the region map here, so naming an unknown region is allowed
// but will silently fall back to default_scope until the operator runs
// `region put` for it.
const char* val = (config[12] == ' ') ? &config[13] : "";
while (*val == ' ') val++;
size_t len = strlen(val);
if (len == 0) {
_mqtt_prefs.alert_region[0] = '\0';
savePrefs();
_callbacks->onAlertConfigChanged();
strcpy(reply, "OK - alert.region cleared (using default scope)");
} else if (len >= sizeof(_mqtt_prefs.alert_region)) {
strcpy(reply, "Error: alert.region too long");
} else {
StrHelper::strncpy(_mqtt_prefs.alert_region, val, sizeof(_mqtt_prefs.alert_region));
StrHelper::stripSurroundingQuotes(_mqtt_prefs.alert_region, sizeof(_mqtt_prefs.alert_region));
savePrefs();
_callbacks->onAlertConfigChanged();
sprintf(reply, "OK - alert.region: %s", _mqtt_prefs.alert_region);
}
} else if (memcmp(config, "alert.wifi ", 11) == 0) {
int mins = (int)_atoi(&config[11]);
if (mins < 0 || mins > 1440) {
strcpy(reply, "Error: alert.wifi must be 0-1440 minutes (0=off)");
} else {
_mqtt_prefs.alert_wifi_minutes = (uint16_t)mins;
savePrefs();
sprintf(reply, "OK - alert.wifi %d min%s", mins, mins == 0 ? " (disabled)" : "");
}
} else if (memcmp(config, "alert.mqtt ", 11) == 0) {
int mins = (int)_atoi(&config[11]);
if (mins < 0 || mins > 10080) {
strcpy(reply, "Error: alert.mqtt must be 0-10080 minutes (0=off)");
} else {
_mqtt_prefs.alert_mqtt_minutes = (uint16_t)mins;
savePrefs();
sprintf(reply, "OK - alert.mqtt %d min%s", mins, mins == 0 ? " (disabled)" : "");
}
} else if (memcmp(config, "alert.interval ", 15) == 0) {
int mins = (int)_atoi(&config[15]);
// Floor at 60 min: faster re-fires would let a flapping link spam the
// mesh with a fresh GRP_TXT flood every minute — terrible for airtime.
if (mins < 60 || mins > 10080) {
strcpy(reply, "Error: alert.interval must be 60-10080 minutes");
} else {
_mqtt_prefs.alert_min_interval_min = (uint16_t)mins;
savePrefs();
sprintf(reply, "OK - alert.interval %d min", mins);
}
} else {
handled = false;
}
return handled;
#else
(void)sender_timestamp; (void)config; (void)reply;
return false;
#endif
}
bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* config, char* reply) {
#ifdef WITH_MQTT_BRIDGE
bool handled = true;
if (memcmp(config, "snmp.community", 14) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.snmp_community);
} else if (memcmp(config, "snmp", 4) == 0 && (config[4] == '\0' || config[4] == '\n' || config[4] == '\r')) {
strcpy(reply, _mqtt_prefs.snmp_enabled ? "> on" : "> off");
} else if (memcmp(config, "radio.watchdog", 14) == 0) {
sprintf(reply, "> %d", (uint32_t)_mqtt_prefs.radio_watchdog_minutes);
#ifdef WITH_MQTT_BRIDGE
} else if (memcmp(config, "mqtt.origin", 11) == 0) {
char effective_origin[32];
MQTTBridge::getEffectiveMqttOrigin(_prefs, &_mqtt_prefs, effective_origin, sizeof(effective_origin));
sprintf(reply, "> %s", effective_origin);
} else if (memcmp(config, "mqtt.iata", 9) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_iata);
} else if (memcmp(config, "mqtt.presets", 12) == 0 && (config[12] == '\0' || config[12] == ' ')) {
int start = 0;
if (config[12] == ' ') {
const char* start_arg = &config[13];
if (*start_arg == '\0') {
strcpy(reply, "Error: usage get mqtt.presets [start]");
return true;
}
for (const char* sp = start_arg; *sp; sp++) {
if (*sp < '0' || *sp > '9') {
strcpy(reply, "Error: usage get mqtt.presets [start]");
return true;
}
}
start = (int)_atoi(start_arg);
}
formatMQTTPresetListReply(reply, 160, start);
} else if (memcmp(config, "mqtt.status", 11) == 0) {
MQTTBridge::formatMqttStatusReply(reply, 160, &_mqtt_prefs);
} else if (memcmp(config, "mqtt.packets", 12) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_packets_enabled ? "on" : "off");
} else if (memcmp(config, "mqtt.raw", 8) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_raw_enabled ? "on" : "off");
} else if (memcmp(config, "mqtt.tx", 7) == 0) {
const char* tx_str = _mqtt_prefs.mqtt_tx_enabled == 2 ? "advert" : (_mqtt_prefs.mqtt_tx_enabled ? "on" : "off");
sprintf(reply, "> %s", tx_str);
} else if (memcmp(config, "mqtt.rx", 7) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_rx_enabled ? "on" : "off");
} else if (memcmp(config, "mqtt.interval", 13) == 0) {
uint32_t minutes = (_mqtt_prefs.mqtt_status_interval + 29999) / 60000;
sprintf(reply, "> %u minutes (%lu ms)", minutes, (unsigned long)_mqtt_prefs.mqtt_status_interval);
} else if (memcmp(config, "mqtt.ntp.diag", 13) == 0 && (config[13] == '\0' || config[13] == ' ')) {
#ifdef ESP_PLATFORM
// Connectivity probe across all configured NTP servers; never updates the clock.
// Serial console (sender_timestamp == 0) gets a detailed table; LoRa gets a compact list.
if (WiFi.status() != WL_CONNECTED) {
strcpy(reply, "Error: WiFi not connected");
} else if (!_callbacks->isMqttBridgeRunning()) {
strcpy(reply, "Error: MQTT bridge not running");
} else if (!_callbacks->runMqttNtpDiag(reply, 160, sender_timestamp == 0)) {
strcpy(reply, "Error: NTP diag unavailable");
}
#else
strcpy(reply, "Error: not supported on this platform");
#endif
} else if (memcmp(config, "mqtt.ntp", 8) == 0 && (config[8] == '\0' || config[8] == ' ')) {
sprintf(reply, "> %s", MQTTBridge::effectiveNtpPrimary(&_mqtt_prefs));
} else if (config[0] == 'm' && config[1] == 'q' && config[2] == 't' && config[3] == 't' &&
config[4] >= '1' && config[4] <= ('0' + MAX_MQTT_SLOTS) && config[5] == '.') {
// Slot-based commands: get mqtt1.preset, get mqtt1.server, etc.
int slot = config[4] - '1'; // 0-5
const char* subcmd = &config[6];
if (memcmp(subcmd, "preset", 6) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_preset[slot]);
} else if (memcmp(subcmd, "server", 6) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_host[slot]);
} else if (memcmp(subcmd, "port", 4) == 0) {
sprintf(reply, "> %d", _mqtt_prefs.mqtt_slot_port[slot]);
} else if (memcmp(subcmd, "username", 8) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_username[slot]);
} else if (memcmp(subcmd, "password", 8) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_password[slot]);
} else if (memcmp(subcmd, "token", 5) == 0) {
if (_mqtt_prefs.mqtt_slot_token[slot][0] != '\0') {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_token[slot]);
} else {
strcpy(reply, "> (not set)");
}
} else if (memcmp(subcmd, "topic", 5) == 0) {
if (_mqtt_prefs.mqtt_slot_topic[slot][0] != '\0') {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_topic[slot]);
} else {
strcpy(reply, "> (default: meshcore/{iata}/{device}/{type})");
}
} else if (memcmp(subcmd, "audience", 8) == 0) {
if (_mqtt_prefs.mqtt_slot_audience[slot][0] != '\0') {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_audience[slot]);
} else {
strcpy(reply, "> (not set — custom slots use username/password auth)");
}
} else if (memcmp(subcmd, "diag", 4) == 0) {
MQTTBridge::formatSlotDiagReply(reply, 160, slot);
} else {
sprintf(reply, "??: %s", config);
}
} else if (memcmp(config, "wifi.ssid", 9) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.wifi_ssid);
} else if (memcmp(config, "wifi.pwd", 8) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.wifi_password);
} else if (memcmp(config, "wifi.status", 11) == 0) {
wl_status_t status = WiFi.status();
const char* status_str;
switch (status) {
case WL_CONNECTED: status_str = "connected"; break;
case WL_NO_SSID_AVAIL: status_str = "no_ssid"; break;
case WL_CONNECT_FAILED: status_str = "connect_failed"; break;
case WL_CONNECTION_LOST: status_str = "connection_lost"; break;
case WL_DISCONNECTED: status_str = "disconnected"; break;
case 255: status_str = "not_started"; break;
default: status_str = "unknown"; break;
}
if (status == WL_CONNECTED) {
sprintf(reply, "> %s, IP: %s, RSSI: %d dBm", status_str, WiFi.localIP().toString().c_str(), WiFi.RSSI());
#ifdef WITH_MQTT_BRIDGE
unsigned long connect_at = MQTTBridge::getWifiConnectedAtMillis();
if (connect_at != 0) {
unsigned long uptime_ms = millis() - connect_at;
unsigned long uptime_sec = uptime_ms / 1000;
unsigned long d = uptime_sec / 86400;
unsigned long h = (uptime_sec % 86400) / 3600;
unsigned long m = (uptime_sec % 3600) / 60;
unsigned long s = uptime_sec % 60;
size_t len = strlen(reply);
const size_t reply_remaining = 128;
if (d > 0) {
snprintf(reply + len, reply_remaining, ", uptime: %lud %luh %lum %lus", d, h, m, s);
} else if (h > 0) {
snprintf(reply + len, reply_remaining, ", uptime: %luh %lum %lus", h, m, s);
} else if (m > 0) {
snprintf(reply + len, reply_remaining, ", uptime: %lum %lus", m, s);
} else {
snprintf(reply + len, reply_remaining, ", uptime: %lus", s);
}
}
#endif
} else {
#ifdef WITH_MQTT_BRIDGE
uint8_t reason = MQTTBridge::getLastWifiDisconnectReason();
if (reason != 0) {
const char* desc = MQTTBridge::wifiReasonStr(reason);
if (desc) {
sprintf(reply, "> %s: %s (reason: %d)", status_str, desc, reason);
} else {
sprintf(reply, "> %s: reason %d", status_str, reason);
}
} else {
sprintf(reply, "> %s (code: %d)", status_str, status);
}
#else
sprintf(reply, "> %s (code: %d)", status_str, status);
#endif
}
} else if (memcmp(config, "wifi.powersave", 14) == 0) {
uint8_t ps = _mqtt_prefs.wifi_power_save;
const char* ps_name = (ps == 1) ? "none" : (ps == 2) ? "max" : "min";
sprintf(reply, "> %s", ps_name);
} else if (memcmp(config, "timezone", 8) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.timezone_string);
} else if (memcmp(config, "timezone.offset", 15) == 0) {
sprintf(reply, "> %d", _mqtt_prefs.timezone_offset);
} else if (memcmp(config, "mqtt.analyzer.us", 17) == 0) {
sprintf(reply, "> %s", strcmp(_mqtt_prefs.mqtt_slot_preset[0], "analyzer-us") == 0 ? "on" : "off");
} else if (memcmp(config, "mqtt.analyzer.eu", 17) == 0) {
sprintf(reply, "> %s", strcmp(_mqtt_prefs.mqtt_slot_preset[1], "analyzer-eu") == 0 ? "on" : "off");
} else if (sender_timestamp == 0 && memcmp(config, "mqtt.owner", 10) == 0) {
if (_mqtt_prefs.mqtt_owner_public_key[0] != '\0') {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_owner_public_key);
} else {
strcpy(reply, "> (not set)");
}
} else if (sender_timestamp == 0 && memcmp(config, "mqtt.email", 10) == 0) {
if (_mqtt_prefs.mqtt_email[0] != '\0') {
sprintf(reply, "> %s", _mqtt_prefs.mqtt_email);
} else {
strcpy(reply, "> (not set)");
}
} else if (memcmp(config, "mqtt.config.valid", 17) == 0) {
bool valid = MQTTBridge::isConfigValid(&_mqtt_prefs);
sprintf(reply, "> %s", valid ? "valid" : "invalid");
#endif
} else if (memcmp(config, "alert.hashtag", 13) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.alert_hashtag[0] ? _mqtt_prefs.alert_hashtag : "(unset)");
} else if (sender_timestamp == 0 && memcmp(config, "alert.psk", 9) == 0) { // from serial command line only
sprintf(reply, "> %s", _mqtt_prefs.alert_psk_hex[0] ? _mqtt_prefs.alert_psk_hex : "(unset)");
} else if (memcmp(config, "alert.region", 12) == 0) {
sprintf(reply, "> %s", _mqtt_prefs.alert_region[0] ? _mqtt_prefs.alert_region : "(unset, using default scope)");
} else if (memcmp(config, "alert.wifi", 10) == 0) {
sprintf(reply, "> %u min%s", (unsigned)_mqtt_prefs.alert_wifi_minutes,
_mqtt_prefs.alert_wifi_minutes == 0 ? " (disabled)" : "");
} else if (memcmp(config, "alert.mqtt", 10) == 0) {
sprintf(reply, "> %u min%s", (unsigned)_mqtt_prefs.alert_mqtt_minutes,
_mqtt_prefs.alert_mqtt_minutes == 0 ? " (disabled)" : "");
} else if (memcmp(config, "alert.interval", 14) == 0) {
sprintf(reply, "> %u min", (unsigned)_mqtt_prefs.alert_min_interval_min);
} else if (memcmp(config, "alert", 5) == 0 && (config[5] == 0 || config[5] == '\n' || config[5] == '\r')) {
sprintf(reply, "> %s", _mqtt_prefs.alert_enabled ? "on" : "off");
} else {
handled = false;
}
return handled;
#else
(void)sender_timestamp; (void)config; (void)reply;
return false;
#endif
}
bool CommonCLI::handleObserverCommand(uint32_t sender_timestamp, char* command, char* reply) {
#ifdef WITH_MQTT_BRIDGE
if (memcmp(command, "tls.bundletest ", 15) == 0) {
#ifdef ESP_PLATFORM
if (WiFi.status() != WL_CONNECTED) {
strcpy(reply, "ERR: WiFi not connected");
} else {
size_t bundle_len = 0;
if (rootca_crt_bundle_start != nullptr &&
rootca_crt_bundle_end != nullptr &&
rootca_crt_bundle_end > rootca_crt_bundle_start) {
bundle_len = static_cast<size_t>(rootca_crt_bundle_end - rootca_crt_bundle_start);
}
if (bundle_len == 0) {
strcpy(reply, "ERR: no embedded cert bundle");
} else {
char host[96];
uint16_t port = 443;
if (!parseTlsBundleTarget(command + 15, host, sizeof(host), &port)) {
strcpy(reply, "ERR: usage tls.bundletest <host[:port]|url>");
} else {
WiFiClientSecure client;
#if ESP_ARDUINO_VERSION_MAJOR >= 3
client.setCACertBundle(rootca_crt_bundle_start, bundle_len);
#else
client.setCACertBundle(rootca_crt_bundle_start);
#endif
client.setTimeout(8000);
bool ok = client.connect(host, port);
if (ok) {
client.stop();
snprintf(reply, 160, "OK: TLS bundle verified %s:%u", host, (unsigned)port);
} else {
snprintf(reply, 160, "ERR: TLS bundle failed %s:%u", host, (unsigned)port);
}
}
}
}
#else
strcpy(reply, "ERR: unsupported on this platform");
#endif
return true;
} else if (memcmp(command, "ota check", 9) == 0 || memcmp(command, "ota update", 10) == 0) {
// Observer pull-OTA: fetch this variant's build from the baked-in manifest
// and flash it. Intentionally a separate command from "start ota" (the
// manual ElegantOTA web-upload SoftAP) so a remote/online update is never
// triggered by someone expecting to hand-upload a binary.
// ota check -> report available build, do not flash
// ota update -> download and flash, then reboot
#if defined(WITH_MQTT_BRIDGE) && defined(OTA_MANIFEST_BASE)
if (WiFi.status() != WL_CONNECTED) {
strcpy(reply, "ERR: WiFi not connected");
} else if (memcmp(command, "ota check", 9) == 0) {
// Check is synchronous so its result lands in this reply, and runs with the
// MQTT bridge UP: the slim per-variant manifest is tiny, so the fetch only
// costs a single TLS handshake (no large JSON doc) — which fits alongside
// the live MQTT sessions even on no-PSRAM boards. No bridge bounce needed.
_board->otaFromManifest(_callbacks->getFirmwareVer(), true, reply);
} else {
// `ota update`: cheap pre-check first (plain HTTP, bridge stays up). Only
// schedule the real update — which tears the bridge down, flashes, and
// reboots — when an applicable build actually exists. otaFromManifest(dry)
// returns true iff so; otherwise it leaves the explanation (up to date /
// cable flash / error) in reply, which we send without disturbing the
// bridge or misleading the user with a "Beginning update..." that no-ops.
if (_board->otaFromManifest(_callbacks->getFirmwareVer(), true, reply)) {
// reply now holds "update available: <cur> -> <target> (N behind|new base)",
// where <target> is "vX.Y.Z.B (hash)". Pull <target> out for a friendlier
// start message. The "-> " ... trailing " (" framing is produced by
// ESP32Board::otaFromManifestImpl; <target> ends at the LAST " (" (the
// "(N behind)"/"(new base)" suffix), since the version's own hash-paren
// comes before it.
char target[48] = {0};
const char* arrow = strstr(reply, "-> ");
if (arrow) {
arrow += 3;
const char* suffix = nullptr;
for (const char* p = arrow; (p = strstr(p, " (")) != nullptr; p++) suffix = p;
size_t len = suffix ? (size_t)(suffix - arrow) : strlen(arrow);
if (len >= sizeof(target)) len = sizeof(target) - 1;
memcpy(target, arrow, len);
target[len] = 0;
}
// Update is DEFERRED so this ack goes out over the mesh before the flash
// blocks the loop and reboots (the app loop runs it shortly).
if (_callbacks->beginDeferredOtaUpdate()) {
if (target[0]) {
snprintf(reply, 160, "Updating to %s; reboots when done (~30s offline). Check 'ver' after.", target);
} else {
strcpy(reply, "Beginning update... (node will reboot if successful)");
}
} else {
strcpy(reply, "ERR: online OTA not available");
}
}
}
#else
strcpy(reply, "ERR: online OTA not supported on this build");
#endif
return true;
} else if (memcmp(command, "alert test", 10) == 0 && (command[10] == 0 || command[10] == ' ')) {
// Send a one-off test alert on the configured alert channel.
const char* extra = command[10] == ' ' ? &command[11] : "";
char text[120];
if (*extra) {
snprintf(text, sizeof(text), "[test] %s", extra);
} else {
strcpy(text, "[test] alert channel ok");
}
if (!_mqtt_prefs.alert_psk_hex[0]) {
strcpy(reply, "Error: alert channel not configured (set alert.psk or set alert.hashtag)");
} else {
bool ok = _callbacks->sendAlertText(text);
strcpy(reply, ok ? "OK - alert sent" : "Error: alert send failed (bad PSK or PUBLIC key refused?)");
}
return true;
}
return false;
#else
(void)sender_timestamp; (void)command; (void)reply;
return false;
#endif
}
+7
View File
@@ -93,6 +93,13 @@ static inline void applyMQTTDefaults(MQTTPrefs* prefs) {
prefs->timezone_string[sizeof(prefs->timezone_string) - 1] = '\0';
}
prefs->timezone_offset = MQTT_DEFAULT_TIMEZONE_OFFSET;
// Observer non-MQTT defaults (moved out of NodePrefs/MyMesh ctor in Phase 2).
strncpy(prefs->snmp_community, "public", sizeof(prefs->snmp_community) - 1);
prefs->radio_watchdog_minutes = 5;
prefs->alert_wifi_minutes = 30;
prefs->alert_mqtt_minutes = 240;
prefs->alert_min_interval_min = 60;
}
#endif // WITH_MQTT_BRIDGE
+99
View File
@@ -0,0 +1,99 @@
#pragma once
#include <MeshCore.h>
#include <helpers/StaticPoolPacketManager.h>
// Fork-owned (not upstream-tracked). Observer builds capture every received packet
// to MQTT, but RX processing needs a free pool packet first — Dispatcher::checkRecv()
// discards the received bytes before logRx() when allocNew() fails. Under duty-cycle
// throttling the outbound queue can park the entire pool waiting on TX budget, which
// starves RX allocation and silently caps MQTT capture at the TX rate (each completed
// TX frees exactly one packet for exactly one more RX). Parked retransmissions also
// absorb every budget refill, starving the node's own CLI responses/ACKs and making
// a heavily-throttled node un-administrable over the mesh.
//
// Two policies fix this, both confined to this manager:
//
// 1. Priority-aware shedding. Below the RX reserve, only low-priority outbound
// (priority > 1: multi-hop flood repeats, adverts, trace) is refused; the node's
// own responses/ACKs (pri 0) and login/PATH replies (pri 1) still queue. Below the
// smaller emergency floor everything is shed to keep capture alive.
// 2. Stale-packet expiry. A queued packet still untransmitted STALE_OUTBOUND_MS past
// its scheduled time is dropped at the next dequeue — a repeat delayed that long is
// noise (the flood has long since propagated), and a CLI response that old has
// already timed out at the client. Under normal load the queue drains in
// milliseconds and this never triggers; under throttle it frees the pool and lets
// fresh traffic (including admin responses) compete for the trickle of TX budget.
class RxReservePacketManager : public StaticPoolPacketManager {
int _rx_reserve, _emergency_floor;
int _cap;
// scheduled_for per queued packet, keyed by packet pointer. The pool is a fixed set
// of _cap Packet objects, so _cap slots cover every possible key with no eviction.
struct AgeEntry { mesh::Packet* pkt; uint32_t scheduled_for; };
AgeEntry* _ages;
static const uint32_t STALE_OUTBOUND_MS = 30000;
static const uint8_t MAX_PROTECTED_PRI = 1; // pri 0-1 = own responses/ACKs/replies
void recordAge(mesh::Packet* packet, uint32_t scheduled_for) {
int empty = -1;
for (int i = 0; i < _cap; i++) {
if (_ages[i].pkt == packet) { _ages[i].scheduled_for = scheduled_for; return; }
if (empty < 0 && _ages[i].pkt == NULL) empty = i;
}
if (empty >= 0) { _ages[empty].pkt = packet; _ages[empty].scheduled_for = scheduled_for; }
}
bool lookupAge(const mesh::Packet* packet, uint32_t* scheduled_for) const {
for (int i = 0; i < _cap; i++) {
if (_ages[i].pkt == packet) { *scheduled_for = _ages[i].scheduled_for; return true; }
}
return false;
}
public:
RxReservePacketManager(int pool_size, int rx_reserve)
: StaticPoolPacketManager(pool_size), _rx_reserve(rx_reserve),
_emergency_floor(rx_reserve / 2), _cap(pool_size) {
_ages = new AgeEntry[pool_size];
for (int i = 0; i < pool_size; i++) { _ages[i].pkt = NULL; _ages[i].scheduled_for = 0; }
}
void queueOutbound(mesh::Packet* packet, uint8_t priority, uint32_t scheduled_for) override {
int free_count = getFreeCount();
if (free_count < _emergency_floor
|| (free_count < _rx_reserve && priority > MAX_PROTECTED_PRI)) {
MESH_DEBUG_PRINTLN("RxReservePacketManager: pool below RX reserve, shedding outbound (pri %d)", (int)priority);
free(packet);
return;
}
recordAge(packet, scheduled_for);
StaticPoolPacketManager::queueOutbound(packet, priority, scheduled_for);
}
mesh::Packet* getNextOutbound(uint32_t now) override {
// Expire queued packets that have waited too long past their scheduled time.
for (int i = getOutboundTotal() - 1; i >= 0; i--) {
mesh::Packet* pkt = getOutboundByIdx(i);
uint32_t scheduled_for;
if (pkt && lookupAge(pkt, &scheduled_for)
&& (int32_t)(now - scheduled_for) > (int32_t)STALE_OUTBOUND_MS) {
MESH_DEBUG_PRINTLN("RxReservePacketManager: dropping stale queued outbound");
removeOutboundByIdx(i);
free(pkt);
}
}
return StaticPoolPacketManager::getNextOutbound(now);
}
};
// The packet manager for an app build: observer builds reserve a quarter of the pool
// for RX so MQTT capture survives duty-cycle throttling; non-observer builds keep the
// upstream pool behavior unchanged.
inline mesh::PacketManager* createObserverPacketManager(int pool_size) {
#ifdef WITH_MQTT_BRIDGE
return new RxReservePacketManager(pool_size, pool_size / 4);
#else
return new StaticPoolPacketManager(pool_size);
#endif
}
+6
View File
@@ -9,6 +9,8 @@ PacketQueue::PacketQueue(int max_entries) {
}
int PacketQueue::countBefore(uint32_t now) const {
if (now == 0xFFFFFFFF) return _num; // sentinel: count all entries regardless of schedule
int n = 0;
for (int j = 0; j < _num; j++) {
if ((int32_t)(_schedule_table[j] - now) > 0) continue; // scheduled for future... ignore for now
@@ -97,6 +99,10 @@ int StaticPoolPacketManager::getOutboundCount(uint32_t now) const {
return send_queue.countBefore(now);
}
int StaticPoolPacketManager::getOutboundTotal() const {
return send_queue.count();
}
int StaticPoolPacketManager::getFreeCount() const {
return unused.count();
}
+1
View File
@@ -29,6 +29,7 @@ public:
void queueOutbound(mesh::Packet* packet, uint8_t priority, uint32_t scheduled_for) override;
mesh::Packet* getNextOutbound(uint32_t now) override;
int getOutboundCount(uint32_t now) const override;
int getOutboundTotal() const override;
int getFreeCount() const override;
mesh::Packet* getOutboundByIdx(int i) override;
mesh::Packet* removeOutboundByIdx(int i) override;
+1 -1
View File
@@ -14,7 +14,7 @@ public:
board.getBattMilliVolts(),
ms.getMillis() / 1000,
err_flags,
mgr->getOutboundCount(0xFFFFFFFF)
mgr->getOutboundTotal()
);
}
+77 -76
View File
@@ -24,12 +24,12 @@
#endif
// Effective MQTT origin: empty mqtt_origin follows node_name; otherwise mqtt_origin override (quotes stripped).
static void applyEffectiveOrigin(const NodePrefs* prefs, char* dest, size_t dest_size) {
if (!prefs || !dest || dest_size == 0) return;
if (prefs->mqtt_origin[0] == '\0') {
strncpy(dest, prefs->node_name, dest_size - 1);
static void applyEffectiveOrigin(const NodePrefs* np, const MQTTPrefs* obs, char* dest, size_t dest_size) {
if (!np || !obs || !dest || dest_size == 0) return;
if (obs->mqtt_origin[0] == '\0') {
strncpy(dest, np->node_name, dest_size - 1);
} else {
strncpy(dest, prefs->mqtt_origin, dest_size - 1);
strncpy(dest, obs->mqtt_origin, dest_size - 1);
}
dest[dest_size - 1] = '\0';
StrHelper::stripSurroundingQuotes(dest, dest_size);
@@ -52,7 +52,7 @@ static bool ntpHostnameEquals(const char* a, const char* b) {
return strcasecmp(a, b) == 0;
}
static void fillNtpServerList(const NodePrefs* prefs, const char* servers[], int& count) {
static void fillNtpServerList(const MQTTPrefs* prefs, const char* servers[], int& count) {
count = 0;
if (prefs && prefs->mqtt_ntp_server[0] != '\0') {
servers[count++] = prefs->mqtt_ntp_server;
@@ -72,31 +72,31 @@ static void fillNtpServerList(const NodePrefs* prefs, const char* servers[], int
}
}
const char* MQTTBridge::effectiveNtpPrimary(const NodePrefs* prefs) {
if (prefs && prefs->mqtt_ntp_server[0] != '\0') {
return prefs->mqtt_ntp_server;
const char* MQTTBridge::effectiveNtpPrimary(const MQTTPrefs* obs) {
if (obs && obs->mqtt_ntp_server[0] != '\0') {
return obs->mqtt_ntp_server;
}
return kNtpBuiltinFallbacks[0];
}
void MQTTBridge::refreshOriginFromPrefs() {
if (!_prefs) return;
applyEffectiveOrigin(_prefs, _origin, sizeof(_origin));
applyEffectiveOrigin(_prefs, _obs, _origin, sizeof(_origin));
}
void MQTTBridge::getEffectiveMqttOrigin(const NodePrefs* prefs, char* buf, size_t buf_size) {
void MQTTBridge::getEffectiveMqttOrigin(const NodePrefs* np, const MQTTPrefs* obs, char* buf, size_t buf_size) {
if (!buf || buf_size == 0) return;
if (!prefs) {
if (!np || !obs) {
buf[0] = '\0';
return;
}
applyEffectiveOrigin(prefs, buf, buf_size);
applyEffectiveOrigin(np, obs, buf, buf_size);
}
// Helper function to check if WiFi credentials are valid
static bool isWiFiConfigValid(const NodePrefs* prefs) {
static bool isWiFiConfigValid(const MQTTPrefs* obs) {
// Check if WiFi SSID is configured (not empty)
if (strlen(prefs->wifi_ssid) == 0) {
if (!obs || strlen(obs->wifi_ssid) == 0) {
return false;
}
@@ -114,13 +114,13 @@ static bool customEndpointComplete(const char* host, uint16_t port) {
return host[0] != '\0' && (port != 0 || strstr(host, "://") != nullptr);
}
bool MQTTBridge::isConfigValid(const NodePrefs* prefs) {
if (!prefs || !isWiFiConfigValid(prefs)) return false;
bool MQTTBridge::isConfigValid(const MQTTPrefs* obs) {
if (!obs || !isWiFiConfigValid(obs)) return false;
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
const char* preset_name = prefs->mqtt_slot_preset[i];
const char* preset_name = obs->mqtt_slot_preset[i];
if (preset_name[0] == '\0' || strcmp(preset_name, MQTT_PRESET_NONE) == 0) continue;
if (strcmp(preset_name, MQTT_PRESET_CUSTOM) == 0) {
if (customEndpointComplete(prefs->mqtt_slot_host[i], prefs->mqtt_slot_port[i])) return true;
if (customEndpointComplete(obs->mqtt_slot_host[i], obs->mqtt_slot_port[i])) return true;
} else if (findMQTTPreset(preset_name) != nullptr) {
return true;
}
@@ -201,9 +201,9 @@ unsigned long MQTTBridge::getWifiConnectedAtMillis() {
return s_wifi_connected_at;
}
void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const NodePrefs* prefs) {
void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPrefs* obs) {
if (buf == nullptr || bufsize == 0) return;
const char* msgs = (prefs->mqtt_status_enabled) ? "on" : "off";
const char* msgs = (obs && obs->mqtt_status_enabled) ? "on" : "off";
if (s_mqtt_bridge_instance == nullptr || !s_mqtt_bridge_instance->_initialized) {
snprintf(buf, bufsize, "> msgs: %s (bridge not running)", msgs);
return;
@@ -391,11 +391,12 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index)
// ---------------------------------------------------------------------------
// Constructor
// ---------------------------------------------------------------------------
MQTTBridge::MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity)
MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity)
: BridgeBase(prefs, mgr, rtc),
_obs(obs),
_queue_count(0),
_last_status_publish(0), _last_status_retry(0), _status_interval(300000),
_ntp_client(_ntp_udp, effectiveNtpPrimary(prefs), 0, 60000), _last_ntp_sync(0), _ntp_synced(false), _ntp_sync_pending(false), _slots_setup_done(false), _max_active_slots(RUNTIME_MQTT_SLOTS),
_ntp_client(_ntp_udp, effectiveNtpPrimary(obs), 0, 60000), _last_ntp_sync(0), _ntp_synced(false), _ntp_sync_pending(false), _slots_setup_done(false), _max_active_slots(RUNTIME_MQTT_SLOTS),
_ntp_force_requested(false), _ntp_force_done(false), _ntp_force_result(false),
_ntp_diag_requested(false), _ntp_diag_done(false), _ntp_diag_count(0),
// Default to UTC; setRules() will be called from syncTimeWithNTP when a
@@ -546,14 +547,14 @@ void MQTTBridge::begin() {
MQTT_DEBUG_PRINTLN("Max active slots: %d", _max_active_slots);
// Check if WiFi credentials are configured first
if (!isWiFiConfigValid(_prefs)) {
if (!isWiFiConfigValid(_obs)) {
MQTT_DEBUG_PRINTLN("MQTT Bridge initialization skipped - WiFi credentials not configured");
return;
}
refreshOriginFromPrefs();
strncpy(_iata, _prefs->mqtt_iata, sizeof(_iata) - 1);
strncpy(_iata, _obs->mqtt_iata, sizeof(_iata) - 1);
_iata[sizeof(_iata) - 1] = '\0';
StrHelper::stripSurroundingQuotes(_iata, sizeof(_iata));
@@ -564,17 +565,17 @@ void MQTTBridge::begin() {
}
// Update enabled flags from preferences
_status_enabled = _prefs->mqtt_status_enabled;
_packets_enabled = _prefs->mqtt_packets_enabled;
_raw_enabled = _prefs->mqtt_raw_enabled;
_rx_enabled = _prefs->mqtt_rx_enabled;
_tx_mode = _prefs->mqtt_tx_enabled; // 0=off, 1=all, 2=advert
_status_enabled = _obs->mqtt_status_enabled;
_packets_enabled = _obs->mqtt_packets_enabled;
_raw_enabled = _obs->mqtt_raw_enabled;
_rx_enabled = _obs->mqtt_rx_enabled;
_tx_mode = _obs->mqtt_tx_enabled; // 0=off, 1=all, 2=advert
// Set status interval to 5 minutes (300000 ms), or use preference if set and valid
if (_prefs->mqtt_status_interval >= 1000 && _prefs->mqtt_status_interval <= 3600000) {
_status_interval = _prefs->mqtt_status_interval;
if (_obs->mqtt_status_interval >= 1000 && _obs->mqtt_status_interval <= 3600000) {
_status_interval = _obs->mqtt_status_interval;
} else {
// Invalid or uninitialized value - fix it in preferences and use default
_prefs->mqtt_status_interval = 300000; // Fix the preference value
_obs->mqtt_status_interval = 300000; // Fix the preference value
_status_interval = 300000; // 5 minutes default
}
@@ -585,12 +586,12 @@ void MQTTBridge::begin() {
// Apply slot presets from preferences
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
const char* preset_name = _prefs->mqtt_slot_preset[i];
const char* preset_name = _obs->mqtt_slot_preset[i];
if (preset_name[0] != '\0' && strcmp(preset_name, MQTT_PRESET_NONE) != 0) {
if (strcmp(preset_name, MQTT_PRESET_CUSTOM) == 0) {
// Custom broker: copy host/port/username/password from prefs
_slots[i].preset = nullptr;
strncpy(_slots[i].host, _prefs->mqtt_slot_host[i], sizeof(_slots[i].host) - 1);
strncpy(_slots[i].host, _obs->mqtt_slot_host[i], sizeof(_slots[i].host) - 1);
_slots[i].host[sizeof(_slots[i].host) - 1] = '\0';
if (strlen(_slots[i].host) == 0) {
MQTT_DEBUG_PRINTLN("MQTT%d: custom preset has no server configured, disabling", i + 1);
@@ -598,12 +599,12 @@ void MQTTBridge::begin() {
continue;
}
_slots[i].enabled = true;
_slots[i].port = _prefs->mqtt_slot_port[i];
strncpy(_slots[i].username, _prefs->mqtt_slot_username[i], sizeof(_slots[i].username) - 1);
_slots[i].port = _obs->mqtt_slot_port[i];
strncpy(_slots[i].username, _obs->mqtt_slot_username[i], sizeof(_slots[i].username) - 1);
_slots[i].username[sizeof(_slots[i].username) - 1] = '\0';
strncpy(_slots[i].password, _prefs->mqtt_slot_password[i], sizeof(_slots[i].password) - 1);
strncpy(_slots[i].password, _obs->mqtt_slot_password[i], sizeof(_slots[i].password) - 1);
_slots[i].password[sizeof(_slots[i].password) - 1] = '\0';
strncpy(_slots[i].audience, _prefs->mqtt_slot_audience[i], sizeof(_slots[i].audience) - 1);
strncpy(_slots[i].audience, _obs->mqtt_slot_audience[i], sizeof(_slots[i].audience) - 1);
_slots[i].audience[sizeof(_slots[i].audience) - 1] = '\0';
} else {
const MQTTPresetDef* preset = findMQTTPreset(preset_name);
@@ -611,9 +612,9 @@ void MQTTBridge::begin() {
_slots[i].enabled = true;
_slots[i].preset = preset;
if (mqttPresetNeedsSlotCredentials(preset)) {
strncpy(_slots[i].username, _prefs->mqtt_slot_username[i], sizeof(_slots[i].username) - 1);
strncpy(_slots[i].username, _obs->mqtt_slot_username[i], sizeof(_slots[i].username) - 1);
_slots[i].username[sizeof(_slots[i].username) - 1] = '\0';
strncpy(_slots[i].password, _prefs->mqtt_slot_password[i], sizeof(_slots[i].password) - 1);
strncpy(_slots[i].password, _obs->mqtt_slot_password[i], sizeof(_slots[i].password) - 1);
_slots[i].password[sizeof(_slots[i].password) - 1] = '\0';
}
} else {
@@ -709,7 +710,7 @@ void MQTTBridge::begin() {
WiFi.mode(WIFI_STA);
WiFi.setAutoReconnect(true);
WiFi.setAutoConnect(true);
WiFi.begin(_prefs->wifi_ssid, _prefs->wifi_password);
WiFi.begin(_obs->wifi_ssid, _obs->wifi_password);
// NOTE: Slot setup deferred until after NTP sync in loop()
#endif
@@ -852,7 +853,7 @@ void MQTTBridge::initializeWiFiInTask() {
// When already connected, the deferred slot setup still fires in mqttTaskLoop()
// because _ntp_synced persists across end() (only _slots_setup_done is reset).
if (WiFi.status() != WL_CONNECTED) {
WiFi.begin(_prefs->wifi_ssid, _prefs->wifi_password);
WiFi.begin(_obs->wifi_ssid, _obs->wifi_password);
} else if (!_ntp_synced && !_ntp_sync_pending) {
_ntp_sync_pending = true; // already connected but never synced — kick NTP now
}
@@ -983,8 +984,8 @@ void MQTTBridge::mqttTaskLoop() {
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
if (_slot_reconfigure_pending[i]) {
_slot_reconfigure_pending[i] = false;
MQTT_DEBUG_PRINTLN("Applying deferred reconfigure for MQTT%d (preset: %s)", i + 1, _prefs->mqtt_slot_preset[i]);
applySlotPreset(i, _prefs->mqtt_slot_preset[i]);
MQTT_DEBUG_PRINTLN("Applying deferred reconfigure for MQTT%d (preset: %s)", i + 1, _obs->mqtt_slot_preset[i]);
applySlotPreset(i, _obs->mqtt_slot_preset[i]);
}
}
@@ -997,9 +998,9 @@ void MQTTBridge::mqttTaskLoop() {
#ifdef WITH_SNMP
// SNMP agent loop — process incoming UDP requests
if (_snmp_agent) {
if (!_snmp_agent->isRunning() && WiFi.isConnected() && _prefs->snmp_enabled) {
_snmp_agent->begin(_prefs->snmp_community);
MQTT_DEBUG_PRINTLN("SNMP agent started on port 161 (community: %s)", _prefs->snmp_community);
if (!_snmp_agent->isRunning() && WiFi.isConnected() && _obs->snmp_enabled) {
_snmp_agent->begin(_obs->snmp_community);
MQTT_DEBUG_PRINTLN("SNMP agent started on port 161 (community: %s)", _obs->snmp_community);
}
if (_snmp_agent->isRunning()) {
// Update MQTT stats from this core
@@ -1633,8 +1634,8 @@ bool MQTTBridge::createSlotAuthToken(int index) {
// Prepare owner key
const char* owner_key = nullptr;
char owner_key_uppercase[65];
if (_prefs->mqtt_owner_public_key[0] != '\0') {
strncpy(owner_key_uppercase, _prefs->mqtt_owner_public_key, sizeof(owner_key_uppercase) - 1);
if (_obs->mqtt_owner_public_key[0] != '\0') {
strncpy(owner_key_uppercase, _obs->mqtt_owner_public_key, sizeof(owner_key_uppercase) - 1);
owner_key_uppercase[sizeof(owner_key_uppercase) - 1] = '\0';
for (int i = 0; owner_key_uppercase[i]; i++) {
owner_key_uppercase[i] = toupper(owner_key_uppercase[i]);
@@ -1644,7 +1645,7 @@ bool MQTTBridge::createSlotAuthToken(int index) {
char client_version[64];
getClientVersion(client_version, sizeof(client_version));
const char* email = (_prefs->mqtt_email[0] != '\0') ? _prefs->mqtt_email : nullptr;
const char* email = (_obs->mqtt_email[0] != '\0') ? _obs->mqtt_email : nullptr;
unsigned long current_time = time(nullptr);
// Stagger token expiry per slot to avoid simultaneous renewal/reconnect
@@ -1719,7 +1720,7 @@ bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, bool
// ---------------------------------------------------------------------------
bool MQTTBridge::substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size) {
const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw";
const char* token = _prefs->mqtt_slot_token[slot_index];
const char* token = _obs->mqtt_slot_token[slot_index];
size_t out = 0;
const char* p = tmpl;
@@ -1769,7 +1770,7 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
if (slot.preset->topic_style == MQTT_TOPIC_MESHRANK) {
// MeshRank: packets only, uses per-slot token in topic path
if (type != MSG_PACKETS) return false;
const char* token = _prefs->mqtt_slot_token[index];
const char* token = _obs->mqtt_slot_token[index];
if (!token || token[0] == '\0') return false;
snprintf(topic_buf, buf_size, "meshrank/uplink/%s/%s/packets", token, _device_id);
return true;
@@ -1782,8 +1783,8 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
}
// Custom slots: use topic template if set, otherwise default meshcore format
if (_prefs->mqtt_slot_topic[index][0] != '\0') {
return substituteTopicTemplate(_prefs->mqtt_slot_topic[index], type, index, topic_buf, buf_size);
if (_obs->mqtt_slot_topic[index][0] != '\0') {
return substituteTopicTemplate(_obs->mqtt_slot_topic[index], type, index, topic_buf, buf_size);
}
// Default: meshcore format
if (!isIATAValid()) return false;
@@ -1943,9 +1944,9 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) {
slot.enabled = true;
slot.preset = preset;
if (mqttPresetNeedsSlotCredentials(preset)) {
strncpy(slot.username, _prefs->mqtt_slot_username[slot_index], sizeof(slot.username) - 1);
strncpy(slot.username, _obs->mqtt_slot_username[slot_index], sizeof(slot.username) - 1);
slot.username[sizeof(slot.username) - 1] = '\0';
strncpy(slot.password, _prefs->mqtt_slot_password[slot_index], sizeof(slot.password) - 1);
strncpy(slot.password, _obs->mqtt_slot_password[slot_index], sizeof(slot.password) - 1);
slot.password[sizeof(slot.password) - 1] = '\0';
}
if (_initialized) {
@@ -1979,7 +1980,7 @@ void MQTTBridge::setSlotCustomBroker(int slot_index, const char* host, uint16_t
void MQTTBridge::checkConfigurationMismatch() {
// Warn if packets are enabled but both rx and tx are off — nothing will be published
if (_prefs->mqtt_packets_enabled && !_prefs->mqtt_rx_enabled && _prefs->mqtt_tx_enabled == 0) {
if (_obs->mqtt_packets_enabled && !_obs->mqtt_rx_enabled && _obs->mqtt_tx_enabled == 0) {
unsigned long now = millis();
if (_last_config_warning == 0 || (now - _last_config_warning > CONFIG_WARNING_INTERVAL)) {
MQTT_DEBUG_PRINTLN("MQTT: Both mqtt.rx and mqtt.tx are off — no packets will be published. Run 'set mqtt.rx on' or 'set mqtt.tx on' to fix.");
@@ -2017,7 +2018,7 @@ bool MQTTBridge::handleWiFiConnection(unsigned long now) {
_wifi_reconnect_backoff_attempt = 0;
#ifdef ESP_PLATFORM
wifi_ps_type_t ps_mode;
uint8_t ps_pref = _prefs->wifi_power_save;
uint8_t ps_pref = _obs->wifi_power_save;
if (ps_pref == 1) {
ps_mode = WIFI_PS_NONE;
} else if (ps_pref == 2) {
@@ -2061,7 +2062,7 @@ bool MQTTBridge::handleWiFiConnection(unsigned long now) {
_wifi_reconnect_backoff_attempt++;
}
WiFi.disconnect();
WiFi.begin(_prefs->wifi_ssid, _prefs->wifi_password);
WiFi.begin(_obs->wifi_ssid, _obs->wifi_password);
}
}
_last_wifi_status = current_wifi_status;
@@ -2070,7 +2071,7 @@ bool MQTTBridge::handleWiFiConnection(unsigned long now) {
}
bool MQTTBridge::isReady() const {
return _initialized && isWiFiConfigValid(_prefs);
return _initialized && isWiFiConfigValid(_obs);
}
bool MQTTBridge::isIATAValid() const {
@@ -2088,7 +2089,7 @@ bool MQTTBridge::isSlotReady(int index, char* reason_buf, size_t reason_size) co
if (slot.preset) {
if (slot.preset->topic_style == MQTT_TOPIC_MESHRANK) {
if (_prefs->mqtt_slot_token[index][0] == '\0') {
if (_obs->mqtt_slot_token[index][0] == '\0') {
if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt%d.token <your_token>", index + 1);
return false;
}
@@ -2099,18 +2100,18 @@ bool MQTTBridge::isSlotReady(int index, char* reason_buf, size_t reason_size) co
}
}
if (mqttPresetNeedsSlotCredentials(slot.preset)) {
if (_prefs->mqtt_slot_username[index][0] == '\0') {
if (_obs->mqtt_slot_username[index][0] == '\0') {
if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt%d.username <user>", index + 1);
return false;
}
if (_prefs->mqtt_slot_password[index][0] == '\0') {
if (_obs->mqtt_slot_password[index][0] == '\0') {
if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt%d.password <pass>", index + 1);
return false;
}
}
} else {
// Custom slot without a topic template uses meshcore format, needs IATA
if (_prefs->mqtt_slot_topic[index][0] == '\0' && !isIATAValid()) {
if (_obs->mqtt_slot_topic[index][0] == '\0' && !isIATAValid()) {
if (reason_buf) snprintf(reason_buf, reason_size, "set mqtt.iata <airport_code> or set mqtt%d.topic <template>", index + 1);
return false;
}
@@ -2161,7 +2162,7 @@ void MQTTBridge::loop() {
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
if (_slot_reconfigure_pending[i]) {
_slot_reconfigure_pending[i] = false;
applySlotPreset(i, _prefs->mqtt_slot_preset[i]);
applySlotPreset(i, _obs->mqtt_slot_preset[i]);
}
}
@@ -2244,7 +2245,7 @@ void MQTTBridge::loop() {
// ---------------------------------------------------------------------------
void MQTTBridge::onPacketReceived(mesh::Packet *packet) {
if (!_initialized || !_prefs->mqtt_packets_enabled || !_prefs->mqtt_rx_enabled) return;
if (!_initialized || !_obs->mqtt_packets_enabled || !_obs->mqtt_rx_enabled) return;
// Check if we have any enabled slots to send to
bool has_valid_slots = false;
@@ -2261,8 +2262,8 @@ void MQTTBridge::onPacketReceived(mesh::Packet *packet) {
}
void MQTTBridge::sendPacket(mesh::Packet *packet) {
uint8_t tx_mode = _prefs->mqtt_tx_enabled; // Read live from prefs (no restart needed)
if (!_initialized || !_prefs->mqtt_packets_enabled || tx_mode == 0) return;
uint8_t tx_mode = _obs->mqtt_tx_enabled; // Read live from prefs (no restart needed)
if (!_initialized || !_obs->mqtt_packets_enabled || tx_mode == 0) return;
// Advert mode: only queue self-originated advert packets
if (tx_mode == 2) {
@@ -2917,7 +2918,7 @@ void MQTTBridge::refreshNTP() {
// Lightweight periodic refresh: just restart SNTP which runs async in the background.
// No blocking DNS, no UDP sockets, no retry loops on the MQTT task loop.
// The heavy syncTimeWithNTP() is only used for initial sync and WiFi reconnect recovery.
configTime(0, 0, effectiveNtpPrimary(_prefs));
configTime(0, 0, effectiveNtpPrimary(_obs));
_last_ntp_sync = millis();
MQTT_DEBUG_PRINTLN("NTP refresh triggered (async SNTP)");
}
@@ -2946,10 +2947,10 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) {
if (primary_only) {
// Validation path (e.g. set mqtt.ntp): test only the configured primary so a
// typo fails fast instead of walking the entire fallback list.
servers[0] = effectiveNtpPrimary(_prefs);
servers[0] = effectiveNtpPrimary(_obs);
server_count = 1;
} else {
fillNtpServerList(_prefs, servers, server_count);
fillNtpServerList(_obs, servers, server_count);
}
bool ntp_ok = false;
@@ -3048,15 +3049,15 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) {
// Reuses the inline _timezone_storage via setRules() instead of
// deleting/newing a Timezone, which was a per-change heap alloc pair.
static char last_timezone[64] = "";
if (strcmp(_prefs->timezone_string, last_timezone) != 0) {
if (strcmp(_obs->timezone_string, last_timezone) != 0) {
TimeChangeRule dst_rule, std_rule;
if (!timezoneRulesFromString(_prefs->timezone_string, dst_rule, std_rule)) {
if (!timezoneRulesFromString(_obs->timezone_string, dst_rule, std_rule)) {
TimeChangeRule utc = {"UTC", Last, Sun, Mar, 0, 0};
dst_rule = utc;
std_rule = utc;
}
_timezone_storage.setRules(dst_rule, std_rule);
strncpy(last_timezone, _prefs->timezone_string, sizeof(last_timezone) - 1);
strncpy(last_timezone, _obs->timezone_string, sizeof(last_timezone) - 1);
last_timezone[sizeof(last_timezone) - 1] = '\0';
}
@@ -3096,7 +3097,7 @@ bool MQTTBridge::requestForcedNtpSync(uint32_t timeout_ms) {
void MQTTBridge::runNtpDiagProbe() {
const char* servers[kMaxNtpServers];
int count = 0;
fillNtpServerList(_prefs, servers, count);
fillNtpServerList(_obs, servers, count);
_ntp_client.begin();
for (int i = 0; i < count; i++) {
+9 -5
View File
@@ -375,8 +375,12 @@ private:
void logMemoryStatus();
void refreshOriginFromPrefs();
// Observer config (MQTT/WiFi/timezone/SNMP/alert), persisted to /mqtt_prefs.
// _prefs (held by BridgeBase) still provides upstream fields (freq/sf/node_name…).
MQTTPrefs* _obs = nullptr;
public:
MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity);
MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity);
void begin() override;
void end() override;
@@ -439,8 +443,8 @@ public:
const char* getSlotPresetName(int slot_index) const;
static int getRuntimeSlotCount() { return RUNTIME_MQTT_SLOTS; }
/** Resolved origin for MQTT JSON: node_name when mqtt_origin is empty, else mqtt_origin (with quote stripping). */
static void getEffectiveMqttOrigin(const NodePrefs* prefs, char* buf, size_t buf_size);
static const char* effectiveNtpPrimary(const NodePrefs* prefs);
static void getEffectiveMqttOrigin(const NodePrefs* np, const MQTTPrefs* obs, char* buf, size_t buf_size);
static const char* effectiveNtpPrimary(const MQTTPrefs* obs);
/** Sync system clock via NTP. force=true bypasses the 5s post-sync rate limit.
* primary_only=true tests just the effective primary server (no fallback walk) so a
* mistyped hostname fails fast instead of blocking through the whole fallback list.
@@ -459,9 +463,9 @@ public:
* summary in reply; verbose=false fills reply with a compact "<server> ok|fail" list
* (for LoRa). Returns false if the bridge is not running. */
bool ntpDiag(char* reply, size_t reply_size, bool verbose);
static void formatMqttStatusReply(char* buf, size_t bufsize, const NodePrefs* prefs);
static void formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPrefs* obs);
/** True when WiFi is set and at least one MQTT slot can run (preset + custom host if needed). */
static bool isConfigValid(const NodePrefs* prefs);
static bool isConfigValid(const MQTTPrefs* obs);
static void formatSlotDiagReply(char* buf, size_t bufsize, int slot_index);
static uint8_t getLastWifiDisconnectReason();
static unsigned long getLastWifiDisconnectTime();
+19 -3
View File
@@ -180,10 +180,26 @@ void RadioLibWrapper::onSendFinished() {
state = STATE_IDLE;
}
int16_t RadioLibWrapper::performChannelScan() {
return _radio->scanChannel();
}
bool RadioLibWrapper::isChannelActive() {
return _threshold == 0
? false // interference check is disabled
: getCurrentRSSI() > _noise_floor + _threshold;
// int.thresh: RSSI-based interference detection (relative to noise floor)
if (_threshold != 0 && getCurrentRSSI() > _noise_floor + _threshold) return true;
// cad: hardware channel activity detection
if (_cad_enabled) {
int16_t result = performChannelScan();
// scanChannel() triggers DIO interrupt (CAD done) which sets STATE_INT_READY
// via setFlag() ISR. Clear it before restarting RX so recvRaw() doesn't
// try to read a non-existent packet and count a spurious recv error.
state = STATE_IDLE;
startRecv();
if (result != RADIOLIB_CHANNEL_FREE) return true;
}
return false;
}
uint8_t RadioLibWrapper::getRadioState() const {
+4
View File
@@ -14,6 +14,7 @@ protected:
unsigned long last_recv_millis;
unsigned long last_radio_interrupt_millis; // updated on any ISR event, even CRC errors
uint8_t _preamble_sf;
bool _cad_enabled;
void idle() override;
void startRecv() override;
@@ -26,6 +27,7 @@ public:
n_recv = n_sent = n_recv_errors = 0;
last_recv_millis = 0;
last_radio_interrupt_millis = 0;
_cad_enabled = false;
}
void begin() override;
@@ -52,9 +54,11 @@ public:
virtual uint8_t getSpreadingFactor() const { return LORA_SF; }
static uint16_t preambleLengthForSF(uint8_t sf) { return sf <= 8 ? 32 : 16; }
void updatePreamble(uint8_t sf) { _preamble_sf = sf; _radio->setPreambleLength(preambleLengthForSF(sf)); }
virtual int16_t performChannelScan();
int getNoiseFloor() const override { return _noise_floor; }
void triggerNoiseFloorCalibrate(int threshold) override;
void setCADEnabled(bool enable) override { _cad_enabled = enable; }
void resetAGC() override;
void loop() override;
+2 -1
View File
@@ -12,8 +12,9 @@ class LoRaFEMControl
void setRxModeEnable(void);
void setRxModeEnableWhenMCUSleep(void);
void setLNAEnable(bool enabled);
bool isLnaCanControl(void) { return lna_can_control; }
bool isLnaCanControl(void) const { return lna_can_control; }
void setLnaCanControl(bool can_control) { lna_can_control = can_control; }
bool isLNAEnabled(void) const { return lna_enabled; }
private:
bool lna_enabled = false;
+18
View File
@@ -123,4 +123,22 @@ void T096Board::powerOff() {
const char* T096Board::getManufacturerName() const {
return "Heltec T096";
}
bool T096Board::setLoRaFemLnaEnabled(bool enable) {
if (!loRaFEMControl.isLnaCanControl()) {
return false;
}
loRaFEMControl.setLNAEnable(enable);
loRaFEMControl.setRxModeEnable();
return true;
}
bool T096Board::canControlLoRaFemLna() const {
return loRaFEMControl.isLnaCanControl();
}
bool T096Board::isLoRaFemLnaEnabled() const {
return loRaFEMControl.isLNAEnabled();
}
+3
View File
@@ -25,4 +25,7 @@ public:
uint16_t getBattMilliVolts() override;
const char* getManufacturerName() const override ;
void powerOff() override;
bool setLoRaFemLnaEnabled(bool enable) override;
bool canControlLoRaFemLna() const override;
bool isLoRaFemLnaEnabled() const override;
};
@@ -82,3 +82,21 @@ void HeltecTrackerV2Board::begin() {
const char* HeltecTrackerV2Board::getManufacturerName() const {
return "Heltec Tracker V2";
}
bool HeltecTrackerV2Board::setLoRaFemLnaEnabled(bool enable) {
if (!loRaFEMControl.isLnaCanControl()) {
return false;
}
loRaFEMControl.setLNAEnable(enable);
loRaFEMControl.setRxModeEnable();
return true;
}
bool HeltecTrackerV2Board::canControlLoRaFemLna() const {
return loRaFEMControl.isLnaCanControl();
}
bool HeltecTrackerV2Board::isLoRaFemLnaEnabled() const {
return loRaFEMControl.isLNAEnabled();
}
@@ -21,5 +21,8 @@ public:
void powerOff() override;
uint16_t getBattMilliVolts() override;
const char* getManufacturerName() const override ;
bool setLoRaFemLnaEnabled(bool enable) override;
bool canControlLoRaFemLna() const override;
bool isLoRaFemLnaEnabled() const override;
};
+2 -1
View File
@@ -12,8 +12,9 @@ class LoRaFEMControl
void setRxModeEnable(void);
void setRxModeEnableWhenMCUSleep(void);
void setLNAEnable(bool enabled);
bool isLnaCanControl(void) { return lna_can_control; }
bool isLnaCanControl(void) const { return lna_can_control; }
void setLnaCanControl(bool can_control) { lna_can_control = can_control; }
bool isLNAEnabled(void) const { return lna_enabled; }
private:
bool lna_enabled = false;
+17
View File
@@ -83,3 +83,20 @@ void HeltecV4Board::begin() {
return loRaFEMControl.getFEMType() == KCT8103L_PA ? "Heltec V4.3 OLED" : "Heltec V4 OLED";
#endif
}
bool HeltecV4Board::setLoRaFemLnaEnabled(bool enable) {
if (!loRaFEMControl.isLnaCanControl()) {
return false;
}
loRaFEMControl.setLNAEnable(enable);
loRaFEMControl.setRxModeEnable();
return true;
}
bool HeltecV4Board::canControlLoRaFemLna() const {
return loRaFEMControl.isLnaCanControl();
}
bool HeltecV4Board::isLoRaFemLnaEnabled() const {
return loRaFEMControl.isLNAEnabled();
}
+4
View File
@@ -36,4 +36,8 @@ public:
}
float getAdcMultiplier() const override { return adc_mult; }
const char* getManufacturerName() const override;
bool setLoRaFemLnaEnabled(bool enable) override;
bool canControlLoRaFemLna() const override;
bool isLoRaFemLnaEnabled() const override;
};
+2 -1
View File
@@ -18,7 +18,8 @@ class LoRaFEMControl
void setRxModeEnable(void);
void setRxModeEnableWhenMCUSleep(void);
void setLNAEnable(bool enabled);
bool isLnaCanControl(void) { return lna_can_control; }
bool isLnaCanControl(void) const { return lna_can_control; }
bool isLNAEnabled(void) const { return lna_enabled; }
void setLnaCanControl(bool can_control) { lna_can_control = can_control; }
LoRaFEMType getFEMType(void) const { return fem_type; }
private: