Diff: MQTTClient.h

Revision:

23:05fc7de97d4a

Parent:

22:aadb79d29330

Child:

25:d13a6c558164

Child:

26:2658bb87c53d

diff -r aadb79d29330 -r 05fc7de97d4a MQTTClient.h
--- a/MQTTClient.h	Wed Apr 30 13:03:45 2014 +0000
+++ b/MQTTClient.h	Tue May 06 09:44:23 2014 +0000
@@ -19,8 +19,13 @@
  TODO: 
  
  log messages - use macros
+ 
  define return code constants
  
+ call connectionLost at appropriate points - in sendPacket and readPacket
+ 
+ match wildcard topics
+ 
  */
 
 #if !defined(MQTTCLIENT_H)
@@ -36,6 +41,8 @@
 
 enum QoS { QOS0, QOS1, QOS2 };
 
+enum returnCode { BUFFER_OVERFLOW = -2, FAILURE = -1, SUCCESS = 0 };
+
 
 struct Message
 {
@@ -58,50 +65,36 @@
 class PacketId
 {
 public:
-    PacketId();
+    PacketId()
+    {
+        next = 0;
+    }
     
-    int getNext();
+    int getNext()
+    {
+        return next = (next == MAX_PACKET_ID) ? 1 : ++next;
+    }
    
 private:
     static const int MAX_PACKET_ID = 65535;
     int next;
 };
-
-typedef void (*messageHandler)(Message*);
-
-typedef struct limits
-{
-    int MAX_MQTT_PACKET_SIZE; // 
-    int MAX_MESSAGE_HANDLERS;  // each subscription requires a message handler
-    long command_timeout_ms;
-        
-    limits()
-    {
-        MAX_MQTT_PACKET_SIZE = 100;
-        MAX_MESSAGE_HANDLERS = 5;
-        command_timeout_ms = 30000;
-    }
-} Limits;
   
   
 /**
  * @class Client
  * @brief blocking, non-threaded MQTT client API
+ * 
+ * This version of the API blocks on all method calls, until they are complete.  This means that only one
+ * MQTT request can be in process at any one time.  
  * @param Network a network class which supports send, receive
  * @param Timer a timer class with the methods: 
  */ 
-template<class Network, class Timer> class Client
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE = 100, int MAX_MESSAGE_HANDLERS = 5> class Client
 {
     
 public:
 
-    /** Construct the client
-     *  @param network - pointer to an instance of the Network class - must be connected to the endpoint
-     *      before calling MQTT connect
-     *  @param limits an instance of the Limit class - to alter limits as required
-     */
-    Client(Network* network, const Limits limits = Limits()); 
-    
     typedef struct
     {
         Client* client;
@@ -109,6 +102,14 @@
     } connectionLostInfo;
     
     typedef int (*connectionLostHandlers)(connectionLostInfo*);
+    typedef void (*messageHandler)(Message*);
+
+    /** Construct the client
+     *  @param network - pointer to an instance of the Network class - must be connected to the endpoint
+     *      before calling MQTT connect
+     *  @param limits an instance of the Limit class - to alter limits as required
+     */
+    Client(Network& network, unsigned int command_timeout_ms = 30000); 
     
     /** Set the connection lost callback - called whenever the connection is lost and we should be connected
      *  @param clh - pointer to the callback function
@@ -162,8 +163,9 @@
     /** A call to this API must be made within the keepAlive interval to keep the MQTT connection alive
      *  yield can be called if no other MQTT operation is needed.  This will also allow messages to be 
      *  received.
+     *  @param timeout_ms the time to wait, in milliseconds
      */
-    void yield(int timeout);
+    void yield(int timeout_ms = 1000);
     
 private:
 
@@ -176,12 +178,11 @@
     int sendPacket(int length, Timer& timer);
     int deliverMessage(MQTTString* topic, Message* message);
     
-    Network* ipstack;
+    Network& ipstack;
+    unsigned int command_timeout_ms;
     
-    Limits limits;
-    
-    char* buf;  
-    char* readbuf;
+    char buf[MAX_MQTT_PACKET_SIZE];  
+    char readbuf[MAX_MQTT_PACKET_SIZE];  
 
     Timer ping_timer;
     unsigned int keepAliveInterval;
@@ -194,12 +195,11 @@
     {
         const char* topic;
         messageHandlerFP fp;
-    } *messageHandlers;      // Message handlers are indexed by subscription topic
+    } messageHandlers[MAX_MESSAGE_HANDLERS];      // Message handlers are indexed by subscription topic
     
     messageHandlerFP defaultMessageHandler;
     
     typedef FP<int, connectionLostInfo*> connectionLostFP;
-    
     connectionLostFP connectionLostHandler;
     
 };
@@ -207,34 +207,46 @@
 }
 
 
-template<class Network, class Timer> MQTT::Client<Network, Timer>::Client(Network* network, Limits limits)  : limits(limits), packetid()
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS> 
+MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::Client(Network& network, unsigned int command_timeout_ms)  : ipstack(network), packetid()
 {
-    this->ipstack = network;
-    this->ping_timer = Timer();
-    this->ping_outstanding = 0;
-       
-    // How to make these memory allocations portable?  I was hoping to avoid the heap
-    buf = new char[limits.MAX_MQTT_PACKET_SIZE];
-    readbuf = new char[limits.MAX_MQTT_PACKET_SIZE];
-    this->messageHandlers = new struct MessageHandlers[limits.MAX_MESSAGE_HANDLERS];
-    for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i)
+    ping_timer = Timer();
+    ping_outstanding = 0;
+    for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
         messageHandlers[i].topic = 0;
+    this->command_timeout_ms = command_timeout_ms;    
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::sendPacket(int length, Timer& timer)
+template<class Network, class Timer, int a, int b> 
+int MQTT::Client<Network, Timer, a, b>::sendPacket(int length, Timer& timer)
 {
-    int sent = 0;
+    int rc = FAILURE, 
+        sent = 0;
     
-    while (sent < length)
-        sent += ipstack->write(&buf[sent], length, timer.left_ms());
+    while (sent < length && !timer.expired())
+    {
+        rc = ipstack.write(&buf[sent], length, timer.left_ms());
+        if (rc == -1)
+        {
+            connectionLostInfo info = {this, &ipstack};
+            connectionLostHandler(&info);
+        }
+        else
+            sent += rc;
+    }
     if (sent == length)
+    {
         ping_timer.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet    
-    return sent;
+        rc = SUCCESS;
+    }
+    else
+        rc = FAILURE;
+    return rc;
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::decodePacket(int* value, int timeout)
+template<class Network, class Timer, int a, int b> int MQTT::Client<Network, Timer, a, b>::decodePacket(int* value, int timeout)
 {
     char c;
     int multiplier = 1;
@@ -251,7 +263,7 @@
             rc = MQTTPACKET_READ_ERROR; /* bad data */
             goto exit;
         }
-        rc = ipstack->read(&c, 1, timeout);
+        rc = ipstack.read(&c, 1, timeout);
         if (rc != 1)
             goto exit;
         *value += (c & 127) * multiplier;
@@ -268,7 +280,8 @@
  * @param timeout the max time to wait for the packet read to complete, in milliseconds
  * @return the MQTT packet type, or -1 if none
  */
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::readPacket(Timer& timer) 
+template<class Network, class Timer, int a, int b> 
+int MQTT::Client<Network, Timer, a, b>::readPacket(Timer& timer) 
 {
     int rc = -1;
     MQTTHeader header = {0};
@@ -276,7 +289,7 @@
     int rem_len = 0;
 
     /* 1. read the header byte.  This has the packet type in it */
-    if (ipstack->read(readbuf, 1, timer.left_ms()) != 1)
+    if (ipstack.read(readbuf, 1, timer.left_ms()) != 1)
         goto exit;
 
     len = 1;
@@ -285,7 +298,7 @@
     len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length back into the buffer */
 
     /* 3. read the rest of the buffer using a callback to supply the rest of the data */
-    if (ipstack->read(readbuf + len, rem_len, timer.left_ms()) != rem_len)
+    if (ipstack.read(readbuf + len, rem_len, timer.left_ms()) != rem_len)
         goto exit;
 
     header.byte = readbuf[0];
@@ -295,12 +308,13 @@
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::deliverMessage(MQTTString* topic, Message* message)
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS> 
+int MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::deliverMessage(MQTTString* topic, Message* message)
 {
     int rc = -1;
 
     // we have to find the right message handler - indexed by topic
-    for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i)
+    for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
     {
         if (messageHandlers[i].topic != 0 && MQTTPacket_equals(topic, (char*)messageHandlers[i].topic))
         {
@@ -317,17 +331,19 @@
 
 
 
-template<class Network, class Timer> void MQTT::Client<Network, Timer>::yield(int timeout)
+template<class Network, class Timer, int a, int b> 
+void MQTT::Client<Network, Timer, a, b>::yield(int timeout_ms)
 {
     Timer timer = Timer();
     
-    timer.countdown_ms(timeout);
+    timer.countdown_ms(timeout_ms);
     while (!timer.expired())
         cycle(timer);
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::cycle(Timer& timer)
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::cycle(Timer& timer)
 {
     /* get one piece of work off the wire and one pass through */
 
@@ -345,26 +361,24 @@
             MQTTString topicName;
             Message msg;
             rc = MQTTDeserialize_publish((int*)&msg.dup, (int*)&msg.qos, (int*)&msg.retained, (int*)&msg.id, &topicName,
-                                 (char**)&msg.payload, (int*)&msg.payloadlen, readbuf, limits.MAX_MQTT_PACKET_SIZE);;
+                                 (char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_MQTT_PACKET_SIZE);;
             deliverMessage(&topicName, &msg);
             if (msg.qos != QOS0)
             {
                 if (msg.qos == QOS1)
-                    len = MQTTSerialize_ack(buf, limits.MAX_MQTT_PACKET_SIZE, PUBACK, 0, msg.id);
+                    len = MQTTSerialize_ack(buf, MAX_MQTT_PACKET_SIZE, PUBACK, 0, msg.id);
                 else if (msg.qos == QOS2)
-                    len = MQTTSerialize_ack(buf, limits.MAX_MQTT_PACKET_SIZE, PUBREC, 0, msg.id);
-                rc = sendPacket(len, timer); 
-                if (rc != len) 
+                    len = MQTTSerialize_ack(buf, MAX_MQTT_PACKET_SIZE, PUBREC, 0, msg.id);
+                if ((rc = sendPacket(len, timer)) != SUCCESS)
                     goto exit; // there was a problem
             }
             break;
         case PUBREC:
             int type, dup, mypacketid;
-            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
                 ; 
-            len = MQTTSerialize_ack(buf, limits.MAX_MQTT_PACKET_SIZE, PUBREL, 0, mypacketid);
-            rc = sendPacket(len, timer); // send the PUBREL packet
-            if (rc != len) 
+            len = MQTTSerialize_ack(buf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, mypacketid);
+            if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet
                 goto exit; // there was a problem
 
             break;
@@ -380,7 +394,8 @@
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::keepalive()
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::keepalive()
 {
     int rc = 0;
 
@@ -394,9 +409,9 @@
         else
         {
             Timer timer = Timer(1000);
-            int len = MQTTSerialize_pingreq(buf, limits.MAX_MQTT_PACKET_SIZE);
+            int len = MQTTSerialize_pingreq(buf, MAX_MQTT_PACKET_SIZE);
             rc = sendPacket(len, timer); // send the ping packet
-            if (rc != len) 
+            if (rc != SUCCESS) 
                 rc = -1; // indicate there's a problem
             else
                 ping_outstanding = true;
@@ -409,7 +424,8 @@
 
 
 // only used in single-threaded mode where one command at a time is in process
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::waitfor(int packet_type, Timer& timer)
+template<class Network, class Timer, int a, int b> 
+int MQTT::Client<Network, Timer, a, b>::waitfor(int packet_type, Timer& timer)
 {
     int rc = -1;
     
@@ -424,9 +440,10 @@
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::connect(MQTTPacket_connectData* options)
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect(MQTTPacket_connectData* options)
 {
-    Timer connect_timer = Timer(limits.command_timeout_ms);
+    Timer connect_timer = Timer(command_timeout_ms);
 
     MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;
     if (options == 0)
@@ -434,16 +451,16 @@
     
     this->keepAliveInterval = options->keepAliveInterval;
     ping_timer.countdown(this->keepAliveInterval);
-    int len = MQTTSerialize_connect(buf, limits.MAX_MQTT_PACKET_SIZE, options);
+    int len = MQTTSerialize_connect(buf, MAX_MQTT_PACKET_SIZE, options);
     int rc = sendPacket(len, connect_timer); // send the connect packet
-    if (rc != len) 
+    if (rc != SUCCESS) 
         goto exit; // there was a problem
     
     // this will be a blocking call, wait for the connack
     if (waitfor(CONNACK, connect_timer) == CONNACK)
     {
         int connack_rc = -1;
-        if (MQTTDeserialize_connack(&connack_rc, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+        if (MQTTDeserialize_connack(&connack_rc, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
             rc = connack_rc;
     }
     
@@ -452,28 +469,29 @@
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::subscribe(const char* topicFilter, enum QoS qos, messageHandler messageHandler)
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS> 
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::subscribe(const char* topicFilter, enum QoS qos, messageHandler messageHandler)
 { 
     int len = -1;
-    Timer timer = Timer(limits.command_timeout_ms);
+    Timer timer = Timer(command_timeout_ms);
     
     MQTTString topic = {(char*)topicFilter, 0, 0};
     
-    int rc = MQTTSerialize_subscribe(buf, limits.MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos);
+    int rc = MQTTSerialize_subscribe(buf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos);
     if (rc <= 0)
         goto exit;
     len = rc;
-    if ((rc = sendPacket(len, timer)) != len) // send the subscribe packet
+    if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet
         goto exit; // there was a problem
     
     if (waitfor(SUBACK, timer) == SUBACK)      // wait for suback 
     {
         int count = 0, grantedQoS = -1, mypacketid;
-        if (MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+        if (MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
             rc = grantedQoS; // 0, 1, 2 or 0x80 
         if (rc != 0x80)
         {
-            for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i)
+            for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
             {
                 if (messageHandlers[i].topic == 0)
                 {
@@ -491,24 +509,25 @@
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::unsubscribe(const char* topicFilter)
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS> 
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::unsubscribe(const char* topicFilter)
 {   
     int len = -1;
-    Timer timer = Timer(limits.command_timeout_ms);
+    Timer timer = Timer(command_timeout_ms);
     
     MQTTString topic = {(char*)topicFilter, 0, 0};
     
-    int rc = MQTTSerialize_unsubscribe(buf, limits.MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic);
+    int rc = MQTTSerialize_unsubscribe(buf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic);
     if (rc <= 0)
         goto exit;
     len = rc;
-    if ((rc = sendPacket(len, timer)) != len) // send the subscribe packet
+    if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet
         goto exit; // there was a problem
     
     if (waitfor(UNSUBACK, timer) == UNSUBACK)
     {
         int mypacketid;  // should be the same as the packetid above
-        if (MQTTDeserialize_unsuback(&mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+        if (MQTTDeserialize_unsuback(&mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
             rc = 0; 
     }
     
@@ -518,19 +537,20 @@
 
 
    
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::publish(const char* topicName, Message* message)
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, Message* message)
 {
-    Timer timer = Timer(limits.command_timeout_ms);
+    Timer timer = Timer(command_timeout_ms);
     
     MQTTString topicString = {(char*)topicName, 0, 0};
 
     if (message->qos == QOS1 || message->qos == QOS2)
         message->id = packetid.getNext();
     
-    int len = MQTTSerialize_publish(buf, limits.MAX_MQTT_PACKET_SIZE, 0, message->qos, message->retained, message->id, 
+    int len = MQTTSerialize_publish(buf, MAX_MQTT_PACKET_SIZE, 0, message->qos, message->retained, message->id, 
               topicString, (char*)message->payload, message->payloadlen);
     int rc = sendPacket(len, timer); // send the subscribe packet
-    if (rc != len) 
+    if (rc != SUCCESS) 
         goto exit; // there was a problem
     
     if (message->qos == QOS1)
@@ -538,7 +558,7 @@
         if (waitfor(PUBACK, timer) == PUBACK)
         {
             int type, dup, mypacketid;
-            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
                 rc = 0; 
         }
     }
@@ -547,7 +567,7 @@
         if (waitfor(PUBCOMP, timer) == PUBCOMP)
         {
             int type, dup, mypacketid;
-            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
                 rc = 0; 
         }
     }
@@ -557,13 +577,14 @@
 }
 
 
-template<class Network, class Timer> int MQTT::Client<Network, Timer>::disconnect()
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::disconnect()
 {  
-    Timer timer = Timer(limits.command_timeout_ms);     // we might wait for incomplete incoming publishes to complete
-    int len = MQTTSerialize_disconnect(buf, limits.MAX_MQTT_PACKET_SIZE);
+    Timer timer = Timer(command_timeout_ms);     // we might wait for incomplete incoming publishes to complete
+    int len = MQTTSerialize_disconnect(buf, MAX_MQTT_PACKET_SIZE);
     int rc = sendPacket(len, timer);   // send the disconnect packet
     
-    return (rc == len) ? 0 : -1;
+    return rc;
 }