MQTT telemetry (build with make MQTT=1)
---------------------------------------
Add optional MQTT publishing to DStarRepeater, following the same
conventions as MMDVMHost: CMQTTConnection wrapping libmosquitto,
topic-prefix convention ({name}/topic), log-level filtering, PID-based
client IDs, and matching config key names. All MQTT code is guarded by
#if defined(MQTT) — without the flag, the binary is unchanged.
Three topics are published:
{name}/log Timestamped log messages filtered by severity
{name}/status JSON repeater state snapshot, once per second
{name}/json Display-Driver-compatible events at state transitions
Display-Driver compatibility
----------------------------
Publish event-driven JSON on the "json" topic in the exact format that
Display-Driver expects, enabling OLED/TFT display output with no
changes to Display-Driver. Events: D-Star start (RF/net with callsign
fields), end, lost (watchdog), idle, and BER. Hooks added at every
state transition point in all four thread variants (TRXThread,
RXThread, TXThread, TXRXThread).
Shutdown safety
---------------
Add mosquitto_loop_stop() before mosquitto_destroy() in close() to
prevent use-after-free from the background network thread. Add Wait()
after kill() in OnExit() so the repeater thread has fully exited
before tearing down MQTT. Both issues also exist in MMDVMHost.
Config parser bug fix
---------------------
The config file parser called GetNextLine() inside comment and
no-equals handlers before continuing, but the for-loop increment
already advances — silently skipping the line after every comment or
blank line. Remove the redundant calls and add an empty-line guard.
Comments and blank lines now work correctly in config files.
Documentation and config
------------------------
Add MQTT.md with build instructions, config keys, topic structure,
JSON examples, and Display-Driver compatibility details. Add
CONFIGURATION.md as a full reference guide for every config setting.
Update README.md with gateway links (ircDDBGateway, DStarGateway) and
MQTT build instructions. Rename and update the example config with
corrected defaults, missing icomPort, and MQTT settings.
pull/16/head
Andy Taylor
5 months ago
parent
fd7f1a181c
commit
6dc8097d20
@ -0,0 +1,236 @@
|
||||
/*
|
||||
* Copyright (C) 2022,2023,2025 by Jonathan Naylor G4KLX
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License as published by
|
||||
* the Free Software Foundation; either version 2 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License
|
||||
* along with this program; if not, write to the Free Software
|
||||
* Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
|
||||
*/
|
||||
|
||||
#if defined(MQTT)
|
||||
|
||||
#include "MQTTConnection.h"
|
||||
|
||||
#include <cassert>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
|
||||
#if defined(_WIN32) || defined(_WIN64)
|
||||
#include <process.h>
|
||||
#else
|
||||
#include <unistd.h>
|
||||
#endif
|
||||
|
||||
|
||||
CMQTTConnection::CMQTTConnection(const std::string& host, unsigned short port, const std::string& name, const bool authEnabled, const std::string& username, const std::string& password, const std::vector<std::pair<std::string, void (*)(const unsigned char*, unsigned int)>>& subs, unsigned int keepalive, MQTT_QOS qos) :
|
||||
m_host(host),
|
||||
m_port(port),
|
||||
m_name(name),
|
||||
m_authEnabled(authEnabled),
|
||||
m_username(username),
|
||||
m_password(password),
|
||||
m_subs(subs),
|
||||
m_keepalive(keepalive),
|
||||
m_qos(qos),
|
||||
m_mosq(nullptr),
|
||||
m_connected(false)
|
||||
{
|
||||
assert(!host.empty());
|
||||
assert(port > 0U);
|
||||
assert(!name.empty());
|
||||
assert(keepalive >= 5U);
|
||||
|
||||
::mosquitto_lib_init();
|
||||
}
|
||||
|
||||
CMQTTConnection::~CMQTTConnection()
|
||||
{
|
||||
::mosquitto_lib_cleanup();
|
||||
}
|
||||
|
||||
bool CMQTTConnection::open()
|
||||
{
|
||||
char name[50U];
|
||||
#if defined(_WIN32) || defined(_WIN64)
|
||||
::sprintf(name, "DStarRepeater.%u", (unsigned)::_getpid());
|
||||
#else
|
||||
::sprintf(name, "DStarRepeater.%u", (unsigned)::getpid());
|
||||
#endif
|
||||
|
||||
::fprintf(stdout, "DStarRepeater (%s) connecting to MQTT as %s\n", m_name.c_str(), name);
|
||||
|
||||
m_mosq = ::mosquitto_new(name, true, this);
|
||||
if (m_mosq == nullptr) {
|
||||
::fprintf(stderr, "MQTT Error newing: Out of memory.\n");
|
||||
return false;
|
||||
}
|
||||
|
||||
if (m_authEnabled)
|
||||
::mosquitto_username_pw_set(m_mosq, m_username.c_str(), m_password.c_str());
|
||||
|
||||
::mosquitto_connect_callback_set(m_mosq, onConnect);
|
||||
::mosquitto_subscribe_callback_set(m_mosq, onSubscribe);
|
||||
::mosquitto_message_callback_set(m_mosq, onMessage);
|
||||
::mosquitto_disconnect_callback_set(m_mosq, onDisconnect);
|
||||
|
||||
int rc = ::mosquitto_connect(m_mosq, m_host.c_str(), m_port, m_keepalive);
|
||||
if (rc != MOSQ_ERR_SUCCESS) {
|
||||
::mosquitto_destroy(m_mosq);
|
||||
m_mosq = nullptr;
|
||||
::fprintf(stderr, "MQTT Error connecting: %s\n", ::mosquitto_strerror(rc));
|
||||
return false;
|
||||
}
|
||||
|
||||
rc = ::mosquitto_loop_start(m_mosq);
|
||||
if (rc != MOSQ_ERR_SUCCESS) {
|
||||
::mosquitto_disconnect(m_mosq);
|
||||
::mosquitto_destroy(m_mosq);
|
||||
m_mosq = nullptr;
|
||||
::fprintf(stderr, "MQTT Error loop starting: %s\n", ::mosquitto_strerror(rc));
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool CMQTTConnection::publish(const char* topic, const char* text)
|
||||
{
|
||||
assert(topic != nullptr);
|
||||
assert(text != nullptr);
|
||||
|
||||
return publish(topic, (unsigned char*)text, (unsigned int)::strlen(text));
|
||||
}
|
||||
|
||||
bool CMQTTConnection::publish(const char* topic, const std::string& text)
|
||||
{
|
||||
assert(topic != nullptr);
|
||||
|
||||
return publish(topic, (unsigned char*)text.c_str(), (unsigned int)text.size());
|
||||
}
|
||||
|
||||
bool CMQTTConnection::publish(const char* topic, const unsigned char* data, unsigned int len)
|
||||
{
|
||||
assert(topic != nullptr);
|
||||
assert(data != nullptr);
|
||||
|
||||
if (!m_connected)
|
||||
return false;
|
||||
|
||||
if (::strchr(topic, '/') == nullptr) {
|
||||
char topicEx[100U];
|
||||
::sprintf(topicEx, "%s/%s", m_name.c_str(), topic);
|
||||
|
||||
int rc = ::mosquitto_publish(m_mosq, nullptr, topicEx, len, data, static_cast<int>(m_qos), false);
|
||||
if (rc != MOSQ_ERR_SUCCESS) {
|
||||
::fprintf(stderr, "MQTT Error publishing: %s\n", ::mosquitto_strerror(rc));
|
||||
return false;
|
||||
}
|
||||
} else {
|
||||
int rc = ::mosquitto_publish(m_mosq, nullptr, topic, len, data, static_cast<int>(m_qos), false);
|
||||
if (rc != MOSQ_ERR_SUCCESS) {
|
||||
::fprintf(stderr, "MQTT Error publishing: %s\n", ::mosquitto_strerror(rc));
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
void CMQTTConnection::close()
|
||||
{
|
||||
if (m_mosq != nullptr) {
|
||||
::mosquitto_disconnect(m_mosq);
|
||||
::mosquitto_loop_stop(m_mosq, true);
|
||||
::mosquitto_destroy(m_mosq);
|
||||
m_mosq = nullptr;
|
||||
}
|
||||
}
|
||||
|
||||
void CMQTTConnection::onConnect(mosquitto* mosq, void* obj, int rc)
|
||||
{
|
||||
assert(mosq != nullptr);
|
||||
assert(obj != nullptr);
|
||||
|
||||
::fprintf(stdout, "MQTT: on_connect: %s\n", ::mosquitto_connack_string(rc));
|
||||
if (rc != 0) {
|
||||
::mosquitto_disconnect(mosq);
|
||||
return;
|
||||
}
|
||||
|
||||
CMQTTConnection* p = static_cast<CMQTTConnection*>(obj);
|
||||
p->m_connected = true;
|
||||
|
||||
for (std::vector<std::pair<std::string, void (*)(const unsigned char*, unsigned int)>>::const_iterator it = p->m_subs.cbegin(); it != p->m_subs.cend(); ++it) {
|
||||
std::string topic = (*it).first;
|
||||
|
||||
if (topic.find_first_of('/') == std::string::npos) {
|
||||
char topicEx[100U];
|
||||
::sprintf(topicEx, "%s/%s", p->m_name.c_str(), topic.c_str());
|
||||
|
||||
rc = ::mosquitto_subscribe(mosq, nullptr, topicEx, static_cast<int>(p->m_qos));
|
||||
if (rc != MOSQ_ERR_SUCCESS) {
|
||||
::fprintf(stderr, "MQTT: error subscribing to %s - %s\n", topicEx, ::mosquitto_strerror(rc));
|
||||
::mosquitto_disconnect(mosq);
|
||||
}
|
||||
} else {
|
||||
rc = ::mosquitto_subscribe(mosq, nullptr, topic.c_str(), static_cast<int>(p->m_qos));
|
||||
if (rc != MOSQ_ERR_SUCCESS) {
|
||||
::fprintf(stderr, "MQTT: error subscribing to %s - %s\n", topic.c_str(), ::mosquitto_strerror(rc));
|
||||
::mosquitto_disconnect(mosq);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void CMQTTConnection::onSubscribe(mosquitto* mosq, void* obj, int mid, int qosCount, const int* grantedQOS)
|
||||
{
|
||||
assert(mosq != nullptr);
|
||||
assert(obj != nullptr);
|
||||
assert(grantedQOS != nullptr);
|
||||
|
||||
for (int i = 0; i < qosCount; i++)
|
||||
::fprintf(stdout, "MQTT: on_subscribe: %d:%d\n", i, grantedQOS[i]);
|
||||
}
|
||||
|
||||
void CMQTTConnection::onMessage(mosquitto* mosq, void* obj, const mosquitto_message* message)
|
||||
{
|
||||
assert(mosq != nullptr);
|
||||
assert(obj != nullptr);
|
||||
assert(message != nullptr);
|
||||
|
||||
CMQTTConnection* p = static_cast<CMQTTConnection*>(obj);
|
||||
|
||||
for (std::vector<std::pair<std::string, void (*)(const unsigned char*, unsigned int)>>::const_iterator it = p->m_subs.cbegin(); it != p->m_subs.cend(); ++it) {
|
||||
std::string topic = (*it).first;
|
||||
|
||||
char topicEx[100U];
|
||||
::sprintf(topicEx, "%s/%s", p->m_name.c_str(), topic.c_str());
|
||||
|
||||
if (::strcmp(topicEx, message->topic) == 0) {
|
||||
(*it).second((unsigned char*)message->payload, message->payloadlen);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void CMQTTConnection::onDisconnect(mosquitto* mosq, void* obj, int rc)
|
||||
{
|
||||
assert(mosq != nullptr);
|
||||
assert(obj != nullptr);
|
||||
|
||||
::fprintf(stdout, "MQTT: on_disconnect: %s\n", ::mosquitto_reason_string(rc));
|
||||
|
||||
CMQTTConnection* p = static_cast<CMQTTConnection*>(obj);
|
||||
p->m_connected = false;
|
||||
}
|
||||
|
||||
#endif
|
||||
@ -0,0 +1,69 @@
|
||||
/*
|
||||
* Copyright (C) 2022,2023,2025 by Jonathan Naylor G4KLX
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License as published by
|
||||
* the Free Software Foundation; either version 2 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License
|
||||
* along with this program; if not, write to the Free Software
|
||||
* Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
|
||||
*/
|
||||
|
||||
#if !defined(MQTTCONNECTION_H)
|
||||
#define MQTTCONNECTION_H
|
||||
|
||||
#if defined(MQTT)
|
||||
|
||||
#include <mosquitto.h>
|
||||
|
||||
#include <vector>
|
||||
#include <string>
|
||||
|
||||
enum class MQTT_QOS : int {
|
||||
AT_MODE_ONCE = 0,
|
||||
AT_LEAST_ONCE = 1,
|
||||
EXACTLY_ONCE = 2
|
||||
};
|
||||
|
||||
class CMQTTConnection {
|
||||
public:
|
||||
CMQTTConnection(const std::string& host, unsigned short port, const std::string& name, const bool authEnabled, const std::string& username, const std::string& password, const std::vector<std::pair<std::string, void (*)(const unsigned char*, unsigned int)>>& subs, unsigned int keepalive, MQTT_QOS qos = MQTT_QOS::EXACTLY_ONCE);
|
||||
~CMQTTConnection();
|
||||
|
||||
bool open();
|
||||
|
||||
bool publish(const char* topic, const char* text);
|
||||
bool publish(const char* topic, const std::string& text);
|
||||
bool publish(const char* topic, const unsigned char* data, unsigned int len);
|
||||
|
||||
void close();
|
||||
|
||||
private:
|
||||
std::string m_host;
|
||||
unsigned short m_port;
|
||||
std::string m_name;
|
||||
bool m_authEnabled;
|
||||
std::string m_username;
|
||||
std::string m_password;
|
||||
std::vector<std::pair<std::string, void (*)(const unsigned char*, unsigned int)>> m_subs;
|
||||
unsigned int m_keepalive;
|
||||
MQTT_QOS m_qos;
|
||||
mosquitto* m_mosq;
|
||||
bool m_connected;
|
||||
|
||||
static void onConnect(mosquitto* mosq, void* obj, int rc);
|
||||
static void onSubscribe(mosquitto* mosq, void* obj, int mid, int qosCount, const int* grantedQOS);
|
||||
static void onMessage(mosquitto* mosq, void* obj, const mosquitto_message* message);
|
||||
static void onDisconnect(mosquitto* mosq, void* obj, int rc);
|
||||
};
|
||||
|
||||
#endif
|
||||
|
||||
#endif
|
||||
@ -0,0 +1,121 @@
|
||||
/*
|
||||
* Copyright (C) 2025 by Andy Taylor MW0MWZ
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License as published by
|
||||
* the Free Software Foundation; either version 2 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License
|
||||
* along with this program; if not, write to the Free Software
|
||||
* Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
|
||||
*/
|
||||
|
||||
#ifndef MQTTPublisher_H
|
||||
#define MQTTPublisher_H
|
||||
|
||||
#if defined(MQTT)
|
||||
|
||||
/*
|
||||
* Display-Driver-compatible MQTT JSON publishing for DStarRepeater.
|
||||
*
|
||||
* Publishes event-driven JSON messages to the "json" topic in the format
|
||||
* expected by Display-Driver, enabling LCD/OLED display output.
|
||||
*
|
||||
* These are separate from the periodic status messages on the "status"
|
||||
* topic — the display needs discrete start/end events, not continuous
|
||||
* state snapshots.
|
||||
*/
|
||||
|
||||
#include "MQTTConnection.h"
|
||||
#include "Logger.h"
|
||||
|
||||
#include <wx/wx.h>
|
||||
|
||||
#include <cstdio>
|
||||
|
||||
// Publish a D-Star call start event (RF or network)
|
||||
static inline void mqttPublishDStarStart(const wxString& myCall1, const wxString& myCall2,
|
||||
const wxString& yourCall, const wxString& rptCall2, const char* source)
|
||||
{
|
||||
if (g_mqtt == NULL)
|
||||
return;
|
||||
|
||||
char buf[512];
|
||||
::snprintf(buf, sizeof(buf),
|
||||
"{\"D-Star\":{\"action\":\"start\","
|
||||
"\"source_cs\":\"%s\",\"source_ext\":\"%s\","
|
||||
"\"destination_cs\":\"%s\",\"reflector\":\"%s\","
|
||||
"\"source\":\"%s\"}}",
|
||||
(const char*)myCall1.mb_str(),
|
||||
(const char*)myCall2.mb_str(),
|
||||
(const char*)yourCall.mb_str(),
|
||||
(const char*)rptCall2.mb_str(),
|
||||
source);
|
||||
|
||||
g_mqtt->publish("json", buf);
|
||||
}
|
||||
|
||||
// Publish a D-Star call end event (normal termination)
|
||||
static inline void mqttPublishDStarEnd()
|
||||
{
|
||||
if (g_mqtt == NULL)
|
||||
return;
|
||||
|
||||
g_mqtt->publish("json", "{\"D-Star\":{\"action\":\"end\"}}");
|
||||
}
|
||||
|
||||
// Publish a D-Star call lost event (watchdog timeout / abnormal)
|
||||
static inline void mqttPublishDStarLost()
|
||||
{
|
||||
if (g_mqtt == NULL)
|
||||
return;
|
||||
|
||||
g_mqtt->publish("json", "{\"D-Star\":{\"action\":\"lost\"}}");
|
||||
}
|
||||
|
||||
// Publish idle mode (no traffic)
|
||||
static inline void mqttPublishIdle()
|
||||
{
|
||||
if (g_mqtt == NULL)
|
||||
return;
|
||||
|
||||
g_mqtt->publish("json", "{\"MMDVM\":{\"mode\":\"idle\"}}");
|
||||
}
|
||||
|
||||
// Publish D-Star BER value
|
||||
static inline void mqttPublishBER(float ber)
|
||||
{
|
||||
if (g_mqtt == NULL)
|
||||
return;
|
||||
|
||||
char buf[128];
|
||||
::snprintf(buf, sizeof(buf),
|
||||
"{\"BER\":{\"mode\":\"D-Star\",\"value\":%.1f}}",
|
||||
ber);
|
||||
|
||||
g_mqtt->publish("json", buf);
|
||||
}
|
||||
|
||||
// Publish D-Star slow data text
|
||||
static inline void mqttPublishText(const wxString& text)
|
||||
{
|
||||
if (g_mqtt == NULL)
|
||||
return;
|
||||
|
||||
char buf[256];
|
||||
::snprintf(buf, sizeof(buf),
|
||||
"{\"Text\":{\"mode\":\"D-Star\",\"value\":\"%s\"}}",
|
||||
(const char*)text.mb_str());
|
||||
|
||||
g_mqtt->publish("json", buf);
|
||||
}
|
||||
|
||||
#endif
|
||||
|
||||
#endif
|
||||
@ -1,14 +1,40 @@
|
||||
This is the D-Star Repeater which controls homebrew repeater hardware and links into the ircDDB Gateway to allow for access to extra non-RF facilities,
|
||||
|
||||
The hardware supported include:
|
||||
* DVAP
|
||||
* DV-Mega (D-Star only)
|
||||
* GMSK Modems
|
||||
* Soundcard repeaters (including UDRC)
|
||||
* MMDVM (D-Star only)
|
||||
* DV-RPTR V1, V2, and V3
|
||||
* Icom Terminal and Access Point modes
|
||||
|
||||
They all build on 32-bit and 64-bit Linux as well as on Windows using Visual Studio 2017 on x86 and x64.
|
||||
# DStarRepeater
|
||||
|
||||
DStarRepeater is the D-Star repeater controller for homebrew repeater hardware. It links into [ircDDBGateway](https://github.com/g4klx/ircDDBGateway) or the newer [DStarGateway](https://github.com/g4klx/DStarGateway) to provide access to D-Star networking facilities such as callsign routing, reflector linking, and ircDDB.
|
||||
|
||||
## Supported Hardware
|
||||
|
||||
- DVAP
|
||||
- DV-Mega (D-Star only)
|
||||
- GMSK Modems
|
||||
- Soundcard repeaters (including UDRC)
|
||||
- MMDVM (D-Star only)
|
||||
- DV-RPTR V1, V2, and V3
|
||||
- Icom Terminal and Access Point modes
|
||||
|
||||
## Building
|
||||
|
||||
Builds on 32-bit and 64-bit Linux as well as on Windows using Visual Studio 2017 (x86 and x64).
|
||||
|
||||
### Standard Build
|
||||
|
||||
```
|
||||
make
|
||||
```
|
||||
|
||||
### With MQTT Support
|
||||
|
||||
DStarRepeater can optionally publish live telemetry to an MQTT broker, including log messages, repeater status, and Display-Driver-compatible events for driving OLED/TFT displays.
|
||||
|
||||
```
|
||||
sudo apt-get install libmosquitto-dev
|
||||
make MQTT=1
|
||||
```
|
||||
|
||||
MQTT support is entirely opt-in. When built without the `MQTT=1` flag, the binary is identical to the tree before MQTT was added — no MQTT code is compiled in and no additional libraries are required.
|
||||
|
||||
See [MQTT.md](MQTT.md) for full configuration details, topic structure, JSON examples, and Display-Driver compatibility information.
|
||||
|
||||
## Licence
|
||||
|
||||
This software is licenced under the GPL v2.
|
||||
|
||||
Loading…
Reference in new issue