Skip to content

[Feature] Add MQTT bridge #3030

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
hickey wants to merge 5 commits into from
Closed

[Feature] Add MQTT bridge #3030

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8ceaca3
feat: Add MQTT bridging support
hickey Jul 21, 2026
679955a
Add mqtt command to CLI
hickey Jul 22, 2026
f3fd749
Add mqtt CLI commands to documentation
hickey Jul 22, 2026
7c39190
Properly reinitialize MQTT connection after changes
hickey Jul 22, 2026
96cc91d
Add support for bridge source of rx and tx
hickey Jul 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 85 additions & 45 deletions docs/cli_commands.md
View file
Open in desktop

Large diffs are not rendered by default.

21 changes: 12 additions & 9 deletions examples/simple_repeater/MyMesh.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,8 +199,10 @@ uint8_t MyMesh::handleAnonClockReq(const mesh::Identity& sender, uint32_t sender
reply_data[8] = 0; // features
#ifdef WITH_RS232_BRIDGE
reply_data[8] |= 0x01; // is bridge, type UART
#elif WITH_ESPNOW_BRIDGE
#elif defined(WITH_ESPNOW_BRIDGE)
reply_data[8] |= 0x03; // is bridge, type ESP-NOW
#elif defined(WITH_MQTT_BRIDGE)
reply_data[8] |= MQTT_BRIDGE; // is bridge, type MQTT
#endif
if (_prefs.disable_fwd) { // is this repeater currently disabled
reply_data[8] |= 0x80; // is disabled
Expand Down Expand Up @@ -344,7 +346,7 @@ int MyMesh::handleRequest(ClientInfo *sender, uint32_t sender_timestamp, uint8_t
int results_offset = 0;
uint8_t results_buffer[130];
for(int index = 0; index < count && index + offset < neighbours_count; index++){

// stop if we can't fit another entry in results
int entry_size = pubkey_prefix_length + 4 + 1;
if(results_offset + entry_size > sizeof(results_buffer)){
Expand Down Expand Up @@ -474,7 +476,7 @@ void MyMesh::logRxRaw(float snr, float rssi, const uint8_t raw[], int len) {

void MyMesh::logRx(mesh::Packet *pkt, int len, float score) {
#ifdef WITH_BRIDGE
if (_prefs.bridge_pkt_src == 1) {
if (_prefs.bridge_pkt_src & BRIDGE_SOURCE_RX) {
bridge.sendPacket(pkt);
}
#endif
Expand All @@ -500,7 +502,7 @@ void MyMesh::logRx(mesh::Packet *pkt, int len, float score) {

void MyMesh::logTx(mesh::Packet *pkt, int len) {
#ifdef WITH_BRIDGE
if (_prefs.bridge_pkt_src == 0) {
if (_prefs.bridge_pkt_src & BRIDGE_SOURCE_TX) {
bridge.sendPacket(pkt);
}
#endif
Expand Down Expand Up @@ -847,15 +849,16 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
mesh::RTCClock &rtc, mesh::MeshTables &tables)
: mesh::Mesh(radio, ms, rng, rtc, *new StaticPoolPacketManager(32), tables),
region_map(key_store), temp_map(key_store),
_cli(board, rtc, sensors, region_map, acl, &_prefs, this),
_cli(board, rtc, bridge, sensors, region_map, acl, &_prefs, this),
telemetry(MAX_PACKET_PAYLOAD - 4),
discover_limiter(4, 120), // max 4 every 2 minutes
anon_limiter(4, 180) // max 4 every 3 minutes
#if defined(WITH_RS232_BRIDGE)
, bridge(&_prefs, WITH_RS232_BRIDGE, _mgr, &rtc)
#endif
#if defined(WITH_ESPNOW_BRIDGE)
#elif defined(WITH_ESPNOW_BRIDGE)
, bridge(&_prefs, _mgr, &rtc)
#elif defined(WITH_MQTT_BRIDGE)
, bridge(&_prefs, _mgr, &rtc, self_id.pub_key)
#endif
{
last_millis = 0;
Expand Down Expand Up @@ -897,7 +900,7 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
// bridge defaults
_prefs.bridge_enabled = 1; // enabled
_prefs.bridge_delay = 500; // milliseconds
_prefs.bridge_pkt_src = 0; // logTx
_prefs.bridge_pkt_src = BRIDGE_SOURCE_TX; // logTx
_prefs.bridge_baud = 115200; // baud rate
_prefs.bridge_channel = 1; // channel 1

Expand Down Expand Up @@ -1148,7 +1151,7 @@ void MyMesh::formatRadioStatsReply(char *reply) {
}

void MyMesh::formatPacketStatsReply(char *reply) {
StatsFormatHelper::formatPacketStats(reply, radio_driver, getNumSentFlood(), getNumSentDirect(),
StatsFormatHelper::formatPacketStats(reply, radio_driver, getNumSentFlood(), getNumSentDirect(),
getNumRecvFlood(), getNumRecvDirect());
}

Expand Down
10 changes: 9 additions & 1 deletion examples/simple_repeater/MyMesh.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,11 @@
#define WITH_BRIDGE
#endif

#ifdef WITH_MQTT_BRIDGE
#include "helpers/bridges/MQTTBridge.h"
#define WITH_BRIDGE
#endif

#include <helpers/AdvertDataHelpers.h>
#include <helpers/ArduinoHelpers.h>
#include <helpers/ClientACL.h>
Expand All @@ -37,6 +42,7 @@

#ifdef WITH_BRIDGE
extern AbstractBridge* bridge;
#define MQTT_BRIDGE 0x06
#endif

struct RepeaterStats {
Expand Down Expand Up @@ -117,6 +123,8 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
RS232Bridge bridge;
#elif defined(WITH_ESPNOW_BRIDGE)
ESPNowBridge bridge;
#elif defined(WITH_MQTT_BRIDGE)
MQTTBridge bridge;
#endif

void putNeighbour(const mesh::Identity& id, uint32_t timestamp, float snr);
Expand Down Expand Up @@ -236,7 +244,7 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
{
bridge.begin();
}
else
else
{
bridge.end();
}
Expand Down
67 changes: 60 additions & 7 deletions src/helpers/CommonCLI.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,10 @@
#include "TxtDataHelpers.h"
#include <RTClib.h>

#ifdef WITH_MQTT_BRIDGE
#include <helpers/bridges/MQTTBridge.h>
#endif

#ifndef BRIDGE_MAX_BAUD
#define BRIDGE_MAX_BAUD 115200
#endif
Expand Down Expand Up @@ -112,7 +116,7 @@ void CommonCLI::loadPrefsInt(FILESYSTEM* fs, const char* filename) {
// sanitise bad bridge pref values
_prefs->bridge_enabled = constrain(_prefs->bridge_enabled, 0, 1);
_prefs->bridge_delay = constrain(_prefs->bridge_delay, 0, 10000);
_prefs->bridge_pkt_src = constrain(_prefs->bridge_pkt_src, 0, 1);
_prefs->bridge_pkt_src = constrain(_prefs->bridge_pkt_src, 0, 3); // log Rx and Tx = 3
_prefs->bridge_baud = constrain(_prefs->bridge_baud, 9600, BRIDGE_MAX_BAUD);
_prefs->bridge_channel = constrain(_prefs->bridge_channel, 0, 14);

Expand Down Expand Up @@ -654,7 +658,7 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep
strcpy(reply, "OK");
} else {
strcpy(reply, "Error, max 64");
}
}
} else if (memcmp(config, "flood.max.advert ", 17) == 0) {
uint8_t m = atoi(&config[17]);
if (m <= 64) {
Expand Down Expand Up @@ -747,9 +751,23 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep
strcpy(reply, "Error: delay must be between 0-10000 ms");
}
} else if (memcmp(config, "bridge.source ", 14) == 0) {
_prefs->bridge_pkt_src = memcmp(&config[14], "rx", 2) == 0;
savePrefs();
strcpy(reply, "OK");
if (memcmp(&config[14], "log", 3) == 0) {
if (memcmp(&config[17], "Rx", 2) == 0 || memcmp(&config[17], "rx", 2) == 0) {
_prefs->bridge_pkt_src = BRIDGE_SOURCE_RX;
strcpy(reply, "OK");
} else if (memcmp(&config[17], "Tx", 2) == 0 || memcmp(&config[17], "tx", 2) == 0) {
_prefs->bridge_pkt_src = BRIDGE_SOURCE_TX;
strcpy(reply, "OK");
} else if (memcmp(&config[17], "Both", 4) == 0 || memcmp(&config[17], "both", 4) == 0) {
_prefs->bridge_pkt_src = BRIDGE_SOURCE_TX | BRIDGE_SOURCE_RX;
strcpy(reply, "OK");
}
}
if (memcmp(reply, "OK", 2) == 0) {
savePrefs();
} else {
strcpy(reply, "Error: source must be logRx, logTx or logBoth.");
}
#endif
#ifdef WITH_RS232_BRIDGE
} else if (memcmp(config, "bridge.baud ", 12) == 0) {
Expand Down Expand Up @@ -779,6 +797,31 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep
_callbacks->restartBridge();
savePrefs();
strcpy(reply, "OK");
#endif
#ifdef WITH_MQTT_BRIDGE
} else if (memcmp(config, "mqtt.host ", 10) == 0) {
static_cast<MQTTBridge*>(_bridge)->end();
StrHelper::strncpy(static_cast<MQTTBridge*>(_bridge)->mqtt_host,
&config[10], 64);
static_cast<MQTTBridge*>(_bridge)->initialize();
static_cast<MQTTBridge*>(_bridge)->reconnect();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.port ", 10) == 0) {
static_cast<MQTTBridge*>(_bridge)->end();
static_cast<MQTTBridge*>(_bridge)->mqtt_port = atoi(&config[10]);
static_cast<MQTTBridge*>(_bridge)->initialize();
static_cast<MQTTBridge*>(_bridge)->reconnect();
strcpy(reply, "OK");
} else if (memcmp(config, "mqtt.topic ", 11) == 0) {
static_cast<MQTTBridge*>(_bridge)->end();
StrHelper::strncpy(static_cast<MQTTBridge*>(_bridge)->mqtt_topic,
&config[11], 64);
static_cast<MQTTBridge*>(_bridge)->initialize();
if (static_cast<MQTTBridge*>(_bridge)->reconnect()) {
strcpy(reply, "OK");
} else {
strcpy(reply, "Error: failed to subscribe to new topic");
}
#endif
} else if (memcmp(config, "adc.multiplier ", 15) == 0) {
_prefs->adc_multiplier = atof(&config[15]);
Expand Down Expand Up @@ -899,6 +942,8 @@ void CommonCLI::handleGetCmd(uint32_t sender_timestamp, char* command, char* rep
"rs232"
#elif WITH_ESPNOW_BRIDGE
"espnow"
#elif WITH_MQTT_BRIDGE
"mqtt"
#else
"none"
#endif
Expand All @@ -920,6 +965,14 @@ void CommonCLI::handleGetCmd(uint32_t sender_timestamp, char* command, char* rep
sprintf(reply, "> %d", (uint32_t)_prefs->bridge_channel);
} else if (memcmp(config, "bridge.secret", 13) == 0) {
sprintf(reply, "> %s", _prefs->bridge_secret);
#endif
#ifdef WITH_MQTT_BRIDGE
} else if (memcmp(config, "mqtt.host", 9) == 0) {
sprintf(reply, "> %s", static_cast<MQTTBridge*>(_bridge)->mqtt_host);
} else if (memcmp(config, "mqtt.port", 9) == 0) {
sprintf(reply, "> %s", static_cast<MQTTBridge*>(_bridge)->mqtt_port);
} else if (memcmp(config, "mqtt.topic", 10) == 0) {
sprintf(reply, "> %s", static_cast<MQTTBridge*>(_bridge)->mqtt_topic);
#endif
} else if (memcmp(config, "bootloader.ver", 14) == 0) {
#ifdef NRF52_PLATFORM
Expand Down Expand Up @@ -1138,7 +1191,7 @@ void CommonCLI::handleRegionCmd(char* command, char* reply) {
} else if (n >= 3 && strcmp(parts[1], "list") == 0) {
uint8_t mask = 0;
bool invert = false;

if (strcmp(parts[2], "allowed") == 0) {
mask = REGION_DENY_FLOOD;
invert = false; // list regions that DON'T have DENY flag
Expand All @@ -1149,7 +1202,7 @@ void CommonCLI::handleRegionCmd(char* command, char* reply) {
strcpy(reply, "Err - use 'allowed' or 'denied'");
return;
}

int len = _region_map->exportNamesTo(reply, 160, mask, invert);
if (len == 0) {
strcpy(reply, "-none-");
Expand Down
12 changes: 8 additions & 4 deletions src/helpers/CommonCLI.h
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,16 @@
#pragma once

#include "Mesh.h"
#include "AbstractBridge.h"
#include <helpers/IdentityStore.h>
#include <helpers/SensorManager.h>
#include <helpers/ClientACL.h>
#include <helpers/RegionMap.h>

#if defined(WITH_RS232_BRIDGE) || defined(WITH_ESPNOW_BRIDGE)
#if defined(WITH_RS232_BRIDGE) || defined(WITH_ESPNOW_BRIDGE) || defined(WITH_MQTT_BRIDGE)
#define WITH_BRIDGE
#define BRIDGE_SOURCE_RX 0x01
#define BRIDGE_SOURCE_TX 0x02
#endif

#define ADVERT_LOC_NONE 0
Expand Down Expand Up @@ -47,7 +50,7 @@ struct NodePrefs { // persisted to file
// Bridge settings
uint8_t bridge_enabled; // boolean
uint16_t bridge_delay; // milliseconds (default 500 ms)
uint8_t bridge_pkt_src; // 0 = logTx, 1 = logRx (default logTx)
uint8_t bridge_pkt_src; // 1 = logTx, 2 = logRx (default logTx)
uint32_t bridge_baud; // 9600, 19200, 38400, 57600, 115200 (default 115200)
uint8_t bridge_channel; // 1-14 (ESP-NOW only)
char bridge_secret[16]; // for XOR encryption of bridge packets (ESP-NOW only)
Expand Down Expand Up @@ -121,6 +124,7 @@ class CommonCLI {
NodePrefs* _prefs;
CommonCLICallbacks* _callbacks;
mesh::MainBoard* _board;
AbstractBridge* _bridge;
SensorManager* _sensors;
RegionMap* _region_map;
ClientACL* _acl;
Expand All @@ -135,8 +139,8 @@ class CommonCLI {
void handleSetCmd(uint32_t sender_timestamp, char* command, char* reply);

public:
CommonCLI(mesh::MainBoard& board, mesh::RTCClock& rtc, SensorManager& sensors, RegionMap& region_map, ClientACL& acl, NodePrefs* prefs, CommonCLICallbacks* callbacks)
: _board(&board), _rtc(&rtc), _sensors(&sensors), _region_map(&region_map), _acl(&acl), _prefs(prefs), _callbacks(callbacks) { }
CommonCLI(mesh::MainBoard& board, mesh::RTCClock& rtc, AbstractBridge& bridge, SensorManager& sensors, RegionMap& region_map, ClientACL& acl, NodePrefs* prefs, CommonCLICallbacks* callbacks)
: _board(&board), _rtc(&rtc), _bridge(&bridge), _sensors(&sensors), _region_map(&region_map), _acl(&acl), _prefs(prefs), _callbacks(callbacks) { }

void loadPrefs(FILESYSTEM* _fs);
void savePrefs(FILESYSTEM* _fs);
Expand Down
Loading