nils/ebusd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

send version, running, and current signal after reconnect, re-set message callback on reconnect (fixes #222)

Browse Source
This commit is contained in:
john30
2018-12-02 11:59:42 +01:00
parent b608d80916
commit c9eae87bd9
1 changed files with 28 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ebusd/mqtthandler.cpp
+28 -23
View File
@@ -430,12 +430,6 @@ void MqttHandler::start() {
} }
} }
void MqttHandler::notifyConnected() {
if (m_mosquitto && isRunning()) {
mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0);
}
}
void on_message( void on_message(
#if (LIBMOSQUITTO_MAJOR >= 1) #if (LIBMOSQUITTO_MAJOR >= 1)
struct mosquitto *mosq, struct mosquitto *mosq,
@@ -450,6 +444,16 @@ void on_message(
handler->notifyTopic(topic, data); handler->notifyTopic(topic, data);
} }
void MqttHandler::notifyConnected() {
if (m_mosquitto && isRunning()) {
const string sep = (g_publishFormat & OF_JSON) ? "\"" : "";
publishTopic(m_globalTopic+"version", sep + (PACKAGE_STRING "." REVISION) + sep, true);
publishTopic(m_globalTopic+"running", "true", true);
mosquitto_message_callback_set(m_mosquitto, on_message);
mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0);
}
}
void MqttHandler::notifyTopic(const string& topic, const string& data) { void MqttHandler::notifyTopic(const string& topic, const string& data) {
size_t pos = topic.rfind('/'); size_t pos = topic.rfind('/');
if (pos == string::npos) { if (pos == string::npos) {
@@ -550,16 +554,14 @@ void MqttHandler::run() {
time(&now); time(&now);
start = lastTaskRun = now; start = lastTaskRun = now;
const string sep = (g_publishFormat & OF_JSON) ? "\"" : "";
publishTopic(m_globalTopic+"version", sep + (PACKAGE_STRING "." REVISION) + sep, true);
publishTopic(m_globalTopic+"running", "true", true);
publishTopic(signalTopic, "false");
mosquitto_message_callback_set(m_mosquitto, on_message);
bool allowReconnect = false; bool allowReconnect = false;
while (isRunning()) { while (isRunning()) {
bool wasConnected = m_connected;
handleTraffic(allowReconnect); handleTraffic(allowReconnect);
bool reconnected = !wasConnected && m_connected;
allowReconnect = false; allowReconnect = false;
time(&now); time(&now);
bool sendSignal = reconnected;
if (now < start) { if (now < start) {
// clock skew // clock skew
if (now < lastSignal) { if (now < lastSignal) {
@@ -568,18 +570,7 @@ void MqttHandler::run() {
lastTaskRun = now; lastTaskRun = now;
} else if (now > lastTaskRun+15) { } else if (now > lastTaskRun+15) {
allowReconnect = true; allowReconnect = true;
if (m_busHandler->hasSignal()) { sendSignal = true;
lastSignal = now;
if (!signal) {
signal = true;
publishTopic(signalTopic, "true");
}
} else {
if (signal) {
signal = false;
publishTopic(signalTopic, "false");
}
}
time_t uptime = now-start; time_t uptime = now-start;
updates.str(""); updates.str("");
updates.clear(); updates.clear();
@@ -587,6 +578,20 @@ void MqttHandler::run() {
publishTopic(uptimeTopic, updates.str()); publishTopic(uptimeTopic, updates.str());
time(&lastTaskRun); time(&lastTaskRun);
} }
if (sendSignal) {
if (m_busHandler->hasSignal()) {
lastSignal = now;
if (!signal || reconnected) {
signal = true;
publishTopic(signalTopic, "true");
}
} else {
if (signal || reconnected) {
signal = false;
publishTopic(signalTopic, "false");
}
}
}
if (!m_updatedMessages.empty()) { if (!m_updatedMessages.empty()) {
if (m_connected) { if (m_connected) {
m_messages->lock(); m_messages->lock();