From 9a442c4f9a4f03d8892091e091d92edfd3e07090 Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Thu, 9 Apr 2015 16:47:53 +0100 Subject: [PATCH] New MQTT-SN high level client API --- MQTTSNClient/src/FP.h | 208 +++++++ MQTTSNClient/src/MQTTLogging.h | 55 ++ MQTTSNClient/src/MQTTSNClient.h | 920 +++++++++++++++++++++++++++++++ MQTTSNClient/src/linux/linux.cpp | 224 ++++++++ 4 files changed, 1407 insertions(+) create mode 100644 MQTTSNClient/src/FP.h create mode 100644 MQTTSNClient/src/MQTTLogging.h create mode 100644 MQTTSNClient/src/MQTTSNClient.h create mode 100644 MQTTSNClient/src/linux/linux.cpp diff --git a/MQTTSNClient/src/FP.h b/MQTTSNClient/src/FP.h new file mode 100644 index 0000000..de5221d --- /dev/null +++ b/MQTTSNClient/src/FP.h @@ -0,0 +1,208 @@ +/******************************************************************************* + * Copyright (c) 2013, 2014 + * + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * 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: + * Sam Grove - initial API and implementation and/or initial documentation + * Ian Craggs - added attached and detached member functions + * Sam Grove - removed need for FP.cpp + *******************************************************************************/ + +#ifndef FP_H +#define FP_H + +/** Example using the FP Class with global functions + * @code + * #include "mbed.h" + * #include "FP.h" + * + * FPfp; + * DigitalOut myled(LED1); + * + * void handler(bool value) + * { + * myled = value; + * return; + * } + * + * int main() + * { + * fp.attach(&handler); + * + * while(1) + * { + * fp(1); + * wait(0.2); + * fp(0); + * wait(0.2); + * } + * } + * @endcode + */ + +/** Example using the FP Class with different class member functions + * @code + * #include "mbed.h" + * #include "FP.h" + * + * FPfp; + * DigitalOut myled(LED4); + * + * class Wrapper + * { + * public: + * Wrapper(){} + * + * void handler(bool value) + * { + * myled = value; + * return; + * } + * }; + * + * int main() + * { + * Wrapper wrapped; + * fp.attach(&wrapped, &Wrapper::handler); + * + * while(1) + * { + * fp(1); + * wait(0.2); + * fp(0); + * wait(0.2); + * } + * } + * @endcode + */ + +/** Example using the FP Class with member FP and member function +* @code +* #include "mbed.h" +* #include "FP.h" +* +* DigitalOut myled(LED2); +* +* class Wrapper +* { +* public: +* Wrapper() +* { +* fp.attach(this, &Wrapper::handler); +* } +* +* void handler(bool value) +* { +* myled = value; +* return; +* } +* +* FPfp; +* }; +* +* int main() +* { +* Wrapper wrapped; +* +* while(1) +* { +* wrapped.fp(1); +* wait(0.2); +* wrapped.fp(0); +* wait(0.2); +* } +* } +* @endcode +*/ + +/** + * @class FP + * @brief API for managing Function Pointers + */ +template +class FP +{ +public: + /** Create the FP object - only one callback can be attached to the object, that is + * a member function or a global function, not both at the same time + */ + FP() + { + obj_callback = 0; + c_callback = 0; + } + + /** Add a callback function to the object + * @param item - Address of the initialized object + * @param member - Address of the member function (dont forget the scope that the function is defined in) + */ + template + void attach(T *item, retT (T::*method)(argT)) + { + obj_callback = (FPtrDummy *)(item); + method_callback = (retT (FPtrDummy::*)(argT))(method); + return; + } + + /** Add a callback function to the object + * @param function - The address of a globally defined function + */ + void attach(retT (*function)(argT)) + { + c_callback = function; + } + + /** Invoke the function attached to the class + * @param arg - An argument that is passed into the function handler that is called + * @return The return from the function hanlder called by this class + */ + retT operator()(argT arg) const + { + if( 0 != c_callback ) { + return obj_callback ? (obj_callback->*method_callback)(arg) : (*c_callback)(arg); + } + return (retT)0; + } + + /** Determine if an callback is currently hooked + * @return 1 if a method is hooked, 0 otherwise + */ + bool attached() + { + return obj_callback || c_callback; + } + + /** Release a function from the callback hook + */ + void detach() + { + obj_callback = 0; + c_callback = 0; + } + +private: + + // empty type used for casting + class FPtrDummy; + + FPtrDummy *obj_callback; + + /** + * @union Funciton + * @brief Member or global callback function + */ + union { + retT (*c_callback)(argT); /*!< Footprint for a global function */ + retT (FPtrDummy::*method_callback)(argT); /*!< Footprint for a member function */ + }; +}; + +#endif diff --git a/MQTTSNClient/src/MQTTLogging.h b/MQTTSNClient/src/MQTTLogging.h new file mode 100644 index 0000000..fd83e61 --- /dev/null +++ b/MQTTSNClient/src/MQTTLogging.h @@ -0,0 +1,55 @@ +/******************************************************************************* + * 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(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 diff --git a/MQTTSNClient/src/MQTTSNClient.h b/MQTTSNClient/src/MQTTSNClient.h new file mode 100644 index 0000000..674d2f4 --- /dev/null +++ b/MQTTSNClient/src/MQTTSNClient.h @@ -0,0 +1,920 @@ +/******************************************************************************* + * Copyright (c) 2014, 2015 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(MQTTSNCLIENT_H) +#define MQTTSNCLIENT_H + +#include "FP.h" +#include "MQTTSNPacket.h" +#include "stdio.h" +#include "MQTTLogging.h" + +// Data limits +#if !defined(MAX_REGISTRATIONS) + #define MAX_REGISTRATIONS 5 +#endif +#if !defined(MAX_REGISTRATION_TOPIC_NAME_LENGTH) + #define MAX_REGISTRATION_TOPIC_NAME_LENGTH 20 +#endif +#if !defined(MAX_INCOMING_QOS2_MESSAGES) + #define MAX_INCOMING_QOS2_MESSAGES 10 +#endif + +#if !defined(MQTTSNCLIENT_QOS1) + #define MQTTSNCLIENT_QOS1 1 +#endif +#if !defined(MQTTSNCLIENT_QOS2) + #define MQTTSNCLIENT_QOS2 0 +#endif + +namespace MQTTSN +{ + + +enum QoS { QOS0, QOS1, QOS2 }; + +// all failure return codes must be negative +enum returnCode { MAX_SUBSCRIPTIONS_EXCEEDED = -3, 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(MQTTSN_topicid &aTopic, struct Message &aMessage) : message(aMessage), topic(aTopic) + { } + + struct Message &message; + MQTTSN_topicid &topic; +}; + + +class PacketId +{ +public: + PacketId() + { + next = 0; + } + + int getNext() + { + return next = (next == MAX_PACKET_ID) ? 1 : ++next; + } + +private: + static const int MAX_PACKET_ID = 65535; + int next; +}; + + +/** + * @class MQTTSNClient + * @brief blocking, non-threaded MQTTSN 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 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 + */ + void setDefaultMessageHandler(messageHandler mh) + { + defaultMessageHandler.attach(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(MQTTSNPacket_connectData& options); + + /** 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(MQTTSN_topicid& topic, 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(MQTTSN_topicid &topic, 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(MQTTSN_topicid& topic, 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(MQTTSN_topicid& topicFilter, enum QoS qos, messageHandler mh); + + /** 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(MQTTSN_topicid& topicFilter); + + /** MQTT Disconnect - send an MQTT disconnect packet, and clean up any state + * @param duration - used for sleeping clients, 0 means no duration + * @return success code - + */ + int disconnect(unsigned short duration = 0); + + /** 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; + } + +protected: + + int cycle(Timer& timer); + int waitfor(int packet_type, Timer& timer); + +private: + + 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(MQTTSN_topicid& topic, Message& message); + bool isTopicMatched(char* topicFilter, MQTTSNString& topicName); + + Network& ipstack; + unsigned long command_timeout_ms; + + unsigned char sendbuf[MAX_PACKET_SIZE]; + unsigned char readbuf[MAX_PACKET_SIZE]; + + Timer last_sent, last_received; + unsigned short duration; + bool ping_outstanding; + bool cleansession; + + PacketId packetid; + + struct MessageHandlers + { + MQTTSN_topicid* topicFilter; + FP fp; + } messageHandlers[MAX_MESSAGE_HANDLERS]; // Message handlers are indexed by subscription topic + + FP defaultMessageHandler; + + bool isconnected; + + struct Registrations + { + unsigned short id; + char name[MAX_REGISTRATION_TOPIC_NAME_LENGTH]; + } registrations[MAX_REGISTRATIONS]; + +#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 + unsigned char pubbuf[MAX_PACKET_SIZE]; // store the last publish for sending on reconnect + int inflightLen; + unsigned short inflightMsgid; + enum QoS inflightQoS; +#endif + +#if MQTTCLIENT_QOS2 + bool pubrel; + unsigned short incomingQoS2messages[MAX_INCOMING_QOS2_MESSAGES]; + bool isQoS2msgidFree(unsigned short id); + bool useQoS2msgid(unsigned short id); +#endif + +}; + +} + + +template +MQTTSN::Client::Client(Network& network, unsigned int command_timeout_ms) : ipstack(network), packetid() +{ + last_sent = Timer(); + last_received = Timer(); + ping_outstanding = false; + for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i) + messageHandlers[i].topicFilter = 0; + this->command_timeout_ms = command_timeout_ms; + isconnected = false; + +#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 +} + +#if MQTTCLIENT_QOS2 +template +bool MQTTSN::Client::isQoS2msgidFree(unsigned short id) +{ + for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) + { + if (incomingQoS2messages[i] == id) + return false; + } + return true; +} + + +template +bool MQTTSN::Client::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; +} +#endif + + +template +int MQTTSN::Client::sendPacket(int length, Timer& timer) +{ + int rc = FAILURE, + sent = 0; + + while (sent < length && !timer.expired()) + { + rc = ipstack.write(&sendbuf[sent], length, timer.left_ms()); + if (rc < 0) // there was an error writing the data + break; + sent += rc; + } + if (sent == length) + { + if (this->duration > 0) + last_sent.countdown(this->duration); // record the fact that we have successfully sent the packet + rc = SUCCESS; + } + else + rc = FAILURE; + +#if defined(MQTT_DEBUG) + char printbuf[50]; + DEBUG("Rc %d from sending packet %s\n", rc, MQTTPacket_toString(printbuf, sizeof(printbuf), sendbuf, length)); +#endif + return rc; +} + + +template +int MQTTSN::Client::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 = MQTTSNPACKET_READ_ERROR; + + if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES) + { + rc = MQTTSNPACKET_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 +int MQTTSN::Client::readPacket(Timer& timer) +{ + int rc = FAILURE; + int len = 0; // the length of the whole packet including length field + int lenlen = 0; + int datalen = 0; + + #define MQTTSN_MIN_PACKET_LENGTH 3 + // 1. read the packet, datagram style + if ((len = ipstack.read(readbuf, MAX_PACKET_SIZE, timer.left_ms())) < MQTTSN_MIN_PACKET_LENGTH) + goto exit; + + // 2. read the length. This is variable in itself + lenlen = MQTTSNPacket_decode(readbuf, len, &datalen); + if (datalen != len) + goto exit; // there was an error + + rc = readbuf[lenlen]; + if (this->duration > 0) + last_received.countdown(this->duration); // record the fact that we have successfully received a packet +exit: + +#if defined(MQTT_DEBUG) + char printbuf[50]; + DEBUG("Rc %d from receiving packet %s\n", rc, MQTTPacket_toString(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 +bool MQTTSN::Client::isTopicMatched(char* topicFilter, MQTTSNString& 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 +int MQTTSN::Client::deliverMessage(MQTTSN_topicid& topic, 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 && (MQTTSNtopic_equals(&topic, messageHandlers[i].topicFilter) || + isTopicMatched(messageHandlers[i].topicFilter, topic))) + { + if (messageHandlers[i].fp.attached()) + { + MessageData md(topicName, message); + messageHandlers[i].fp(md); + rc = SUCCESS; + } + } + */ + } + + if (rc == FAILURE && defaultMessageHandler.attached()) + { + MessageData md(topic, message); + defaultMessageHandler(md); + rc = SUCCESS; + } + + return rc; +} + + + +template +int MQTTSN::Client::yield(unsigned long timeout_ms) +{ + int rc = SUCCESS; + Timer timer = Timer(); + + timer.countdown_ms(timeout_ms); + while (!timer.expired()) + { + if (cycle(timer) == FAILURE) + { + rc = FAILURE; + break; + } + } + + return rc; +} + + +template +int MQTTSN::Client::cycle(Timer& timer) +{ + /* get one piece of work off the wire and one pass through */ + + // read the socket, see what work is due + unsigned short packet_type = readPacket(timer); + + int len = 0; + unsigned char rc = SUCCESS; + + switch (packet_type) + { + case MQTTSN_CONNACK: + case MQTTSN_PUBACK: + case MQTTSN_SUBACK: + case MQTTSN_REGACK: + break; + case MQTTSN_REGISTER: + { + unsigned short topicid, packetid; + MQTTSNString topicName; + rc = MQTTSN_RC_ACCEPTED; + if (MQTTSNDeserialize_register(&topicid, &packetid, &topicName, readbuf, MAX_PACKET_SIZE) != 1) + goto exit; + len = MQTTSNSerialize_regack(sendbuf, MAX_PACKET_SIZE, topicid, packetid, rc); + if (len <= 0) + rc = FAILURE; + else + rc = sendPacket(len, timer); + break; + } + case MQTTSN_PUBLISH: + MQTTSN_topicid topicid; + Message msg; + if (MQTTSNDeserialize_publish((unsigned char*)&msg.dup, (int*)&msg.qos, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicid, + (unsigned char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_PACKET_SIZE) != 1) + goto exit; +#if MQTTCLIENT_QOS2 + if (msg.qos != QOS2) +#endif + deliverMessage(topicid, 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 = MQTTSNSerialize_puback(sendbuf, MAX_PACKET_SIZE, topicid.data.id, msg.id, 0); + else if (msg.qos == QOS2) + len = MQTTSNSerialize_pubrec(sendbuf, MAX_PACKET_SIZE, 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: + unsigned short mypacketid; + unsigned char dup, type; + if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_PACKET_SIZE) != 1) + rc = FAILURE; + else if ((len = MQTTSerialize_ack(sendbuf, MAX_PACKET_SIZE, PUBREL, 0, mypacketid)) <= 0) + rc = FAILURE; + else if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet + rc = FAILURE; // there was a problem + if (rc == FAILURE) + goto exit; // there was a problem + break; + case PUBCOMP: + break; +#endif + case MQTTSN_PINGRESP: + ping_outstanding = false; + break; + } + keepalive(); +exit: + if (rc == SUCCESS) + rc = packet_type; + return rc; +} + + +template +int MQTTSN::Client::keepalive() +{ + int rc = FAILURE; + + if (duration == 0) + { + rc = SUCCESS; + goto exit; + } + + if (last_sent.expired() || last_received.expired()) + { + if (!ping_outstanding) + { + MQTTSNString clientid = MQTTSNString_initializer; + Timer timer = Timer(1000); + int len = MQTTSNSerialize_pingreq(sendbuf, MAX_PACKET_SIZE, clientid); + if (len > 0 && (rc = sendPacket(len, timer)) == SUCCESS) // send the ping packet + ping_outstanding = true; + } + } + +exit: + return rc; +} + + +// only used in single-threaded mode where one command at a time is in process +template +int MQTTSN::Client::waitfor(int packet_type, Timer& timer) +{ + int rc = FAILURE; + + do + { + if (timer.expired()) + break; // we timed out + } + while ((rc = cycle(timer)) != packet_type); + + return rc; +} + + +template +int MQTTSN::Client::connect(MQTTSNPacket_connectData& options) +{ + Timer connect_timer = 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->duration = options.duration; + this->cleansession = options.cleansession; + if ((len = MQTTSNSerialize_connect(sendbuf, MAX_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->duration > 0) + last_received.countdown(this->duration); + // this will be a blocking call, wait for the connack + if (waitfor(MQTTSN_CONNACK, connect_timer) == MQTTSN_CONNACK) + { + //unsigned char connack_rc = 255; + int connack_rc = 255; + if (MQTTSNDeserialize_connack(&connack_rc, readbuf, MAX_PACKET_SIZE) == 1) + rc = connack_rc; + else + rc = FAILURE; + } + else + rc = FAILURE; + +#if MQTTCLIENT_QOS2 + // resend an inflight publish + if (inflightMsgid >0 && inflightQoS == QOS2 && pubrel) + { + if ((len = MQTTSerialize_ack(sendbuf, MAX_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_PACKET_SIZE); + rc = publish(inflightLen, connect_timer, inflightQoS); + } +#endif + +exit: + if (rc == SUCCESS) + isconnected = true; + return rc; +} + + +template +int MQTTSN::Client::connect() +{ + MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer; + return connect(default_options); +} + + +template +int MQTTSN::Client::subscribe(MQTTSN_topicid& topicFilter, enum QoS qos, messageHandler messageHandler) +{ + int rc = FAILURE; + Timer timer = Timer(command_timeout_ms); + int len = 0; + + if (!isconnected) + goto exit; + + bool freeHandler = false; + for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i) + { + if (messageHandlers[i].topicFilter == 0) + { + freeHandler = true; + break; + } + } + if (!freeHandler) + { // No message handler free + rc = MAX_SUBSCRIPTIONS_EXCEEDED; + goto exit; + } + + len = MQTTSNSerialize_subscribe(sendbuf, MAX_PACKET_SIZE, 0, qos, packetid.getNext(), &topicFilter); + if (len <= 0) + goto exit; + if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet + goto exit; // there was a problem + + if (waitfor(MQTTSN_SUBACK, timer) == MQTTSN_SUBACK) // wait for suback + { + int grantedQoS = -1; + unsigned short mypacketid; + unsigned char rc; + if (MQTTSNDeserialize_suback(&grantedQoS, &topicFilter.data.id, &mypacketid, &rc, readbuf, MAX_PACKET_SIZE) == 1) + rc = grantedQoS; + if (rc == MQTTSN_RC_ACCEPTED) + { + for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i) + { + if (messageHandlers[i].topicFilter == 0) + { + messageHandlers[i].topicFilter = &topicFilter; + messageHandlers[i].fp.attach(messageHandler); + rc = 0; + break; + } + } + } + } + else + rc = FAILURE; + +exit: + if (rc != SUCCESS) + isconnected = false; + return rc; +} + + +template +int MQTTSN::Client::unsubscribe(MQTTSN_topicid& topicFilter) +{ + int rc = FAILURE; + Timer timer = Timer(command_timeout_ms); + int len = 0; + + if (!isconnected) + goto exit; + + if ((len = MQTTSNSerialize_unsubscribe(sendbuf, MAX_PACKET_SIZE, packetid.getNext(), &topicFilter)) <= 0) + goto exit; + if ((rc = sendPacket(len, timer)) != SUCCESS) // send the unsubscribe packet + goto exit; // there was a problem + + if (waitfor(MQTTSN_UNSUBACK, timer) == MQTTSN_UNSUBACK) + { + unsigned short mypacketid; // should be the same as the packetid above + if (MQTTSNDeserialize_unsuback(&mypacketid, readbuf, MAX_PACKET_SIZE) == 1) + rc = 0; + } + else + rc = FAILURE; + +exit: + if (rc != SUCCESS) + isconnected = false; + return rc; +} + + +template +int MQTTSN::Client::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(MQTTSN_PUBACK, timer) == MQTTSN_PUBACK) + { + unsigned short mypacketid; + unsigned char type; + if (MQTTSNDeserialize_ack(&type, &mypacketid, readbuf, MAX_PACKET_SIZE) != 1) + rc = FAILURE; + else if (inflightMsgid == mypacketid) + inflightMsgid = 0; + } + else + rc = FAILURE; + } +#elif MQTTCLIENT_QOS2 + else if (qos == QOS2) + { + if (waitfor(PUBCOMP, timer) == PUBCOMP) + { + unsigned short mypacketid; + unsigned char type; + if (MQTTDeserialize_ack(&type, &mypacketid, readbuf, MAX_PACKET_SIZE) != 1) + rc = FAILURE; + else if (inflightMsgid == mypacketid) + inflightMsgid = 0; + } + else + rc = FAILURE; + } +#endif + +exit: + if (rc != SUCCESS) + isconnected = false; + return rc; +} + + + +template +int MQTTSN::Client::publish(MQTTSN_topicid& topic, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos, bool retained) +{ + int rc = FAILURE; + Timer timer = Timer(command_timeout_ms); + int len = 0; + + if (!isconnected) + goto exit; + +#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 + if (qos == QOS1 || qos == QOS2) + id = packetid.getNext(); +#endif + + len = MQTTSNSerialize_publish(sendbuf, MAX_PACKET_SIZE, 0, qos, retained, id, + topic, (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 +int MQTTSN::Client::publish(MQTTSN_topicid& 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 +int MQTTSN::Client::publish(MQTTSN_topicid& topicName, Message& message) +{ + return publish(topicName, message.payload, message.payloadlen, message.qos, message.retained); +} + + +template +int MQTTSN::Client::disconnect(unsigned short duration) +{ + int rc = FAILURE; + Timer timer = Timer(command_timeout_ms); // we might wait for incomplete incoming publishes to complete + int int_duration = (duration == 0) ? -1 : (int)duration; + int len = MQTTSNSerialize_disconnect(sendbuf, MAX_PACKET_SIZE, int_duration); + if (len > 0) + rc = sendPacket(len, timer); // send the disconnect packet + + isconnected = false; + return rc; +} + + +#endif + diff --git a/MQTTSNClient/src/linux/linux.cpp b/MQTTSNClient/src/linux/linux.cpp new file mode 100644 index 0000000..222acda --- /dev/null +++ b/MQTTSNClient/src/linux/linux.cpp @@ -0,0 +1,224 @@ +/******************************************************************************* + * Copyright (c) 2014, 2015 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 +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + + +class IPStack +{ +public: + IPStack() + { + + } + + int Socket_error(const char* aString) + { + int rc = 0; + //if (errno != EINTR && errno != EAGAIN && errno != EINPROGRESS && errno != EWOULDBLOCK) + //{ + if (strcmp(aString, "shutdown") != 0 || (errno != ENOTCONN && errno != ECONNRESET)) + { + if (errno != EINTR && errno != EAGAIN && errno != EINPROGRESS && errno != EWOULDBLOCK) + printf("Socket error %s in %s for socket %d\n", strerror(errno), aString, mysock); + rc = errno; + } + //} + return errno; + } + + int connect(const char* hostname, int port) + { + int type = SOCK_STREAM; + struct sockaddr_in address; + int rc = -1; + sa_family_t family = AF_INET; + struct addrinfo *result = NULL; + struct addrinfo hints = {0, AF_UNSPEC, SOCK_STREAM, IPPROTO_TCP, 0, NULL, NULL, NULL}; + + if ((rc = getaddrinfo(hostname, NULL, &hints, &result)) == 0) + { + struct addrinfo* res = result; + + /* prefer ip4 addresses */ + while (res) + { + if (res->ai_family == AF_INET) + { + result = res; + break; + } + res = res->ai_next; + } + + if (result->ai_family == AF_INET) + { + address.sin_port = htons(port); + address.sin_family = family = AF_INET; + address.sin_addr = ((struct sockaddr_in*)(result->ai_addr))->sin_addr; + } + else + rc = -1; + + freeaddrinfo(result); + } + + if (rc == 0) + { + mysock = socket(family, type, 0); + if (mysock != -1) + { + int opt = 1; + + //if (setsockopt(mysock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0) + // printf("Could not set SO_NOSIGPIPE for socket %d", mysock); + + rc = ::connect(mysock, (struct sockaddr*)&address, sizeof(address)); + } + } + + return rc; + } + + int read(unsigned char* buffer, int len, int timeout_ms) + { + struct timeval interval = {timeout_ms / 1000, (timeout_ms % 1000) * 1000}; + if (interval.tv_sec < 0 || (interval.tv_sec == 0 && interval.tv_usec <= 0)) + { + interval.tv_sec = 0; + interval.tv_usec = 100; + } + + setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO, (char *)&interval, sizeof(struct timeval)); + + int bytes = 0; + while (bytes < len) + { + int rc = ::recv(mysock, &buffer[bytes], (size_t)(len - bytes), 0); + if (rc == -1) + { + if (Socket_error("read") != 0) + { + bytes = -1; + break; + } + } + else + bytes += rc; + } + return bytes; + } + + int write(unsigned char* buffer, int len, int timeout) + { + struct timeval tv; + + tv.tv_sec = 0; /* 30 Secs Timeout */ + tv.tv_usec = timeout * 1000; // Not init'ing this can cause strange errors + + setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv,sizeof(struct timeval)); + int rc = ::write(mysock, buffer, len); + //printf("write rc %d\n", rc); + return rc; + } + + int disconnect() + { + return ::close(mysock); + } + +private: + + int mysock; + +}; + + +class Countdown +{ +public: + Countdown() + { + + } + + Countdown(int ms) + { + countdown_ms(ms); + } + + + bool expired() + { + struct timeval now, res; + gettimeofday(&now, NULL); + timersub(&end_time, &now, &res); + //printf("left %d ms\n", (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000); + //if (res.tv_sec > 0 || res.tv_usec > 0) + // printf("expired %d %d\n", res.tv_sec, res.tv_usec); + return res.tv_sec < 0 || (res.tv_sec == 0 && res.tv_usec <= 0); + } + + + void countdown_ms(int ms) + { + struct timeval now; + gettimeofday(&now, NULL); + struct timeval interval = {ms / 1000, (ms % 1000) * 1000}; + //printf("interval %d %d\n", interval.tv_sec, interval.tv_usec); + timeradd(&now, &interval, &end_time); + } + + + void countdown(int seconds) + { + struct timeval now; + gettimeofday(&now, NULL); + struct timeval interval = {seconds, 0}; + timeradd(&now, &interval, &end_time); + } + + + int left_ms() + { + struct timeval now, res; + gettimeofday(&now, NULL); + timersub(&end_time, &now, &res); + //printf("left %d ms\n", (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000); + return (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000; + } + +private: + + struct timeval end_time; +}; +