Merge remote-tracking branch 'origin/master' into enhanced_device
This commit is contained in:
@@ -934,7 +934,7 @@ result_t BusHandler::handleSymbol() {
|
||||
return setState(bs_skip, RESULT_ERR_INVALID_ARG);
|
||||
}
|
||||
istringstream input; // TODO create input from database of internal variables
|
||||
if (message == m_messages->getScanMessage()) {
|
||||
if (message == m_messages->getScanMessage() || message == m_messages->getScanMessage(m_ownSlaveAddress)) {
|
||||
input.str(SCAN_ANSWER);
|
||||
}
|
||||
// build response and store in m_response for sending back to requesting master
|
||||
|
||||
Executable → Regular
+11
-3
@@ -298,6 +298,10 @@ void on_connect(
|
||||
void *obj, int rc) {
|
||||
if (rc == 0) {
|
||||
logOtherNotice("mqtt", "connection established");
|
||||
MqttHandler* handler = reinterpret_cast<MqttHandler*>(obj);
|
||||
if (handler) {
|
||||
handler->notifyConnected();
|
||||
}
|
||||
} else {
|
||||
if (rc >= 1 && rc <= 3) {
|
||||
logOtherError("mqtt", "connection refused: %s",
|
||||
@@ -335,6 +339,7 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap*
|
||||
}
|
||||
}
|
||||
m_globalTopic = getTopic(nullptr, "global/");
|
||||
m_subscribeTopic = getTopic(nullptr, "#");
|
||||
m_mosquitto = nullptr;
|
||||
if (mosquitto_lib_init() != MOSQ_ERR_SUCCESS) {
|
||||
logOtherError("mqtt", "unable to initialize");
|
||||
@@ -382,7 +387,6 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap*
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
mosquitto_connect_callback_set(m_mosquitto, on_connect);
|
||||
int ret;
|
||||
#if (LIBMOSQUITTO_MAJOR >= 1)
|
||||
@@ -426,6 +430,12 @@ 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,
|
||||
@@ -545,8 +555,6 @@ void MqttHandler::run() {
|
||||
publishTopic(m_globalTopic+"running", "true", true);
|
||||
publishTopic(signalTopic, "false");
|
||||
mosquitto_message_callback_set(m_mosquitto, on_message);
|
||||
string subTopic = getTopic(nullptr, "#");
|
||||
mosquitto_subscribe(m_mosquitto, nullptr, subTopic.c_str(), 0);
|
||||
bool allowReconnect = false;
|
||||
while (isRunning()) {
|
||||
handleTraffic(allowReconnect);
|
||||
|
||||
Executable → Regular
+8
@@ -76,6 +76,11 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread {
|
||||
// @copydoc
|
||||
void start() override;
|
||||
|
||||
/**
|
||||
* Notify the handler of a (re-)established connection to the broker.
|
||||
*/
|
||||
void notifyConnected();
|
||||
|
||||
/**
|
||||
* Notify the handler of a received MQTT message.
|
||||
* @param topic the topic string.
|
||||
@@ -128,6 +133,9 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread {
|
||||
/** the global topic prefix. */
|
||||
string m_globalTopic;
|
||||
|
||||
/** the topic to subscribe to. */
|
||||
string m_subscribeTopic;
|
||||
|
||||
/** whether to publish a separate topic for each message field. */
|
||||
bool m_publishByField;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user