james kang
Provides Javascript wrappers for MQTT.
Dependencies: mbed-http DEVI2C_JS MQTTPacket FP
Revision 0:f4dbe435e64c, committed 2018-01-17
- Comitter:
- akhtar.syedzeeshan@gmail.com
- Date:
- Wed Jan 17 11:30:51 2018 +0100
- Child:
- 1:a8ffffd1beab
- Commit message:
- First release
Changed in this revision
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTT/FP.lib Wed Jan 17 11:30:51 2018 +0100 @@ -0,0 +1,1 @@ +http://mbed.org/users/sam_grove/code/FP/#3c62ba1807ac
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT/MQTTAsync.h Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,607 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ * Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#if !defined(MQTTASYNC_H)
+#define MQTTASYNC_H
+
+#include "FP.h"
+#include "MQTTPacket.h"
+#include "stdio.h"
+
+namespace MQTT
+{
+
+
+enum QoS { QOS0, QOS1, QOS2 };
+
+
+struct Message
+{
+ enum QoS qos;
+ bool retained;
+ bool dup;
+ unsigned short id;
+ void *payload;
+ size_t payloadlen;
+};
+
+
+class PacketId
+{
+public:
+ PacketId();
+
+ int getNext();
+
+private:
+ static const int MAX_PACKET_ID = 65535;
+ int next;
+};
+
+typedef void (*messageHandler)(Message*);
+
+typedef struct limits
+{
+ int MAX_MQTT_PACKET_SIZE; //
+ int MAX_MESSAGE_HANDLERS; // each subscription requires a message handler
+ int MAX_CONCURRENT_OPERATIONS; // each command which runs concurrently can have a result handler, when we are in multi-threaded mode
+ int command_timeout_ms;
+
+ limits()
+ {
+ MAX_MQTT_PACKET_SIZE = 100;
+ MAX_MESSAGE_HANDLERS = 5;
+ MAX_CONCURRENT_OPERATIONS = 1; // 1 indicates single-threaded mode - set to >1 for multithreaded mode
+ command_timeout_ms = 30000;
+ }
+} Limits;
+
+
+/**
+ * @class Async
+ * @brief non-blocking, threaded MQTT client API
+ * @param Network a network class which supports send, receive
+ * @param Timer a timer class with the methods:
+ */
+template<class Network, class Timer, class Thread, class Mutex> class Async
+{
+
+public:
+
+ struct Result
+ {
+ /* success or failure result data */
+ Async<Network, Timer, Thread, Mutex>* client;
+ int rc;
+ };
+
+ typedef void (*resultHandler)(Result*);
+
+ Async(Network* network, const Limits limits = Limits());
+
+ typedef struct
+ {
+ Async* client;
+ Network* network;
+ } connectionLostInfo;
+
+ typedef int (*connectionLostHandlers)(connectionLostInfo*);
+
+ /** Set the connection lost callback - called whenever the connection is lost and we should be connected
+ * @param clh - pointer to the callback function
+ */
+ void setConnectionLostHandler(connectionLostHandlers clh)
+ {
+ connectionLostHandler.attach(clh);
+ }
+
+ /** Set the default message handling callback - used for any message which does not match a subscription message handler
+ * @param mh - pointer to the callback function
+ */
+ void setDefaultMessageHandler(messageHandler mh)
+ {
+ defaultMessageHandler.attach(mh);
+ }
+
+ int connect(resultHandler fn, MQTTPacket_connectData* options = 0);
+
+ template<class T>
+ int connect(void(T::*method)(Result *), MQTTPacket_connectData* options = 0, T *item = 0); // alternative to pass in pointer to member function
+
+ int publish(resultHandler rh, const char* topic, Message* message);
+
+ int subscribe(resultHandler rh, const char* topicFilter, enum QoS qos, messageHandler mh);
+
+ int unsubscribe(resultHandler rh, const char* topicFilter);
+
+ int disconnect(resultHandler rh);
+
+private:
+
+ void run(void const *argument);
+ int cycle(int timeout);
+ int waitfor(int packet_type, Timer& atimer);
+ int keepalive();
+ int findFreeOperation();
+
+ int decodePacket(int* value, int timeout);
+ int readPacket(int timeout);
+ int sendPacket(int length, int timeout);
+ int deliverMessage(MQTTString* topic, Message* message);
+
+ Thread* thread;
+ Network* ipstack;
+
+ Limits limits;
+
+ char* buf;
+ char* readbuf;
+
+ Timer ping_timer, connect_timer;
+ unsigned int keepAliveInterval;
+ bool ping_outstanding;
+
+ PacketId packetid;
+
+ typedef FP<void, Result*> resultHandlerFP;
+ resultHandlerFP connectHandler;
+
+ typedef FP<void, Message*> messageHandlerFP;
+ struct MessageHandlers
+ {
+ const char* topic;
+ messageHandlerFP fp;
+ } *messageHandlers; // Message handlers are indexed by subscription topic
+
+ // how many concurrent operations should we allow? Each one will require a function pointer
+ struct Operations
+ {
+ unsigned short id;
+ resultHandlerFP fp;
+ const char* topic; // if this is a publish, store topic name in case republishing is required
+ Message* message; // for publish,
+ Timer timer; // to check if the command has timed out
+ } *operations; // result handlers are indexed by packet ids
+
+ static void threadfn(void* arg);
+
+ messageHandlerFP defaultMessageHandler;
+
+ typedef FP<int, connectionLostInfo*> connectionLostFP;
+
+ connectionLostFP connectionLostHandler;
+
+};
+
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> void MQTT::Async<Network, Timer, Thread, Mutex>::threadfn(void* arg)
+{
+ ((Async<Network, Timer, Thread, Mutex>*) arg)->run(NULL);
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> MQTT::Async<Network, Timer, Thread, Mutex>::Async(Network* network, Limits limits) : limits(limits), packetid()
+{
+ this->thread = 0;
+ this->ipstack = network;
+ this->ping_timer = Timer();
+ this->ping_outstanding = 0;
+
+ // How to make these memory allocations portable? I was hoping to avoid the heap
+ buf = new char[limits.MAX_MQTT_PACKET_SIZE];
+ readbuf = new char[limits.MAX_MQTT_PACKET_SIZE];
+ this->operations = new struct Operations[limits.MAX_CONCURRENT_OPERATIONS];
+ for (int i = 0; i < limits.MAX_CONCURRENT_OPERATIONS; ++i)
+ operations[i].id = 0;
+ this->messageHandlers = new struct MessageHandlers[limits.MAX_MESSAGE_HANDLERS];
+ for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i)
+ messageHandlers[i].topic = 0;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::sendPacket(int length, int timeout)
+{
+ int sent = 0;
+
+ while (sent < length)
+ sent += ipstack->write(&buf[sent], length, timeout);
+ if (sent == length)
+ ping_timer.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet
+ return sent;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::decodePacket(int* value, int timeout)
+{
+ char c;
+ int multiplier = 1;
+ int len = 0;
+ const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4;
+
+ *value = 0;
+ do
+ {
+ int rc = MQTTPACKET_READ_ERROR;
+
+ if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
+ {
+ rc = MQTTPACKET_READ_ERROR; /* bad data */
+ goto exit;
+ }
+ rc = ipstack->read(&c, 1, timeout);
+ if (rc != 1)
+ goto exit;
+ *value += (c & 127) * multiplier;
+ multiplier *= 128;
+ } while ((c & 128) != 0);
+exit:
+ return len;
+}
+
+
+/**
+ * If any read fails in this method, then we should disconnect from the network, as on reconnect
+ * the packets can be retried.
+ * @param timeout the max time to wait for the packet read to complete, in milliseconds
+ * @return the MQTT packet type, or -1 if none
+ */
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::readPacket(int timeout)
+{
+ int rc = -1;
+ MQTTHeader header = {0};
+ int len = 0;
+ int rem_len = 0;
+
+ /* 1. read the header byte. This has the packet type in it */
+ if (ipstack->read(readbuf, 1, timeout) != 1)
+ goto exit;
+
+ len = 1;
+ /* 2. read the remaining length. This is variable in itself */
+ decodePacket(&rem_len, timeout);
+ len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length back into the buffer */
+
+ /* 3. read the rest of the buffer using a callback to supply the rest of the data */
+ if (ipstack->read(readbuf + len, rem_len, timeout) != rem_len)
+ goto exit;
+
+ header.byte = readbuf[0];
+ rc = header.bits.type;
+exit:
+ return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::deliverMessage(MQTTString* topic, Message* message)
+{
+ int rc = -1;
+
+ // we have to find the right message handler - indexed by topic
+ for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i)
+ {
+ if (messageHandlers[i].topic != 0 && MQTTPacket_equals(topic, (char*)messageHandlers[i].topic))
+ {
+ messageHandlers[i].fp(message);
+ rc = 0;
+ break;
+ }
+ }
+
+ return rc;
+}
+
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::cycle(int timeout)
+{
+ /* get one piece of work off the wire and one pass through */
+
+ // read the socket, see what work is due
+ int packet_type = readPacket(timeout);
+
+ int len, rc;
+ switch (packet_type)
+ {
+ case CONNACK:
+ if (this->thread)
+ {
+ Result res = {this, 0};
+ if (MQTTDeserialize_connack(&res.rc, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+ ;
+ connectHandler(&res);
+ connectHandler.detach(); // only invoke the callback once
+ }
+ break;
+ case PUBACK:
+ if (this->thread)
+ ; //call resultHandler
+ case SUBACK:
+ break;
+ case PUBLISH:
+ MQTTString topicName;
+ Message msg;
+ rc = MQTTDeserialize_publish((int*)&msg.dup, (int*)&msg.qos, (int*)&msg.retained, (int*)&msg.id, &topicName,
+ (char**)&msg.payload, (int*)&msg.payloadlen, readbuf, limits.MAX_MQTT_PACKET_SIZE);;
+ if (msg.qos == QOS0)
+ deliverMessage(&topicName, &msg);
+ break;
+ case PUBREC:
+ int type, dup, mypacketid;
+ if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+ ;
+ // must lock this access against the application thread, if we are multi-threaded
+ len = MQTTSerialize_ack(buf, limits.MAX_MQTT_PACKET_SIZE, PUBREL, 0, mypacketid);
+ rc = sendPacket(len, timeout); // send the PUBREL packet
+ if (rc != len)
+ goto exit; // there was a problem
+
+ break;
+ case PUBCOMP:
+ break;
+ case PINGRESP:
+ ping_outstanding = false;
+ break;
+ }
+ keepalive();
+exit:
+ return packet_type;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::keepalive()
+{
+ int rc = 0;
+
+ if (keepAliveInterval == 0)
+ goto exit;
+
+ if (ping_timer.expired())
+ {
+ if (ping_outstanding)
+ rc = -1;
+ else
+ {
+ int len = MQTTSerialize_pingreq(buf, limits.MAX_MQTT_PACKET_SIZE);
+ rc = sendPacket(len, 1000); // send the ping packet
+ if (rc != len)
+ rc = -1; // indicate there's a problem
+ else
+ ping_outstanding = true;
+ }
+ }
+
+exit:
+ return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> void MQTT::Async<Network, Timer, Thread, Mutex>::run(void const *argument)
+{
+ while (true)
+ cycle(ping_timer.left_ms());
+}
+
+
+// only used in single-threaded mode where one command at a time is in process
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::waitfor(int packet_type, Timer& atimer)
+{
+ int rc = -1;
+
+ do
+ {
+ if (atimer.expired())
+ break; // we timed out
+ }
+ while ((rc = cycle(atimer.left_ms())) != packet_type);
+
+ return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::connect(resultHandler resultHandler, MQTTPacket_connectData* options)
+{
+ connect_timer.countdown(limits.command_timeout_ms);
+
+ MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;
+ if (options == 0)
+ options = &default_options; // set default options if none were supplied
+
+ this->keepAliveInterval = options->keepAliveInterval;
+ ping_timer.countdown(this->keepAliveInterval);
+ int len = MQTTSerialize_connect(buf, limits.MAX_MQTT_PACKET_SIZE, options);
+ int rc = sendPacket(len, connect_timer.left_ms()); // send the connect packet
+ if (rc != len)
+ goto exit; // there was a problem
+
+ if (resultHandler == 0) // wait until the connack is received
+ {
+ // this will be a blocking call, wait for the connack
+ if (waitfor(CONNACK, connect_timer) == CONNACK)
+ {
+ int connack_rc = -1;
+ if (MQTTDeserialize_connack(&connack_rc, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+ rc = connack_rc;
+ }
+ }
+ else
+ {
+ // set connect response callback function
+ connectHandler.attach(resultHandler);
+
+ // start background thread
+ this->thread = new Thread((void (*)(void const *argument))&MQTT::Async<Network, Timer, Thread, Mutex>::threadfn, (void*)this);
+ }
+
+exit:
+ return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::findFreeOperation()
+{
+ int found = -1;
+ for (int i = 0; i < limits.MAX_CONCURRENT_OPERATIONS; ++i)
+ {
+ if (operations[i].id == 0)
+ {
+ found = i;
+ break;
+ }
+ }
+ return found;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::subscribe(resultHandler resultHandler, const char* topicFilter, enum QoS qos, messageHandler messageHandler)
+{
+ int index = 0;
+ if (this->thread)
+ index = findFreeOperation();
+ Timer& atimer = operations[index].timer;
+
+ atimer.countdown(limits.command_timeout_ms);
+ MQTTString topic = {(char*)topicFilter, 0, 0};
+
+ int len = MQTTSerialize_subscribe(buf, limits.MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos);
+ int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet
+ if (rc != len)
+ goto exit; // there was a problem
+
+ /* wait for suback */
+ if (resultHandler == 0)
+ {
+ // this will block
+ if (waitfor(SUBACK, atimer) == SUBACK)
+ {
+ int count = 0, grantedQoS = -1, mypacketid;
+ if (MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+ rc = grantedQoS; // 0, 1, 2 or 0x80
+ if (rc != 0x80)
+ {
+ for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i)
+ {
+ if (messageHandlers[i].topic == 0)
+ {
+ messageHandlers[i].topic = topicFilter;
+ messageHandlers[i].fp.attach(messageHandler);
+ rc = 0;
+ break;
+ }
+ }
+ }
+ }
+ }
+ else
+ {
+ // set subscribe response callback function
+
+ }
+
+exit:
+ return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::unsubscribe(resultHandler resultHandler, const char* topicFilter)
+{
+ int index = 0;
+ if (this->thread)
+ index = findFreeOperation();
+ Timer& atimer = operations[index].timer;
+
+ atimer.countdown(limits.command_timeout_ms);
+ MQTTString topic = {(char*)topicFilter, 0, 0};
+
+ int len = MQTTSerialize_unsubscribe(buf, limits.MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic);
+ int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet
+ if (rc != len)
+ goto exit; // there was a problem
+
+ // set unsubscribe response callback function
+
+
+exit:
+ return rc;
+}
+
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::publish(resultHandler resultHandler, const char* topicName, Message* message)
+{
+ int index = 0;
+ if (this->thread)
+ index = findFreeOperation();
+ Timer& atimer = operations[index].timer;
+
+ atimer.countdown(limits.command_timeout_ms);
+ MQTTString topic = {(char*)topicName, 0, 0};
+
+ if (message->qos == QOS1 || message->qos == QOS2)
+ message->id = packetid.getNext();
+
+ int len = MQTTSerialize_publish(buf, limits.MAX_MQTT_PACKET_SIZE, 0, message->qos, message->retained, message->id, topic, (char*)message->payload, message->payloadlen);
+ int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet
+ if (rc != len)
+ goto exit; // there was a problem
+
+ /* wait for acks */
+ if (resultHandler == 0)
+ {
+ if (message->qos == QOS1)
+ {
+ if (waitfor(PUBACK, atimer) == PUBACK)
+ {
+ int type, dup, mypacketid;
+ if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+ rc = 0;
+ }
+ }
+ else if (message->qos == QOS2)
+ {
+ if (waitfor(PUBCOMP, atimer) == PUBCOMP)
+ {
+ int type, dup, mypacketid;
+ if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1)
+ rc = 0;
+ }
+
+ }
+ }
+ else
+ {
+ // set publish response callback function
+
+ }
+
+exit:
+ return rc;
+}
+
+
+template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::disconnect(resultHandler resultHandler)
+{
+ Timer timer = Timer(limits.command_timeout_ms); // we might wait for incomplete incoming publishes to complete
+ int len = MQTTSerialize_disconnect(buf, limits.MAX_MQTT_PACKET_SIZE);
+ int rc = sendPacket(len, timer.left_ms()); // send the disconnect packet
+
+ return (rc == len) ? 0 : -1;
+}
+
+
+
+#endif
\ No newline at end of file
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT/MQTTClient.h Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,1052 @@
+/*******************************************************************************
+ * Copyright (c) 2014, 2017 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ * Ian Craggs - initial API and implementation and/or initial documentation
+ * Ian Craggs - fix for bug 458512 - QoS 2 messages
+ * Ian Craggs - fix for bug 460389 - send loop uses wrong length
+ * Ian Craggs - fix for bug 464169 - clearing subscriptions
+ * Ian Craggs - fix for bug 464551 - enums and ints can be different size
+ * Mark Sonnentag - fix for bug 475204 - inefficient instantiation of Timer
+ * Ian Craggs - fix for bug 475749 - packetid modified twice
+ * Ian Craggs - add ability to set message handler separately #6
+ *******************************************************************************/
+
+#if !defined(MQTTCLIENT_H)
+#define MQTTCLIENT_H
+
+#include "FP.h"
+#include "MQTTPacket.h"
+#include <stdio.h>
+#include "MQTTLogging.h"
+
+#if !defined(MQTTCLIENT_QOS1)
+ #define MQTTCLIENT_QOS1 1
+#endif
+#if !defined(MQTTCLIENT_QOS2)
+ #define MQTTCLIENT_QOS2 0
+#endif
+
+namespace MQTT
+{
+
+
+enum QoS { QOS0, QOS1, QOS2 };
+
+// all failure return codes must be negative
+enum returnCode { BUFFER_OVERFLOW = -2, FAILURE = -1, SUCCESS = 0 };
+
+
+struct Message
+{
+ enum QoS qos;
+ bool retained;
+ bool dup;
+ unsigned short id;
+ void *payload;
+ size_t payloadlen;
+};
+
+
+struct MessageData
+{
+ MessageData(MQTTString &aTopicName, struct Message &aMessage) : message(aMessage), topicName(aTopicName)
+ { }
+
+ struct Message &message;
+ MQTTString &topicName;
+};
+
+
+struct connackData
+{
+ int rc;
+ bool sessionPresent;
+};
+
+
+struct subackData
+{
+ int grantedQoS;
+};
+
+
+class PacketId
+{
+public:
+ PacketId()
+ {
+ next = 0;
+ }
+
+ int getNext()
+ {
+ return next = (next == MAX_PACKET_ID) ? 1 : next + 1;
+ }
+
+private:
+ static const int MAX_PACKET_ID = 65535;
+ int next;
+};
+
+
+/**
+ * @class Client
+ * @brief blocking, non-threaded MQTT client API
+ *
+ * This version of the API blocks on all method calls, until they are complete. This means that only one
+ * MQTT request can be in process at any one time.
+ * @param Network a network class which supports send, receive
+ * @param Timer a timer class with the methods:
+ */
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE = 100, int MAX_MESSAGE_HANDLERS = 5>
+class Client
+{
+
+public:
+
+ typedef void (*messageHandler)(MessageData&);
+
+ /** Construct the client
+ * @param network - pointer to an instance of the Network class - must be connected to the endpoint
+ * before calling MQTT connect
+ * @param limits an instance of the Limit class - to alter limits as required
+ */
+ Client(Network& network, unsigned int command_timeout_ms = 30000);
+
+ /** Set the default message handling callback - used for any message which does not match a subscription message handler
+ * @param mh - pointer to the callback function. Set to 0 to remove.
+ */
+ void setDefaultMessageHandler(messageHandler mh)
+ {
+ if (mh != 0)
+ defaultMessageHandler.attach(mh);
+ else
+ defaultMessageHandler.detach();
+ }
+
+ /** Set a message handling callback. This can be used outside of the the subscribe method.
+ * @param topicFilter - a topic pattern which can include wildcards
+ * @param mh - pointer to the callback function. If 0, removes the callback if any
+ */
+ int setMessageHandler(const char* topicFilter, messageHandler mh);
+
+ /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
+ * The nework object must be connected to the network endpoint before calling this
+ * Default connect options are used
+ * @return success code -
+ */
+ int connect();
+
+ /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
+ * The nework object must be connected to the network endpoint before calling this
+ * @param options - connect options
+ * @return success code -
+ */
+ int connect(MQTTPacket_connectData& options);
+
+ /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
+ * The nework object must be connected to the network endpoint before calling this
+ * @param options - connect options
+ * @param connackData - connack data to be returned
+ * @return success code -
+ */
+ int connect(MQTTPacket_connectData& options, connackData& data);
+
+ /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
+ * @param topic - the topic to publish to
+ * @param message - the message to send
+ * @return success code -
+ */
+ int publish(const char* topicName, Message& message);
+
+ /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
+ * @param topic - the topic to publish to
+ * @param payload - the data to send
+ * @param payloadlen - the length of the data
+ * @param qos - the QoS to send the publish at
+ * @param retained - whether the message should be retained
+ * @return success code -
+ */
+ int publish(const char* topicName, void* payload, size_t payloadlen, enum QoS qos = QOS0, bool retained = false);
+
+ /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
+ * @param topic - the topic to publish to
+ * @param payload - the data to send
+ * @param payloadlen - the length of the data
+ * @param id - the packet id used - returned
+ * @param qos - the QoS to send the publish at
+ * @param retained - whether the message should be retained
+ * @return success code -
+ */
+ int publish(const char* topicName, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos = QOS1, bool retained = false);
+
+ /** MQTT Subscribe - send an MQTT subscribe packet and wait for the suback
+ * @param topicFilter - a topic pattern which can include wildcards
+ * @param qos - the MQTT QoS to subscribe at
+ * @param mh - the callback function to be invoked when a message is received for this subscription
+ * @return success code -
+ */
+ int subscribe(const char* topicFilter, enum QoS qos, messageHandler mh);
+
+ /** MQTT Subscribe - send an MQTT subscribe packet and wait for the suback
+ * @param topicFilter - a topic pattern which can include wildcards
+ * @param qos - the MQTT QoS to subscribe at©
+ * @param mh - the callback function to be invoked when a message is received for this subscription
+ * @param
+ * @return success code -
+ */
+ int subscribe(const char* topicFilter, enum QoS qos, messageHandler mh, subackData &data);
+
+ /** MQTT Unsubscribe - send an MQTT unsubscribe packet and wait for the unsuback
+ * @param topicFilter - a topic pattern which can include wildcards
+ * @return success code -
+ */
+ int unsubscribe(const char* topicFilter);
+
+ /** MQTT Disconnect - send an MQTT disconnect packet, and clean up any state
+ * @return success code -
+ */
+ int disconnect();
+
+ /** A call to this API must be made within the keepAlive interval to keep the MQTT connection alive
+ * yield can be called if no other MQTT operation is needed. This will also allow messages to be
+ * received.
+ * @param timeout_ms the time to wait, in milliseconds
+ * @return success code - on failure, this means the client has disconnected
+ */
+ int yield(unsigned long timeout_ms = 1000L);
+
+ /** Is the client connected?
+ * @return flag - is the client connected or not?
+ */
+ bool isConnected()
+ {
+ return isconnected;
+ }
+
+private:
+
+ void closeSession();
+ void cleanSession();
+ int cycle(Timer& timer);
+ int waitfor(int packet_type, Timer& timer);
+ int keepalive();
+ int publish(int len, Timer& timer, enum QoS qos);
+
+ int decodePacket(int* value, int timeout);
+ int readPacket(Timer& timer);
+ int sendPacket(int length, Timer& timer);
+ int deliverMessage(MQTTString& topicName, Message& message);
+ bool isTopicMatched(char* topicFilter, MQTTString& topicName);
+
+ Network& ipstack;
+ unsigned long command_timeout_ms;
+
+ unsigned char sendbuf[MAX_MQTT_PACKET_SIZE];
+ unsigned char readbuf[MAX_MQTT_PACKET_SIZE];
+
+ Timer last_sent, last_received;
+ unsigned int keepAliveInterval;
+ bool ping_outstanding;
+ bool cleansession;
+
+ PacketId packetid;
+
+ struct MessageHandlers
+ {
+ const char* topicFilter;
+ FP<void, MessageData&> fp;
+ } messageHandlers[MAX_MESSAGE_HANDLERS]; // Message handlers are indexed by subscription topic
+
+ FP<void, MessageData&> defaultMessageHandler;
+
+ bool isconnected;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+ unsigned char pubbuf[MAX_MQTT_PACKET_SIZE]; // store the last publish for sending on reconnect
+ int inflightLen;
+ unsigned short inflightMsgid;
+ enum QoS inflightQoS;
+#endif
+
+#if MQTTCLIENT_QOS2
+ bool pubrel;
+ #if !defined(MAX_INCOMING_QOS2_MESSAGES)
+ #define MAX_INCOMING_QOS2_MESSAGES 10
+ #endif
+ unsigned short incomingQoS2messages[MAX_INCOMING_QOS2_MESSAGES];
+ bool isQoS2msgidFree(unsigned short id);
+ bool useQoS2msgid(unsigned short id);
+ void freeQoS2msgid(unsigned short id);
+#endif
+
+};
+
+}
+
+
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
+void MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::cleanSession()
+{
+ for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+ messageHandlers[i].topicFilter = 0;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+ inflightMsgid = 0;
+ inflightQoS = QOS0;
+#endif
+
+#if MQTTCLIENT_QOS2
+ pubrel = false;
+ for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+ incomingQoS2messages[i] = 0;
+#endif
+}
+
+
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
+void MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::closeSession()
+{
+ ping_outstanding = false;
+ isconnected = false;
+ if (cleansession)
+ cleanSession();
+}
+
+
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
+MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::Client(Network& network, unsigned int command_timeout_ms) : ipstack(network), packetid()
+{
+ this->command_timeout_ms = command_timeout_ms;
+ cleansession = true;
+ closeSession();
+}
+
+
+#if MQTTCLIENT_QOS2
+template<class Network, class Timer, int a, int b>
+bool MQTT::Client<Network, Timer, a, b>::isQoS2msgidFree(unsigned short id)
+{
+ for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+ {
+ if (incomingQoS2messages[i] == id)
+ return false;
+ }
+ return true;
+}
+
+
+template<class Network, class Timer, int a, int b>
+bool MQTT::Client<Network, Timer, a, b>::useQoS2msgid(unsigned short id)
+{
+ for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+ {
+ if (incomingQoS2messages[i] == 0)
+ {
+ incomingQoS2messages[i] = id;
+ return true;
+ }
+ }
+ return false;
+}
+
+
+template<class Network, class Timer, int a, int b>
+void MQTT::Client<Network, Timer, a, b>::freeQoS2msgid(unsigned short id)
+{
+ for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+ {
+ if (incomingQoS2messages[i] == id)
+ {
+ incomingQoS2messages[i] = 0;
+ return;
+ }
+ }
+}
+#endif
+
+
+template<class Network, class Timer, int a, int b>
+int MQTT::Client<Network, Timer, a, b>::sendPacket(int length, Timer& timer)
+{
+ int rc = FAILURE,
+ sent = 0;
+
+ while (sent < length)
+ {
+ rc = ipstack.write(&sendbuf[sent], length - sent, timer.left_ms());
+ if (rc < 0) // there was an error writing the data
+ break;
+ sent += rc;
+ if (timer.expired()) // only check expiry after at least one attempt to write
+ break;
+ }
+ if (sent == length)
+ {
+ if (this->keepAliveInterval > 0)
+ last_sent.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet
+ rc = SUCCESS;
+ }
+ else
+ rc = FAILURE;
+
+#if defined(MQTT_DEBUG)
+ char printbuf[150];
+ DEBUG("Rc %d from sending packet %s\r\n", rc,
+ MQTTFormat_toServerString(printbuf, sizeof(printbuf), sendbuf, length));
+#endif
+ return rc;
+}
+
+
+template<class Network, class Timer, int a, int b>
+int MQTT::Client<Network, Timer, a, b>::decodePacket(int* value, int timeout)
+{
+ unsigned char c;
+ int multiplier = 1;
+ int len = 0;
+ const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4;
+
+ *value = 0;
+ do
+ {
+ int rc = MQTTPACKET_READ_ERROR;
+
+ if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
+ {
+ rc = MQTTPACKET_READ_ERROR; /* bad data */
+ goto exit;
+ }
+ rc = ipstack.read(&c, 1, timeout);
+ if (rc != 1)
+ goto exit;
+ *value += (c & 127) * multiplier;
+ multiplier *= 128;
+ } while ((c & 128) != 0);
+exit:
+ return len;
+}
+
+
+/**
+ * If any read fails in this method, then we should disconnect from the network, as on reconnect
+ * the packets can be retried.
+ * @param timeout the max time to wait for the packet read to complete, in milliseconds
+ * @return the MQTT packet type, 0 if none, -1 if error
+ */
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::readPacket(Timer& timer)
+{
+ int rc = FAILURE;
+ MQTTHeader header = {0};
+ int len = 0;
+ int rem_len = 0;
+
+ /* 1. read the header byte. This has the packet type in it */
+ rc = ipstack.read(readbuf, 1, timer.left_ms());
+ if (rc != 1)
+ goto exit;
+
+ len = 1;
+ /* 2. read the remaining length. This is variable in itself */
+ decodePacket(&rem_len, timer.left_ms());
+ len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length into the buffer */
+
+ if (rem_len > (MAX_MQTT_PACKET_SIZE - len))
+ {
+ rc = BUFFER_OVERFLOW;
+ goto exit;
+ }
+
+ /* 3. read the rest of the buffer using a callback to supply the rest of the data */
+ if (rem_len > 0 && (ipstack.read(readbuf + len, rem_len, timer.left_ms()) != rem_len))
+ goto exit;
+
+ header.byte = readbuf[0];
+ rc = header.bits.type;
+ if (this->keepAliveInterval > 0)
+ last_received.countdown(this->keepAliveInterval); // record the fact that we have successfully received a packet
+exit:
+
+#if defined(MQTT_DEBUG)
+ if (rc >= 0)
+ {
+ char printbuf[50];
+ DEBUG("Rc %d receiving packet %s\r\n", rc,
+ MQTTFormat_toClientString(printbuf, sizeof(printbuf), readbuf, len));
+ }
+#endif
+ return rc;
+}
+
+
+// assume topic filter and name is in correct format
+// # can only be at end
+// + and # can only be next to separator
+template<class Network, class Timer, int a, int b>
+bool MQTT::Client<Network, Timer, a, b>::isTopicMatched(char* topicFilter, MQTTString& topicName)
+{
+ char* curf = topicFilter;
+ char* curn = topicName.lenstring.data;
+ char* curn_end = curn + topicName.lenstring.len;
+
+ while (*curf && curn < curn_end)
+ {
+ if (*curn == '/' && *curf != '/')
+ break;
+ if (*curf != '+' && *curf != '#' && *curf != *curn)
+ break;
+ if (*curf == '+')
+ { // skip until we meet the next separator, or end of string
+ char* nextpos = curn + 1;
+ while (nextpos < curn_end && *nextpos != '/')
+ nextpos = ++curn + 1;
+ }
+ else if (*curf == '#')
+ curn = curn_end - 1; // skip until end of string
+ curf++;
+ curn++;
+ };
+
+ return (curn == curn_end) && (*curf == '\0');
+}
+
+
+
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
+int MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::deliverMessage(MQTTString& topicName, Message& message)
+{
+ int rc = FAILURE;
+
+ // we have to find the right message handler - indexed by topic
+ for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+ {
+ if (messageHandlers[i].topicFilter != 0 && (MQTTPacket_equals(&topicName, (char*)messageHandlers[i].topicFilter) ||
+ isTopicMatched((char*)messageHandlers[i].topicFilter, topicName)))
+ {
+ if (messageHandlers[i].fp.attached())
+ {
+ MessageData md(topicName, message);
+ messageHandlers[i].fp(md);
+ rc = SUCCESS;
+ }
+ }
+ }
+
+ if (rc == FAILURE && defaultMessageHandler.attached())
+ {
+ MessageData md(topicName, message);
+ defaultMessageHandler(md);
+ rc = SUCCESS;
+ }
+
+ return rc;
+}
+
+
+
+template<class Network, class Timer, int a, int b>
+int MQTT::Client<Network, Timer, a, b>::yield(unsigned long timeout_ms)
+{
+ int rc = SUCCESS;
+ Timer timer;
+
+ timer.countdown_ms(timeout_ms);
+ while (!timer.expired())
+ {
+ if (cycle(timer) < 0)
+ {
+ rc = FAILURE;
+ break;
+ }
+ }
+
+ return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::cycle(Timer& timer)
+{
+ // get one piece of work off the wire and one pass through
+ int len = 0,
+ rc = SUCCESS;
+
+ int packet_type = readPacket(timer); // read the socket, see what work is due
+
+ switch (packet_type)
+ {
+ default:
+ // no more data to read, unrecoverable. Or read packet fails due to unexpected network error
+ rc = packet_type;
+ goto exit;
+ case NSAPI_ERROR_WOULD_BLOCK:
+ case NSAPI_ERROR_OK: // timed out reading packet
+ break;
+ case CONNACK:
+ case PUBACK:
+ case SUBACK:
+ break;
+ case PUBLISH:
+ {
+ MQTTString topicName = MQTTString_initializer;
+ Message msg;
+ int intQoS;
+ msg.payloadlen = 0; /* this is a size_t, but deserialize publish sets this as int */
+ if (MQTTDeserialize_publish((unsigned char*)&msg.dup, &intQoS, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicName,
+ (unsigned char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+ goto exit;
+ msg.qos = (enum QoS)intQoS;
+#if MQTTCLIENT_QOS2
+ if (msg.qos != QOS2)
+#endif
+ deliverMessage(topicName, msg);
+#if MQTTCLIENT_QOS2
+ else if (isQoS2msgidFree(msg.id))
+ {
+ if (useQoS2msgid(msg.id))
+ deliverMessage(topicName, msg);
+ else
+ WARN("Maximum number of incoming QoS2 messages exceeded");
+ }
+#endif
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+ if (msg.qos != QOS0)
+ {
+ if (msg.qos == QOS1)
+ len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBACK, 0, msg.id);
+ else if (msg.qos == QOS2)
+ len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREC, 0, msg.id);
+ if (len <= 0)
+ rc = FAILURE;
+ else
+ rc = sendPacket(len, timer);
+ if (rc == FAILURE)
+ goto exit; // there was a problem
+ }
+ break;
+#endif
+ }
+#if MQTTCLIENT_QOS2
+ case PUBREC:
+ case PUBREL:
+ unsigned short mypacketid;
+ unsigned char dup, type;
+ if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+ rc = FAILURE;
+ else if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE,
+ (packet_type == PUBREC) ? PUBREL : PUBCOMP, 0, mypacketid)) <= 0)
+ rc = FAILURE;
+ else if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet
+ rc = FAILURE; // there was a problem
+ if (rc == FAILURE)
+ goto exit; // there was a problem
+ if (packet_type == PUBREL)
+ freeQoS2msgid(mypacketid);
+ break;
+
+ case PUBCOMP:
+ break;
+#endif
+ case PINGRESP:
+ ping_outstanding = false;
+ break;
+ }
+
+ if (keepalive() != SUCCESS)
+ //check only keepalive FAILURE status so that previous FAILURE status can be considered as FAULT
+ rc = FAILURE;
+
+exit:
+ if (rc == SUCCESS)
+ rc = packet_type;
+ else if (isconnected)
+ closeSession();
+ return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::keepalive()
+{
+ int rc = SUCCESS;
+ static Timer ping_sent;
+
+ if (keepAliveInterval == 0)
+ goto exit;
+
+ if (ping_outstanding)
+ {
+ if (ping_sent.expired())
+ {
+ rc = FAILURE; // session failure
+ #if defined(MQTT_DEBUG)
+ DEBUG("PINGRESP not received in keepalive interval\r\n");
+ #endif
+ }
+ }
+ else if (last_sent.expired() || last_received.expired())
+ {
+ Timer timer(1000);
+ int len = MQTTSerialize_pingreq(sendbuf, MAX_MQTT_PACKET_SIZE);
+ if (len > 0 && (rc = sendPacket(len, timer)) == SUCCESS) // send the ping packet
+ {
+ ping_outstanding = true;
+ ping_sent.countdown(this->keepAliveInterval);
+ }
+ }
+exit:
+ return rc;
+}
+
+
+// only used in single-threaded mode where one command at a time is in process
+template<class Network, class Timer, int a, int b>
+int MQTT::Client<Network, Timer, a, b>::waitfor(int packet_type, Timer& timer)
+{
+ int rc = FAILURE;
+
+ do
+ {
+ if (timer.expired())
+ break; // we timed out
+ rc = cycle(timer);
+ }
+ while (rc != packet_type && rc >= 0);
+
+ return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect(MQTTPacket_connectData& options, connackData& data)
+{
+ Timer connect_timer(command_timeout_ms);
+ int rc = FAILURE;
+ int len = 0;
+
+ if (isconnected) // don't send connect packet again if we are already connected
+ goto exit;
+
+ this->keepAliveInterval = options.keepAliveInterval;
+ this->cleansession = options.cleansession;
+ if ((len = MQTTSerialize_connect(sendbuf, MAX_MQTT_PACKET_SIZE, &options)) <= 0)
+ goto exit;
+ if ((rc = sendPacket(len, connect_timer)) != SUCCESS) // send the connect packet
+ goto exit; // there was a problem
+
+ if (this->keepAliveInterval > 0)
+ last_received.countdown(this->keepAliveInterval);
+ // this will be a blocking call, wait for the connack
+ if (waitfor(CONNACK, connect_timer) == CONNACK)
+ {
+ data.rc = 0;
+ data.sessionPresent = false;
+ if (MQTTDeserialize_connack((unsigned char*)&data.sessionPresent,
+ (unsigned char*)&data.rc, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
+ rc = data.rc;
+ else
+ rc = FAILURE;
+ }
+ else
+ rc = FAILURE;
+
+#if MQTTCLIENT_QOS2
+ // resend any inflight publish
+ if (inflightMsgid > 0 && inflightQoS == QOS2 && pubrel)
+ {
+ if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, inflightMsgid)) <= 0)
+ rc = FAILURE;
+ else
+ rc = publish(len, connect_timer, inflightQoS);
+ }
+ else
+#endif
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+ if (inflightMsgid > 0)
+ {
+ memcpy(sendbuf, pubbuf, MAX_MQTT_PACKET_SIZE);
+ rc = publish(inflightLen, connect_timer, inflightQoS);
+ }
+#endif
+
+exit:
+ if (rc == SUCCESS)
+ {
+ isconnected = true;
+ ping_outstanding = false;
+ }
+ return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect(MQTTPacket_connectData& options)
+{
+ connackData data;
+ return connect(options, data);
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect()
+{
+ MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;
+ return connect(default_options);
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::setMessageHandler(const char* topicFilter, messageHandler messageHandler)
+{
+ int rc = FAILURE;
+ int i = -1;
+
+ // first check for an existing matching slot
+ for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+ {
+ if (messageHandlers[i].topicFilter != 0 && strcmp(messageHandlers[i].topicFilter, topicFilter) == 0)
+ {
+ if (messageHandler == 0) // remove existing
+ {
+ messageHandlers[i].topicFilter = 0;
+ messageHandlers[i].fp.detach();
+ }
+ rc = SUCCESS; // return i when adding new subscription
+ break;
+ }
+ }
+ // if no existing, look for empty slot (unless we are removing)
+ if (messageHandler != 0) {
+ if (rc == FAILURE)
+ {
+ for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+ {
+ if (messageHandlers[i].topicFilter == 0)
+ {
+ rc = SUCCESS;
+ break;
+ }
+ }
+ }
+ if (i < MAX_MESSAGE_HANDLERS)
+ {
+ messageHandlers[i].topicFilter = topicFilter;
+ messageHandlers[i].fp.attach(messageHandler);
+ }
+ }
+ return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::subscribe(const char* topicFilter,
+ enum QoS qos, messageHandler messageHandler, subackData& data)
+{
+ int rc = FAILURE;
+ Timer timer(command_timeout_ms);
+ int len = 0;
+ MQTTString topic = {(char*)topicFilter, {0, 0}};
+
+ if (!isconnected)
+ goto exit;
+
+ len = MQTTSerialize_subscribe(sendbuf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos);
+ if (len <= 0)
+ goto exit;
+ if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet
+ goto exit; // there was a problem
+
+ if (waitfor(SUBACK, timer) == SUBACK) // wait for suback
+ {
+ int count = 0;
+ unsigned short mypacketid;
+ data.grantedQoS = 0;
+ if (MQTTDeserialize_suback(&mypacketid, 1, &count, &data.grantedQoS, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
+ {
+ if (data.grantedQoS != 0x80)
+ rc = setMessageHandler(topicFilter, messageHandler);
+ }
+ }
+ else
+ rc = FAILURE;
+
+exit:
+ if (rc == FAILURE)
+ closeSession();
+ return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::subscribe(const char* topicFilter, enum QoS qos, messageHandler messageHandler)
+{
+ subackData data;
+ return subscribe(topicFilter, qos, messageHandler, data);
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::unsubscribe(const char* topicFilter)
+{
+ int rc = FAILURE;
+ Timer timer(command_timeout_ms);
+ MQTTString topic = {(char*)topicFilter, {0, 0}};
+ int len = 0;
+
+ if (!isconnected)
+ goto exit;
+
+ if ((len = MQTTSerialize_unsubscribe(sendbuf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic)) <= 0)
+ goto exit;
+ if ((rc = sendPacket(len, timer)) != SUCCESS) // send the unsubscribe packet
+ goto exit; // there was a problem
+
+ if (waitfor(UNSUBACK, timer) == UNSUBACK)
+ {
+ unsigned short mypacketid; // should be the same as the packetid above
+ if (MQTTDeserialize_unsuback(&mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
+ {
+ // remove the subscription message handler associated with this topic, if there is one
+ setMessageHandler(topicFilter, 0);
+ }
+ }
+ else
+ rc = FAILURE;
+
+exit:
+ if (rc != SUCCESS)
+ closeSession();
+ return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(int len, Timer& timer, enum QoS qos)
+{
+ int rc;
+
+ if ((rc = sendPacket(len, timer)) != SUCCESS) // send the publish packet
+ goto exit; // there was a problem
+
+#if MQTTCLIENT_QOS1
+ if (qos == QOS1)
+ {
+ if (waitfor(PUBACK, timer) == PUBACK)
+ {
+ unsigned short mypacketid;
+ unsigned char dup, type;
+ if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+ rc = FAILURE;
+ else if (inflightMsgid == mypacketid)
+ inflightMsgid = 0;
+ }
+ else
+ rc = FAILURE;
+ }
+#endif
+#if MQTTCLIENT_QOS2
+ else if (qos == QOS2)
+ {
+ if (waitfor(PUBCOMP, timer) == PUBCOMP)
+ {
+ unsigned short mypacketid;
+ unsigned char dup, type;
+ if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
+ rc = FAILURE;
+ else if (inflightMsgid == mypacketid)
+ inflightMsgid = 0;
+ }
+ else
+ rc = FAILURE;
+ }
+#endif
+
+exit:
+ if (rc != SUCCESS)
+ closeSession();
+ return rc;
+}
+
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos, bool retained)
+{
+ int rc = FAILURE;
+ Timer timer(command_timeout_ms);
+ MQTTString topicString = MQTTString_initializer;
+ int len = 0;
+
+ if (!isconnected)
+ goto exit;
+
+ topicString.cstring = (char*)topicName;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+ if (qos == QOS1 || qos == QOS2)
+ id = packetid.getNext();
+#endif
+
+ len = MQTTSerialize_publish(sendbuf, MAX_MQTT_PACKET_SIZE, 0, qos, retained, id,
+ topicString, (unsigned char*)payload, payloadlen);
+ if (len <= 0)
+ goto exit;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+ if (!cleansession)
+ {
+ memcpy(pubbuf, sendbuf, len);
+ inflightMsgid = id;
+ inflightLen = len;
+ inflightQoS = qos;
+#if MQTTCLIENT_QOS2
+ pubrel = false;
+#endif
+ }
+#endif
+
+ rc = publish(len, timer, qos);
+exit:
+ return rc;
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, void* payload, size_t payloadlen, enum QoS qos, bool retained)
+{
+ unsigned short id = 0; // dummy - not used for anything
+ return publish(topicName, payload, payloadlen, id, qos, retained);
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, Message& message)
+{
+ return publish(topicName, message.payload, message.payloadlen, message.qos, message.retained);
+}
+
+
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::disconnect()
+{
+ int rc = FAILURE;
+ Timer timer(command_timeout_ms); // we might wait for incomplete incoming publishes to complete
+ int len = MQTTSerialize_disconnect(sendbuf, MAX_MQTT_PACKET_SIZE);
+ if (len > 0)
+ rc = sendPacket(len, timer); // send the disconnect packet
+ closeSession();
+ return rc;
+}
+
+#endif
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT/MQTTEthernet.h Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,29 @@
+
+#if !defined(MQTTETHERNET_H)
+#define MQTTETHERNET_H
+
+#include "MQTTmbed.h"
+#include "EthernetInterface.h"
+#include "MQTTSocket.h"
+
+class MQTTEthernet : public MQTTSocket
+{
+public:
+ MQTTEthernet() : MQTTSocket(ð)
+ {
+ eth.connect();
+ }
+
+ EthernetInterface& getEth()
+ {
+ return eth;
+ }
+
+private:
+
+ EthernetInterface eth;
+
+};
+
+
+#endif
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT/MQTTLogging.h Wed Jan 17 11:30:51 2018 +0100
@@ -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/MQTT/MQTTNetwork.h Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,40 @@
+#ifndef _MQTTNETWORK_H_
+#define _MQTTNETWORK_H_
+
+#include "NetworkInterface.h"
+
+class MQTTNetwork {
+public:
+ MQTTNetwork(NetworkInterface* aNetwork) : network(aNetwork) {
+ socket = new TCPSocket();
+ }
+
+ ~MQTTNetwork() {
+ delete socket;
+ }
+
+ int read(unsigned char* buffer, int len, int timeout) {
+ socket->set_timeout(timeout);
+ return socket->recv(buffer, len);
+ }
+
+ int write(unsigned char* buffer, int len, int timeout) {
+ socket->set_timeout(timeout);
+ return socket->send(buffer, len);
+ }
+
+ int connect(const char* hostname, int port) {
+ socket->open(network);
+ return socket->connect(hostname, port);
+ }
+
+ int disconnect() {
+ return socket->close();
+ }
+
+private:
+ NetworkInterface* network;
+ TCPSocket* socket;
+};
+
+#endif // _MQTTNETWORK_H_
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/MQTT/MQTTPacket.lib Wed Jan 17 11:30:51 2018 +0100 @@ -0,0 +1,1 @@ +http://mbed.org/teams/mqtt/code/MQTTPacket/#aedcaf7984d5
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT/MQTTSocket.h Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,97 @@
+#if !defined(MQTTSOCKET_H)
+#define MQTTSOCKET_H
+
+#include "MQTTmbed.h"
+#include <EthernetInterface.h>
+#include <Timer.h>
+
+class MQTTSocket
+{
+public:
+ MQTTSocket(EthernetInterface *anet)
+ {
+ net = anet;
+ open = false;
+ }
+
+ int connect(char* hostname, int port, int timeout=1000)
+ {
+ if (open)
+ disconnect();
+ nsapi_error_t rc = mysock.open(net);
+ open = true;
+ mysock.set_blocking(true);
+ mysock.set_timeout((unsigned int)timeout);
+ rc = mysock.connect(hostname, port);
+ mysock.set_blocking(false); // blocking timeouts seem not to work
+ return rc;
+ }
+
+ // common read/write routine, avoiding blocking timeouts
+ int common(unsigned char* buffer, int len, int timeout, bool read)
+ {
+ timer.start();
+ mysock.set_blocking(false); // blocking timeouts seem not to work
+ int bytes = 0;
+ bool first = true;
+ do
+ {
+ if (first)
+ first = false;
+ else
+ wait_ms(timeout < 100 ? timeout : 100);
+ int rc;
+ if (read)
+ rc = mysock.recv((char*)buffer, len);
+ else
+ rc = mysock.send((char*)buffer, len);
+ if (rc < 0)
+ {
+ if (rc != NSAPI_ERROR_WOULD_BLOCK)
+ {
+ bytes = -1;
+ break;
+ }
+ }
+ else
+ bytes += rc;
+ }
+ while (bytes < len && timer.read_ms() < timeout);
+ timer.stop();
+ return bytes;
+ }
+
+ /* returns the number of bytes read, which could be 0.
+ -1 if there was an error on the socket
+ */
+ int read(unsigned char* buffer, int len, int timeout)
+ {
+ return common(buffer, len, timeout, true);
+ }
+
+ int write(unsigned char* buffer, int len, int timeout)
+ {
+ return common(buffer, len, timeout, false);
+ }
+
+ int disconnect()
+ {
+ open = false;
+ return mysock.close();
+ }
+
+ /*bool is_connected()
+ {
+ return mysock.is_connected();
+ }*/
+
+private:
+
+ bool open;
+ TCPSocket mysock;
+ EthernetInterface *net;
+ Timer timer;
+
+};
+
+#endif
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT/MQTTmbed.h Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,48 @@
+#if !defined(MQTT_MBED_H)
+#define MQTT_MBED_H
+
+#include "mbed.h"
+
+class Countdown
+{
+public:
+ Countdown() : t()
+ {
+
+ }
+
+ Countdown(int ms) : t()
+ {
+ countdown_ms(ms);
+ }
+
+
+ bool expired()
+ {
+ return t.read_ms() >= interval_end_ms;
+ }
+
+ void countdown_ms(unsigned long ms)
+ {
+ t.stop();
+ interval_end_ms = ms;
+ t.reset();
+ t.start();
+ }
+
+ void countdown(int seconds)
+ {
+ countdown_ms((unsigned long)seconds * 1000L);
+ }
+
+ int left_ms()
+ {
+ return interval_end_ms - t.read_ms();
+ }
+
+private:
+ Timer t;
+ unsigned long interval_end_ms;
+};
+
+#endif
\ No newline at end of file
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT_JS-js.cpp Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,344 @@
+/*
+ * @file MQTT_JS-js.cpp
+ * @author ST
+ * @version V1.0.0
+ * @date 9 October 2017
+ * @brief Implementation of MQTT for Javascript.
+ ******************************************************************************
+ * @attention
+ *
+ * <h2><center>© COPYRIGHT(c) 2017 STMicroelectronics</center></h2>
+ *
+ * Redistribution and use in source and binary forms, with or without modification,
+ * are permitted provided that the following conditions are met:
+ * 1. Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * 3. Neither the name of STMicroelectronics nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+ * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ ******************************************************************************
+ */
+
+/* Includes ------------------------------------------------------------------*/
+
+#include "jerryscript-mbed-library-registry/wrap_tools.h"
+#include "jerryscript-mbed-event-loop/EventLoop.h"
+
+#include "MQTT_JS.h"
+
+/* Class Implementation ------------------------------------------------------*/
+
+/**
+ * MQTT_JS#destructor
+ *
+ * Called if/when the MQTT_JS object is GC'ed.
+ */
+void NAME_FOR_CLASS_NATIVE_DESTRUCTOR(MQTT_JS) (void *void_ptr) {
+ delete static_cast<MQTT_JS*>(void_ptr);
+}
+
+/**
+ * Type infomation of the native MQTT_JS pointer
+ *
+ * Set MQTT_JS#destructor as the free callback.
+ */
+static const jerry_object_native_info_t native_obj_type_info = {
+ .free_cb = NAME_FOR_CLASS_NATIVE_DESTRUCTOR(MQTT_JS)
+};
+
+
+/**
+ * MQTT_JS#init (native JavaScript method)
+ *
+ * Initializes the MQTT service.
+ */
+DECLARE_CLASS_FUNCTION(MQTT_JS, init) {
+ CHECK_ARGUMENT_COUNT(MQTT_JS, init, (args_count == 4));
+ CHECK_ARGUMENT_TYPE_ALWAYS(MQTT_JS, init, 0, string);
+ CHECK_ARGUMENT_TYPE_ALWAYS(MQTT_JS, init, 1, string);
+ CHECK_ARGUMENT_TYPE_ALWAYS(MQTT_JS, init, 2, string);
+ CHECK_ARGUMENT_TYPE_ALWAYS(MQTT_JS, init, 3, string);
+
+ size_t id_length = jerry_get_string_length(args[0]);
+ size_t token_length = jerry_get_string_length(args[1]);
+ size_t url_length = jerry_get_string_length(args[2]);
+ size_t port_length = jerry_get_string_length(args[3]);
+
+ if(id_length > 32){
+ return jerry_create_number(1);
+ }
+ if(token_length > 32){
+ return jerry_create_number(2);
+ }
+ if(url_length > 128){
+ return jerry_create_number(3);
+ }
+ if(port_length > 16){
+ return jerry_create_number(4);
+ }
+
+ // add an extra character to ensure there's a null character after the device name
+ char* id = (char*)calloc(id_length + 1, sizeof(char));
+ jerry_string_to_char_buffer(args[0], (jerry_char_t*)id, id_length);
+ char* token = (char*)calloc(token_length + 1, sizeof(char));
+ jerry_string_to_char_buffer(args[1], (jerry_char_t*)token, token_length);
+ char* url = (char*)calloc(url_length + 1, sizeof(char));
+ jerry_string_to_char_buffer(args[2], (jerry_char_t*)url, url_length);
+ char* port = (char*)calloc(port_length + 1, sizeof(char));
+ jerry_string_to_char_buffer(args[3], (jerry_char_t*)port, port_length);
+
+ // Unwrap native MQTT_JS object
+ void *void_ptr;
+ const jerry_object_native_info_t *type_ptr;
+ bool has_ptr = jerry_get_object_native_pointer(this_obj, &void_ptr, &type_ptr);
+
+ if (!has_ptr || type_ptr != &native_obj_type_info) {
+ return jerry_create_error(JERRY_ERROR_TYPE,
+ (const jerry_char_t *) "Failed to get native MQTT_JS pointer");
+ }
+
+ MQTT_JS *native_ptr = static_cast<MQTT_JS*>(void_ptr);
+
+ //int ret = 0; //native_ptr->init();
+ NetworkInterface_JS::getInstance()->connect();
+
+ int res = native_ptr->init(NetworkInterface_JS::getInstance()->getNetworkInterface(),
+ id, token, url, port);
+
+ free(id);
+ free(token);
+ free(url);
+ free(port);
+
+ return jerry_create_number(res);
+}
+
+
+/**
+ * MQTT_JS#yield (native JavaScript method)
+ *
+ * Waits to receive from the MQTT Broker.
+ */
+DECLARE_CLASS_FUNCTION(MQTT_JS, yield) {
+ CHECK_ARGUMENT_COUNT(MQTT_JS, yield, (args_count == 1));
+ CHECK_ARGUMENT_TYPE_ALWAYS(MQTT_JS, init, 0, number);
+
+ int time = jerry_get_number_value(args[0]);
+
+ printf("Yielding for %d ms.\n", time);
+ // Unwrap native MQTT_JS object
+ void *void_ptr;
+ const jerry_object_native_info_t *type_ptr;
+ bool has_ptr = jerry_get_object_native_pointer(this_obj, &void_ptr, &type_ptr);
+
+ if (!has_ptr || type_ptr != &native_obj_type_info) {
+ return jerry_create_error(JERRY_ERROR_TYPE,
+ (const jerry_char_t *) "Failed to get native MQTT_JS pointer");
+ }
+
+ MQTT_JS *native_ptr = static_cast<MQTT_JS*>(void_ptr);
+
+ //int ret = 0; //native_ptr->yield();
+ //NetworkInterface_JS::getInstance()->yield();
+
+ int result = native_ptr->yield(time);
+
+ return jerry_create_number(result);
+}
+
+
+/**
+ * MQTT_JS#connect (native JavaScript method)
+ *
+ * Connects to the MQTT Broker.
+ */
+DECLARE_CLASS_FUNCTION(MQTT_JS, connect) {
+ CHECK_ARGUMENT_COUNT(MQTT_JS, connect, (args_count == 0));
+
+ // Unwrap native MQTT_JS object
+ void *void_ptr;
+ const jerry_object_native_info_t *type_ptr;
+ bool has_ptr = jerry_get_object_native_pointer(this_obj, &void_ptr, &type_ptr);
+
+ if (!has_ptr || type_ptr != &native_obj_type_info) {
+ return jerry_create_error(JERRY_ERROR_TYPE,
+ (const jerry_char_t *) "Failed to get native MQTT_JS pointer");
+ }
+
+ MQTT_JS *native_ptr = static_cast<MQTT_JS*>(void_ptr);
+
+ //int ret = 0; //native_ptr->connect();
+ //NetworkInterface_JS::getInstance()->connect();
+
+ int result = native_ptr->connect(NetworkInterface_JS::getInstance()->getNetworkInterface());
+
+ return jerry_create_number(result);
+}
+
+
+/**
+ * MQTT_JS#publish (native JavaScript method)
+ *
+ * Publishes to the MQTT.
+ */
+DECLARE_CLASS_FUNCTION(MQTT_JS, publish) {
+ CHECK_ARGUMENT_COUNT(MQTT_JS, subscribe, (args_count == 1));
+ CHECK_ARGUMENT_TYPE_ALWAYS(MQTT_JS, subscribe, 0, string);
+
+ size_t buf_length = jerry_get_string_length(args[0]);
+
+ // add an extra character to ensure there's a null character after the device name
+ char* buf = (char*)calloc(buf_length + 1, sizeof(char));
+ jerry_string_to_char_buffer(args[0], (jerry_char_t*)buf, buf_length);
+
+ // Unwrap native MQTT_JS object
+ void *void_ptr;
+ const jerry_object_native_info_t *type_ptr;
+ bool has_ptr = jerry_get_object_native_pointer(this_obj, &void_ptr, &type_ptr);
+
+ if (!has_ptr || type_ptr != &native_obj_type_info) {
+ return jerry_create_error(JERRY_ERROR_TYPE,
+ (const jerry_char_t *) "Failed to get native MQTT_JS pointer");
+ }
+
+ MQTT_JS *native_ptr = static_cast<MQTT_JS*>(void_ptr);
+
+ int result = native_ptr->publish(buf);
+
+ free(buf);
+ return jerry_create_number(result);
+
+}
+
+/**
+ * MQTT_JS#run (native JavaScript method)
+ *
+ * Runs the MQTT demo.
+ */
+DECLARE_CLASS_FUNCTION(MQTT_JS, run) {
+ CHECK_ARGUMENT_COUNT(MQTT_JS, run, (args_count == 0));
+
+ // Unwrap native MQTT_JS object
+ void *void_ptr;
+ const jerry_object_native_info_t *type_ptr;
+ bool has_ptr = jerry_get_object_native_pointer(this_obj, &void_ptr, &type_ptr);
+
+ if (!has_ptr || type_ptr != &native_obj_type_info) {
+ return jerry_create_error(JERRY_ERROR_TYPE,
+ (const jerry_char_t *) "Failed to get native MQTT_JS pointer");
+ }
+
+ MQTT_JS *native_ptr = static_cast<MQTT_JS*>(void_ptr);
+
+ //int ret = 0; //native_ptr->run();
+ NetworkInterface_JS::getInstance()->connect();
+
+ //MQTT_JS *mqtt = new MQTT_JS();
+ native_ptr->start_mqtt(NetworkInterface_JS::getInstance()->getNetworkInterface());
+
+ //native_ptr->run();
+
+ return jerry_create_undefined();
+}
+
+DECLARE_CLASS_FUNCTION(MQTT_JS, onSubscribe) {
+ CHECK_ARGUMENT_COUNT(MQTT_JS, onUpdate, (args_count == 1));
+ CHECK_ARGUMENT_TYPE_ALWAYS(MQTT_JS, onUpdate, 0, function);
+
+ void *void_ptr;
+ const jerry_object_native_info_t *type_ptr;
+ bool has_ptr = jerry_get_object_native_pointer(this_obj, &void_ptr, &type_ptr);
+
+ if (!has_ptr || type_ptr != &native_obj_type_info) {
+ return jerry_create_error(JERRY_ERROR_TYPE,
+ (const jerry_char_t *) "Failed to get native MQTT_JS pointer");
+ }
+
+ MQTT_JS *native_ptr = static_cast<MQTT_JS*>(void_ptr);
+
+ jerry_value_t fn = args[0];
+ jerry_acquire_value(fn);
+
+
+ //BLEJS* this_ble = &BLEJS::Instance();
+ //this_ble->setWriteCallback(native_ptr, f);
+ int result = native_ptr->onSubscribe(fn);
+
+ return jerry_create_number(result);
+}
+
+/**
+ * MQTT_JS#subscribe (native JavaScript method)
+ *
+ * Subscribes to MQTT
+ *
+ * @param topic
+ */
+DECLARE_CLASS_FUNCTION(MQTT_JS, subscribe) {
+ CHECK_ARGUMENT_COUNT(MQTT_JS, subscribe, (args_count == 1));
+ CHECK_ARGUMENT_TYPE_ALWAYS(MQTT_JS, subscribe, 0, string);
+
+ size_t topic_length = jerry_get_string_length(args[0]);
+
+ // add an extra character to ensure there's a null character after the device name
+ char* topic = (char*)calloc(topic_length + 1, sizeof(char));
+ jerry_string_to_char_buffer(args[0], (jerry_char_t*)topic, topic_length);
+
+ // Unwrap native MQTT_JS object
+ void *void_ptr;
+ const jerry_object_native_info_t *type_ptr;
+ bool has_ptr = jerry_get_object_native_pointer(this_obj, &void_ptr, &type_ptr);
+
+ if (!has_ptr || type_ptr != &native_obj_type_info) {
+ return jerry_create_error(JERRY_ERROR_TYPE,
+ (const jerry_char_t *) "Failed to get native MQTT_JS pointer");
+ }
+
+ MQTT_JS *native_ptr = static_cast<MQTT_JS*>(void_ptr);
+
+ int result = native_ptr->subscribe(topic);
+
+ free(topic);
+ return jerry_create_number(result);
+}
+
+
+/**
+ * MQTT_JS (native JavaScript constructor)
+ *
+ * @returns a JavaScript object representing the MQTT_JS.
+ */
+DECLARE_CLASS_CONSTRUCTOR(MQTT_JS) {
+ CHECK_ARGUMENT_COUNT(MQTT_JS, __constructor, (args_count == 0));
+
+ MQTT_JS *native_ptr = new MQTT_JS();
+
+ jerry_value_t js_object = jerry_create_object();
+ jerry_set_object_native_pointer(js_object, native_ptr, &native_obj_type_info);
+
+
+ ATTACH_CLASS_FUNCTION(js_object, MQTT_JS, run);
+ ATTACH_CLASS_FUNCTION(js_object, MQTT_JS, onSubscribe);
+ ATTACH_CLASS_FUNCTION(js_object, MQTT_JS, init);
+ ATTACH_CLASS_FUNCTION(js_object, MQTT_JS, connect);
+ ATTACH_CLASS_FUNCTION(js_object, MQTT_JS, subscribe);
+ ATTACH_CLASS_FUNCTION(js_object, MQTT_JS, publish);
+ ATTACH_CLASS_FUNCTION(js_object, MQTT_JS, yield);
+
+ return js_object;
+}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT_JS-js.h Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,57 @@
+/*
+ * @file MQTT_JS-js.h
+ * @author ST
+ * @version V1.0.0
+ * @date 9 October 2017
+ * @brief Implementation of MQTT for Javascript.
+ ******************************************************************************
+ * @attention
+ *
+ * <h2><center>© COPYRIGHT(c) 2017 STMicroelectronics</center></h2>
+ *
+ * Redistribution and use in source and binary forms, with or without modification,
+ * are permitted provided that the following conditions are met:
+ * 1. Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * 3. Neither the name of STMicroelectronics nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+ * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ ******************************************************************************
+ */
+
+
+/* Prevent recursive inclusion -----------------------------------------------*/
+
+#ifndef _MQTT_JS_JS_H
+#define _MQTT_JS_JS_H
+
+/* Includes ------------------------------------------------------------------*/
+
+// This file contains all the macros
+#include "jerryscript-mbed-library-registry/wrap_tools.h"
+
+// Class constructor
+DECLARE_CLASS_CONSTRUCTOR(MQTT_JS);
+
+// Define a wrapper, we can load the wrapper in `main.cpp`.
+// This makes it possible to load libraries optionally.
+DECLARE_JS_WRAPPER_REGISTRATION (MQTT_JS_library) {
+ REGISTER_CLASS_CONSTRUCTOR(MQTT_JS);
+}
+
+#endif
\ No newline at end of file
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT_JS.cpp Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,352 @@
+/*
+ * @file MQTT_JS.cpp
+ * @author ST
+ * @version V1.0.0
+ * @date 9 October 2017
+ * @brief Implementation of MQTT for Javascript.
+ ******************************************************************************
+ * @attention
+ *
+ * <h2><center>© COPYRIGHT(c) 2017 STMicroelectronics</center></h2>
+ *
+ * Redistribution and use in source and binary forms, with or without modification,
+ * are permitted provided that the following conditions are met:
+ * 1. Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * 3. Neither the name of STMicroelectronics nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+ * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ ******************************************************************************
+ */
+
+
+/* Includes ------------------------------------------------------------------*/
+
+#include "MQTT_JS.h"
+
+/** onSubscribeCallback
+ * @brief onSubscribeCallback.
+ */
+jerry_value_t MQTT_JS::onSubscribeCallback;
+
+/** Constructor
+ * @brief Constructor.
+ */
+MQTT_JS::MQTT_JS(){
+ connack_rc = 0; // MQTT connack return code
+ ip_addr = NULL;
+ netConnecting = false;
+ connectTimeout = 1000;
+ mqttConnecting = false;
+ netConnected = false;
+ connected = false;
+ retryAttempt = 0;
+
+ client = NULL;
+ mqttNetwork = NULL;
+
+ onSubscribeCallback = NULL;
+
+}
+
+/** Destructor
+ * @brief Destructor.
+ */
+MQTT_JS::~MQTT_JS(){
+ if(client){
+ delete client;
+ client = NULL;
+ }
+ if(mqttNetwork){
+ delete mqttNetwork;
+ mqttNetwork = NULL;
+ }
+}
+
+/** subscribe_cb
+ * @brief Connects the subscription callback.
+ * @param Message Data
+ */
+void MQTT_JS::subscribe_cb(MQTT::MessageData & msgMQTT) {
+ char msg[MQTT_MAX_PAYLOAD_SIZE];
+ msg[0]='\0';
+ strncat (msg, (char*)msgMQTT.message.payload, msgMQTT.message.payloadlen);
+ //printf ("--->>> subscribe_cb msg: %s\n\r", msg);
+
+ if (onSubscribeCallback && jerry_value_is_function(onSubscribeCallback)) {
+
+ jerry_value_t this_val = jerry_create_undefined ();
+ const jerry_value_t args[1] = {
+ jerry_create_string ((const jerry_char_t *)msg)
+ //jerry_create_number(3)
+ };
+
+ jerry_value_t ret_val = jerry_call_function (onSubscribeCallback, this_val, args, 1);
+
+ if (!jerry_value_has_error_flag (ret_val))
+ {
+ // handle return value
+ }
+ jerry_release_value(args[0]);
+
+ jerry_release_value (ret_val);
+ jerry_release_value (this_val);
+
+ }
+}
+
+/** onSubscribe
+ * @brief Calls the subscription callback.
+ * @param Jerry Callback
+ * @return Return code
+ */
+int MQTT_JS::onSubscribe(jerry_value_t cb){
+
+ if (jerry_value_is_function(cb)) {
+ onSubscribeCallback = cb;
+ return 0;
+ }
+ return 1;
+}
+
+/** subscribe
+ * @brief Subscribes to the topic.
+ * @param Topic
+ * @return Return code
+ */
+int MQTT_JS::subscribe (char *_topic)
+{
+ strcpy(topic, _topic);
+ if(!topic){
+ return 1; // invalid topic
+ }
+ return client->subscribe(topic, MQTT::QOS1, subscribe_cb);
+}
+
+/** unsubscribe
+ * @brief Unsubscribes the callback
+ * @param Topic
+ * @return Return code
+ */
+int MQTT_JS::unsubscribe(char *pubTopic)
+{
+ return client->unsubscribe(pubTopic);
+}
+
+
+/** init
+ * @brief Initializes the MQTT.
+ * @param NetworkInterface
+ * @param ID
+ * @param Token
+ * @param URL
+ * @param Port
+ * @return Return code
+ */
+int MQTT_JS::init(NetworkInterface* network, char* _id, char* _token, char* _url, char* _port)
+{
+ sprintf (id, "%s", _id);
+ sprintf (auth_token, "%s", _token);
+ sprintf (hostname, "%s", _url);
+ sprintf (subscription_url, "%s", _url);
+ sprintf (port, "%s", _port);
+
+ if (!network) {
+ printf ("Error easy_connect\n\r");
+ return -1;
+ }
+
+ mqttNetwork = new MQTTNetwork(network);
+
+ client = new MQTT::Client<MQTTNetwork, Countdown, MQTT_MAX_PACKET_SIZE>(*mqttNetwork);
+
+ return 0;
+}
+
+/** connect
+ * @brief Connects to the MQTT Server.
+ * @param NetworkInterface
+ * @return Return code
+ */
+int MQTT_JS::connect(NetworkInterface* network)
+{
+ netConnecting = true;
+ int rc = mqttNetwork->connect(hostname, atoi(port));
+ if (rc != 0)
+ {
+ //WARN("IP Stack connect returned: %d\n", rc);
+ return rc;
+ }
+ printf ("--->TCP Connected\n\r");
+ netConnected = true;
+ netConnecting = false;
+
+ // MQTT Connect
+ mqttConnecting = true;
+ MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
+ data.MQTTVersion = 4;
+ data.struct_version=0;
+ data.clientID.cstring = id;
+ data.username.cstring = id;
+ data.password.cstring = auth_token;
+ data.keepAliveInterval = 15; // in Sec
+ if ((rc = client->connect(data)) == 0)
+ {
+ connected = true;
+ printf ("--->MQTT Connected\n\r");
+ }
+ else {
+ WARN("MQTT connect returned %d\n", rc);
+ }
+ if (rc >= 0)
+ connack_rc = rc;
+ mqttConnecting = false;
+ return rc;
+}
+
+/** getConnTimeout
+ * @brief Returns the timeout in seconds.
+ * @param Attempt number
+ * @return Return code
+ */
+int MQTT_JS::getConnTimeout(int attemptNumber)
+{
+ // First 10 attempts try within 3 seconds, next 10 attempts retry after every 1 minute
+ // after 20 attempts, retry every 10 minutes
+ return (attemptNumber < 10) ? 3 : (attemptNumber < 20) ? 60 : 600;
+}
+
+/** attemptConnect
+ * @brief Attempt connection to MQTT server.
+ * @param NetworkInterface
+ */
+void MQTT_JS::attemptConnect(NetworkInterface* network)
+{
+ connected = false;
+
+ while (connect(network) != MQTT_CONNECTION_ACCEPTED)
+ {
+ if (connack_rc == MQTT_NOT_AUTHORIZED || connack_rc == MQTT_BAD_USERNAME_OR_PASSWORD) {
+ printf ("File: %s, Line: %d Error: %d\n\r",__FILE__,__LINE__, connack_rc);
+ return; // don't reattempt to connect if credentials are wrong
+ }
+ int timeout = getConnTimeout(++retryAttempt);
+ WARN("Retry attempt number %d waiting %d\n", retryAttempt, timeout);
+
+ // if ipstack and client were on the heap we could deconstruct and goto a label where they are constructed
+ // or maybe just add the proper members to do this disconnect and call attemptConnect(...)
+
+ // this works - reset the system when the retry count gets to a threshold
+ if (retryAttempt == 5)
+ NVIC_SystemReset();
+ else
+ wait(timeout);
+ }
+}
+
+/** publish
+ * @brief Publishes to the MQTT broker.
+ * @param Data
+ * @param Optional: retry number
+ * @return Return code
+ */
+int MQTT_JS::publish(char* buf, int n)
+{
+ MQTT::Message message;
+ message.qos = MQTT::QOS0;
+ message.retained = false;
+ message.dup = false;
+ message.payload = (void*)buf;
+ message.payloadlen = strlen(buf);
+
+ //LOG("Publishing %s\n\r", buf);
+ int result = client->publish(topic, message);
+ if(result != 0){
+ if(n < 2){
+ printf("\33[31mCould not publish message. Trying again...\33[0m\n");
+ return publish(buf, n+1);
+ }
+ else{
+ printf("\33[31mError publishing message!\33[0m\n");
+ return result;
+ }
+ }
+
+ /*
+ if(result == 0){
+ client->yield(5000); // allow the MQTT client to receive messages
+ }
+ */
+ return result;
+}
+
+
+/** yield
+ * @brief Waits for the MQTT broker for subscription callback.
+ * @param Time to wait
+ * @return Return code
+ */
+int MQTT_JS::yield(int time)
+{
+ client->yield(time); // allow the MQTT client to receive messages
+ return 0;
+}
+
+/** start_mqtt
+ * @brief Starts a demo for MQTT.
+ * @param NetworkInterface
+ * @return Return code
+ */
+int MQTT_JS::start_mqtt(NetworkInterface* network)
+{
+ sprintf (id, "hsojbpev");
+ sprintf (auth_token, "4H5vbg1KAhYi");
+ sprintf (hostname, "m20.cloudmqtt.com");
+ sprintf (subscription_url, "m20.cloudmqtt.com");
+ sprintf (port, "10023");
+
+ if (!network) {
+ printf ("Error easy_connect\n\r");
+ return -1;
+ }
+
+ mqttNetwork = new MQTTNetwork(network);
+
+ client = new MQTT::Client<MQTTNetwork, Countdown, MQTT_MAX_PACKET_SIZE>(*mqttNetwork);
+
+ attemptConnect(network);
+ if (connack_rc == MQTT_NOT_AUTHORIZED || connack_rc == MQTT_BAD_USERNAME_OR_PASSWORD)
+ {
+ while (true)
+ wait(1.0); // Permanent failures - don't retry
+ }
+
+ int count = 0;
+ while (true)
+ {
+ if (++count == 6)
+ {
+ // Publish a message every ~3 second
+ if (publish((char*)"TestTest") != 0) {
+ attemptConnect(network); // if we have lost the connection
+ }
+ count = 0;
+ }
+ client->yield(500); // allow the MQTT client to receive messages
+ }
+}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/MQTT_JS.h Wed Jan 17 11:30:51 2018 +0100
@@ -0,0 +1,137 @@
+/*
+ * @file MQTT_JS.h
+ * @author ST
+ * @version V1.0.0
+ * @date 9 October 2017
+ * @brief Implementation of MQTT for Javascript.
+ ******************************************************************************
+ * @attention
+ *
+ * <h2><center>© COPYRIGHT(c) 2017 STMicroelectronics</center></h2>
+ *
+ * Redistribution and use in source and binary forms, with or without modification,
+ * are permitted provided that the following conditions are met:
+ * 1. Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * 3. Neither the name of STMicroelectronics nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+ * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ ******************************************************************************
+ */
+
+/* Prevent recursive inclusion -----------------------------------------------*/
+
+#ifndef _MQTT_JS_H_
+#define _MQTT_JS_H_
+
+/* Includes ------------------------------------------------------------------*/
+
+#include "mbed.h"
+#include "TCPSocket.h"
+#include "MQTTClient.h"
+#include "MQTTNetwork.h"
+#include "MQTTmbed.h"
+
+#include "NetworkInterface_JS.h"
+
+#include "jerryscript-mbed-library-registry/wrap_tools.h"
+
+#include <ctype.h>
+
+/* Constants -----------------------------------------------------------------*/
+
+#define MQTT_MAX_PACKET_SIZE 250
+#define MQTT_MAX_PAYLOAD_SIZE 300
+
+#define MAX_SSID_LEN 80
+#define MAX_PASSW_LEN 80
+
+#define HTTP_BROKER_URL "http://customer.cloudmqtt.com/login"
+
+typedef void (* subscribeCallbackType)(MQTT::MessageData & msgMQTT);
+
+/* Class Declaration ---------------------------------------------------------*/
+
+/**
+ * Abstract class of MQTT for Javascript.
+ */
+class MQTT_JS{
+private:
+ char ssid[MAX_SSID_LEN];
+ char seckey[MAX_PASSW_LEN];
+
+ char id[32];
+ char topic[32];
+ char auth_token[32];
+ char hostname[128];
+ char port[16];
+
+ int connack_rc; // MQTT connack return code
+ char* ip_addr;
+ char type[30];
+ bool netConnecting;
+ int connectTimeout;
+ bool mqttConnecting;
+ bool netConnected;
+ bool connected;
+ int retryAttempt;
+ char subscription_url[300];
+ MQTT::Client<MQTTNetwork, Countdown, MQTT_MAX_PACKET_SIZE>* client;
+ MQTTNetwork* mqttNetwork;
+
+ static jerry_value_t onSubscribeCallback;
+
+public:
+
+ /* Constructors */
+ MQTT_JS();
+
+ /* Destructors */
+ ~MQTT_JS();
+
+ /* Functions */
+
+ NetworkInterface* getNetworkInterface();
+
+ int onSubscribe(jerry_value_t cb);
+ int onDisconnect(jerry_value_t cb, subscribeCallbackType fn);
+
+ static void subscribe_cb(MQTT::MessageData & msgMQTT);
+
+ int init(NetworkInterface* network, char* _id, char* _token, char* _url, char* _port);
+
+ int connect();
+
+ int subscribe(char *pubTopic);
+
+ int unsubscribe(char *pubTopic);
+
+ int connect(NetworkInterface* network);
+
+ int getConnTimeout(int attemptNumber);
+
+ void attemptConnect(NetworkInterface* network) ;
+
+ int publish(char* buf, int n = 0);
+
+ int yield(int time);
+
+ int start_mqtt(NetworkInterface* network);
+};
+
+#endif
\ No newline at end of file
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/mbed-http.lib Wed Jan 17 11:30:51 2018 +0100 @@ -0,0 +1,1 @@ +https://developer.mbed.org/teams/sandbox/code/mbed-http/#f7a85895a941