An API for using MQTT over multiple transports

Dependencies:   FP MQTTPacket

Dependents:   Cellular_HelloMQTT IoTStarterKit GSwifiInterface_HelloMQTT IBMIoTClientEthernetExample ... more

Embed: (wiki syntax)

« Back to documentation index

Show/hide line numbers MQTTClient.h Source File

MQTTClient.h

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