Revision 6:4d312a49200b, committed 2014-04-08

Comitter:

icraggs

Date:

Tue Apr 08 22:54:37 2014 +0000

Parent:

5:389ccac5a50c

Child:

7:f9d690fb6dad

Commit message:

Subscribing

Changed in this revision

--- a/MQTTClient.cpp	Mon Apr 07 23:52:57 2014 +0100
+++ b/MQTTClient.cpp	Tue Apr 08 22:54:37 2014 +0000
@@ -1,41 +1,43 @@
-/**
- * @file    MQTTPubSub.cpp
- * @brief   API - for MQTT
- * @author  
- * @version 1.0
- * @see     
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
  *
- * Copyright (c) 2014
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
  *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
  *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
  
 #include "MQTTClient.h"
 #include "MQTTPacket.h"
 
-template<class Network, class Thread> MQTT::Client<Network, Thread>::Client(Network* network, const int buffer_size, const int command_timeout)
+template<class Network, class Timer, class Thread> MQTT::Client<Network, Timer, Thread>::Client(Network* network, Timer* timer, const int buffer_size, const int command_timeout)
 {
     
-   buf = new char[buffer_size];
+   buf = new char[buffer_size];
    readbuf = new char[buffer_size];
    this->ipstack = ipstack;
    this->command_timeout = command_timeout;
    //this->thread = new Thread(0); // only need a background thread for non-blocking mode
    this->ipstack = network;
+   this->packetid = 0;
+   this->timer = timer;
 }
 
 
-template<class Network, class Thread> int MQTT::Client<Network, Thread>::sendPacket(int length)
+template<class Network, class Timer, class Thread> int MQTT::Client<Network, Timer, Thread>::getPacketId()
+{
+	return this->packetid = (this->packetid == MAX_PACKET_ID) ? 1 : ++this->packetid;
+}
+
+
+template<class Network, class Timer, class Thread> int MQTT::Client<Network, Timer, Thread>::sendPacket(int length, int timeout)
 {
     int sent = 0;
     
@@ -46,7 +48,7 @@
 }
 
 
-template<class Network, class Thread> int MQTT::Client<Network, Thread>::decodePacket(int* value, int timeout)
+template<class Network, class Timer, class Thread> int MQTT::Client<Network, Timer, Thread>::decodePacket(int* value, int timeout)
 {
     char c;
     int multiplier = 1;
@@ -74,7 +76,13 @@
 }
 
 
-template<class Network, class Thread> int MQTT::Client<Network,Thread>::readPacket(int timeout) 
+/**
+ * If any read fails in this method, then we should disconnect from the network, as on reconnect
+ * the packets can be retried. 
+ * @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, class Thread> int MQTT::Client<Network, Timer, Thread>::readPacket(int timeout) 
 {
     int rc = -1;
     MQTTHeader header = {0};
@@ -82,7 +90,7 @@
     int rem_len = 0;
 
     /* 1. read the header byte.  This has the packet type in it */
-    if ((rc = ipstack->read(readbuf, 1, timeout)) != 1)
+    if (ipstack->read(readbuf, 1, timeout) != 1)
         goto exit;
 
     len = 1;
@@ -91,7 +99,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 ((rc = ipstack->read(readbuf + len, rem_len, timeout)) != rem_len)
+    if (ipstack->read(readbuf + len, rem_len, timeout) != rem_len)
         goto exit;
 
     header.byte = readbuf[0];
@@ -101,19 +109,19 @@
 }
 
 
-template<class Network, class Thread> int MQTT::Client<Network, Thread>::cycle()
+template<class Network, class Timer, class Thread> int MQTT::Client<Network, Timer, Thread>::cycle()
 {
     int timeout = 1000L;
     /* get one piece of work off the wire and one pass through */
     
     // 1. read the socket, see what work is due. 
-    int packet_type = readPacket(-1);
-
+    int packet_type = readPacket(-1);
+
 	printf("packet type %d\n", packet_type);
     
     switch (packet_type)
     {
-        case CONNACK:
+        case CONNACK:
 			printf("connack received\n");
             break;
         case PUBACK:
@@ -131,32 +139,33 @@
 }
 
 
-template<class Network, class Thread> int MQTT::Client<Network, Thread>::connect(MQTTPacket_connectData* options, FP<void, MQTT::Result*> *resultHandler)
+template<class Network, class Timer, class Thread> int MQTT::Client<Network, Timer, Thread>::connect(MQTTPacket_connectData* options, FP<void, MQTT::Result*> *resultHandler)
 {
-	int len = 0;
-	int rc = -99;
-
-    /* 2. if the connect was successful, send the MQTT connect packet */   
-	if (options == 0)
-	{
-		MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;
-		default_options.clientID.cstring = "me";
-		len = MQTTSerialize_connect(buf, buflen, &default_options);
-	}
-	else
-		len = MQTTSerialize_connect(buf, buflen, options);
-    rc = sendPacket(len); // send the connect packet
+	int len = 0;
+	int rc = -99;
+	MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;
+
+    /* 2. if the connect was successful, send the MQTT connect packet */   
+	if (options == 0)
+	{
+		default_options.clientID.cstring = "me";
+		options = &default_options;
+	}
+	
+	this->keepalive = options->keepAliveInterval;
+	len = MQTTSerialize_connect(buf, buflen, options);
+    rc = sendPacket(len); // send the connect packet
 	printf("rc from send is %d\n", rc);
     
     /* 3. wait until the connack is received */
     if (resultHandler == 0)
     {
-        // this will be a blocking call, wait for the connack
-		if (cycle() == CONNACK)
-		{
-			int connack_rc = -1;
-			if (MQTTDeserialize_connack(&connack_rc, readbuf, readbuflen) == 1)
-				rc = connack_rc;
+        // this will be a blocking call, wait for the connack
+		if (cycle() == CONNACK)
+		{
+			int connack_rc = -1;
+			if (MQTTDeserialize_connack(&connack_rc, readbuf, readbuflen) == 1)
+				rc = connack_rc;
 		}
     }
     else
@@ -166,3 +175,34 @@
     
     return rc;
 }
+
+
+template<class Network, class Timer, class Thread> int MQTT::Client<Network, Timer, Thread>::subscribe(const char* topicFilter, enum QoS qos, FP<void, Message*> messageHandler, 
+		FP<void, Result*> *resultHandler)
+{
+	int rc = -1, 
+	    len = 0;
+	MQTTString topic = {(char*)topicFilter, 0, 0};
+	
+	len = MQTTSerialize_subscribe(buf, buflen, 0, getPacketId(), 1, &topic, (int*)&qos);
+	rc = sendPacket(len); // send the subscribe packet
+	
+	/* wait for suback */
+    if (resultHandler == 0)
+    {
+        // this will block
+		if (cycle() == SUBACK)
+		{
+			int count = 0, grantedQoS = -1;
+			if (MQTTDeserialize_suback(&packetid, 1, &count, &grantedQoS, readbuf, readbuflen) == 1)
+				rc = grantedQoS; // 0, 1, 2 or 0x80 
+		}
+    }
+    else
+    {
+        // set subscribe response callback function
+        
+    }
+	
+	return rc;
+}
\ No newline at end of file

--- a/MQTTClient.h	Mon Apr 07 23:52:57 2014 +0100
+++ b/MQTTClient.h	Tue Apr 08 22:54:37 2014 +0000
@@ -1,33 +1,30 @@
-/**
- * @file    MQTTPubSub.h
- * @brief   API - for MQTT
- * @author  
- * @version 1.0
- * @see     
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
  *
- * Copyright (c) 2014
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
  *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
  *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
 
 #if !defined(MQTTCLIENT_H)
 #define MQTTCLIENT_H
 
 #include "FP.h"
 #include "MQTTPacket.h"
+#include "stdio.h"
 
 namespace MQTT
 {
+    
+const int MAX_PACKET_ID = 65535;
 
 
 enum QoS { QOS0, QOS1, QOS2 };
@@ -43,27 +40,27 @@
     size_t payloadlen;
 };
 
-template<class Network, class Thread> class Client;
+template<class Network, class Timer, class Thread> class Client;
 
 class Result
 {
     /* success or failure result data */
-    Client<class Network, class Thread>* client;
+    Client<class Network, class Timer, class Thread>* client;
 };
 
   
-template<class Network, class Thread> class Client
+template<class Network, class Timer, class Thread> class Client
 {
     
 public:    
    
-    Client(Network* network, const int buffer_size = 100, const int command_timeout = 30);  
+    Client(Network* network, Timer* timer, const int buffer_size = 100, const int command_timeout = 30);  
        
     int connect(MQTTPacket_connectData* options = 0, FP<void, Result*> *resultHandler = 0);
         
-    int publish(char* topic, Message* message, FP<void, Result*> *resultHandler = 0);
+    int publish(const char* topic, Message* message, FP<void, Result*> *resultHandler = 0);
     
-    int subscribe(char* topicFilter, int qos, FP<void, Message*> messageHandler, FP<void, Result*> *resultHandler = 0);
+    int subscribe(const char* topicFilter, enum QoS qos, FP<void, Message*> messageHandler, FP<void, Result*> *resultHandler = 0);
     
     int unsubscribe(char* topicFilter, FP<void, Result*> *resultHandler = 0);
     
@@ -71,14 +68,16 @@
     
 private:
 
+    int getPacketId();
     int cycle();
 
     int decodePacket(int* value, int timeout);
     int readPacket(int timeout = -1);
-    int sendPacket(int length);
+    int sendPacket(int length, int timeout = -1);
     
     Thread* thread;
     Network* ipstack;
+    Timer* timer;
     
     char* buf;
     int buflen;
@@ -86,7 +85,9 @@
     char* readbuf;
     int readbuflen;
     
-    int command_timeout;
+    int command_timeout; // max time to wait for any MQTT command to complete, in seconds
+    int keepalive;
+    int packetid;
     
 };