Diff: MQTTClient.h

Revision:

26:2658bb87c53d

Parent:

23:05fc7de97d4a

Child:

28:8b2abe9bd814

--- a/MQTTClient.h	Tue May 06 09:44:23 2014 +0000
+++ b/MQTTClient.h	Sun May 11 18:19:07 2014 +0000
@@ -18,13 +18,7 @@
  
  TODO: 
  
- log messages - use macros
- 
- define return code constants
- 
- call connectionLost at appropriate points - in sendPacket and readPacket
- 
- match wildcard topics
+ ensure publish packets are retried on reconnect
  
  */
 
@@ -94,14 +88,7 @@
 {
     
 public:
-
-    typedef struct
-    {
-        Client* client;
-        Network* network;
-    } connectionLostInfo;
-    
-    typedef int (*connectionLostHandlers)(connectionLostInfo*);
+   
     typedef void (*messageHandler)(Message*);
 
     /** Construct the client
@@ -111,14 +98,6 @@
      */
     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
-     */
-    void setConnectionLostHandler(connectionLostHandlers clh)
-    {
-        connectionLostHandler.attach(clh);
-    }
-    
     /** Set the default message handling callback - used for any message which does not match a subscription message handler
      *  @param mh - pointer to the callback function
      */
@@ -164,8 +143,9 @@
      *  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
+     *  @return success code - on failure, this means the client has disconnected
      */
-    void yield(int timeout_ms = 1000);
+    int yield(int timeout_ms = 1000);
     
 private:
 
@@ -176,7 +156,8 @@
     int decodePacket(int* value, int timeout);
     int readPacket(Timer& timer);
     int sendPacket(int length, Timer& timer);
-    int deliverMessage(MQTTString* topic, Message* message);
+    int deliverMessage(MQTTString& topicName, Message& message);
+    bool isTopicMatched(char* topicFilter, MQTTString& topicName);
     
     Network& ipstack;
     unsigned int command_timeout_ms;
@@ -193,15 +174,13 @@
     typedef FP<void, Message*> messageHandlerFP;
     struct MessageHandlers
     {
-        const char* topic;
+        const char* topicFilter;
         messageHandlerFP fp;
     } messageHandlers[MAX_MESSAGE_HANDLERS];      // Message handlers are indexed by subscription topic
     
     messageHandlerFP defaultMessageHandler;
     
-    typedef FP<int, connectionLostInfo*> connectionLostFP;
-    connectionLostFP connectionLostHandler;
-    
+    bool isconnected;
 };
 
 }
@@ -213,8 +192,9 @@
     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;    
+        messageHandlers[i].topicFilter = 0;
+    this->command_timeout_ms = command_timeout_ms; 
+    isconnected = false;
 }
 
 
@@ -227,13 +207,9 @@
     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 (rc < 0)  // there was an error writing the data
+            break;
+        sent += rc;
     }
     if (sent == length)
     {
@@ -246,7 +222,8 @@
 }
 
 
-template<class Network, class Timer, int a, int b> int MQTT::Client<Network, Timer, a, b>::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;
@@ -283,7 +260,7 @@
 template<class Network, class Timer, int a, int b> 
 int MQTT::Client<Network, Timer, a, b>::readPacket(Timer& timer) 
 {
-    int rc = -1;
+    int rc = FAILURE;
     MQTTHeader header = {0};
     int len = 0;
     int rem_len = 0;
@@ -308,23 +285,63 @@
 }
 
 
+// assume topic filter and name is in correct format
+// # can only be at end
+// + and # can only be next to separator
+template<class Network, class Timer, int a, int b> 
+bool MQTT::Client<Network, Timer, a, b>::isTopicMatched(char* topicFilter, MQTTString& topicName)
+{
+    char* curf = topicFilter;
+    char* curn = topicName.lenstring.data;
+    char* curn_end = curn + topicName.lenstring.len;
+    
+    while (*curf && curn < curn_end)
+    {
+        if (*curn == '/' && *curf != '/')
+            break;
+        if (*curf != '+' && *curf != '#' && *curf != *curn)
+            break;
+        if (*curf == '+')
+        {   // skip until we meet the next separator, or end of string
+            char* nextpos = curn + 1;
+            while (nextpos < curn_end && *nextpos != '/')
+                nextpos = ++curn + 1;
+        }
+        else if (*curf == '#')
+            curn = curn_end - 1;    // skip until end of string
+        curf++;
+        curn++;
+    };
+    
+    return (curn == curn_end) && (*curf == '\0');
+}
+
+
+
 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 MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::deliverMessage(MQTTString& topicName, Message& message)
 {
-    int rc = -1;
+    int rc = FAILURE;
 
     // we have to find the right message handler - indexed by topic
     for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
     {
-        if (messageHandlers[i].topic != 0 && MQTTPacket_equals(topic, (char*)messageHandlers[i].topic))
+        if (messageHandlers[i].topicFilter != 0 && (MQTTPacket_equals(&topicName, (char*)messageHandlers[i].topicFilter) ||
+                isTopicMatched((char*)messageHandlers[i].topicFilter, topicName)))
         {
-            messageHandlers[i].fp(message);
-            rc = 0;
-            break;
+            if (messageHandlers[i].fp.attached())
+            {
+                messageHandlers[i].fp(&message);
+                rc = SUCCESS;
+            }
         }
     }
-    if (rc == -1)
-        defaultMessageHandler(message);
+    
+    if (rc == FAILURE && defaultMessageHandler.attached()) 
+    {
+        defaultMessageHandler(&message);
+        rc = SUCCESS;
+    }   
     
     return rc;
 }
@@ -332,13 +349,22 @@
 
 
 template<class Network, class Timer, int a, int b> 
-void MQTT::Client<Network, Timer, a, b>::yield(int timeout_ms)
+int MQTT::Client<Network, Timer, a, b>::yield(int timeout_ms)
 {
+    int rc = SUCCESS;
     Timer timer = Timer();
     
     timer.countdown_ms(timeout_ms);
     while (!timer.expired())
-        cycle(timer);
+    {
+        if (cycle(timer) == FAILURE)
+        {
+            rc = FAILURE;
+            break;
+        }
+    }
+        
+    return rc;
 }
 
 
@@ -350,7 +376,8 @@
     // read the socket, see what work is due
     int packet_type = readPacket(timer);
     
-    int len, rc;
+    int len = 0,
+        rc = SUCCESS;
     switch (packet_type)
     {
         case CONNACK:
@@ -360,27 +387,34 @@
         case PUBLISH:
             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, MAX_MQTT_PACKET_SIZE);;
-            deliverMessage(&topicName, &msg);
+            if (MQTTDeserialize_publish((int*)&msg.dup, (int*)&msg.qos, (int*)&msg.retained, (int*)&msg.id, &topicName,
+                                 (char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+                goto exit;
+            deliverMessage(topicName, msg);
             if (msg.qos != QOS0)
             {
                 if (msg.qos == QOS1)
                     len = MQTTSerialize_ack(buf, MAX_MQTT_PACKET_SIZE, PUBACK, 0, msg.id);
                 else if (msg.qos == QOS2)
                     len = MQTTSerialize_ack(buf, MAX_MQTT_PACKET_SIZE, PUBREC, 0, msg.id);
-                if ((rc = sendPacket(len, timer)) != SUCCESS)
+                if (len <= 0)
+                    rc = FAILURE;
+                else
+                    rc = sendPacket(len, timer);
+                if (rc == FAILURE)
                     goto exit; // there was a problem
             }
             break;
         case PUBREC:
             int type, dup, mypacketid;
-            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
-                ; 
-            len = MQTTSerialize_ack(buf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, mypacketid);
-            if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+                rc = FAILURE;
+            else if ((len = MQTTSerialize_ack(buf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, mypacketid)) <= 0)
+                rc = FAILURE;
+            else if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet
+                rc = FAILURE; // there was a problem
+            if (rc == FAILURE)
                 goto exit; // there was a problem
-
             break;
         case PUBCOMP:
             break;
@@ -390,30 +424,30 @@
     }
     keepalive();
 exit:
-    return packet_type;
+    if (rc == SUCCESS)
+        rc = packet_type;
+    return rc;
 }
 
 
 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;
+    int rc = FAILURE;
 
     if (keepAliveInterval == 0)
+    {
+        rc = SUCCESS;
         goto exit;
+    }
 
     if (ping_timer.expired())
     {
-        if (ping_outstanding)
-            rc = -1;
-        else
+        if (!ping_outstanding)
         {
             Timer timer = Timer(1000);
             int len = MQTTSerialize_pingreq(buf, MAX_MQTT_PACKET_SIZE);
-            rc = sendPacket(len, timer); // send the ping packet
-            if (rc != SUCCESS) 
-                rc = -1; // indicate there's a problem
-            else
+            if (len > 0 && (rc = sendPacket(len, timer)) == SUCCESS) // send the ping packet
                 ping_outstanding = true;
         }
     }
@@ -427,7 +461,7 @@
 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;
+    int rc = FAILURE;
     
     do
     {
@@ -444,6 +478,7 @@
 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect(MQTTPacket_connectData* options)
 {
     Timer connect_timer = Timer(command_timeout_ms);
+    int rc = FAILURE;
 
     MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;
     if (options == 0)
@@ -452,8 +487,9 @@
     this->keepAliveInterval = options->keepAliveInterval;
     ping_timer.countdown(this->keepAliveInterval);
     int len = MQTTSerialize_connect(buf, MAX_MQTT_PACKET_SIZE, options);
-    int rc = sendPacket(len, connect_timer); // send the connect packet
-    if (rc != SUCCESS) 
+    if (len <= 0)
+        goto exit;
+    if ((rc = sendPacket(len, connect_timer)) != SUCCESS)  // send the connect packet
         goto exit; // there was a problem
     
     // this will be a blocking call, wait for the connack
@@ -462,9 +498,15 @@
         int connack_rc = -1;
         if (MQTTDeserialize_connack(&connack_rc, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
             rc = connack_rc;
+        else
+            rc = FAILURE;
     }
+    else
+        rc = FAILURE;
     
 exit:
+    if (rc == SUCCESS)
+        isconnected = true;
     return rc;
 }
 
@@ -472,17 +514,19 @@
 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;
+    int rc = FAILURE;
     Timer timer = Timer(command_timeout_ms);
+    int len = 0;
     
     MQTTString topic = {(char*)topicFilter, 0, 0};
-    
-    int rc = MQTTSerialize_subscribe(buf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos);
-    if (rc <= 0)
+    if (!isconnected)
         goto exit;
-    len = rc;
+    
+    len = MQTTSerialize_subscribe(buf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos);
+    if (len <= 0)
+        goto exit;
     if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet
-        goto exit; // there was a problem
+        goto exit;             // there was a problem
     
     if (waitfor(SUBACK, timer) == SUBACK)      // wait for suback 
     {
@@ -493,9 +537,9 @@
         {
             for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
             {
-                if (messageHandlers[i].topic == 0)
+                if (messageHandlers[i].topicFilter == 0)
                 {
-                    messageHandlers[i].topic = topicFilter;
+                    messageHandlers[i].topicFilter = topicFilter;
                     messageHandlers[i].fp.attach(messageHandler);
                     rc = 0;
                     break;
@@ -503,8 +547,12 @@
             }
         }
     }
-    
+    else 
+        rc = FAILURE;
+        
 exit:
+    //if (rc == FAILURE)
+    //   closesession();
     return rc;
 }
 
@@ -512,15 +560,14 @@
 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;
+    int rc = FAILURE;
     Timer timer = Timer(command_timeout_ms);
     
     MQTTString topic = {(char*)topicFilter, 0, 0};
     
-    int rc = MQTTSerialize_unsubscribe(buf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic);
-    if (rc <= 0)
+    int len = MQTTSerialize_unsubscribe(buf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic);
+    if (len <= 0)
         goto exit;
-    len = rc;
     if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet
         goto exit; // there was a problem
     
@@ -530,6 +577,8 @@
         if (MQTTDeserialize_unsuback(&mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
             rc = 0; 
     }
+    else
+        rc = FAILURE;
     
 exit:
     return rc;
@@ -540,6 +589,7 @@
 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)
 {
+    int rc = FAILURE;
     Timer timer = Timer(command_timeout_ms);
     
     MQTTString topicString = {(char*)topicName, 0, 0};
@@ -549,8 +599,9 @@
     
     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 != SUCCESS) 
+    if (len <= 0)
+        goto exit;
+    if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet
         goto exit; // there was a problem
     
     if (message->qos == QOS1)
@@ -558,18 +609,22 @@
         if (waitfor(PUBACK, timer) == PUBACK)
         {
             int type, dup, mypacketid;
-            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
-                rc = 0; 
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+                rc = FAILURE;
         }
+        else
+            rc = FAILURE;
     }
     else if (message->qos == QOS2)
     {
         if (waitfor(PUBCOMP, timer) == PUBCOMP)
         {
             int type, dup, mypacketid;
-            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
-                rc = 0; 
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+                rc = FAILURE;
         }
+        else
+            rc = FAILURE;
     }
     
 exit:
@@ -580,10 +635,12 @@
 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b> 
 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::disconnect()
 {  
+    int rc = FAILURE;
     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
-    
+    if (len > 0)
+        rc = sendPacket(len, timer);            // send the disconnect packet
+
     return rc;
 }