feat(mqtt): add neighbor-scope discovery + periodic publish to MyMesh
Port the neighbor-discovery state machine from mqtt-bridge-implementation-flex, adapted to this branch's MyMesh (WebConfig members shifted the insertion points; applied by content). - Two-stage periodic refresh in loop() driven by mqtt_neighbors_interval: stage 1 is a zero-hop sendNodeDiscoverReq() (reuses the existing 60s window), stage 2 (startNeighborDiscover) fires one anon-regions scope query per heard neighbour, then finishNeighborDiscover() builds the table JSON via MQTTMessageBuilder::buildNeighborsMessage and hands it to bridge->requestPublishNeighbors(). - Peer overlay at NEIGHBOR_DISCOVER_PEER_BASE lets scope-query RESPONSE packets from non-ACL neighbours decrypt: searchPeersByHash prepends heard neighbours (bounded by MAX_CLIENTS), getPeerSharedSecret derives the secret on the fly, and onPeerDataRecv routes both overlay-index and ACL-client-that-is-a-neighbour responses into handleNeighborDiscoverResponse. - Entries ordered most- to least-useful (recent, then stronger SNR) so the JSON builder's tail-drop keeps the useful head. - `discover.scopes` CLI command (manual trigger), with a WITH_MQTT_BRIDGE stub replying "requires PSRAM" on non-PSRAM builds. - Reports schedule to the bridge each loop via setNeighborsSchedule(). - Uses ArduinoJson v7 JsonDocument (not deprecated DynamicJsonDocument). All gated on WITH_MQTT_NEIGHBORS. Reuses the existing MQTTBridge* member. Verified: T_Beam_S3_Supreme_SX1262_repeater_observer_mqtt builds [SUCCESS].
This commit is contained in:
1 parent
de320bc4df
commit
34a8220111
2 files changed
+403
-1
No files matched your search
@@ -666,7 +666,22 @@ void MyMesh::onAnonDataRecv(mesh::Packet *packet, const uint8_t *secret, const m
|
||||
|
||||
int MyMesh::searchPeersByHash(const uint8_t *hash) {
|
||||
int n = 0;
|
||||
for (int i = 0; i < acl.getNumClients(); i++) {
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
// While a neighbor-scope discovery is active, overlay the heard neighbours
|
||||
// that are NOT already ACL clients so their RESPONSE packets can be decoded.
|
||||
// Overlay indices are offset by NEIGHBOR_DISCOVER_PEER_BASE to keep them
|
||||
// distinct from real ACL indices.
|
||||
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 (acl.getClient(nb.id.pub_key, PUB_KEY_SIZE) != nullptr) continue;
|
||||
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() && n < MAX_CLIENTS; 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 +691,16 @@ 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)
|
||||
// Overlay entries have no precomputed shared secret; derive it on the fly.
|
||||
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,11 +732,34 @@ 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)
|
||||
// Overlay response: a heard neighbour (not an ACL client) answering our
|
||||
// anon-regions scope query. Consume it and stop — it is not a client packet.
|
||||
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;
|
||||
}
|
||||
ClientInfo* client = acl.getClientByIdx(i);
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
// A neighbour that IS an ACL client resolves to a normal index above, so a
|
||||
// scope-query response from it lands here — match it against the overlay.
|
||||
if (neighbor_discover_active && type == PAYLOAD_TYPE_RESPONSE) {
|
||||
for (int oi = 0; oi < neighbor_discover_count; oi++) {
|
||||
auto& nb = neighbours[neighbor_discover[oi].neighbour_idx];
|
||||
if (client->id.matches(nb.id) && handleNeighborDiscoverResponse(oi, data, len)) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
if (type == PAYLOAD_TYPE_REQ) { // request (from a Known admin client!)
|
||||
uint32_t timestamp;
|
||||
@@ -986,6 +1034,16 @@ 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_table_refresh_active = false;
|
||||
neighbor_table_refresh_periodic = false;
|
||||
neighbor_discover_until = 0;
|
||||
next_neighbors_publish = 0;
|
||||
self_scopes_buf[0] = 0;
|
||||
#endif
|
||||
|
||||
memset(default_scope.key, 0, sizeof(default_scope.key));
|
||||
}
|
||||
|
||||
@@ -1519,6 +1577,36 @@ 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 (pending_discover_tag != 0 &&
|
||||
!millisHasNowPassed(pending_discover_until) &&
|
||||
!neighbor_discover_active) {
|
||||
// A zero-hop table refresh is already collecting; queue the scope pass
|
||||
// behind it (as a manual, non-periodic request) rather than starting a
|
||||
// second refresh.
|
||||
if (!neighborDiscoverReady(reply)) {
|
||||
// reply already set by neighborDiscoverReady
|
||||
} 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 by startNeighborDiscover
|
||||
}
|
||||
#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
|
||||
}
|
||||
@@ -1615,6 +1703,66 @@ void MyMesh::loop() {
|
||||
_alerter.onLoop(now);
|
||||
#endif
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
// Two-stage periodic neighbors publication:
|
||||
// stage 1 - zero-hop node-discover refreshes the neighbour table (60s window)
|
||||
// stage 2 - anon-regions scope query per neighbour (startNeighborDiscover)
|
||||
// then the table JSON is published and the next cycle is rescheduled.
|
||||
bool periodic_neighbors_enabled = _cli.getObserverPrefs()->mqtt_neighbors_enabled;
|
||||
if (neighbor_discover_active) {
|
||||
loopNeighborDiscover();
|
||||
} else if (neighbor_table_refresh_active) {
|
||||
if (neighbor_table_refresh_periodic && !periodic_neighbors_enabled) {
|
||||
// periodic switched off mid-refresh -> cancel (leave pending_discover_tag alone)
|
||||
neighbor_table_refresh_active = false;
|
||||
neighbor_table_refresh_periodic = false;
|
||||
next_neighbors_publish = 0;
|
||||
} else if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) {
|
||||
// 60s zero-hop window done -> begin the per-neighbour scope queries
|
||||
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 %s %s", origin_str, tmp_reply);
|
||||
} else {
|
||||
if (periodic_neighbors_enabled) {
|
||||
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()) {
|
||||
if (next_neighbors_publish == 0 ||
|
||||
(next_neighbors_publish != 0 && millisHasNowPassed(next_neighbors_publish))) {
|
||||
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;
|
||||
neighbor_table_refresh_periodic = true;
|
||||
}
|
||||
}
|
||||
|
||||
// Report the schedule state back to the bridge for `get mqtt.status`.
|
||||
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
|
||||
|
||||
#ifdef WITH_SNMP
|
||||
// Push radio stats to SNMP agent every 2 seconds
|
||||
if (_snmp_agent.isRunning()) {
|
||||
@@ -1635,6 +1783,222 @@ void MyMesh::loop() {
|
||||
#endif
|
||||
}
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
#include "helpers/MQTTMessageBuilder.h"
|
||||
#if defined(ESP_PLATFORM)
|
||||
#include <esp_heap_caps.h>
|
||||
#endif
|
||||
|
||||
// This node's own non-flood scope names, same source the anon-regions server
|
||||
// reply uses. Empty string when the node has no scoped regions.
|
||||
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);
|
||||
}
|
||||
|
||||
// Client side of the anon-regions request (the server side is handleAnonRegionsReq).
|
||||
// Inner payload: {tag(4)}{ANON_REQ_TYPE_REGIONS}{0x00 = zero-hop reply path}.
|
||||
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] = ANON_REQ_TYPE_REGIONS;
|
||||
inner[5] = 0x00; // request a 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;
|
||||
}
|
||||
|
||||
// Match a RESPONSE against the pending overlay entry by tag; copy its scope
|
||||
// string (payload after the 8-byte {tag}{clock} header) into the entry.
|
||||
bool MyMesh::handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len) {
|
||||
if (overlay_idx < 0 || overlay_idx >= neighbor_discover_count) return false;
|
||||
NeighborDiscoverEntry& entry = neighbor_discover[overlay_idx];
|
||||
if (entry.status != ND_PENDING || len < 8) return false;
|
||||
|
||||
uint32_t tag;
|
||||
memcpy(&tag, data, 4);
|
||||
if (tag != entry.tag) return false;
|
||||
|
||||
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;
|
||||
return true;
|
||||
}
|
||||
|
||||
// Publish-ordering: most recently heard first, then stronger SNR, then pubkey.
|
||||
// The JSON builder drops the tail if the buffer fills, so the head must be the
|
||||
// most useful entries.
|
||||
static bool neighborPublishEntryComesBefore(
|
||||
const MQTTMessageBuilder::NeighborsMessageEntry& lhs,
|
||||
const MQTTMessageBuilder::NeighborsMessageEntry& rhs) {
|
||||
if (lhs.heard_secs_ago != rhs.heard_secs_ago) {
|
||||
return lhs.heard_secs_ago < rhs.heard_secs_ago; // newer first
|
||||
}
|
||||
if (lhs.snr != rhs.snr) {
|
||||
return lhs.snr > rhs.snr; // stronger first when equally recent
|
||||
}
|
||||
return strcmp(lhs.pubkey_hex, rhs.pubkey_hex) < 0;
|
||||
}
|
||||
|
||||
// Build the neighbors-table JSON and hand it to the bridge, then reschedule.
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
// insertion sort: most useful first (JSON builder drops the tail on overflow)
|
||||
for (int i = 1; i < neighbor_discover_count; i++) {
|
||||
MQTTMessageBuilder::NeighborsMessageEntry entry = entries[i];
|
||||
int j = i;
|
||||
while (j > 0 && neighborPublishEntryComesBefore(entry, entries[j - 1])) {
|
||||
entries[j] = entries[j - 1];
|
||||
j--;
|
||||
}
|
||||
entries[j] = entry;
|
||||
}
|
||||
|
||||
#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;
|
||||
}
|
||||
|
||||
JsonDocument doc;
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
// Advance the scope-query phase; publish once all entries resolve or the window
|
||||
// times out (stragglers marked ND_TIMEOUT).
|
||||
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();
|
||||
}
|
||||
|
||||
// Shared precondition for starting a discovery: PSRAM present + bridge running.
|
||||
bool MyMesh::neighborDiscoverReady(char* reply) {
|
||||
#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; }
|
||||
return true;
|
||||
}
|
||||
|
||||
// Snapshot the neighbor table into the overlay and fire one anon-regions query
|
||||
// per heard neighbour; arm the 30s scope-query window.
|
||||
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;
|
||||
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)
|
||||
|
||||
@@ -147,6 +147,44 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
|
||||
uint8_t _wc_slot_restart_mask = 0;
|
||||
#endif
|
||||
|
||||
#if defined(WITH_MQTT_NEIGHBORS)
|
||||
// Neighbor-scope discovery: a snapshot of the neighbor table overlaid with an
|
||||
// in-flight anon-regions query per neighbor, published to the MQTT neighbors
|
||||
// topic once every neighbor has responded or the window times out.
|
||||
enum NeighborDiscoverStatus : uint8_t {
|
||||
ND_PENDING = 1,
|
||||
ND_RESPONDED = 2,
|
||||
ND_TIMEOUT = 3,
|
||||
ND_SEND_FAILED = 4,
|
||||
};
|
||||
struct NeighborDiscoverEntry {
|
||||
uint8_t neighbour_idx; // index into neighbours[]
|
||||
uint32_t tag; // anon-regions request tag we're waiting on
|
||||
char scopes[96]; // scope names from the response
|
||||
uint8_t status; // NeighborDiscoverStatus
|
||||
};
|
||||
NeighborDiscoverEntry neighbor_discover[MAX_NEIGHBOURS];
|
||||
uint8_t neighbor_discover_count;
|
||||
bool neighbor_discover_active; // scope-query phase in flight
|
||||
bool neighbor_table_refresh_active; // zero-hop table refresh (stage 1) in flight
|
||||
bool neighbor_table_refresh_periodic; // that refresh was kicked by the periodic timer
|
||||
unsigned long neighbor_discover_until; // scope-query timeout deadline
|
||||
unsigned long next_neighbors_publish; // periodic publish deadline (0 = fire ASAP)
|
||||
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();
|
||||
bool handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len);
|
||||
void getLocalScopes(char* buf, size_t len);
|
||||
// Overlay peer indices are offset by this base so onPeerDataRecv can tell a
|
||||
// discovery response apart from a normal ACL-client index.
|
||||
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);
|
||||
|
||||
Reference in new issue
Block a user