v-16

Dependencies:   FP MQTTPacket

Files at this revision

API Documentation at this revision

Comitter:
DuyLionTran
Date:
Tue Dec 12 15:58:09 2017 +0000
Commit message:
version 1.6;

Changed in this revision

FP.lib Show annotated file Show diff for this revision Revisions of this file
MQTTAsync.h Show annotated file Show diff for this revision Revisions of this file
MQTTClient.h Show annotated file Show diff for this revision Revisions of this file
MQTTEthernet.h Show annotated file Show diff for this revision Revisions of this file
MQTTLogging.h Show annotated file Show diff for this revision Revisions of this file
MQTTPacket.lib Show annotated file Show diff for this revision Revisions of this file
MQTTSocket.h Show annotated file Show diff for this revision Revisions of this file
MQTTmbed.h Show annotated file Show diff for this revision Revisions of this file
diff -r 000000000000 -r 818366b014c9 FP.lib
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/FP.lib	Tue Dec 12 15:58:09 2017 +0000
@@ -0,0 +1,1 @@
+http://mbed.org/users/sam_grove/code/FP/#3c62ba1807ac
diff -r 000000000000 -r 818366b014c9 MQTTAsync.h
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTTAsync.h	Tue Dec 12 15:58:09 2017 +0000
@@ -0,0 +1,607 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#if !defined(MQTTASYNC_H)
+#define MQTTASYNC_H
+
+#include "FP.h"
+#include "MQTTPacket.h"
+#include "stdio.h"
+
+namespace MQTT
+{
+
+
+enum QoS { QOS0, QOS1, QOS2 };
+
+
+struct Message
+{
+    enum QoS qos;
+    bool retained;
+    bool dup;
+    unsigned short id;
+    void *payload;
+    size_t payloadlen;
+};
+
+
+class PacketId
+{
+public:
+    PacketId();
+    
+    int getNext();
+   
+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
+	int MAX_CONCURRENT_OPERATIONS;  // each command which runs concurrently can have a result handler, when we are in multi-threaded mode
+	int command_timeout_ms;
+		
+	limits()
+	{
+		MAX_MQTT_PACKET_SIZE = 100;
+		MAX_MESSAGE_HANDLERS = 5;
+		MAX_CONCURRENT_OPERATIONS = 1; // 1 indicates single-threaded mode - set to >1 for multithreaded mode
+		command_timeout_ms = 30000;
+	}
+} Limits;
+  
+
+/**
+ * @class Async
+ * @brief non-blocking, threaded MQTT client API
+ * @param Network a network class which supports send, receive
+ * @param Timer a timer class with the methods: 
+ */ 
+template<class Network, class Timer, class Thread, class Mutex> class Async
+{
+    
+public:    
+
+	struct Result
+	{
+    	/* success or failure result data */
+    	Async<Network, Timer, Thread, Mutex>* client;
+		int rc;
+	};
+
+	typedef void (*resultHandler)(Result*);	
+   
+    Async(Network* network, const Limits limits = Limits()); 
+        
+    typedef struct
+    {
+        Async* client;
+        Network* network;
+    } connectionLostInfo;
+    
+    typedef int (*connectionLostHandlers)(connectionLostInfo*);
+    
+    /** 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
+     */
+    void setDefaultMessageHandler(messageHandler mh)
+    {
+        defaultMessageHandler.attach(mh);
+    }
+           
+    int connect(resultHandler fn, MQTTPacket_connectData* options = 0);
+    
+     template<class T>
+    int connect(void(T::*method)(Result *), MQTTPacket_connectData* options = 0, T *item = 0);  // alternative to pass in pointer to member function
+        
+    int publish(resultHandler rh, const char* topic, Message* message);
+    
+    int subscribe(resultHandler rh, const char* topicFilter, enum QoS qos, messageHandler mh);
+    
+    int unsubscribe(resultHandler rh, const char* topicFilter);
+    
+    int disconnect(resultHandler rh);
+    
+private:
+
+    void run(void const *argument);
+    int cycle(int timeout);
+    int waitfor(int packet_type, Timer& atimer);
+	int keepalive();
+	int findFreeOperation();
+
+    int decodePacket(int* value, int timeout);
+    int readPacket(int timeout);
+    int sendPacket(int length, int timeout);
+	int deliverMessage(MQTTString* topic, Message* message);
+    
+    Thread* thread;
+    Network* ipstack;
+    
+    Limits limits;
+    
+    char* buf;  
+    char* readbuf;
+
+    Timer ping_timer, connect_timer;
+    unsigned int keepAliveInterval;
+	bool ping_outstanding;
+    
+    PacketId packetid;
+    
+    typedef FP<void, Result*> resultHandlerFP;    
+    resultHandlerFP connectHandler; 
+    
+    typedef FP<void, Message*> messageHandlerFP;
+    struct MessageHandlers
+    {
+    	const char* topic;
+    	messageHandlerFP fp;
+    } *messageHandlers;      // Message handlers are indexed by subscription topic
+    
+    // how many concurrent operations should we allow?  Each one will require a function pointer
+    struct Operations
+    {
+    	unsigned short id;
+    	resultHandlerFP fp;
+    	const char* topic;         // if this is a publish, store topic name in case republishing is required
+    	Message* message;    // for publish, 
+    	Timer timer;         // to check if the command has timed out
+    } *operations;           // result handlers are indexed by packet ids
+
+	static void threadfn(void* arg);
+	
+	messageHandlerFP defaultMessageHandler;
+    
+    typedef FP<int, connectionLostInfo*> connectionLostFP;
+    
+    connectionLostFP connectionLostHandler;
+    
+};
+
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> void MQTT::Async<Network, Timer, Thread, Mutex>::threadfn(void* arg)
+{
+    ((Async<Network, Timer, Thread, Mutex>*) arg)->run(NULL);
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> MQTT::Async<Network, Timer, Thread, Mutex>::Async(Network* network, Limits limits)  : limits(limits), packetid()
+{
+	this->thread = 0;
+	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->operations = new struct Operations[limits.MAX_CONCURRENT_OPERATIONS];
+	for (int i = 0; i < limits.MAX_CONCURRENT_OPERATIONS; ++i)
+		operations[i].id = 0;
+	this->messageHandlers = new struct MessageHandlers[limits.MAX_MESSAGE_HANDLERS];
+	for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i)
+		messageHandlers[i].topic = 0;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::sendPacket(int length, int timeout)
+{
+    int sent = 0;
+    
+    while (sent < length)
+        sent += ipstack->write(&buf[sent], length, timeout);
+	if (sent == length)
+	    ping_timer.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet    
+    return sent;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::decodePacket(int* value, int timeout)
+{
+    char c;
+    int multiplier = 1;
+    int len = 0;
+	const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4;
+
+    *value = 0;
+    do
+    {
+        int rc = MQTTPACKET_READ_ERROR;
+
+        if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
+        {
+            rc = MQTTPACKET_READ_ERROR; /* bad data */
+            goto exit;
+        }
+        rc = ipstack->read(&c, 1, timeout);
+        if (rc != 1)
+            goto exit;
+        *value += (c & 127) * multiplier;
+        multiplier *= 128;
+    } while ((c & 128) != 0);
+exit:
+    return len;
+}
+
+
+/**
+ * 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, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::readPacket(int timeout) 
+{
+    int rc = -1;
+    MQTTHeader header = {0};
+    int len = 0;
+    int rem_len = 0;
+
+    /* 1. read the header byte.  This has the packet type in it */
+    if (ipstack->read(readbuf, 1, timeout) != 1)
+        goto exit;
+
+    len = 1;
+    /* 2. read the remaining length.  This is variable in itself */
+    decodePacket(&rem_len, timeout);
+    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, timeout) != rem_len)
+        goto exit;
+
+    header.byte = readbuf[0];
+    rc = header.bits.type;
+exit:
+    return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::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)
+	{
+		if (messageHandlers[i].topic != 0 && MQTTPacket_equals(topic, (char*)messageHandlers[i].topic))
+		{
+			messageHandlers[i].fp(message);
+			rc = 0;
+			break;
+		}
+	}
+	
+	return rc;
+}
+
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::cycle(int timeout)
+{
+    /* get one piece of work off the wire and one pass through */
+
+    // read the socket, see what work is due
+    int packet_type = readPacket(timeout);
+    
+	int len, rc;
+    switch (packet_type)
+    {
+        case CONNACK:
+			if (this->thread)
+			{
+				Result res = {this, 0};
+            	if (MQTTDeserialize_connack(&res.rc, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+                	;
+				connectHandler(&res);
+				connectHandler.detach(); // only invoke the callback once
+			}
+			break;
+        case PUBACK:
+        	if (this->thread)
+        		; //call resultHandler
+        case SUBACK:
+            break;
+        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, limits.MAX_MQTT_PACKET_SIZE);;
+			if (msg.qos == QOS0)
+				deliverMessage(&topicName, &msg);
+            break;
+        case PUBREC:
+   	        int type, dup, mypacketid;
+   	        if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+   	            ; 
+   	        // must lock this access against the application thread, if we are multi-threaded
+			len = MQTTSerialize_ack(buf, limits.MAX_MQTT_PACKET_SIZE, PUBREL, 0, mypacketid);
+		    rc = sendPacket(len, timeout); // send the PUBREL packet
+			if (rc != len) 
+				goto exit; // there was a problem
+
+            break;
+        case PUBCOMP:
+            break;
+        case PINGRESP:
+			ping_outstanding = false;
+            break;
+    }
+	keepalive();
+exit:
+    return packet_type;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::keepalive()
+{
+	int rc = 0;
+
+	if (keepAliveInterval == 0)
+		goto exit;
+
+	if (ping_timer.expired())
+	{
+		if (ping_outstanding)
+			rc = -1;
+		else
+		{
+			int len = MQTTSerialize_pingreq(buf, limits.MAX_MQTT_PACKET_SIZE);
+			rc = sendPacket(len, 1000); // send the ping packet
+			if (rc != len) 
+				rc = -1; // indicate there's a problem
+			else
+				ping_outstanding = true;
+		}
+	}
+
+exit:
+	return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> void MQTT::Async<Network, Timer, Thread, Mutex>::run(void const *argument)
+{
+	while (true)
+		cycle(ping_timer.left_ms());
+}
+
+
+// only used in single-threaded mode where one command at a time is in process
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::waitfor(int packet_type, Timer& atimer)
+{
+	int rc = -1;
+	
+	do
+    {
+		if (atimer.expired()) 
+			break; // we timed out
+	}
+	while ((rc = cycle(atimer.left_ms())) != packet_type);	
+	
+	return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::connect(resultHandler resultHandler, MQTTPacket_connectData* options)
+{
+	connect_timer.countdown(limits.command_timeout_ms);
+
+    MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;
+    if (options == 0)
+        options = &default_options; // set default options if none were supplied
+    
+    this->keepAliveInterval = options->keepAliveInterval;
+	ping_timer.countdown(this->keepAliveInterval);
+    int len = MQTTSerialize_connect(buf, limits.MAX_MQTT_PACKET_SIZE, options);
+    int rc = sendPacket(len, connect_timer.left_ms()); // send the connect packet
+	if (rc != len) 
+		goto exit; // there was a problem
+    
+    if (resultHandler == 0)     // wait until the connack is received 
+    {
+        // 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)
+	        	rc = connack_rc;
+	    }
+    }
+    else
+    {
+        // set connect response callback function
+        connectHandler.attach(resultHandler);
+        
+        // start background thread            
+        this->thread = new Thread((void (*)(void const *argument))&MQTT::Async<Network, Timer, Thread, Mutex>::threadfn, (void*)this);
+    }
+    
+exit:
+    return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::findFreeOperation()
+{
+	int found = -1;
+	for (int i = 0; i < limits.MAX_CONCURRENT_OPERATIONS; ++i)
+	{
+		if (operations[i].id == 0)
+		{
+			found = i;
+			break;
+		}
+	}
+	return found;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::subscribe(resultHandler resultHandler, const char* topicFilter, enum QoS qos, messageHandler messageHandler)
+{
+	int index = 0;
+	if (this->thread)
+		index = findFreeOperation();	
+	Timer& atimer = operations[index].timer;
+	
+	atimer.countdown(limits.command_timeout_ms);
+    MQTTString topic = {(char*)topicFilter, 0, 0};
+    
+    int len = MQTTSerialize_subscribe(buf, limits.MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos);
+    int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet
+	if (rc != len) 
+		goto exit; // there was a problem
+    
+    /* wait for suback */
+    if (resultHandler == 0)
+    {
+        // this will block
+        if (waitfor(SUBACK, atimer) == SUBACK)
+        {
+            int count = 0, grantedQoS = -1, mypacketid;
+            if (MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, readbuf, limits.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)
+				{
+					if (messageHandlers[i].topic == 0)
+					{
+						messageHandlers[i].topic = topicFilter;
+						messageHandlers[i].fp.attach(messageHandler);
+						rc = 0;
+						break;
+					}
+				}
+            }
+        }
+    }
+    else
+    {
+        // set subscribe response callback function
+        
+    }
+    
+exit:
+    return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::unsubscribe(resultHandler resultHandler, const char* topicFilter)
+{
+	int index = 0;
+	if (this->thread)
+		index = findFreeOperation();	
+	Timer& atimer = operations[index].timer;
+
+	atimer.countdown(limits.command_timeout_ms);
+    MQTTString topic = {(char*)topicFilter, 0, 0};
+    
+    int len = MQTTSerialize_unsubscribe(buf, limits.MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic);
+    int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet
+	if (rc != len) 
+		goto exit; // there was a problem
+    
+    // set unsubscribe response callback function
+        
+    
+exit:
+    return rc;
+}
+
+
+   
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::publish(resultHandler resultHandler, const char* topicName, Message* message)
+{
+	int index = 0;
+	if (this->thread)
+		index = findFreeOperation();	
+	Timer& atimer = operations[index].timer;
+
+	atimer.countdown(limits.command_timeout_ms);
+    MQTTString topic = {(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, topic, (char*)message->payload, message->payloadlen);
+    int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet
+	if (rc != len) 
+		goto exit; // there was a problem
+    
+    /* wait for acks */
+    if (resultHandler == 0)
+    {
+ 		if (message->qos == QOS1)
+		{
+	        if (waitfor(PUBACK, atimer) == PUBACK)
+    	    {
+    	        int type, dup, mypacketid;
+    	        if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+    	            rc = 0; 
+    	    }
+		}
+		else if (message->qos == QOS2)
+		{
+	        if (waitfor(PUBCOMP, atimer) == PUBCOMP)
+	   	    {
+	   	    	int type, dup, mypacketid;
+            	if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+    	           	rc = 0; 
+			}
+
+		}
+    }
+    else
+    {
+        // set publish response callback function
+        
+    }
+    
+exit:
+    return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::disconnect(resultHandler resultHandler)
+{  
+    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);
+    int rc = sendPacket(len, timer.left_ms());   // send the disconnect packet
+    
+    return (rc == len) ? 0 : -1;
+}
+
+
+
+#endif
\ No newline at end of file
diff -r 000000000000 -r 818366b014c9 MQTTClient.h
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTTClient.h	Tue Dec 12 15:58:09 2017 +0000
@@ -0,0 +1,1052 @@
+/*******************************************************************************
+ * Copyright (c) 2014, 2017 IBM Corp.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Ian Craggs - fix for bug 458512 - QoS 2 messages
+ *    Ian Craggs - fix for bug 460389 - send loop uses wrong length
+ *    Ian Craggs - fix for bug 464169 - clearing subscriptions
+ *    Ian Craggs - fix for bug 464551 - enums and ints can be different size
+ *    Mark Sonnentag - fix for bug 475204 - inefficient instantiation of Timer
+ *    Ian Craggs - fix for bug 475749 - packetid modified twice
+ *    Ian Craggs - add ability to set message handler separately #6
+ *******************************************************************************/
+
+#if !defined(MQTTCLIENT_H)
+#define MQTTCLIENT_H
+
+#include "FP.h"
+#include "MQTTPacket.h"
+#include <stdio.h>
+#include "MQTTLogging.h"
+
+#if !defined(MQTTCLIENT_QOS1)
+    #define MQTTCLIENT_QOS1 1
+#endif
+#if !defined(MQTTCLIENT_QOS2)
+    #define MQTTCLIENT_QOS2 0
+#endif
+
+namespace MQTT
+{
+
+
+enum QoS { QOS0, QOS1, QOS2 };
+
+// all failure return codes must be negative
+enum returnCode { BUFFER_OVERFLOW = -2, FAILURE = -1, SUCCESS = 0 };
+
+
+struct Message
+{
+    enum QoS qos;
+    bool retained;
+    bool dup;
+    unsigned short id;
+    void *payload;
+    size_t payloadlen;
+};
+
+
+struct MessageData
+{
+    MessageData(MQTTString &aTopicName, struct Message &aMessage)  : message(aMessage), topicName(aTopicName)
+    { }
+
+    struct Message &message;
+    MQTTString &topicName;
+};
+
+
+struct connackData
+{
+    int rc;
+    bool sessionPresent;
+};
+
+
+struct subackData
+{
+    int grantedQoS;
+};
+
+
+class PacketId
+{
+public:
+    PacketId()
+    {
+        next = 0;
+    }
+
+    int getNext()
+    {
+        return next = (next == MAX_PACKET_ID) ? 1 : next + 1;
+    }
+
+private:
+    static const int MAX_PACKET_ID = 65535;
+    int next;
+};
+
+
+/**
+ * @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, int MAX_MQTT_PACKET_SIZE = 100, int MAX_MESSAGE_HANDLERS = 5>
+class Client
+{
+
+public:
+
+    typedef void (*messageHandler)(MessageData&);
+
+    /** 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 default message handling callback - used for any message which does not match a subscription message handler
+     *  @param mh - pointer to the callback function.  Set to 0 to remove.
+     */
+    void setDefaultMessageHandler(messageHandler mh)
+    {
+        if (mh != 0)
+            defaultMessageHandler.attach(mh);
+        else
+            defaultMessageHandler.detach();
+    }
+
+    /** Set a message handling callback.  This can be used outside of the the subscribe method.
+     *  @param topicFilter - a topic pattern which can include wildcards
+     *  @param mh - pointer to the callback function. If 0, removes the callback if any
+     */
+    int setMessageHandler(const char* topicFilter, messageHandler mh);
+
+    /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
+     *  The nework object must be connected to the network endpoint before calling this
+     *  Default connect options are used
+     *  @return success code -
+     */
+    int connect();
+
+    /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
+     *  The nework object must be connected to the network endpoint before calling this
+     *  @param options - connect options
+     *  @return success code -
+     */
+    int connect(MQTTPacket_connectData& options);
+
+    /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
+     *  The nework object must be connected to the network endpoint before calling this
+     *  @param options - connect options
+     *  @param connackData - connack data to be returned
+     *  @return success code -
+     */
+    int connect(MQTTPacket_connectData& options, connackData& data);
+
+    /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
+     *  @param topic - the topic to publish to
+     *  @param message - the message to send
+     *  @return success code -
+     */
+    int publish(const char* topicName, Message& message);
+
+    /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
+     *  @param topic - the topic to publish to
+     *  @param payload - the data to send
+     *  @param payloadlen - the length of the data
+     *  @param qos - the QoS to send the publish at
+     *  @param retained - whether the message should be retained
+     *  @return success code -
+     */
+    int publish(const char* topicName, void* payload, size_t payloadlen, enum QoS qos = QOS0, bool retained = false);
+
+    /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
+     *  @param topic - the topic to publish to
+     *  @param payload - the data to send
+     *  @param payloadlen - the length of the data
+     *  @param id - the packet id used - returned
+     *  @param qos - the QoS to send the publish at
+     *  @param retained - whether the message should be retained
+     *  @return success code -
+     */
+    int publish(const char* topicName, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos = QOS1, bool retained = false);
+
+    /** MQTT Subscribe - send an MQTT subscribe packet and wait for the suback
+     *  @param topicFilter - a topic pattern which can include wildcards
+     *  @param qos - the MQTT QoS to subscribe at
+     *  @param mh - the callback function to be invoked when a message is received for this subscription
+     *  @return success code -
+     */
+    int subscribe(const char* topicFilter, enum QoS qos, messageHandler mh);
+
+    /** MQTT Subscribe - send an MQTT subscribe packet and wait for the suback
+     *  @param topicFilter - a topic pattern which can include wildcards
+     *  @param qos - the MQTT QoS to subscribe at©
+     *  @param mh - the callback function to be invoked when a message is received for this subscription
+     *  @param
+     *  @return success code -
+     */
+    int subscribe(const char* topicFilter, enum QoS qos, messageHandler mh, subackData &data);
+
+    /** MQTT Unsubscribe - send an MQTT unsubscribe packet and wait for the unsuback
+     *  @param topicFilter - a topic pattern which can include wildcards
+     *  @return success code -
+     */
+    int unsubscribe(const char* topicFilter);
+
+    /** MQTT Disconnect - send an MQTT disconnect packet, and clean up any state
+     *  @return success code -
+     */
+    int disconnect();
+
+    /** 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
+     *  @return success code - on failure, this means the client has disconnected
+     */
+    int yield(unsigned long timeout_ms = 1000L);
+
+    /** Is the client connected?
+     *  @return flag - is the client connected or not?
+     */
+    bool isConnected()
+    {
+        return isconnected;
+    }
+
+private:
+
+    void closeSession();
+    void cleanSession();
+    int cycle(Timer& timer);
+    int waitfor(int packet_type, Timer& timer);
+    int keepalive();
+    int publish(int len, Timer& timer, enum QoS qos);
+
+    int decodePacket(int* value, int timeout);
+    int readPacket(Timer& timer);
+    int sendPacket(int length, Timer& timer);
+    int deliverMessage(MQTTString& topicName, Message& message);
+    bool isTopicMatched(char* topicFilter, MQTTString& topicName);
+
+    Network& ipstack;
+    unsigned long command_timeout_ms;
+
+    unsigned char sendbuf[MAX_MQTT_PACKET_SIZE];
+    unsigned char readbuf[MAX_MQTT_PACKET_SIZE];
+
+    Timer last_sent, last_received;
+    unsigned int keepAliveInterval;
+    bool ping_outstanding;
+    bool cleansession;
+
+    PacketId packetid;
+
+    struct MessageHandlers
+    {
+        const char* topicFilter;
+        FP<void, MessageData&> fp;
+    } messageHandlers[MAX_MESSAGE_HANDLERS];      // Message handlers are indexed by subscription topic
+
+    FP<void, MessageData&> defaultMessageHandler;
+
+    bool isconnected;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+    unsigned char pubbuf[MAX_MQTT_PACKET_SIZE];  // store the last publish for sending on reconnect
+    int inflightLen;
+    unsigned short inflightMsgid;
+    enum QoS inflightQoS;
+#endif
+
+#if MQTTCLIENT_QOS2
+    bool pubrel;
+    #if !defined(MAX_INCOMING_QOS2_MESSAGES)
+        #define MAX_INCOMING_QOS2_MESSAGES 10
+    #endif
+    unsigned short incomingQoS2messages[MAX_INCOMING_QOS2_MESSAGES];
+    bool isQoS2msgidFree(unsigned short id);
+    bool useQoS2msgid(unsigned short id);
+    void freeQoS2msgid(unsigned short id);
+#endif
+
+};
+
+}
+
+
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
+void MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::cleanSession()
+{
+    for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+        messageHandlers[i].topicFilter = 0;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+    inflightMsgid = 0;
+    inflightQoS = QOS0;
+#endif
+
+#if MQTTCLIENT_QOS2
+    pubrel = false;
+    for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+        incomingQoS2messages[i] = 0;
+#endif
+}
+
+
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
+void MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::closeSession()
+{
+    ping_outstanding = false;
+    isconnected = false;
+    if (cleansession)
+        cleanSession();
+}
+
+
+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->command_timeout_ms = command_timeout_ms;
+    cleansession = true;
+      closeSession();
+}
+
+
+#if MQTTCLIENT_QOS2
+template<class Network, class Timer, int a, int b>
+bool MQTT::Client<Network, Timer, a, b>::isQoS2msgidFree(unsigned short id)
+{
+    for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+    {
+        if (incomingQoS2messages[i] == id)
+            return false;
+    }
+    return true;
+}
+
+
+template<class Network, class Timer, int a, int b>
+bool MQTT::Client<Network, Timer, a, b>::useQoS2msgid(unsigned short id)
+{
+    for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+    {
+        if (incomingQoS2messages[i] == 0)
+        {
+            incomingQoS2messages[i] = id;
+            return true;
+        }
+    }
+    return false;
+}
+
+
+template<class Network, class Timer, int a, int b>
+void MQTT::Client<Network, Timer, a, b>::freeQoS2msgid(unsigned short id)
+{
+    for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+    {
+        if (incomingQoS2messages[i] == id)
+        {
+            incomingQoS2messages[i] = 0;
+            return;
+        }
+    }
+}
+#endif
+
+
+template<class Network, class Timer, int a, int b>
+int MQTT::Client<Network, Timer, a, b>::sendPacket(int length, Timer& timer)
+{
+    int rc = FAILURE,
+        sent = 0;
+
+    while (sent < length)
+    {
+        rc = ipstack.write(&sendbuf[sent], length - sent, timer.left_ms());
+        if (rc < 0)  // there was an error writing the data
+            break;
+        sent += rc;
+        if (timer.expired()) // only check expiry after at least one attempt to write
+            break;
+    }
+    if (sent == length)
+    {
+        if (this->keepAliveInterval > 0)
+            last_sent.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet
+        rc = SUCCESS;
+    }
+    else
+        rc = FAILURE;
+
+#if defined(MQTT_DEBUG)
+    char printbuf[150];
+    DEBUG("Rc %d from sending packet %s\r\n", rc, 
+        MQTTFormat_toServerString(printbuf, sizeof(printbuf), sendbuf, length));
+#endif
+    return rc;
+}
+
+
+template<class Network, class Timer, int a, int b>
+int MQTT::Client<Network, Timer, a, b>::decodePacket(int* value, int timeout)
+{
+    unsigned char c;
+    int multiplier = 1;
+    int len = 0;
+    const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4;
+
+    *value = 0;
+    do
+    {
+        int rc = MQTTPACKET_READ_ERROR;
+
+        if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
+        {
+            rc = MQTTPACKET_READ_ERROR; /* bad data */
+            goto exit;
+        }
+        rc = ipstack.read(&c, 1, timeout);
+        if (rc != 1)
+            goto exit;
+        *value += (c & 127) * multiplier;
+        multiplier *= 128;
+    } while ((c & 128) != 0);
+exit:
+    return len;
+}
+
+
+/**
+ * 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, 0 if none, -1 if error
+ */
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::readPacket(Timer& timer)
+{
+    int rc = FAILURE;
+    MQTTHeader header = {0};
+    int len = 0;
+    int rem_len = 0;
+
+    /* 1. read the header byte.  This has the packet type in it */
+    rc = ipstack.read(readbuf, 1, timer.left_ms());
+    if (rc != 1)
+        goto exit;
+
+    len = 1;
+    /* 2. read the remaining length.  This is variable in itself */
+    decodePacket(&rem_len, timer.left_ms());
+    len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length into the buffer */
+
+    if (rem_len > (MAX_MQTT_PACKET_SIZE - len))
+    {
+        rc = BUFFER_OVERFLOW;
+        goto exit;
+    }
+
+    /* 3. read the rest of the buffer using a callback to supply the rest of the data */
+    if (rem_len > 0 && (ipstack.read(readbuf + len, rem_len, timer.left_ms()) != rem_len))
+        goto exit;
+
+    header.byte = readbuf[0];
+    rc = header.bits.type;
+    if (this->keepAliveInterval > 0)
+        last_received.countdown(this->keepAliveInterval); // record the fact that we have successfully received a packet
+exit:
+
+#if defined(MQTT_DEBUG)
+    if (rc >= 0)
+    {
+        char printbuf[50];
+        DEBUG("Rc %d receiving packet %s\r\n", rc, 
+            MQTTFormat_toClientString(printbuf, sizeof(printbuf), readbuf, len));
+    }
+#endif
+    return rc;
+}
+
+
+// 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& topicName, Message& message)
+{
+    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].topicFilter != 0 && (MQTTPacket_equals(&topicName, (char*)messageHandlers[i].topicFilter) ||
+                isTopicMatched((char*)messageHandlers[i].topicFilter, topicName)))
+        {
+            if (messageHandlers[i].fp.attached())
+            {
+                MessageData md(topicName, message);
+                messageHandlers[i].fp(md);
+                rc = SUCCESS;
+            }
+        }
+    }
+
+    if (rc == FAILURE && defaultMessageHandler.attached())
+    {
+        MessageData md(topicName, message);
+        defaultMessageHandler(md);
+        rc = SUCCESS;
+    }
+
+    return rc;
+}
+
+
+
+template<class Network, class Timer, int a, int b>
+int MQTT::Client<Network, Timer, a, b>::yield(unsigned long timeout_ms)
+{
+    int rc = SUCCESS;
+    Timer timer;
+
+    timer.countdown_ms(timeout_ms);
+    while (!timer.expired())
+    {
+        if (cycle(timer) < 0)
+        {
+            rc = FAILURE;
+            break;
+        }
+    }
+
+    return rc;
+}
+
+
+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
+    int len = 0,
+        rc = SUCCESS;
+
+    int packet_type = readPacket(timer);    // read the socket, see what work is due
+
+    switch (packet_type)
+    {
+        default:
+            // no more data to read, unrecoverable. Or read packet fails due to unexpected network error
+            rc = packet_type;
+            goto exit;
+        case NSAPI_ERROR_WOULD_BLOCK:
+        case NSAPI_ERROR_OK: // timed out reading packet            
+            break;
+        case CONNACK:
+        case PUBACK:
+        case SUBACK:
+            break;
+        case PUBLISH:
+        {
+            MQTTString topicName = MQTTString_initializer;
+            Message msg;
+            int intQoS;
+            msg.payloadlen = 0; /* this is a size_t, but deserialize publish sets this as int */
+            if (MQTTDeserialize_publish((unsigned char*)&msg.dup, &intQoS, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicName,
+                                 (unsigned char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+                goto exit;
+            msg.qos = (enum QoS)intQoS;
+#if MQTTCLIENT_QOS2
+            if (msg.qos != QOS2)
+#endif
+                deliverMessage(topicName, msg);
+#if MQTTCLIENT_QOS2
+            else if (isQoS2msgidFree(msg.id))
+            {
+                if (useQoS2msgid(msg.id))
+                    deliverMessage(topicName, msg);
+                else
+                    WARN("Maximum number of incoming QoS2 messages exceeded");
+            }
+#endif
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+            if (msg.qos != QOS0)
+            {
+                if (msg.qos == QOS1)
+                    len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBACK, 0, msg.id);
+                else if (msg.qos == QOS2)
+                    len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREC, 0, msg.id);
+                if (len <= 0)
+                    rc = FAILURE;
+                else
+                    rc = sendPacket(len, timer);
+                if (rc == FAILURE)
+                    goto exit; // there was a problem
+            }
+            break;
+#endif
+        }
+#if MQTTCLIENT_QOS2
+        case PUBREC:
+        case PUBREL:
+            unsigned short mypacketid;
+            unsigned char dup, type;
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+                rc = FAILURE;
+            else if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE,
+                                 (packet_type == PUBREC) ? PUBREL : PUBCOMP, 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
+            if (packet_type == PUBREL)
+                freeQoS2msgid(mypacketid);
+            break;
+
+        case PUBCOMP:
+            break;
+#endif
+        case PINGRESP:
+            ping_outstanding = false;
+            break;
+    }
+
+    if (keepalive() != SUCCESS)
+        //check only keepalive FAILURE status so that previous FAILURE status can be considered as FAULT
+        rc = FAILURE;
+
+exit:
+    if (rc == SUCCESS)
+        rc = packet_type;
+    else if (isconnected)
+        closeSession();
+    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 = SUCCESS;
+    static Timer ping_sent;
+
+    if (keepAliveInterval == 0)
+        goto exit;
+    
+    if (ping_outstanding)
+    {
+        if (ping_sent.expired())
+        {
+            rc = FAILURE; // session failure
+            #if defined(MQTT_DEBUG)
+                DEBUG("PINGRESP not received in keepalive interval\r\n");
+            #endif
+        }
+    }
+    else if (last_sent.expired() || last_received.expired())
+    {
+        Timer timer(1000);
+        int len = MQTTSerialize_pingreq(sendbuf, MAX_MQTT_PACKET_SIZE);
+        if (len > 0 && (rc = sendPacket(len, timer)) == SUCCESS) // send the ping packet
+        {
+            ping_outstanding = true;
+            ping_sent.countdown(this->keepAliveInterval);
+        }
+    }
+exit:
+    return rc;
+}
+
+
+// only used in single-threaded mode where one command at a time is in process
+template<class Network, class Timer, int a, int b>
+int MQTT::Client<Network, Timer, a, b>::waitfor(int packet_type, Timer& timer)
+{
+    int rc = FAILURE;
+
+    do
+    {
+        if (timer.expired())
+            break; // we timed out
+        rc = cycle(timer);
+    }
+    while (rc != packet_type && rc >= 0);
+
+    return rc;
+}
+
+
+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, connackData& data)
+{
+    Timer connect_timer(command_timeout_ms);
+    int rc = FAILURE;
+    int len = 0;
+
+    if (isconnected) // don't send connect packet again if we are already connected
+        goto exit;
+
+    this->keepAliveInterval = options.keepAliveInterval;
+    this->cleansession = options.cleansession;
+    if ((len = MQTTSerialize_connect(sendbuf, MAX_MQTT_PACKET_SIZE, &options)) <= 0)
+        goto exit;
+    if ((rc = sendPacket(len, connect_timer)) != SUCCESS)  // send the connect packet
+        goto exit; // there was a problem
+
+    if (this->keepAliveInterval > 0)
+        last_received.countdown(this->keepAliveInterval);
+    // this will be a blocking call, wait for the connack
+    if (waitfor(CONNACK, connect_timer) == CONNACK)
+    {
+        data.rc = 0;
+        data.sessionPresent = false;
+        if (MQTTDeserialize_connack((unsigned char*)&data.sessionPresent,
+                            (unsigned char*)&data.rc, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
+            rc = data.rc;
+        else
+            rc = FAILURE;
+    }
+    else
+        rc = FAILURE;
+
+#if MQTTCLIENT_QOS2
+    // resend any inflight publish
+    if (inflightMsgid > 0 && inflightQoS == QOS2 && pubrel)
+    {
+        if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, inflightMsgid)) <= 0)
+            rc = FAILURE;
+        else
+            rc = publish(len, connect_timer, inflightQoS);
+    }
+    else
+#endif
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+    if (inflightMsgid > 0)
+    {
+        memcpy(sendbuf, pubbuf, MAX_MQTT_PACKET_SIZE);
+        rc = publish(inflightLen, connect_timer, inflightQoS);
+    }
+#endif
+
+exit:
+    if (rc == SUCCESS)
+    {
+        isconnected = true;
+        ping_outstanding = false;
+    }
+    return rc;
+}
+
+
+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)
+{
+    connackData data;
+    return connect(options, data);
+}
+
+
+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 default_options = MQTTPacket_connectData_initializer;
+    return connect(default_options);
+}
+
+
+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>::setMessageHandler(const char* topicFilter, messageHandler messageHandler)
+{
+    int rc = FAILURE;
+    int i = -1;
+
+    // first check for an existing matching slot
+    for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+    {
+        if (messageHandlers[i].topicFilter != 0 && strcmp(messageHandlers[i].topicFilter, topicFilter) == 0)
+        {
+            if (messageHandler == 0) // remove existing
+            {
+                messageHandlers[i].topicFilter = 0;
+                messageHandlers[i].fp.detach();
+            }
+            rc = SUCCESS; // return i when adding new subscription
+            break;
+        }
+    }
+    // if no existing, look for empty slot (unless we are removing)
+    if (messageHandler != 0) {
+        if (rc == FAILURE)
+        {
+            for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+            {
+                if (messageHandlers[i].topicFilter == 0)
+                {
+                    rc = SUCCESS;
+                    break;
+                }
+            }
+        }
+        if (i < MAX_MESSAGE_HANDLERS)
+        {
+            messageHandlers[i].topicFilter = topicFilter;
+            messageHandlers[i].fp.attach(messageHandler);
+        }
+    }
+    return rc;
+}
+
+
+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, subackData& data)
+{
+    int rc = FAILURE;
+    Timer timer(command_timeout_ms);
+    int len = 0;
+    MQTTString topic = {(char*)topicFilter, {0, 0}};
+
+    if (!isconnected)
+        goto exit;
+
+    len = MQTTSerialize_subscribe(sendbuf, 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
+
+    if (waitfor(SUBACK, timer) == SUBACK)      // wait for suback
+    {
+        int count = 0;
+        unsigned short mypacketid;
+        data.grantedQoS = 0;
+        if (MQTTDeserialize_suback(&mypacketid, 1, &count, &data.grantedQoS, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
+        {
+            if (data.grantedQoS != 0x80)
+                rc = setMessageHandler(topicFilter, messageHandler);
+        }
+    }
+    else
+        rc = FAILURE;
+
+exit:
+    if (rc == FAILURE)
+        closeSession();
+    return rc;
+}
+
+
+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)
+{
+    subackData data;
+    return subscribe(topicFilter, qos, messageHandler, data);
+}
+
+
+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 rc = FAILURE;
+    Timer timer(command_timeout_ms);
+    MQTTString topic = {(char*)topicFilter, {0, 0}};
+    int len = 0;
+
+    if (!isconnected)
+        goto exit;
+
+    if ((len = MQTTSerialize_unsubscribe(sendbuf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic)) <= 0)
+        goto exit;
+    if ((rc = sendPacket(len, timer)) != SUCCESS) // send the unsubscribe packet
+        goto exit; // there was a problem
+
+    if (waitfor(UNSUBACK, timer) == UNSUBACK)
+    {
+        unsigned short mypacketid;  // should be the same as the packetid above
+        if (MQTTDeserialize_unsuback(&mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
+        {
+            // remove the subscription message handler associated with this topic, if there is one
+            setMessageHandler(topicFilter, 0);
+        }
+    }
+    else
+        rc = FAILURE;
+
+exit:
+    if (rc != SUCCESS)
+        closeSession();
+    return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(int len, Timer& timer, enum QoS qos)
+{
+    int rc;
+
+    if ((rc = sendPacket(len, timer)) != SUCCESS) // send the publish packet
+        goto exit; // there was a problem
+
+#if MQTTCLIENT_QOS1
+    if (qos == QOS1)
+    {
+        if (waitfor(PUBACK, timer) == PUBACK)
+        {
+            unsigned short mypacketid;
+            unsigned char dup, type;
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+                rc = FAILURE;
+            else if (inflightMsgid == mypacketid)
+                inflightMsgid = 0;
+        }
+        else
+            rc = FAILURE;
+    }
+#endif
+#if MQTTCLIENT_QOS2
+    else if (qos == QOS2)
+    {
+        if (waitfor(PUBCOMP, timer) == PUBCOMP)
+        {
+            unsigned short mypacketid;
+            unsigned char dup, type;
+            if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+                rc = FAILURE;
+            else if (inflightMsgid == mypacketid)
+                inflightMsgid = 0;
+        }
+        else
+            rc = FAILURE;
+    }
+#endif
+
+exit:
+    if (rc != SUCCESS)
+        closeSession();
+    return rc;
+}
+
+
+
+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, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos, bool retained)
+{
+    int rc = FAILURE;
+    Timer timer(command_timeout_ms);
+    MQTTString topicString = MQTTString_initializer;
+    int len = 0;
+
+    if (!isconnected)
+        goto exit;
+
+    topicString.cstring = (char*)topicName;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+    if (qos == QOS1 || qos == QOS2)
+        id = packetid.getNext();
+#endif
+
+    len = MQTTSerialize_publish(sendbuf, MAX_MQTT_PACKET_SIZE, 0, qos, retained, id,
+              topicString, (unsigned char*)payload, payloadlen);
+    if (len <= 0)
+        goto exit;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+    if (!cleansession)
+    {
+        memcpy(pubbuf, sendbuf, len);
+        inflightMsgid = id;
+        inflightLen = len;
+        inflightQoS = qos;
+#if MQTTCLIENT_QOS2
+        pubrel = false;
+#endif
+    }
+#endif
+
+    rc = publish(len, timer, qos);
+exit:
+    return rc;
+}
+
+
+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, void* payload, size_t payloadlen, enum QoS qos, bool retained)
+{
+    unsigned short id = 0;  // dummy - not used for anything
+    return publish(topicName, payload, payloadlen, id, qos, retained);
+}
+
+
+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)
+{
+    return publish(topicName, message.payload, message.payloadlen, message.qos, message.retained);
+}
+
+
+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(command_timeout_ms);     // we might wait for incomplete incoming publishes to complete
+    int len = MQTTSerialize_disconnect(sendbuf, MAX_MQTT_PACKET_SIZE);
+    if (len > 0)
+        rc = sendPacket(len, timer);            // send the disconnect packet
+    closeSession();
+    return rc;
+}
+
+#endif
diff -r 000000000000 -r 818366b014c9 MQTTEthernet.h
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTTEthernet.h	Tue Dec 12 15:58:09 2017 +0000
@@ -0,0 +1,29 @@
+
+#if !defined(MQTTETHERNET_H)
+#define MQTTETHERNET_H
+
+#include "MQTTmbed.h"
+#include "EthernetInterface.h"
+#include "MQTTSocket.h"
+
+class MQTTEthernet : public MQTTSocket
+{
+public:    
+    MQTTEthernet() : MQTTSocket(&eth)
+    {
+        eth.connect();
+    }
+    
+    EthernetInterface& getEth()
+    {
+        return eth;
+    }
+    
+private:
+
+    EthernetInterface eth;
+    
+};
+
+
+#endif
diff -r 000000000000 -r 818366b014c9 MQTTLogging.h
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTTLogging.h	Tue Dec 12 15:58:09 2017 +0000
@@ -0,0 +1,39 @@
+#if !defined(MQTT_LOGGING_H)
+#define MQTT_LOGGING_H
+
+#define STREAM      stdout
+#if !defined(DEBUG)
+#define DEBUG(...)    \
+    {\
+    fprintf(STREAM, "DEBUG:   %s L#%d ", __PRETTY_FUNCTION__, __LINE__);  \
+    fprintf(STREAM, ##__VA_ARGS__); \
+    fflush(STREAM); \
+    }
+#endif
+#if !defined(LOG)
+#define LOG(...)    \
+    {\
+    fprintf(STREAM, "LOG:   %s L#%d ", __PRETTY_FUNCTION__, __LINE__);  \
+    fprintf(STREAM, ##__VA_ARGS__); \
+    fflush(STREAM); \
+    }
+#endif
+#if !defined(WARN)
+#define WARN(...)   \
+    { \
+    fprintf(STREAM, "WARN:  %s L#%d ", __PRETTY_FUNCTION__, __LINE__);  \
+    fprintf(STREAM, ##__VA_ARGS__); \
+    fflush(STREAM); \
+    }
+#endif 
+#if !defined(ERROR)
+#define ERROR(...)  \
+    { \
+    fprintf(STREAM, "ERROR: %s L#%d ", __PRETTY_FUNCTION__, __LINE__); \
+    fprintf(STREAM, ##__VA_ARGS__); \
+    fflush(STREAM); \
+    exit(1); \
+    }
+#endif
+
+#endif
\ No newline at end of file
diff -r 000000000000 -r 818366b014c9 MQTTPacket.lib
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTTPacket.lib	Tue Dec 12 15:58:09 2017 +0000
@@ -0,0 +1,1 @@
+http://mbed.org/teams/mqtt/code/MQTTPacket/#aedcaf7984d5
diff -r 000000000000 -r 818366b014c9 MQTTSocket.h
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTTSocket.h	Tue Dec 12 15:58:09 2017 +0000
@@ -0,0 +1,97 @@
+#if !defined(MQTTSOCKET_H)
+#define MQTTSOCKET_H
+
+#include "MQTTmbed.h"
+#include <EthernetInterface.h>
+#include <Timer.h>
+
+class MQTTSocket
+{
+public:
+    MQTTSocket(EthernetInterface *anet)
+    {
+        net = anet;
+        open = false;
+    }
+    
+    int connect(char* hostname, int port, int timeout=1000)
+    {
+        if (open)
+            disconnect();
+        nsapi_error_t rc = mysock.open(net);
+        open = true;
+        mysock.set_blocking(true);
+        mysock.set_timeout((unsigned int)timeout);  
+        rc = mysock.connect(hostname, port);
+        mysock.set_blocking(false);  // blocking timeouts seem not to work
+        return rc;
+    }
+
+    // common read/write routine, avoiding blocking timeouts
+    int common(unsigned char* buffer, int len, int timeout, bool read)
+    {
+        timer.start();
+        mysock.set_blocking(false); // blocking timeouts seem not to work
+        int bytes = 0;
+        bool first = true;
+        do 
+        {
+            if (first)
+                first = false;
+            else
+                wait_ms(timeout < 100 ? timeout : 100);
+            int rc;
+            if (read)
+                rc = mysock.recv((char*)buffer, len);
+            else
+                rc = mysock.send((char*)buffer, len);
+            if (rc < 0)
+            {
+                if (rc != NSAPI_ERROR_WOULD_BLOCK)
+                {
+                    bytes = -1;
+                    break;
+                }
+            } 
+            else
+                bytes += rc;
+        }
+        while (bytes < len && timer.read_ms() < timeout);
+        timer.stop();
+        return bytes;
+    }
+
+    /* returns the number of bytes read, which could be 0.
+       -1 if there was an error on the socket
+    */
+    int read(unsigned char* buffer, int len, int timeout)
+    {
+        return common(buffer, len, timeout, true);
+    }
+
+    int write(unsigned char* buffer, int len, int timeout)
+    {
+        return common(buffer, len, timeout, false);
+    }
+
+    int disconnect()
+    {
+        open = false;
+        return mysock.close();
+    }
+
+    /*bool is_connected()
+    {
+        return mysock.is_connected();
+    }*/
+
+private:
+
+    bool open;
+    TCPSocket mysock;
+    EthernetInterface *net;
+    Timer timer;
+
+};
+
+#endif
diff -r 000000000000 -r 818366b014c9 MQTTmbed.h
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTTmbed.h	Tue Dec 12 15:58:09 2017 +0000
@@ -0,0 +1,48 @@
+#if !defined(MQTT_MBED_H)
+#define MQTT_MBED_H
+
+#include "mbed.h"
+
+class Countdown
+{
+public:
+    Countdown() : t()
+    {
+  
+    }
+    
+    Countdown(int ms) : t()
+    {
+        countdown_ms(ms);   
+    }
+    
+    
+    bool expired()
+    {
+        return t.read_ms() >= interval_end_ms;
+    }
+    
+    void countdown_ms(unsigned long ms)  
+    {
+        t.stop();
+        interval_end_ms = ms;
+        t.reset();
+        t.start();
+    }
+    
+    void countdown(int seconds)
+    {
+        countdown_ms((unsigned long)seconds * 1000L);
+    }
+    
+    int left_ms()
+    {
+        return interval_end_ms - t.read_ms();
+    }
+    
+private:
+    Timer t;
+    unsigned long interval_end_ms; 
+};
+
+#endif
\ No newline at end of file