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