Add MQTT neighbors functionality for PSRAM builds
- Introduced commands for periodic neighbors publishing: `get mqtt.neighbors`, `set mqtt.neighbors on|off`, `get mqtt.neighbors.interval`, and `set mqtt.neighbors.interval <hours>`. - Updated MQTT topics to include a new neighbors topic for cached zero-hop repeater neighbors. - Added JSON message format for neighbors, including fields for SNR, last-heard age, and status. - Enhanced CLI documentation to reflect new commands and settings. - Implemented neighbor discovery logic in MyMesh class, including handling responses and publishing neighbor data. - Adjusted MQTTBridge to manage neighbors JSON buffer and publish logic. This update enhances the MQTT implementation by allowing devices to share neighbor information, improving network awareness and connectivity.
This commit is contained in:
1 parent
ae045395f9
commit
6b3c543cba
11 files changed
+525
-5
No files matched your search
+33
-1
@@ -396,6 +396,8 @@ These settings apply across all MQTT slots:
|
||||
- `get mqtt.rx` - Get RX packet uplinking setting (on/off)
|
||||
- `get mqtt.tx` - Get TX packet uplinking setting (on/off/advert)
|
||||
- `get mqtt.interval` - Get status publish interval
|
||||
- `get mqtt.neighbors` - Get periodic neighbors publish (PSRAM builds: on/off)
|
||||
- `get mqtt.neighbors.interval` - Get neighbors publish interval in hours
|
||||
- `get mqtt.ntp` - Get effective NTP server hostname
|
||||
- `get mqtt.ntp.diag` - Probe every configured NTP server for connectivity (does not change the clock; serial console shows each server's reported time, LoRa shows a compact `<server> ok|fail` list)
|
||||
- `get mqtt.owner` - Get owner public key (serial console only)
|
||||
@@ -413,6 +415,8 @@ These settings apply across all MQTT slots:
|
||||
- `advert` - Uplink only this node's own advert packets (self-originated)
|
||||
- `off` - Disable TX packet uplinking
|
||||
- `set mqtt.interval <minutes>` - Set status publish interval (1-60 minutes)
|
||||
- `set mqtt.neighbors on|off` - Enable/disable periodic neighbors/scopes publish (PSRAM builds only)
|
||||
- `set mqtt.neighbors.interval <hours>` - Set neighbors publish interval (12-8760 hours, default 24)
|
||||
- `set mqtt.ntp <hostname>` - Set custom NTP server (validated with immediate sync); `none` reverts to default
|
||||
- `set mqtt.owner <64-hex-char-public-key>` - Set owner public key
|
||||
- `set mqtt.email <email>` - Set owner email address
|
||||
@@ -507,7 +511,7 @@ The CLI commands are organized into two levels:
|
||||
|
||||
## MQTT Topics
|
||||
|
||||
The bridge publishes to three main topics with the following structure:
|
||||
The bridge publishes to four main topics with the following structure:
|
||||
|
||||
### Status Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/status`
|
||||
Device connection status and metadata (retained messages).
|
||||
@@ -518,6 +522,9 @@ Full packet data with RF characteristics and metadata.
|
||||
### Raw Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/raw`
|
||||
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).
|
||||
|
||||
**Note**: `{DEVICE_PUBLIC_KEY}` is the device's public key in hexadecimal format (64 characters).
|
||||
|
||||
## JSON Message Formats
|
||||
@@ -596,6 +603,31 @@ Minimal raw packet data for map integration.
|
||||
}
|
||||
```
|
||||
|
||||
### Neighbors Message (PSRAM observer builds)
|
||||
```json
|
||||
{
|
||||
"timestamp": "2026-06-07T12:00:00.000000+00:00",
|
||||
"origin": "MQTT Observer",
|
||||
"origin_id": "DEVICE_PUBLIC_KEY",
|
||||
"self": {
|
||||
"scopes": "Europe,UK,France"
|
||||
},
|
||||
"neighbors": [
|
||||
{
|
||||
"pubkey": "NEIGHBOR_PUBLIC_KEY",
|
||||
"snr": 8.5,
|
||||
"heard_secs_ago": 120,
|
||||
"scopes": "*,Europe",
|
||||
"status": "responded"
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
**Notes:**
|
||||
- `status` per neighbor: `responded`, `timeout`, or `send_failed` (scope query only; `self` has no status field).
|
||||
- Trigger with `discover.scopes` or enable periodic publish via `set mqtt.neighbors on`.
|
||||
|
||||
## Key Features
|
||||
|
||||
### Slot-Based Preset System
|
||||
|
||||
@@ -113,6 +113,17 @@ This document provides an overview of CLI commands that can be sent to MeshCore
|
||||
|
||||
---
|
||||
|
||||
### Discover neighbor scopes and publish to MQTT (PSRAM observer builds)
|
||||
|
||||
**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.
|
||||
|
||||
**Note:** Requires `WITH_MQTT_BRIDGE`, `MAX_NEIGHBOURS`, and PSRAM (`BOARD_HAS_PSRAM`). Returns `Err - not supported (requires PSRAM)` on other builds.
|
||||
|
||||
---
|
||||
|
||||
## Statistics
|
||||
|
||||
### Clear Stats
|
||||
@@ -1103,6 +1114,21 @@ Requires WiFi connected and the MQTT bridge running.
|
||||
|
||||
---
|
||||
|
||||
#### Periodic neighbors/scopes MQTT publish (PSRAM observer builds)
|
||||
|
||||
**Usage:**
|
||||
- `get mqtt.neighbors`
|
||||
- `set mqtt.neighbors on|off`
|
||||
- `get mqtt.neighbors.interval`
|
||||
- `set mqtt.neighbors.interval <hours>`
|
||||
|
||||
**Parameters:**
|
||||
- `hours`: Interval between automatic `discover.scopes` runs (minimum **12**, default **24**)
|
||||
|
||||
**Note:** Requires PSRAM. Publishes to `meshcore/{IATA}/{device}/neighbors` (retained, QoS 1). Use `discover.scopes` for a one-shot publish.
|
||||
|
||||
---
|
||||
|
||||
#### View or change the speed of the bridge (RS-232 only)
|
||||
**Usage:**
|
||||
- `get bridge.baud`
|
||||
|
||||
@@ -666,6 +666,16 @@ void MyMesh::onAnonDataRecv(mesh::Packet *packet, const uint8_t *secret, const m
|
||||
|
||||
int MyMesh::searchPeersByHash(const uint8_t *hash) {
|
||||
int n = 0;
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
if (neighbor_discover_active) {
|
||||
for (int i = 0; i < neighbor_discover_count && n < MAX_CLIENTS; i++) {
|
||||
auto& nb = neighbours[neighbor_discover[i].neighbour_idx];
|
||||
if (nb.heard_timestamp > 0 && nb.id.isHashMatch(hash)) {
|
||||
matching_peer_indexes[n++] = NEIGHBOR_DISCOVER_PEER_BASE + i;
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
for (int i = 0; i < acl.getNumClients(); i++) {
|
||||
if (acl.getClientByIdx(i)->id.isHashMatch(hash)) {
|
||||
matching_peer_indexes[n++] = i; // store the INDEXES of matching contacts (for subsequent 'peer' methods)
|
||||
@@ -676,6 +686,15 @@ int MyMesh::searchPeersByHash(const uint8_t *hash) {
|
||||
|
||||
void MyMesh::getPeerSharedSecret(uint8_t *dest_secret, int peer_idx) {
|
||||
int i = matching_peer_indexes[peer_idx];
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
if (neighbor_discover_active && i >= NEIGHBOR_DISCOVER_PEER_BASE) {
|
||||
int oi = i - NEIGHBOR_DISCOVER_PEER_BASE;
|
||||
if (oi >= 0 && oi < neighbor_discover_count) {
|
||||
self_id.calcSharedSecret(dest_secret, neighbours[neighbor_discover[oi].neighbour_idx].id);
|
||||
return;
|
||||
}
|
||||
}
|
||||
#endif
|
||||
if (i >= 0 && i < acl.getNumClients()) {
|
||||
// lookup pre-calculated shared_secret
|
||||
memcpy(dest_secret, acl.getClientByIdx(i)->shared_secret, PUB_KEY_SIZE);
|
||||
@@ -707,6 +726,15 @@ void MyMesh::onAdvertRecv(mesh::Packet *packet, const mesh::Identity &id, uint32
|
||||
void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx, const uint8_t *secret,
|
||||
uint8_t *data, size_t len) {
|
||||
int i = matching_peer_indexes[sender_idx];
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
if (neighbor_discover_active && i >= NEIGHBOR_DISCOVER_PEER_BASE) {
|
||||
int oi = i - NEIGHBOR_DISCOVER_PEER_BASE;
|
||||
if (type == PAYLOAD_TYPE_RESPONSE && oi >= 0 && oi < neighbor_discover_count) {
|
||||
handleNeighborDiscoverResponse(oi, data, len);
|
||||
}
|
||||
return;
|
||||
}
|
||||
#endif
|
||||
if (i < 0 || i >= acl.getNumClients()) { // get from our known_clients table (sender SHOULD already be known in this context)
|
||||
MESH_DEBUG_PRINTLN("onPeerDataRecv: invalid peer idx: %d", i);
|
||||
return;
|
||||
@@ -986,6 +1014,14 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
|
||||
pending_discover_tag = 0;
|
||||
pending_discover_until = 0;
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
neighbor_discover_count = 0;
|
||||
neighbor_discover_active = false;
|
||||
neighbor_discover_until = 0;
|
||||
next_neighbors_publish = 0;
|
||||
self_scopes_buf[0] = 0;
|
||||
#endif
|
||||
|
||||
memset(default_scope.key, 0, sizeof(default_scope.key));
|
||||
}
|
||||
|
||||
@@ -1390,6 +1426,19 @@ void MyMesh::handleCommand(uint32_t sender_timestamp, char *command, char *reply
|
||||
sendNodeDiscoverReq();
|
||||
strcpy(reply, "OK - Discover sent");
|
||||
}
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
} else if (memcmp(command, "discover.scopes", 15) == 0) {
|
||||
const char* sub = command + 15;
|
||||
while (*sub == ' ') sub++;
|
||||
if (*sub != 0) {
|
||||
strcpy(reply, "Err - discover.scopes has no options");
|
||||
} else if (!startNeighborDiscover(reply)) {
|
||||
// reply already set
|
||||
}
|
||||
#elif defined(WITH_MQTT_BRIDGE)
|
||||
} else if (memcmp(command, "discover.scopes", 15) == 0) {
|
||||
strcpy(reply, "Err - not supported (requires PSRAM)");
|
||||
#endif
|
||||
} else{
|
||||
_cli.handleCommand(sender_timestamp, command, reply); // common CLI commands
|
||||
}
|
||||
@@ -1462,6 +1511,20 @@ void MyMesh::loop() {
|
||||
_alerter.onLoop(now);
|
||||
#endif
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
if (neighbor_discover_active) {
|
||||
loopNeighborDiscover();
|
||||
} else if (_cli.getObserverPrefs()->mqtt_neighbors_enabled && bridge && bridge->isRunning()) {
|
||||
if (next_neighbors_publish == 0 ||
|
||||
(next_neighbors_publish != 0 && millisHasNowPassed(next_neighbors_publish))) {
|
||||
char tmp_reply[80];
|
||||
if (startNeighborDiscover(tmp_reply)) {
|
||||
MESH_DEBUG_PRINTLN("%s", tmp_reply);
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
#ifdef WITH_SNMP
|
||||
// Push radio stats to SNMP agent every 2 seconds
|
||||
if (_snmp_agent.isRunning()) {
|
||||
@@ -1482,6 +1545,193 @@ void MyMesh::loop() {
|
||||
#endif
|
||||
}
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
|
||||
#include "helpers/MQTTMessageBuilder.h"
|
||||
#if defined(ESP_PLATFORM)
|
||||
#include <esp_heap_caps.h>
|
||||
#endif
|
||||
|
||||
void MyMesh::getLocalScopes(char* buf, size_t len) {
|
||||
if (!buf || len == 0) return;
|
||||
buf[0] = 0;
|
||||
region_map.exportNamesTo(buf, (int)len, REGION_DENY_FLOOD);
|
||||
}
|
||||
|
||||
bool MyMesh::sendAnonRegionsReq(const mesh::Identity& target, uint32_t& tag) {
|
||||
uint8_t secret[PUB_KEY_SIZE];
|
||||
self_id.calcSharedSecret(secret, target);
|
||||
|
||||
tag = getRTCClock()->getCurrentTimeUnique();
|
||||
uint8_t inner[6];
|
||||
memcpy(inner, &tag, 4);
|
||||
inner[4] = 0x01; // ANON_REQ_TYPE_REGIONS
|
||||
inner[5] = 0x00; // zero-hop reply path
|
||||
|
||||
mesh::Packet* pkt = createAnonDatagram(PAYLOAD_TYPE_ANON_REQ, self_id, target, secret, inner, sizeof(inner));
|
||||
if (!pkt) return false;
|
||||
sendDirect(pkt, NULL, 0, 0);
|
||||
return true;
|
||||
}
|
||||
|
||||
void MyMesh::handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len) {
|
||||
if (overlay_idx < 0 || overlay_idx >= neighbor_discover_count) return;
|
||||
NeighborDiscoverEntry& entry = neighbor_discover[overlay_idx];
|
||||
if (entry.status != ND_PENDING || len < 8) return;
|
||||
|
||||
uint32_t tag;
|
||||
memcpy(&tag, data, 4);
|
||||
if (tag != entry.tag) return;
|
||||
|
||||
size_t scope_len = len - 8;
|
||||
if (scope_len >= sizeof(entry.scopes)) {
|
||||
scope_len = sizeof(entry.scopes) - 1;
|
||||
}
|
||||
memcpy(entry.scopes, &data[8], scope_len);
|
||||
entry.scopes[scope_len] = 0;
|
||||
entry.status = ND_RESPONDED;
|
||||
}
|
||||
|
||||
void MyMesh::finishNeighborDiscover() {
|
||||
getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf));
|
||||
|
||||
char self_pubkey_hex[65];
|
||||
mesh::Utils::toHex(self_pubkey_hex, self_id.pub_key, PUB_KEY_SIZE);
|
||||
|
||||
char origin[32];
|
||||
MQTTBridge::getEffectiveMqttOrigin(&_prefs, _cli.getObserverPrefs(), origin, sizeof(origin));
|
||||
|
||||
char timestamp[40];
|
||||
MQTTMessageBuilder::formatIsoTimestampForMqtt(getRTCClock()->getCurrentTime(), 0, nullptr, timestamp, sizeof(timestamp));
|
||||
|
||||
char pubkey_hex[MAX_NEIGHBOURS][65];
|
||||
MQTTMessageBuilder::NeighborsMessageEntry entries[MAX_NEIGHBOURS];
|
||||
uint32_t now_secs = getRTCClock()->getCurrentTime();
|
||||
|
||||
for (int i = 0; i < neighbor_discover_count; i++) {
|
||||
auto& nb = neighbours[neighbor_discover[i].neighbour_idx];
|
||||
mesh::Utils::toHex(pubkey_hex[i], nb.id.pub_key, PUB_KEY_SIZE);
|
||||
entries[i].pubkey_hex = pubkey_hex[i];
|
||||
entries[i].snr = nb.snr / 4.0f;
|
||||
entries[i].heard_secs_ago = (nb.heard_timestamp > 0 && now_secs >= nb.heard_timestamp)
|
||||
? (now_secs - nb.heard_timestamp) : 0;
|
||||
entries[i].scopes = neighbor_discover[i].scopes;
|
||||
switch (neighbor_discover[i].status) {
|
||||
case ND_RESPONDED: entries[i].status = "responded"; break;
|
||||
case ND_SEND_FAILED: entries[i].status = "send_failed"; break;
|
||||
default: entries[i].status = "timeout"; break;
|
||||
}
|
||||
}
|
||||
|
||||
#if defined(ESP_PLATFORM)
|
||||
char* json_buf = (char*)heap_caps_malloc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE, MALLOC_CAP_SPIRAM);
|
||||
#else
|
||||
char* json_buf = (char*)malloc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE);
|
||||
#endif
|
||||
if (!json_buf) {
|
||||
neighbor_discover_active = false;
|
||||
neighbor_discover_count = 0;
|
||||
if (_cli.getObserverPrefs()->mqtt_neighbors_enabled) {
|
||||
next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
DynamicJsonDocument doc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE);
|
||||
int json_len = MQTTMessageBuilder::buildNeighborsMessage(
|
||||
doc, origin, self_pubkey_hex, timestamp, self_scopes_buf,
|
||||
entries, neighbor_discover_count,
|
||||
json_buf, MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE);
|
||||
|
||||
if (json_len > 0 && bridge) {
|
||||
bridge->requestPublishNeighbors(json_buf, (size_t)json_len);
|
||||
}
|
||||
|
||||
#if defined(ESP_PLATFORM)
|
||||
heap_caps_free(json_buf);
|
||||
#else
|
||||
free(json_buf);
|
||||
#endif
|
||||
|
||||
neighbor_discover_active = false;
|
||||
neighbor_discover_count = 0;
|
||||
if (_cli.getObserverPrefs()->mqtt_neighbors_enabled) {
|
||||
next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval);
|
||||
}
|
||||
}
|
||||
|
||||
void MyMesh::loopNeighborDiscover() {
|
||||
if (!neighbor_discover_active) return;
|
||||
|
||||
bool all_done = true;
|
||||
for (int i = 0; i < neighbor_discover_count; i++) {
|
||||
if (neighbor_discover[i].status == ND_PENDING) {
|
||||
all_done = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (!all_done && !millisHasNowPassed(neighbor_discover_until)) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!all_done) {
|
||||
for (int i = 0; i < neighbor_discover_count; i++) {
|
||||
if (neighbor_discover[i].status == ND_PENDING) {
|
||||
neighbor_discover[i].status = ND_TIMEOUT;
|
||||
}
|
||||
}
|
||||
}
|
||||
finishNeighborDiscover();
|
||||
}
|
||||
|
||||
bool MyMesh::startNeighborDiscover(char* reply) {
|
||||
if (neighbor_discover_active) {
|
||||
strcpy(reply, "Err - neighbor discover already active");
|
||||
return false;
|
||||
}
|
||||
#if defined(ESP_PLATFORM)
|
||||
if (!psramFound()) {
|
||||
strcpy(reply, "Err - PSRAM not available");
|
||||
return false;
|
||||
}
|
||||
#endif
|
||||
if (!bridge || !bridge->isRunning()) {
|
||||
strcpy(reply, "Err - MQTT bridge not running");
|
||||
return false;
|
||||
}
|
||||
|
||||
getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf));
|
||||
neighbor_discover_count = 0;
|
||||
for (int i = 0; i < MAX_NEIGHBOURS; i++) {
|
||||
if (neighbours[i].heard_timestamp > 0) {
|
||||
neighbor_discover[neighbor_discover_count].neighbour_idx = (uint8_t)i;
|
||||
neighbor_discover[neighbor_discover_count].scopes[0] = 0;
|
||||
neighbor_discover[neighbor_discover_count].status = ND_PENDING;
|
||||
uint32_t tag;
|
||||
if (sendAnonRegionsReq(neighbours[i].id, tag)) {
|
||||
neighbor_discover[neighbor_discover_count].tag = tag;
|
||||
} else {
|
||||
neighbor_discover[neighbor_discover_count].status = ND_SEND_FAILED;
|
||||
}
|
||||
neighbor_discover_count++;
|
||||
}
|
||||
}
|
||||
|
||||
neighbor_discover_active = true;
|
||||
neighbor_discover_until = futureMillis(NEIGHBOR_DISCOVER_TIMEOUT_MS);
|
||||
|
||||
if (neighbor_discover_count == 0) {
|
||||
finishNeighborDiscover();
|
||||
strcpy(reply, "OK - neighbor discover started (0 neighbors, self only)");
|
||||
} else {
|
||||
sprintf(reply, "OK - neighbor discover started (%u neighbors)", (unsigned)neighbor_discover_count);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
#endif // WITH_MQTT_NEIGHBORS
|
||||
|
||||
// To check if there is pending work
|
||||
bool MyMesh::hasPendingWork() const {
|
||||
#if defined(WITH_BRIDGE)
|
||||
|
||||
@@ -136,6 +136,36 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
|
||||
AlertReporter _alerter;
|
||||
#endif
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
enum NeighborDiscoverStatus : uint8_t {
|
||||
ND_PENDING = 1,
|
||||
ND_RESPONDED = 2,
|
||||
ND_TIMEOUT = 3,
|
||||
ND_SEND_FAILED = 4,
|
||||
};
|
||||
struct NeighborDiscoverEntry {
|
||||
uint8_t neighbour_idx;
|
||||
uint32_t tag;
|
||||
char scopes[96];
|
||||
uint8_t status;
|
||||
};
|
||||
NeighborDiscoverEntry neighbor_discover[MAX_NEIGHBOURS];
|
||||
uint8_t neighbor_discover_count;
|
||||
bool neighbor_discover_active;
|
||||
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 startNeighborDiscover(char* reply);
|
||||
void loopNeighborDiscover();
|
||||
void finishNeighborDiscover();
|
||||
void handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len);
|
||||
void getLocalScopes(char* buf, size_t len);
|
||||
static const int NEIGHBOR_DISCOVER_PEER_BASE = 1000;
|
||||
static const unsigned long NEIGHBOR_DISCOVER_TIMEOUT_MS = 30000;
|
||||
#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);
|
||||
uint8_t handleAnonRegionsReq(const mesh::Identity& sender, uint32_t sender_timestamp, const uint8_t* data);
|
||||
|
||||
@@ -156,6 +156,12 @@ struct MQTTPrefs {
|
||||
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
|
||||
|
||||
// Neighbors publishing (PSRAM boards only). Appended at the end of the observer
|
||||
// tail so a shorter /mqtt_prefs payload from earlier firmware still loads with
|
||||
// these defaulting off (see applyMQTTDefaults); keeps the format at VERSION 1.
|
||||
uint8_t mqtt_neighbors_enabled; // Periodic neighbors/scopes MQTT publish (PSRAM boards only)
|
||||
uint32_t mqtt_neighbors_interval; // Neighbors publish interval (ms), default 24h, min 12h
|
||||
};
|
||||
|
||||
// /mqtt_prefs is written with an 8-byte header so the format is self-describing.
|
||||
|
||||
@@ -246,6 +246,26 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf
|
||||
} else {
|
||||
strcpy(reply, "Error: interval must be between 1-60 minutes");
|
||||
}
|
||||
#if defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0
|
||||
} else if (memcmp(config, "mqtt.neighbors ", 14) == 0) {
|
||||
_mqtt_prefs.mqtt_neighbors_enabled = memcmp(&config[14], "on", 2) == 0;
|
||||
savePrefs();
|
||||
strcpy(reply, "OK");
|
||||
} else if (memcmp(config, "mqtt.neighbors.interval ", 24) == 0) {
|
||||
uint32_t hours = _atoi(&config[24]);
|
||||
if (hours >= 12 && hours <= 8760) {
|
||||
_mqtt_prefs.mqtt_neighbors_interval = hours * 3600000UL;
|
||||
savePrefs();
|
||||
sprintf(reply, "OK - neighbors interval set to %u hours (%lu ms)", hours,
|
||||
(unsigned long)_mqtt_prefs.mqtt_neighbors_interval);
|
||||
} else {
|
||||
strcpy(reply, "Error: neighbors interval must be between 12-8760 hours");
|
||||
}
|
||||
#elif defined(WITH_MQTT_BRIDGE)
|
||||
} else if (memcmp(config, "mqtt.neighbors ", 14) == 0 ||
|
||||
memcmp(config, "mqtt.neighbors.interval ", 24) == 0) {
|
||||
strcpy(reply, "Err - not supported (requires PSRAM)");
|
||||
#endif
|
||||
} else if (memcmp(config, "mqtt.ntp ", 9) == 0) {
|
||||
const char* host = &config[9];
|
||||
while (*host == ' ') host++;
|
||||
@@ -709,6 +729,17 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf
|
||||
} 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);
|
||||
#if defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0
|
||||
} else if (memcmp(config, "mqtt.neighbors.interval", 23) == 0) {
|
||||
uint32_t hours = (_mqtt_prefs.mqtt_neighbors_interval + 3599999) / 3600000;
|
||||
sprintf(reply, "> %u hours (%lu ms)", hours, (unsigned long)_mqtt_prefs.mqtt_neighbors_interval);
|
||||
} else if (memcmp(config, "mqtt.neighbors", 14) == 0) {
|
||||
sprintf(reply, "> %s", _mqtt_prefs.mqtt_neighbors_enabled ? "on" : "off");
|
||||
#elif defined(WITH_MQTT_BRIDGE)
|
||||
} else if (memcmp(config, "mqtt.neighbors", 14) == 0 ||
|
||||
memcmp(config, "mqtt.neighbors.interval", 23) == 0) {
|
||||
strcpy(reply, "Err - not supported (requires PSRAM)");
|
||||
#endif
|
||||
} 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.
|
||||
|
||||
@@ -68,6 +68,8 @@ static inline void applyMQTTDefaults(MQTTPrefs* prefs) {
|
||||
prefs->mqtt_tx_enabled = 2;
|
||||
prefs->mqtt_rx_enabled = 1;
|
||||
prefs->mqtt_status_interval = 300000;
|
||||
prefs->mqtt_neighbors_enabled = 0;
|
||||
prefs->mqtt_neighbors_interval = 24UL * 3600UL * 1000UL;
|
||||
prefs->wifi_power_save = 1;
|
||||
|
||||
mqttDefaultSlotPreset(prefs->mqtt_slot_preset[0], sizeof(prefs->mqtt_slot_preset[0]),
|
||||
|
||||
@@ -227,6 +227,41 @@ int MQTTMessageBuilder::buildRawMessage(
|
||||
return (len > 0 && len < buffer_size) ? len : 0;
|
||||
}
|
||||
|
||||
int MQTTMessageBuilder::buildNeighborsMessage(
|
||||
JsonDocument& doc,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
const char* timestamp,
|
||||
const char* self_scopes,
|
||||
const NeighborsMessageEntry* neighbors,
|
||||
int neighbor_count,
|
||||
char* buffer,
|
||||
size_t buffer_size
|
||||
) {
|
||||
doc.clear();
|
||||
JsonObject root = doc.to<JsonObject>();
|
||||
|
||||
root["timestamp"] = timestamp;
|
||||
root["origin"] = origin;
|
||||
root["origin_id"] = origin_id;
|
||||
|
||||
JsonObject self = root.createNestedObject("self");
|
||||
self["scopes"] = self_scopes ? self_scopes : "";
|
||||
|
||||
JsonArray arr = root.createNestedArray("neighbors");
|
||||
for (int i = 0; i < neighbor_count; i++) {
|
||||
JsonObject nb = arr.createNestedObject();
|
||||
nb["pubkey"] = neighbors[i].pubkey_hex;
|
||||
nb["snr"] = neighbors[i].snr;
|
||||
nb["heard_secs_ago"] = neighbors[i].heard_secs_ago;
|
||||
nb["scopes"] = neighbors[i].scopes ? neighbors[i].scopes : "";
|
||||
nb["status"] = neighbors[i].status;
|
||||
}
|
||||
|
||||
size_t len = serializeJson(root, buffer, buffer_size);
|
||||
return (len > 0 && len < buffer_size) ? len : 0;
|
||||
}
|
||||
|
||||
int MQTTMessageBuilder::buildPacketJSON(
|
||||
JsonDocument& doc,
|
||||
mesh::Packet* packet,
|
||||
|
||||
@@ -153,6 +153,29 @@ public:
|
||||
size_t buffer_size
|
||||
);
|
||||
|
||||
struct NeighborsMessageEntry {
|
||||
const char* pubkey_hex;
|
||||
float snr;
|
||||
uint32_t heard_secs_ago;
|
||||
const char* scopes;
|
||||
const char* status;
|
||||
};
|
||||
|
||||
/**
|
||||
* Build neighbors table JSON for meshcore/{iata}/{device}/neighbors topic.
|
||||
*/
|
||||
static int buildNeighborsMessage(
|
||||
JsonDocument& doc,
|
||||
const char* origin,
|
||||
const char* origin_id,
|
||||
const char* timestamp,
|
||||
const char* self_scopes,
|
||||
const NeighborsMessageEntry* neighbors,
|
||||
int neighbor_count,
|
||||
char* buffer,
|
||||
size_t buffer_size
|
||||
);
|
||||
|
||||
/**
|
||||
* Convert packet to JSON message
|
||||
*
|
||||
|
||||
@@ -506,6 +506,11 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg
|
||||
_last_raw_data = (uint8_t*)psram_malloc(LAST_RAW_DATA_SIZE);
|
||||
_publish_json_buffer = (char*)psram_malloc(PUBLISH_JSON_BUFFER_SIZE);
|
||||
_status_json_buffer = (char*)psram_malloc(STATUS_JSON_BUFFER_SIZE);
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
_neighbors_json_buffer = (char*)psram_malloc(NEIGHBORS_JSON_BUFFER_SIZE);
|
||||
_neighbors_publish_pending = false;
|
||||
_neighbors_publish_len = 0;
|
||||
#endif
|
||||
#else
|
||||
memset(_last_raw_data, 0, sizeof(_last_raw_data));
|
||||
#endif
|
||||
@@ -804,6 +809,11 @@ void MQTTBridge::end() {
|
||||
psram_free(_last_raw_data); _last_raw_data = nullptr;
|
||||
psram_free(_publish_json_buffer); _publish_json_buffer = nullptr;
|
||||
psram_free(_status_json_buffer); _status_json_buffer = nullptr;
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
psram_free(_neighbors_json_buffer); _neighbors_json_buffer = nullptr;
|
||||
_neighbors_publish_pending = false;
|
||||
_neighbors_publish_len = 0;
|
||||
#endif
|
||||
#endif
|
||||
// JSON documents are now StaticJsonDocument inline members — no heap allocation to free.
|
||||
|
||||
@@ -1023,6 +1033,17 @@ void MQTTBridge::mqttTaskLoop() {
|
||||
// Process packet queue
|
||||
processPacketQueue();
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
if (_neighbors_publish_pending) {
|
||||
_neighbors_publish_pending = false;
|
||||
if (publishNeighbors()) {
|
||||
MQTT_DEBUG_PRINTLN("Neighbors published");
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("Neighbors publish failed");
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
#ifdef WITH_SNMP
|
||||
// SNMP agent loop — process incoming UDP requests
|
||||
if (_snmp_agent) {
|
||||
@@ -1820,7 +1841,10 @@ bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, bool
|
||||
// Presets use hardcoded topic logic; custom slots support user-defined templates.
|
||||
// ---------------------------------------------------------------------------
|
||||
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* type_str = (type == MSG_STATUS) ? "status"
|
||||
: (type == MSG_PACKETS) ? "packets"
|
||||
: (type == MSG_RAW) ? "raw"
|
||||
: "neighbors";
|
||||
const char* token = _obs->mqtt_slot_token[slot_index];
|
||||
|
||||
size_t out = 0;
|
||||
@@ -1878,7 +1902,10 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
|
||||
}
|
||||
// MQTT_TOPIC_MESHCORE (default for all other presets)
|
||||
if (!isIATAValid()) return false;
|
||||
const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw";
|
||||
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);
|
||||
return true;
|
||||
}
|
||||
@@ -1889,7 +1916,10 @@ 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" : "raw";
|
||||
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);
|
||||
return true;
|
||||
}
|
||||
@@ -2716,6 +2746,44 @@ bool MQTTBridge::publishStatus() {
|
||||
return false;
|
||||
}
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
void MQTTBridge::requestPublishNeighbors(const char* json, size_t len) {
|
||||
if (!_neighbors_json_buffer || !json || len == 0) return;
|
||||
if (len >= NEIGHBORS_JSON_BUFFER_SIZE) {
|
||||
len = NEIGHBORS_JSON_BUFFER_SIZE - 1;
|
||||
}
|
||||
memcpy(_neighbors_json_buffer, json, len);
|
||||
_neighbors_json_buffer[len] = '\0';
|
||||
_neighbors_publish_len = len;
|
||||
_neighbors_publish_pending = true;
|
||||
}
|
||||
|
||||
bool MQTTBridge::publishNeighbors() {
|
||||
if (!_neighbors_json_buffer || _neighbors_publish_len == 0) {
|
||||
return false;
|
||||
}
|
||||
if (!_cached_has_connected_slots) {
|
||||
return false;
|
||||
}
|
||||
|
||||
refreshOriginFromPrefs();
|
||||
|
||||
bool published = false;
|
||||
char topic[128];
|
||||
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
|
||||
if (_slots[i].enabled && _slots[i].client && _slots[i].connected) {
|
||||
if (buildTopicForSlot(i, MSG_NEIGHBORS, topic, sizeof(topic))) {
|
||||
bool use_retain = _slots[i].preset ? _slots[i].preset->allow_retain : false;
|
||||
if (publishToSlot(i, topic, _neighbors_json_buffer, use_retain, 1)) {
|
||||
published = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return published;
|
||||
}
|
||||
#endif
|
||||
|
||||
bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx,
|
||||
const uint8_t* raw_data, int raw_len,
|
||||
float snr, float rssi) {
|
||||
|
||||
@@ -35,6 +35,10 @@ class MeshSNMPAgent; // Forward declaration
|
||||
|
||||
#ifdef WITH_MQTT_BRIDGE
|
||||
|
||||
#if defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0
|
||||
#define WITH_MQTT_NEIGHBORS 1
|
||||
#endif
|
||||
|
||||
/**
|
||||
* @brief Bridge implementation using MQTT protocol for packet transport
|
||||
*
|
||||
@@ -257,6 +261,11 @@ private:
|
||||
// class object so these allocations don't interleave with large TLS buffers at startup.
|
||||
static const size_t PUBLISH_JSON_BUFFER_SIZE = 2048;
|
||||
static const size_t STATUS_JSON_BUFFER_SIZE = 768;
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
char* _neighbors_json_buffer;
|
||||
size_t _neighbors_publish_len;
|
||||
bool _neighbors_publish_pending;
|
||||
#endif
|
||||
#if defined(BOARD_HAS_PSRAM)
|
||||
char* _publish_json_buffer;
|
||||
char* _status_json_buffer;
|
||||
@@ -318,7 +327,7 @@ private:
|
||||
mesh::MillisecondClock* _ms; // For uptime
|
||||
|
||||
// Topic building
|
||||
enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW };
|
||||
enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW, MSG_NEIGHBORS };
|
||||
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);
|
||||
|
||||
@@ -349,6 +358,9 @@ private:
|
||||
|
||||
void processPacketQueue();
|
||||
bool publishStatus(); // Returns true if status was successfully published
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
bool publishNeighbors();
|
||||
#endif
|
||||
bool handleWiFiConnection(unsigned long now);
|
||||
|
||||
// FreeRTOS task function (runs on Core 0)
|
||||
@@ -477,6 +489,11 @@ public:
|
||||
static void formatSlotDiagReply(char* buf, size_t bufsize, int slot_index);
|
||||
static uint8_t getLastWifiDisconnectReason();
|
||||
static unsigned long getLastWifiDisconnectTime();
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
void requestPublishNeighbors(const char* json, size_t len);
|
||||
static const size_t NEIGHBORS_JSON_BUFFER_SIZE = 10240;
|
||||
#endif
|
||||
static const char* wifiReasonStr(uint8_t reason);
|
||||
static const char* tlsErrorStr(int32_t err);
|
||||
|
||||
|
||||
Reference in new issue
Block a user