MySmartHome/MySensors
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix mqtt gateway doesn't distribute node ids

Browse Source
This commit is contained in:
Marcelo Aquino
2015-02-26 21:19:47 -03:00
parent 7c82728633
commit 8d3f24424d
3 changed files with 129 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libraries/MySensors/MyMQTT.cpp
+90 -62
View File
@@ -89,7 +89,6 @@ PROGMEM const char *vType[] = {
V_61, V_62, V_63 V_61, V_62, V_63
}; };
char broker[] PROGMEM = MQTT_BROKER_PREFIX; char broker[] PROGMEM = MQTT_BROKER_PREFIX;
#define S_FIRSTCUSTOM 60 #define S_FIRSTCUSTOM 60
@@ -132,7 +131,7 @@ void MyMQTT::begin(rf24_pa_dbm_e paLevel, uint8_t channel, rf24_datarate_e dataR
Serial.begin(BAUD_RATE); Serial.begin(BAUD_RATE);
repeaterMode = true; repeaterMode = true;
isGateway = true; isGateway = true;
MQTTClientConnected = false; MQTTClients = 0;
setupRepeaterMode(); setupRepeaterMode();
dataCallback = inDataCallback; dataCallback = inDataCallback;
@@ -163,56 +162,94 @@ void MyMQTT::processRadioMessage() {
if (process()) { if (process()) {
// A new message was received from one of the sensors // A new message was received from one of the sensors
MyMessage message = getLastMessage(); MyMessage message = getLastMessage();
// Pass along the message from sensors to serial line
rxBlink(1); rxBlink(1);
SendMQTT(message);
if (msg.isAck()) {
Serial.println("msg is ack!");
if (msg.sender == 255 && mGetCommand(msg) == C_INTERNAL && msg.type == I_ID_REQUEST) {
// TODO: sending ACK request on id_response fucks node up. doesn't work.
// The idea was to confirm id and save to EEPROM_LATEST_NODE_ADDRESS.
}
} else {
// we have to check every message if its a newly assigned id or not.
// Ack on I_ID_RESPONSE does not work, and checking on C_PRESENTATION isn't reliable.
uint8_t newNodeID = loadState(EEPROM_LATEST_NODE_ADDRESS)+1;
if (newNodeID <= MQTT_FIRST_SENSORID) {
newNodeID = MQTT_FIRST_SENSORID;
}
if (msg.sender == newNodeID) {
saveState(EEPROM_LATEST_NODE_ADDRESS,newNodeID);
} }
if (mGetCommand(msg) == C_INTERNAL) {
if (msg.type == I_CONFIG) {
txBlink(1);
if (!sendRoute(build(msg, GATEWAY_ADDRESS, msg.sender, 255, C_INTERNAL, I_CONFIG, 0).set("M"))) {
errBlink(1);
}
} else if (msg.type == I_ID_REQUEST && msg.sender == 255) {
uint8_t newNodeID = loadState(EEPROM_LATEST_NODE_ADDRESS)+1;
if (newNodeID <= MQTT_FIRST_SENSORID) {
newNodeID = MQTT_FIRST_SENSORID;
}
if (newNodeID >= MQTT_LAST_SENSORID) {
// Sorry no more id's left :(
newNodeID = AUTO;
}
txBlink(1);
if (!sendRoute(build(msg, GATEWAY_ADDRESS, msg.sender, 255, C_INTERNAL, I_ID_RESPONSE, 0).set(newNodeID))) {
errBlink(1);
}
}
} else if (mGetCommand(msg)!= C_PRESENTATION) {
// Pass along the message from sensors to MQTT
SendMQTT(message);
}
}
}
} }
void MyMQTT::processMQTTMessage(char *inputString, uint8_t inputPos) { void MyMQTT::processMQTTMessage(char *inputString, uint8_t inputPos) {
char *str, *p; char *str, *p, *payload=NULL;
uint8_t i = 0; uint8_t i = 0;
buffer[0]= 0; buffer[0]= 0;
buffsize = 0; buffsize = 0;
uint8_t mqttMsgType = (uint8_t)inputString[0] >> 4;
if ((uint8_t)inputString[0] >> 4 == MQTTCONNECT) { if (mqttMsgType == MQTTCONNECT) {
buffer[buffsize++] = MQTTCONNACK << 4; buffer[buffsize++] = MQTTCONNACK << 4;
buffer[buffsize++] = 0x02; // Remaining length buffer[buffsize++] = 0x02; // Remaining length
buffer[buffsize++] = 0x00; // Connection accepted buffer[buffsize++] = 0x00; // Connection accepted
buffer[buffsize++] = 0x00; // Reserved buffer[buffsize++] = 0x00; // Reserved
MQTTClientConnected=true; // We have connection! MQTTClients++; // We have a new client connected!
} } else if (mqttMsgType == MQTTPINGREQ) {
if ((uint8_t)inputString[0] >> 4 == MQTTPINGREQ) {
buffer[buffsize++] = MQTTPINGRESP << 4; buffer[buffsize++] = MQTTPINGRESP << 4;
buffer[buffsize++] = 0x00; buffer[buffsize++] = 0x00;
} } else if (mqttMsgType == MQTTSUBSCRIBE) {
if ((uint8_t)inputString[0] >> 4 == MQTTSUBSCRIBE) {
buffer[buffsize++] = MQTTSUBACK << 4; // Just ack everything, we actually dont really care! buffer[buffsize++] = MQTTSUBACK << 4; // Just ack everything, we actually dont really care!
buffer[buffsize++] = 0x03; // Remaining length buffer[buffsize++] = 0x03; // Remaining length
buffer[buffsize++] = (uint8_t)inputString[2]; // Message ID MSB buffer[buffsize++] = (uint8_t)inputString[2]; // Message ID MSB
buffer[buffsize++] = (uint8_t)inputString[3]; // Message ID LSB buffer[buffsize++] = (uint8_t)inputString[3]; // Message ID LSB
buffer[buffsize++] = MQTTQOS0; // QOS level buffer[buffsize++] = MQTTQOS0; // QOS level
} } else if (mqttMsgType == MQTTUNSUBSCRIBE) {
if ((uint8_t)inputString[0] >> 4 == MQTTUNSUBSCRIBE) {
buffer[buffsize++] = MQTTUNSUBACK << 4; buffer[buffsize++] = MQTTUNSUBACK << 4;
buffer[buffsize++] = 0x02; // Remaining length buffer[buffsize++] = 0x02; // Remaining length
buffer[buffsize++] = (uint8_t)inputString[2]; // Message ID MSB buffer[buffsize++] = (uint8_t)inputString[2]; // Message ID MSB
buffer[buffsize++] = (uint8_t)inputString[3]; // Message ID LSB buffer[buffsize++] = (uint8_t)inputString[3]; // Message ID LSB
} else if (mqttMsgType == MQTTDISCONNECT) {
MQTTClients--; // Client disconnected!
} }
if ((uint8_t)inputString[0] >> 4 == MQTTDISCONNECT) {
MQTTClientConnected=false; // We lost connection!
}
if (buffsize > 0) { if (buffsize > 0) {
dataCallback(buffer, &buffsize); dataCallback(buffer, &buffsize);
} }
// We publish everything we get, we dont care if its subscribed or not! // We publish everything we get, we dont care if its subscribed or not!
if ((uint8_t)inputString[0] >> 4 == MQTTPUBLISH || (MQTT_SEND_SUBSCRIPTION && (uint8_t)inputString[0] >> 4 == MQTTSUBSCRIBE)) { if (mqttMsgType == MQTTPUBLISH || (MQTT_SEND_SUBSCRIPTION && mqttMsgType == MQTTSUBSCRIBE)) {
buffer[0] = 0; buffer[0] = 0;
buffsize = 0; buffsize = 0;
// Cut out address and payload depending on message type. // Cut out address and payload depending on message type.
if ((uint8_t)inputString[0] >> 4 == MQTTSUBSCRIBE) { if (mqttMsgType == MQTTSUBSCRIBE) {
strncat(buffer, inputString+6, inputString[5]); strncat(buffer, inputString+6, inputString[5]);
} else { } else {
strncat(buffer, inputString+4, inputString[3]); strncat(buffer, inputString+4, inputString[3]);
@@ -221,19 +258,24 @@ void MyMQTT::processMQTTMessage(char *inputString, uint8_t inputPos) {
// TODO: Check if we should send ack or not. // TODO: Check if we should send ack or not.
for (str = strtok_r(buffer, "/", &p) ; str && i<4 ; str = strtok_r(NULL, "/", &p)) { for (str = strtok_r(buffer, "/", &p) ; str && i<4 ; str = strtok_r(NULL, "/", &p)) {
if (i == 0) { if (i == 0) {
if (strcmp_P(str,broker)!=0) { //look for MQTT_BROKER_PREFIX //look for MQTT_BROKER_PREFIX
return; //Message not for us or malformatted! if (strcmp_P(str, broker) != 0) {
//Message not for us or malformatted!
return;
} }
} else if (i==1) { } else if (i==1) {
msg.destination = atoi(str); //NodeID //NodeID
msg.destination = atoi(str);
} else if (i==2) { } else if (i==2) {
msg.sensor = atoi(str); //SensorID //SensorID
msg.sensor = atoi(str);
} else if (i==3) { } else if (i==3) {
char match=0; //SensorType //SensorType
//strcpy(str,(char*)&str[2]); //Strip V_ char match=0;
for (uint8_t j=0; strcpy_P(convBuf, (char*)pgm_read_word(&(vType[j]))) ; j++) { for (uint8_t j=0; strcpy_P(convBuf, (char*)pgm_read_word(&(vType[j]))) ; j++) {
if (strcmp((char*)&str[2],convBuf)==0) { //Strip V_ and compare //Strip V_ and compare
if (strcmp((char*)&str[2], convBuf)==0) {
match=j; match=j;
break; break;
} }
@@ -245,55 +287,45 @@ void MyMQTT::processMQTTMessage(char *inputString, uint8_t inputPos) {
msg.type = match; msg.type = match;
} }
i++; i++;
} //Check if packge has payload
if ((uint8_t)inputString[1] > (uint8_t)(inputString[3]+2) && !((uint8_t)inputString[0] >> 4 == MQTTSUBSCRIBE)) {
strcpy(convBuf,inputString+(inputString[3]+4));
msg.set(convBuf); //Payload
} else {
msg.set(""); //No payload
} }
// Check if package has payload
if (mqttMsgType == MQTTPUBLISH) {
uint8_t length = (uint8_t)inputString[1] - (uint8_t)(inputString[3]+2);
if (length && length < MAX_PAYLOAD*2) {
// Payload
memcpy(convBuf, inputString+(inputString[3]+4), length);
convBuf[length] = 0;
payload = convBuf;
}
}
msg.set(payload);
txBlink(1); txBlink(1);
if (!sendRoute(build(msg, GATEWAY_ADDRESS, msg.destination, msg.sensor, C_SET, msg.type, 0))) errBlink(1); if (!sendRoute(build(msg, GATEWAY_ADDRESS, msg.destination, msg.sensor, C_SET, msg.type, 0))) errBlink(1);
} }
} }
void MyMQTT::SendMQTT(MyMessage &msg) { void MyMQTT::SendMQTT(MyMessage &msg) {
buffsize = 0; buffsize = 0;
if (!MQTTClientConnected) return; //We have no connections - return if (!MQTTClients) {
if (msg.isAck()) { //We have no clients connected - return
Serial.println("msg is ack!"); return;
if (msg.sender==255 && mGetCommand(msg)==C_INTERNAL && msg.type==I_ID_REQUEST) {
// TODO: sending ACK request on id_response fucks node up. doesn't work.
// The idea was to confirm id and save to EEPROM_LATEST_NODE_ADDRESS.
}
} else {
// we have to check every message if its a newly assigned id or not.
// Ack on I_ID_RESPONSE does not work, and checking on C_PRESENTATION isn't reliable.
uint8_t newNodeID = loadState(EEPROM_LATEST_NODE_ADDRESS)+1;
if (newNodeID <= MQTT_FIRST_SENSORID) newNodeID = MQTT_FIRST_SENSORID;
if (msg.sender==newNodeID) {
saveState(EEPROM_LATEST_NODE_ADDRESS,newNodeID);
} }
if (mGetCommand(msg) == C_INTERNAL) { if (mGetCommand(msg) == C_INTERNAL) {
if (msg.type==I_CONFIG) { //Special message
txBlink(1); msg.type = msg.type+(S_FIRSTCUSTOM-10);
if (!sendRoute(build(msg, GATEWAY_ADDRESS, msg.sender, 255, C_INTERNAL, I_CONFIG, 0).set("M"))) errBlink(1);
} else if (msg.type==I_ID_REQUEST && msg.sender==255) {
uint8_t newNodeID = loadState(EEPROM_LATEST_NODE_ADDRESS)+1;
if (newNodeID <= MQTT_FIRST_SENSORID) newNodeID = MQTT_FIRST_SENSORID;
if (newNodeID >= MQTT_LAST_SENSORID) return; // Sorry no more id's left :(
txBlink(1);
if (!sendRoute(build(msg, GATEWAY_ADDRESS, msg.sender, 255, C_INTERNAL, I_ID_RESPONSE, 0).set(newNodeID))) errBlink(1);
} }
} else if (mGetCommand(msg)!=0) {
if (mGetCommand(msg)==3) msg.type=msg.type+(S_FIRSTCUSTOM-10); //Special message
buffer[buffsize++] = MQTTPUBLISH << 4; // 0: buffer[buffsize++] = MQTTPUBLISH << 4; // 0:
buffer[buffsize++] = 0x09; // 1: Remaining length with no payload, we'll set this later to correct value, buffsize -2 buffer[buffsize++] = 0x09; // 1: Remaining length with no payload, we'll set this later to correct value, buffsize -2
buffer[buffsize++] = 0x00; // 2: Length MSB (Remaing length can never exceed ff,so MSB must be 0!) buffer[buffsize++] = 0x00; // 2: Length MSB (Remaing length can never exceed ff,so MSB must be 0!)
buffer[buffsize++] = 0x08; // 3: Length LSB (ADDR), We'll set this later buffer[buffsize++] = 0x08; // 3: Length LSB (ADDR), We'll set this later
if (msg.type > V_TOTAL) msg.type=V_TOTAL;// If type > defined types set to unknown. if (msg.type > V_TOTAL) {
// If type > defined types set to unknown.
msg.type=V_TOTAL;
}
strcpy_P(buffer+4, broker); strcpy_P(buffer+4, broker);
buffsize+=strlen_P(broker); buffsize+=strlen_P(broker);
buffsize+=sprintf(&buffer[buffsize], "/%i/%i/V_%s", msg.sender, msg.sensor, getType(convBuf, &vType[msg.type])); buffsize+=sprintf(&buffer[buffsize], "/%i/%i/V_%s", msg.sender, msg.sensor, getType(convBuf, &vType[msg.type]));
@@ -308,8 +340,6 @@ void MyMQTT::SendMQTT(MyMessage &msg) {
buffer[1]=buffsize-2; // Set correct Remaining length on byte 2. buffer[1]=buffsize-2; // Set correct Remaining length on byte 2.
dataCallback(buffer, &buffsize); dataCallback(buffer, &buffsize);
} }
}
}
void MyMQTT::rxBlink(uint8_t cnt) { void MyMQTT::rxBlink(uint8_t cnt) {
if(countRx == 255) { countRx = cnt; } if(countRx == 255) { countRx = cnt; }
@@ -322,5 +352,3 @@ void MyMQTT::txBlink(uint8_t cnt) {
void MyMQTT::errBlink(uint8_t cnt) { void MyMQTT::errBlink(uint8_t cnt) {
if(countErr == 255) { countErr = cnt; } if(countErr == 255) { countErr = cnt; }
} }
libraries/MySensors/MyMQTT.h
+1 -1
View File
@@ -71,7 +71,7 @@ private:
void txBlink(uint8_t cnt); void txBlink(uint8_t cnt);
void errBlink(uint8_t cnt); void errBlink(uint8_t cnt);
bool MQTTClientConnected; uint8_t MQTTClients;
char buffer[MQTT_MAX_PACKET_SIZE]; char buffer[MQTT_MAX_PACKET_SIZE];
char convBuf[MAX_PAYLOAD*2+1]; char convBuf[MAX_PAYLOAD*2+1];
uint8_t buffsize; uint8_t buffsize;
libraries/MySensors/examples/MQTTGateway/MQTTGateway.ino
+1 -2
View File
@@ -99,7 +99,7 @@ void processEthernetMessages() {
uint8_t inChar = client.read(); uint8_t inChar = client.read();
readCnt++; readCnt++;
if (inputSize < MQTT_MAX_PACKET_SIZE-1) { if (inputSize < MQTT_MAX_PACKET_SIZE) {
inputString[inputSize] = (char)inChar; inputString[inputSize] = (char)inChar;
inputSize++; inputSize++;
} }
@@ -111,7 +111,6 @@ void processEthernetMessages() {
break; break;
} }
} }
inputString[inputSize] = 0;
#ifdef TCPDUMP #ifdef TCPDUMP
Serial.print("<<"); Serial.print("<<");
char buf[4]; char buf[4];