Added message type to broker

code/espurna/alexa.ino

@@ -46,8 +46,11 @@ bool _alexaRequestCallback(AsyncWebServerRequest *request) {
 }
 
 #if BROKER_SUPPORT
-void _alexaBrokerCallback(const char * topic, unsigned char id, const char * payload) {
+void _alexaBrokerCallback(const unsigned char type, const char * topic, unsigned char id, const char * payload) {
     
+    // Only process status messages
+    if (BROKER_MSG_TYPE_STATUS != type) return;
+
     unsigned char value = atoi(payload);
 
     if (strcmp(MQTT_TOPIC_CHANNEL, topic) == 0) {

code/espurna/broker.ino

@@ -10,23 +10,23 @@ Copyright (C) 2017-2018 by Xose Pérez <xose dot perez at gmail dot com>
 
 #include <vector>
 
-std::vector<void (*)(const char *, unsigned char, const char *)> _broker_callbacks;
+std::vector<void (*)(const unsigned char, const char *, unsigned char, const char *)> _broker_callbacks;
 
 // -----------------------------------------------------------------------------
 
-void brokerRegister(void (*callback)(const char *, unsigned char, const char *)) {
+void brokerRegister(void (*callback)(const unsigned char, const char *, unsigned char, const char *)) {
     _broker_callbacks.push_back(callback);
 }
 
-void brokerPublish(const char * topic, unsigned char id, const char * message) {
+void brokerPublish(const unsigned char type, const char * topic, unsigned char id, const char * message) {
     //DEBUG_MSG_P(PSTR("[BROKER] Message %s[%u] => %s\n"), topic, id, message);
     for (unsigned char i=0; i<_broker_callbacks.size(); i++) {
-        (_broker_callbacks[i])(topic, id, message);
+        (_broker_callbacks[i])(type, topic, id, message);
     }
 }
 
-void brokerPublish(const char * topic, const char * message) {
-    brokerPublish(topic, 0, message);
+void brokerPublish(const unsigned char type, const char * topic, const char * message) {
+    brokerPublish(type, topic, 0, message);
 }
 
 #endif // BROKER_SUPPORT

code/espurna/config/prototypes.h

@@ -24,7 +24,7 @@ extern "C" {
 // Broker
 // -----------------------------------------------------------------------------
 #if BROKER_SUPPORT
-    void brokerRegister(void (*)(const char *, unsigned char, const char *));
+    void brokerRegister(void (*)(const unsigned char, const char *, unsigned char, const char *));
 #endif
 
 // -----------------------------------------------------------------------------

code/espurna/config/types.h

@@ -3,6 +3,15 @@
 // Do not touch this definitions
 //------------------------------------------------------------------------------
 
+// -----------------------------------------------------------------------------
+// BROKER
+// -----------------------------------------------------------------------------
+
+#define BROKER_MSG_TYPE_SYSTEM      0
+#define BROKER_MSG_TYPE_DATETIME    1
+#define BROKER_MSG_TYPE_STATUS      2
+#define BROKER_MSG_TYPE_SENSOR      3
+
 // -----------------------------------------------------------------------------
 // WIFI
 // -----------------------------------------------------------------------------

code/espurna/domoticz.ino

@@ -149,11 +149,16 @@ void _domoticzMqtt(unsigned int type, const char * topic, const char * payload)
 };
 
 #if BROKER_SUPPORT
-void _domoticzBrokerCallback(const char * topic, unsigned char id, const char * payload) {
+void _domoticzBrokerCallback(const unsigned char type, const char * topic, unsigned char id, const char * payload) {
+
+    // Only process status messages
+    if (BROKER_MSG_TYPE_STATUS != type) return;
+
     if (strcmp(MQTT_TOPIC_RELAY, topic) == 0) {
         unsigned char value = atoi(payload);
         domoticzSendRelay(id, value == 1);
     }
+    
 }
 #endif // BROKER_SUPPORT
 

code/espurna/influxdb.ino

@@ -39,10 +39,13 @@ void _idbConfigure() {
 }
 
 #if BROKER_SUPPORT
-void _idbBrokerCallback(const char * topic, unsigned char id, const char * payload) {
-    if (strcmp(MQTT_TOPIC_RELAY, topic) == 0) {
+void _idbBrokerCallback(const unsigned char type, const char * topic, unsigned char id, const char * payload) {
+    
+    // Only process status & senssor messages
+    if ((BROKER_MSG_TYPE_STATUS == type) || (BROKER_MSG_TYPE_SENSOR == type)) {
         idbSend(topic, id, (char *) payload);
     }
+
 }
 #endif // BROKER_SUPPORT
 

code/espurna/led.ino

@@ -74,10 +74,15 @@ void _ledWebSocketOnSend(JsonObject& root) {
 #endif
 
 #if BROKER_SUPPORT
-void _ledBrokerCallback(const char * topic, unsigned char id, const char * payload) {
+void _ledBrokerCallback(const unsigned char type, const char * topic, unsigned char id, const char * payload) {
+
+    // Only process status messages
+    if (BROKER_MSG_TYPE_STATUS != type) return;
+    
     if (strcmp(MQTT_TOPIC_RELAY, topic) == 0) {
         ledUpdate(true);
     }
+
 }
 #endif // BROKER_SUPPORT
 

code/espurna/light.ino

@@ -640,7 +640,7 @@ void lightBroker() {
     char buffer[10];
     for (unsigned int i=0; i < _light_channel.size(); i++) {
         itoa(_light_channel[i].inputValue, buffer, 10);
-        brokerPublish(MQTT_TOPIC_CHANNEL, i, buffer);
+        brokerPublish(BROKER_MSG_TYPE_STATUS, MQTT_TOPIC_CHANNEL, i, buffer);
     }
 }
 

code/espurna/ntp.ino

@@ -108,7 +108,7 @@ void _ntpLoop() {
         static unsigned char last_minute = 60;
         if (ntpSynced() && (minute() != last_minute)) {
             last_minute = minute();
-            brokerPublish(MQTT_TOPIC_DATETIME, ntpDateTime().c_str());
+            brokerPublish(BROKER_MSG_TYPE_DATETIME, MQTT_TOPIC_DATETIME, ntpDateTime().c_str());
         }
     #endif
 

code/espurna/relay.ino

@@ -177,7 +177,7 @@ void _relayProcess(bool mode) {
 
         // Send to Broker
         #if BROKER_SUPPORT
-            brokerPublish(MQTT_TOPIC_RELAY, id, target ? "1" : "0");
+            brokerPublish(BROKER_MSG_TYPE_STATUS, MQTT_TOPIC_RELAY, id, target ? "1" : "0");
         #endif
 
         // Send MQTT

code/espurna/sensor.ino

@@ -1172,7 +1172,7 @@ void _sensorReport(unsigned char index, double value) {
     dtostrf(value, 1-sizeof(buffer), decimals, buffer);
 
     #if BROKER_SUPPORT
-        brokerPublish(magnitudeTopic(magnitude.type).c_str(), magnitude.local, buffer);
+        brokerPublish(BROKER_MSG_TYPE_SENSOR ,magnitudeTopic(magnitude.type).c_str(), magnitude.local, buffer);
     #endif
 
     #if MQTT_SUPPORT
@@ -1191,14 +1191,6 @@ void _sensorReport(unsigned char index, double value) {
 
     #endif // MQTT_SUPPORT
 
-    #if INFLUXDB_SUPPORT
-        if (SENSOR_USE_INDEX || (_counts[magnitude.type] > 1)) {
-            idbSend(magnitudeTopic(magnitude.type).c_str(), magnitude.global, buffer);
-        } else {
-            idbSend(magnitudeTopic(magnitude.type).c_str(), buffer);
-        }
-    #endif // INFLUXDB_SUPPORT
-
     #if THINGSPEAK_SUPPORT
         tspkEnqueueMeasurement(index, buffer);
     #endif

code/espurna/thinkspeak.ino

@@ -36,11 +36,20 @@ unsigned char _tspk_tries = 0;
 // -----------------------------------------------------------------------------
 
 #if BROKER_SUPPORT
-void _tspkBrokerCallback(const char * topic, unsigned char id, const char * payload) {
-    if (strcmp(MQTT_TOPIC_RELAY, topic) == 0) {
+void _tspkBrokerCallback(const unsigned char type, const char * topic, unsigned char id, const char * payload) {
+
+    // Process status messages
+    if (BROKER_MSG_TYPE_STATUS == type) {
         tspkEnqueueRelay(id, (char *) payload);
         tspkFlush();
     }
+
+    // Porcess sensor messages
+    if (BROKER_MSG_TYPE_SENSOR == type) {
+        //tspkEnqueueMeasurement(id, (char *) payload);
+        //tspkFlush();
+    }
+
 }
 #endif // BROKER_SUPPORT