Version to make it easier to reuse without source modifications
Revision 0:5cd4781e0c88, committed 2018-03-27
- Comitter:
- JMF
- Date:
- Tue Mar 27 17:26:35 2018 +0000
- Commit message:
- Initial commit
Changed in this revision
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTT.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,46 @@ +#ifndef __MQTT_H__ +#define __MQTT_H__ + +#include "MQTTmbed.h" +#include "MQTTSocket.h" + +template <class T> +class MQTTconnect : public MQTTSocket +{ + T eth; + public: + MQTTconnect() : MQTTSocket(ð) + { eth.connect(); } + + T& getEth() + { return eth; } +}; + +// +// To add additional transports, Define it as: +// +#define MQTT_BG96 100 +#define MQTT_WNC14A2A 101 +#define MQTT_ETHERNET 102 + +// +// Then define the interface that will be used: +// +#if MBED_CONF_APP_MQTT_INTERFACE == MQTT_BG96 +#define _ift BG96Interface + +#elif MBED_CONF_APP_MQTT_INTERFACE == MQTT_WNC14A2A +#define _ift WNC14A2AInterface + +#elif MBED_CONF_APP_MQTT_INTERFACE == MQTT_ETHERNET +#define _ift EthernetInterface + +#else +#error "No MQTT Network Transport chosen. Please add 'config.mqtt-interfaces.value' to your mbed_app.json (see README.md for more information)." +#endif + +typedef MQTTconnect<_ift> MQTTct; +typedef _ift MQTTnet; + +#endif +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTAsync.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,606 @@ +/******************************************************************************* + * 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 "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 Callback<void(Result*)> resultHandlerFP; + resultHandlerFP connectHandler; + + typedef Callback<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 Callback<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
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTClient.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,1050 @@ +/******************************************************************************* + * 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 "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=mh; + else + defaultMessageHandler=NULL; + } + + /** 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; + Callback<void( MessageData&)> fp; + } messageHandlers[MAX_MESSAGE_HANDLERS]; // Message handlers are indexed by subscription topic + + Callback<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) //test to see if it is attached + { + MessageData md(topicName, message); + messageHandlers[i].fp(md); + rc = SUCCESS; + } + } + } + + if (rc == FAILURE && defaultMessageHandler ) + { + 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 0: // 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
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTLogging.h Tue Mar 27 17:26:35 2018 +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
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTConnect.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,149 @@ +/******************************************************************************* + * 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 - add connack return code definitions + * Xiang Rong - 442039 Add makefile to Embedded C client + * Ian Craggs - fix for issue #64, bit order in connack response + *******************************************************************************/ + +#ifndef MQTTCONNECT_H_ +#define MQTTCONNECT_H_ + +enum connack_return_codes +{ + MQTT_CONNECTION_ACCEPTED = 0, + MQTT_UNNACCEPTABLE_PROTOCOL = 1, + MQTT_CLIENTID_REJECTED = 2, + MQTT_SERVER_UNAVAILABLE = 3, + MQTT_BAD_USERNAME_OR_PASSWORD = 4, + MQTT_NOT_AUTHORIZED = 5, +}; + +#if !defined(DLLImport) + #define DLLImport +#endif +#if !defined(DLLExport) + #define DLLExport +#endif + + +typedef union +{ + unsigned char all; /**< all connect flags */ +#if defined(REVERSED) + struct + { + unsigned int username : 1; /**< 3.1 user name */ + unsigned int password : 1; /**< 3.1 password */ + unsigned int willRetain : 1; /**< will retain setting */ + unsigned int willQoS : 2; /**< will QoS value */ + unsigned int will : 1; /**< will flag */ + unsigned int cleansession : 1; /**< clean session flag */ + unsigned int : 1; /**< unused */ + } bits; +#else + struct + { + unsigned int : 1; /**< unused */ + unsigned int cleansession : 1; /**< cleansession flag */ + unsigned int will : 1; /**< will flag */ + unsigned int willQoS : 2; /**< will QoS value */ + unsigned int willRetain : 1; /**< will retain setting */ + unsigned int password : 1; /**< 3.1 password */ + unsigned int username : 1; /**< 3.1 user name */ + } bits; +#endif +} MQTTConnectFlags; /**< connect flags byte */ + + + +/** + * Defines the MQTT "Last Will and Testament" (LWT) settings for + * the connect packet. + */ +typedef struct +{ + /** The eyecatcher for this structure. must be MQTW. */ + char struct_id[4]; + /** The version number of this structure. Must be 0 */ + int struct_version; + /** The LWT topic to which the LWT message will be published. */ + MQTTString topicName; + /** The LWT payload. */ + MQTTString message; + /** + * The retained flag for the LWT message (see MQTTAsync_message.retained). + */ + unsigned char retained; + /** + * The quality of service setting for the LWT message (see + * MQTTAsync_message.qos and @ref qos). + */ + char qos; +} MQTTPacket_willOptions; + + +#define MQTTPacket_willOptions_initializer { {'M', 'Q', 'T', 'W'}, 0, {NULL, {0, NULL}}, {NULL, {0, NULL}}, 0, 0 } + + +typedef struct +{ + /** The eyecatcher for this structure. must be MQTC. */ + char struct_id[4]; + /** The version number of this structure. Must be 0 */ + int struct_version; + /** Version of MQTT to be used. 3 = 3.1 4 = 3.1.1 + */ + unsigned char MQTTVersion; + MQTTString clientID; + unsigned short keepAliveInterval; + unsigned char cleansession; + unsigned char willFlag; + MQTTPacket_willOptions will; + MQTTString username; + MQTTString password; +} MQTTPacket_connectData; + +typedef union +{ + unsigned char all; /**< all connack flags */ +#if defined(REVERSED) + struct + { + unsigned int reserved : 7; /**< unused */ + unsigned int sessionpresent : 1; /**< session present flag */ + } bits; +#else + struct + { + unsigned int sessionpresent : 1; /**< session present flag */ + unsigned int reserved: 7; /**< unused */ + } bits; +#endif +} MQTTConnackFlags; /**< connack flags byte */ + +#define MQTTPacket_connectData_initializer { {'M', 'Q', 'T', 'C'}, 0, 4, {NULL, {0, NULL}}, 60, 1, 0, \ + MQTTPacket_willOptions_initializer, {NULL, {0, NULL}}, {NULL, {0, NULL}} } + +DLLExport int MQTTSerialize_connect(unsigned char* buf, int buflen, MQTTPacket_connectData* options); +DLLExport int MQTTDeserialize_connect(MQTTPacket_connectData* data, unsigned char* buf, int len); + +DLLExport int MQTTSerialize_connack(unsigned char* buf, int buflen, unsigned char connack_rc, unsigned char sessionPresent); +DLLExport int MQTTDeserialize_connack(unsigned char* sessionPresent, unsigned char* connack_rc, unsigned char* buf, int buflen); + +DLLExport int MQTTSerialize_disconnect(unsigned char* buf, int buflen); +DLLExport int MQTTSerialize_pingreq(unsigned char* buf, int buflen); + +#endif /* MQTTCONNECT_H_ */ +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTConnectClient.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,215 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "MQTTPacket.h" +#include "StackTrace.h" + +#include <string.h> + +/** + * Determines the length of the MQTT connect packet that would be produced using the supplied connect options. + * @param options the options to be used to build the connect packet + * @return the length of buffer needed to contain the serialized version of the packet + */ +int MQTTSerialize_connectLength(MQTTPacket_connectData* options) +{ + int len = 0; + + FUNC_ENTRY; + + if (options->MQTTVersion == 3) + len = 12; /* variable depending on MQTT or MQIsdp */ + else if (options->MQTTVersion == 4) + len = 10; + + len += MQTTstrlen(options->clientID)+2; + if (options->willFlag) + len += MQTTstrlen(options->will.topicName)+2 + MQTTstrlen(options->will.message)+2; + if (options->username.cstring || options->username.lenstring.data) + len += MQTTstrlen(options->username)+2; + if (options->password.cstring || options->password.lenstring.data) + len += MQTTstrlen(options->password)+2; + + FUNC_EXIT_RC(len); + return len; +} + + +/** + * Serializes the connect options into the buffer. + * @param buf the buffer into which the packet will be serialized + * @param len the length in bytes of the supplied buffer + * @param options the options to be used to build the connect packet + * @return serialized length, or error if 0 + */ +int MQTTSerialize_connect(unsigned char* buf, int buflen, MQTTPacket_connectData* options) +{ + unsigned char *ptr = buf; + MQTTHeader header = {0}; + MQTTConnectFlags flags = {0}; + int len = 0; + int rc = -1; + + FUNC_ENTRY; + if (MQTTPacket_len(len = MQTTSerialize_connectLength(options)) > buflen) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + + header.byte = 0; + header.bits.type = CONNECT; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, len); /* write remaining length */ + + if (options->MQTTVersion == 4) + { + writeCString(&ptr, "MQTT"); + writeChar(&ptr, (char) 4); + } + else + { + writeCString(&ptr, "MQIsdp"); + writeChar(&ptr, (char) 3); + } + + flags.all = 0; + flags.bits.cleansession = options->cleansession; + flags.bits.will = (options->willFlag) ? 1 : 0; + if (flags.bits.will) + { + flags.bits.willQoS = options->will.qos; + flags.bits.willRetain = options->will.retained; + } + + if (options->username.cstring || options->username.lenstring.data) + flags.bits.username = 1; + if (options->password.cstring || options->password.lenstring.data) + flags.bits.password = 1; + + writeChar(&ptr, flags.all); + writeInt(&ptr, options->keepAliveInterval); + writeMQTTString(&ptr, options->clientID); + if (options->willFlag) + { + writeMQTTString(&ptr, options->will.topicName); + writeMQTTString(&ptr, options->will.message); + } + if (flags.bits.username) + writeMQTTString(&ptr, options->username); + if (flags.bits.password) + writeMQTTString(&ptr, options->password); + + rc = ptr - buf; + + exit: FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Deserializes the supplied (wire) buffer into connack data - return code + * @param sessionPresent the session present flag returned (only for MQTT 3.1.1) + * @param connack_rc returned integer value of the connack return code + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param len the length in bytes of the data in the supplied buffer + * @return error code. 1 is success, 0 is failure + */ +int MQTTDeserialize_connack(unsigned char* sessionPresent, unsigned char* connack_rc, unsigned char* buf, int buflen) +{ + MQTTHeader header = {0}; + unsigned char* curdata = buf; + unsigned char* enddata = NULL; + int rc = 0; + int mylen; + MQTTConnackFlags flags = {0}; + + FUNC_ENTRY; + header.byte = readChar(&curdata); + if (header.bits.type != CONNACK) + goto exit; + + curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */ + enddata = curdata + mylen; + if (enddata - curdata < 2) + goto exit; + + flags.all = readChar(&curdata); + *sessionPresent = flags.bits.sessionpresent; + *connack_rc = readChar(&curdata); + + rc = 1; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + + +/** + * Serializes a 0-length packet into the supplied buffer, ready for writing to a socket + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer, to avoid overruns + * @param packettype the message type + * @return serialized length, or error if 0 + */ +int MQTTSerialize_zero(unsigned char* buf, int buflen, unsigned char packettype) +{ + MQTTHeader header = {0}; + int rc = -1; + unsigned char *ptr = buf; + + FUNC_ENTRY; + if (buflen < 2) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + header.byte = 0; + header.bits.type = packettype; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, 0); /* write remaining length */ + rc = ptr - buf; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Serializes a disconnect packet into the supplied buffer, ready for writing to a socket + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer, to avoid overruns + * @return serialized length, or error if 0 + */ +int MQTTSerialize_disconnect(unsigned char* buf, int buflen) +{ + return MQTTSerialize_zero(buf, buflen, DISCONNECT); +} + + +/** + * Serializes a disconnect packet into the supplied buffer, ready for writing to a socket + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer, to avoid overruns + * @return serialized length, or error if 0 + */ +int MQTTSerialize_pingreq(unsigned char* buf, int buflen) +{ + return MQTTSerialize_zero(buf, buflen, PINGREQ); +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTConnectServer.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,148 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "StackTrace.h" +#include "MQTTPacket.h" +#include <string.h> + +#define min(a, b) ((a < b) ? a : b) + + +/** + * Validates MQTT protocol name and version combinations + * @param protocol the MQTT protocol name as an MQTTString + * @param version the MQTT protocol version number, as in the connect packet + * @return correct MQTT combination? 1 is true, 0 is false + */ +int MQTTPacket_checkVersion(MQTTString* protocol, int version) +{ + int rc = 0; + + if (version == 3 && memcmp(protocol->lenstring.data, "MQIdsp", + min(6, protocol->lenstring.len)) == 0) + rc = 1; + else if (version == 4 && memcmp(protocol->lenstring.data, "MQTT", + min(4, protocol->lenstring.len)) == 0) + rc = 1; + return rc; +} + + +/** + * Deserializes the supplied (wire) buffer into connect data structure + * @param data the connect data structure to be filled out + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param len the length in bytes of the data in the supplied buffer + * @return error code. 1 is success, 0 is failure + */ +int MQTTDeserialize_connect(MQTTPacket_connectData* data, unsigned char* buf, int len) +{ + MQTTHeader header = {0}; + MQTTConnectFlags flags = {0}; + unsigned char* curdata = buf; + unsigned char* enddata = &buf[len]; + int rc = 0; + MQTTString Protocol; + int version; + int mylen = 0; + + FUNC_ENTRY; + header.byte = readChar(&curdata); + if (header.bits.type != CONNECT) + goto exit; + + curdata += MQTTPacket_decodeBuf(curdata, &mylen); /* read remaining length */ + + if (!readMQTTLenString(&Protocol, &curdata, enddata) || + enddata - curdata < 0) /* do we have enough data to read the protocol version byte? */ + goto exit; + + version = (int)readChar(&curdata); /* Protocol version */ + /* If we don't recognize the protocol version, we don't parse the connect packet on the + * basis that we don't know what the format will be. + */ + if (MQTTPacket_checkVersion(&Protocol, version)) + { + flags.all = readChar(&curdata); + data->cleansession = flags.bits.cleansession; + data->keepAliveInterval = readInt(&curdata); + if (!readMQTTLenString(&data->clientID, &curdata, enddata)) + goto exit; + if (flags.bits.will) + { + data->willFlag = 1; + data->will.qos = flags.bits.willQoS; + data->will.retained = flags.bits.willRetain; + if (!readMQTTLenString(&data->will.topicName, &curdata, enddata) || + !readMQTTLenString(&data->will.message, &curdata, enddata)) + goto exit; + } + if (flags.bits.username) + { + if (enddata - curdata < 3 || !readMQTTLenString(&data->username, &curdata, enddata)) + goto exit; /* username flag set, but no username supplied - invalid */ + if (flags.bits.password && + (enddata - curdata < 3 || !readMQTTLenString(&data->password, &curdata, enddata))) + goto exit; /* password flag set, but no password supplied - invalid */ + } + else if (flags.bits.password) + goto exit; /* password flag set without username - invalid */ + rc = 1; + } +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Serializes the connack packet into the supplied buffer. + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param connack_rc the integer connack return code to be used + * @param sessionPresent the MQTT 3.1.1 sessionPresent flag + * @return serialized length, or error if 0 + */ +int MQTTSerialize_connack(unsigned char* buf, int buflen, unsigned char connack_rc, unsigned char sessionPresent) +{ + MQTTHeader header = {0}; + int rc = 0; + unsigned char *ptr = buf; + MQTTConnackFlags flags = {0}; + + FUNC_ENTRY; + if (buflen < 2) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + header.byte = 0; + header.bits.type = CONNACK; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, 2); /* write remaining length */ + + flags.all = 0; + flags.bits.sessionpresent = sessionPresent; + writeChar(&ptr, flags.all); + writeChar(&ptr, connack_rc); + + rc = ptr - buf; +exit: + FUNC_EXIT_RC(rc); + return rc; +} +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTDeserializePublish.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,107 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "StackTrace.h" +#include "MQTTPacket.h" +#include <string.h> + +#define min(a, b) ((a < b) ? 1 : 0) + +/** + * Deserializes the supplied (wire) buffer into publish data + * @param dup returned integer - the MQTT dup flag + * @param qos returned integer - the MQTT QoS value + * @param retained returned integer - the MQTT retained flag + * @param packetid returned integer - the MQTT packet identifier + * @param topicName returned MQTTString - the MQTT topic in the publish + * @param payload returned byte buffer - the MQTT publish payload + * @param payloadlen returned integer - the length of the MQTT payload + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param buflen the length in bytes of the data in the supplied buffer + * @return error code. 1 is success + */ +int MQTTDeserialize_publish(unsigned char* dup, int* qos, unsigned char* retained, unsigned short* packetid, MQTTString* topicName, + unsigned char** payload, int* payloadlen, unsigned char* buf, int buflen) +{ + MQTTHeader header = {0}; + unsigned char* curdata = buf; + unsigned char* enddata = NULL; + int rc = 0; + int mylen = 0; + + FUNC_ENTRY; + header.byte = readChar(&curdata); + if (header.bits.type != PUBLISH) + goto exit; + *dup = header.bits.dup; + *qos = header.bits.qos; + *retained = header.bits.retain; + + curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */ + enddata = curdata + mylen; + + if (!readMQTTLenString(topicName, &curdata, enddata) || + enddata - curdata < 0) /* do we have enough data to read the protocol version byte? */ + goto exit; + + if (*qos > 0) + *packetid = readInt(&curdata); + + *payloadlen = enddata - curdata; + *payload = curdata; + rc = 1; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + + +/** + * Deserializes the supplied (wire) buffer into an ack + * @param packettype returned integer - the MQTT packet type + * @param dup returned integer - the MQTT dup flag + * @param packetid returned integer - the MQTT packet identifier + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param buflen the length in bytes of the data in the supplied buffer + * @return error code. 1 is success, 0 is failure + */ +int MQTTDeserialize_ack(unsigned char* packettype, unsigned char* dup, unsigned short* packetid, unsigned char* buf, int buflen) +{ + MQTTHeader header = {0}; + unsigned char* curdata = buf; + unsigned char* enddata = NULL; + int rc = 0; + int mylen; + + FUNC_ENTRY; + header.byte = readChar(&curdata); + *dup = header.bits.dup; + *packettype = header.bits.type; + + curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */ + enddata = curdata + mylen; + + if (enddata - curdata < 2) + goto exit; + *packetid = readInt(&curdata); + + rc = 1; +exit: + FUNC_EXIT_RC(rc); + return rc; +} +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTFormat.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,259 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "StackTrace.h" +#include "MQTTPacket.h" + +#include <string.h> + + +const char* MQTTPacket_names[] = +{ + "RESERVED", "CONNECT", "CONNACK", "PUBLISH", "PUBACK", "PUBREC", "PUBREL", + "PUBCOMP", "SUBSCRIBE", "SUBACK", "UNSUBSCRIBE", "UNSUBACK", + "PINGREQ", "PINGRESP", "DISCONNECT" +}; + + +const char* MQTTPacket_getName(unsigned short packetid) +{ + return MQTTPacket_names[packetid]; +} + + +int MQTTStringFormat_connect(char* strbuf, int strbuflen, MQTTPacket_connectData* data) +{ + int strindex = 0; + + strindex = snprintf(strbuf, strbuflen, + "CONNECT MQTT version %d, client id %.*s, clean session %d, keep alive %d", + (int)data->MQTTVersion, data->clientID.lenstring.len, data->clientID.lenstring.data, + (int)data->cleansession, data->keepAliveInterval); + if (data->willFlag) + strindex += snprintf(&strbuf[strindex], strbuflen - strindex, + ", will QoS %d, will retain %d, will topic %.*s, will message %.*s", + data->will.qos, data->will.retained, + data->will.topicName.lenstring.len, data->will.topicName.lenstring.data, + data->will.message.lenstring.len, data->will.message.lenstring.data); + if (data->username.lenstring.data && data->username.lenstring.len > 0) + strindex += snprintf(&strbuf[strindex], strbuflen - strindex, + ", user name %.*s", data->username.lenstring.len, data->username.lenstring.data); + if (data->password.lenstring.data && data->password.lenstring.len > 0) + strindex += snprintf(&strbuf[strindex], strbuflen - strindex, + ", password %.*s", data->password.lenstring.len, data->password.lenstring.data); + return strindex; +} + + +int MQTTStringFormat_connack(char* strbuf, int strbuflen, unsigned char connack_rc, unsigned char sessionPresent) +{ + int strindex = snprintf(strbuf, strbuflen, "CONNACK session present %d, rc %d", sessionPresent, connack_rc); + return strindex; +} + + +int MQTTStringFormat_publish(char* strbuf, int strbuflen, unsigned char dup, int qos, unsigned char retained, + unsigned short packetid, MQTTString topicName, unsigned char* payload, int payloadlen) +{ + int strindex = snprintf(strbuf, strbuflen, + "PUBLISH dup %d, QoS %d, retained %d, packet id %d, topic %.*s, payload length %d, payload %.*s", + dup, qos, retained, packetid, + (topicName.lenstring.len < 20) ? topicName.lenstring.len : 20, topicName.lenstring.data, + payloadlen, (payloadlen < 20) ? payloadlen : 20, payload); + return strindex; +} + + +int MQTTStringFormat_ack(char* strbuf, int strbuflen, unsigned char packettype, unsigned char dup, unsigned short packetid) +{ + int strindex = snprintf(strbuf, strbuflen, "%s, packet id %d", MQTTPacket_names[packettype], packetid); + if (dup) + strindex += snprintf(strbuf + strindex, strbuflen - strindex, ", dup %d", dup); + return strindex; +} + + +int MQTTStringFormat_subscribe(char* strbuf, int strbuflen, unsigned char dup, unsigned short packetid, int count, + MQTTString topicFilters[], int requestedQoSs[]) +{ + return snprintf(strbuf, strbuflen, + "SUBSCRIBE dup %d, packet id %d count %d topic %.*s qos %d", + dup, packetid, count, + topicFilters[0].lenstring.len, topicFilters[0].lenstring.data, + requestedQoSs[0]); +} + + +int MQTTStringFormat_suback(char* strbuf, int strbuflen, unsigned short packetid, int count, int* grantedQoSs) +{ + return snprintf(strbuf, strbuflen, + "SUBACK packet id %d count %d granted qos %d", packetid, count, grantedQoSs[0]); +} + + +int MQTTStringFormat_unsubscribe(char* strbuf, int strbuflen, unsigned char dup, unsigned short packetid, + int count, MQTTString topicFilters[]) +{ + return snprintf(strbuf, strbuflen, + "UNSUBSCRIBE dup %d, packet id %d count %d topic %.*s", + dup, packetid, count, + topicFilters[0].lenstring.len, topicFilters[0].lenstring.data); +} + + +#if defined(MQTT_CLIENT) +char* MQTTFormat_toClientString(char* strbuf, int strbuflen, unsigned char* buf, int buflen) +{ + int index = 0; + int rem_length = 0; + MQTTHeader header = {0}; + + header.byte = buf[index++]; + index += MQTTPacket_decodeBuf(&buf[index], &rem_length); + + switch (header.bits.type) + { + + case CONNACK: + { + unsigned char sessionPresent, connack_rc; + if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) == 1) + MQTTStringFormat_connack(strbuf, strbuflen, connack_rc, sessionPresent); + } + break; + case PUBLISH: + { + unsigned char dup, retained, *payload; + unsigned short packetid; + int qos, payloadlen; + MQTTString topicName = MQTTString_initializer; + if (MQTTDeserialize_publish(&dup, &qos, &retained, &packetid, &topicName, + &payload, &payloadlen, buf, buflen) == 1) + MQTTStringFormat_publish(strbuf, strbuflen, dup, qos, retained, packetid, + topicName, payload, payloadlen); + } + break; + case PUBACK: + case PUBREC: + case PUBREL: + case PUBCOMP: + { + unsigned char packettype, dup; + unsigned short packetid; + if (MQTTDeserialize_ack(&packettype, &dup, &packetid, buf, buflen) == 1) + MQTTStringFormat_ack(strbuf, strbuflen, packettype, dup, packetid); + } + break; + case SUBACK: + { + unsigned short packetid; + int maxcount = 1, count = 0; + int grantedQoSs[1]; + if (MQTTDeserialize_suback(&packetid, maxcount, &count, grantedQoSs, buf, buflen) == 1) + MQTTStringFormat_suback(strbuf, strbuflen, packetid, count, grantedQoSs); + } + break; + case UNSUBACK: + { + unsigned short packetid; + if (MQTTDeserialize_unsuback(&packetid, buf, buflen) == 1) + MQTTStringFormat_ack(strbuf, strbuflen, UNSUBACK, 0, packetid); + } + break; + case PINGREQ: + case PINGRESP: + case DISCONNECT: + snprintf(strbuf, strbuflen, "%s", MQTTPacket_names[header.bits.type]); + break; + } + return strbuf; +} +#endif + +#if defined(MQTT_SERVER) +char* MQTTFormat_toServerString(char* strbuf, int strbuflen, unsigned char* buf, int buflen) +{ + int index = 0; + int rem_length = 0; + MQTTHeader header = {0}; + + header.byte = buf[index++]; + index += MQTTPacket_decodeBuf(&buf[index], &rem_length); + + switch (header.bits.type) + { + case CONNECT: + { + MQTTPacket_connectData data; + if ((MQTTDeserialize_connect(&data, buf, buflen)) == 1) + MQTTStringFormat_connect(strbuf, strbuflen, &data); + } + break; + case PUBLISH: + { + unsigned char dup, retained, *payload; + unsigned short packetid; + int qos, payloadlen; + MQTTString topicName = MQTTString_initializer; + if (MQTTDeserialize_publish(&dup, &qos, &retained, &packetid, &topicName, + &payload, &payloadlen, buf, buflen) == 1) + MQTTStringFormat_publish(strbuf, strbuflen, dup, qos, retained, packetid, + topicName, payload, payloadlen); + } + break; + case PUBACK: + case PUBREC: + case PUBREL: + case PUBCOMP: + { + unsigned char packettype, dup; + unsigned short packetid; + if (MQTTDeserialize_ack(&packettype, &dup, &packetid, buf, buflen) == 1) + MQTTStringFormat_ack(strbuf, strbuflen, packettype, dup, packetid); + } + break; + case SUBSCRIBE: + { + unsigned char dup; + unsigned short packetid; + int maxcount = 1, count = 0; + MQTTString topicFilters[1]; + int requestedQoSs[1]; + if (MQTTDeserialize_subscribe(&dup, &packetid, maxcount, &count, + topicFilters, requestedQoSs, buf, buflen) == 1) + MQTTStringFormat_subscribe(strbuf, strbuflen, dup, packetid, count, topicFilters, requestedQoSs);; + } + break; + case UNSUBSCRIBE: + { + unsigned char dup; + unsigned short packetid; + int maxcount = 1, count = 0; + MQTTString topicFilters[1]; + if (MQTTDeserialize_unsubscribe(&dup, &packetid, maxcount, &count, topicFilters, buf, buflen) == 1) + MQTTStringFormat_unsubscribe(strbuf, strbuflen, dup, packetid, count, topicFilters); + } + break; + case PINGREQ: + case PINGRESP: + case DISCONNECT: + snprintf(strbuf, strbuflen, "%s", MQTTPacket_names[header.bits.type]); + break; + } + strbuf[strbuflen] = '\0'; + return strbuf; +} +#endif
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTFormat.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,40 @@ +/******************************************************************************* + * 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(MQTTFORMAT_H) +#define MQTTFORMAT_H + +#define MQTT_CLIENT +#define MQTT_SERVER + +#include "StackTrace.h" +#include "MQTTPacket.h" + +const char* MQTTPacket_getName(unsigned short packetid); +int MQTTStringFormat_connect(char* strbuf, int strbuflen, MQTTPacket_connectData* data); +int MQTTStringFormat_connack(char* strbuf, int strbuflen, unsigned char connack_rc, unsigned char sessionPresent); +int MQTTStringFormat_publish(char* strbuf, int strbuflen, unsigned char dup, int qos, unsigned char retained, + unsigned short packetid, MQTTString topicName, unsigned char* payload, int payloadlen); +int MQTTStringFormat_ack(char* strbuf, int strbuflen, unsigned char packettype, unsigned char dup, unsigned short packetid); +int MQTTStringFormat_subscribe(char* strbuf, int strbuflen, unsigned char dup, unsigned short packetid, int count, + MQTTString topicFilters[], int requestedQoSs[]); +int MQTTStringFormat_suback(char* strbuf, int strbuflen, unsigned short packetid, int count, int* grantedQoSs); +int MQTTStringFormat_unsubscribe(char* strbuf, int strbuflen, unsigned char dup, unsigned short packetid, + int count, MQTTString topicFilters[]); +char* MQTTFormat_toClientString(char* strbuf, int strbuflen, unsigned char* buf, int buflen); +char* MQTTFormat_toServerString(char* strbuf, int strbuflen, unsigned char* buf, int buflen); + +#endif
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTPacket.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,411 @@ +/******************************************************************************* + * 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 + * Sergio R. Caprile - non-blocking packet read functions for stream transport + *******************************************************************************/ + +#include "StackTrace.h" +#include "MQTTPacket.h" + +#include <string.h> + +/** + * Encodes the message length according to the MQTT algorithm + * @param buf the buffer into which the encoded data is written + * @param length the length to be encoded + * @return the number of bytes written to buffer + */ +int MQTTPacket_encode(unsigned char* buf, int length) +{ + int rc = 0; + + FUNC_ENTRY; + do + { + char d = length % 128; + length /= 128; + /* if there are more digits to encode, set the top bit of this digit */ + if (length > 0) + d |= 0x80; + buf[rc++] = d; + } while (length > 0); + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Decodes the message length according to the MQTT algorithm + * @param getcharfn pointer to function to read the next character from the data source + * @param value the decoded length returned + * @return the number of bytes read from the socket + */ +int MQTTPacket_decode(int (*getcharfn)(unsigned char*, int), int* value) +{ + unsigned char c; + int multiplier = 1; + int len = 0; +#define MAX_NO_OF_REMAINING_LENGTH_BYTES 4 + + FUNC_ENTRY; + *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 = (*getcharfn)(&c, 1); + if (rc != 1) + goto exit; + *value += (c & 127) * multiplier; + multiplier *= 128; + } while ((c & 128) != 0); +exit: + FUNC_EXIT_RC(len); + return len; +} + + +int MQTTPacket_len(int rem_len) +{ + rem_len += 1; /* header byte */ + + /* now remaining_length field */ + if (rem_len < 128) + rem_len += 1; + else if (rem_len < 16384) + rem_len += 2; + else if (rem_len < 2097151) + rem_len += 3; + else + rem_len += 4; + return rem_len; +} + + +static unsigned char* bufptr; + +int bufchar(unsigned char* c, int count) +{ + int i; + + for (i = 0; i < count; ++i) + *c = *bufptr++; + return count; +} + + +int MQTTPacket_decodeBuf(unsigned char* buf, int* value) +{ + bufptr = buf; + return MQTTPacket_decode(bufchar, value); +} + + +/** + * Calculates an integer from two bytes read from the input buffer + * @param pptr pointer to the input buffer - incremented by the number of bytes used & returned + * @return the integer value calculated + */ +int readInt(unsigned char** pptr) +{ + unsigned char* ptr = *pptr; + int len = 256*(*ptr) + (*(ptr+1)); + *pptr += 2; + return len; +} + + +/** + * Reads one character from the input buffer. + * @param pptr pointer to the input buffer - incremented by the number of bytes used & returned + * @return the character read + */ +char readChar(unsigned char** pptr) +{ + char c = **pptr; + (*pptr)++; + return c; +} + + +/** + * Writes one character to an output buffer. + * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned + * @param c the character to write + */ +void writeChar(unsigned char** pptr, char c) +{ + **pptr = c; + (*pptr)++; +} + + +/** + * Writes an integer as 2 bytes to an output buffer. + * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned + * @param anInt the integer to write + */ +void writeInt(unsigned char** pptr, int anInt) +{ + **pptr = (unsigned char)(anInt / 256); + (*pptr)++; + **pptr = (unsigned char)(anInt % 256); + (*pptr)++; +} + + +/** + * Writes a "UTF" string to an output buffer. Converts C string to length-delimited. + * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned + * @param string the C string to write + */ +void writeCString(unsigned char** pptr, const char* string) +{ + int len = strlen(string); + writeInt(pptr, len); + memcpy(*pptr, string, len); + *pptr += len; +} + + +int getLenStringLen(char* ptr) +{ + int len = 256*((unsigned char)(*ptr)) + (unsigned char)(*(ptr+1)); + return len; +} + + +void writeMQTTString(unsigned char** pptr, MQTTString mqttstring) +{ + if (mqttstring.lenstring.len > 0) + { + writeInt(pptr, mqttstring.lenstring.len); + memcpy(*pptr, mqttstring.lenstring.data, mqttstring.lenstring.len); + *pptr += mqttstring.lenstring.len; + } + else if (mqttstring.cstring) + writeCString(pptr, mqttstring.cstring); + else + writeInt(pptr, 0); +} + + +/** + * @param mqttstring the MQTTString structure into which the data is to be read + * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned + * @param enddata pointer to the end of the data: do not read beyond + * @return 1 if successful, 0 if not + */ +int readMQTTLenString(MQTTString* mqttstring, unsigned char** pptr, unsigned char* enddata) +{ + int rc = 0; + + FUNC_ENTRY; + /* the first two bytes are the length of the string */ + if (enddata - (*pptr) > 1) /* enough length to read the integer? */ + { + mqttstring->lenstring.len = readInt(pptr); /* increments pptr to point past length */ + if (&(*pptr)[mqttstring->lenstring.len] <= enddata) + { + mqttstring->lenstring.data = (char*)*pptr; + *pptr += mqttstring->lenstring.len; + rc = 1; + } + } + mqttstring->cstring = NULL; + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Return the length of the MQTTstring - C string if there is one, otherwise the length delimited string + * @param mqttstring the string to return the length of + * @return the length of the string + */ +int MQTTstrlen(MQTTString mqttstring) +{ + int rc = 0; + + if (mqttstring.cstring) + rc = strlen(mqttstring.cstring); + else + rc = mqttstring.lenstring.len; + return rc; +} + + +/** + * Compares an MQTTString to a C string + * @param a the MQTTString to compare + * @param bptr the C string to compare + * @return boolean - equal or not + */ +int MQTTPacket_equals(MQTTString* a, char* bptr) +{ + int alen = 0, + blen = 0; + char *aptr; + + if (a->cstring) + { + aptr = a->cstring; + alen = strlen(a->cstring); + } + else + { + aptr = a->lenstring.data; + alen = a->lenstring.len; + } + blen = strlen(bptr); + + return (alen == blen) && (strncmp(aptr, bptr, alen) == 0); +} + + +/** + * Helper function to read packet data from some source into a buffer + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param getfn pointer to a function which will read any number of bytes from the needed source + * @return integer MQTT packet type, or -1 on error + * @note the whole message must fit into the caller's buffer + */ +int MQTTPacket_read(unsigned char* buf, int buflen, int (*getfn)(unsigned char*, int)) +{ + 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 ((*getfn)(buf, 1) != 1) + goto exit; + + len = 1; + /* 2. read the remaining length. This is variable in itself */ + MQTTPacket_decode(getfn, &rem_len); + len += MQTTPacket_encode(buf + 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((rem_len + len) > buflen) + goto exit; + if (rem_len && ((*getfn)(buf + len, rem_len) != rem_len)) + goto exit; + + header.byte = buf[0]; + rc = header.bits.type; +exit: + return rc; +} + +/** + * Decodes the message length according to the MQTT algorithm, non-blocking + * @param trp pointer to a transport structure holding what is needed to solve getting data from it + * @param value the decoded length returned + * @return integer the number of bytes read from the socket, 0 for call again, or -1 on error + */ +static int MQTTPacket_decodenb(MQTTTransport *trp) +{ + unsigned char c; + int rc = MQTTPACKET_READ_ERROR; + + FUNC_ENTRY; + if(trp->len == 0){ /* initialize on first call */ + trp->multiplier = 1; + trp->rem_len = 0; + } + do { + int frc; + if (trp->len >= MAX_NO_OF_REMAINING_LENGTH_BYTES) + goto exit; + if ((frc=(*trp->getfn)(trp->sck, &c, 1)) == -1) + goto exit; + if (frc == 0){ + rc = 0; + goto exit; + } + ++(trp->len); + trp->rem_len += (c & 127) * trp->multiplier; + trp->multiplier *= 128; + } while ((c & 128) != 0); + rc = trp->len; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + +/** + * Helper function to read packet data from some source into a buffer, non-blocking + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param trp pointer to a transport structure holding what is needed to solve getting data from it + * @return integer MQTT packet type, 0 for call again, or -1 on error + * @note the whole message must fit into the caller's buffer + */ +int MQTTPacket_readnb(unsigned char* buf, int buflen, MQTTTransport *trp) +{ + int rc = -1, frc; + MQTTHeader header = {0}; + + switch(trp->state){ + default: + trp->state = 0; + /*FALLTHROUGH*/ + case 0: + /* read the header byte. This has the packet type in it */ + if ((frc=(*trp->getfn)(trp->sck, buf, 1)) == -1) + goto exit; + if (frc == 0) + return 0; + trp->len = 0; + ++trp->state; + /*FALLTHROUGH*/ + /* read the remaining length. This is variable in itself */ + case 1: + if((frc=MQTTPacket_decodenb(trp)) == MQTTPACKET_READ_ERROR) + goto exit; + if(frc == 0) + return 0; + trp->len = 1 + MQTTPacket_encode(buf + 1, trp->rem_len); /* put the original remaining length back into the buffer */ + if((trp->rem_len + trp->len) > buflen) + goto exit; + ++trp->state; + /*FALLTHROUGH*/ + case 2: + if(trp->rem_len){ + /* read the rest of the buffer using a callback to supply the rest of the data */ + if ((frc=(*trp->getfn)(trp->sck, buf + trp->len, trp->rem_len)) == -1) + goto exit; + if (frc == 0) + return 0; + trp->rem_len -= frc; + trp->len += frc; + if(trp->rem_len) + return 0; + } + header.byte = buf[0]; + rc = header.bits.type; + break; + } + +exit: + trp->state = 0; + return rc; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTPacket.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,133 @@ +/******************************************************************************* + * 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 + * Xiang Rong - 442039 Add makefile to Embedded C client + *******************************************************************************/ + +#ifndef MQTTPACKET_H_ +#define MQTTPACKET_H_ + +#if defined(__cplusplus) /* If this is a C++ compiler, use C linkage */ +extern "C" { +#endif + +#if defined(WIN32_DLL) || defined(WIN64_DLL) + #define DLLImport __declspec(dllimport) + #define DLLExport __declspec(dllexport) +#elif defined(LINUX_SO) + #define DLLImport extern + #define DLLExport __attribute__ ((visibility ("default"))) +#else + #define DLLImport + #define DLLExport +#endif + +enum errors +{ + MQTTPACKET_BUFFER_TOO_SHORT = -2, + MQTTPACKET_READ_ERROR = -1, + MQTTPACKET_READ_COMPLETE +}; + +enum msgTypes +{ + CONNECT = 1, CONNACK, PUBLISH, PUBACK, PUBREC, PUBREL, + PUBCOMP, SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK, + PINGREQ, PINGRESP, DISCONNECT +}; + +/** + * Bitfields for the MQTT header byte. + */ +typedef union +{ + unsigned char byte; /**< the whole byte */ +#if defined(REVERSED) + struct + { + unsigned int type : 4; /**< message type nibble */ + unsigned int dup : 1; /**< DUP flag bit */ + unsigned int qos : 2; /**< QoS value, 0, 1 or 2 */ + unsigned int retain : 1; /**< retained flag bit */ + } bits; +#else + struct + { + unsigned int retain : 1; /**< retained flag bit */ + unsigned int qos : 2; /**< QoS value, 0, 1 or 2 */ + unsigned int dup : 1; /**< DUP flag bit */ + unsigned int type : 4; /**< message type nibble */ + } bits; +#endif +} MQTTHeader; + +typedef struct +{ + int len; + char* data; +} MQTTLenString; + +typedef struct +{ + char* cstring; + MQTTLenString lenstring; +} MQTTString; + +#define MQTTString_initializer {NULL, {0, NULL}} + +int MQTTstrlen(MQTTString mqttstring); + +#include "MQTTConnect.h" +#include "MQTTPublish.h" +#include "MQTTSubscribe.h" +#include "MQTTUnsubscribe.h" +#include "MQTTFormat.h" + +DLLExport int MQTTSerialize_ack(unsigned char* buf, int buflen, unsigned char type, unsigned char dup, unsigned short packetid); +DLLExport int MQTTDeserialize_ack(unsigned char* packettype, unsigned char* dup, unsigned short* packetid, unsigned char* buf, int buflen); + +int MQTTPacket_len(int rem_len); +DLLExport int MQTTPacket_equals(MQTTString* a, char* b); + +DLLExport int MQTTPacket_encode(unsigned char* buf, int length); +int MQTTPacket_decode(int (*getcharfn)(unsigned char*, int), int* value); +int MQTTPacket_decodeBuf(unsigned char* buf, int* value); + +int readInt(unsigned char** pptr); +char readChar(unsigned char** pptr); +void writeChar(unsigned char** pptr, char c); +void writeInt(unsigned char** pptr, int anInt); +int readMQTTLenString(MQTTString* mqttstring, unsigned char** pptr, unsigned char* enddata); +void writeCString(unsigned char** pptr, const char* string); +void writeMQTTString(unsigned char** pptr, MQTTString mqttstring); + +DLLExport int MQTTPacket_read(unsigned char* buf, int buflen, int (*getfn)(unsigned char*, int)); + +typedef struct { + int (*getfn)(void *, unsigned char*, int); /* must return -1 for error, 0 for call again, or the number of bytes read */ + void *sck; /* pointer to whatever the system may use to identify the transport */ + int multiplier; + int rem_len; + int len; + char state; +}MQTTTransport; + +int MQTTPacket_readnb(unsigned char* buf, int buflen, MQTTTransport *trp); + +#ifdef __cplusplus /* If this is a C++ compiler, use C linkage */ +} +#endif + + +#endif /* MQTTPACKET_H_ */
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTPublish.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,30 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#ifndef MQTTPUBLISH_H_ +#define MQTTPUBLISH_H_ + +int MQTTSerialize_publish(unsigned char* buf, int buflen, unsigned char dup, int qos, unsigned char retained, unsigned short packetid, + MQTTString topicName, unsigned char* payload, int payloadlen); + +int MQTTDeserialize_publish(unsigned char* dup, int* qos, unsigned char* retained, unsigned short* packetid, MQTTString* topicName, + unsigned char** payload, int* payloadlen, unsigned char* buf, int len); + +int MQTTSerialize_puback(unsigned char* buf, int buflen, unsigned short packetid); +int MQTTSerialize_pubrel(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid); +int MQTTSerialize_pubcomp(unsigned char* buf, int buflen, unsigned short packetid); + +#endif /* MQTTPUBLISH_H_ */
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTSerializePublish.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,168 @@ +/******************************************************************************* + * 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 + * Ian Craggs - fix for https://bugs.eclipse.org/bugs/show_bug.cgi?id=453144 + *******************************************************************************/ + +#include "MQTTPacket.h" +#include "StackTrace.h" + +#include <string.h> + + +/** + * Determines the length of the MQTT publish packet that would be produced using the supplied parameters + * @param qos the MQTT QoS of the publish (packetid is omitted for QoS 0) + * @param topicName the topic name to be used in the publish + * @param payloadlen the length of the payload to be sent + * @return the length of buffer needed to contain the serialized version of the packet + */ +int MQTTSerialize_publishLength(int qos, MQTTString topicName, int payloadlen) +{ + int len = 0; + + len += 2 + MQTTstrlen(topicName) + payloadlen; + if (qos > 0) + len += 2; /* packetid */ + return len; +} + + +/** + * Serializes the supplied publish data into the supplied buffer, ready for sending + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param dup integer - the MQTT dup flag + * @param qos integer - the MQTT QoS value + * @param retained integer - the MQTT retained flag + * @param packetid integer - the MQTT packet identifier + * @param topicName MQTTString - the MQTT topic in the publish + * @param payload byte buffer - the MQTT publish payload + * @param payloadlen integer - the length of the MQTT payload + * @return the length of the serialized data. <= 0 indicates error + */ +int MQTTSerialize_publish(unsigned char* buf, int buflen, unsigned char dup, int qos, unsigned char retained, unsigned short packetid, + MQTTString topicName, unsigned char* payload, int payloadlen) +{ + unsigned char *ptr = buf; + MQTTHeader header = {0}; + int rem_len = 0; + int rc = 0; + + FUNC_ENTRY; + if (MQTTPacket_len(rem_len = MQTTSerialize_publishLength(qos, topicName, payloadlen)) > buflen) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + + header.bits.type = PUBLISH; + header.bits.dup = dup; + header.bits.qos = qos; + header.bits.retain = retained; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, rem_len); /* write remaining length */; + + writeMQTTString(&ptr, topicName); + + if (qos > 0) + writeInt(&ptr, packetid); + + memcpy(ptr, payload, payloadlen); + ptr += payloadlen; + + rc = ptr - buf; + +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + + +/** + * Serializes the ack packet into the supplied buffer. + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param type the MQTT packet type + * @param dup the MQTT dup flag + * @param packetid the MQTT packet identifier + * @return serialized length, or error if 0 + */ +int MQTTSerialize_ack(unsigned char* buf, int buflen, unsigned char packettype, unsigned char dup, unsigned short packetid) +{ + MQTTHeader header = {0}; + int rc = 0; + unsigned char *ptr = buf; + + FUNC_ENTRY; + if (buflen < 4) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + header.bits.type = packettype; + header.bits.dup = dup; + header.bits.qos = (packettype == PUBREL) ? 1 : 0; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, 2); /* write remaining length */ + writeInt(&ptr, packetid); + rc = ptr - buf; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Serializes a puback packet into the supplied buffer. + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param packetid integer - the MQTT packet identifier + * @return serialized length, or error if 0 + */ +int MQTTSerialize_puback(unsigned char* buf, int buflen, unsigned short packetid) +{ + return MQTTSerialize_ack(buf, buflen, PUBACK, 0, packetid); +} + + +/** + * Serializes a pubrel packet into the supplied buffer. + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param dup integer - the MQTT dup flag + * @param packetid integer - the MQTT packet identifier + * @return serialized length, or error if 0 + */ +int MQTTSerialize_pubrel(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid) +{ + return MQTTSerialize_ack(buf, buflen, PUBREL, dup, packetid); +} + + +/** + * Serializes a pubrel packet into the supplied buffer. + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param packetid integer - the MQTT packet identifier + * @return serialized length, or error if 0 + */ +int MQTTSerialize_pubcomp(unsigned char* buf, int buflen, unsigned short packetid) +{ + return MQTTSerialize_ack(buf, buflen, PUBCOMP, 0, packetid); +} +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTSubscribe.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,31 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#ifndef MQTTSUBSCRIBE_H_ +#define MQTTSUBSCRIBE_H_ + +int MQTTSerialize_subscribe(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid, + int count, MQTTString topicFilters[], int requestedQoSs[]); + +int MQTTDeserialize_subscribe(unsigned char* dup, unsigned short* packetid, + int maxcount, int* count, MQTTString topicFilters[], int requestedQoSs[], unsigned char* buf, int len); + +int MQTTSerialize_suback(unsigned char* buf, int buflen, unsigned short packetid, int count, int* grantedQoSs); + +int MQTTDeserialize_suback(unsigned short* packetid, int maxcount, int* count, int grantedQoSs[], unsigned char* buf, int len); + + +#endif /* MQTTSUBSCRIBE_H_ */
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTSubscribeClient.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,137 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "MQTTPacket.h" +#include "StackTrace.h" + +#include <string.h> + +/** + * Determines the length of the MQTT subscribe packet that would be produced using the supplied parameters + * @param count the number of topic filter strings in topicFilters + * @param topicFilters the array of topic filter strings to be used in the publish + * @return the length of buffer needed to contain the serialized version of the packet + */ +int MQTTSerialize_subscribeLength(int count, MQTTString topicFilters[]) +{ + int i; + int len = 2; /* packetid */ + + for (i = 0; i < count; ++i) + len += 2 + MQTTstrlen(topicFilters[i]) + 1; /* length + topic + req_qos */ + return len; +} + + +/** + * Serializes the supplied subscribe data into the supplied buffer, ready for sending + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied bufferr + * @param dup integer - the MQTT dup flag + * @param packetid integer - the MQTT packet identifier + * @param count - number of members in the topicFilters and reqQos arrays + * @param topicFilters - array of topic filter names + * @param requestedQoSs - array of requested QoS + * @return the length of the serialized data. <= 0 indicates error + */ +int MQTTSerialize_subscribe(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid, int count, + MQTTString topicFilters[], int requestedQoSs[]) +{ + unsigned char *ptr = buf; + MQTTHeader header = {0}; + int rem_len = 0; + int rc = 0; + int i = 0; + + FUNC_ENTRY; + if (MQTTPacket_len(rem_len = MQTTSerialize_subscribeLength(count, topicFilters)) > buflen) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + + header.byte = 0; + header.bits.type = SUBSCRIBE; + header.bits.dup = dup; + header.bits.qos = 1; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, rem_len); /* write remaining length */; + + writeInt(&ptr, packetid); + + for (i = 0; i < count; ++i) + { + writeMQTTString(&ptr, topicFilters[i]); + writeChar(&ptr, requestedQoSs[i]); + } + + rc = ptr - buf; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + + +/** + * Deserializes the supplied (wire) buffer into suback data + * @param packetid returned integer - the MQTT packet identifier + * @param maxcount - the maximum number of members allowed in the grantedQoSs array + * @param count returned integer - number of members in the grantedQoSs array + * @param grantedQoSs returned array of integers - the granted qualities of service + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param buflen the length in bytes of the data in the supplied buffer + * @return error code. 1 is success, 0 is failure + */ +int MQTTDeserialize_suback(unsigned short* packetid, int maxcount, int* count, int grantedQoSs[], unsigned char* buf, int buflen) +{ + MQTTHeader header = {0}; + unsigned char* curdata = buf; + unsigned char* enddata = NULL; + int rc = 0; + int mylen; + + FUNC_ENTRY; + header.byte = readChar(&curdata); + if (header.bits.type != SUBACK) + goto exit; + + curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */ + enddata = curdata + mylen; + if (enddata - curdata < 2) + goto exit; + + *packetid = readInt(&curdata); + + *count = 0; + while (curdata < enddata) + { + if (*count > maxcount) + { + rc = -1; + goto exit; + } + grantedQoSs[(*count)++] = readChar(&curdata); + } + + rc = 1; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTSubscribeServer.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,112 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "MQTTPacket.h" +#include "StackTrace.h" + +#include <string.h> + + +/** + * Deserializes the supplied (wire) buffer into subscribe data + * @param dup integer returned - the MQTT dup flag + * @param packetid integer returned - the MQTT packet identifier + * @param maxcount - the maximum number of members allowed in the topicFilters and requestedQoSs arrays + * @param count - number of members in the topicFilters and requestedQoSs arrays + * @param topicFilters - array of topic filter names + * @param requestedQoSs - array of requested QoS + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param buflen the length in bytes of the data in the supplied buffer + * @return the length of the serialized data. <= 0 indicates error + */ +int MQTTDeserialize_subscribe(unsigned char* dup, unsigned short* packetid, int maxcount, int* count, MQTTString topicFilters[], + int requestedQoSs[], unsigned char* buf, int buflen) +{ + MQTTHeader header = {0}; + unsigned char* curdata = buf; + unsigned char* enddata = NULL; + int rc = -1; + int mylen = 0; + + FUNC_ENTRY; + header.byte = readChar(&curdata); + if (header.bits.type != SUBSCRIBE) + goto exit; + *dup = header.bits.dup; + + curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */ + enddata = curdata + mylen; + + *packetid = readInt(&curdata); + + *count = 0; + while (curdata < enddata) + { + if (!readMQTTLenString(&topicFilters[*count], &curdata, enddata)) + goto exit; + if (curdata >= enddata) /* do we have enough data to read the req_qos version byte? */ + goto exit; + requestedQoSs[*count] = readChar(&curdata); + (*count)++; + } + + rc = 1; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Serializes the supplied suback data into the supplied buffer, ready for sending + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param packetid integer - the MQTT packet identifier + * @param count - number of members in the grantedQoSs array + * @param grantedQoSs - array of granted QoS + * @return the length of the serialized data. <= 0 indicates error + */ +int MQTTSerialize_suback(unsigned char* buf, int buflen, unsigned short packetid, int count, int* grantedQoSs) +{ + MQTTHeader header = {0}; + int rc = -1; + unsigned char *ptr = buf; + int i; + + FUNC_ENTRY; + if (buflen < 2 + count) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + header.byte = 0; + header.bits.type = SUBACK; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, 2 + count); /* write remaining length */ + + writeInt(&ptr, packetid); + + for (i = 0; i < count; ++i) + writeChar(&ptr, grantedQoSs[i]); + + rc = ptr - buf; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTUnsubscribe.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,30 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#ifndef MQTTUNSUBSCRIBE_H_ +#define MQTTUNSUBSCRIBE_H_ + +int MQTTSerialize_unsubscribe(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid, + int count, MQTTString topicFilters[]); + +int MQTTDeserialize_unsubscribe(unsigned char* dup, unsigned short* packetid, int max_count, int* count, MQTTString topicFilters[], + unsigned char* buf, int len); + +int MQTTSerialize_unsuback(unsigned char* buf, int buflen, unsigned short packetid); + +int MQTTDeserialize_unsuback(unsigned short* packetid, unsigned char* buf, int len); + +#endif /* MQTTUNSUBSCRIBE_H_ */
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTUnsubscribeClient.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,106 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "MQTTPacket.h" +#include "StackTrace.h" + +#include <string.h> + +/** + * Determines the length of the MQTT unsubscribe packet that would be produced using the supplied parameters + * @param count the number of topic filter strings in topicFilters + * @param topicFilters the array of topic filter strings to be used in the publish + * @return the length of buffer needed to contain the serialized version of the packet + */ +int MQTTSerialize_unsubscribeLength(int count, MQTTString topicFilters[]) +{ + int i; + int len = 2; /* packetid */ + + for (i = 0; i < count; ++i) + len += 2 + MQTTstrlen(topicFilters[i]); /* length + topic*/ + return len; +} + + +/** + * Serializes the supplied unsubscribe data into the supplied buffer, ready for sending + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param buflen the length in bytes of the data in the supplied buffer + * @param dup integer - the MQTT dup flag + * @param packetid integer - the MQTT packet identifier + * @param count - number of members in the topicFilters array + * @param topicFilters - array of topic filter names + * @return the length of the serialized data. <= 0 indicates error + */ +int MQTTSerialize_unsubscribe(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid, + int count, MQTTString topicFilters[]) +{ + unsigned char *ptr = buf; + MQTTHeader header = {0}; + int rem_len = 0; + int rc = -1; + int i = 0; + + FUNC_ENTRY; + if (MQTTPacket_len(rem_len = MQTTSerialize_unsubscribeLength(count, topicFilters)) > buflen) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + + header.byte = 0; + header.bits.type = UNSUBSCRIBE; + header.bits.dup = dup; + header.bits.qos = 1; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, rem_len); /* write remaining length */; + + writeInt(&ptr, packetid); + + for (i = 0; i < count; ++i) + writeMQTTString(&ptr, topicFilters[i]); + + rc = ptr - buf; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Deserializes the supplied (wire) buffer into unsuback data + * @param packetid returned integer - the MQTT packet identifier + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param buflen the length in bytes of the data in the supplied buffer + * @return error code. 1 is success, 0 is failure + */ +int MQTTDeserialize_unsuback(unsigned short* packetid, unsigned char* buf, int buflen) +{ + unsigned char type = 0; + unsigned char dup = 0; + int rc = 0; + + FUNC_ENTRY; + rc = MQTTDeserialize_ack(&type, &dup, packetid, buf, buflen); + if (type == UNSUBACK) + rc = 1; + FUNC_EXIT_RC(rc); + return rc; +} + +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/MQTTUnsubscribeServer.c Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,102 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "MQTTPacket.h" +#include "StackTrace.h" + +#include <string.h> + + +/** + * Deserializes the supplied (wire) buffer into unsubscribe data + * @param dup integer returned - the MQTT dup flag + * @param packetid integer returned - the MQTT packet identifier + * @param maxcount - the maximum number of members allowed in the topicFilters and requestedQoSs arrays + * @param count - number of members in the topicFilters and requestedQoSs arrays + * @param topicFilters - array of topic filter names + * @param buf the raw buffer data, of the correct length determined by the remaining length field + * @param buflen the length in bytes of the data in the supplied buffer + * @return the length of the serialized data. <= 0 indicates error + */ +int MQTTDeserialize_unsubscribe(unsigned char* dup, unsigned short* packetid, int maxcount, int* count, MQTTString topicFilters[], + unsigned char* buf, int len) +{ + MQTTHeader header = {0}; + unsigned char* curdata = buf; + unsigned char* enddata = NULL; + int rc = 0; + int mylen = 0; + + FUNC_ENTRY; + header.byte = readChar(&curdata); + if (header.bits.type != UNSUBSCRIBE) + goto exit; + *dup = header.bits.dup; + + curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */ + enddata = curdata + mylen; + + *packetid = readInt(&curdata); + + *count = 0; + while (curdata < enddata) + { + if (!readMQTTLenString(&topicFilters[*count], &curdata, enddata)) + goto exit; + (*count)++; + } + + rc = 1; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + + +/** + * Serializes the supplied unsuback data into the supplied buffer, ready for sending + * @param buf the buffer into which the packet will be serialized + * @param buflen the length in bytes of the supplied buffer + * @param packetid integer - the MQTT packet identifier + * @return the length of the serialized data. <= 0 indicates error + */ +int MQTTSerialize_unsuback(unsigned char* buf, int buflen, unsigned short packetid) +{ + MQTTHeader header = {0}; + int rc = 0; + unsigned char *ptr = buf; + + FUNC_ENTRY; + if (buflen < 2) + { + rc = MQTTPACKET_BUFFER_TOO_SHORT; + goto exit; + } + header.byte = 0; + header.bits.type = UNSUBACK; + writeChar(&ptr, header.byte); /* write header */ + + ptr += MQTTPacket_encode(ptr, 2); /* write remaining length */ + + writeInt(&ptr, packetid); + + rc = ptr - buf; +exit: + FUNC_EXIT_RC(rc); + return rc; +} + +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/StackTrace.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,78 @@ +/******************************************************************************* + * 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 + * Ian Craggs - fix for bug #434081 + *******************************************************************************/ + +#ifndef STACKTRACE_H_ +#define STACKTRACE_H_ + +#include <stdio.h> +#define NOSTACKTRACE 1 + +#if defined(NOSTACKTRACE) +#define FUNC_ENTRY +#define FUNC_ENTRY_NOLOG +#define FUNC_ENTRY_MED +#define FUNC_ENTRY_MAX +#define FUNC_EXIT +#define FUNC_EXIT_NOLOG +#define FUNC_EXIT_MED +#define FUNC_EXIT_MAX +#define FUNC_EXIT_RC(x) +#define FUNC_EXIT_MED_RC(x) +#define FUNC_EXIT_MAX_RC(x) + +#else + +#if defined(WIN32) +#define inline __inline +#define FUNC_ENTRY StackTrace_entry(__FUNCTION__, __LINE__, TRACE_MINIMUM) +#define FUNC_ENTRY_NOLOG StackTrace_entry(__FUNCTION__, __LINE__, -1) +#define FUNC_ENTRY_MED StackTrace_entry(__FUNCTION__, __LINE__, TRACE_MEDIUM) +#define FUNC_ENTRY_MAX StackTrace_entry(__FUNCTION__, __LINE__, TRACE_MAXIMUM) +#define FUNC_EXIT StackTrace_exit(__FUNCTION__, __LINE__, NULL, TRACE_MINIMUM) +#define FUNC_EXIT_NOLOG StackTrace_exit(__FUNCTION__, __LINE__, -1) +#define FUNC_EXIT_MED StackTrace_exit(__FUNCTION__, __LINE__, NULL, TRACE_MEDIUM) +#define FUNC_EXIT_MAX StackTrace_exit(__FUNCTION__, __LINE__, NULL, TRACE_MAXIMUM) +#define FUNC_EXIT_RC(x) StackTrace_exit(__FUNCTION__, __LINE__, &x, TRACE_MINIMUM) +#define FUNC_EXIT_MED_RC(x) StackTrace_exit(__FUNCTION__, __LINE__, &x, TRACE_MEDIUM) +#define FUNC_EXIT_MAX_RC(x) StackTrace_exit(__FUNCTION__, __LINE__, &x, TRACE_MAXIMUM) +#else +#define FUNC_ENTRY StackTrace_entry(__func__, __LINE__, TRACE_MINIMUM) +#define FUNC_ENTRY_NOLOG StackTrace_entry(__func__, __LINE__, -1) +#define FUNC_ENTRY_MED StackTrace_entry(__func__, __LINE__, TRACE_MEDIUM) +#define FUNC_ENTRY_MAX StackTrace_entry(__func__, __LINE__, TRACE_MAXIMUM) +#define FUNC_EXIT StackTrace_exit(__func__, __LINE__, NULL, TRACE_MINIMUM) +#define FUNC_EXIT_NOLOG StackTrace_exit(__func__, __LINE__, NULL, -1) +#define FUNC_EXIT_MED StackTrace_exit(__func__, __LINE__, NULL, TRACE_MEDIUM) +#define FUNC_EXIT_MAX StackTrace_exit(__func__, __LINE__, NULL, TRACE_MAXIMUM) +#define FUNC_EXIT_RC(x) StackTrace_exit(__func__, __LINE__, &x, TRACE_MINIMUM) +#define FUNC_EXIT_MED_RC(x) StackTrace_exit(__func__, __LINE__, &x, TRACE_MEDIUM) +#define FUNC_EXIT_MAX_RC(x) StackTrace_exit(__func__, __LINE__, &x, TRACE_MAXIMUM) + +void StackTrace_entry(const char* name, int line, int trace); +void StackTrace_exit(const char* name, int line, void* return_value, int trace); + +void StackTrace_printStack(FILE* dest); +char* StackTrace_get(unsigned long); + +#endif + +#endif + + + + +#endif /* STACKTRACE_H_ */
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/samples/publish-subscribe.txt Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,115 @@ +#include "MQTTPacket.h" + +#include <errno.h> +#include <stdlib.h> +#include <string.h> + +#include "EthernetInterface.h" + + +TCPSocketConnection mysock; + +int getdata(char* buf, int count) +{ + return mysock.receive(buf, (size_t)count); +} + +int toStop = 0; + + +int main() +{ + MQTTPacket_connectData data = MQTTPacket_connectData_initializer; + int rc = 0; + char buf[200]; + int buflen = sizeof(buf); + int msgid = 1; + MQTTString topicString = MQTTString_initializer; + int req_qos = 0; + char* payload = "mypayload"; + int payloadlen = strlen(payload); + int len = 0; + EthernetInterface eth; + + eth.init(); //Use DHCP + eth.connect(); + + rc = mysock.connect("m2m.eclipse.org", 1883); + + data.clientID.cstring = "SendReceive mbed MQTT "; + data.keepAliveInterval = 20; + data.cleansession = 1; + + mysock.set_blocking(true, 1000); /* 1 second Timeout */ + + len = MQTTSerialize_connect(buf, buflen, &data); + rc = mysock.send(buf, len); + + /* wait for connack */ + if (MQTTPacket_read(buf, buflen, getdata) == CONNACK) + { + int connack_rc; + + if (MQTTDeserialize_connack(&connack_rc, buf, buflen) != 1 || connack_rc != 0) + { + printf("Unable to connect, return code %d\n", connack_rc); + goto exit; + } + } + else + goto exit; + + /* subscribe */ + topicString.cstring = "substopic"; + len = MQTTSerialize_subscribe(buf, buflen, 0, msgid, 1, &topicString, &req_qos); + + rc = mysock.send(buf, len); + if (MQTTPacket_read(buf, buflen, getdata) == SUBACK) /* wait for suback */ + { + int submsgid; + int subcount; + int granted_qos; + + rc = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf, buflen); + if (granted_qos != 0) + { + printf("granted qos != 0, %d\n", granted_qos); + goto exit; + } + } + else + goto exit; + + topicString.cstring = "pubtopic"; + while (!toStop) + { + if (MQTTPacket_read(buf, buflen, getdata) == PUBLISH) + { + int dup; + int qos; + int retained; + int msgid; + int payloadlen_in; + char* payload_in; + int rc; + MQTTString receivedTopic; + + rc = MQTTDeserialize_publish(&dup, &qos, &retained, &msgid, &receivedTopic, + &payload_in, &payloadlen_in, buf, buflen); + printf("message arrived %.*s\n", payloadlen_in, payload_in); + } + + printf("publishing reading\n"); + len = MQTTSerialize_publish(buf, buflen, 0, 0, 0, 0, topicString, payload, payloadlen); + rc = mysock.send(buf, len); + } + + printf("disconnecting\n"); + len = MQTTSerialize_disconnect(buf, buflen); + rc = mysock.send(buf, len); + +exit: + eth.disconnect(); + + return 0; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/samples/simple-publish.txt Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,89 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + +#include "mbed.h" +#include "EthernetInterface.h" +#include "C12832_lcd.h" + +#include "MQTTPacket.h" + +DigitalOut myled(LED2); +C12832_LCD lcd; + +int publish() +{ + MQTTPacket_connectData data = MQTTPacket_connectData_initializer; + int rc = 0; + char buf[200]; + int buflen = sizeof(buf); + TCPSocketConnection mysock; + MQTTString topicString = MQTTString_initializer; + char* payload = "I'm alive!"; + int payloadlen = strlen(payload); + int len = 0; + + mysock.connect("m2m.eclipse.org", 1883); + + data.clientID.cstring = "mbed test client - Ian Craggs"; + data.keepAliveInterval = 20; + data.cleansession = 1; + data.MQTTVersion = 3; + + len = MQTTSerialize_connect(buf, buflen, &data); + + topicString.cstring = "mbed NXP LPC1768"; + len += MQTTSerialize_publish(buf + len, buflen - len, 0, 0, 0, 0, topicString, payload, payloadlen); + + len += MQTTSerialize_disconnect(buf + len, buflen - len); + + rc = 0; + while (rc < len) + { + int rc1 = mysock.send(buf, len); + if (rc1 == -1) + { + lcd.printf("Send failed\n"); + break; + } + else + rc += rc1; + } + if (rc == len) + lcd.printf("Send succeeded\n"); + wait(0.2); + + return 0; +} + +int main() +{ + EthernetInterface eth; + eth.init(); //Use DHCP + eth.connect(); + lcd.printf("IP Address is %s\n", eth.getIPAddress()); + + while(1) + { + myled = 1; + publish(); + wait(0.2); + myled = 0; + publish(); + wait(0.2); + } + + eth.disconnect(); +} \ No newline at end of file
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTPacket/test/test1.txt Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,631 @@ +/******************************************************************************* + * 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 + *******************************************************************************/ + + +#include "MQTTPacket.h" +#include <string.h> +#include <stdlib.h> +#include <stdio.h> + +#if !defined(_WINDOWS) + #include <sys/time.h> + #include <sys/socket.h> + #include <unistd.h> + #include <errno.h> +#else +#include <winsock2.h> +#include <ws2tcpip.h> +#define MAXHOSTNAMELEN 256 +#define EAGAIN WSAEWOULDBLOCK +#define EINTR WSAEINTR +#define EINPROGRESS WSAEINPROGRESS +#define EWOULDBLOCK WSAEWOULDBLOCK +#define ENOTCONN WSAENOTCONN +#define ECONNRESET WSAECONNRESET +#endif + +#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0])) + +struct Options +{ + char* connection; /**< connection to system under test. */ + char** haconnections; + int hacount; + int verbose; + int test_no; +} options = +{ + "tcp://m2m.eclipse.org:1883", + NULL, + 0, + 0, + 0, +}; + +void usage() +{ + +} + +void getopts(int argc, char** argv) +{ + int count = 1; + + while (count < argc) + { + if (strcmp(argv[count], "--test_no") == 0) + { + if (++count < argc) + options.test_no = atoi(argv[count]); + else + usage(); + } + else if (strcmp(argv[count], "--connection") == 0) + { + if (++count < argc) + { + options.connection = argv[count]; + printf("\nSetting connection to %s\n", options.connection); + } + else + usage(); + } + else if (strcmp(argv[count], "--haconnections") == 0) + { + if (++count < argc) + { + char* tok = strtok(argv[count], " "); + options.hacount = 0; + options.haconnections = malloc(sizeof(char*) * 5); + while (tok) + { + options.haconnections[options.hacount] = malloc(strlen(tok) + 1); + strcpy(options.haconnections[options.hacount], tok); + options.hacount++; + tok = strtok(NULL, " "); + } + } + else + usage(); + } + else if (strcmp(argv[count], "--verbose") == 0) + { + options.verbose = 1; + printf("\nSetting verbose on\n"); + } + count++; + } +} + + +#define LOGA_DEBUG 0 +#define LOGA_INFO 1 +#include <stdarg.h> +#include <time.h> +#include <sys/timeb.h> +void MyLog(int LOGA_level, char* format, ...) +{ + static char msg_buf[256]; + va_list args; + struct timeb ts; + + struct tm *timeinfo; + + if (LOGA_level == LOGA_DEBUG && options.verbose == 0) + return; + + ftime(&ts); + timeinfo = localtime(&ts.time); + strftime(msg_buf, 80, "%Y%m%d %H%M%S", timeinfo); + + sprintf(&msg_buf[strlen(msg_buf)], ".%.3hu ", ts.millitm); + + va_start(args, format); + vsnprintf(&msg_buf[strlen(msg_buf)], sizeof(msg_buf) - strlen(msg_buf), format, args); + va_end(args); + + printf("%s\n", msg_buf); + fflush(stdout); +} + + +#if defined(WIN32) || defined(_WINDOWS) +#define mqsleep(A) Sleep(1000*A) +#define START_TIME_TYPE DWORD +static DWORD start_time = 0; +START_TIME_TYPE start_clock(void) +{ + return GetTickCount(); +} +#elif defined(AIX) +#define mqsleep sleep +#define START_TIME_TYPE struct timespec +START_TIME_TYPE start_clock(void) +{ + static struct timespec start; + clock_gettime(CLOCK_REALTIME, &start); + return start; +} +#else +#define mqsleep sleep +#define START_TIME_TYPE struct timeval +/* TODO - unused - remove? static struct timeval start_time; */ +START_TIME_TYPE start_clock(void) +{ + struct timeval start_time; + gettimeofday(&start_time, NULL); + return start_time; +} +#endif + + +#if defined(WIN32) +long elapsed(START_TIME_TYPE start_time) +{ + return GetTickCount() - start_time; +} +#elif defined(AIX) +#define assert(a) +long elapsed(struct timespec start) +{ + struct timespec now, res; + + clock_gettime(CLOCK_REALTIME, &now); + ntimersub(now, start, res); + return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L; +} +#else +long elapsed(START_TIME_TYPE start_time) +{ + struct timeval now, res; + + gettimeofday(&now, NULL); + timersub(&now, &start_time, &res); + return (res.tv_sec)*1000 + (res.tv_usec)/1000; +} +#endif + + +#define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d) +#define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e) + +int tests = 0; +int failures = 0; +FILE* xml; +START_TIME_TYPE global_start_time; +char output[3000]; +char* cur_output = output; + + +void write_test_result() +{ + long duration = elapsed(global_start_time); + + fprintf(xml, " time=\"%ld.%.3ld\" >\n", duration / 1000, duration % 1000); + if (cur_output != output) + { + fprintf(xml, "%s", output); + cur_output = output; + } + fprintf(xml, "</testcase>\n"); +} + + +void myassert(char* filename, int lineno, char* description, int value, char* format, ...) +{ + ++tests; + if (!value) + { + va_list args; + + ++failures; + printf("Assertion failed, file %s, line %d, description: %s\n", filename, lineno, description); + + va_start(args, format); + vprintf(format, args); + va_end(args); + + cur_output += sprintf(cur_output, "<failure type=\"%s\">file %s, line %d </failure>\n", + description, filename, lineno); + } + else + MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description); +} + +#define min(a, b) ((a < b) ? a : b) + +int checkMQTTStrings(MQTTString a, MQTTString b) +{ + if (!a.lenstring.data) + { + a.lenstring.data = a.cstring; + if (a.cstring) + a.lenstring.len = strlen(a.cstring); + } + if (!b.lenstring.data) + { + b.lenstring.data = b.cstring; + if (b.cstring) + b.lenstring.len = strlen(b.cstring); + } + return memcmp(a.lenstring.data, b.lenstring.data, min(a.lenstring.len, b.lenstring.len)) == 0; +} + + +int checkConnectPackets(MQTTPacket_connectData* before, MQTTPacket_connectData* after) +{ + int rc = 0; + int start_failures = failures; + + assert("struct_ids should be the same", + memcmp(before->struct_id, after->struct_id, 4) == 0, "struct_ids were different %.4s\n", after->struct_id); + + assert("struct_versions should be the same", + before->struct_version == after->struct_version, "struct_versions were different\n", rc); + + assert("MQTT versions should be the same", + before->MQTTVersion == after->MQTTVersion, "MQTT versions were different\n", rc); + + assert("ClientIDs should be the same", + checkMQTTStrings(before->clientID, after->clientID), "ClientIDs were different\n", rc); + + assert("keepAliveIntervals should be the same", + before->keepAliveInterval == after->keepAliveInterval, "keepAliveIntervals were different %d\n", after->keepAliveInterval); + + assert("cleansessions should be the same", + before->cleansession == after->cleansession, "cleansessions were different\n", rc); + + assert("willFlags should be the same", + before->willFlag == after->willFlag, "willFlags were different\n", rc); + + if (before->willFlag) + { + assert("will struct_ids should be the same", + memcmp(before->will.struct_id, after->will.struct_id, 4) == 0, "will struct_ids were different %.4s\n", after->struct_id); + + assert("will struct_versions should be the same", + before->will.struct_version == after->will.struct_version, "will struct_versions were different\n", rc); + + assert("topic names should be the same", + checkMQTTStrings(before->will.topicName, after->will.topicName), "topic names were different\n", rc); + + assert("messages should be the same", + checkMQTTStrings(before->will.message, after->will.message), "messages were different\n", rc); + + assert("retained flags should be the same", + before->will.retained == after->will.retained, "retained flags were different\n", rc); + + assert("will qos should be the same", + before->will.qos == after->will.qos, "will qos were different\n", rc); + } + + assert("usernames should be the same", + checkMQTTStrings(before->clientID, after->clientID), "usernames were different\n", rc); + assert("passwords should be the same", + checkMQTTStrings(before->password, after->password), "passwords were different\n", rc); + return failures == start_failures; +} + +int test1(struct Options options) +{ + MQTTPacket_connectData data = MQTTPacket_connectData_initializer; + MQTTPacket_connectData data_after = MQTTPacket_connectData_initializer; + int rc = 0; + char buf[100]; + int buflen = sizeof(buf); + + fprintf(xml, "<testcase classname=\"test1\" name=\"de/serialization\""); + global_start_time = start_clock(); + failures = 0; + MyLog(LOGA_INFO, "Starting test 1 - serialization of connect and back"); + + data.clientID.cstring = "me"; + + data.keepAliveInterval = 20; + data.cleansession = 1; + data.username.cstring = "testuser"; + data.password.cstring = "testpassword"; + + data.willFlag = 1; + data.will.message.cstring = "will message"; + data.will.qos = 1; + data.will.retained = 0; + data.will.topicName.cstring = "will topic"; + + rc = MQTTSerialize_connect(buf, buflen, &data); + assert("good rc from serialize connect", rc > 0, "rc was %d\n", rc); + + rc = MQTTDeserialize_connect(&data_after, buf, buflen); + assert("good rc from deserialize connect", rc == 1, "rc was %d\n", rc); + + /* data after should be the same as data before */ + rc = checkConnectPackets(&data, &data_after); + assert("packets should be the same", rc == 1, "packets were different\n", rc); + +/* exit: */ + MyLog(LOGA_INFO, "TEST1: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", tests, failures); + write_test_result(); + return failures; +} + + +int test2(struct Options options) +{ + int rc = 0; + char buf[100]; + int buflen = sizeof(buf); + + unsigned char dup = 0; + int qos = 2; + unsigned char retained = 0; + int msgid = 23; + MQTTString topicString = MQTTString_initializer; + char *payload = "kkhkhkjkj jkjjk jk jk "; + int payloadlen = strlen(payload); + + unsigned char dup2 = 1; + int qos2 = 1; + unsigned char retained2 = 1; + int msgid2 = 3243; + MQTTString topicString2 = MQTTString_initializer; + char *payload2 = NULL; + int payloadlen2 = 0; + + fprintf(xml, "<testcase classname=\"test1\" name=\"de/serialization\""); + global_start_time = start_clock(); + failures = 0; + MyLog(LOGA_INFO, "Starting test 2 - serialization of publish and back"); + + topicString.cstring = "mytopic"; + rc = MQTTSerialize_publish(buf, buflen, dup, qos, retained, msgid, topicString, + payload, payloadlen); + assert("good rc from serialize publish", rc > 0, "rc was %d\n", rc); + + rc = MQTTDeserialize_publish(&dup2, &qos2, &retained2, &msgid2, &topicString2, + &payload2, &payloadlen2, buf, buflen); + assert("good rc from deserialize publish", rc == 1, "rc was %d\n", rc); + + /* data after should be the same as data before */ + assert("dups should be the same", dup == dup2, "dups were different %d\n", dup2); + assert("qoss should be the same", qos == qos2, "qoss were different %d\n", qos2); + assert("retaineds should be the same", retained == retained2, "retaineds were different %d\n", retained2); + assert("msgids should be the same", msgid == msgid2, "msgids were different %d\n", msgid2); + + assert("topics should be the same", + checkMQTTStrings(topicString, topicString2), "topics were different %s\n", ""); //topicString2); + + assert("payload lengths should be the same", + payloadlen == payloadlen2, "payload lengths were different %d\n", payloadlen2); + + assert("payloads should be the same", + memcmp(payload, payload2, payloadlen) == 0, "payloads were different %s\n", ""); + +/*exit:*/ + MyLog(LOGA_INFO, "TEST2: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", tests, failures); + write_test_result(); + return failures; +} + + + +int test3(struct Options options) +{ + int i = 0; + int rc = 0; + char buf[100]; + int buflen = sizeof(buf); +#define TOPIC_COUNT 2 + + int dup = 0; + int msgid = 23; + int count = TOPIC_COUNT; + MQTTString topicStrings[TOPIC_COUNT] = { MQTTString_initializer, MQTTString_initializer }; + int req_qoss[TOPIC_COUNT] = {2, 1}; + + int dup2 = 1; + int msgid2 = 2223; + int count2 = 0; + MQTTString topicStrings2[TOPIC_COUNT] = { MQTTString_initializer, MQTTString_initializer }; + int req_qoss2[TOPIC_COUNT] = {0, 0}; + + fprintf(xml, "<testcase classname=\"test1\" name=\"de/serialization\""); + global_start_time = start_clock(); + failures = 0; + MyLog(LOGA_INFO, "Starting test 2 - serialization of subscribe and back"); + + topicStrings[0].cstring = "mytopic"; + topicStrings[1].cstring = "mytopic2"; + rc = MQTTSerialize_subscribe(buf, buflen, dup, msgid, count, topicStrings, req_qoss); + assert("good rc from serialize subscribe", rc > 0, "rc was %d\n", rc); + + rc = MQTTDeserialize_subscribe(&dup2, &msgid2, 2, &count2, topicStrings2, req_qoss2, buf, buflen); + assert("good rc from deserialize subscribe", rc == 1, "rc was %d\n", rc); + + /* data after should be the same as data before */ + assert("dups should be the same", dup == dup2, "dups were different %d\n", dup2); + assert("msgids should be the same", msgid == msgid2, "msgids were different %d\n", msgid2); + + assert("count should be the same", count == count2, "counts were different %d\n", count2); + + for (i = 0; i < count2; ++i) + { + assert("topics should be the same", + checkMQTTStrings(topicStrings[i], topicStrings2[i]), "topics were different %s\n", ""); + + assert("qoss should be the same", req_qoss[i] == req_qoss2[i], "qoss were different %d\n", req_qoss2[i]); + } + +/*exit:*/ + MyLog(LOGA_INFO, "TEST3: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", tests, failures); + write_test_result(); + return failures; +} + + +int test4(struct Options options) +{ + int i = 0; + int rc = 0; + char buf[100]; + int buflen = sizeof(buf); +#define TOPIC_COUNT 2 + + int msgid = 23; + int count = TOPIC_COUNT; + int granted_qoss[TOPIC_COUNT] = {2, 1}; +; + int msgid2 = 2223; + int count2 = 0; + int granted_qoss2[TOPIC_COUNT] = {0, 0}; + + fprintf(xml, "<testcase classname=\"test1\" name=\"de/serialization\""); + global_start_time = start_clock(); + failures = 0; + MyLog(LOGA_INFO, "Starting test 4 - serialization of suback and back"); + + rc = MQTTSerialize_suback(buf, buflen, msgid, count, granted_qoss); + assert("good rc from serialize suback", rc > 0, "rc was %d\n", rc); + + rc = MQTTDeserialize_suback(&msgid2, 2, &count2, granted_qoss2, buf, buflen); + assert("good rc from deserialize suback", rc == 1, "rc was %d\n", rc); + + /* data after should be the same as data before */ + assert("msgids should be the same", msgid == msgid2, "msgids were different %d\n", msgid2); + + assert("count should be the same", count == count2, "counts were different %d\n", count2); + + for (i = 0; i < count2; ++i) + assert("qoss should be the same", granted_qoss[i] == granted_qoss2[i], "qoss were different %d\n", granted_qoss2[i]); + +/* exit: */ + MyLog(LOGA_INFO, "TEST4: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", tests, failures); + write_test_result(); + return failures; +} + + +int test5(struct Options options) +{ + int i = 0; + int rc = 0; + char buf[100]; + int buflen = sizeof(buf); +#define TOPIC_COUNT 2 + + int dup = 0; + int msgid = 23; + int count = TOPIC_COUNT; + MQTTString topicStrings[TOPIC_COUNT] = { MQTTString_initializer, MQTTString_initializer }; + + int dup2 = 1; + int msgid2 = 2223; + int count2 = 0; + MQTTString topicStrings2[TOPIC_COUNT] = { MQTTString_initializer, MQTTString_initializer }; + + fprintf(xml, "<testcase classname=\"test1\" name=\"de/serialization\""); + global_start_time = start_clock(); + failures = 0; + MyLog(LOGA_INFO, "Starting test 2 - serialization of unsubscribe and back"); + + topicStrings[0].cstring = "mytopic"; + topicStrings[1].cstring = "mytopic2"; + rc = MQTTSerialize_unsubscribe(buf, buflen, dup, msgid, count, topicStrings); + assert("good rc from serialize unsubscribe", rc > 0, "rc was %d\n", rc); + + rc = MQTTDeserialize_unsubscribe(&dup2, &msgid2, 2, &count2, topicStrings2, buf, buflen); + assert("good rc from deserialize unsubscribe", rc == 1, "rc was %d\n", rc); + + /* data after should be the same as data before */ + assert("dups should be the same", dup == dup2, "dups were different %d\n", dup2); + assert("msgids should be the same", msgid == msgid2, "msgids were different %d\n", msgid2); + + assert("count should be the same", count == count2, "counts were different %d\n", count2); + + for (i = 0; i < count2; ++i) + assert("topics should be the same", + checkMQTTStrings(topicStrings[i], topicStrings2[i]), "topics were different %s\n", ""); + +/* exit: */ + MyLog(LOGA_INFO, "TEST5: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", tests, failures); + write_test_result(); + return failures; +} + + +int test6(struct Options options) +{ + int rc = 0; + char buf[100]; + int buflen = sizeof(buf); + + int connack_rc = 77; + + int connack_rc2 = 0; + + fprintf(xml, "<testcase classname=\"test1\" name=\"de/serialization\""); + global_start_time = start_clock(); + failures = 0; + MyLog(LOGA_INFO, "Starting test 2 - serialization of connack and back"); + + rc = MQTTSerialize_connack(buf, buflen, connack_rc); + assert("good rc from serialize connack", rc > 0, "rc was %d\n", rc); + + rc = MQTTDeserialize_connack(&connack_rc2, buf, buflen); + assert("good rc from deserialize connack", rc == 1, "rc was %d\n", rc); + + /* data after should be the same as data before */ + assert("dups should be the same", connack_rc == connack_rc2, "dups were different %d\n", connack_rc2); + +/* exit: */ + MyLog(LOGA_INFO, "TEST6: test %s. %d tests run, %d failures.", + (failures == 0) ? "passed" : "failed", tests, failures); + write_test_result(); + return failures; +} + + +int main(int argc, char** argv) +{ + int rc = 0; + int (*tests[])() = {NULL, test1, test2, test3, test4, test5, test6}; + + xml = fopen("TEST-test1.xml", "w"); + fprintf(xml, "<testsuite name=\"test1\" tests=\"%d\">\n", (int)(ARRAY_SIZE(tests) - 1)); + + getopts(argc, argv); + + if (options.test_no == 0) + { /* run all the tests */ + for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no) + rc += tests[options.test_no](options); /* return number of failures. 0 = test succeeded */ + } + else + rc = tests[options.test_no](options); /* run just the selected test */ + + if (rc == 0) + MyLog(LOGA_INFO, "verdict pass"); + else + MyLog(LOGA_INFO, "verdict fail"); + + fprintf(xml, "</testsuite>\n"); + fclose(xml); + return rc; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTSocket.h Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,95 @@ +#if !defined(MQTTSOCKET_H) +#define MQTTSOCKET_H + +#include "MQTTmbed.h" +#include <Timer.h> + +class MQTTSocket +{ +public: + MQTTSocket(NetworkInterface *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; + NetworkInterface *net; + Timer timer; + +}; + +#endif
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTTmbed.h Tue Mar 27 17:26:35 2018 +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; + long interval_end_ms; +}; + +#endif
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/README.md Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,20 @@ +# MQTT -- An API for using MQTT over multiple transports + +This library originated from https://os.mbed.com/teams/mqtt/code/MQTT/ but has been modified to allow you to select different transports by +modifing the mbed_app.json file. + +## Specifying the connectivity method + +Add the following to your `mbed_app.json` file: + +```json +{ + "config": { + "mqtt-interface":{ + "help": "options are MQTT_ETHERNET, MQTT_WNC14A2A, MQTT_BG96" + "value": "MQTT_ETHERNET" + } + } +} +``` +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/mbed_lib.json Tue Mar 27 17:26:35 2018 +0000 @@ -0,0 +1,10 @@ +{ + "name": "MQTT", + "config": { + "mqtt-interface":{ + "help": "Options are MQTT_ETHERNET, MQTT_WNC14A2A, MQTT_BG96", + "value": "MQTT_ETHERNET" + } + } +} +