mqtt specific components for the impact mbed endpoint library
Dependents: mbed_mqtt_endpoint_ublox_ethernet mbed_mqtt_endpoint_ublox_cellular mbed_mqtt_endpoint_nxp
MQTTTransport.cpp@49:965a65122c0f, 2014-07-08 (annotated)
- Committer:
- ansond
- Date:
- Tue Jul 08 14:59:39 2014 +0000
- Revision:
- 49:965a65122c0f
- Parent:
- 43:14ccc830d6a6
ported to new MQTT library
Who changed what in which revision?
User | Revision | Line number | New contents of line |
---|---|---|---|
ansond | 0:a3fc1c6ef150 | 1 | /* Copyright C2013 Doug Anson, MIT License |
ansond | 0:a3fc1c6ef150 | 2 | * |
ansond | 0:a3fc1c6ef150 | 3 | * Permission is hereby granted, free of charge, to any person obtaining a copy of this software |
ansond | 0:a3fc1c6ef150 | 4 | * and associated documentation files the "Software", to deal in the Software without restriction, |
ansond | 0:a3fc1c6ef150 | 5 | * including without limitation the rights to use, copy, modify, merge, publish, distribute, |
ansond | 0:a3fc1c6ef150 | 6 | * sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is |
ansond | 0:a3fc1c6ef150 | 7 | * furnished to do so, subject to the following conditions: |
ansond | 0:a3fc1c6ef150 | 8 | * |
ansond | 0:a3fc1c6ef150 | 9 | * The above copyright notice and this permission notice shall be included in all copies or |
ansond | 0:a3fc1c6ef150 | 10 | * substantial portions of the Software. |
ansond | 0:a3fc1c6ef150 | 11 | * |
ansond | 0:a3fc1c6ef150 | 12 | * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING |
ansond | 0:a3fc1c6ef150 | 13 | * BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND |
ansond | 0:a3fc1c6ef150 | 14 | * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, |
ansond | 0:a3fc1c6ef150 | 15 | * DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
ansond | 0:a3fc1c6ef150 | 16 | * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
ansond | 0:a3fc1c6ef150 | 17 | */ |
ansond | 0:a3fc1c6ef150 | 18 | |
ansond | 0:a3fc1c6ef150 | 19 | #include "mbed.h" |
ansond | 42:297585f8e7bd | 20 | |
ansond | 0:a3fc1c6ef150 | 21 | #include "MQTTTransport.h" |
ansond | 0:a3fc1c6ef150 | 22 | |
ansond | 0:a3fc1c6ef150 | 23 | // Endpoint Support |
ansond | 0:a3fc1c6ef150 | 24 | #include "MBEDEndpoint.h" |
ansond | 0:a3fc1c6ef150 | 25 | |
ansond | 0:a3fc1c6ef150 | 26 | // EmulatedResourceFactory support |
ansond | 0:a3fc1c6ef150 | 27 | #include "EmulatedResourceFactory.h" |
ansond | 43:14ccc830d6a6 | 28 | |
ansond | 49:965a65122c0f | 29 | // our instance pointer |
ansond | 49:965a65122c0f | 30 | MQTTTransport *_mqtt_instance = NULL; |
ansond | 49:965a65122c0f | 31 | |
ansond | 0:a3fc1c6ef150 | 32 | // MQTT callback to handle received messages |
ansond | 49:965a65122c0f | 33 | void _mqtt_message_handler(MQTT::MessageData& md) { |
ansond | 0:a3fc1c6ef150 | 34 | char buffer[MAX_MQTT_MESSAGE_LENGTH+1]; |
ansond | 0:a3fc1c6ef150 | 35 | char rcv_topic[MQTT_IOC_TOPIC_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 36 | |
ansond | 49:965a65122c0f | 37 | MQTT::Message &message = md.message; |
ansond | 49:965a65122c0f | 38 | |
ansond | 0:a3fc1c6ef150 | 39 | memset(buffer,0,MAX_MQTT_MESSAGE_LENGTH+1); |
ansond | 0:a3fc1c6ef150 | 40 | memset(rcv_topic,0,MQTT_IOC_TOPIC_LEN+1); |
ansond | 49:965a65122c0f | 41 | |
ansond | 49:965a65122c0f | 42 | int length = message.payloadlen; |
ansond | 49:965a65122c0f | 43 | if (length > MAX_MQTT_MESSAGE_LENGTH) length = MAX_MQTT_MESSAGE_LENGTH; |
ansond | 49:965a65122c0f | 44 | memcpy(buffer,message.payload,length); |
ansond | 49:965a65122c0f | 45 | strcpy(rcv_topic,(char *)md.topicName.cstring); |
ansond | 0:a3fc1c6ef150 | 46 | |
ansond | 0:a3fc1c6ef150 | 47 | if (_mqtt_instance != NULL) { |
ansond | 49:965a65122c0f | 48 | if (_mqtt_instance->isPongMessage(rcv_topic,buffer,length)) |
ansond | 0:a3fc1c6ef150 | 49 | _mqtt_instance->processPongMessage(buffer,length); |
ansond | 49:965a65122c0f | 50 | else |
ansond | 0:a3fc1c6ef150 | 51 | _mqtt_instance->processMessage(_mqtt_instance->getEndpointNameFromTopic(rcv_topic),buffer,length); |
ansond | 0:a3fc1c6ef150 | 52 | } |
ansond | 0:a3fc1c6ef150 | 53 | } |
ansond | 49:965a65122c0f | 54 | |
ansond | 0:a3fc1c6ef150 | 55 | // default constructor |
ansond | 0:a3fc1c6ef150 | 56 | MQTTTransport::MQTTTransport(ErrorHandler *error_handler,void *endpoint,MBEDToIOCResourceMap *map) : Transport(error_handler,endpoint) { |
ansond | 49:965a65122c0f | 57 | this->m_ipstack = new MQTTEthernet(); |
ansond | 49:965a65122c0f | 58 | this->m_mqtt = new MQTT::Client<MQTTEthernet, Countdown>(*this->m_ipstack); |
ansond | 0:a3fc1c6ef150 | 59 | _mqtt_instance = this; |
ansond | 0:a3fc1c6ef150 | 60 | this->m_map = map; |
ansond | 0:a3fc1c6ef150 | 61 | this->m_ping_counter = 1; |
ansond | 0:a3fc1c6ef150 | 62 | this->m_ping_countdown = MQTT_PING_COUNTDOWN; |
ansond | 0:a3fc1c6ef150 | 63 | this->initTopic(); |
ansond | 0:a3fc1c6ef150 | 64 | } |
ansond | 0:a3fc1c6ef150 | 65 | |
ansond | 0:a3fc1c6ef150 | 66 | // default destructor |
ansond | 0:a3fc1c6ef150 | 67 | MQTTTransport::~MQTTTransport() { |
ansond | 0:a3fc1c6ef150 | 68 | this->disconnect(); |
ansond | 0:a3fc1c6ef150 | 69 | } |
ansond | 0:a3fc1c6ef150 | 70 | |
ansond | 0:a3fc1c6ef150 | 71 | // init our topic |
ansond | 0:a3fc1c6ef150 | 72 | void MQTTTransport::initTopic() { |
ansond | 0:a3fc1c6ef150 | 73 | MBEDEndpoint *endpoint = (MBEDEndpoint *)this->getEndpoint(); |
ansond | 0:a3fc1c6ef150 | 74 | char *endpoint_name = endpoint->getEndpointName(); |
ansond | 0:a3fc1c6ef150 | 75 | memset(this->m_topic,0,MQTT_IOC_TOPIC_LEN+1); |
ansond | 0:a3fc1c6ef150 | 76 | sprintf(this->m_topic,MQTT_IOC_TOPIC,endpoint_name); |
ansond | 0:a3fc1c6ef150 | 77 | } |
ansond | 0:a3fc1c6ef150 | 78 | |
ansond | 0:a3fc1c6ef150 | 79 | // get our topic |
ansond | 0:a3fc1c6ef150 | 80 | char *MQTTTransport::getTopic() { return this->m_topic; } |
ansond | 0:a3fc1c6ef150 | 81 | |
ansond | 0:a3fc1c6ef150 | 82 | // get the IOC <--> MBED resource map |
ansond | 0:a3fc1c6ef150 | 83 | MBEDToIOCResourceMap *MQTTTransport::getMap() { return this->m_map; } |
ansond | 0:a3fc1c6ef150 | 84 | |
ansond | 0:a3fc1c6ef150 | 85 | // pull the endpoint name from the MQTT topic |
ansond | 0:a3fc1c6ef150 | 86 | char *MQTTTransport::getEndpointNameFromTopic(char *topic) { |
ansond | 0:a3fc1c6ef150 | 87 | if (topic != NULL) { |
ansond | 7:8a4a61202b36 | 88 | memset(this->m_endpoint_name,0,PERSONALITY_NAME_LEN+1); |
ansond | 0:a3fc1c6ef150 | 89 | char trash[MQTT_IOC_TOPIC_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 90 | char ep[MQTT_IOC_TOPIC_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 91 | memset(trash,0,MQTT_IOC_TOPIC_LEN+1); |
ansond | 0:a3fc1c6ef150 | 92 | memset(ep,0,MQTT_IOC_TOPIC_LEN+1); |
ansond | 0:a3fc1c6ef150 | 93 | bool done = false; |
ansond | 0:a3fc1c6ef150 | 94 | int length = 0; if (topic != NULL) length = strlen(topic); |
ansond | 0:a3fc1c6ef150 | 95 | for(int i=length-1;i>=0 && !done;--i) if (topic[i] == '/') { topic[i] = ' ' ; done = true; } |
ansond | 0:a3fc1c6ef150 | 96 | sscanf(topic,"%s%s",trash,ep); |
ansond | 0:a3fc1c6ef150 | 97 | //this->logger()->log("MQTT: Topic:[%s] trash:[%s] ep:[%s]",topic,trash,ep); |
ansond | 0:a3fc1c6ef150 | 98 | if (strlen(ep) > 0) { |
ansond | 0:a3fc1c6ef150 | 99 | if (strcmp(ep,MQTT_IOC_ALL_ENDPOINT) != 0) { |
ansond | 0:a3fc1c6ef150 | 100 | // just insert the name and let the parser determine if its for us or not... |
ansond | 0:a3fc1c6ef150 | 101 | strncpy(this->m_endpoint_name,ep,strlen(ep)); |
ansond | 0:a3fc1c6ef150 | 102 | } |
ansond | 0:a3fc1c6ef150 | 103 | else { |
ansond | 0:a3fc1c6ef150 | 104 | // this is a broadcast message - so we need to process it |
ansond | 0:a3fc1c6ef150 | 105 | MBEDEndpoint *endpoint = (MBEDEndpoint *)this->getEndpoint(); |
ansond | 0:a3fc1c6ef150 | 106 | char *endpoint_name = endpoint->getEndpointName(); |
ansond | 0:a3fc1c6ef150 | 107 | strcpy(this->m_endpoint_name,endpoint_name); |
ansond | 0:a3fc1c6ef150 | 108 | } |
ansond | 0:a3fc1c6ef150 | 109 | } |
ansond | 0:a3fc1c6ef150 | 110 | //this->logger()->log("MQTT Topic (discovered): %s Original: %s",this->m_endpoint_name,topic); |
ansond | 0:a3fc1c6ef150 | 111 | return this->m_endpoint_name; |
ansond | 0:a3fc1c6ef150 | 112 | } |
ansond | 0:a3fc1c6ef150 | 113 | //this->logger()->log("MQTT Topic (discovered): NULL Original: %s",topic); |
ansond | 0:a3fc1c6ef150 | 114 | return NULL; |
ansond | 0:a3fc1c6ef150 | 115 | } |
ansond | 0:a3fc1c6ef150 | 116 | |
ansond | 0:a3fc1c6ef150 | 117 | // process a MQTT Message |
ansond | 0:a3fc1c6ef150 | 118 | void MQTTTransport::processMessage(char *message_name,char *payload, unsigned int payload_length) { |
ansond | 0:a3fc1c6ef150 | 119 | char message_type[MQTT_PAYLOAD_SEGMENT_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 120 | char message_verb[MQTT_PAYLOAD_SEGMENT_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 121 | char message_value[MQTT_PAYLOAD_SEGMENT_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 122 | char message_opt[MQTT_PAYLOAD_SEGMENT_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 123 | |
ansond | 0:a3fc1c6ef150 | 124 | // initialize |
ansond | 0:a3fc1c6ef150 | 125 | memset(message_type,0,MQTT_PAYLOAD_SEGMENT_LEN+1); |
ansond | 0:a3fc1c6ef150 | 126 | memset(message_verb,0,MQTT_PAYLOAD_SEGMENT_LEN+1); |
ansond | 0:a3fc1c6ef150 | 127 | memset(message_value,0,MQTT_PAYLOAD_SEGMENT_LEN+1); |
ansond | 0:a3fc1c6ef150 | 128 | memset(message_opt,0,MQTT_PAYLOAD_SEGMENT_LEN+1); |
ansond | 0:a3fc1c6ef150 | 129 | |
ansond | 0:a3fc1c6ef150 | 130 | // get our endpoint |
ansond | 0:a3fc1c6ef150 | 131 | MBEDEndpoint *endpoint = (MBEDEndpoint *)this->getEndpoint(); |
ansond | 0:a3fc1c6ef150 | 132 | char *endpoint_name = endpoint->getEndpointName(); |
ansond | 0:a3fc1c6ef150 | 133 | |
ansond | 0:a3fc1c6ef150 | 134 | // DEBUG |
ansond | 0:a3fc1c6ef150 | 135 | //this->logger()->log("Endpoint:[%s] Target: [%s]",endpoint_name,message_name); |
ansond | 0:a3fc1c6ef150 | 136 | |
ansond | 0:a3fc1c6ef150 | 137 | // only respond if its for our node |
ansond | 0:a3fc1c6ef150 | 138 | if (strcmp(endpoint_name,message_name) == 0) { |
ansond | 0:a3fc1c6ef150 | 139 | // format of the MQTT message: message_type:verb|Parameter_X:value|keyword:optional_data |
ansond | 0:a3fc1c6ef150 | 140 | char buffer[MAX_MQTT_MESSAGE_LENGTH+1]; |
ansond | 0:a3fc1c6ef150 | 141 | memset(buffer,0,MAX_MQTT_MESSAGE_LENGTH+1); |
ansond | 0:a3fc1c6ef150 | 142 | memcpy(buffer,payload,payload_length); |
ansond | 0:a3fc1c6ef150 | 143 | int count = 0; for(int i=0;i<payload_length;++i) if (payload[i] == ':') ++count; |
ansond | 0:a3fc1c6ef150 | 144 | for(int i=0;i<payload_length;++i) { |
ansond | 0:a3fc1c6ef150 | 145 | if (buffer[i] == ':') { |
ansond | 0:a3fc1c6ef150 | 146 | if (i < (payload_length-1)) buffer[i] = ' '; |
ansond | 0:a3fc1c6ef150 | 147 | else buffer[i] = '\0'; |
ansond | 0:a3fc1c6ef150 | 148 | } |
ansond | 0:a3fc1c6ef150 | 149 | } |
ansond | 0:a3fc1c6ef150 | 150 | if (count == 1) sscanf(buffer,"%s%s",message_type,message_verb); |
ansond | 0:a3fc1c6ef150 | 151 | if (count == 2) sscanf(buffer,"%s%s%s",message_type,message_verb,message_value); |
ansond | 0:a3fc1c6ef150 | 152 | if (count == 3) sscanf(buffer,"%s%s%s%s",message_type,message_verb,message_value,message_opt); |
ansond | 0:a3fc1c6ef150 | 153 | |
ansond | 0:a3fc1c6ef150 | 154 | // DEBUG |
ansond | 0:a3fc1c6ef150 | 155 | //this->logger()->log("Raw Payload: %s, length: %d",payload,payload_length); |
ansond | 0:a3fc1c6ef150 | 156 | //this->logger()->log("Buffer: %s, length: %d",buffer,strlen(buffer)); |
ansond | 0:a3fc1c6ef150 | 157 | //this->logger()->log("Parsed Payload: Type: [%s] Name: [%s] Verb: [%s] Value: [%s]",message_type,message_name,message_verb,message_value); |
ansond | 0:a3fc1c6ef150 | 158 | |
ansond | 0:a3fc1c6ef150 | 159 | // load endpoints |
ansond | 0:a3fc1c6ef150 | 160 | if (message_type != NULL && strcmp(message_type,IOC_ENDPOINT_VERB) == 0) { // Endpoint |
ansond | 0:a3fc1c6ef150 | 161 | if (message_verb != NULL && strcmp(message_verb,IOC_REQUEST_LOAD_ALL_VERB) == 0) { // load |
ansond | 0:a3fc1c6ef150 | 162 | if (message_value != NULL && strcmp(message_value,IOC_ENDPOINT_ALL_VERB) == 0) { // all |
ansond | 0:a3fc1c6ef150 | 163 | // load up our endpoints |
ansond | 7:8a4a61202b36 | 164 | endpoint->loadPersonalities(); |
ansond | 7:8a4a61202b36 | 165 | endpoint->updatePersonalities(); |
ansond | 0:a3fc1c6ef150 | 166 | } |
ansond | 0:a3fc1c6ef150 | 167 | else if (message_value != NULL && strcmp(message_value,this->m_endpoint_name) == 0) { |
ansond | 0:a3fc1c6ef150 | 168 | // load up our endpoints (us only) |
ansond | 7:8a4a61202b36 | 169 | endpoint->loadPersonalities(); |
ansond | 7:8a4a61202b36 | 170 | endpoint->updatePersonalities(); |
ansond | 0:a3fc1c6ef150 | 171 | } |
ansond | 0:a3fc1c6ef150 | 172 | } |
ansond | 0:a3fc1c6ef150 | 173 | |
ansond | 0:a3fc1c6ef150 | 174 | else if (message_verb != NULL && strcmp(message_verb,IOC_REQUEST_UPDATE_ALL_VERB) == 0) { // update |
ansond | 0:a3fc1c6ef150 | 175 | if (message_value != NULL && strcmp(message_value,IOC_ENDPOINT_ALL_VERB) == 0) { // all |
ansond | 0:a3fc1c6ef150 | 176 | // update our endpoints |
ansond | 7:8a4a61202b36 | 177 | endpoint->updatePersonalities(); |
ansond | 0:a3fc1c6ef150 | 178 | } |
ansond | 0:a3fc1c6ef150 | 179 | else { |
ansond | 0:a3fc1c6ef150 | 180 | // update just our endpoint |
ansond | 0:a3fc1c6ef150 | 181 | int index = -1; |
ansond | 0:a3fc1c6ef150 | 182 | if (message_name != NULL) { |
ansond | 7:8a4a61202b36 | 183 | index = endpoint->indexOfPersonality((char *)message_name); |
ansond | 0:a3fc1c6ef150 | 184 | if (index >= 0) { |
ansond | 0:a3fc1c6ef150 | 185 | if (message_verb != NULL && strcmp(message_verb,IOC_REQUEST_UPDATE_ALL_VERB) == 0) { |
ansond | 0:a3fc1c6ef150 | 186 | // update our endpoint |
ansond | 7:8a4a61202b36 | 187 | endpoint->updatePersonality(index); |
ansond | 0:a3fc1c6ef150 | 188 | } |
ansond | 0:a3fc1c6ef150 | 189 | } |
ansond | 0:a3fc1c6ef150 | 190 | } |
ansond | 0:a3fc1c6ef150 | 191 | } |
ansond | 0:a3fc1c6ef150 | 192 | } |
ansond | 0:a3fc1c6ef150 | 193 | } |
ansond | 0:a3fc1c6ef150 | 194 | |
ansond | 0:a3fc1c6ef150 | 195 | // change a resource value |
ansond | 0:a3fc1c6ef150 | 196 | if (message_type != NULL && strcmp(message_type,IOC_CHANGE_VERB) == 0) { |
ansond | 0:a3fc1c6ef150 | 197 | if (message_name != NULL) { |
ansond | 0:a3fc1c6ef150 | 198 | // destined for our lights? |
ansond | 7:8a4a61202b36 | 199 | int index = endpoint->indexOfPersonality((char *)message_name); |
ansond | 0:a3fc1c6ef150 | 200 | if (index >= 0) { |
ansond | 0:a3fc1c6ef150 | 201 | if (message_verb != NULL) { |
ansond | 0:a3fc1c6ef150 | 202 | // map the parameter to one of ours |
ansond | 0:a3fc1c6ef150 | 203 | char *mapped_resource = this->mapIOCResourceToEndpointResource((char *)message_verb); |
ansond | 0:a3fc1c6ef150 | 204 | if (mapped_resource != NULL) { |
ansond | 0:a3fc1c6ef150 | 205 | if (message_value != NULL) { |
ansond | 0:a3fc1c6ef150 | 206 | EmulatedResourceFactory *factory = (EmulatedResourceFactory *)endpoint->getResources(index); |
ansond | 0:a3fc1c6ef150 | 207 | bool success = factory->setResourceValue(mapped_resource,message_value); |
ansond | 0:a3fc1c6ef150 | 208 | |
ansond | 0:a3fc1c6ef150 | 209 | // end the resource value back over MQTT |
ansond | 0:a3fc1c6ef150 | 210 | this->sendResult(message_name,message_verb,message_value,success); |
ansond | 0:a3fc1c6ef150 | 211 | } |
ansond | 0:a3fc1c6ef150 | 212 | } |
ansond | 31:e5950e0677be | 213 | |
ansond | 31:e5950e0677be | 214 | // for Dimming, we also want to refresh our record (in whole) at the IOC |
ansond | 31:e5950e0677be | 215 | if (this->isDimmingResource(message_verb)) { |
ansond | 31:e5950e0677be | 216 | // send a fresh update to the IOC - just us... |
ansond | 31:e5950e0677be | 217 | int index = endpoint->indexOfPersonality((char *)message_name); |
ansond | 31:e5950e0677be | 218 | if (index >= 0) endpoint->updatePersonality(index); |
ansond | 31:e5950e0677be | 219 | } |
ansond | 0:a3fc1c6ef150 | 220 | } |
ansond | 0:a3fc1c6ef150 | 221 | } |
ansond | 0:a3fc1c6ef150 | 222 | } |
ansond | 0:a3fc1c6ef150 | 223 | } |
ansond | 0:a3fc1c6ef150 | 224 | |
ansond | 0:a3fc1c6ef150 | 225 | // get a resource value |
ansond | 0:a3fc1c6ef150 | 226 | if (message_type != NULL && strcmp(message_type,IOC_REQUEST_VALUE_VERB) == 0) { |
ansond | 0:a3fc1c6ef150 | 227 | if (message_name != NULL) { |
ansond | 0:a3fc1c6ef150 | 228 | // destined for our lights? |
ansond | 7:8a4a61202b36 | 229 | int index = endpoint->indexOfPersonality((char *)message_name); |
ansond | 0:a3fc1c6ef150 | 230 | if (index >= 0) { |
ansond | 0:a3fc1c6ef150 | 231 | if (message_verb != NULL) { |
ansond | 0:a3fc1c6ef150 | 232 | // map the parameter to one of ours |
ansond | 0:a3fc1c6ef150 | 233 | char *mapped_resource = this->mapIOCResourceToEndpointResource((char *)message_verb); |
ansond | 0:a3fc1c6ef150 | 234 | if (mapped_resource != NULL) { |
ansond | 0:a3fc1c6ef150 | 235 | EmulatedResourceFactory *factory = (EmulatedResourceFactory *)endpoint->getResources(index); |
ansond | 0:a3fc1c6ef150 | 236 | strcpy(message_value,factory->getResourceValue((char *)mapped_resource)); |
ansond | 0:a3fc1c6ef150 | 237 | bool success = false; if (message_value != NULL) success = true; |
ansond | 0:a3fc1c6ef150 | 238 | |
ansond | 0:a3fc1c6ef150 | 239 | // log resource get |
ansond | 0:a3fc1c6ef150 | 240 | if (success) this->logger()->log("Resource: %s (%s) Value: %s",message_verb,mapped_resource,message_value); |
ansond | 0:a3fc1c6ef150 | 241 | |
ansond | 0:a3fc1c6ef150 | 242 | // end the resource value back over MQTT |
ansond | 0:a3fc1c6ef150 | 243 | this->sendResult(message_name,message_verb,message_value,success); |
ansond | 0:a3fc1c6ef150 | 244 | } |
ansond | 0:a3fc1c6ef150 | 245 | } |
ansond | 0:a3fc1c6ef150 | 246 | } |
ansond | 0:a3fc1c6ef150 | 247 | } |
ansond | 0:a3fc1c6ef150 | 248 | } |
ansond | 0:a3fc1c6ef150 | 249 | } |
ansond | 0:a3fc1c6ef150 | 250 | else { |
ansond | 0:a3fc1c6ef150 | 251 | // message not bound for our node |
ansond | 0:a3fc1c6ef150 | 252 | //this->logger()->log("MQTT Message: %s not for us: %s... ignoring...",payload,endpoint_name); |
ansond | 0:a3fc1c6ef150 | 253 | ; |
ansond | 0:a3fc1c6ef150 | 254 | } |
ansond | 0:a3fc1c6ef150 | 255 | } |
ansond | 0:a3fc1c6ef150 | 256 | |
ansond | 31:e5950e0677be | 257 | // is this the dimming resource? |
ansond | 31:e5950e0677be | 258 | bool MQTTTransport::isDimmingResource(char *resource) { |
ansond | 31:e5950e0677be | 259 | bool isDimming = false; |
ansond | 31:e5950e0677be | 260 | |
ansond | 31:e5950e0677be | 261 | if (resource != NULL && strcmp(resource,"Dimming") == 0) isDimming = true; |
ansond | 31:e5950e0677be | 262 | |
ansond | 31:e5950e0677be | 263 | return isDimming; |
ansond | 31:e5950e0677be | 264 | } |
ansond | 31:e5950e0677be | 265 | |
ansond | 0:a3fc1c6ef150 | 266 | // send result back to MQTT |
ansond | 0:a3fc1c6ef150 | 267 | void MQTTTransport::sendResult(char *endpoint_name,char *resource_name,char *value,bool success) { |
ansond | 0:a3fc1c6ef150 | 268 | if (this->m_connected == true) { |
ansond | 0:a3fc1c6ef150 | 269 | // send the response back to MQTT |
ansond | 0:a3fc1c6ef150 | 270 | this->logger()->log("Sending Response back to MQTT..."); |
ansond | 0:a3fc1c6ef150 | 271 | char message[MAX_MQTT_MESSAGE_LENGTH+1]; |
ansond | 0:a3fc1c6ef150 | 272 | memset(message,0,MAX_MQTT_MESSAGE_LENGTH+1); |
ansond | 0:a3fc1c6ef150 | 273 | char *str_success = IOC_RESPONSE_OK; if (!success) str_success = IOC_RESPONSE_FAILED; |
ansond | 0:a3fc1c6ef150 | 274 | sprintf(message,IOC_RESPONSE_TEMPLATE,IOC_RESPONSE_VERB,endpoint_name,resource_name,value,str_success); |
ansond | 49:965a65122c0f | 275 | |
ansond | 49:965a65122c0f | 276 | // build out the message |
ansond | 49:965a65122c0f | 277 | MQTT::Message mqtt_message; |
ansond | 49:965a65122c0f | 278 | mqtt_message.qos = MQTT::QOS0; |
ansond | 49:965a65122c0f | 279 | mqtt_message.retained = false; |
ansond | 49:965a65122c0f | 280 | mqtt_message.dup = false; |
ansond | 49:965a65122c0f | 281 | mqtt_message.payload = (void*)message; |
ansond | 49:965a65122c0f | 282 | mqtt_message.payloadlen = strlen(message)+1; |
ansond | 49:965a65122c0f | 283 | |
ansond | 49:965a65122c0f | 284 | // publish... |
ansond | 49:965a65122c0f | 285 | int sent = this->m_mqtt->publish(this->getTopic(),&mqtt_message); |
ansond | 49:965a65122c0f | 286 | if (sent == 0) { |
ansond | 0:a3fc1c6ef150 | 287 | this->logger()->log("Result sent successfully"); |
ansond | 0:a3fc1c6ef150 | 288 | this->logger()->blinkTransportTxLED(); |
ansond | 0:a3fc1c6ef150 | 289 | } |
ansond | 0:a3fc1c6ef150 | 290 | else { |
ansond | 0:a3fc1c6ef150 | 291 | this->logger()->log("Result send FAILED"); |
ansond | 0:a3fc1c6ef150 | 292 | } |
ansond | 0:a3fc1c6ef150 | 293 | } |
ansond | 0:a3fc1c6ef150 | 294 | else { |
ansond | 0:a3fc1c6ef150 | 295 | // unable to send the response |
ansond | 0:a3fc1c6ef150 | 296 | this->logger()->log("Unable to send response back to MQTT. Not connected."); |
ansond | 0:a3fc1c6ef150 | 297 | } |
ansond | 0:a3fc1c6ef150 | 298 | } |
ansond | 0:a3fc1c6ef150 | 299 | |
ansond | 0:a3fc1c6ef150 | 300 | char *MQTTTransport::mapIOCResourceToEndpointResource(char *ioc_name) { return this->getMap()->iocNameToEndpointName(ioc_name); } |
ansond | 0:a3fc1c6ef150 | 301 | |
ansond | 0:a3fc1c6ef150 | 302 | char *MQTTTransport::makeID(char *id_template,char *buffer) { |
ansond | 0:a3fc1c6ef150 | 303 | MBEDEndpoint *endpoint = (MBEDEndpoint *)this->getEndpoint(); |
ansond | 0:a3fc1c6ef150 | 304 | int instance_id = rand()%100; |
ansond | 0:a3fc1c6ef150 | 305 | if (endpoint != NULL) instance_id = endpoint->getInstanceID(); |
ansond | 0:a3fc1c6ef150 | 306 | srand(time(0)); |
ansond | 0:a3fc1c6ef150 | 307 | srand(rand()); |
ansond | 0:a3fc1c6ef150 | 308 | sprintf(buffer,id_template,rand()%MQTT_MAXID_VALUE,instance_id); |
ansond | 0:a3fc1c6ef150 | 309 | return buffer; |
ansond | 0:a3fc1c6ef150 | 310 | } |
ansond | 0:a3fc1c6ef150 | 311 | |
ansond | 0:a3fc1c6ef150 | 312 | // is this message a PONG message? |
ansond | 0:a3fc1c6ef150 | 313 | bool MQTTTransport::isPongMessage(char *topic,char *payload,int payload_length) { |
ansond | 0:a3fc1c6ef150 | 314 | bool isPong = false; |
ansond | 0:a3fc1c6ef150 | 315 | char verb[MQTT_PING_VERB_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 316 | char end[MQTT_PING_VERB_LEN+1]; |
ansond | 0:a3fc1c6ef150 | 317 | int counter = 0; |
ansond | 0:a3fc1c6ef150 | 318 | |
ansond | 0:a3fc1c6ef150 | 319 | // clean |
ansond | 0:a3fc1c6ef150 | 320 | memset(verb,0,MQTT_PING_VERB_LEN+1); |
ansond | 0:a3fc1c6ef150 | 321 | memset(end,0,MQTT_PING_VERB_LEN+1); |
ansond | 0:a3fc1c6ef150 | 322 | |
ansond | 0:a3fc1c6ef150 | 323 | // make sure this is for us... |
ansond | 0:a3fc1c6ef150 | 324 | char *topic_ep_name = this->getEndpointNameFromTopic(topic); |
ansond | 0:a3fc1c6ef150 | 325 | if (topic_ep_name != NULL && strcmp(topic_ep_name,this->m_endpoint_name) == 0) { |
ansond | 0:a3fc1c6ef150 | 326 | // parse the payload |
ansond | 0:a3fc1c6ef150 | 327 | for(int i=0;payload != NULL && i<payload_length;++i) if (payload[i] == ':') payload[i] = ' '; |
ansond | 0:a3fc1c6ef150 | 328 | sscanf(payload,"%s%d",verb,&counter); |
ansond | 0:a3fc1c6ef150 | 329 | |
ansond | 0:a3fc1c6ef150 | 330 | // check the contents to make sure its for us... |
ansond | 0:a3fc1c6ef150 | 331 | //this->logger()->log("isPongMessage: verb: %s counter %d ping_counter: %d",verb,counter,this->m_ping_counter); |
ansond | 0:a3fc1c6ef150 | 332 | if (strcmp(verb,"pong") == 0 && counter == this->m_ping_counter) { |
ansond | 0:a3fc1c6ef150 | 333 | // its a PONG message to our PING... |
ansond | 0:a3fc1c6ef150 | 334 | isPong = true; |
ansond | 0:a3fc1c6ef150 | 335 | } |
ansond | 0:a3fc1c6ef150 | 336 | } |
ansond | 0:a3fc1c6ef150 | 337 | |
ansond | 0:a3fc1c6ef150 | 338 | // return isPong status |
ansond | 0:a3fc1c6ef150 | 339 | return isPong; |
ansond | 0:a3fc1c6ef150 | 340 | } |
ansond | 0:a3fc1c6ef150 | 341 | |
ansond | 0:a3fc1c6ef150 | 342 | |
ansond | 0:a3fc1c6ef150 | 343 | // process this PONG message |
ansond | 0:a3fc1c6ef150 | 344 | void MQTTTransport::processPongMessage(char *payload,int payload_length) { |
ansond | 0:a3fc1c6ef150 | 345 | // DEBUG |
ansond | 0:a3fc1c6ef150 | 346 | //this->logger()->log("Received PONG: counter=%d",this->m_ping_counter); |
ansond | 0:a3fc1c6ef150 | 347 | |
ansond | 0:a3fc1c6ef150 | 348 | // simply increment the counter |
ansond | 0:a3fc1c6ef150 | 349 | ++this->m_ping_counter; |
ansond | 0:a3fc1c6ef150 | 350 | |
ansond | 0:a3fc1c6ef150 | 351 | // reset counter if maxed |
ansond | 0:a3fc1c6ef150 | 352 | if (this->m_ping_counter >= MQTT_MAX_COUNTER) this->m_ping_counter = 1; |
ansond | 0:a3fc1c6ef150 | 353 | } |
ansond | 0:a3fc1c6ef150 | 354 | |
ansond | 0:a3fc1c6ef150 | 355 | // send a PING message |
ansond | 0:a3fc1c6ef150 | 356 | bool MQTTTransport::sendPingMessage() { |
ansond | 0:a3fc1c6ef150 | 357 | bool sent = false; |
ansond | 0:a3fc1c6ef150 | 358 | char message[MAX_MQTT_MESSAGE_LENGTH+1]; |
ansond | 0:a3fc1c6ef150 | 359 | |
ansond | 0:a3fc1c6ef150 | 360 | // initialize... |
ansond | 0:a3fc1c6ef150 | 361 | memset(message,0,MAX_MQTT_MESSAGE_LENGTH+1); |
ansond | 0:a3fc1c6ef150 | 362 | |
ansond | 0:a3fc1c6ef150 | 363 | // build message |
ansond | 0:a3fc1c6ef150 | 364 | sprintf(message,"ping:%s:%d:",this->m_endpoint_name,this->m_ping_counter); |
ansond | 0:a3fc1c6ef150 | 365 | |
ansond | 0:a3fc1c6ef150 | 366 | // send the message over the ping/pong topic |
ansond | 0:a3fc1c6ef150 | 367 | //this->logger()->log("Sending PING: counter=%d",this->m_ping_counter); |
ansond | 49:965a65122c0f | 368 | |
ansond | 49:965a65122c0f | 369 | // build out the message |
ansond | 49:965a65122c0f | 370 | MQTT::Message mqtt_message; |
ansond | 49:965a65122c0f | 371 | mqtt_message.qos = MQTT::QOS0; |
ansond | 49:965a65122c0f | 372 | mqtt_message.retained = false; |
ansond | 49:965a65122c0f | 373 | mqtt_message.dup = false; |
ansond | 49:965a65122c0f | 374 | mqtt_message.payload = (void*)message; |
ansond | 49:965a65122c0f | 375 | mqtt_message.payloadlen = strlen(message)+1; |
ansond | 49:965a65122c0f | 376 | |
ansond | 49:965a65122c0f | 377 | // publish... |
ansond | 49:965a65122c0f | 378 | sent = this->m_mqtt->publish(MQTT_PING_SEND_TOPIC,&mqtt_message); |
ansond | 49:965a65122c0f | 379 | if (sent == 0) { |
ansond | 0:a3fc1c6ef150 | 380 | // send succeeded |
ansond | 0:a3fc1c6ef150 | 381 | //this->logger()->log("PING %d sent successfully",this->m_ping_counter); |
ansond | 0:a3fc1c6ef150 | 382 | this->logger()->blinkTransportTxLED(); |
ansond | 0:a3fc1c6ef150 | 383 | |
ansond | 0:a3fc1c6ef150 | 384 | // wait for 1 second |
ansond | 0:a3fc1c6ef150 | 385 | wait_ms(1000); |
ansond | 0:a3fc1c6ef150 | 386 | } |
ansond | 0:a3fc1c6ef150 | 387 | else { |
ansond | 20:ee03f10c074e | 388 | // fail silently now... |
ansond | 20:ee03f10c074e | 389 | ; |
ansond | 20:ee03f10c074e | 390 | |
ansond | 0:a3fc1c6ef150 | 391 | // send failed! - reconnect |
ansond | 20:ee03f10c074e | 392 | //this->logger()->log("PING send %d FAILED... (re)connecting...",this->m_ping_counter); |
ansond | 0:a3fc1c6ef150 | 393 | |
ansond | 0:a3fc1c6ef150 | 394 | // attempt reconnect |
ansond | 20:ee03f10c074e | 395 | //this->logger()->log("PING send failed - re-connecting MQTT..."); |
ansond | 20:ee03f10c074e | 396 | //this->disconnect(); |
ansond | 20:ee03f10c074e | 397 | //sent = this->connect(); |
ansond | 20:ee03f10c074e | 398 | //if (sent) this->logger()->log("PING %d: MQTT reconnection successful...",this->m_ping_counter); |
ansond | 20:ee03f10c074e | 399 | //else this->logger()->log("PING %d: resend failed giving up...",this->m_ping_counter); |
ansond | 0:a3fc1c6ef150 | 400 | } |
ansond | 0:a3fc1c6ef150 | 401 | |
ansond | 0:a3fc1c6ef150 | 402 | // return our status |
ansond | 0:a3fc1c6ef150 | 403 | return sent; |
ansond | 0:a3fc1c6ef150 | 404 | } |
ansond | 0:a3fc1c6ef150 | 405 | |
ansond | 29:ac6390032cec | 406 | static char _mqtt_id[MQTT_ENDPOINT_IDLEN+1]; |
ansond | 29:ac6390032cec | 407 | |
ansond | 0:a3fc1c6ef150 | 408 | // connect up MQTT |
ansond | 0:a3fc1c6ef150 | 409 | bool MQTTTransport::connect() { |
ansond | 29:ac6390032cec | 410 | memset(_mqtt_id,0,(MQTT_ENDPOINT_IDLEN+1)); |
ansond | 0:a3fc1c6ef150 | 411 | if (this->m_connected == false) { |
ansond | 49:965a65122c0f | 412 | this->logger()->log("MQTT Init: %s:%d...",MQTT_HOSTNAME,MQTT_HOSTPORT); |
ansond | 49:965a65122c0f | 413 | if (this->m_ipstack->connect(MQTT_HOSTNAME,MQTT_HOSTPORT)) { |
ansond | 49:965a65122c0f | 414 | |
ansond | 49:965a65122c0f | 415 | MQTTPacket_connectData data = MQTTPacket_connectData_initializer; |
ansond | 49:965a65122c0f | 416 | data.MQTTVersion = 3; |
ansond | 49:965a65122c0f | 417 | data.clientID.cstring = this->makeID(MQTT_ENDPOINT_ID,_mqtt_id); |
ansond | 49:965a65122c0f | 418 | this->logger()->log("MQTT Connect: ID: %s...",data.clientID.cstring); |
ansond | 49:965a65122c0f | 419 | |
ansond | 49:965a65122c0f | 420 | int rc = this->m_mqtt->connect(&data); |
ansond | 49:965a65122c0f | 421 | if (rc != 0) { |
ansond | 49:965a65122c0f | 422 | this->logger()->log("Error from MQTT connect: %d", rc); |
ansond | 49:965a65122c0f | 423 | } |
ansond | 49:965a65122c0f | 424 | else { |
ansond | 49:965a65122c0f | 425 | this->logger()->log("MQTT Subscribe: Topic: %s...",this->getTopic()); |
ansond | 49:965a65122c0f | 426 | if ((rc = this->m_mqtt->subscribe(this->getTopic(),MQTT::QOS1,_mqtt_message_handler)) != 0) { |
ansond | 49:965a65122c0f | 427 | this->logger()->log("MQTT Subscribe: Topic: %s FAILED: %d", this->getTopic(),rc); |
ansond | 49:965a65122c0f | 428 | this->logger()->turnLEDRed(); |
ansond | 49:965a65122c0f | 429 | this->m_connected = false; |
ansond | 49:965a65122c0f | 430 | } |
ansond | 49:965a65122c0f | 431 | else { |
ansond | 49:965a65122c0f | 432 | if ((rc = this->m_mqtt->subscribe(MQTT_IOC_ALL_TOPIC, MQTT::QOS1,_mqtt_message_handler)) == 0) { |
ansond | 0:a3fc1c6ef150 | 433 | this->logger()->log("MQTT CONNECTED."); |
ansond | 0:a3fc1c6ef150 | 434 | this->m_connected = true; |
ansond | 0:a3fc1c6ef150 | 435 | } |
ansond | 0:a3fc1c6ef150 | 436 | else { |
ansond | 49:965a65122c0f | 437 | this->logger()->log("MQTT Subscribe(ALL): Topic: %s FAILED: %d", MQTT_IOC_ALL_TOPIC,rc); |
ansond | 0:a3fc1c6ef150 | 438 | this->logger()->turnLEDRed(); |
ansond | 0:a3fc1c6ef150 | 439 | this->m_connected = false; |
ansond | 0:a3fc1c6ef150 | 440 | } |
ansond | 49:965a65122c0f | 441 | } |
ansond | 49:965a65122c0f | 442 | } |
ansond | 49:965a65122c0f | 443 | } |
ansond | 49:965a65122c0f | 444 | else { |
ansond | 49:965a65122c0f | 445 | this->logger()->log("MQTT Connect FAILED"); |
ansond | 0:a3fc1c6ef150 | 446 | this->logger()->turnLEDRed(); |
ansond | 0:a3fc1c6ef150 | 447 | this->m_connected = false; |
ansond | 49:965a65122c0f | 448 | } |
ansond | 49:965a65122c0f | 449 | } |
ansond | 49:965a65122c0f | 450 | else { |
ansond | 0:a3fc1c6ef150 | 451 | this->logger()->log("MQTT already connected (OK)"); |
ansond | 49:965a65122c0f | 452 | } |
ansond | 49:965a65122c0f | 453 | return this->m_connected; |
ansond | 0:a3fc1c6ef150 | 454 | } |
ansond | 0:a3fc1c6ef150 | 455 | |
ansond | 0:a3fc1c6ef150 | 456 | // disconnect from MQTT |
ansond | 0:a3fc1c6ef150 | 457 | bool MQTTTransport::disconnect() { |
ansond | 0:a3fc1c6ef150 | 458 | if (this->m_mqtt != NULL) { |
ansond | 0:a3fc1c6ef150 | 459 | this->logger()->log("MQTT Unsubscribing from: %s...",this->getTopic()); |
ansond | 49:965a65122c0f | 460 | int rc = this->m_mqtt->unsubscribe(this->getTopic()); |
ansond | 49:965a65122c0f | 461 | if (rc != 0) |
ansond | 49:965a65122c0f | 462 | this->logger()->log("Error in unsubscribing from %s ErrorCode: %d", this->getTopic(),rc); |
ansond | 49:965a65122c0f | 463 | if ((rc = this->m_mqtt->unsubscribe(MQTT_IOC_ALL_TOPIC)) != 0) |
ansond | 49:965a65122c0f | 464 | this->logger()->log("Error in unsubscribing from %s ErrorCode: %d", MQTT_IOC_ALL_TOPIC,rc); |
ansond | 0:a3fc1c6ef150 | 465 | this->logger()->log("MQTT Disconnecting..."); |
ansond | 49:965a65122c0f | 466 | if ((rc = this->m_mqtt->disconnect()) != 0) |
ansond | 49:965a65122c0f | 467 | this->logger()->log("Error from disconnect: %d", rc); |
ansond | 0:a3fc1c6ef150 | 468 | } |
ansond | 0:a3fc1c6ef150 | 469 | else { |
ansond | 0:a3fc1c6ef150 | 470 | this->logger()->log("MQTT already disconnected (OK)"); |
ansond | 0:a3fc1c6ef150 | 471 | } |
ansond | 0:a3fc1c6ef150 | 472 | this->m_connected = false; |
ansond | 0:a3fc1c6ef150 | 473 | return true; |
ansond | 0:a3fc1c6ef150 | 474 | } |
ansond | 0:a3fc1c6ef150 | 475 | |
ansond | 0:a3fc1c6ef150 | 476 | // check transport and process stuff |
ansond | 0:a3fc1c6ef150 | 477 | void MQTTTransport::checkAndProcess() { |
ansond | 0:a3fc1c6ef150 | 478 | // process any MQTT messages |
ansond | 0:a3fc1c6ef150 | 479 | if (this->m_mqtt != NULL && this->m_connected == true) { |
ansond | 49:965a65122c0f | 480 | bool connected = this->m_mqtt->yield(200); |
ansond | 0:a3fc1c6ef150 | 481 | if (connected) { |
ansond | 0:a3fc1c6ef150 | 482 | this->logger()->blinkTransportRxLED(); |
ansond | 0:a3fc1c6ef150 | 483 | } |
ansond | 32:9a024a6af2fb | 484 | //else { |
ansond | 32:9a024a6af2fb | 485 | // this->logger()->log("Attempting reconnection to MQTT in checkAndProcess..."); |
ansond | 32:9a024a6af2fb | 486 | // this->disconnect(); |
ansond | 32:9a024a6af2fb | 487 | // Thread::wait(15000); |
ansond | 32:9a024a6af2fb | 488 | // this->connect(); |
ansond | 32:9a024a6af2fb | 489 | //} |
ansond | 0:a3fc1c6ef150 | 490 | } |
ansond | 0:a3fc1c6ef150 | 491 | |
ansond | 0:a3fc1c6ef150 | 492 | // send a PING if time for it |
ansond | 0:a3fc1c6ef150 | 493 | --this->m_ping_countdown; |
ansond | 0:a3fc1c6ef150 | 494 | if (this->m_ping_countdown <= 0) { |
ansond | 0:a3fc1c6ef150 | 495 | //this->logger()->log("MQTT: Sending PING..."); |
ansond | 0:a3fc1c6ef150 | 496 | this->m_ping_countdown = MQTT_PING_COUNTDOWN; |
ansond | 0:a3fc1c6ef150 | 497 | this->sendPingMessage(); |
ansond | 0:a3fc1c6ef150 | 498 | } |
ansond | 0:a3fc1c6ef150 | 499 | } |