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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
john
2018-12-02 12:00:29 +01:00
parent aec6dfccbe c9eae87bd9
commit da56691e41
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(
#if (LIBMOSQUITTO_MAJOR >= 1)
struct mosquitto *mosq,
@@ -450,6 +444,16 @@ void on_message(
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) {
size_t pos = topic.rfind('/');
if (pos == string::npos) {
@@ -550,16 +554,14 @@ void MqttHandler::run() {
time(&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;
while (isRunning()) {
bool wasConnected = m_connected;
handleTraffic(allowReconnect);
bool reconnected = !wasConnected && m_connected;
allowReconnect = false;
time(&now);
bool sendSignal = reconnected;
if (now < start) {
// clock skew
if (now < lastSignal) {
@@ -568,18 +570,7 @@ void MqttHandler::run() {
lastTaskRun = now;
} else if (now > lastTaskRun+15) {
allowReconnect = true;
if (m_busHandler->hasSignal()) {
lastSignal = now;
if (!signal) {
signal = true;
publishTopic(signalTopic, "true");
}
} else {
if (signal) {
signal = false;
publishTopic(signalTopic, "false");
}
}
sendSignal = true;
time_t uptime = now-start;
updates.str("");
updates.clear();
@@ -587,6 +578,20 @@ void MqttHandler::run() {
publishTopic(uptimeTopic, updates.str());
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_connected) {
m_messages->lock();