log loop communication error at most every 10 seconds and wait at least 1 second before doing the loop again on error (#262)

This commit is contained in:
john30
2019-02-16 17:38:36 +01:00
parent 5af41dfab0
commit d366bbb6b3
2 changed files with 18 additions and 8 deletions
+13 -7
View File
@@ -425,7 +425,7 @@ void on_message(
MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages) MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages)
: DataSink(userInfo, "mqtt"), DataSource(busHandler), WaitThread(), m_messages(messages), m_connected(false), : DataSink(userInfo, "mqtt"), DataSource(busHandler), WaitThread(), m_messages(messages), m_connected(false),
m_initialConnectFailed(false), m_lastUpdateCheckResult(".") { m_initialConnectFailed(false), m_lastUpdateCheckResult("."), m_lastErrorLogTime(0) {
m_publishByField = false; m_publishByField = false;
m_mosquitto = nullptr; m_mosquitto = nullptr;
if (g_topicFields.empty()) { if (g_topicFields.empty()) {
@@ -673,7 +673,7 @@ void MqttHandler::run() {
bool allowReconnect = false; bool allowReconnect = false;
while (isRunning()) { while (isRunning()) {
bool wasConnected = m_connected; bool wasConnected = m_connected;
handleTraffic(allowReconnect); bool needsWait = handleTraffic(allowReconnect);
bool reconnected = !wasConnected && m_connected; bool reconnected = !wasConnected && m_connected;
allowReconnect = false; allowReconnect = false;
time(&now); time(&now);
@@ -732,16 +732,16 @@ void MqttHandler::run() {
m_updatedMessages.clear(); m_updatedMessages.clear();
} }
} }
if (!m_connected && !Wait(5)) { if ((!m_connected && !Wait(5)) || (needsWait && !Wait(1))) {
break; break;
} }
} }
publishTopic(signalTopic, "false", true); publishTopic(signalTopic, "false", true);
} }
void MqttHandler::handleTraffic(bool allowReconnect) { bool MqttHandler::handleTraffic(bool allowReconnect) {
if (!m_mosquitto) { if (!m_mosquitto) {
return; return false;
} }
int ret; int ret;
#if (LIBMOSQUITTO_MAJOR >= 1) #if (LIBMOSQUITTO_MAJOR >= 1)
@@ -771,15 +771,21 @@ void MqttHandler::handleTraffic(bool allowReconnect) {
logOtherNotice("mqtt", "connection re-established"); logOtherNotice("mqtt", "connection re-established");
} }
if (!m_connected || ret == MOSQ_ERR_SUCCESS) { if (!m_connected || ret == MOSQ_ERR_SUCCESS) {
return; return false;
} }
if (ret == MOSQ_ERR_NO_CONN || ret == MOSQ_ERR_CONN_LOST || ret == MOSQ_ERR_CONN_REFUSED) { if (ret == MOSQ_ERR_NO_CONN || ret == MOSQ_ERR_CONN_LOST || ret == MOSQ_ERR_CONN_REFUSED) {
logOtherError("mqtt", "communication error: %s", ret == MOSQ_ERR_NO_CONN ? "not connected" logOtherError("mqtt", "communication error: %s", ret == MOSQ_ERR_NO_CONN ? "not connected"
: (ret == MOSQ_ERR_CONN_LOST ? "connection lost" : "connection refused")); : (ret == MOSQ_ERR_CONN_LOST ? "connection lost" : "connection refused"));
m_connected = false; m_connected = false;
} else { } else {
check(ret, "communication error"); time_t now;
time(&now);
if (now > m_lastErrorLogTime + 10) { // log at most every 10 seconds
m_lastErrorLogTime = now;
check(ret, "communication error");
}
} }
return true;
} }
string MqttHandler::getTopic(const Message* message, const string& suffix, const string& fieldName) { string MqttHandler::getTopic(const Message* message, const string& suffix, const string& fieldName) {
+5 -1
View File
@@ -100,8 +100,9 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread {
/** /**
* Called regularly to handle MQTT traffic. * Called regularly to handle MQTT traffic.
* @param allowReconnect true when reconnecting to the broker is allowed. * @param allowReconnect true when reconnecting to the broker is allowed.
* @return true on error for waiting a bit until next call, or false otherwise.
*/ */
void handleTraffic(bool allowReconnect); bool handleTraffic(bool allowReconnect);
/** /**
* Build the MQTT topic string for the @a Message. * Build the MQTT topic string for the @a Message.
@@ -150,6 +151,9 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread {
/** the last update check result. */ /** the last update check result. */
string m_lastUpdateCheckResult; string m_lastUpdateCheckResult;
/** the last system time when a communication error was logged. */
time_t m_lastErrorLogTime;
}; };
} // namespace ebusd } // namespace ebusd