Modified MQTT for Mbed OS.
Dependents: mbed-os-mqtt door_lock co657_IoT nucleo-f429zi-mbed-os-mqtt
Fork of MQTT by
MQTTClient.h
00001 /******************************************************************************* 00002 * Copyright (c) 2014, 2015 IBM Corp. 00003 * 00004 * All rights reserved. This program and the accompanying materials 00005 * are made available under the terms of the Eclipse Public License v1.0 00006 * and Eclipse Distribution License v1.0 which accompany this distribution. 00007 * 00008 * The Eclipse Public License is available at 00009 * http://www.eclipse.org/legal/epl-v10.html 00010 * and the Eclipse Distribution License is available at 00011 * http://www.eclipse.org/org/documents/edl-v10.php. 00012 * 00013 * Contributors: 00014 * Ian Craggs - initial API and implementation and/or initial documentation 00015 * Ian Craggs - fix for bug 458512 - QoS 2 messages 00016 * Ian Craggs - fix for bug 460389 - send loop uses wrong length 00017 * Ian Craggs - fix for bug 464169 - clearing subscriptions 00018 * Ian Craggs - fix for bug 464551 - enums and ints can be different size 00019 *******************************************************************************/ 00020 00021 #if !defined(MQTTCLIENT_H) 00022 #define MQTTCLIENT_H 00023 00024 #include "FP.h" 00025 #include "MQTTPacket.h" 00026 #include "stdio.h" 00027 #include "MQTTLogging.h" 00028 00029 #if !defined(MQTTCLIENT_QOS1) 00030 #define MQTTCLIENT_QOS1 1 00031 #endif 00032 #if !defined(MQTTCLIENT_QOS2) 00033 #define MQTTCLIENT_QOS2 0 00034 #endif 00035 00036 namespace MQTT 00037 { 00038 00039 00040 enum QoS { QOS0, QOS1, QOS2 }; 00041 00042 // all failure return codes must be negative 00043 enum returnCode { BUFFER_OVERFLOW = -2, FAILURE = -1, SUCCESS = 0 }; 00044 00045 00046 struct Message 00047 { 00048 enum QoS qos; 00049 bool retained; 00050 bool dup; 00051 unsigned short id; 00052 void *payload; 00053 size_t payloadlen; 00054 }; 00055 00056 00057 struct MessageData 00058 { 00059 MessageData(MQTTString &aTopicName, struct Message &aMessage) : message(aMessage), topicName(aTopicName) 00060 { } 00061 00062 struct Message &message; 00063 MQTTString &topicName; 00064 }; 00065 00066 00067 class PacketId 00068 { 00069 public: 00070 PacketId() 00071 { 00072 next = 0; 00073 } 00074 00075 int getNext() 00076 { 00077 return next = (next == MAX_PACKET_ID) ? 1 : ++next; 00078 } 00079 00080 private: 00081 static const int MAX_PACKET_ID = 65535; 00082 int next; 00083 }; 00084 00085 00086 /** 00087 * @class Client 00088 * @brief blocking, non-threaded MQTT client API 00089 * 00090 * This version of the API blocks on all method calls, until they are complete. This means that only one 00091 * MQTT request can be in process at any one time. 00092 * @param Network a network class which supports send, receive 00093 * @param Timer a timer class with the methods: 00094 */ 00095 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE = 200, int MAX_MESSAGE_HANDLERS = 5> 00096 class Client 00097 { 00098 00099 public: 00100 00101 typedef void (*messageHandler)(MessageData&); 00102 00103 /** Construct the client 00104 * @param network - pointer to an instance of the Network class - must be connected to the endpoint 00105 * before calling MQTT connect 00106 * @param limits an instance of the Limit class - to alter limits as required 00107 */ 00108 Client(Network& network, unsigned int command_timeout_ms = 30000); 00109 00110 /** Set the default message handling callback - used for any message which does not match a subscription message handler 00111 * @param mh - pointer to the callback function 00112 */ 00113 void setDefaultMessageHandler(messageHandler mh) 00114 { 00115 defaultMessageHandler.attach(mh); 00116 } 00117 00118 /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack 00119 * The nework object must be connected to the network endpoint before calling this 00120 * Default connect options are used 00121 * @return success code - 00122 */ 00123 int connect(); 00124 00125 /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack 00126 * The nework object must be connected to the network endpoint before calling this 00127 * @param options - connect options 00128 * @return success code - 00129 */ 00130 int connect(MQTTPacket_connectData& options); 00131 00132 /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs 00133 * @param topic - the topic to publish to 00134 * @param message - the message to send 00135 * @return success code - 00136 */ 00137 int publish(const char* topicName, Message& message); 00138 00139 /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs 00140 * @param topic - the topic to publish to 00141 * @param payload - the data to send 00142 * @param payloadlen - the length of the data 00143 * @param qos - the QoS to send the publish at 00144 * @param retained - whether the message should be retained 00145 * @return success code - 00146 */ 00147 int publish(const char* topicName, void* payload, size_t payloadlen, enum QoS qos = QOS0, bool retained = false); 00148 00149 /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs 00150 * @param topic - the topic to publish to 00151 * @param payload - the data to send 00152 * @param payloadlen - the length of the data 00153 * @param id - the packet id used - returned 00154 * @param qos - the QoS to send the publish at 00155 * @param retained - whether the message should be retained 00156 * @return success code - 00157 */ 00158 int publish(const char* topicName, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos = QOS1, bool retained = false); 00159 00160 /** MQTT Subscribe - send an MQTT subscribe packet and wait for the suback 00161 * @param topicFilter - a topic pattern which can include wildcards 00162 * @param qos - the MQTT QoS to subscribe at 00163 * @param mh - the callback function to be invoked when a message is received for this subscription 00164 * @return success code - 00165 */ 00166 int subscribe(const char* topicFilter, enum QoS qos, messageHandler mh); 00167 00168 /** MQTT Unsubscribe - send an MQTT unsubscribe packet and wait for the unsuback 00169 * @param topicFilter - a topic pattern which can include wildcards 00170 * @return success code - 00171 */ 00172 int unsubscribe(const char* topicFilter); 00173 00174 /** MQTT Disconnect - send an MQTT disconnect packet, and clean up any state 00175 * @return success code - 00176 */ 00177 int disconnect(); 00178 00179 /** A call to this API must be made within the keepAlive interval to keep the MQTT connection alive 00180 * yield can be called if no other MQTT operation is needed. This will also allow messages to be 00181 * received. 00182 * @param timeout_ms the time to wait, in milliseconds 00183 * @return success code - on failure, this means the client has disconnected 00184 */ 00185 int yield(unsigned long timeout_ms = 1000L); 00186 00187 /** Is the client connected? 00188 * @return flag - is the client connected or not? 00189 */ 00190 bool isConnected() 00191 { 00192 return isconnected; 00193 } 00194 00195 private: 00196 00197 void cleanSession(); 00198 int cycle(Timer& timer); 00199 int waitfor(int packet_type, Timer& timer); 00200 int keepalive(); 00201 int publish(int len, Timer& timer, enum QoS qos); 00202 00203 int decodePacket(int* value, int timeout); 00204 int readPacket(Timer& timer); 00205 int sendPacket(int length, Timer& timer); 00206 int deliverMessage(MQTTString& topicName, Message& message); 00207 bool isTopicMatched(char* topicFilter, MQTTString& topicName); 00208 00209 Network& ipstack; 00210 unsigned long command_timeout_ms; 00211 00212 unsigned char sendbuf[MAX_MQTT_PACKET_SIZE]; 00213 unsigned char readbuf[MAX_MQTT_PACKET_SIZE]; 00214 00215 Timer last_sent, last_received; 00216 unsigned int keepAliveInterval; 00217 bool ping_outstanding; 00218 bool cleansession; 00219 00220 PacketId packetid; 00221 00222 struct MessageHandlers 00223 { 00224 const char* topicFilter; 00225 FP<void, MessageData&> fp; 00226 } messageHandlers[MAX_MESSAGE_HANDLERS]; // Message handlers are indexed by subscription topic 00227 00228 FP<void, MessageData&> defaultMessageHandler; 00229 00230 bool isconnected; 00231 00232 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 00233 unsigned char pubbuf[MAX_MQTT_PACKET_SIZE]; // store the last publish for sending on reconnect 00234 int inflightLen; 00235 unsigned short inflightMsgid; 00236 enum QoS inflightQoS; 00237 #endif 00238 00239 #if MQTTCLIENT_QOS2 00240 bool pubrel; 00241 #if !defined(MAX_INCOMING_QOS2_MESSAGES) 00242 #define MAX_INCOMING_QOS2_MESSAGES 10 00243 #endif 00244 unsigned short incomingQoS2messages[MAX_INCOMING_QOS2_MESSAGES]; 00245 bool isQoS2msgidFree(unsigned short id); 00246 bool useQoS2msgid(unsigned short id); 00247 void freeQoS2msgid(unsigned short id); 00248 #endif 00249 00250 }; 00251 00252 } 00253 00254 00255 template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS> 00256 void MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::cleanSession() 00257 { 00258 ping_outstanding = false; 00259 for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i) 00260 messageHandlers[i].topicFilter = 0; 00261 isconnected = false; 00262 00263 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 00264 inflightMsgid = 0; 00265 inflightQoS = QOS0; 00266 #endif 00267 00268 #if MQTTCLIENT_QOS2 00269 pubrel = false; 00270 for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) 00271 incomingQoS2messages[i] = 0; 00272 #endif 00273 } 00274 00275 00276 template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS> 00277 MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::Client(Network& network, unsigned int command_timeout_ms) : ipstack(network), packetid() 00278 { 00279 last_sent = Timer(); 00280 last_received = Timer(); 00281 this->command_timeout_ms = command_timeout_ms; 00282 cleanSession(); 00283 } 00284 00285 00286 #if MQTTCLIENT_QOS2 00287 template<class Network, class Timer, int a, int b> 00288 bool MQTT::Client<Network, Timer, a, b>::isQoS2msgidFree(unsigned short id) 00289 { 00290 for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) 00291 { 00292 if (incomingQoS2messages[i] == id) 00293 return false; 00294 } 00295 return true; 00296 } 00297 00298 00299 template<class Network, class Timer, int a, int b> 00300 bool MQTT::Client<Network, Timer, a, b>::useQoS2msgid(unsigned short id) 00301 { 00302 for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) 00303 { 00304 if (incomingQoS2messages[i] == 0) 00305 { 00306 incomingQoS2messages[i] = id; 00307 return true; 00308 } 00309 } 00310 return false; 00311 } 00312 00313 00314 template<class Network, class Timer, int a, int b> 00315 void MQTT::Client<Network, Timer, a, b>::freeQoS2msgid(unsigned short id) 00316 { 00317 for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) 00318 { 00319 if (incomingQoS2messages[i] == id) 00320 { 00321 incomingQoS2messages[i] = 0; 00322 return; 00323 } 00324 } 00325 } 00326 #endif 00327 00328 00329 template<class Network, class Timer, int a, int b> 00330 int MQTT::Client<Network, Timer, a, b>::sendPacket(int length, Timer& timer) 00331 { 00332 int rc = FAILURE, 00333 sent = 0; 00334 00335 while (sent < length && !timer.expired()) 00336 { 00337 rc = ipstack.write(&sendbuf[sent], length - sent, timer.left_ms()); 00338 if (rc < 0) // there was an error writing the data 00339 break; 00340 sent += rc; 00341 } 00342 if (sent == length) 00343 { 00344 if (this->keepAliveInterval > 0) 00345 last_sent.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet 00346 rc = SUCCESS; 00347 } 00348 else 00349 rc = FAILURE; 00350 00351 #if defined(MQTT_DEBUG) 00352 char printbuf[150]; 00353 DEBUG("Rc %d from sending packet %s\n", rc, MQTTFormat_toServerString(printbuf, sizeof(printbuf), sendbuf, length)); 00354 #endif 00355 return rc; 00356 } 00357 00358 00359 template<class Network, class Timer, int a, int b> 00360 int MQTT::Client<Network, Timer, a, b>::decodePacket(int* value, int timeout) 00361 { 00362 unsigned char c; 00363 int multiplier = 1; 00364 int len = 0; 00365 const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4; 00366 00367 *value = 0; 00368 do 00369 { 00370 int rc = MQTTPACKET_READ_ERROR; 00371 00372 if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES) 00373 { 00374 rc = MQTTPACKET_READ_ERROR; /* bad data */ 00375 goto exit; 00376 } 00377 rc = ipstack.read(&c, 1, timeout); 00378 if (rc != 1) 00379 goto exit; 00380 *value += (c & 127) * multiplier; 00381 multiplier *= 128; 00382 } while ((c & 128) != 0); 00383 exit: 00384 return len; 00385 } 00386 00387 00388 /** 00389 * If any read fails in this method, then we should disconnect from the network, as on reconnect 00390 * the packets can be retried. 00391 * @param timeout the max time to wait for the packet read to complete, in milliseconds 00392 * @return the MQTT packet type, or -1 if none 00393 */ 00394 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00395 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::readPacket(Timer& timer) 00396 { 00397 int rc = FAILURE; 00398 MQTTHeader header = {0}; 00399 int len = 0; 00400 int rem_len = 0; 00401 00402 /* 1. read the header byte. This has the packet type in it */ 00403 if (ipstack.read(readbuf, 1, timer.left_ms()) != 1) 00404 goto exit; 00405 00406 len = 1; 00407 /* 2. read the remaining length. This is variable in itself */ 00408 decodePacket(&rem_len, timer.left_ms()); 00409 len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length into the buffer */ 00410 00411 if (rem_len > (MAX_MQTT_PACKET_SIZE - len)) 00412 { 00413 rc = BUFFER_OVERFLOW; 00414 goto exit; 00415 } 00416 00417 /* 3. read the rest of the buffer using a callback to supply the rest of the data */ 00418 if (rem_len > 0 && (ipstack.read(readbuf + len, rem_len, timer.left_ms()) != rem_len)) 00419 goto exit; 00420 00421 header.byte = readbuf[0]; 00422 rc = header.bits.type; 00423 if (this->keepAliveInterval > 0) 00424 last_received.countdown(this->keepAliveInterval); // record the fact that we have successfully received a packet 00425 exit: 00426 00427 #if defined(MQTT_DEBUG) 00428 if (rc >= 0) 00429 { 00430 char printbuf[50]; 00431 DEBUG("Rc %d from receiving packet %s\n", rc, MQTTFormat_toClientString(printbuf, sizeof(printbuf), readbuf, len)); 00432 } 00433 #endif 00434 return rc; 00435 } 00436 00437 00438 // assume topic filter and name is in correct format 00439 // # can only be at end 00440 // + and # can only be next to separator 00441 template<class Network, class Timer, int a, int b> 00442 bool MQTT::Client<Network, Timer, a, b>::isTopicMatched(char* topicFilter, MQTTString& topicName) 00443 { 00444 char* curf = topicFilter; 00445 char* curn = topicName.lenstring.data; 00446 char* curn_end = curn + topicName.lenstring.len; 00447 00448 while (*curf && curn < curn_end) 00449 { 00450 if (*curn == '/' && *curf != '/') 00451 break; 00452 if (*curf != '+' && *curf != '#' && *curf != *curn) 00453 break; 00454 if (*curf == '+') 00455 { // skip until we meet the next separator, or end of string 00456 char* nextpos = curn + 1; 00457 while (nextpos < curn_end && *nextpos != '/') 00458 nextpos = ++curn + 1; 00459 } 00460 else if (*curf == '#') 00461 curn = curn_end - 1; // skip until end of string 00462 curf++; 00463 curn++; 00464 }; 00465 00466 return (curn == curn_end) && (*curf == '\0'); 00467 } 00468 00469 00470 00471 template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS> 00472 int MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::deliverMessage(MQTTString& topicName, Message& message) 00473 { 00474 int rc = FAILURE; 00475 00476 // we have to find the right message handler - indexed by topic 00477 for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i) 00478 { 00479 if (messageHandlers[i].topicFilter != 0 && (MQTTPacket_equals(&topicName, (char*)messageHandlers[i].topicFilter) || 00480 isTopicMatched((char*)messageHandlers[i].topicFilter, topicName))) 00481 { 00482 if (messageHandlers[i].fp.attached()) 00483 { 00484 MessageData md(topicName, message); 00485 messageHandlers[i].fp(md); 00486 rc = SUCCESS; 00487 } 00488 } 00489 } 00490 00491 if (rc == FAILURE && defaultMessageHandler.attached()) 00492 { 00493 MessageData md(topicName, message); 00494 defaultMessageHandler(md); 00495 rc = SUCCESS; 00496 } 00497 00498 return rc; 00499 } 00500 00501 00502 00503 template<class Network, class Timer, int a, int b> 00504 int MQTT::Client<Network, Timer, a, b>::yield(unsigned long timeout_ms) 00505 { 00506 int rc = SUCCESS; 00507 Timer timer = Timer(); 00508 00509 timer.countdown_ms(timeout_ms); 00510 while (!timer.expired()) 00511 { 00512 if (cycle(timer) < 0) 00513 { 00514 rc = FAILURE; 00515 break; 00516 } 00517 } 00518 00519 return rc; 00520 } 00521 00522 00523 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00524 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::cycle(Timer& timer) 00525 { 00526 /* get one piece of work off the wire and one pass through */ 00527 00528 // read the socket, see what work is due 00529 int packet_type = readPacket(timer); 00530 00531 int len = 0, 00532 rc = SUCCESS; 00533 00534 switch (packet_type) 00535 { 00536 case FAILURE: 00537 case BUFFER_OVERFLOW: 00538 rc = packet_type; 00539 break; 00540 case CONNACK: 00541 case PUBACK: 00542 case SUBACK: 00543 break; 00544 case PUBLISH: 00545 { 00546 MQTTString topicName = MQTTString_initializer; 00547 Message msg; 00548 int intQoS; 00549 if (MQTTDeserialize_publish((unsigned char*)&msg.dup, &intQoS, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicName, 00550 (unsigned char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_MQTT_PACKET_SIZE) != 1) 00551 goto exit; 00552 msg.qos = (enum QoS)intQoS; 00553 #if MQTTCLIENT_QOS2 00554 if (msg.qos != QOS2) 00555 #endif 00556 deliverMessage(topicName, msg); 00557 #if MQTTCLIENT_QOS2 00558 else if (isQoS2msgidFree(msg.id)) 00559 { 00560 if (useQoS2msgid(msg.id)) 00561 deliverMessage(topicName, msg); 00562 else 00563 WARN("Maximum number of incoming QoS2 messages exceeded"); 00564 } 00565 #endif 00566 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 00567 if (msg.qos != QOS0) 00568 { 00569 if (msg.qos == QOS1) 00570 len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBACK, 0, msg.id); 00571 else if (msg.qos == QOS2) 00572 len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREC, 0, msg.id); 00573 if (len <= 0) 00574 rc = FAILURE; 00575 else 00576 rc = sendPacket(len, timer); 00577 if (rc == FAILURE) 00578 goto exit; // there was a problem 00579 } 00580 break; 00581 #endif 00582 } 00583 #if MQTTCLIENT_QOS2 00584 case PUBREC: 00585 case PUBREL: 00586 unsigned short mypacketid; 00587 unsigned char dup, type; 00588 if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1) 00589 rc = FAILURE; 00590 else if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, 00591 (packet_type == PUBREC) ? PUBREL : PUBCOMP, 0, mypacketid)) <= 0) 00592 rc = FAILURE; 00593 else if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet 00594 rc = FAILURE; // there was a problem 00595 if (rc == FAILURE) 00596 goto exit; // there was a problem 00597 if (packet_type == PUBREL) 00598 freeQoS2msgid(mypacketid); 00599 break; 00600 00601 case PUBCOMP: 00602 break; 00603 #endif 00604 case PINGRESP: 00605 ping_outstanding = false; 00606 break; 00607 } 00608 keepalive(); 00609 exit: 00610 if (rc == SUCCESS) 00611 rc = packet_type; 00612 return rc; 00613 } 00614 00615 00616 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00617 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::keepalive() 00618 { 00619 int rc = FAILURE; 00620 00621 if (keepAliveInterval == 0) 00622 { 00623 rc = SUCCESS; 00624 goto exit; 00625 } 00626 00627 if (last_sent.expired() || last_received.expired()) 00628 { 00629 if (!ping_outstanding) 00630 { 00631 Timer timer(1000); 00632 int len = MQTTSerialize_pingreq(sendbuf, MAX_MQTT_PACKET_SIZE); 00633 if (len > 0 && (rc = sendPacket(len, timer)) == SUCCESS) // send the ping packet 00634 ping_outstanding = true; 00635 } 00636 } 00637 00638 exit: 00639 return rc; 00640 } 00641 00642 00643 // only used in single-threaded mode where one command at a time is in process 00644 template<class Network, class Timer, int a, int b> 00645 int MQTT::Client<Network, Timer, a, b>::waitfor(int packet_type, Timer& timer) 00646 { 00647 int rc = FAILURE; 00648 00649 do 00650 { 00651 if (timer.expired()) 00652 break; // we timed out 00653 } 00654 while ((rc = cycle(timer)) != packet_type); 00655 00656 return rc; 00657 } 00658 00659 00660 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00661 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect(MQTTPacket_connectData& options) 00662 { 00663 Timer connect_timer(command_timeout_ms); 00664 int rc = FAILURE; 00665 int len = 0; 00666 00667 if (isconnected) // don't send connect packet again if we are already connected 00668 goto exit; 00669 00670 this->keepAliveInterval = options.keepAliveInterval; 00671 this->cleansession = options.cleansession; 00672 if ((len = MQTTSerialize_connect(sendbuf, MAX_MQTT_PACKET_SIZE, &options)) <= 0) 00673 goto exit; 00674 if ((rc = sendPacket(len, connect_timer)) != SUCCESS) // send the connect packet 00675 goto exit; // there was a problem 00676 00677 if (this->keepAliveInterval > 0) 00678 last_received.countdown(this->keepAliveInterval); 00679 // this will be a blocking call, wait for the connack 00680 if (waitfor(CONNACK, connect_timer) == CONNACK) 00681 { 00682 unsigned char connack_rc = 255; 00683 bool sessionPresent = false; 00684 if (MQTTDeserialize_connack((unsigned char*)&sessionPresent, &connack_rc, readbuf, MAX_MQTT_PACKET_SIZE) == 1) 00685 rc = connack_rc; 00686 else 00687 rc = FAILURE; 00688 } 00689 else 00690 rc = FAILURE; 00691 00692 #if MQTTCLIENT_QOS2 00693 // resend any inflight publish 00694 if (inflightMsgid > 0 && inflightQoS == QOS2 && pubrel) 00695 { 00696 if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, inflightMsgid)) <= 0) 00697 rc = FAILURE; 00698 else 00699 rc = publish(len, connect_timer, inflightQoS); 00700 } 00701 else 00702 #endif 00703 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 00704 if (inflightMsgid > 0) 00705 { 00706 memcpy(sendbuf, pubbuf, MAX_MQTT_PACKET_SIZE); 00707 rc = publish(inflightLen, connect_timer, inflightQoS); 00708 } 00709 #endif 00710 00711 exit: 00712 if (rc == SUCCESS) 00713 isconnected = true; 00714 return rc; 00715 } 00716 00717 00718 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00719 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect() 00720 { 00721 MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer; 00722 return connect(default_options); 00723 } 00724 00725 00726 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS> 00727 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::subscribe(const char* topicFilter, enum QoS qos, messageHandler messageHandler) 00728 { 00729 int rc = FAILURE; 00730 Timer timer(command_timeout_ms); 00731 int len = 0; 00732 MQTTString topic = {(char*)topicFilter, {0, 0}}; 00733 00734 if (!isconnected) 00735 goto exit; 00736 00737 len = MQTTSerialize_subscribe(sendbuf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos); 00738 if (len <= 0) 00739 goto exit; 00740 if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet 00741 goto exit; // there was a problem 00742 00743 if (waitfor(SUBACK, timer) == SUBACK) // wait for suback 00744 { 00745 int count = 0, grantedQoS = -1; 00746 unsigned short mypacketid; 00747 if (MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, readbuf, MAX_MQTT_PACKET_SIZE) == 1) 00748 rc = grantedQoS; // 0, 1, 2 or 0x80 00749 if (rc != 0x80) 00750 { 00751 for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i) 00752 { 00753 if (messageHandlers[i].topicFilter == 0) 00754 { 00755 messageHandlers[i].topicFilter = topicFilter; 00756 messageHandlers[i].fp.attach(messageHandler); 00757 rc = 0; 00758 break; 00759 } 00760 } 00761 } 00762 } 00763 else 00764 rc = FAILURE; 00765 00766 exit: 00767 if (rc != SUCCESS) 00768 cleanSession(); 00769 return rc; 00770 } 00771 00772 00773 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS> 00774 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::unsubscribe(const char* topicFilter) 00775 { 00776 int rc = FAILURE; 00777 Timer timer(command_timeout_ms); 00778 MQTTString topic = {(char*)topicFilter, {0, 0}}; 00779 int len = 0; 00780 00781 if (!isconnected) 00782 goto exit; 00783 00784 if ((len = MQTTSerialize_unsubscribe(sendbuf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic)) <= 0) 00785 goto exit; 00786 if ((rc = sendPacket(len, timer)) != SUCCESS) // send the unsubscribe packet 00787 goto exit; // there was a problem 00788 00789 if (waitfor(UNSUBACK, timer) == UNSUBACK) 00790 { 00791 unsigned short mypacketid; // should be the same as the packetid above 00792 if (MQTTDeserialize_unsuback(&mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1) 00793 { 00794 rc = 0; 00795 00796 // remove the subscription message handler associated with this topic, if there is one 00797 for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i) 00798 { 00799 if (messageHandlers[i].topicFilter && strcmp(messageHandlers[i].topicFilter, topicFilter) == 0) 00800 { 00801 messageHandlers[i].topicFilter = 0; 00802 break; 00803 } 00804 } 00805 } 00806 } 00807 else 00808 rc = FAILURE; 00809 00810 exit: 00811 if (rc != SUCCESS) 00812 cleanSession(); 00813 return rc; 00814 } 00815 00816 00817 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00818 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(int len, Timer& timer, enum QoS qos) 00819 { 00820 int rc; 00821 00822 if ((rc = sendPacket(len, timer)) != SUCCESS) // send the publish packet 00823 goto exit; // there was a problem 00824 00825 #if MQTTCLIENT_QOS1 00826 if (qos == QOS1) 00827 { 00828 if (waitfor(PUBACK, timer) == PUBACK) 00829 { 00830 unsigned short mypacketid; 00831 unsigned char dup, type; 00832 if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1) 00833 rc = FAILURE; 00834 else if (inflightMsgid == mypacketid) 00835 inflightMsgid = 0; 00836 } 00837 else 00838 rc = FAILURE; 00839 } 00840 #elif MQTTCLIENT_QOS2 00841 else if (qos == QOS2) 00842 { 00843 if (waitfor(PUBCOMP, timer) == PUBCOMP) 00844 { 00845 unsigned short mypacketid; 00846 unsigned char dup, type; 00847 if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1) 00848 rc = FAILURE; 00849 else if (inflightMsgid == mypacketid) 00850 inflightMsgid = 0; 00851 } 00852 else 00853 rc = FAILURE; 00854 } 00855 #endif 00856 00857 exit: 00858 if (rc != SUCCESS) 00859 cleanSession(); 00860 return rc; 00861 } 00862 00863 00864 00865 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00866 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos, bool retained) 00867 { 00868 int rc = FAILURE; 00869 Timer timer(command_timeout_ms); 00870 MQTTString topicString = MQTTString_initializer; 00871 int len = 0; 00872 00873 if (!isconnected) 00874 goto exit; 00875 00876 topicString.cstring = (char*)topicName; 00877 00878 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 00879 if (qos == QOS1 || qos == QOS2) 00880 id = packetid.getNext(); 00881 #endif 00882 00883 len = MQTTSerialize_publish(sendbuf, MAX_MQTT_PACKET_SIZE, 0, qos, retained, id, 00884 topicString, (unsigned char*)payload, payloadlen); 00885 if (len <= 0) 00886 goto exit; 00887 00888 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 00889 if (!cleansession) 00890 { 00891 memcpy(pubbuf, sendbuf, len); 00892 inflightMsgid = id; 00893 inflightLen = len; 00894 inflightQoS = qos; 00895 #if MQTTCLIENT_QOS2 00896 pubrel = false; 00897 #endif 00898 } 00899 #endif 00900 00901 rc = publish(len, timer, qos); 00902 exit: 00903 return rc; 00904 } 00905 00906 00907 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00908 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, void* payload, size_t payloadlen, enum QoS qos, bool retained) 00909 { 00910 unsigned short id = 0; // dummy - not used for anything 00911 return publish(topicName, payload, payloadlen, id, qos, retained); 00912 } 00913 00914 00915 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00916 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, Message& message) 00917 { 00918 return publish(topicName, message.payload, message.payloadlen, message.qos, message.retained); 00919 } 00920 00921 00922 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 00923 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::disconnect() 00924 { 00925 int rc = FAILURE; 00926 Timer timer(command_timeout_ms); // we might wait for incomplete incoming publishes to complete 00927 int len = MQTTSerialize_disconnect(sendbuf, MAX_MQTT_PACKET_SIZE); 00928 if (len > 0) 00929 rc = sendPacket(len, timer); // send the disconnect packet 00930 00931 if (cleansession) 00932 cleanSession(); 00933 else 00934 isconnected = false; 00935 return rc; 00936 } 00937 00938 00939 #endif
Generated on Tue Jul 12 2022 23:04:24 by 1.7.2