Bemfa_temp.zip

/* PubSubClient.cpp - A simple client for MQTT. Nick O'Leary http://knolleary.net */ #include "PubSubClient.h" #include "Arduino.h" PubSubClient::PubSubClient() { this->_state = MQTT_DISCONNECTED; this->_client = NULL; this->stream = NULL; setCallback(NULL); } PubSubClient::PubSubClient(Client& client) { this->_state = MQTT_DISCONNECTED; setClient(client); this->stream = NULL; } PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(addr, port); setClient(client); this->stream = NULL; } PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(addr,port); setClient(client); setStream(stream); } PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(addr, port); setCallback(callback); setClient(client); this->stream = NULL; } PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(addr,port); setCallback(callback); setClient(client); setStream(stream); } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(ip, port); setClient(client); this->stream = NULL; } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(ip,port); setClient(client); setStream(stream); } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(ip, port); setCallback(callback); setClient(client); this->stream = NULL; } PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(ip,port); setCallback(callback); setClient(client); setStream(stream); } PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(domain,port); setClient(client); this->stream = NULL; } PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(domain,port); setClient(client); setStream(stream); } PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) { this->_state = MQTT_DISCONNECTED; setServer(domain,port); setCallback(callback); setClient(client); this->stream = NULL; } PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) { this->_state = MQTT_DISCONNECTED; setServer(domain,port); setCallback(callback); setClient(client); setStream(stream); } boolean PubSubClient::connect(const char *id) { return connect(id,NULL,NULL,0,0,0,0); } boolean PubSubClient::connect(const char *id, const char *user, const char *pass) { return connect(id,user,pass,0,0,0,0); } boolean PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage); } boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) { if (!connected()) { int result = 0; if (domain != NULL) { result = _client->connect(this->domain, this->port); } else { result = _client->connect(this->ip, this->port); } if (result == 1) { nextMsgId = 1; // Leave room in the buffer for header and variable length field uint16_t length = 5; unsigned int j; #if MQTT_VERSION == MQTT_VERSION_3_1 uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p', MQTT_VERSION}; #define MQTT_HEADER_VERSION_LENGTH 9 #elif MQTT_VERSION == MQTT_VERSION_3_1_1 uint8_t d[7] = {0x00,0x04,'M','Q','T','T',MQTT_VERSION}; #define MQTT_HEADER_VERSION_LENGTH 7 #endif for (j = 0;j<MQTT_HEADER_VERSION_LENGTH;j++) { buffer[length++] = d[j]; } uint8_t v; if (willTopic) { v = 0x06|(willQos<<3)|(willRetain<<5); } else { v = 0x02; } if(user != NULL) { v = v|0x80; if(pass != NULL) { v = v|(0x80>>1); } } buffer[length++] = v; buffer[length++] = ((MQTT_KEEPALIVE) >> 8); buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF); length = writeString(id,buffer,length); if (willTopic) { length = writeString(willTopic,buffer,length); length = writeString(willMessage,buffer,length); } if(user != NULL) { length = writeString(user,buffer,length); if(pass != NULL) { length = writeString(pass,buffer,length); } } write(MQTTCONNECT,buffer,length-5); lastInActivity = lastOutActivity = millis(); while (!_client->available()) { unsigned long t = millis(); if (t-lastInActivity >= ((int32_t) MQTT_SOCKET_TIMEOUT*1000UL)) { _state = MQTT_CONNECTION_TIMEOUT; _client->stop(); return false; } } uint8_t llen; uint16_t len = readPacket(&llen); if (len == 4) { if (buffer[3] == 0) { lastInActivity = millis(); pingOutstanding = false; _state = MQTT_CONNECTED; return true; } else { _state = buffer[3]; } } _client->stop(); } else { _state = MQTT_CONNECT_FAILED; } return false; } return true; } // reads a byte into result boolean PubSubClient::readByte(uint8_t * result) { uint32_t previousMillis = millis(); while(!_client->available()) { uint32_t currentMillis = millis(); if(currentMillis - previousMillis >= ((int32_t) MQTT_SOCKET_TIMEOUT * 1000)){ return false; } } *result = _client->read(); return true; } // reads a byte into result[*index] and increments index boolean PubSubClient::readByte(uint8_t * result, uint16_t * index){ uint16_t current_index = *index; uint8_t * write_address = &(result[current_index]); if(readByte(write_address)){ *index = current_index + 1; return true; } return false; } uint16_t PubSubClient::readPacket(uint8_t* lengthLength) { uint16_t len = 0; if(!readByte(buffer, &len)) return 0; bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH; uint32_t multiplier = 1; uint16_t length = 0; uint8_t digit = 0; uint16_t skip = 0; uint8_t start = 0; do { if(!readByte(&digit)) return 0; buffer[len++] = digit; length += (digit & 127) * multiplier; multiplier *= 128; } while ((digit & 128) != 0); *lengthLength = len-1; if (isPublish) { // Read in topic length to calculate bytes to skip over for Stream writing if(!readByte(buffer, &len)) return 0; if(!readByte(buffer, &len)) return 0; skip = (buffer[*lengthLength+1]<<8)+buffer[*lengthLength+2]; start = 2; if (buffer[0]&MQTTQOS1) { // skip messag