From 85c0e1de11cfbfb61b2f20460afbcfedd20c6ed5 Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 20 Jul 2026 08:47:37 -0700 Subject: [PATCH] feat(mqtt): implement neighbor discovery and management in MyMesh Add support for neighbor discovery and management in the MyMesh class. This includes structures for neighbor information, methods for handling neighbor advertisements, and control data reception. The new features allow for dynamic neighbor management, enhancing the mesh network's capabilities. Additionally, the maximum number of neighbors is set to 50 across various platform configurations. --- examples/simple_room_server/MyMesh.cpp | 575 +++++++++++++++++- examples/simple_room_server/MyMesh.h | 53 +- variants/heltec_t190/platformio.ini | 1 + variants/heltec_tracker_v2/platformio.ini | 2 + variants/heltec_v3/platformio.ini | 2 + variants/heltec_v4/platformio.ini | 2 + variants/lilygo_t3s3/platformio.ini | 1 + variants/lilygo_tbeam_1w/platformio.ini | 1 + variants/lilygo_tbeam_SX1262/platformio.ini | 1 + variants/lilygo_tbeam_SX1276/platformio.ini | 1 + .../platformio.ini | 1 + variants/rak3112/platformio.ini | 1 + variants/station_g2/platformio.ini | 1 + variants/xiao_s3_wio/platformio.ini | 1 + 14 files changed, 639 insertions(+), 4 deletions(-) diff --git a/examples/simple_room_server/MyMesh.cpp b/examples/simple_room_server/MyMesh.cpp index 4f2e6d82c..1fc1ef88a 100644 --- a/examples/simple_room_server/MyMesh.cpp +++ b/examples/simple_room_server/MyMesh.cpp @@ -1,6 +1,9 @@ #include "MyMesh.h" +#include #include +#define ANON_REQ_TYPE_REGIONS 0x01 // client side of the anon-regions scope query (neighbors feature) + #define REPLY_DELAY_MILLIS 1500 #define PUSH_NOTIFY_DELAY_MILLIS 2000 #define SYNC_PUSH_INTERVAL 1200 @@ -416,7 +419,21 @@ 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) + if (neighbor_discover_active) { + for (int i = 0; i < neighbor_discover_count && n < MAX_CLIENTS; i++) { + auto& nb = neighbours[neighbor_discover[i].neighbour_idx]; + // ACL clients already have a matching peer entry and shared secret. Adding + // a second overlay entry would decrypt first and intercept their normal + // CLI/request traffic for the duration of discovery. + 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) } @@ -426,6 +443,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); @@ -437,11 +463,34 @@ void MyMesh::getPeerSharedSecret(uint8_t *dest_secret, int peer_idx) { 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; } auto 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_TXT_MSG && len > 5) { // a CLI command or new Post uint32_t sender_timestamp; memcpy(&sender_timestamp, data, 4); // timestamp (by sender's RTC clock - which could be wrong) @@ -631,6 +680,107 @@ void MyMesh::onAckRecv(mesh::Packet *packet, uint32_t ack_crc) { } } +#if defined(WITH_MQTT_NEIGHBORS) + +#define CTL_TYPE_NODE_DISCOVER_REQ 0x80 +#define CTL_TYPE_NODE_DISCOVER_RESP 0x90 + +void MyMesh::putNeighbour(const mesh::Identity &id, uint32_t timestamp, float snr) { + // find existing neighbour, else use least recently updated + uint32_t oldest_timestamp = 0xFFFFFFFF; + NeighbourInfo *neighbour = &neighbours[0]; + for (int i = 0; i < MAX_NEIGHBOURS; i++) { + // if neighbour already known, we should update it + if (id.matches(neighbours[i].id)) { + neighbour = &neighbours[i]; + break; + } + + // otherwise we should update the least recently updated neighbour + if (neighbours[i].heard_timestamp < oldest_timestamp) { + neighbour = &neighbours[i]; + oldest_timestamp = neighbour->heard_timestamp; + } + } + + // update neighbour info + neighbour->id = id; + neighbour->advert_timestamp = timestamp; + neighbour->heard_timestamp = getRTCClock()->getCurrentTime(); + neighbour->snr = (int8_t)(snr * 4); +} + +static bool isShare(const mesh::Packet *packet) { + if (packet->hasTransportCodes()) { + return packet->transport_codes[0] == 0 && packet->transport_codes[1] == 0; // codes { 0, 0 } means 'send to nowhere' + } + return false; +} + +void MyMesh::onAdvertRecv(mesh::Packet *packet, const mesh::Identity &id, uint32_t timestamp, + const uint8_t *app_data, size_t app_data_len) { + mesh::Mesh::onAdvertRecv(packet, id, timestamp, app_data, app_data_len); // chain to super impl + + // if this a zero hop advert (and not via 'Share'), add it to neighbours + if (packet->getPathHashCount() == 0 && !isShare(packet)) { + AdvertDataParser parser(app_data, app_data_len); + if (parser.isValid() && parser.getType() == ADV_TYPE_REPEATER) { // just keep neigbouring Repeaters + putNeighbour(id, timestamp, packet->getSNR()); + } + } +} + +void MyMesh::onControlDataRecv(mesh::Packet* packet) { + uint8_t type = packet->payload[0] & 0xF0; // just test upper 4 bits + // A room server is ADV_TYPE_ROOM, so it does NOT answer node-discover requests + // (those filter for repeaters). It only records repeater responses to its own + // discovery, to build the neighbour table. + if (type == CTL_TYPE_NODE_DISCOVER_RESP && packet->payload_len >= 6) { + uint8_t node_type = packet->payload[0] & 0x0F; + if (node_type != ADV_TYPE_REPEATER) { + return; + } + if (packet->payload_len < 6 + PUB_KEY_SIZE) { + MESH_DEBUG_PRINTLN("onControlDataRecv: DISCOVER_RESP pubkey too short: %d", (uint32_t)packet->payload_len); + return; + } + + if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) { + pending_discover_tag = 0; + return; + } + uint32_t tag; + memcpy(&tag, &packet->payload[2], 4); + if (tag != pending_discover_tag) { + return; + } + + mesh::Identity id(&packet->payload[6]); + if (id.matches(self_id)) { + return; + } + putNeighbour(id, getRTCClock()->getCurrentTime(), packet->getSNR()); + } +} + +void MyMesh::sendNodeDiscoverReq() { + uint8_t data[10]; + data[0] = CTL_TYPE_NODE_DISCOVER_REQ; // prefix_only=0 + data[1] = (1 << ADV_TYPE_REPEATER); + getRNG()->random(&data[2], 4); // tag + memcpy(&pending_discover_tag, &data[2], 4); + pending_discover_until = futureMillis(60000); + uint32_t since = 0; + memcpy(&data[6], &since, 4); + + auto pkt = createControlData(data, sizeof(data)); + if (pkt) { + sendZeroHop(pkt); + } +} + +#endif // WITH_MQTT_NEIGHBORS + 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, *createObserverPacketManager(32), tables), @@ -702,6 +852,19 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc memset(posts, 0, sizeof(posts)); _num_posted = _num_post_pushes = 0; +#if defined(WITH_MQTT_NEIGHBORS) + pending_discover_tag = 0; + pending_discover_until = 0; + 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; + memset(neighbours, 0, sizeof(neighbours)); +#endif + memset(default_scope.key, 0, sizeof(default_scope.key)); } @@ -946,6 +1109,65 @@ void MyMesh::clearStats() { ((SimpleMeshTables *)getTables())->resetStats(); } +void MyMesh::formatNeighborsReply(char *reply) { +#if defined(WITH_MQTT_NEIGHBORS) + char *dp = reply; + + // create copy of neighbours list, skipping empty entries so we can sort it separately from main list + int16_t neighbours_count = 0; + NeighbourInfo* sorted_neighbours[MAX_NEIGHBOURS]; + for (int i = 0; i < MAX_NEIGHBOURS; i++) { + auto neighbour = &neighbours[i]; + if (neighbour->heard_timestamp > 0) { + sorted_neighbours[neighbours_count] = neighbour; + neighbours_count++; + } + } + + // sort neighbours newest to oldest + std::sort(sorted_neighbours, sorted_neighbours + neighbours_count, [](const NeighbourInfo* a, const NeighbourInfo* b) { + return a->heard_timestamp > b->heard_timestamp; // desc + }); + + for (int i = 0; i < neighbours_count && dp - reply < 134; i++) { + NeighbourInfo *neighbour = sorted_neighbours[i]; + + // add new line if not first item + if (i > 0) *dp++ = '\n'; + + char hex[10]; + // get 4 bytes of neighbour id as hex + mesh::Utils::toHex(hex, neighbour->id.pub_key, 4); + + // add next neighbour + uint32_t secs_ago = getRTCClock()->getCurrentTime() - neighbour->heard_timestamp; + sprintf(dp, "%s:%d:%d", hex, secs_ago, neighbour->snr); + while (*dp) + dp++; // find end of string + } + if (dp == reply) { // no neighbours, need empty response + strcpy(dp, "-none-"); + dp += 6; + } + *dp = 0; // null terminator +#else + strcpy(reply, "not supported"); +#endif +} + +void MyMesh::removeNeighbor(const uint8_t *pubkey, int key_len) { +#if defined(WITH_MQTT_NEIGHBORS) + for (int i = 0; i < MAX_NEIGHBOURS; i++) { + NeighbourInfo *neighbour = &neighbours[i]; + if (memcmp(neighbour->id.pub_key, pubkey, key_len) == 0) { + neighbours[i] = NeighbourInfo(); // clear neighbour entry + } + } +#else + (void)pubkey; (void)key_len; +#endif +} + void MyMesh::formatStatsReply(char *reply) { StatsFormatHelper::formatCoreStats(reply, board, *_ms, _err_flags, _mgr); } @@ -1137,6 +1359,45 @@ void MyMesh::handleCommand(uint32_t sender_timestamp, char *command, char *reply Serial.printf("\n"); } reply[0] = 0; +#if defined(WITH_MQTT_NEIGHBORS) + } else if (memcmp(command, "discover.neighbors", 18) == 0) { + const char* sub = command + 18; + while (*sub == ' ') sub++; + if (*sub != 0) { + strcpy(reply, "Err - discover.neighbors has no options"); + } else { + sendNodeDiscoverReq(); + strcpy(reply, "OK - Discover sent"); + } + } 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 } @@ -1249,4 +1510,316 @@ void MyMesh::loop() { #ifdef WITH_MQTT_BRIDGE _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 } + +#if defined(WITH_MQTT_NEIGHBORS) +#include "helpers/MQTTMessageBuilder.h" +#if defined(ESP_PLATFORM) +#include +#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; +} + +#if defined(ESP_PLATFORM) +// ArduinoJson v7 JsonDocument has no real capacity cap (DynamicJsonDocument(N) +// is a no-op shim). Keep the pool off internal DRAM and soft-cap peak growth to +// the publish buffer size. used only rises on allocate — conservative for this +// single-shot doc (overflow path removes+breaks, so no further growth after free). +struct NeighborsDocAllocator : ArduinoJson::Allocator { + size_t used = 0; + static const size_t kBudget = MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE; + + void* allocate(size_t size) override { + if (used >= kBudget || size > kBudget - used) return nullptr; + void* p = heap_caps_malloc(size, MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT); + if (p) used += size; + return p; + } + + void deallocate(void* ptr) override { + heap_caps_free(ptr); + } + + void* reallocate(void* ptr, size_t new_size) override { + size_t old_size = ptr ? heap_caps_get_allocated_size(ptr) : 0; + size_t next_used = (used >= old_size) ? (used - old_size) : 0; + if (next_used >= kBudget || new_size > kBudget - next_used) return nullptr; + void* p = heap_caps_realloc(ptr, new_size, MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT); + if (p) used = next_used + new_size; + return p; + } +}; +#endif + +// 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; + } + +#if defined(ESP_PLATFORM) + NeighborsDocAllocator doc_alloc; + JsonDocument doc(&doc_alloc); +#else + JsonDocument doc; +#endif + 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 diff --git a/examples/simple_room_server/MyMesh.h b/examples/simple_room_server/MyMesh.h index 3e18b2b37..747820414 100644 --- a/examples/simple_room_server/MyMesh.h +++ b/examples/simple_room_server/MyMesh.h @@ -95,6 +95,13 @@ struct PostInfo { char text[MAX_POST_TEXT_LEN+1]; }; +struct NeighbourInfo { + mesh::Identity id; + uint32_t advert_timestamp; + uint32_t heard_timestamp; + int8_t snr; // multiplied by 4, user should divide to get float value +}; + class MyMesh : public mesh::Mesh, public CommonCLICallbacks #ifdef WITH_WEBCONFIG , public WebConfigServer::Callbacks @@ -128,6 +135,43 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks uint8_t pending_sf; uint8_t pending_cr; int matching_peer_indexes[MAX_CLIENTS]; +#if defined(WITH_MQTT_NEIGHBORS) + NeighbourInfo neighbours[MAX_NEIGHBOURS]; + uint32_t pending_discover_tag; + unsigned long pending_discover_until; + 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; + 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]; + + void putNeighbour(const mesh::Identity& id, uint32_t timestamp, float snr); + void sendNodeDiscoverReq(); + 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); + static const int NEIGHBOR_DISCOVER_PEER_BASE = 1000; + static const unsigned long NEIGHBOR_DISCOVER_TIMEOUT_MS = 30000; +#endif #ifdef WITH_MQTT_BRIDGE MQTTBridge* bridge; #endif @@ -187,6 +231,10 @@ protected: void onPeerDataRecv(mesh::Packet* packet, uint8_t type, int sender_idx, const uint8_t* secret, uint8_t* data, size_t len) override; bool onPeerPathRecv(mesh::Packet* packet, int sender_idx, const uint8_t* secret, uint8_t* path, uint8_t path_len, uint8_t extra_type, uint8_t* extra, uint8_t extra_len) override; void onAckRecv(mesh::Packet* packet, uint32_t ack_crc) override; +#if defined(WITH_MQTT_NEIGHBORS) + void onAdvertRecv(mesh::Packet* packet, const mesh::Identity& id, uint32_t timestamp, const uint8_t* app_data, size_t app_data_len) override; + void onControlDataRecv(mesh::Packet* packet) override; +#endif #if ENV_INCLUDE_GPS == 1 void applyGpsPrefs() { @@ -237,9 +285,8 @@ public: void dumpLogFile() override; void setTxPower(int8_t power_dbm) override; - void formatNeighborsReply(char *reply) override { - strcpy(reply, "not supported"); - } + void formatNeighborsReply(char *reply) override; + void removeNeighbor(const uint8_t* pubkey, int key_len) override; void formatStatsReply(char *reply) override; void formatRadioStatsReply(char *reply) override; void formatPacketStatsReply(char *reply) override; diff --git a/variants/heltec_t190/platformio.ini b/variants/heltec_t190/platformio.ini index 11bfaec6f..9b0db55ea 100644 --- a/variants/heltec_t190/platformio.ini +++ b/variants/heltec_t190/platformio.ini @@ -204,6 +204,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/heltec_tracker_v2/platformio.ini b/variants/heltec_tracker_v2/platformio.ini index 3571de2ac..b5e127745 100644 --- a/variants/heltec_tracker_v2/platformio.ini +++ b/variants/heltec_tracker_v2/platformio.ini @@ -257,6 +257,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 @@ -293,6 +294,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/heltec_v3/platformio.ini b/variants/heltec_v3/platformio.ini index 5c29fa2c4..49385eaef 100644 --- a/variants/heltec_v3/platformio.ini +++ b/variants/heltec_v3/platformio.ini @@ -203,6 +203,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 @@ -540,6 +541,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/heltec_v4/platformio.ini b/variants/heltec_v4/platformio.ini index 3facd287a..c3b02b530 100644 --- a/variants/heltec_v4/platformio.ini +++ b/variants/heltec_v4/platformio.ini @@ -315,6 +315,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 @@ -354,6 +355,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/lilygo_t3s3/platformio.ini b/variants/lilygo_t3s3/platformio.ini index 446307e3b..577be0241 100644 --- a/variants/lilygo_t3s3/platformio.ini +++ b/variants/lilygo_t3s3/platformio.ini @@ -161,6 +161,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/lilygo_tbeam_1w/platformio.ini b/variants/lilygo_tbeam_1w/platformio.ini index 6849018e2..53078e1f7 100644 --- a/variants/lilygo_tbeam_1w/platformio.ini +++ b/variants/lilygo_tbeam_1w/platformio.ini @@ -255,6 +255,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D PERSISTANT_GPS=1 diff --git a/variants/lilygo_tbeam_SX1262/platformio.ini b/variants/lilygo_tbeam_SX1262/platformio.ini index 6d381a9d7..56208b622 100644 --- a/variants/lilygo_tbeam_SX1262/platformio.ini +++ b/variants/lilygo_tbeam_SX1262/platformio.ini @@ -203,6 +203,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/lilygo_tbeam_SX1276/platformio.ini b/variants/lilygo_tbeam_SX1276/platformio.ini index 048214711..46bf131fd 100644 --- a/variants/lilygo_tbeam_SX1276/platformio.ini +++ b/variants/lilygo_tbeam_SX1276/platformio.ini @@ -202,6 +202,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/lilygo_tbeam_supreme_SX1262/platformio.ini b/variants/lilygo_tbeam_supreme_SX1262/platformio.ini index 53871835c..9ca87c0b8 100644 --- a/variants/lilygo_tbeam_supreme_SX1262/platformio.ini +++ b/variants/lilygo_tbeam_supreme_SX1262/platformio.ini @@ -162,6 +162,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/rak3112/platformio.ini b/variants/rak3112/platformio.ini index b1b575256..6869a2977 100644 --- a/variants/rak3112/platformio.ini +++ b/variants/rak3112/platformio.ini @@ -176,6 +176,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/station_g2/platformio.ini b/variants/station_g2/platformio.ini index d64ce398b..f149f1c42 100644 --- a/variants/station_g2/platformio.ini +++ b/variants/station_g2/platformio.ini @@ -187,6 +187,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1 diff --git a/variants/xiao_s3_wio/platformio.ini b/variants/xiao_s3_wio/platformio.ini index 8e3ec15d3..09fab2fb9 100644 --- a/variants/xiao_s3_wio/platformio.ini +++ b/variants/xiao_s3_wio/platformio.ini @@ -146,6 +146,7 @@ build_flags = -D ADMIN_PASSWORD='"password"' -D ROOM_PASSWORD='"hello"' -D WITH_MQTT_BRIDGE=1 + -D MAX_NEIGHBOURS=50 -D MAX_MQTT_BROKERS=3 -D MQTT_MAX_PACKET_SIZE=1024 -D MQTT_DEBUG=1