dandri/MeshCore-mqtt-observer
1
0
Fork 0
mirror of https://github.com/agessaman/MeshCore.git synced 2026-08-29 09:48:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): refactor message building to use MQTTPayloadBuilder

Browse Source 
Replace direct JSON construction in MQTTMessageBuilder with calls to
MQTTPayloadBuilder for building status, packet, and raw messages.
This change improves code maintainability and reduces duplication by
centralizing message formatting logic. Additionally, update platformio.ini
to include ArduinoJson dependency for JSON handling.
This commit is contained in:
agessaman
2026-07-18 15:47:11 -07:00
parent b7b3b4dcf1
commit 26a8a3cb58
27 changed files with 1514 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platformio.ini
+2
View File
@@ -172,5 +172,7 @@ test_build_src = yes
build_src_filter =
-<*>
+<../src/Utils.cpp>
+<../src/helpers/MQTTPayloadBuilder.cpp>
lib_deps =
google/googletest @ 1.17.0
bblanchon/ArduinoJson @ 7.4.3
src/helpers/MQTTConnectionPolicy.h
+135
View File
@@ -0,0 +1,135 @@
#pragma once
#include <stdint.h>
// Pure timing and state-transition policy used by MQTTBridge's connection
// maintenance loop. Keeping these decisions independent of Arduino, WiFi, and
// the MQTT client lets host tests exercise the exact production policy with a
// deterministic clock.
namespace MQTTConnectionPolicy {
static const uint32_t kReconnectGuardMs = 15000UL;
static const uint32_t kStableResetMs = 120000UL;
static const uint32_t kCircuitBreakerProbeMs = 1800000UL;
static const uint32_t kRenewalThrottleMs = 60000UL;
static const uint32_t kSlotStaggerMs = 3000UL;
static const uint8_t kMaxFailuresAtMaxBackoff = 3;
static const uint32_t kDefaultJwtLifetimeSecs = 86400UL;
static const uint32_t kMaxJwtStaggerSecs = 300UL;
static const uint32_t kMinimumValidEpoch = 1000000000UL;
static const uint32_t kJwtClockThreshold = 1735689600UL; // 2025-01-01 UTC
// Unsigned subtraction is intentionally used: it is the standard millis()
// idiom and remains correct across a single 32-bit counter rollover.
static inline uint32_t elapsedMs(uint32_t now, uint32_t then) {
return now - then;
}
static inline bool reconnectGuardActive(uint32_t now, uint32_t last_reconnect) {
return elapsedMs(now, last_reconnect) < kReconnectGuardMs;
}
static inline bool stableConnection(uint32_t now, uint32_t connected_at) {
return connected_at != 0 && elapsedMs(now, connected_at) >= kStableResetMs;
}
static inline uint32_t reconnectBackoffMs(uint8_t reconnect_backoff) {
static const uint32_t kBackoffMs[] = {
10000UL, 30000UL, 60000UL, 120000UL, 300000UL
};
const uint8_t index = reconnect_backoff < 5 ? reconnect_backoff : 4;
return kBackoffMs[index];
}
static inline uint32_t reconnectDelayMs(uint8_t reconnect_backoff, uint8_t slot_index) {
return reconnectBackoffMs(reconnect_backoff) +
static_cast<uint32_t>(slot_index) * kSlotStaggerMs;
}
static inline bool reconnectDue(uint32_t now, uint32_t last_attempt,
uint8_t reconnect_backoff, uint8_t slot_index) {
return elapsedMs(now, last_attempt) >= reconnectDelayMs(reconnect_backoff, slot_index);
}
struct BackoffAdvance {
uint8_t reconnect_backoff;
uint8_t max_backoff_failures;
bool circuit_breaker_tripped;
bool should_reconnect;
};
// Advance the ladder immediately before a due reconnect. The first visit to
// the 300-second rung changes level 4 to the saturated marker 5. Three later
// failures at that rung trip the breaker; the third does not launch another
// connection attempt.
static inline BackoffAdvance advanceBackoff(uint8_t reconnect_backoff,
uint8_t max_backoff_failures) {
BackoffAdvance result = {
reconnect_backoff, max_backoff_failures, false, true
};
if (result.reconnect_backoff < 5) {
result.reconnect_backoff++;
return result;
}
if (result.max_backoff_failures < UINT8_MAX) {
result.max_backoff_failures++;
}
if (result.max_backoff_failures >= kMaxFailuresAtMaxBackoff) {
result.circuit_breaker_tripped = true;
result.should_reconnect = false;
}
return result;
}
static inline bool circuitBreakerProbeDue(uint32_t now, uint32_t last_attempt) {
return elapsedMs(now, last_attempt) >= kCircuitBreakerProbeMs;
}
// Each later slot expires up to five percent of the base lifetime earlier,
// capped at five minutes per slot. Runtime slot indexes are bounded by the
// persisted MQTT slot count; the final clamp also prevents underflow if this
// helper is used with unexpected input.
static inline uint32_t jwtLifetimeSecs(uint32_t base_lifetime, uint8_t slot_index) {
uint32_t per_slot_stagger = base_lifetime / 20UL;
if (per_slot_stagger > kMaxJwtStaggerSecs) {
per_slot_stagger = kMaxJwtStaggerSecs;
}
uint64_t stagger = static_cast<uint64_t>(slot_index) * per_slot_stagger;
if (stagger > base_lifetime) {
stagger = base_lifetime;
}
return base_lifetime - static_cast<uint32_t>(stagger);
}
static inline uint32_t renewalBufferSecs(uint32_t lifetime_secs) {
uint32_t buffer = lifetime_secs / 10UL;
if (buffer < 60UL) buffer = 60UL;
if (buffer > 300UL) buffer = 300UL;
return buffer;
}
static inline bool tokenNeedsRenewal(bool time_synced, uint32_t current_time,
uint32_t token_expires_at,
uint32_t renewal_buffer_secs) {
if (!time_synced) {
return token_expires_at == 0;
}
if (token_expires_at < kMinimumValidEpoch) {
return true;
}
if (current_time >= token_expires_at) {
return true;
}
return current_time >= token_expires_at - renewal_buffer_secs;
}
static inline bool renewalAttemptAllowed(uint32_t now, uint32_t last_attempt) {
return elapsedMs(now, last_attempt) >= kRenewalThrottleMs;
}
static inline bool jwtClockAvailable(bool ntp_synced, uint32_t current_time) {
return ntp_synced || current_time >= kJwtClockThreshold;
}
} // namespace MQTTConnectionPolicy
src/helpers/MQTTMessageBuilder.cpp
+13 -138
View File
@@ -1,4 +1,5 @@
#include "MQTTMessageBuilder.h"
#include "MQTTPayloadBuilder.h"
#include <ArduinoJson.h>
#include <cstring>
#include <math.h>
@@ -53,66 +54,11 @@ int MQTTMessageBuilder::buildStatusMessage(
int packets_received,
const char* repeat
) {
// doc is provided by the caller (heap-allocated DynamicJsonDocument in MQTTBridge),
// keeping this 768-byte scratch space off the MQTT task stack.
doc.clear();
JsonObject root = doc.to<JsonObject>();
root["status"] = status;
root["timestamp"] = timestamp;
root["origin"] = origin;
root["origin_id"] = origin_id;
root["model"] = model;
root["firmware_version"] = firmware_version;
root["radio"] = radio;
root["client_version"] = client_version;
if (repeat != nullptr) {
root["repeat"] = repeat;
}
// Add stats object if any stats are provided
if (battery_mv >= 0 || uptime_secs >= 0 || errors >= 0 || queue_len >= 0 ||
noise_floor > -999 || tx_air_secs >= 0 || rx_air_secs >= 0 || recv_errors >= 0 ||
internal_heap >= 0 || packets_sent >= 0 || packets_received >= 0) {
JsonObject stats = root.createNestedObject("stats");
if (battery_mv >= 0) {
stats["battery_mv"] = battery_mv;
}
if (uptime_secs >= 0) {
stats["uptime_secs"] = uptime_secs;
}
if (packets_sent >= 0) {
stats["packets_sent"] = packets_sent;
}
if (packets_received >= 0) {
stats["packets_received"] = packets_received;
}
if (errors >= 0) {
stats["errors"] = errors;
}
if (queue_len >= 0) {
stats["queue_len"] = queue_len;
}
if (noise_floor > -999) {
stats["noise_floor"] = noise_floor;
}
if (tx_air_secs >= 0) {
stats["tx_air_secs"] = tx_air_secs;
}
if (rx_air_secs >= 0) {
stats["rx_air_secs"] = rx_air_secs;
}
if (recv_errors >= 0) {
stats["recv_errors"] = recv_errors;
}
if (internal_heap >= 0) {
stats["internal_heap"] = internal_heap;
}
}
size_t len = serializeJson(root, buffer, buffer_size);
return (len > 0 && len < buffer_size) ? len : 0;
return MQTTPayloadBuilder::buildStatusMessage(
doc, origin, origin_id, model, firmware_version, radio, client_version,
status, timestamp, buffer, buffer_size, battery_mv, uptime_secs, errors,
queue_len, noise_floor, tx_air_secs, rx_air_secs, recv_errors, internal_heap,
packets_sent, packets_received, repeat);
}
int MQTTMessageBuilder::buildPacketMessage(
@@ -138,71 +84,10 @@ int MQTTMessageBuilder::buildPacketMessage(
char* buffer,
size_t buffer_size
) {
// doc is provided by the caller (heap-allocated DynamicJsonDocument in MQTTBridge),
// keeping this 2048-byte scratch space off the MQTT task stack.
doc.clear();
JsonObject root = doc.to<JsonObject>();
// Format numeric values as strings to avoid String object allocations
char len_str[16];
char packet_type_str[16];
char payload_len_str[16];
char snr_str[16];
char rssi_str[16];
char score_str[16];
snprintf(len_str, sizeof(len_str), "%d", len);
snprintf(packet_type_str, sizeof(packet_type_str), "%d", packet_type);
snprintf(payload_len_str, sizeof(payload_len_str), "%d", payload_len);
snprintf(snr_str, sizeof(snr_str), "%.1f", snr);
snprintf(rssi_str, sizeof(rssi_str), "%d", rssi);
root["timestamp"] = timestamp;
root["hash"] = hash;
root["origin"] = origin;
root["type"] = "PACKET";
root["direction"] = direction;
root["time"] = time;
root["date"] = date;
root["len"] = len_str;
root["packet_type"] = packet_type_str;
root["route"] = route;
root["payload_len"] = payload_len_str;
root["raw"] = raw;
root["origin_id"] = origin_id;
// SNR and RSSI are only meaningful for RX packets (received from radio)
if (strcmp(direction, "rx") == 0) {
root["SNR"] = snr_str;
root["RSSI"] = rssi_str;
// Firmware's rebroadcast "score" for this RX packet, scaled x1000 to match the
// integer form printed in the serial RX log (see Dispatcher::checkRecv()).
if (!isnan(score)) {
snprintf(score_str, sizeof(score_str), "%d", (int)(score * 1000));
root["score"] = score_str;
}
}
// Routing path as an array of lowercase hex hop tokens, one element per hop
// (e.g. ["aa","bb","cc"], or ["aaaa","bbbb"] for multi-byte hashes). This matches
// meshcore-packet-capture's _split_path_hops() representation.
if (path_bytes && path_hop_count > 0 && path_hash_size > 0) {
JsonArray path_arr = root.createNestedArray("path");
char hop_hex[2 * 4 + 1]; // hop hash is 1-4 bytes -> up to 8 hex chars + null
for (int i = 0; i < path_hop_count; i++) {
size_t pos = 0;
for (int b = 0; b < path_hash_size && b < 4; b++) {
size_t idx = (size_t)i * path_hash_size + b;
if (idx >= MAX_PATH_SIZE) break;
snprintf(hop_hex + pos, 3, "%02x", path_bytes[idx]);
pos += 2;
}
hop_hex[pos] = '\0';
path_arr.add(hop_hex); // char[] (non-const) -> ArduinoJson copies the string
}
}
size_t json_len = serializeJson(root, buffer, buffer_size);
return (json_len > 0 && json_len < buffer_size) ? json_len : 0;
return MQTTPayloadBuilder::buildPacketMessage(
doc, origin, origin_id, timestamp, direction, time, date, len, packet_type,
route, payload_len, raw, snr, rssi, score, hash, path_bytes, path_hop_count,
path_hash_size, MAX_PATH_SIZE, buffer, buffer_size);
}
int MQTTMessageBuilder::buildRawMessage(
@@ -213,18 +98,8 @@ int MQTTMessageBuilder::buildRawMessage(
char* buffer,
size_t buffer_size
) {
// Use StaticJsonDocument to avoid heap fragmentation (fixed-size stack allocation)
StaticJsonDocument<512> doc;
JsonObject root = doc.to<JsonObject>();
root["origin"] = origin;
root["origin_id"] = origin_id;
root["timestamp"] = timestamp;
root["type"] = "RAW";
root["data"] = raw;
size_t len = serializeJson(root, buffer, buffer_size);
return (len > 0 && len < buffer_size) ? len : 0;
return MQTTPayloadBuilder::buildRawMessage(
origin, origin_id, timestamp, raw, buffer, buffer_size);
}
int MQTTMessageBuilder::buildPacketJSON(
@@ -435,4 +310,4 @@ void MQTTMessageBuilder::packetToHex(mesh::Packet* packet, char* hex, size_t hex
// Convert serialized packet to hex
bytesToHex(raw_buf, raw_len, hex, hex_size);
}
}
src/helpers/MQTTPacketQueuePolicy.h
+105
View File
@@ -0,0 +1,105 @@
#pragma once
#include <stddef.h>
#include <stdint.h>
// Pure queue/backpressure policy shared by the FreeRTOS and circular-buffer
// MQTT packet queues. Keeping the timing and retry decisions here makes the
// production behavior deterministic under host tests without mocking either
// queue implementation or the MQTT client.
namespace MQTTPacketQueuePolicy {
static const uint32_t kDisconnectedStaleMs = 300000UL;
static const size_t kBacklogThreshold = 5;
static const uint8_t kGentleDrainCount = 1;
static const uint8_t kBurstDrainCount = 5;
static const uint32_t kGentleDrainBudgetMs = 30UL;
static const uint32_t kBurstDrainBudgetMs = 100UL;
static const uint8_t kMaxQos0RetryAttempts = 3;
static const uint32_t kRetryDelayBaseMs = 300UL;
static const uint32_t kRetryDelayJitterMs = 200UL;
// Unsigned subtraction is the standard millis() idiom and remains correct
// across one 32-bit counter rollover.
static inline uint32_t elapsedMs(uint32_t now, uint32_t then) {
return now - then;
}
enum class EnqueueAction : uint8_t {
Enqueue,
EvictOldestThenEnqueue,
Reject
};
static inline EnqueueAction enqueueAction(size_t queue_count, size_t capacity) {
if (capacity == 0) return EnqueueAction::Reject;
return queue_count >= capacity
? EnqueueAction::EvictOldestThenEnqueue
: EnqueueAction::Enqueue;
}
// disconnected_since == 0 means tracking has not started. The bridge records
// the first disconnected observation and asks this helper on later cycles.
static inline bool shouldFlushDisconnected(uint32_t now,
uint32_t disconnected_since,
uint32_t stale_ms = kDisconnectedStaleMs) {
return disconnected_since != 0 && elapsedMs(now, disconnected_since) >= stale_ms;
}
struct DrainBudget {
uint8_t max_packets;
uint32_t max_time_ms;
};
static inline DrainBudget drainBudget(size_t queue_count) {
if (queue_count > kBacklogThreshold) {
return {kBurstDrainCount, kBurstDrainBudgetMs};
}
return {kGentleDrainCount, kGentleDrainBudgetMs};
}
static inline bool drainTimeAvailable(uint32_t now, uint32_t started_at,
uint32_t budget_ms) {
// Preserve the bridge's inclusive boundary: work may begin at exactly the
// configured limit, but not one millisecond later.
return elapsedMs(now, started_at) <= budget_ms;
}
// retry_attempts distinguishes an unscheduled packet from a scheduled retry
// whose deadline wrapped to exactly zero. Deadlines are always less than
// 500 ms away, so the half-range comparison is unambiguous.
static inline bool retryReady(uint32_t now, uint32_t next_retry_ms,
uint8_t retry_attempts) {
if (retry_attempts == 0) return true;
return elapsedMs(now, next_retry_ms) < 0x80000000UL;
}
enum class RetryAction : uint8_t {
Complete,
Schedule,
Drop
};
struct RetryDecision {
RetryAction action;
uint8_t retry_attempts;
uint32_t delay_ms;
uint32_t next_retry_ms;
};
static inline RetryDecision retryDecision(bool any_published,
uint8_t retry_attempts,
uint32_t now) {
if (any_published) {
return {RetryAction::Complete, retry_attempts, 0, 0};
}
if (retry_attempts >= kMaxQos0RetryAttempts) {
return {RetryAction::Drop, retry_attempts, 0, 0};
}
const uint32_t delay = kRetryDelayBaseMs + (now % kRetryDelayJitterMs);
return {RetryAction::Schedule, static_cast<uint8_t>(retry_attempts + 1),
delay, now + delay};
}
} // namespace MQTTPacketQueuePolicy
src/helpers/MQTTPayloadBuilder.cpp
+186
View File
@@ -0,0 +1,186 @@
#include "MQTTPayloadBuilder.h"
#include <math.h>
#include <stdio.h>
#include <string.h>
namespace {
static int serializeComplete(JsonObject root, char* buffer, size_t buffer_size) {
if (!buffer || buffer_size == 0) return 0;
size_t written = serializeJson(root, buffer, buffer_size);
// Preserve MQTTMessageBuilder's existing success criterion while clearing
// ArduinoJson's truncated prefix on failure. Callers publish only a positive
// return value, and now a failed buffer cannot be mistaken for complete JSON.
if (written == 0 || written >= buffer_size) {
buffer[0] = '\0';
return 0;
}
return static_cast<int>(written);
}
} // namespace
int MQTTPayloadBuilder::buildStatusMessage(
JsonDocument& doc,
const char* origin,
const char* origin_id,
const char* model,
const char* firmware_version,
const char* radio,
const char* client_version,
const char* status,
const char* timestamp,
char* buffer,
size_t buffer_size,
int battery_mv,
int uptime_secs,
int errors,
int queue_len,
int noise_floor,
int tx_air_secs,
int rx_air_secs,
int recv_errors,
int internal_heap,
int packets_sent,
int packets_received,
const char* repeat
) {
doc.clear();
JsonObject root = doc.to<JsonObject>();
root["status"] = status;
root["timestamp"] = timestamp;
root["origin"] = origin;
root["origin_id"] = origin_id;
root["model"] = model;
root["firmware_version"] = firmware_version;
root["radio"] = radio;
root["client_version"] = client_version;
if (repeat != nullptr) {
root["repeat"] = repeat;
}
if (battery_mv >= 0 || uptime_secs >= 0 || errors >= 0 || queue_len >= 0 ||
noise_floor > -999 || tx_air_secs >= 0 || rx_air_secs >= 0 || recv_errors >= 0 ||
internal_heap >= 0 || packets_sent >= 0 || packets_received >= 0) {
JsonObject stats = root["stats"].to<JsonObject>();
if (battery_mv >= 0) stats["battery_mv"] = battery_mv;
if (uptime_secs >= 0) stats["uptime_secs"] = uptime_secs;
if (packets_sent >= 0) stats["packets_sent"] = packets_sent;
if (packets_received >= 0) stats["packets_received"] = packets_received;
if (errors >= 0) stats["errors"] = errors;
if (queue_len >= 0) stats["queue_len"] = queue_len;
if (noise_floor > -999) stats["noise_floor"] = noise_floor;
if (tx_air_secs >= 0) stats["tx_air_secs"] = tx_air_secs;
if (rx_air_secs >= 0) stats["rx_air_secs"] = rx_air_secs;
if (recv_errors >= 0) stats["recv_errors"] = recv_errors;
if (internal_heap >= 0) stats["internal_heap"] = internal_heap;
}
return serializeComplete(root, buffer, buffer_size);
}
int MQTTPayloadBuilder::buildPacketMessage(
JsonDocument& doc,
const char* origin,
const char* origin_id,
const char* timestamp,
const char* direction,
const char* time,
const char* date,
int len,
int packet_type,
const char* route,
int payload_len,
const char* raw,
float snr,
int rssi,
float score,
const char* hash,
const uint8_t* path_bytes,
int path_hop_count,
int path_hash_size,
size_t max_path_bytes,
char* buffer,
size_t buffer_size
) {
doc.clear();
JsonObject root = doc.to<JsonObject>();
char len_str[16];
char packet_type_str[16];
char payload_len_str[16];
char snr_str[16];
char rssi_str[16];
char score_str[16];
snprintf(len_str, sizeof(len_str), "%d", len);
snprintf(packet_type_str, sizeof(packet_type_str), "%d", packet_type);
snprintf(payload_len_str, sizeof(payload_len_str), "%d", payload_len);
snprintf(snr_str, sizeof(snr_str), "%.1f", snr);
snprintf(rssi_str, sizeof(rssi_str), "%d", rssi);
root["timestamp"] = timestamp;
root["hash"] = hash;
root["origin"] = origin;
root["type"] = "PACKET";
root["direction"] = direction;
root["time"] = time;
root["date"] = date;
root["len"] = len_str;
root["packet_type"] = packet_type_str;
root["route"] = route;
root["payload_len"] = payload_len_str;
root["raw"] = raw;
root["origin_id"] = origin_id;
if (direction && strcmp(direction, "rx") == 0) {
root["SNR"] = snr_str;
root["RSSI"] = rssi_str;
if (!isnan(score)) {
snprintf(score_str, sizeof(score_str), "%d", static_cast<int>(score * 1000));
root["score"] = score_str;
}
}
if (path_bytes && path_hop_count > 0 && path_hash_size > 0) {
JsonArray path_arr = root["path"].to<JsonArray>();
char hop_hex[2 * 4 + 1];
for (int i = 0; i < path_hop_count; i++) {
size_t pos = 0;
for (int b = 0; b < path_hash_size && b < 4; b++) {
size_t idx = static_cast<size_t>(i) * path_hash_size + b;
if (idx >= max_path_bytes) break;
snprintf(hop_hex + pos, 3, "%02x", path_bytes[idx]);
pos += 2;
}
hop_hex[pos] = '\0';
path_arr.add(hop_hex);
}
}
return serializeComplete(root, buffer, buffer_size);
}
int MQTTPayloadBuilder::buildRawMessage(
const char* origin,
const char* origin_id,
const char* timestamp,
const char* raw,
char* buffer,
size_t buffer_size
) {
JsonDocument doc;
JsonObject root = doc.to<JsonObject>();
root["origin"] = origin;
root["origin_id"] = origin_id;
root["timestamp"] = timestamp;
root["type"] = "RAW";
root["data"] = raw;
return serializeComplete(root, buffer, buffer_size);
}
src/helpers/MQTTPayloadBuilder.h
+72
View File
@@ -0,0 +1,72 @@
#pragma once
#include <ArduinoJson.h>
#include <stddef.h>
#include <stdint.h>
// Mesh-independent JSON serialization core for MQTT publication payloads.
// MQTTMessageBuilder keeps the firmware-facing API and delegates these three
// deterministic contracts here so they can be exercised by native tests.
class MQTTPayloadBuilder {
public:
static int buildStatusMessage(
JsonDocument& doc,
const char* origin,
const char* origin_id,
const char* model,
const char* firmware_version,
const char* radio,
const char* client_version,
const char* status,
const char* timestamp,
char* buffer,
size_t buffer_size,
int battery_mv = -1,
int uptime_secs = -1,
int errors = -1,
int queue_len = -1,
int noise_floor = -999,
int tx_air_secs = -1,
int rx_air_secs = -1,
int recv_errors = -1,
int internal_heap = -1,
int packets_sent = -1,
int packets_received = -1,
const char* repeat = nullptr
);
static int buildPacketMessage(
JsonDocument& doc,
const char* origin,
const char* origin_id,
const char* timestamp,
const char* direction,
const char* time,
const char* date,
int len,
int packet_type,
const char* route,
int payload_len,
const char* raw,
float snr,
int rssi,
float score,
const char* hash,
const uint8_t* path_bytes,
int path_hop_count,
int path_hash_size,
size_t max_path_bytes,
char* buffer,
size_t buffer_size
);
static int buildRawMessage(
const char* origin,
const char* origin_id,
const char* timestamp,
const char* raw,
char* buffer,
size_t buffer_size
);
};
src/helpers/MQTTTopicRouter.h
+85
View File
@@ -0,0 +1,85 @@
#pragma once
#include <stddef.h>
#include <stdio.h>
#include <string.h>
#include "MQTTObserverValidation.h"
#include "MQTTTopicTemplate.h"
// Pure MQTT publication-topic policy shared by MQTTBridge and the native tests.
// Keep these values aligned with MQTTBridge::MQTTMessageType; the bridge passes
// its enum value as an int so this helper stays independent of ESP/Arduino types.
enum MQTTPublicationType {
MQTT_PUBLICATION_STATUS = 0,
MQTT_PUBLICATION_PACKETS = 1,
MQTT_PUBLICATION_RAW = 2,
};
enum MQTTTopicRouteStyle {
MQTT_ROUTE_MESHCORE,
MQTT_ROUTE_MESHRANK,
MQTT_ROUTE_CUSTOM,
};
static inline bool mqttTopicSlotIndexValid(int index, size_t slot_count) {
return index >= 0 && (size_t)index < slot_count;
}
static inline const char* mqttPublicationTypeName(int type) {
switch (type) {
case MQTT_PUBLICATION_STATUS: return "status";
case MQTT_PUBLICATION_PACKETS: return "packets";
case MQTT_PUBLICATION_RAW: return "raw";
default: return NULL;
}
}
static inline bool mqttWriteTopic(char* buf, size_t buf_size, const char* format,
const char* first, const char* second,
const char* third) {
if (!buf || buf_size == 0 || !format || !first || !second || !third) return false;
buf[0] = '\0';
int written = snprintf(buf, buf_size, format, first, second, third);
return written > 0 && (size_t)written < buf_size;
}
// Build the complete topic for one publication. MeshRank is deliberately
// packets-only; status and raw are unsupported by the current broker contract.
// MeshCore routes require a configured IATA and device id. Custom templates may
// omit either placeholder, so their individual values are allowed to be empty.
static inline bool mqttBuildPublicationTopic(MQTTTopicRouteStyle style, int type,
const char* custom_template,
const char* iata, const char* device,
const char* token,
char* buf, size_t buf_size) {
if (!buf || buf_size == 0) return false;
buf[0] = '\0';
const char* type_name = mqttPublicationTypeName(type);
if (!type_name) return false;
switch (style) {
case MQTT_ROUTE_MESHCORE:
if (!mqttIataValid(iata) || strcmp(iata, "XXX") == 0 || !device || device[0] == '\0') {
return false;
}
return mqttWriteTopic(buf, buf_size, "meshcore/%s/%s/%s", iata, device, type_name);
case MQTT_ROUTE_MESHRANK:
if (type != MQTT_PUBLICATION_PACKETS || !token || token[0] == '\0' ||
!device || device[0] == '\0') {
return false;
}
return mqttWriteTopic(buf, buf_size, "meshrank/uplink/%s/%s/%s",
token, device, type_name);
case MQTT_ROUTE_CUSTOM:
return mqttSubstituteTopic(custom_template, iata, device, token, type_name,
buf, buf_size);
default:
return false;
}
}
src/helpers/bridges/MQTTBridge.cpp
+150 -121
View File
@@ -1,6 +1,8 @@
#include "MQTTBridge.h"
#include "../MQTTConnectionPolicy.h"
#include "../MQTTMessageBuilder.h"
#include "../MQTTTopicTemplate.h"
#include "../MQTTPacketQueuePolicy.h"
#include "../MQTTTopicRouter.h"
#include "../TxtDataHelpers.h"
#include <NTPClient.h>
#include <WiFiUdp.h>
@@ -1520,8 +1522,8 @@ void MQTTBridge::maintainSlotConnections() {
// JWT tokens require valid timestamps
unsigned long clock_sec = current_time;
bool clock_looks_set = (clock_sec >= 1735689600); // 2025-01-01 00:00:00 UTC
bool can_do_jwt = _ntp_synced || clock_looks_set;
bool can_do_jwt = MQTTConnectionPolicy::jwtClockAvailable(
_ntp_synced, static_cast<uint32_t>(clock_sec));
// Count connected slots to inform reconnect decisions
int connected_count = 0;
@@ -1534,8 +1536,8 @@ void MQTTBridge::maintainSlotConnections() {
// Time-based guard: block reconnects if any slot reconnected within the last 15 s,
// ensuring the previous TLS handshake (and its Core-0-expensive completion events)
// finish before the next slot begins its own handshake.
const unsigned long RECONNECT_GUARD_MS = 15000UL;
bool reconnect_attempted_this_cycle = (now_millis - _last_slot_reconnect_ms < RECONNECT_GUARD_MS);
bool reconnect_attempted_this_cycle = MQTTConnectionPolicy::reconnectGuardActive(
static_cast<uint32_t>(now_millis), static_cast<uint32_t>(_last_slot_reconnect_ms));
// Only allow one full teardown+setup per cycle to limit heap fragmentation
// when multiple slots fail simultaneously
bool teardown_attempted_this_cycle = false;
@@ -1564,11 +1566,10 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
// (worst case the 300 s rung / 30-minute breaker probes) instead of
// hammering full TLS handshakes at the 10 s rung — see the onConnect
// handler in initSlotClients() for why this doesn't happen on CONNACK.
static const unsigned long BACKOFF_STABLE_RESET_MS = 120000UL;
if (slot.connected &&
(slot.reconnect_backoff != 0 || slot.max_backoff_failures != 0) &&
slot.connected_at_ms != 0 &&
(now_millis - slot.connected_at_ms) >= BACKOFF_STABLE_RESET_MS) {
MQTTConnectionPolicy::stableConnection(static_cast<uint32_t>(now_millis),
static_cast<uint32_t>(slot.connected_at_ms))) {
MQTT_DEBUG_PRINTLN("MQTT%d stable for %lus - clearing reconnect backoff (was level %d)",
index + 1, (now_millis - slot.connected_at_ms) / 1000UL, slot.reconnect_backoff);
slot.reconnect_backoff = 0;
@@ -1580,23 +1581,19 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
(!slot.preset && slot.audience[0] != '\0');
if (slot_uses_jwt) {
// Renew (and below, reconnect) this many seconds before the token's exp
// claim. Scaled to the slot's token lifetime — see tokenRenewalBufferSecs
// claim. Scaled to the slot's token lifetime — see renewalBufferSecs()
// for why a flat 60 s lost the renewal race against brokers that enforce
// exp on live sessions (waev's 55-minute tokens).
const unsigned long renewal_buffer = tokenRenewalBufferSecs(slotTokenLifetime(index));
bool token_needs_renewal = false;
if (!time_synced) {
token_needs_renewal = (slot.token_expires_at == 0);
} else {
token_needs_renewal = (slot.token_expires_at == 0) ||
!(slot.token_expires_at >= 1000000000) ||
(current_time >= slot.token_expires_at) ||
(current_time >= (slot.token_expires_at - renewal_buffer));
}
const unsigned long renewal_buffer = MQTTConnectionPolicy::renewalBufferSecs(
static_cast<uint32_t>(slotTokenLifetime(index)));
bool token_needs_renewal = MQTTConnectionPolicy::tokenNeedsRenewal(
time_synced, static_cast<uint32_t>(current_time),
static_cast<uint32_t>(slot.token_expires_at),
static_cast<uint32_t>(renewal_buffer));
// Throttle renewal attempts to once per minute
const unsigned long RENEWAL_THROTTLE_MS = 60000;
bool can_attempt_renewal = (now_millis - slot.last_token_renewal) >= RENEWAL_THROTTLE_MS;
bool can_attempt_renewal = MQTTConnectionPolicy::renewalAttemptAllowed(
static_cast<uint32_t>(now_millis), static_cast<uint32_t>(slot.last_token_renewal));
if (token_needs_renewal && can_attempt_renewal) {
slot.last_token_renewal = now_millis;
@@ -1653,11 +1650,10 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
// Periodic probe for circuit-breaker-tripped slots (recovery from transient outages)
// Attempts a single reconnect every 30 minutes to see if the server has come back
if (slot.circuit_breaker_tripped && !reconnect_attempted) {
static const unsigned long CIRCUIT_BREAKER_PROBE_INTERVAL_MS = 1800000UL; // 30 minutes
unsigned long probe_elapsed = (now_millis >= slot.last_reconnect_attempt) ?
(now_millis - slot.last_reconnect_attempt) :
(ULONG_MAX - slot.last_reconnect_attempt + now_millis + 1);
if (probe_elapsed >= CIRCUIT_BREAKER_PROBE_INTERVAL_MS) {
unsigned long probe_elapsed = MQTTConnectionPolicy::elapsedMs(
static_cast<uint32_t>(now_millis), static_cast<uint32_t>(slot.last_reconnect_attempt));
if (MQTTConnectionPolicy::circuitBreakerProbeDue(
static_cast<uint32_t>(now_millis), static_cast<uint32_t>(slot.last_reconnect_attempt))) {
slot.last_reconnect_attempt = now_millis;
reconnect_attempted = true;
_last_slot_reconnect_ms = now_millis;
@@ -1686,24 +1682,18 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
// Reconnect with exponential backoff (for disconnected slots that already have valid config)
// Only one reconnect per maintenance cycle to prevent TLS handshakes from blocking other slots
if (!slot.connected && slot.initial_connect_done && !slot.circuit_breaker_tripped && !reconnect_attempted) {
static const unsigned long SLOT_BACKOFF_MS[] = { 10000, 30000, 60000, 120000, 300000 };
static const uint8_t MAX_FAILURES_AT_MAX_BACKOFF = 3; // ~15 min at max backoff before giving up
unsigned long reconnect_elapsed = (now_millis >= slot.last_reconnect_attempt) ?
(now_millis - slot.last_reconnect_attempt) :
(ULONG_MAX - slot.last_reconnect_attempt + now_millis + 1);
unsigned int idx = (slot.reconnect_backoff < 5) ? slot.reconnect_backoff : 4;
unsigned long delay_ms = SLOT_BACKOFF_MS[idx] + (index * 3000UL); // stagger by slot index
if (reconnect_elapsed >= delay_ms) {
if (MQTTConnectionPolicy::reconnectDue(
static_cast<uint32_t>(now_millis), static_cast<uint32_t>(slot.last_reconnect_attempt),
slot.reconnect_backoff, static_cast<uint8_t>(index))) {
slot.last_reconnect_attempt = now_millis;
if (slot.reconnect_backoff < 5) {
slot.reconnect_backoff++;
} else {
slot.max_backoff_failures++;
if (slot.max_backoff_failures >= MAX_FAILURES_AT_MAX_BACKOFF) {
slot.circuit_breaker_tripped = true;
MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker tripped after %d failures at max backoff - stopping reconnect attempts. Reconfigure slot to retry.", index + 1, slot.max_backoff_failures);
return;
}
MQTTConnectionPolicy::BackoffAdvance advance = MQTTConnectionPolicy::advanceBackoff(
slot.reconnect_backoff, slot.max_backoff_failures);
slot.reconnect_backoff = advance.reconnect_backoff;
slot.max_backoff_failures = advance.max_backoff_failures;
slot.circuit_breaker_tripped = advance.circuit_breaker_tripped;
if (!advance.should_reconnect) {
MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker tripped after %d failures at max backoff - stopping reconnect attempts. Reconfigure slot to retry.", index + 1, slot.max_backoff_failures);
return;
}
MQTT_DEBUG_PRINTLN("MQTT%d reconnecting (backoff level %d, failures at max: %d, int_heap=%d)", index + 1, slot.reconnect_backoff, slot.max_backoff_failures,
(int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL));
@@ -1740,14 +1730,12 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
// renewal scheduling in maintainSlotConnection() can be derived from it.
unsigned long MQTTBridge::slotTokenLifetime(int index) const {
const MQTTSlot& slot = _slots[index];
unsigned long base_lifetime = 86400; // default 24h
unsigned long base_lifetime = MQTTConnectionPolicy::kDefaultJwtLifetimeSecs;
if (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT && slot.preset->token_lifetime > 0) {
base_lifetime = slot.preset->token_lifetime;
}
// Stagger token expiry per slot to avoid simultaneous renewal/reconnect.
// Use 5% of lifetime per slot, capped at 300s, so short-lived tokens aren't over-reduced.
unsigned long stagger = index * min((unsigned long)300, base_lifetime / 20);
return base_lifetime - stagger;
return MQTTConnectionPolicy::jwtLifetimeSecs(
static_cast<uint32_t>(base_lifetime), static_cast<uint8_t>(index));
}
// How early (seconds before the token's exp claim) to renew the token AND
@@ -1763,13 +1751,6 @@ unsigned long MQTTBridge::slotTokenLifetime(int index) const {
// lifetime/10 with a 60 s floor and 300 s cap: 24 h tokens renew 5 min early
// (unchanged in practice), waev renews ~5 min early with ~5 throttled retry
// windows, and degenerate short lifetimes still renew inside their validity.
unsigned long MQTTBridge::tokenRenewalBufferSecs(unsigned long lifetime_secs) {
unsigned long buffer = lifetime_secs / 10;
if (buffer < 60) buffer = 60;
if (buffer > 300) buffer = 300;
return buffer;
}
bool MQTTBridge::createSlotAuthToken(int index) {
if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false;
MQTTSlot& slot = _slots[index];
@@ -1885,31 +1866,28 @@ 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";
// Pure expansion lives in helpers/MQTTTopicTemplate.h (host-tested).
return mqttSubstituteTopic(tmpl, _iata, _device_id,
_obs->mqtt_slot_token[slot_index], type_str, buf, buf_size);
return mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, (int)type, tmpl,
_iata, _device_id, _obs->mqtt_slot_token[slot_index],
buf, buf_size);
}
bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_buf, size_t buf_size) {
if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false;
static_assert(
static_cast<int>(MSG_STATUS) == MQTT_PUBLICATION_STATUS &&
static_cast<int>(MSG_PACKETS) == MQTT_PUBLICATION_PACKETS &&
static_cast<int>(MSG_RAW) == MQTT_PUBLICATION_RAW,
"topic router enum drift");
if (!mqttTopicSlotIndexValid(index, RUNTIME_MQTT_SLOTS)) return false;
const MQTTSlot& slot = _slots[index];
// Preset slots: use hardcoded topic logic
if (slot.preset) {
if (slot.preset->topic_style == MQTT_TOPIC_MESHRANK) {
// MeshRank: packets only, uses per-slot token in topic path
if (type != MSG_PACKETS) return false;
const char* token = _obs->mqtt_slot_token[index];
if (!token || token[0] == '\0') return false;
snprintf(topic_buf, buf_size, "meshrank/uplink/%s/%s/packets", token, _device_id);
return true;
}
// MQTT_TOPIC_MESHCORE (default for all other presets)
if (!isIATAValid()) return false;
const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw";
snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str);
return true;
MQTTTopicRouteStyle style = (slot.preset->topic_style == MQTT_TOPIC_MESHRANK)
? MQTT_ROUTE_MESHRANK : MQTT_ROUTE_MESHCORE;
return mqttBuildPublicationTopic(style, (int)type, nullptr,
_iata, _device_id, _obs->mqtt_slot_token[index],
topic_buf, buf_size);
}
// Custom slots: use topic template if set, otherwise default meshcore format
@@ -1917,10 +1895,9 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
return substituteTopicTemplate(_obs->mqtt_slot_topic[index], type, index, topic_buf, buf_size);
}
// Default: meshcore format
if (!isIATAValid()) return false;
const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw";
snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str);
return true;
return mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, (int)type, nullptr,
_iata, _device_id, _obs->mqtt_slot_token[index],
topic_buf, buf_size);
}
void MQTTBridge::publishStatusToSlot(int index) {
@@ -2451,7 +2428,9 @@ void MQTTBridge::processPacketQueue() {
// Flush stale packets after extended disconnect
if (_queue_disconnected_since == 0) {
_queue_disconnected_since = now;
} else if ((now - _queue_disconnected_since) >= QUEUE_STALE_MS) {
} else if (MQTTPacketQueuePolicy::shouldFlushDisconnected(
static_cast<uint32_t>(now),
static_cast<uint32_t>(_queue_disconnected_since))) {
QueuedPacket discard;
while (xQueueReceive(_packet_queue_handle, &discard, 0) == pdTRUE) {}
_queue_count = 0;
@@ -2467,19 +2446,18 @@ void MQTTBridge::processPacketQueue() {
// Adaptive drain: burst-process when queue has backlog, gentle otherwise
int processed = 0;
int max_per_loop = (_queue_count > 5) ? 5 : 1;
const MQTTPacketQueuePolicy::DrainBudget drain_budget =
MQTTPacketQueuePolicy::drainBudget(static_cast<size_t>(_queue_count));
unsigned long loop_start_time = millis();
const unsigned long MAX_PROCESSING_TIME_MS = (_queue_count > 5) ? 100 : 30;
static const uint8_t MAX_QOS0_RETRY_ATTEMPTS = 3;
static const unsigned long RETRY_DELAY_BASE_MS = 300UL;
static const unsigned long RETRY_DELAY_JITTER_MS = 200UL;
#ifdef MQTT_DIAG_VERBOSE
static unsigned long last_retry_schedule_log = 0;
#endif
while (processed < max_per_loop) {
unsigned long elapsed = millis() - loop_start_time;
if (elapsed > MAX_PROCESSING_TIME_MS) {
while (processed < drain_budget.max_packets) {
if (!MQTTPacketQueuePolicy::drainTimeAvailable(
static_cast<uint32_t>(millis()),
static_cast<uint32_t>(loop_start_time),
drain_budget.max_time_ms)) {
break;
}
@@ -2490,7 +2468,10 @@ void MQTTBridge::processPacketQueue() {
}
unsigned long now_ms = millis();
if (queued.next_retry_ms != 0 && now_ms < queued.next_retry_ms) {
if (!MQTTPacketQueuePolicy::retryReady(
static_cast<uint32_t>(now_ms),
static_cast<uint32_t>(queued.next_retry_ms),
queued.retry_attempts)) {
// Not ready yet; put it back and stop draining this cycle.
xQueueSend(_packet_queue_handle, &queued, 0);
break;
@@ -2520,16 +2501,22 @@ void MQTTBridge::processPacketQueue() {
}
bool any_published = packet_published || raw_published;
if (!any_published && queued.retry_attempts < MAX_QOS0_RETRY_ATTEMPTS) {
queued.retry_attempts++;
unsigned long retry_delay_ms = RETRY_DELAY_BASE_MS + (now_ms % RETRY_DELAY_JITTER_MS);
queued.next_retry_ms = now_ms + retry_delay_ms;
const MQTTPacketQueuePolicy::RetryDecision retry =
MQTTPacketQueuePolicy::retryDecision(
any_published, queued.retry_attempts,
static_cast<uint32_t>(now_ms));
if (retry.action == MQTTPacketQueuePolicy::RetryAction::Schedule) {
queued.retry_attempts = retry.retry_attempts;
queued.next_retry_ms = retry.next_retry_ms;
#ifdef MQTT_DIAG_VERBOSE
if (now_ms - last_retry_schedule_log > 5000UL) {
unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0;
unsigned long age_ms = queued.timestamp > 0
? MQTTPacketQueuePolicy::elapsedMs(static_cast<uint32_t>(now_ms),
static_cast<uint32_t>(queued.timestamp))
: 0;
MQTT_DEBUG_PRINTLN("Retry scheduled: attempt=%u/%u delay=%lu age=%lu q=%u pkt_type=%u packet_ok=%d raw_ok=%d",
(unsigned)queued.retry_attempts, (unsigned)MAX_QOS0_RETRY_ATTEMPTS,
retry_delay_ms, age_ms, (unsigned)uxQueueMessagesWaiting(_packet_queue_handle),
(unsigned)queued.retry_attempts, (unsigned)MQTTPacketQueuePolicy::kMaxQos0RetryAttempts,
(unsigned long)retry.delay_ms, age_ms, (unsigned)uxQueueMessagesWaiting(_packet_queue_handle),
(unsigned)queued.packet_copy.getPayloadType(), packet_published ? 1 : 0, raw_published ? 1 : 0);
last_retry_schedule_log = now_ms;
}
@@ -2537,13 +2524,16 @@ void MQTTBridge::processPacketQueue() {
if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) {
MQTT_DEBUG_PRINTLN("Retry requeue failed, dropping packet (attempt=%u)", queued.retry_attempts);
}
} else if (!any_published) {
} else if (retry.action == MQTTPacketQueuePolicy::RetryAction::Drop) {
// Intentional: QoS0 best-effort packets are dropped silently in normal
// builds; detailed exhaustion logs are only emitted in verbose mode.
#ifdef MQTT_DIAG_VERBOSE
static unsigned long last_retry_drop_log = 0;
if (now_ms - last_retry_drop_log > 60000UL) {
unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0;
unsigned long age_ms = queued.timestamp > 0
? MQTTPacketQueuePolicy::elapsedMs(static_cast<uint32_t>(now_ms),
static_cast<uint32_t>(queued.timestamp))
: 0;
MQTT_DEBUG_PRINTLN("Packet dropped after retry exhaustion (attempts=%u age=%lu pkt_type=%u packet_ok=%d raw_ok=%d)",
queued.retry_attempts, age_ms, (unsigned)queued.packet_copy.getPayloadType(),
packet_published ? 1 : 0, raw_published ? 1 : 0);
@@ -2558,6 +2548,7 @@ void MQTTBridge::processPacketQueue() {
#else
// Non-ESP32: Use circular buffer
if (_queue_count == 0) {
_queue_disconnected_since = 0;
return;
}
@@ -2570,33 +2561,48 @@ void MQTTBridge::processPacketQueue() {
MQTT_DEBUG_PRINTLN("Queue has %d packets but no slots connected", _queue_count);
_last_no_broker_log = now;
}
if (_queue_disconnected_since == 0) {
_queue_disconnected_since = now;
} else if (MQTTPacketQueuePolicy::shouldFlushDisconnected(
static_cast<uint32_t>(now),
static_cast<uint32_t>(_queue_disconnected_since))) {
while (_queue_count > 0) {
dequeuePacket();
}
MQTT_DEBUG_PRINTLN("Flushed stale packet queue after %lu ms disconnected",
now - _queue_disconnected_since);
_queue_disconnected_since = now;
}
}
return;
}
_queue_disconnected_since = 0;
_last_no_broker_log = 0;
// Adaptive drain: burst-process when queue has backlog, gentle otherwise
int processed = 0;
int max_per_loop = (_queue_count > 5) ? 5 : 1;
const MQTTPacketQueuePolicy::DrainBudget drain_budget =
MQTTPacketQueuePolicy::drainBudget(static_cast<size_t>(_queue_count));
unsigned long loop_start_time = millis();
const unsigned long MAX_PROCESSING_TIME_MS = (_queue_count > 5) ? 100 : 30;
static const uint8_t MAX_QOS0_RETRY_ATTEMPTS = 3;
static const unsigned long RETRY_DELAY_BASE_MS = 300UL;
static const unsigned long RETRY_DELAY_JITTER_MS = 200UL;
#ifdef MQTT_DIAG_VERBOSE
static unsigned long last_retry_schedule_log = 0;
#endif
while (_queue_count > 0 && processed < max_per_loop) {
unsigned long elapsed = millis() - loop_start_time;
if (elapsed > MAX_PROCESSING_TIME_MS) {
while (_queue_count > 0 && processed < drain_budget.max_packets) {
if (!MQTTPacketQueuePolicy::drainTimeAvailable(
static_cast<uint32_t>(millis()),
static_cast<uint32_t>(loop_start_time),
drain_budget.max_time_ms)) {
break;
}
QueuedPacket& queued = _packet_queue[_queue_head];
unsigned long now_ms = millis();
if (queued.next_retry_ms != 0 && now_ms < queued.next_retry_ms) {
if (!MQTTPacketQueuePolicy::retryReady(
static_cast<uint32_t>(now_ms),
static_cast<uint32_t>(queued.next_retry_ms),
queued.retry_attempts)) {
break;
}
@@ -2621,28 +2627,37 @@ void MQTTBridge::processPacketQueue() {
}
bool any_published = packet_published || raw_published;
if (!any_published && queued.retry_attempts < MAX_QOS0_RETRY_ATTEMPTS) {
queued.retry_attempts++;
unsigned long retry_delay_ms = RETRY_DELAY_BASE_MS + (now_ms % RETRY_DELAY_JITTER_MS);
queued.next_retry_ms = now_ms + retry_delay_ms;
const MQTTPacketQueuePolicy::RetryDecision retry =
MQTTPacketQueuePolicy::retryDecision(
any_published, queued.retry_attempts,
static_cast<uint32_t>(now_ms));
if (retry.action == MQTTPacketQueuePolicy::RetryAction::Schedule) {
queued.retry_attempts = retry.retry_attempts;
queued.next_retry_ms = retry.next_retry_ms;
#ifdef MQTT_DIAG_VERBOSE
if (now_ms - last_retry_schedule_log > 5000UL) {
unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0;
unsigned long age_ms = queued.timestamp > 0
? MQTTPacketQueuePolicy::elapsedMs(static_cast<uint32_t>(now_ms),
static_cast<uint32_t>(queued.timestamp))
: 0;
MQTT_DEBUG_PRINTLN("Retry scheduled: attempt=%u/%u delay=%lu age=%lu q=%d pkt_type=%u packet_ok=%d raw_ok=%d",
(unsigned)queued.retry_attempts, (unsigned)MAX_QOS0_RETRY_ATTEMPTS,
retry_delay_ms, age_ms, _queue_count,
(unsigned)queued.retry_attempts, (unsigned)MQTTPacketQueuePolicy::kMaxQos0RetryAttempts,
(unsigned long)retry.delay_ms, age_ms, _queue_count,
(unsigned)queued.packet_copy.getPayloadType(), packet_published ? 1 : 0, raw_published ? 1 : 0);
last_retry_schedule_log = now_ms;
}
#endif
break; // keep packet at head for delayed retry
} else if (!any_published) {
} else if (retry.action == MQTTPacketQueuePolicy::RetryAction::Drop) {
// Intentional: QoS0 best-effort packets are dropped silently in normal
// builds; detailed exhaustion logs are only emitted in verbose mode.
#ifdef MQTT_DIAG_VERBOSE
static unsigned long last_retry_drop_log = 0;
if (now_ms - last_retry_drop_log > 60000UL) {
unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0;
unsigned long age_ms = queued.timestamp > 0
? MQTTPacketQueuePolicy::elapsedMs(static_cast<uint32_t>(now_ms),
static_cast<uint32_t>(queued.timestamp))
: 0;
MQTT_DEBUG_PRINTLN("Packet dropped after retry exhaustion (attempts=%u age=%lu pkt_type=%u packet_ok=%d raw_ok=%d)",
queued.retry_attempts, age_ms, (unsigned)queued.packet_copy.getPayloadType(),
packet_published ? 1 : 0, raw_published ? 1 : 0);
@@ -2970,27 +2985,41 @@ void MQTTBridge::queuePacket(mesh::Packet* packet, bool is_tx) {
// Try to send to queue (non-blocking)
if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) {
const MQTTPacketQueuePolicy::EnqueueAction action =
MQTTPacketQueuePolicy::enqueueAction(
static_cast<size_t>(uxQueueMessagesWaiting(_packet_queue_handle)),
static_cast<size_t>(MAX_QUEUE_SIZE));
QueuedPacket oldest;
if (xQueueReceive(_packet_queue_handle, &oldest, 0) == pdTRUE) {
if (action == MQTTPacketQueuePolicy::EnqueueAction::EvictOldestThenEnqueue &&
xQueueReceive(_packet_queue_handle, &oldest, 0) == pdTRUE) {
MQTT_DEBUG_PRINTLN("Queue full, dropping oldest packet reference");
if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) {
MQTT_DEBUG_PRINTLN("Failed to queue packet after dropping oldest");
return;
}
} else {
} else if (action == MQTTPacketQueuePolicy::EnqueueAction::Reject) {
MQTT_DEBUG_PRINTLN("Queue has no capacity");
return;
} else if (action == MQTTPacketQueuePolicy::EnqueueAction::EvictOldestThenEnqueue) {
MQTT_DEBUG_PRINTLN("Queue full and cannot remove oldest packet");
return;
}
// If the consumer made room after the failed send, retry without evicting.
if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) {
MQTT_DEBUG_PRINTLN("Failed to queue packet after overflow handling");
return;
}
}
UBaseType_t queue_messages = uxQueueMessagesWaiting(_packet_queue_handle);
_queue_count = queue_messages;
#else
// Non-ESP32: Use circular buffer
if (_queue_count >= MAX_QUEUE_SIZE) {
const MQTTPacketQueuePolicy::EnqueueAction action =
MQTTPacketQueuePolicy::enqueueAction(
static_cast<size_t>(_queue_count), static_cast<size_t>(MAX_QUEUE_SIZE));
if (action == MQTTPacketQueuePolicy::EnqueueAction::EvictOldestThenEnqueue) {
QueuedPacket& oldest = _packet_queue[_queue_head];
MQTT_DEBUG_PRINTLN("Queue full, dropping oldest packet (queue size: %d)", _queue_count);
dequeuePacket();
} else if (action == MQTTPacketQueuePolicy::EnqueueAction::Reject) {
return;
}
QueuedPacket& queued = _packet_queue[_queue_tail];
src/helpers/bridges/MQTTBridge.h
-2
View File
@@ -289,7 +289,6 @@ private:
// Queue staleness tracking
unsigned long _queue_disconnected_since; // 0 = has connected slots
static const unsigned long QUEUE_STALE_MS = 300000UL; // Flush queue after 5 min disconnected
#ifdef WITH_SNMP
MeshSNMPAgent* _snmp_agent;
@@ -341,7 +340,6 @@ private:
void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted, bool& teardown_attempted);
bool createSlotAuthToken(int index); // Create/renew JWT token for a slot
unsigned long slotTokenLifetime(int index) const; // effective JWT lifetime (preset/default minus slot stagger), seconds
static unsigned long tokenRenewalBufferSecs(unsigned long lifetime_secs); // how early to renew+reconnect before exp
bool publishToSlot(int index, const char* topic, const char* payload, bool retained = false, uint8_t qos = 0);
bool publishToAllSlots(const char* topic, const char* payload, bool retained = false, uint8_t qos = 0);
void publishStatusToSlot(int index);
test/README.md
+4
View File
@@ -27,6 +27,10 @@ does not reflect the GoogleTest count — run the built binary directly
| `test_observer_validation` | `src/helpers/MQTTObserverValidation.h` | IATA (exactly 3 alphanumerics), owner key (64 hex), NTP hostname, and the buffer-fit check behind the #17 length validation — including boundaries and nulls |
| `test_webconfig_keys` | `src/helpers/WebConfigKeys.h` | POST-key allowlist, secret detection, slot-index bounds, and the short-key out-of-bounds guard (attacker-supplied keys) |
| `test_topic_template` | `src/helpers/MQTTTopicTemplate.h` | `{iata}/{device}/{token}/{type}` expansion, overflow/NUL-termination, and a buffer-size fuzz |
| `test_mqtt_topic_router` | `src/helpers/MQTTTopicRouter.h` | complete preset/custom topic-routing contract; MeshRank packets-only behavior; required identifiers; invalid inputs/slots; exact buffer boundaries |
| `test_mqtt_connection_policy` | `src/helpers/MQTTConnectionPolicy.h` | reconnect guard/backoff/stagger and breaker transitions; stable reset; JWT lifetime/renewal policy; exact timing boundaries and 32-bit `millis()` rollover |
| `test_mqtt_packet_queue_policy` | `src/helpers/MQTTPacketQueuePolicy.h` | queue-full eviction; stale-disconnect flush; adaptive drain limits; bounded QoS0 retries; exact timing boundaries and 32-bit `millis()` rollover |
| `test_mqtt_payload_builder` | `src/helpers/MQTTPayloadBuilder.cpp` | status/packet/raw JSON contracts; optional fields; escaping; RX metrics and path; score handling; exact buffer bounds; maximum representative payloads |
| `test_utils` | `src/Utils.cpp` | `Utils::toHex` (upstream) |
## Conventions (and how to add a suite)
test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp
+158
View File
@@ -0,0 +1,158 @@
#include <gtest/gtest.h>
#include <stdint.h>
#include <limits>
#include "helpers/MQTTConnectionPolicy.h"
namespace Policy = MQTTConnectionPolicy;
TEST(MQTTConnectionPolicy, ElapsedTimeHandlesNormalAndWrappedClocks) {
EXPECT_EQ(4000U, Policy::elapsedMs(5000U, 1000U));
const uint32_t before_wrap = std::numeric_limits<uint32_t>::max() - 99U;
EXPECT_EQ(150U, Policy::elapsedMs(50U, before_wrap));
}
TEST(MQTTConnectionPolicy, CrossSlotReconnectGuardHasExactBoundary) {
EXPECT_TRUE(Policy::reconnectGuardActive(14999U, 0U));
EXPECT_FALSE(Policy::reconnectGuardActive(15000U, 0U));
const uint32_t last = std::numeric_limits<uint32_t>::max() - 9999U;
EXPECT_TRUE(Policy::reconnectGuardActive(4999U, last));
EXPECT_FALSE(Policy::reconnectGuardActive(5000U, last));
}
TEST(MQTTConnectionPolicy, StableResetRequiresARealStartAndFullWindow) {
EXPECT_FALSE(Policy::stableConnection(500000U, 0U));
EXPECT_FALSE(Policy::stableConnection(120999U, 1000U));
EXPECT_TRUE(Policy::stableConnection(121000U, 1000U));
}
TEST(MQTTConnectionPolicy, StableResetWindowSurvivesMillisRollover) {
const uint32_t connected_at = std::numeric_limits<uint32_t>::max() - 59999U;
EXPECT_FALSE(Policy::stableConnection(59999U, connected_at));
EXPECT_TRUE(Policy::stableConnection(60000U, connected_at));
}
TEST(MQTTConnectionPolicy, BackoffLadderSaturatesAtFiveMinutes) {
EXPECT_EQ(10000U, Policy::reconnectBackoffMs(0));
EXPECT_EQ(30000U, Policy::reconnectBackoffMs(1));
EXPECT_EQ(60000U, Policy::reconnectBackoffMs(2));
EXPECT_EQ(120000U, Policy::reconnectBackoffMs(3));
EXPECT_EQ(300000U, Policy::reconnectBackoffMs(4));
EXPECT_EQ(300000U, Policy::reconnectBackoffMs(5));
EXPECT_EQ(300000U, Policy::reconnectBackoffMs(255));
}
TEST(MQTTConnectionPolicy, LaterSlotsReceiveThreeSecondStagger) {
EXPECT_EQ(10000U, Policy::reconnectDelayMs(0, 0));
EXPECT_EQ(13000U, Policy::reconnectDelayMs(0, 1));
EXPECT_EQ(315000U, Policy::reconnectDelayMs(5, 5));
}
TEST(MQTTConnectionPolicy, ReconnectDueUsesDelayBoundaryAndWrapSafeElapsedTime) {
EXPECT_FALSE(Policy::reconnectDue(12999U, 0U, 0, 1));
EXPECT_TRUE(Policy::reconnectDue(13000U, 0U, 0, 1));
const uint32_t last = std::numeric_limits<uint32_t>::max() - 4999U;
EXPECT_FALSE(Policy::reconnectDue(4999U, last, 0, 0));
EXPECT_TRUE(Policy::reconnectDue(5000U, last, 0, 0));
}
TEST(MQTTConnectionPolicy, BackoffAdvanceClimbsThenCountsFailuresAtMaximum) {
Policy::BackoffAdvance first = Policy::advanceBackoff(0, 0);
EXPECT_EQ(1, first.reconnect_backoff);
EXPECT_EQ(0, first.max_backoff_failures);
EXPECT_FALSE(first.circuit_breaker_tripped);
EXPECT_TRUE(first.should_reconnect);
Policy::BackoffAdvance enters_maximum = Policy::advanceBackoff(4, 0);
EXPECT_EQ(5, enters_maximum.reconnect_backoff);
EXPECT_EQ(0, enters_maximum.max_backoff_failures);
EXPECT_FALSE(enters_maximum.circuit_breaker_tripped);
EXPECT_TRUE(enters_maximum.should_reconnect);
Policy::BackoffAdvance first_max_failure = Policy::advanceBackoff(5, 0);
EXPECT_EQ(5, first_max_failure.reconnect_backoff);
EXPECT_EQ(1, first_max_failure.max_backoff_failures);
EXPECT_FALSE(first_max_failure.circuit_breaker_tripped);
EXPECT_TRUE(first_max_failure.should_reconnect);
}
TEST(MQTTConnectionPolicy, ThirdFailureAtMaximumTripsWithoutAnotherHandshake) {
Policy::BackoffAdvance result = Policy::advanceBackoff(5, 2);
EXPECT_EQ(5, result.reconnect_backoff);
EXPECT_EQ(3, result.max_backoff_failures);
EXPECT_TRUE(result.circuit_breaker_tripped);
EXPECT_FALSE(result.should_reconnect);
}
TEST(MQTTConnectionPolicy, CircuitBreakerProbeHasExactThirtyMinuteBoundary) {
EXPECT_FALSE(Policy::circuitBreakerProbeDue(1799999U, 0U));
EXPECT_TRUE(Policy::circuitBreakerProbeDue(1800000U, 0U));
const uint32_t last = std::numeric_limits<uint32_t>::max() - 899999U;
EXPECT_FALSE(Policy::circuitBreakerProbeDue(899999U, last));
EXPECT_TRUE(Policy::circuitBreakerProbeDue(900000U, last));
}
TEST(MQTTConnectionPolicy, JwtLifetimeUsesCappedPerSlotStagger) {
EXPECT_EQ(86400U, Policy::jwtLifetimeSecs(86400U, 0));
EXPECT_EQ(86100U, Policy::jwtLifetimeSecs(86400U, 1));
EXPECT_EQ(84900U, Policy::jwtLifetimeSecs(86400U, 5));
}
TEST(MQTTConnectionPolicy, ShortJwtLifetimeUsesFivePercentPerSlot) {
EXPECT_EQ(3300U, Policy::jwtLifetimeSecs(3300U, 0));
EXPECT_EQ(3135U, Policy::jwtLifetimeSecs(3300U, 1));
EXPECT_EQ(2970U, Policy::jwtLifetimeSecs(3300U, 2));
EXPECT_EQ(85U, Policy::jwtLifetimeSecs(100U, 3));
}
TEST(MQTTConnectionPolicy, JwtLifetimeCannotUnderflowForUnexpectedSlotInput) {
EXPECT_EQ(0U, Policy::jwtLifetimeSecs(100U, 255));
}
TEST(MQTTConnectionPolicy, RenewalBufferHasOneMinuteFloorAndFiveMinuteCap) {
EXPECT_EQ(60U, Policy::renewalBufferSecs(0U));
EXPECT_EQ(60U, Policy::renewalBufferSecs(599U));
EXPECT_EQ(60U, Policy::renewalBufferSecs(600U));
EXPECT_EQ(299U, Policy::renewalBufferSecs(2999U));
EXPECT_EQ(300U, Policy::renewalBufferSecs(3000U));
EXPECT_EQ(300U, Policy::renewalBufferSecs(86400U));
}
TEST(MQTTConnectionPolicy, UnsynchronizedClockOnlyCreatesAMissingToken) {
EXPECT_TRUE(Policy::tokenNeedsRenewal(false, 0U, 0U, 300U));
EXPECT_FALSE(Policy::tokenNeedsRenewal(false, 0U, 1735693200U, 300U));
}
TEST(MQTTConnectionPolicy, SyncedClockRenewsInvalidExpiredOrImminentTokens) {
const uint32_t expires = 1735693200U;
EXPECT_TRUE(Policy::tokenNeedsRenewal(true, 1735689000U, 0U, 300U));
EXPECT_TRUE(Policy::tokenNeedsRenewal(true, 1735689000U, 999999999U, 300U));
EXPECT_FALSE(Policy::tokenNeedsRenewal(true, expires - 301U, expires, 300U));
EXPECT_TRUE(Policy::tokenNeedsRenewal(true, expires - 300U, expires, 300U));
EXPECT_TRUE(Policy::tokenNeedsRenewal(true, expires, expires, 300U));
EXPECT_TRUE(Policy::tokenNeedsRenewal(true, expires + 1U, expires, 300U));
}
TEST(MQTTConnectionPolicy, RenewalThrottleHasExactBoundaryAndHandlesRollover) {
EXPECT_FALSE(Policy::renewalAttemptAllowed(59999U, 0U));
EXPECT_TRUE(Policy::renewalAttemptAllowed(60000U, 0U));
const uint32_t last = std::numeric_limits<uint32_t>::max() - 29999U;
EXPECT_FALSE(Policy::renewalAttemptAllowed(29999U, last));
EXPECT_TRUE(Policy::renewalAttemptAllowed(30000U, last));
}
TEST(MQTTConnectionPolicy, JwtClockNeedsNtpOrAReasonableWallClock) {
EXPECT_FALSE(Policy::jwtClockAvailable(false, Policy::kJwtClockThreshold - 1U));
EXPECT_TRUE(Policy::jwtClockAvailable(false, Policy::kJwtClockThreshold));
EXPECT_TRUE(Policy::jwtClockAvailable(true, 0U));
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
test/test_mqtt_packet_queue_policy/test_mqtt_packet_queue_policy.cpp
+157
View File
@@ -0,0 +1,157 @@
#include <gtest/gtest.h>
#include <stdint.h>
#include "helpers/MQTTPacketQueuePolicy.h"
namespace QueuePolicy = MQTTPacketQueuePolicy;
TEST(MQTTPacketQueuePolicy, EnqueuesWhileCapacityRemains) {
EXPECT_EQ(QueuePolicy::EnqueueAction::Enqueue,
QueuePolicy::enqueueAction(0, 6));
EXPECT_EQ(QueuePolicy::EnqueueAction::Enqueue,
QueuePolicy::enqueueAction(5, 6));
}
TEST(MQTTPacketQueuePolicy, EvictsOldestAtOrAboveCapacity) {
EXPECT_EQ(QueuePolicy::EnqueueAction::EvictOldestThenEnqueue,
QueuePolicy::enqueueAction(6, 6));
EXPECT_EQ(QueuePolicy::EnqueueAction::EvictOldestThenEnqueue,
QueuePolicy::enqueueAction(7, 6));
}
TEST(MQTTPacketQueuePolicy, RejectsQueueWithZeroCapacity) {
EXPECT_EQ(QueuePolicy::EnqueueAction::Reject,
QueuePolicy::enqueueAction(0, 0));
}
TEST(MQTTPacketQueuePolicy, DisconnectedQueueFlushesAtExactStaleBoundary) {
const uint32_t started = 1000;
EXPECT_FALSE(QueuePolicy::shouldFlushDisconnected(
started + QueuePolicy::kDisconnectedStaleMs - 1, started));
EXPECT_TRUE(QueuePolicy::shouldFlushDisconnected(
started + QueuePolicy::kDisconnectedStaleMs, started));
EXPECT_TRUE(QueuePolicy::shouldFlushDisconnected(
started + QueuePolicy::kDisconnectedStaleMs + 1, started));
}
TEST(MQTTPacketQueuePolicy, ZeroDisconnectedTimestampMeansNotStarted) {
EXPECT_FALSE(QueuePolicy::shouldFlushDisconnected(UINT32_MAX, 0));
}
TEST(MQTTPacketQueuePolicy, DisconnectedStaleTimerSurvivesMillisWrap) {
const uint32_t started = UINT32_MAX - 100;
EXPECT_FALSE(QueuePolicy::shouldFlushDisconnected(198, started, 300));
EXPECT_TRUE(QueuePolicy::shouldFlushDisconnected(199, started, 300));
}
TEST(MQTTPacketQueuePolicy, DrainIsGentleThroughFivePackets) {
for (size_t count = 0; count <= QueuePolicy::kBacklogThreshold; ++count) {
const QueuePolicy::DrainBudget budget = QueuePolicy::drainBudget(count);
EXPECT_EQ(QueuePolicy::kGentleDrainCount, budget.max_packets) << count;
EXPECT_EQ(QueuePolicy::kGentleDrainBudgetMs, budget.max_time_ms) << count;
}
}
TEST(MQTTPacketQueuePolicy, DrainBurstsAboveFivePackets) {
const QueuePolicy::DrainBudget budget =
QueuePolicy::drainBudget(QueuePolicy::kBacklogThreshold + 1);
EXPECT_EQ(QueuePolicy::kBurstDrainCount, budget.max_packets);
EXPECT_EQ(QueuePolicy::kBurstDrainBudgetMs, budget.max_time_ms);
}
TEST(MQTTPacketQueuePolicy, DrainTimeBudgetUsesInclusiveBoundary) {
EXPECT_TRUE(QueuePolicy::drainTimeAvailable(129, 100, 30));
EXPECT_TRUE(QueuePolicy::drainTimeAvailable(130, 100, 30));
EXPECT_FALSE(QueuePolicy::drainTimeAvailable(131, 100, 30));
}
TEST(MQTTPacketQueuePolicy, DrainTimeBudgetSurvivesMillisWrap) {
const uint32_t started = UINT32_MAX - 10;
EXPECT_TRUE(QueuePolicy::drainTimeAvailable(19, started, 30));
EXPECT_FALSE(QueuePolicy::drainTimeAvailable(20, started, 30));
}
TEST(MQTTPacketQueuePolicy, NewPacketIsReadyWithoutRetryDeadline) {
EXPECT_TRUE(QueuePolicy::retryReady(100, 0, 0));
EXPECT_TRUE(QueuePolicy::retryReady(100, 500, 0));
}
TEST(MQTTPacketQueuePolicy, RetryBecomesReadyAtExactDeadline) {
EXPECT_FALSE(QueuePolicy::retryReady(499, 500, 1));
EXPECT_TRUE(QueuePolicy::retryReady(500, 500, 1));
EXPECT_TRUE(QueuePolicy::retryReady(501, 500, 1));
}
TEST(MQTTPacketQueuePolicy, RetryDeadlineSurvivesMillisWrap) {
const uint32_t deadline = 100;
EXPECT_FALSE(QueuePolicy::retryReady(UINT32_MAX - 50, deadline, 1));
EXPECT_FALSE(QueuePolicy::retryReady(99, deadline, 1));
EXPECT_TRUE(QueuePolicy::retryReady(100, deadline, 1));
}
TEST(MQTTPacketQueuePolicy, WrappedZeroIsARealRetryDeadline) {
EXPECT_FALSE(QueuePolicy::retryReady(UINT32_MAX, 0, 1));
EXPECT_TRUE(QueuePolicy::retryReady(0, 0, 1));
}
TEST(MQTTPacketQueuePolicy, SuccessfulPublishCompletesWithoutChangingAttempts) {
const QueuePolicy::RetryDecision decision =
QueuePolicy::retryDecision(true, 2, 1234);
EXPECT_EQ(QueuePolicy::RetryAction::Complete, decision.action);
EXPECT_EQ(2, decision.retry_attempts);
EXPECT_EQ(0U, decision.delay_ms);
EXPECT_EQ(0U, decision.next_retry_ms);
}
TEST(MQTTPacketQueuePolicy, FailedPublishSchedulesBoundedRetry) {
const QueuePolicy::RetryDecision minimum =
QueuePolicy::retryDecision(false, 0, 400);
EXPECT_EQ(QueuePolicy::RetryAction::Schedule, minimum.action);
EXPECT_EQ(1, minimum.retry_attempts);
EXPECT_EQ(300U, minimum.delay_ms);
EXPECT_EQ(700U, minimum.next_retry_ms);
const QueuePolicy::RetryDecision maximum =
QueuePolicy::retryDecision(false, 1, 599);
EXPECT_EQ(QueuePolicy::RetryAction::Schedule, maximum.action);
EXPECT_EQ(2, maximum.retry_attempts);
EXPECT_EQ(499U, maximum.delay_ms);
EXPECT_EQ(1098U, maximum.next_retry_ms);
}
TEST(MQTTPacketQueuePolicy, ThirdFailedPublishSchedulesFinalRetry) {
const QueuePolicy::RetryDecision decision =
QueuePolicy::retryDecision(false, 2, 1000);
EXPECT_EQ(QueuePolicy::RetryAction::Schedule, decision.action);
EXPECT_EQ(QueuePolicy::kMaxQos0RetryAttempts, decision.retry_attempts);
}
TEST(MQTTPacketQueuePolicy, FailureAfterFinalRetryDropsPacket) {
const QueuePolicy::RetryDecision decision =
QueuePolicy::retryDecision(false, QueuePolicy::kMaxQos0RetryAttempts, 1000);
EXPECT_EQ(QueuePolicy::RetryAction::Drop, decision.action);
EXPECT_EQ(QueuePolicy::kMaxQos0RetryAttempts, decision.retry_attempts);
EXPECT_EQ(0U, decision.delay_ms);
EXPECT_EQ(0U, decision.next_retry_ms);
}
TEST(MQTTPacketQueuePolicy, RetrySchedulingDeadlineMayWrapToZero) {
// This timestamp has jitter 98, so its 398 ms delay wraps to exactly zero.
const uint32_t now = UINT32_MAX - 397;
ASSERT_EQ(98U, now % QueuePolicy::kRetryDelayJitterMs);
const QueuePolicy::RetryDecision decision =
QueuePolicy::retryDecision(false, 0, now);
EXPECT_EQ(QueuePolicy::RetryAction::Schedule, decision.action);
EXPECT_EQ(398U, decision.delay_ms);
EXPECT_EQ(0U, decision.next_retry_ms);
EXPECT_FALSE(QueuePolicy::retryReady(UINT32_MAX, decision.next_retry_ms,
decision.retry_attempts));
EXPECT_TRUE(QueuePolicy::retryReady(0, decision.next_retry_ms,
decision.retry_attempts));
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp
+236
View File
@@ -0,0 +1,236 @@
#include <gtest/gtest.h>
#include <ArduinoJson.h>
#include <cmath>
#include <cstring>
#include <string>
#include <vector>
#include "helpers/MQTTPayloadBuilder.h"
namespace {
constexpr const char* kTimestamp = "2026-07-18T12:34:56.123456+00:00";
static int buildMinimalStatus(JsonDocument& scratch, char* buffer, size_t buffer_size) {
return MQTTPayloadBuilder::buildStatusMessage(
scratch, "DEN Repeater", "0123456789ABCDEF", "Heltec V3", "v1.16.0",
"915.000000,62.5,7,5", "MeshCore", "online", kTimestamp,
buffer, buffer_size);
}
static int buildRepresentativePacket(JsonDocument& scratch, const char* direction,
float score, const uint8_t* path, int path_hops,
int path_hash_size, const char* raw,
char* buffer, size_t buffer_size) {
return MQTTPayloadBuilder::buildPacketMessage(
scratch, "DEN Repeater", "0123456789ABCDEF", kTimestamp, direction,
"12:34:56", "18/07/2026", 42, 4, "D", 20, raw,
10.26f, -87, score, "89ABCDEF01234567", path, path_hops,
path_hash_size, 64, buffer, buffer_size);
}
TEST(MQTTPayloadBuilder, MinimalStatusHasExactRequiredContract) {
JsonDocument scratch;
char buffer[768];
int len = buildMinimalStatus(scratch, buffer, sizeof(buffer));
ASSERT_GT(len, 0);
EXPECT_EQ(static_cast<size_t>(len), strlen(buffer));
EXPECT_STREQ(
"{\"status\":\"online\",\"timestamp\":\"2026-07-18T12:34:56.123456+00:00\","
"\"origin\":\"DEN Repeater\",\"origin_id\":\"0123456789ABCDEF\","
"\"model\":\"Heltec V3\",\"firmware_version\":\"v1.16.0\","
"\"radio\":\"915.000000,62.5,7,5\",\"client_version\":\"MeshCore\"}",
buffer);
JsonDocument parsed;
ASSERT_FALSE(deserializeJson(parsed, buffer));
EXPECT_FALSE(parsed["repeat"].is<JsonVariant>());
EXPECT_FALSE(parsed["stats"].is<JsonVariant>());
}
TEST(MQTTPayloadBuilder, StatusIncludesRepeatAndEveryRequestedStatistic) {
JsonDocument scratch;
char buffer[1024];
int len = MQTTPayloadBuilder::buildStatusMessage(
scratch, "node", "id", "model", "firmware", "radio", "client", "online",
kTimestamp, buffer, sizeof(buffer), 4200, 86400, 3, 6, -112,
11, 22, 4, 180864, 31, 47, "on");
ASSERT_GT(len, 0);
JsonDocument parsed;
ASSERT_FALSE(deserializeJson(parsed, buffer));
EXPECT_STREQ("on", parsed["repeat"].as<const char*>());
JsonObject stats = parsed["stats"].as<JsonObject>();
ASSERT_FALSE(stats.isNull());
EXPECT_EQ(4200, stats["battery_mv"].as<int>());
EXPECT_EQ(86400, stats["uptime_secs"].as<int>());
EXPECT_EQ(3, stats["errors"].as<int>());
EXPECT_EQ(6, stats["queue_len"].as<int>());
EXPECT_EQ(-112, stats["noise_floor"].as<int>());
EXPECT_EQ(11, stats["tx_air_secs"].as<int>());
EXPECT_EQ(22, stats["rx_air_secs"].as<int>());
EXPECT_EQ(4, stats["recv_errors"].as<int>());
EXPECT_EQ(180864, stats["internal_heap"].as<int>());
EXPECT_EQ(31, stats["packets_sent"].as<int>());
EXPECT_EQ(47, stats["packets_received"].as<int>());
}
TEST(MQTTPayloadBuilder, StatusOmissionSentinelsRemainOmitted) {
JsonDocument scratch;
char buffer[768];
int len = MQTTPayloadBuilder::buildStatusMessage(
scratch, "node", "id", "model", "firmware", "radio", "client", "online",
kTimestamp, buffer, sizeof(buffer), -1, -1, -1, -1, -999,
-1, -1, -1, -1, -1, -1, nullptr);
ASSERT_GT(len, 0);
JsonDocument parsed;
ASSERT_FALSE(deserializeJson(parsed, buffer));
EXPECT_FALSE(parsed["stats"].is<JsonVariant>());
EXPECT_FALSE(parsed["repeat"].is<JsonVariant>());
}
TEST(MQTTPayloadBuilder, StringsAreEscapedAndRoundTrip) {
const char* origin = "node \"north\"\\rack\nline";
const char* model = "Heltec\tV3";
JsonDocument scratch;
char buffer[1024];
int len = MQTTPayloadBuilder::buildStatusMessage(
scratch, origin, "id", model, "v1", "radio", "client", "online",
kTimestamp, buffer, sizeof(buffer));
ASSERT_GT(len, 0);
EXPECT_NE(std::string::npos, std::string(buffer).find("\\\"north\\\""));
EXPECT_NE(std::string::npos, std::string(buffer).find("\\\\rack\\nline"));
JsonDocument parsed;
ASSERT_FALSE(deserializeJson(parsed, buffer));
EXPECT_STREQ(origin, parsed["origin"].as<const char*>());
EXPECT_STREQ(model, parsed["model"].as<const char*>());
}
TEST(MQTTPayloadBuilder, RxPacketIncludesMetricsScaledScoreAndPath) {
const uint8_t path[] = {0xAA, 0xBB, 0x01, 0x2F};
JsonDocument scratch;
char buffer[2048];
int len = buildRepresentativePacket(
scratch, "rx", 0.125f, path, 2, 2, "A0B1C2D3", buffer, sizeof(buffer));
ASSERT_GT(len, 0);
JsonDocument parsed;
ASSERT_FALSE(deserializeJson(parsed, buffer));
EXPECT_STREQ("PACKET", parsed["type"].as<const char*>());
EXPECT_STREQ("rx", parsed["direction"].as<const char*>());
EXPECT_STREQ("42", parsed["len"].as<const char*>());
EXPECT_STREQ("4", parsed["packet_type"].as<const char*>());
EXPECT_STREQ("20", parsed["payload_len"].as<const char*>());
EXPECT_STREQ("10.3", parsed["SNR"].as<const char*>());
EXPECT_STREQ("-87", parsed["RSSI"].as<const char*>());
EXPECT_STREQ("125", parsed["score"].as<const char*>());
JsonArray parsed_path = parsed["path"].as<JsonArray>();
ASSERT_EQ(2U, parsed_path.size());
EXPECT_STREQ("aabb", parsed_path[0].as<const char*>());
EXPECT_STREQ("012f", parsed_path[1].as<const char*>());
}
TEST(MQTTPayloadBuilder, TxPacketOmitsReceiveOnlyMetricsAndAbsentPath) {
JsonDocument scratch;
char buffer[2048];
int len = buildRepresentativePacket(
scratch, "tx", 0.5f, nullptr, 0, 0, "A0B1", buffer, sizeof(buffer));
ASSERT_GT(len, 0);
JsonDocument parsed;
ASSERT_FALSE(deserializeJson(parsed, buffer));
EXPECT_FALSE(parsed["SNR"].is<JsonVariant>());
EXPECT_FALSE(parsed["RSSI"].is<JsonVariant>());
EXPECT_FALSE(parsed["score"].is<JsonVariant>());
EXPECT_FALSE(parsed["path"].is<JsonVariant>());
}
TEST(MQTTPayloadBuilder, RxPacketOmitsUnknownNanScore) {
JsonDocument scratch;
char buffer[2048];
int len = buildRepresentativePacket(
scratch, "rx", std::nanf(""), nullptr, 0, 0, "A0B1", buffer, sizeof(buffer));
ASSERT_GT(len, 0);
JsonDocument parsed;
ASSERT_FALSE(deserializeJson(parsed, buffer));
EXPECT_TRUE(parsed["SNR"].is<const char*>());
EXPECT_TRUE(parsed["RSSI"].is<const char*>());
EXPECT_FALSE(parsed["score"].is<JsonVariant>());
}
TEST(MQTTPayloadBuilder, RawMessageHasExactContractAndEscapesData) {
char buffer[512];
int len = MQTTPayloadBuilder::buildRawMessage(
"node \"A\"", "id\\1", kTimestamp, "AA\nBB", buffer, sizeof(buffer));
ASSERT_GT(len, 0);
EXPECT_EQ(static_cast<size_t>(len), strlen(buffer));
EXPECT_STREQ(
"{\"origin\":\"node \\\"A\\\"\",\"origin_id\":\"id\\\\1\","
"\"timestamp\":\"2026-07-18T12:34:56.123456+00:00\","
"\"type\":\"RAW\",\"data\":\"AA\\nBB\"}",
buffer);
JsonDocument parsed;
ASSERT_FALSE(deserializeJson(parsed, buffer));
EXPECT_STREQ("AA\nBB", parsed["data"].as<const char*>());
}
TEST(MQTTPayloadBuilder, ExactOutputSizeSucceedsAndOneByteShortFailsCleanly) {
JsonDocument scratch;
char reference[768];
int reference_len = buildMinimalStatus(scratch, reference, sizeof(reference));
ASSERT_GT(reference_len, 0);
std::vector<char> exact(static_cast<size_t>(reference_len) + 1);
EXPECT_EQ(reference_len, buildMinimalStatus(scratch, exact.data(), exact.size()));
EXPECT_STREQ(reference, exact.data());
std::vector<char> short_buffer(static_cast<size_t>(reference_len), 'x');
EXPECT_EQ(0, buildMinimalStatus(scratch, short_buffer.data(), short_buffer.size()));
EXPECT_EQ('\0', short_buffer[0]);
EXPECT_EQ(0, buildMinimalStatus(scratch, short_buffer.data(), 1));
EXPECT_EQ('\0', short_buffer[0]);
EXPECT_EQ(0, buildMinimalStatus(scratch, nullptr, exact.size()));
}
TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) {
uint8_t path[64];
for (size_t i = 0; i < sizeof(path); ++i) path[i] = static_cast<uint8_t>(i);
std::string raw(510, 'A');
JsonDocument scratch;
char packet_buffer[2048];
int packet_len = buildRepresentativePacket(
scratch, "rx", 1.0f, path, 16, 4, raw.c_str(),
packet_buffer, sizeof(packet_buffer));
ASSERT_GT(packet_len, 0);
JsonDocument packet;
ASSERT_FALSE(deserializeJson(packet, packet_buffer));
EXPECT_EQ(510U, strlen(packet["raw"].as<const char*>()));
JsonArray parsed_path = packet["path"].as<JsonArray>();
ASSERT_EQ(16U, parsed_path.size());
EXPECT_STREQ("00010203", parsed_path[0].as<const char*>());
EXPECT_STREQ("3c3d3e3f", parsed_path[15].as<const char*>());
char raw_buffer[1024];
int raw_len = MQTTPayloadBuilder::buildRawMessage(
"node", "0123456789ABCDEF", kTimestamp, raw.c_str(),
raw_buffer, sizeof(raw_buffer));
ASSERT_GT(raw_len, 0);
JsonDocument parsed_raw;
ASSERT_FALSE(deserializeJson(parsed_raw, raw_buffer));
EXPECT_EQ(510U, strlen(parsed_raw["data"].as<const char*>()));
}
} // namespace
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
test/test_mqtt_topic_router/test_mqtt_topic_router.cpp
+179
View File
@@ -0,0 +1,179 @@
// Host contract tests for the complete MQTT publication-topic routing policy.
#define WITH_MQTT_BRIDGE 1
#define PROGMEM
#include <gtest/gtest.h>
#include <cstring>
#include <string>
#include "helpers/MQTTPresets.h"
#include "helpers/MQTTTopicRouter.h"
namespace {
constexpr const char* IATA = "DEN";
constexpr const char* DEVICE = "0123456789ABCDEF";
constexpr const char* TOKEN = "account-token";
struct TypeCase {
int type;
const char* name;
};
const TypeCase kTypes[] = {
{MQTT_PUBLICATION_STATUS, "status"},
{MQTT_PUBLICATION_PACKETS, "packets"},
{MQTT_PUBLICATION_RAW, "raw"},
};
TEST(MQTTTopicRouter, EveryMeshCorePresetSupportsEveryPublicationType) {
for (int preset_index = 0; preset_index < MQTT_PRESET_COUNT; ++preset_index) {
const MQTTPresetDef& preset = MQTT_PRESETS[preset_index];
if (preset.topic_style != MQTT_TOPIC_MESHCORE) continue;
for (const TypeCase& type : kTypes) {
char topic[128];
ASSERT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, type.type, nullptr,
IATA, DEVICE, TOKEN, topic, sizeof(topic)))
<< preset.name << " / " << type.name;
EXPECT_EQ(std::string("meshcore/DEN/0123456789ABCDEF/") + type.name, topic)
<< preset.name;
}
}
}
TEST(MQTTTopicRouter, MeshRankContractIsPacketsOnly) {
const MQTTPresetDef* preset = findMQTTPreset("meshrank");
ASSERT_NE(nullptr, preset);
ASSERT_EQ(MQTT_TOPIC_MESHRANK, preset->topic_style);
char topic[128];
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_STATUS,
nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_STREQ("", topic);
ASSERT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_PACKETS,
nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_STREQ("meshrank/uplink/account-token/0123456789ABCDEF/packets", topic);
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_RAW,
nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_STREQ("", topic);
}
TEST(MQTTTopicRouter, MeshCoreRequiresUsableIataAndDevice) {
char topic[64];
const char* invalid_iatas[] = {nullptr, "", "XX", "XXXX", "X/X", "XXX"};
for (const char* iata : invalid_iatas) {
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS,
nullptr, iata, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_STREQ("", topic);
}
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS,
nullptr, IATA, nullptr, TOKEN, topic, sizeof(topic)));
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS,
nullptr, IATA, "", TOKEN, topic, sizeof(topic)));
}
TEST(MQTTTopicRouter, MeshRankRequiresTokenAndDeviceButNotIata) {
char topic[128];
const char* missing_tokens[] = {nullptr, ""};
for (const char* token : missing_tokens) {
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_PACKETS,
nullptr, nullptr, DEVICE, token, topic, sizeof(topic)));
EXPECT_STREQ("", topic);
}
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_PACKETS,
nullptr, nullptr, nullptr, TOKEN, topic, sizeof(topic)));
EXPECT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_PACKETS,
nullptr, nullptr, DEVICE, TOKEN, topic, sizeof(topic)));
}
TEST(MQTTTopicRouter, CustomTemplateExpandsEveryType) {
for (const TypeCase& type : kTypes) {
char topic[128];
ASSERT_TRUE(mqttBuildPublicationTopic(
MQTT_ROUTE_CUSTOM, type.type, "custom/{iata}/{token}/{device}/{type}",
IATA, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_EQ(std::string("custom/DEN/account-token/0123456789ABCDEF/") + type.name, topic);
}
}
TEST(MQTTTopicRouter, CustomLiteralDoesNotRequireIataTokenOrDevice) {
char topic[32];
ASSERT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_RAW,
"private/raw", nullptr, nullptr, nullptr,
topic, sizeof(topic)));
EXPECT_STREQ("private/raw", topic);
}
TEST(MQTTTopicRouter, EmptyCustomTemplateFailsRatherThanFallingBackImplicitly) {
char topic[64];
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_STATUS,
"", IATA, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_STREQ("", topic);
// MQTTBridge selects the MeshCore style explicitly for a custom slot whose
// template is empty; make that default contract visible here.
EXPECT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS,
nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_STREQ("meshcore/DEN/0123456789ABCDEF/status", topic);
}
TEST(MQTTTopicRouter, FormattedTopicsRequireRoomForTerminator) {
const char* expected = "meshcore/DEN/0123456789ABCDEF/status";
const size_t exact_size = strlen(expected) + 1;
char exact[64];
ASSERT_LE(exact_size, sizeof(exact));
EXPECT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS,
nullptr, IATA, DEVICE, TOKEN, exact, exact_size));
EXPECT_STREQ(expected, exact);
char short_buf[64];
memset(short_buf, 0x7f, sizeof(short_buf));
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS,
nullptr, IATA, DEVICE, TOKEN,
short_buf, exact_size - 1));
EXPECT_EQ('\0', short_buf[exact_size - 2]);
}
TEST(MQTTTopicRouter, CustomTopicHonorsExactBoundary) {
const char* expected = "custom/DEN/raw";
char exact[15];
static_assert(sizeof(exact) == 15, "fixture includes the terminator");
EXPECT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_RAW,
"custom/{iata}/{type}", IATA, DEVICE, TOKEN,
exact, sizeof(exact)));
EXPECT_STREQ(expected, exact);
char short_buf[14];
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_RAW,
"custom/{iata}/{type}", IATA, DEVICE, TOKEN,
short_buf, sizeof(short_buf)));
EXPECT_LT(strlen(short_buf), sizeof(short_buf));
}
TEST(MQTTTopicRouter, RejectsInvalidStyleTypeSlotAndOutput) {
char topic[64] = "dirty";
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, 99, nullptr,
IATA, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_STREQ("", topic);
EXPECT_FALSE(mqttBuildPublicationTopic(static_cast<MQTTTopicRouteStyle>(99),
MQTT_PUBLICATION_STATUS, nullptr,
IATA, DEVICE, TOKEN, topic, sizeof(topic)));
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS,
nullptr, IATA, DEVICE, TOKEN, nullptr, 64));
EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS,
nullptr, IATA, DEVICE, TOKEN, topic, 0));
EXPECT_FALSE(mqttTopicSlotIndexValid(-1, RUNTIME_MQTT_SLOTS));
EXPECT_TRUE(mqttTopicSlotIndexValid(0, RUNTIME_MQTT_SLOTS));
EXPECT_TRUE(mqttTopicSlotIndexValid(RUNTIME_MQTT_SLOTS - 1, RUNTIME_MQTT_SLOTS));
EXPECT_FALSE(mqttTopicSlotIndexValid(RUNTIME_MQTT_SLOTS, RUNTIME_MQTT_SLOTS));
EXPECT_FALSE(mqttTopicSlotIndexValid(0, 0));
}
} // namespace
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
variants/heltec_t190/platformio.ini
+2 -2
View File
@@ -128,7 +128,7 @@ lib_deps =
${Heltec_T190_base.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -221,7 +221,7 @@ lib_deps =
${Heltec_T190_base.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/heltec_tracker_v2/platformio.ini
+4 -4
View File
@@ -200,7 +200,7 @@ lib_deps =
${Heltec_tracker_v1_1.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -236,7 +236,7 @@ lib_deps =
${Heltec_tracker_v2.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -272,7 +272,7 @@ lib_deps =
${Heltec_tracker_v1_1.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -308,7 +308,7 @@ lib_deps =
${Heltec_tracker_v2.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/heltec_v3/platformio.ini
+4 -4
View File
@@ -153,7 +153,7 @@ lib_deps =
${Heltec_lora32_v3.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
0neblock/SNMP_Agent
@@ -223,7 +223,7 @@ lib_deps =
${Heltec_lora32_v3.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -522,7 +522,7 @@ lib_deps =
${Heltec_lora32_v3.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
0neblock/SNMP_Agent
@@ -562,7 +562,7 @@ lib_deps =
${Heltec_lora32_v3.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/heltec_v4/platformio.ini
+4 -4
View File
@@ -192,7 +192,7 @@ lib_deps =
${heltec_v4_oled.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -246,7 +246,7 @@ lib_deps =
${heltec_v4_oled.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -336,7 +336,7 @@ lib_deps =
${heltec_v4_oled.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -378,7 +378,7 @@ lib_deps =
${heltec_v4_oled.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/lilygo_t3s3/platformio.ini
+2 -2
View File
@@ -140,7 +140,7 @@ lib_deps =
${LilyGo_T3S3_sx1262.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -180,7 +180,7 @@ lib_deps =
${LilyGo_T3S3_sx1262.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/lilygo_tbeam_1w/platformio.ini
+2 -2
View File
@@ -235,7 +235,7 @@ lib_deps =
${LilyGo_TBeam_1W.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -276,7 +276,7 @@ lib_deps =
${LilyGo_TBeam_1W.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/lilygo_tbeam_SX1262/platformio.ini
+2 -2
View File
@@ -184,7 +184,7 @@ lib_deps =
${LilyGo_TBeam_SX1262.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -221,7 +221,7 @@ lib_deps =
${LilyGo_TBeam_SX1262.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/lilygo_tbeam_SX1276/platformio.ini
+2 -2
View File
@@ -183,7 +183,7 @@ lib_deps =
${LilyGo_TBeam_SX1276.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -220,7 +220,7 @@ lib_deps =
${LilyGo_TBeam_SX1276.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/lilygo_tbeam_supreme_SX1262/platformio.ini
+2 -2
View File
@@ -143,7 +143,7 @@ lib_deps =
${T_Beam_S3_Supreme_SX1262.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -180,7 +180,7 @@ lib_deps =
${T_Beam_S3_Supreme_SX1262.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/lilygo_tlora_v2_1/platformio.ini
+2 -2
View File
@@ -182,7 +182,7 @@ lib_deps =
${LilyGo_TLora_V2_1_1_6_core.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -225,7 +225,7 @@ lib_deps =
${LilyGo_TLora_V2_1_1_6_core.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/rak3112/platformio.ini
+2 -2
View File
@@ -139,7 +139,7 @@ lib_deps =
${rak3112.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
0neblock/SNMP_Agent
@@ -195,7 +195,7 @@ lib_deps =
${rak3112.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/station_g2/platformio.ini
+2 -2
View File
@@ -205,7 +205,7 @@ lib_deps =
${Station_G2.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -332,7 +332,7 @@ lib_deps =
${Station_G2.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
variants/xiao_s3_wio/platformio.ini
+2 -2
View File
@@ -127,7 +127,7 @@ lib_deps =
${Xiao_S3_WIO.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1
@@ -164,7 +164,7 @@ lib_deps =
${Xiao_S3_WIO.lib_deps}
${esp32_ota.lib_deps}
elims/PsychicMqttClient@^0.2.4
bblanchon/ArduinoJson
bblanchon/ArduinoJson @ 7.4.3
arduino-libraries/NTPClient
JChristensen/Timezone
paulstoffregen/Time@1.6.1