fork of wolfSSL/MQTTS library with increased message buffer & increased number of subscribed topics
Fork of MQTTS by
Revision 3:dbff6b768d28, committed 2014-03-31
- Comitter:
- icraggs
- Date:
- Mon Mar 31 15:48:45 2014 +0000
- Parent:
- 2:dcfdd2abfe71
- Child:
- 4:4ef00243708e
- Commit message:
- Move parameters around to avoid storing data
Changed in this revision
--- a/MQTTClient.cpp Fri Mar 28 13:39:25 2014 +0000 +++ b/MQTTClient.cpp Mon Mar 31 15:48:45 2014 +0000 @@ -23,32 +23,102 @@ #include "MQTTClient.h" #include "MQTTPacket.h" -MQTTClient::MQTTClient(char* serverURI, char* clientId, const int buffer_size) +MQTT::Client::Client(IPStack* ipstack, const int buffer_size) { buf = new char[buffer_size]; - this->clientId = clientId; - this->serverURI = serverURI; + this->ipstack = ipstack; +} + + +int MQTT::Client::sendPacket(int length) +{ + int sent = 0; + + while (sent < length) + sent += ipstack->write(&buf[sent], length); + + return sent; +} + + +int MQTT::Client::decodePacket(int* value, int timeout) +{ + char c; + int multiplier = 1; + int len = 0; +#define 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; } -int MQTTClient::connect(MQTTConnectOptions* options, FP<void, MQTTResult*> resultHandler) +int MQTT::Client::readPacket(int timeout) { - MQTTPacket_connectData data = MQTTPacket_connectData_initializer; - - data.clientID.cstring = clientId; - if (options) - { - data.keepAliveInterval = options->keepAliveInterval; - data.cleansession = options->cleansession; - data.username.cstring = options->username; - data.password.cstring = options->password; - } - - int len = MQTTSerialize_connect(buf, buflen, &data); + int rc = -1; + MQTTHeader header; + int len = 0; + int rem_len = 0; + + /* 1. read the header byte. This has the packet type in it */ + if ((rc = 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 ((rc = ipstack->read(readbuf + len, rem_len, timeout)) != rem_len) + goto exit; + + header.byte = buf[0]; + rc = header.bits.type; +exit: + return rc; +} + + +void MQTT::Client::cycle() +{ + int timeout = 1000L; + /* get one piece of work off the wire and one pass through */ - sendPacket(buf, buflen); // send the connect packet + // 1. read the socket, see what work is due. + int packet_type = readPacket(buf, buflen, -1); +} + + +int MQTT::Client::connect(MQTTPacket_connectData* options, FP<void, MQTT::Result*> resultHandler) +{ + /* 1. connect to the server with the desired transport */ + if (!ipstack->connect()) + return -99; + + /* 2. if the connect was successful, send the MQTT connect packet */ + int len = MQTTSerialize_connect(buf, buflen, options); + sendPacket(len); // send the connect packet + + /* 3. wait until the connack is received */ /* how to make this check work? if (resultHandler == None) {
--- a/MQTTClient.h Fri Mar 28 13:39:25 2014 +0000 +++ b/MQTTClient.h Mon Mar 31 15:48:45 2014 +0000 @@ -27,18 +27,23 @@ #include "mbed.h" #include "FP.h" +#include "MQTTPacket.h" +#include "include_me.h" -class MQTTClient; +namespace MQTT +{ + +class Client; enum QoS { QOS0, QOS1, QOS2 }; -class MQTTResult +class Result { /* success or failure result data */ - MQTTClient* client; + Client* client; }; -struct MQTTMessage +struct Message { enum QoS qos; bool retained; @@ -48,45 +53,40 @@ size_t payloadlen; }; -struct MQTTConnectOptions -{ - unsigned short keepAliveInterval; - bool cleansession; - char* username; - char* password; - int timeout; - std::vector<char*> serverURIs; -}; - -class MQTTClient +class Client { public: - static FP<void, MQTTResult*> None; // default argument of no result handler to indicate call should be blocking + static FP<void, Result*> None; // default argument of no result handler to indicate call should be blocking - MQTTClient(char* serverURI, char* clientId = "", const int buffer_size = 100); + Client(IPStack* ipstack, const int buffer_size = 100); - int connect(MQTTConnectOptions* options = 0, FP<void, MQTTResult*> resultHandler = None); + int connect(MQTTPacket_connectData* options = 0, FP<void, Result*> resultHandler = None); - int publish(char* topic, MQTTMessage* message, FP<void, MQTTResult*> resultHandler = None); + int publish(char* topic, Message* message, FP<void, Result*> resultHandler = None); - int subscribe(char* topicFilter, int qos, FP<void, MQTTMessage*> messageHandler, FP<void, MQTTResult*> resultHandler = None); + int subscribe(char* topicFilter, int qos, FP<void, Message*> messageHandler, FP<void, Result*> resultHandler = None); - int unsubscribe(char* topicFilter, FP<void, MQTTResult*> resultHandler = None); + int unsubscribe(char* topicFilter, FP<void, Result*> resultHandler = None); - int disconnect(int timeout, FP<void, MQTTResult*> resultHandler = None); + int disconnect(int timeout, FP<void, Result*> resultHandler = None); private: - int sendPacket(char* buf, int buflen); + void cycle(); - char* clientId; - char* serverURI; + int decodePacket(int* value, int timeout); + int readPacket(char* buf, int buflen, int timeout = -1); + int sendPacket(int length); + + IPStack* ipstack; char* buf; int buflen; }; +} + #endif
--- a/MQTTClient.lib Fri Mar 28 13:39:25 2014 +0000 +++ b/MQTTClient.lib Mon Mar 31 15:48:45 2014 +0000 @@ -1,1 +1,1 @@ -http://mbed.org/users/icraggs/code/MQTTClient/#7734401cc1b4 +http://mbed.org/users/icraggs/code/MQTTClient/#069ae45b7070