feat(mqtt): send all message types to MeshRank and harden neighbors flow
MeshRank slots now publish every message type (status/packets/raw/neighbors)
under meshrank/uplink/{token}/{device}/, matching the standard MeshCore topic
layout, instead of packets only. The four duplicated topic-suffix ternaries are
folded into messageTypeSuffix().
Fix the discover.neighbors -> discover.scopes race: a scopes request issued
while a zero-hop discovery is still collecting responses (from discover.neighbors
or the periodic timer) now queues behind that 60-second window and runs against
the refreshed table, reporting the wait in its reply. A queued one-shot request
is no longer cancelled by `mqtt.neighbors off`; only the periodic timer's own
refresh is. Precondition checks are shared via neighborDiscoverReady() so queuing
cannot report OK for a request that will fail once the window closes.
Add the periodic-neighbors schedule to `get mqtt.status` as a trailing
nbr:<next>/<last> field (time to next publish + last result). The mesh owns the
timer and reports a summary to the bridge across the core boundary via atomics.
Also clamp the running offset in formatMqttStatusReply: snprintf returns the
untruncated length, so a full 6-slot line could push the offset past the buffer
and underflow the remaining size on the next append. Reachable today with real
preset-name combinations at the 160-byte reply size.
This commit is contained in:
1 parent
df13d181b2
commit
b3bd5ecd3c
7 files changed
+185
-43
No files matched your search
@@ -296,7 +296,7 @@ set mqtt3.preset meshrank
|
||||
set mqtt3.token FE1B34242C5938C39225310081FD6718
|
||||
```
|
||||
|
||||
The token is generated on the MeshRank website and is tied to your account. MeshRank only receives packet data (no status or raw messages).
|
||||
The token is generated on the MeshRank website and is tied to your account. MeshRank receives the same message types as any other preset (`status`, `packets`, `raw`, `neighbors`), each under `meshrank/uplink/{token}/{device_id}/`, subject to the usual `mqtt.status`/`mqtt.raw`/`mqtt.neighbors` toggles.
|
||||
|
||||
#### Example: Configure MeshMapper on Slot 3
|
||||
```bash
|
||||
@@ -376,7 +376,7 @@ These settings apply across all MQTT slots:
|
||||
- `get mqtt.iata` - Get IATA code
|
||||
- `get mqtt.presets` - List available MQTT presets (paginated, comma-separated)
|
||||
- `get mqtt.presets <start>` - Continue list from index shown in `... next:<idx>`
|
||||
- `get mqtt.status` - Get MQTT status summary (connection info per slot)
|
||||
- `get mqtt.status` - Get MQTT status summary (connection info per slot, plus the periodic neighbors schedule when `mqtt.neighbors` is on)
|
||||
- `get mqtt.packets` - Get packet message setting (on/off)
|
||||
- `get mqtt.raw` - Get raw message setting (on/off)
|
||||
- `get mqtt.rx` - Get RX packet uplinking setting (on/off)
|
||||
@@ -509,11 +509,15 @@ Full packet data with RF characteristics and metadata.
|
||||
Minimal raw packet data for map integration.
|
||||
|
||||
### Neighbors Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/neighbors`
|
||||
Cached zero-hop repeater neighbors with SNR, last-heard age, and flood-allowed scopes (retained, QoS 1). Published on `discover.scopes` or periodically when `mqtt.neighbors` is enabled (PSRAM observer builds only).
|
||||
Cached zero-hop repeater neighbors with SNR, last-heard age, and flood-allowed scopes. Published on `discover.scopes` or periodically when `mqtt.neighbors` is enabled (PSRAM observer builds only). Goes to every configured slot's `neighbors` topic at QoS 1, retained only where the preset allows it.
|
||||
|
||||
Periodic publishing first runs a 60-second zero-hop neighbor refresh equivalent to `discover.neighbors`, then queries the refreshed table for scopes and publishes when the scope-query phase completes. Manual `discover.scopes` remains a one-shot query of the current cache.
|
||||
Periodic publishing first runs a 60-second zero-hop neighbor refresh equivalent to `discover.neighbors`, then queries the refreshed table for scopes and publishes when the scope-query phase completes.
|
||||
|
||||
**Note**: `{DEVICE_PUBLIC_KEY}` is the device's public key in hexadecimal format (64 characters).
|
||||
Manual `discover.scopes` normally queries the current cache in one shot. If a `discover.neighbors` refresh is already collecting responses — whether started from the CLI or by the periodic timer — the scope queries are queued behind its 60-second window instead, so they run against the refreshed table. The reply reports the wait, e.g. `OK - scopes queued (47s discovery remaining)`. A queued one-shot request survives `set mqtt.neighbors off`; only the periodic timer's own refresh is cancelled by it.
|
||||
|
||||
While `mqtt.neighbors` is on, `get mqtt.status` appends `nbr: <next>/<last>` — time to the next automatic publish (`3h12m`, `12m`, `45s`, or `active`/`due`) and the last publish result (`ok`, `failed`, or `none`).
|
||||
|
||||
**Note**: `{DEVICE_PUBLIC_KEY}` is the device's public key in hexadecimal format (64 characters). MeshRank slots use `meshrank/uplink/{token}/{DEVICE_PUBLIC_KEY}/neighbors` instead.
|
||||
|
||||
## JSON Message Formats
|
||||
|
||||
|
||||
@@ -118,7 +118,9 @@ This document provides an overview of CLI commands that can be sent to MeshCore
|
||||
**Usage:**
|
||||
- `discover.scopes`
|
||||
|
||||
Queries cached zero-hop repeater neighbors for flood-allowed region scopes (`ANON_REQ_TYPE_REGIONS`), then publishes a retained JSON snapshot to `meshcore/{IATA}/{device}/neighbors`. Includes this node's own scopes in the `self` object.
|
||||
Queries cached zero-hop repeater neighbors for flood-allowed region scopes (`ANON_REQ_TYPE_REGIONS`), then publishes a JSON snapshot to each configured MQTT slot's `neighbors` topic. Includes this node's own scopes in the `self` object.
|
||||
|
||||
If a zero-hop neighbor discovery is already collecting responses — from `discover.neighbors` or from the periodic `mqtt.neighbors` refresh — the scope queries are queued until its 60-second response window ends, so they run against the refreshed table rather than the stale cache. The reply reports the wait, e.g. `OK - scopes queued (47s discovery remaining)`. A queued request is not cancelled by `set mqtt.neighbors off`.
|
||||
|
||||
**Note:** Requires `WITH_MQTT_BRIDGE`, `MAX_NEIGHBOURS`, and PSRAM (`BOARD_HAS_PSRAM`). Returns `Err - not supported (requires PSRAM)` on other builds.
|
||||
|
||||
@@ -1125,10 +1127,15 @@ Requires WiFi connected and the MQTT bridge running.
|
||||
**Parameters:**
|
||||
- `hours`: Interval between automatic `discover.scopes` runs (**12-336**, default **24**)
|
||||
|
||||
**Note:** Requires PSRAM. Publishes to `meshcore/{IATA}/{device}/neighbors` (retained, QoS 1). Use `discover.scopes` for a one-shot publish.
|
||||
**Note:** Requires PSRAM. Publishes to each configured MQTT slot's `neighbors` topic at QoS 1 (retained only when that preset allows it). Use `discover.scopes` for a one-shot publish.
|
||||
|
||||
Automatic publishes run in two stages: first the repeater performs the same 60-second zero-hop refresh as `discover.neighbors`, then it queries the refreshed neighbors for scopes and publishes after those queries complete.
|
||||
|
||||
While enabled, `get mqtt.status` reports the schedule as a trailing `nbr: <next>/<last>` field, e.g. `nbr: 3h12m/ok`:
|
||||
|
||||
- `<next>` — time until the next automatic publish (`3h12m`, `12m`, `45s`), or `active` while a refresh or scope query is in flight, or `due` when a publish is waiting on the bridge or WiFi to come up.
|
||||
- `<last>` — result of the most recent publish attempt: `ok`, `failed`, or `none` if none has been attempted since boot.
|
||||
|
||||
---
|
||||
|
||||
#### View or change the speed of the bridge (RS-232 only)
|
||||
|
||||
@@ -1036,6 +1036,7 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
|
||||
neighbor_discover_count = 0;
|
||||
neighbor_discover_active = false;
|
||||
neighbor_table_refresh_active = false;
|
||||
neighbor_table_refresh_periodic = false;
|
||||
neighbor_discover_until = 0;
|
||||
next_neighbors_publish = 0;
|
||||
self_scopes_buf[0] = 0;
|
||||
@@ -1451,6 +1452,25 @@ void MyMesh::handleCommand(uint32_t sender_timestamp, char *command, char *reply
|
||||
while (*sub == ' ') sub++;
|
||||
if (*sub != 0) {
|
||||
strcpy(reply, "Err - discover.scopes has no options");
|
||||
} else if (pending_discover_tag != 0 &&
|
||||
!millisHasNowPassed(pending_discover_until) &&
|
||||
!neighbor_discover_active) {
|
||||
// Reuse an in-flight zero-hop discovery rather than querying the old
|
||||
// neighbor table. An explicit command takes ownership of a periodic
|
||||
// refresh so `mqtt.neighbors off` cannot cancel this one-shot request.
|
||||
// Validate up front so queuing cannot report OK for a request that is
|
||||
// certain to fail once the discovery window closes.
|
||||
if (!neighborDiscoverReady(reply)) {
|
||||
// reply already set
|
||||
} else {
|
||||
neighbor_table_refresh_active = true;
|
||||
neighbor_table_refresh_periodic = false;
|
||||
long remaining_ms = (long)(pending_discover_until - futureMillis(0));
|
||||
unsigned remaining_secs = remaining_ms > 0
|
||||
? (unsigned)(((unsigned long)remaining_ms + 999UL) / 1000UL) : 0;
|
||||
sprintf(reply, "OK - scopes queued (%us discovery remaining)", remaining_secs);
|
||||
MESH_DEBUG_PRINTLN("Neighbor scopes queued behind active discovery (%us remaining)", remaining_secs);
|
||||
}
|
||||
} else if (!startNeighborDiscover(reply)) {
|
||||
// reply already set
|
||||
}
|
||||
@@ -1535,25 +1555,33 @@ void MyMesh::loop() {
|
||||
if (neighbor_discover_active) {
|
||||
loopNeighborDiscover();
|
||||
} else if (neighbor_table_refresh_active) {
|
||||
if (!periodic_neighbors_enabled) {
|
||||
if (neighbor_table_refresh_periodic && !periodic_neighbors_enabled) {
|
||||
// Turning periodic publishing off cancels a refresh that has not yet
|
||||
// reached the scope-query/publish phase. Re-enabling starts fresh.
|
||||
// pending_discover_tag is left alone: this refresh may have joined a
|
||||
// discover.neighbors the user started, and clearing the tag would drop
|
||||
// its remaining responses. An unused tag expires on its own.
|
||||
neighbor_table_refresh_active = false;
|
||||
pending_discover_tag = 0;
|
||||
neighbor_table_refresh_periodic = false;
|
||||
next_neighbors_publish = 0;
|
||||
} else if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) {
|
||||
// The zero-hop discover.neighbors response window is complete. Freeze the
|
||||
// refreshed table into the scope-query overlay, then publish when those
|
||||
// per-neighbor requests finish.
|
||||
bool was_periodic = neighbor_table_refresh_periodic;
|
||||
pending_discover_tag = 0;
|
||||
neighbor_table_refresh_active = false;
|
||||
neighbor_table_refresh_periodic = false;
|
||||
char tmp_reply[80];
|
||||
const char* origin_str = was_periodic ? "periodic" : "manual";
|
||||
if (startNeighborDiscover(tmp_reply)) {
|
||||
MESH_DEBUG_PRINTLN("MQTT periodic %s", tmp_reply);
|
||||
MESH_DEBUG_PRINTLN("MQTT %s %s", origin_str, tmp_reply);
|
||||
} else {
|
||||
// Avoid retrying a failed refresh on every loop iteration.
|
||||
next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval);
|
||||
MESH_DEBUG_PRINTLN("MQTT periodic neighbor scope discovery failed: %s", tmp_reply);
|
||||
if (periodic_neighbors_enabled) {
|
||||
// Avoid retrying a failed refresh on every loop iteration.
|
||||
next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval);
|
||||
}
|
||||
MESH_DEBUG_PRINTLN("MQTT %s neighbor scope discovery failed: %s", origin_str, tmp_reply);
|
||||
}
|
||||
}
|
||||
} else if (periodic_neighbors_enabled && bridge && bridge->isRunning()) {
|
||||
@@ -1562,9 +1590,29 @@ void MyMesh::loop() {
|
||||
// Refresh the zero-hop neighbor cache first (the same operation as the
|
||||
// discover.neighbors CLI command). Scope queries begin after its 60-second
|
||||
// collection window instead of publishing a stale cached table.
|
||||
sendNodeDiscoverReq();
|
||||
if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) {
|
||||
pending_discover_tag = 0;
|
||||
sendNodeDiscoverReq();
|
||||
MESH_DEBUG_PRINTLN("MQTT periodic neighbor table refresh started");
|
||||
} else {
|
||||
MESH_DEBUG_PRINTLN("MQTT periodic refresh joined active neighbor discovery");
|
||||
}
|
||||
neighbor_table_refresh_active = true;
|
||||
MESH_DEBUG_PRINTLN("MQTT periodic neighbor table refresh started");
|
||||
neighbor_table_refresh_periodic = true;
|
||||
}
|
||||
}
|
||||
|
||||
// Report the periodic-neighbors schedule for `get mqtt.status`. The timer lives
|
||||
// here, so the wrap-safe millis math stays on this side of the handoff.
|
||||
if (bridge) {
|
||||
if (neighbor_discover_active || neighbor_table_refresh_active) {
|
||||
bridge->setNeighborsSchedule(MQTTBridge::NBR_ACTIVE, 0);
|
||||
} else if (next_neighbors_publish == 0 || millisHasNowPassed(next_neighbors_publish)) {
|
||||
bridge->setNeighborsSchedule(MQTTBridge::NBR_DUE, 0);
|
||||
} else {
|
||||
long remaining_ms = (long)(next_neighbors_publish - futureMillis(0));
|
||||
uint32_t remaining_secs = remaining_ms > 0 ? (uint32_t)(remaining_ms / 1000) : 0;
|
||||
bridge->setNeighborsSchedule(MQTTBridge::NBR_SCHEDULED, remaining_secs);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
@@ -1755,11 +1803,7 @@ void MyMesh::loopNeighborDiscover() {
|
||||
finishNeighborDiscover();
|
||||
}
|
||||
|
||||
bool MyMesh::startNeighborDiscover(char* reply) {
|
||||
if (neighbor_discover_active) {
|
||||
strcpy(reply, "Err - neighbor discover already active");
|
||||
return false;
|
||||
}
|
||||
bool MyMesh::neighborDiscoverReady(char* reply) {
|
||||
#if defined(ESP_PLATFORM)
|
||||
if (!psramFound()) {
|
||||
strcpy(reply, "Err - PSRAM not available");
|
||||
@@ -1770,6 +1814,17 @@ bool MyMesh::startNeighborDiscover(char* reply) {
|
||||
strcpy(reply, "Err - MQTT bridge not running");
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool MyMesh::startNeighborDiscover(char* reply) {
|
||||
if (neighbor_discover_active) {
|
||||
strcpy(reply, "Err - neighbor discover already active");
|
||||
return false;
|
||||
}
|
||||
if (!neighborDiscoverReady(reply)) {
|
||||
return false; // reply already set
|
||||
}
|
||||
|
||||
getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf));
|
||||
neighbor_discover_count = 0;
|
||||
|
||||
@@ -153,11 +153,13 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
|
||||
uint8_t neighbor_discover_count;
|
||||
bool neighbor_discover_active;
|
||||
bool neighbor_table_refresh_active;
|
||||
bool neighbor_table_refresh_periodic;
|
||||
unsigned long neighbor_discover_until;
|
||||
unsigned long next_neighbors_publish;
|
||||
char self_scopes_buf[96];
|
||||
|
||||
bool sendAnonRegionsReq(const mesh::Identity& target, uint32_t& tag);
|
||||
bool neighborDiscoverReady(char* reply);
|
||||
bool startNeighborDiscover(char* reply);
|
||||
void loopNeighborDiscover();
|
||||
void finishNeighborDiscover();
|
||||
|
||||
@@ -23,7 +23,7 @@ enum MQTTAuthType : uint8_t {
|
||||
|
||||
enum MQTTTopicStyle : uint8_t {
|
||||
MQTT_TOPIC_MESHCORE, // meshcore/{iata}/{device_id}/{status|packets|raw}
|
||||
MQTT_TOPIC_MESHRANK, // meshrank/uplink/{token}/{device_id}/packets (packets only)
|
||||
MQTT_TOPIC_MESHRANK, // meshrank/uplink/{token}/{device_id}/{status|packets|raw|neighbors}
|
||||
};
|
||||
|
||||
struct MQTTPresetDef {
|
||||
|
||||
@@ -201,6 +201,19 @@ unsigned long MQTTBridge::getWifiConnectedAtMillis() {
|
||||
return s_wifi_connected_at;
|
||||
}
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
// Compact "3h12m" / "12m" / "45s" for CLI replies that share a 160-byte buffer.
|
||||
static void formatDuration(char* buf, size_t bufsize, uint32_t secs) {
|
||||
if (secs >= 3600) {
|
||||
snprintf(buf, bufsize, "%uh%um", (unsigned)(secs / 3600), (unsigned)((secs % 3600) / 60));
|
||||
} else if (secs >= 60) {
|
||||
snprintf(buf, bufsize, "%um", (unsigned)(secs / 60));
|
||||
} else {
|
||||
snprintf(buf, bufsize, "%us", (unsigned)secs);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPrefs* obs) {
|
||||
if (buf == nullptr || bufsize == 0) return;
|
||||
const char* msgs = (obs && obs->mqtt_status_enabled) ? "on" : "off";
|
||||
@@ -246,8 +259,40 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref
|
||||
state = "disc";
|
||||
}
|
||||
pos += snprintf(buf + pos, bufsize - pos, ", %d: %s (%s)", i + 1, name, state);
|
||||
// snprintf returns the untruncated length, so clamp before reusing pos as an
|
||||
// offset: an overflowing pos would walk past buf and underflow bufsize - pos.
|
||||
if (pos >= (int)bufsize) { pos = (int)bufsize - 1; break; }
|
||||
}
|
||||
snprintf(buf + pos, bufsize - pos, ", q:%d", q);
|
||||
if (pos < (int)bufsize - 1) {
|
||||
pos += snprintf(buf + pos, bufsize - pos, ", q:%d", q);
|
||||
if (pos >= (int)bufsize) pos = (int)bufsize - 1;
|
||||
}
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
// Periodic neighbors: time to next publish + how the last one went.
|
||||
if (obs && obs->mqtt_neighbors_enabled && pos < (int)bufsize - 1) {
|
||||
char when[16];
|
||||
switch (b->_neighbors_phase.load(std::memory_order_relaxed)) {
|
||||
case NBR_ACTIVE:
|
||||
strcpy(when, "active");
|
||||
break;
|
||||
case NBR_DUE:
|
||||
strcpy(when, "due");
|
||||
break;
|
||||
default:
|
||||
formatDuration(when, sizeof(when),
|
||||
b->_neighbors_secs_until_next.load(std::memory_order_relaxed));
|
||||
break;
|
||||
}
|
||||
const char* last;
|
||||
switch (b->_neighbors_last_result.load(std::memory_order_relaxed)) {
|
||||
case NBR_RESULT_OK: last = "ok"; break;
|
||||
case NBR_RESULT_FAIL: last = "failed"; break;
|
||||
default: last = "none"; break;
|
||||
}
|
||||
snprintf(buf + pos, bufsize - pos, ", nbr: %s/%s", when, last);
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
// On-demand publish-health + heap snapshot for the `get mqtt.stats` CLI command.
|
||||
@@ -533,6 +578,9 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg
|
||||
_neighbors_json_buffer = (char*)psram_malloc(NEIGHBORS_JSON_BUFFER_SIZE);
|
||||
_neighbors_publish_pending = false;
|
||||
_neighbors_publish_len = 0;
|
||||
_neighbors_last_result = NBR_RESULT_NONE;
|
||||
_neighbors_phase = NBR_SCHEDULED;
|
||||
_neighbors_secs_until_next = 0;
|
||||
#endif
|
||||
#else
|
||||
memset(_last_raw_data, 0, sizeof(_last_raw_data));
|
||||
@@ -1059,7 +1107,10 @@ void MQTTBridge::mqttTaskLoop() {
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
if (_neighbors_publish_pending.load(std::memory_order_acquire)) {
|
||||
if (publishNeighbors()) {
|
||||
bool ok = publishNeighbors();
|
||||
_neighbors_last_result.store(ok ? NBR_RESULT_OK : NBR_RESULT_FAIL,
|
||||
std::memory_order_relaxed);
|
||||
if (ok) {
|
||||
MQTT_DEBUG_PRINTLN("Neighbors published");
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("Neighbors publish failed");
|
||||
@@ -1864,11 +1915,17 @@ bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, bool
|
||||
// Topic building - resolves the correct topic for a given slot and message type.
|
||||
// Presets use hardcoded topic logic; custom slots support user-defined templates.
|
||||
// ---------------------------------------------------------------------------
|
||||
const char* MQTTBridge::messageTypeSuffix(MQTTMessageType type) {
|
||||
switch (type) {
|
||||
case MSG_STATUS: return "status";
|
||||
case MSG_PACKETS: return "packets";
|
||||
case MSG_RAW: return "raw";
|
||||
default: return "neighbors";
|
||||
}
|
||||
}
|
||||
|
||||
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"
|
||||
: (type == MSG_RAW) ? "raw"
|
||||
: "neighbors";
|
||||
const char* type_str = messageTypeSuffix(type);
|
||||
const char* token = _obs->mqtt_slot_token[slot_index];
|
||||
|
||||
size_t out = 0;
|
||||
@@ -1917,20 +1974,18 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
|
||||
// Preset slots: use hardcoded topic logic
|
||||
if (slot.preset) {
|
||||
if (slot.preset->topic_style == MQTT_TOPIC_MESHRANK) {
|
||||
// MeshRank: packets only, uses per-slot token in topic path
|
||||
if (type != MSG_PACKETS) return false;
|
||||
// MeshRank uses a per-slot token in its topic path. Publish every message
|
||||
// type using the same suffixes as the standard MeshCore topic layout.
|
||||
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);
|
||||
snprintf(topic_buf, buf_size, "meshrank/uplink/%s/%s/%s", token, _device_id,
|
||||
messageTypeSuffix(type));
|
||||
return true;
|
||||
}
|
||||
// MQTT_TOPIC_MESHCORE (default for all other presets)
|
||||
if (!isIATAValid()) return false;
|
||||
const char* type_str = (type == MSG_STATUS) ? "status"
|
||||
: (type == MSG_PACKETS) ? "packets"
|
||||
: (type == MSG_RAW) ? "raw"
|
||||
: "neighbors";
|
||||
snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str);
|
||||
snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id,
|
||||
messageTypeSuffix(type));
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -1940,11 +1995,8 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
|
||||
}
|
||||
// Default: meshcore format
|
||||
if (!isIATAValid()) return false;
|
||||
const char* type_str = (type == MSG_STATUS) ? "status"
|
||||
: (type == MSG_PACKETS) ? "packets"
|
||||
: (type == MSG_RAW) ? "raw"
|
||||
: "neighbors";
|
||||
snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str);
|
||||
snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id,
|
||||
messageTypeSuffix(type));
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -1958,7 +2010,7 @@ void MQTTBridge::publishStatusToSlot(int index) {
|
||||
// Build per-slot topic (handles IATA check for meshcore, token check for meshrank)
|
||||
char status_topic[128];
|
||||
if (!buildTopicForSlot(index, MSG_STATUS, status_topic, sizeof(status_topic))) {
|
||||
return; // Slot doesn't support status (e.g., meshrank) or missing required config
|
||||
return; // Slot is missing required topic configuration
|
||||
}
|
||||
|
||||
// Reuse pre-allocated buffer to avoid heap alloc/free churn under memory pressure.
|
||||
@@ -2755,8 +2807,8 @@ bool MQTTBridge::publishStatus() {
|
||||
}
|
||||
}
|
||||
}
|
||||
// If no connected slot accepts status topics (e.g. meshrank is packets-only),
|
||||
// treat as success to avoid infinite retry loops
|
||||
// If no connected slot accepts status topics, treat as success to avoid
|
||||
// infinite retry loops.
|
||||
if (published || !any_slot_wants_status) {
|
||||
if (published) MQTT_DEBUG_PRINTLN("Status published");
|
||||
return true;
|
||||
@@ -2767,6 +2819,11 @@ bool MQTTBridge::publishStatus() {
|
||||
}
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
void MQTTBridge::setNeighborsSchedule(NeighborsPhase phase, uint32_t secs_until_next) {
|
||||
_neighbors_phase.store((uint8_t)phase, std::memory_order_relaxed);
|
||||
_neighbors_secs_until_next.store(secs_until_next, std::memory_order_relaxed);
|
||||
}
|
||||
|
||||
void MQTTBridge::requestPublishNeighbors(const char* json, size_t len) {
|
||||
if (!_neighbors_json_buffer || !json || len == 0) return;
|
||||
if (_neighbors_publish_pending.load(std::memory_order_acquire)) return;
|
||||
|
||||
@@ -268,6 +268,12 @@ private:
|
||||
// Release/acquire handoff from the Arduino loop (Core 1) to the MQTT task
|
||||
// (Core 0). A second snapshot is dropped while the current one is publishing.
|
||||
std::atomic<bool> _neighbors_publish_pending;
|
||||
// Written by the MQTT task (Core 0), read by the CLI (Core 1).
|
||||
enum NeighborsResult : uint8_t { NBR_RESULT_NONE, NBR_RESULT_OK, NBR_RESULT_FAIL };
|
||||
std::atomic<uint8_t> _neighbors_last_result;
|
||||
// Written by the mesh loop (Core 1), read by the CLI (Core 1).
|
||||
std::atomic<uint8_t> _neighbors_phase;
|
||||
std::atomic<uint32_t> _neighbors_secs_until_next;
|
||||
#endif
|
||||
#if defined(BOARD_HAS_PSRAM)
|
||||
char* _publish_json_buffer;
|
||||
@@ -331,6 +337,7 @@ private:
|
||||
|
||||
// Topic building
|
||||
enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW, MSG_NEIGHBORS };
|
||||
static const char* messageTypeSuffix(MQTTMessageType type);
|
||||
bool buildTopicForSlot(int index, MQTTMessageType type, char* topic_buf, size_t buf_size);
|
||||
bool substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size);
|
||||
|
||||
@@ -495,6 +502,16 @@ public:
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
void requestPublishNeighbors(const char* json, size_t len);
|
||||
static const size_t NEIGHBORS_JSON_BUFFER_SIZE = 10240;
|
||||
|
||||
// Periodic-neighbors schedule, reported by the mesh loop for `get mqtt.status`.
|
||||
// The mesh owns the timer; the bridge only caches the summary, which keeps the
|
||||
// wrap-safe millis math on the one side that already has those helpers.
|
||||
enum NeighborsPhase : uint8_t {
|
||||
NBR_SCHEDULED, // waiting for the next publish; secs_until_next is valid
|
||||
NBR_ACTIVE, // zero-hop refresh or scope queries in flight
|
||||
NBR_DUE, // publish is due, waiting on the bridge/WiFi to come up
|
||||
};
|
||||
void setNeighborsSchedule(NeighborsPhase phase, uint32_t secs_until_next);
|
||||
#endif
|
||||
static const char* wifiReasonStr(uint8_t reason);
|
||||
static const char* tlsErrorStr(int32_t err);
|
||||
|
||||
Reference in new issue
Block a user