Junichi Katsu / Mbed OS mbed-os-example-milkcocoa

mbed OS5に対応したMilkcocoaライブラリのテストバージョンです。

Fork of mbed-os-example-mbed5-blinky by mbed-os-examples

Milkcocoa/MQTT/MQTTClient.h@24:6ba1245bf049, 2017-01-24 (annotated)

Committer:
jksoft
Date:
Tue Jan 24 13:41:36 2017 +0000
Revision:
24:6ba1245bf049
??????????

Who changed what in which revision?

UserRevisionLine numberNew contents of line
jksoft 24:6ba1245bf049 1 /*******************************************************************************
jksoft 24:6ba1245bf049 2 * Copyright (c) 2014, 2015 IBM Corp.
jksoft 24:6ba1245bf049 3 *
jksoft 24:6ba1245bf049 4 * All rights reserved. This program and the accompanying materials
jksoft 24:6ba1245bf049 5 * are made available under the terms of the Eclipse Public License v1.0
jksoft 24:6ba1245bf049 6 * and Eclipse Distribution License v1.0 which accompany this distribution.
jksoft 24:6ba1245bf049 7 *
jksoft 24:6ba1245bf049 8 * The Eclipse Public License is available at
jksoft 24:6ba1245bf049 9 * http://www.eclipse.org/legal/epl-v10.html
jksoft 24:6ba1245bf049 10 * and the Eclipse Distribution License is available at
jksoft 24:6ba1245bf049 11 * http://www.eclipse.org/org/documents/edl-v10.php.
jksoft 24:6ba1245bf049 12 *
jksoft 24:6ba1245bf049 13 * Contributors:
jksoft 24:6ba1245bf049 14 * Ian Craggs - initial API and implementation and/or initial documentation
jksoft 24:6ba1245bf049 15 * Ian Craggs - fix for bug 458512 - QoS 2 messages
jksoft 24:6ba1245bf049 16 * Ian Craggs - fix for bug 460389 - send loop uses wrong length
jksoft 24:6ba1245bf049 17 * Ian Craggs - fix for bug 464169 - clearing subscriptions
jksoft 24:6ba1245bf049 18 * Ian Craggs - fix for bug 464551 - enums and ints can be different size
jksoft 24:6ba1245bf049 19 *******************************************************************************/
jksoft 24:6ba1245bf049 20
jksoft 24:6ba1245bf049 21 #if !defined(MQTTCLIENT_H)
jksoft 24:6ba1245bf049 22 #define MQTTCLIENT_H
jksoft 24:6ba1245bf049 23
jksoft 24:6ba1245bf049 24 #include "FP.h"
jksoft 24:6ba1245bf049 25 #include "MQTTPacket.h"
jksoft 24:6ba1245bf049 26 #include "stdio.h"
jksoft 24:6ba1245bf049 27 #include "MQTTLogging.h"
jksoft 24:6ba1245bf049 28
jksoft 24:6ba1245bf049 29 #if !defined(MQTTCLIENT_QOS1)
jksoft 24:6ba1245bf049 30 #define MQTTCLIENT_QOS1 1
jksoft 24:6ba1245bf049 31 #endif
jksoft 24:6ba1245bf049 32 #if !defined(MQTTCLIENT_QOS2)
jksoft 24:6ba1245bf049 33 #define MQTTCLIENT_QOS2 0
jksoft 24:6ba1245bf049 34 #endif
jksoft 24:6ba1245bf049 35
jksoft 24:6ba1245bf049 36 namespace MQTT
jksoft 24:6ba1245bf049 37 {
jksoft 24:6ba1245bf049 38
jksoft 24:6ba1245bf049 39
jksoft 24:6ba1245bf049 40 enum QoS { QOS0, QOS1, QOS2 };
jksoft 24:6ba1245bf049 41
jksoft 24:6ba1245bf049 42 // all failure return codes must be negative
jksoft 24:6ba1245bf049 43 enum returnCode { BUFFER_OVERFLOW = -2, FAILURE = -1, SUCCESS = 0 };
jksoft 24:6ba1245bf049 44
jksoft 24:6ba1245bf049 45
jksoft 24:6ba1245bf049 46 struct Message
jksoft 24:6ba1245bf049 47 {
jksoft 24:6ba1245bf049 48 enum QoS qos;
jksoft 24:6ba1245bf049 49 bool retained;
jksoft 24:6ba1245bf049 50 bool dup;
jksoft 24:6ba1245bf049 51 unsigned short id;
jksoft 24:6ba1245bf049 52 void *payload;
jksoft 24:6ba1245bf049 53 size_t payloadlen;
jksoft 24:6ba1245bf049 54 };
jksoft 24:6ba1245bf049 55
jksoft 24:6ba1245bf049 56
jksoft 24:6ba1245bf049 57 struct MessageData
jksoft 24:6ba1245bf049 58 {
jksoft 24:6ba1245bf049 59 MessageData(MQTTString &aTopicName, struct Message &aMessage) : message(aMessage), topicName(aTopicName)
jksoft 24:6ba1245bf049 60 { }
jksoft 24:6ba1245bf049 61
jksoft 24:6ba1245bf049 62 struct Message &message;
jksoft 24:6ba1245bf049 63 MQTTString &topicName;
jksoft 24:6ba1245bf049 64 };
jksoft 24:6ba1245bf049 65
jksoft 24:6ba1245bf049 66
jksoft 24:6ba1245bf049 67 class PacketId
jksoft 24:6ba1245bf049 68 {
jksoft 24:6ba1245bf049 69 public:
jksoft 24:6ba1245bf049 70 PacketId()
jksoft 24:6ba1245bf049 71 {
jksoft 24:6ba1245bf049 72 next = 0;
jksoft 24:6ba1245bf049 73 }
jksoft 24:6ba1245bf049 74
jksoft 24:6ba1245bf049 75 int getNext()
jksoft 24:6ba1245bf049 76 {
jksoft 24:6ba1245bf049 77 return next = (next == MAX_PACKET_ID) ? 1 : ++next;
jksoft 24:6ba1245bf049 78 }
jksoft 24:6ba1245bf049 79
jksoft 24:6ba1245bf049 80 private:
jksoft 24:6ba1245bf049 81 static const int MAX_PACKET_ID = 65535;
jksoft 24:6ba1245bf049 82 int next;
jksoft 24:6ba1245bf049 83 };
jksoft 24:6ba1245bf049 84
jksoft 24:6ba1245bf049 85
jksoft 24:6ba1245bf049 86 /**
jksoft 24:6ba1245bf049 87 * @class Client
jksoft 24:6ba1245bf049 88 * @brief blocking, non-threaded MQTT client API
jksoft 24:6ba1245bf049 89 *
jksoft 24:6ba1245bf049 90 * This version of the API blocks on all method calls, until they are complete. This means that only one
jksoft 24:6ba1245bf049 91 * MQTT request can be in process at any one time.
jksoft 24:6ba1245bf049 92 * @param Network a network class which supports send, receive
jksoft 24:6ba1245bf049 93 * @param Timer a timer class with the methods:
jksoft 24:6ba1245bf049 94 */
jksoft 24:6ba1245bf049 95 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE = 100, int MAX_MESSAGE_HANDLERS = 5>
jksoft 24:6ba1245bf049 96 class Client
jksoft 24:6ba1245bf049 97 {
jksoft 24:6ba1245bf049 98
jksoft 24:6ba1245bf049 99 public:
jksoft 24:6ba1245bf049 100
jksoft 24:6ba1245bf049 101 typedef void (*messageHandler)(MessageData&);
jksoft 24:6ba1245bf049 102
jksoft 24:6ba1245bf049 103 /** Construct the client
jksoft 24:6ba1245bf049 104 * @param network - pointer to an instance of the Network class - must be connected to the endpoint
jksoft 24:6ba1245bf049 105 * before calling MQTT connect
jksoft 24:6ba1245bf049 106 * @param limits an instance of the Limit class - to alter limits as required
jksoft 24:6ba1245bf049 107 */
jksoft 24:6ba1245bf049 108 Client(Network& network, unsigned int command_timeout_ms = 30000);
jksoft 24:6ba1245bf049 109
jksoft 24:6ba1245bf049 110 /** Set the default message handling callback - used for any message which does not match a subscription message handler
jksoft 24:6ba1245bf049 111 * @param mh - pointer to the callback function
jksoft 24:6ba1245bf049 112 */
jksoft 24:6ba1245bf049 113 void setDefaultMessageHandler(messageHandler mh)
jksoft 24:6ba1245bf049 114 {
jksoft 24:6ba1245bf049 115 defaultMessageHandler.attach(mh);
jksoft 24:6ba1245bf049 116 }
jksoft 24:6ba1245bf049 117
jksoft 24:6ba1245bf049 118 /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
jksoft 24:6ba1245bf049 119 * The nework object must be connected to the network endpoint before calling this
jksoft 24:6ba1245bf049 120 * Default connect options are used
jksoft 24:6ba1245bf049 121 * @return success code -
jksoft 24:6ba1245bf049 122 */
jksoft 24:6ba1245bf049 123 int connect();
jksoft 24:6ba1245bf049 124
jksoft 24:6ba1245bf049 125 /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
jksoft 24:6ba1245bf049 126 * The nework object must be connected to the network endpoint before calling this
jksoft 24:6ba1245bf049 127 * @param options - connect options
jksoft 24:6ba1245bf049 128 * @return success code -
jksoft 24:6ba1245bf049 129 */
jksoft 24:6ba1245bf049 130 int connect(MQTTPacket_connectData& options);
jksoft 24:6ba1245bf049 131
jksoft 24:6ba1245bf049 132 /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
jksoft 24:6ba1245bf049 133 * @param topic - the topic to publish to
jksoft 24:6ba1245bf049 134 * @param message - the message to send
jksoft 24:6ba1245bf049 135 * @return success code -
jksoft 24:6ba1245bf049 136 */
jksoft 24:6ba1245bf049 137 int publish(const char* topicName, Message& message);
jksoft 24:6ba1245bf049 138
jksoft 24:6ba1245bf049 139 /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
jksoft 24:6ba1245bf049 140 * @param topic - the topic to publish to
jksoft 24:6ba1245bf049 141 * @param payload - the data to send
jksoft 24:6ba1245bf049 142 * @param payloadlen - the length of the data
jksoft 24:6ba1245bf049 143 * @param qos - the QoS to send the publish at
jksoft 24:6ba1245bf049 144 * @param retained - whether the message should be retained
jksoft 24:6ba1245bf049 145 * @return success code -
jksoft 24:6ba1245bf049 146 */
jksoft 24:6ba1245bf049 147 int publish(const char* topicName, void* payload, size_t payloadlen, enum QoS qos = QOS0, bool retained = false);
jksoft 24:6ba1245bf049 148
jksoft 24:6ba1245bf049 149 /** MQTT Publish - send an MQTT publish packet and wait for all acks to complete for all QoSs
jksoft 24:6ba1245bf049 150 * @param topic - the topic to publish to
jksoft 24:6ba1245bf049 151 * @param payload - the data to send
jksoft 24:6ba1245bf049 152 * @param payloadlen - the length of the data
jksoft 24:6ba1245bf049 153 * @param id - the packet id used - returned
jksoft 24:6ba1245bf049 154 * @param qos - the QoS to send the publish at
jksoft 24:6ba1245bf049 155 * @param retained - whether the message should be retained
jksoft 24:6ba1245bf049 156 * @return success code -
jksoft 24:6ba1245bf049 157 */
jksoft 24:6ba1245bf049 158 int publish(const char* topicName, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos = QOS1, bool retained = false);
jksoft 24:6ba1245bf049 159
jksoft 24:6ba1245bf049 160 /** MQTT Subscribe - send an MQTT subscribe packet and wait for the suback
jksoft 24:6ba1245bf049 161 * @param topicFilter - a topic pattern which can include wildcards
jksoft 24:6ba1245bf049 162 * @param qos - the MQTT QoS to subscribe at
jksoft 24:6ba1245bf049 163 * @param mh - the callback function to be invoked when a message is received for this subscription
jksoft 24:6ba1245bf049 164 * @return success code -
jksoft 24:6ba1245bf049 165 */
jksoft 24:6ba1245bf049 166 int subscribe(const char* topicFilter, enum QoS qos, messageHandler mh);
jksoft 24:6ba1245bf049 167
jksoft 24:6ba1245bf049 168 /** MQTT Unsubscribe - send an MQTT unsubscribe packet and wait for the unsuback
jksoft 24:6ba1245bf049 169 * @param topicFilter - a topic pattern which can include wildcards
jksoft 24:6ba1245bf049 170 * @return success code -
jksoft 24:6ba1245bf049 171 */
jksoft 24:6ba1245bf049 172 int unsubscribe(const char* topicFilter);
jksoft 24:6ba1245bf049 173
jksoft 24:6ba1245bf049 174 /** MQTT Disconnect - send an MQTT disconnect packet, and clean up any state
jksoft 24:6ba1245bf049 175 * @return success code -
jksoft 24:6ba1245bf049 176 */
jksoft 24:6ba1245bf049 177 int disconnect();
jksoft 24:6ba1245bf049 178
jksoft 24:6ba1245bf049 179 /** A call to this API must be made within the keepAlive interval to keep the MQTT connection alive
jksoft 24:6ba1245bf049 180 * yield can be called if no other MQTT operation is needed. This will also allow messages to be
jksoft 24:6ba1245bf049 181 * received.
jksoft 24:6ba1245bf049 182 * @param timeout_ms the time to wait, in milliseconds
jksoft 24:6ba1245bf049 183 * @return success code - on failure, this means the client has disconnected
jksoft 24:6ba1245bf049 184 */
jksoft 24:6ba1245bf049 185 int yield(unsigned long timeout_ms = 1000L);
jksoft 24:6ba1245bf049 186
jksoft 24:6ba1245bf049 187 /** Is the client connected?
jksoft 24:6ba1245bf049 188 * @return flag - is the client connected or not?
jksoft 24:6ba1245bf049 189 */
jksoft 24:6ba1245bf049 190 bool isConnected()
jksoft 24:6ba1245bf049 191 {
jksoft 24:6ba1245bf049 192 return isconnected;
jksoft 24:6ba1245bf049 193 }
jksoft 24:6ba1245bf049 194
jksoft 24:6ba1245bf049 195 private:
jksoft 24:6ba1245bf049 196
jksoft 24:6ba1245bf049 197 void cleanSession();
jksoft 24:6ba1245bf049 198 int cycle(Timer& timer);
jksoft 24:6ba1245bf049 199 int waitfor(int packet_type, Timer& timer);
jksoft 24:6ba1245bf049 200 int keepalive();
jksoft 24:6ba1245bf049 201 int publish(int len, Timer& timer, enum QoS qos);
jksoft 24:6ba1245bf049 202
jksoft 24:6ba1245bf049 203 int decodePacket(int* value, int timeout);
jksoft 24:6ba1245bf049 204 int readPacket(Timer& timer);
jksoft 24:6ba1245bf049 205 int sendPacket(int length, Timer& timer);
jksoft 24:6ba1245bf049 206 int deliverMessage(MQTTString& topicName, Message& message);
jksoft 24:6ba1245bf049 207 bool isTopicMatched(char* topicFilter, MQTTString& topicName);
jksoft 24:6ba1245bf049 208
jksoft 24:6ba1245bf049 209 Network& ipstack;
jksoft 24:6ba1245bf049 210 unsigned long command_timeout_ms;
jksoft 24:6ba1245bf049 211
jksoft 24:6ba1245bf049 212 unsigned char sendbuf[MAX_MQTT_PACKET_SIZE];
jksoft 24:6ba1245bf049 213 unsigned char readbuf[MAX_MQTT_PACKET_SIZE];
jksoft 24:6ba1245bf049 214
jksoft 24:6ba1245bf049 215 Timer last_sent, last_received;
jksoft 24:6ba1245bf049 216 unsigned int keepAliveInterval;
jksoft 24:6ba1245bf049 217 bool ping_outstanding;
jksoft 24:6ba1245bf049 218 bool cleansession;
jksoft 24:6ba1245bf049 219
jksoft 24:6ba1245bf049 220 PacketId packetid;
jksoft 24:6ba1245bf049 221
jksoft 24:6ba1245bf049 222 struct MessageHandlers
jksoft 24:6ba1245bf049 223 {
jksoft 24:6ba1245bf049 224 const char* topicFilter;
jksoft 24:6ba1245bf049 225 FP<void, MessageData&> fp;
jksoft 24:6ba1245bf049 226 } messageHandlers[MAX_MESSAGE_HANDLERS]; // Message handlers are indexed by subscription topic
jksoft 24:6ba1245bf049 227
jksoft 24:6ba1245bf049 228 FP<void, MessageData&> defaultMessageHandler;
jksoft 24:6ba1245bf049 229
jksoft 24:6ba1245bf049 230 bool isconnected;
jksoft 24:6ba1245bf049 231
jksoft 24:6ba1245bf049 232 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 233 unsigned char pubbuf[MAX_MQTT_PACKET_SIZE]; // store the last publish for sending on reconnect
jksoft 24:6ba1245bf049 234 int inflightLen;
jksoft 24:6ba1245bf049 235 unsigned short inflightMsgid;
jksoft 24:6ba1245bf049 236 enum QoS inflightQoS;
jksoft 24:6ba1245bf049 237 #endif
jksoft 24:6ba1245bf049 238
jksoft 24:6ba1245bf049 239 #if MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 240 bool pubrel;
jksoft 24:6ba1245bf049 241 #if !defined(MAX_INCOMING_QOS2_MESSAGES)
jksoft 24:6ba1245bf049 242 #define MAX_INCOMING_QOS2_MESSAGES 10
jksoft 24:6ba1245bf049 243 #endif
jksoft 24:6ba1245bf049 244 unsigned short incomingQoS2messages[MAX_INCOMING_QOS2_MESSAGES];
jksoft 24:6ba1245bf049 245 bool isQoS2msgidFree(unsigned short id);
jksoft 24:6ba1245bf049 246 bool useQoS2msgid(unsigned short id);
jksoft 24:6ba1245bf049 247 void freeQoS2msgid(unsigned short id);
jksoft 24:6ba1245bf049 248 #endif
jksoft 24:6ba1245bf049 249
jksoft 24:6ba1245bf049 250 };
jksoft 24:6ba1245bf049 251
jksoft 24:6ba1245bf049 252 }
jksoft 24:6ba1245bf049 253
jksoft 24:6ba1245bf049 254
jksoft 24:6ba1245bf049 255 template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
jksoft 24:6ba1245bf049 256 void MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::cleanSession()
jksoft 24:6ba1245bf049 257 {
jksoft 24:6ba1245bf049 258 ping_outstanding = false;
jksoft 24:6ba1245bf049 259 for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
jksoft 24:6ba1245bf049 260 messageHandlers[i].topicFilter = 0;
jksoft 24:6ba1245bf049 261 isconnected = false;
jksoft 24:6ba1245bf049 262
jksoft 24:6ba1245bf049 263 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 264 inflightMsgid = 0;
jksoft 24:6ba1245bf049 265 inflightQoS = QOS0;
jksoft 24:6ba1245bf049 266 #endif
jksoft 24:6ba1245bf049 267
jksoft 24:6ba1245bf049 268 #if MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 269 pubrel = false;
jksoft 24:6ba1245bf049 270 for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
jksoft 24:6ba1245bf049 271 incomingQoS2messages[i] = 0;
jksoft 24:6ba1245bf049 272 #endif
jksoft 24:6ba1245bf049 273 }
jksoft 24:6ba1245bf049 274
jksoft 24:6ba1245bf049 275
jksoft 24:6ba1245bf049 276 template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
jksoft 24:6ba1245bf049 277 MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::Client(Network& network, unsigned int command_timeout_ms) : ipstack(network), packetid()
jksoft 24:6ba1245bf049 278 {
jksoft 24:6ba1245bf049 279 last_sent = Timer();
jksoft 24:6ba1245bf049 280 last_received = Timer();
jksoft 24:6ba1245bf049 281 this->command_timeout_ms = command_timeout_ms;
jksoft 24:6ba1245bf049 282 cleanSession();
jksoft 24:6ba1245bf049 283 }
jksoft 24:6ba1245bf049 284
jksoft 24:6ba1245bf049 285
jksoft 24:6ba1245bf049 286 #if MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 287 template<class Network, class Timer, int a, int b>
jksoft 24:6ba1245bf049 288 bool MQTT::Client<Network, Timer, a, b>::isQoS2msgidFree(unsigned short id)
jksoft 24:6ba1245bf049 289 {
jksoft 24:6ba1245bf049 290 for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
jksoft 24:6ba1245bf049 291 {
jksoft 24:6ba1245bf049 292 if (incomingQoS2messages[i] == id)
jksoft 24:6ba1245bf049 293 return false;
jksoft 24:6ba1245bf049 294 }
jksoft 24:6ba1245bf049 295 return true;
jksoft 24:6ba1245bf049 296 }
jksoft 24:6ba1245bf049 297
jksoft 24:6ba1245bf049 298
jksoft 24:6ba1245bf049 299 template<class Network, class Timer, int a, int b>
jksoft 24:6ba1245bf049 300 bool MQTT::Client<Network, Timer, a, b>::useQoS2msgid(unsigned short id)
jksoft 24:6ba1245bf049 301 {
jksoft 24:6ba1245bf049 302 for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
jksoft 24:6ba1245bf049 303 {
jksoft 24:6ba1245bf049 304 if (incomingQoS2messages[i] == 0)
jksoft 24:6ba1245bf049 305 {
jksoft 24:6ba1245bf049 306 incomingQoS2messages[i] = id;
jksoft 24:6ba1245bf049 307 return true;
jksoft 24:6ba1245bf049 308 }
jksoft 24:6ba1245bf049 309 }
jksoft 24:6ba1245bf049 310 return false;
jksoft 24:6ba1245bf049 311 }
jksoft 24:6ba1245bf049 312
jksoft 24:6ba1245bf049 313
jksoft 24:6ba1245bf049 314 template<class Network, class Timer, int a, int b>
jksoft 24:6ba1245bf049 315 void MQTT::Client<Network, Timer, a, b>::freeQoS2msgid(unsigned short id)
jksoft 24:6ba1245bf049 316 {
jksoft 24:6ba1245bf049 317 for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
jksoft 24:6ba1245bf049 318 {
jksoft 24:6ba1245bf049 319 if (incomingQoS2messages[i] == id)
jksoft 24:6ba1245bf049 320 {
jksoft 24:6ba1245bf049 321 incomingQoS2messages[i] = 0;
jksoft 24:6ba1245bf049 322 return;
jksoft 24:6ba1245bf049 323 }
jksoft 24:6ba1245bf049 324 }
jksoft 24:6ba1245bf049 325 }
jksoft 24:6ba1245bf049 326 #endif
jksoft 24:6ba1245bf049 327
jksoft 24:6ba1245bf049 328
jksoft 24:6ba1245bf049 329 template<class Network, class Timer, int a, int b>
jksoft 24:6ba1245bf049 330 int MQTT::Client<Network, Timer, a, b>::sendPacket(int length, Timer& timer)
jksoft 24:6ba1245bf049 331 {
jksoft 24:6ba1245bf049 332 int rc = FAILURE,
jksoft 24:6ba1245bf049 333 sent = 0;
jksoft 24:6ba1245bf049 334
jksoft 24:6ba1245bf049 335 while (sent < length && !timer.expired())
jksoft 24:6ba1245bf049 336 {
jksoft 24:6ba1245bf049 337 rc = ipstack.write(&sendbuf[sent], length - sent, timer.left_ms());
jksoft 24:6ba1245bf049 338 if (rc < 0) // there was an error writing the data
jksoft 24:6ba1245bf049 339 break;
jksoft 24:6ba1245bf049 340 sent += rc;
jksoft 24:6ba1245bf049 341 }
jksoft 24:6ba1245bf049 342 if (sent == length)
jksoft 24:6ba1245bf049 343 {
jksoft 24:6ba1245bf049 344 if (this->keepAliveInterval > 0)
jksoft 24:6ba1245bf049 345 last_sent.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet
jksoft 24:6ba1245bf049 346 rc = SUCCESS;
jksoft 24:6ba1245bf049 347 }
jksoft 24:6ba1245bf049 348 else
jksoft 24:6ba1245bf049 349 rc = FAILURE;
jksoft 24:6ba1245bf049 350
jksoft 24:6ba1245bf049 351 #if defined(MQTT_DEBUG)
jksoft 24:6ba1245bf049 352 char printbuf[150];
jksoft 24:6ba1245bf049 353 DEBUG("Rc %d from sending packet %s\n", rc, MQTTFormat_toServerString(printbuf, sizeof(printbuf), sendbuf, length));
jksoft 24:6ba1245bf049 354 #endif
jksoft 24:6ba1245bf049 355 return rc;
jksoft 24:6ba1245bf049 356 }
jksoft 24:6ba1245bf049 357
jksoft 24:6ba1245bf049 358
jksoft 24:6ba1245bf049 359 template<class Network, class Timer, int a, int b>
jksoft 24:6ba1245bf049 360 int MQTT::Client<Network, Timer, a, b>::decodePacket(int* value, int timeout)
jksoft 24:6ba1245bf049 361 {
jksoft 24:6ba1245bf049 362 unsigned char c;
jksoft 24:6ba1245bf049 363 int multiplier = 1;
jksoft 24:6ba1245bf049 364 int len = 0;
jksoft 24:6ba1245bf049 365 const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4;
jksoft 24:6ba1245bf049 366
jksoft 24:6ba1245bf049 367 *value = 0;
jksoft 24:6ba1245bf049 368 do
jksoft 24:6ba1245bf049 369 {
jksoft 24:6ba1245bf049 370 int rc = MQTTPACKET_READ_ERROR;
jksoft 24:6ba1245bf049 371
jksoft 24:6ba1245bf049 372 if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
jksoft 24:6ba1245bf049 373 {
jksoft 24:6ba1245bf049 374 rc = MQTTPACKET_READ_ERROR; /* bad data */
jksoft 24:6ba1245bf049 375 goto exit;
jksoft 24:6ba1245bf049 376 }
jksoft 24:6ba1245bf049 377 rc = ipstack.read(&c, 1, timeout);
jksoft 24:6ba1245bf049 378 if (rc != 1)
jksoft 24:6ba1245bf049 379 goto exit;
jksoft 24:6ba1245bf049 380 *value += (c & 127) * multiplier;
jksoft 24:6ba1245bf049 381 multiplier *= 128;
jksoft 24:6ba1245bf049 382 } while ((c & 128) != 0);
jksoft 24:6ba1245bf049 383 exit:
jksoft 24:6ba1245bf049 384 return len;
jksoft 24:6ba1245bf049 385 }
jksoft 24:6ba1245bf049 386
jksoft 24:6ba1245bf049 387
jksoft 24:6ba1245bf049 388 /**
jksoft 24:6ba1245bf049 389 * If any read fails in this method, then we should disconnect from the network, as on reconnect
jksoft 24:6ba1245bf049 390 * the packets can be retried.
jksoft 24:6ba1245bf049 391 * @param timeout the max time to wait for the packet read to complete, in milliseconds
jksoft 24:6ba1245bf049 392 * @return the MQTT packet type, or -1 if none
jksoft 24:6ba1245bf049 393 */
jksoft 24:6ba1245bf049 394 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 395 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::readPacket(Timer& timer)
jksoft 24:6ba1245bf049 396 {
jksoft 24:6ba1245bf049 397 int rc = FAILURE;
jksoft 24:6ba1245bf049 398 MQTTHeader header = {0};
jksoft 24:6ba1245bf049 399 int len = 0;
jksoft 24:6ba1245bf049 400 int rem_len = 0;
jksoft 24:6ba1245bf049 401
jksoft 24:6ba1245bf049 402 /* 1. read the header byte. This has the packet type in it */
jksoft 24:6ba1245bf049 403 if (ipstack.read(readbuf, 1, timer.left_ms()) != 1)
jksoft 24:6ba1245bf049 404 goto exit;
jksoft 24:6ba1245bf049 405
jksoft 24:6ba1245bf049 406 len = 1;
jksoft 24:6ba1245bf049 407 /* 2. read the remaining length. This is variable in itself */
jksoft 24:6ba1245bf049 408 decodePacket(&rem_len, timer.left_ms());
jksoft 24:6ba1245bf049 409 len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length into the buffer */
jksoft 24:6ba1245bf049 410
jksoft 24:6ba1245bf049 411 if (rem_len > (MAX_MQTT_PACKET_SIZE - len))
jksoft 24:6ba1245bf049 412 {
jksoft 24:6ba1245bf049 413 rc = BUFFER_OVERFLOW;
jksoft 24:6ba1245bf049 414 goto exit;
jksoft 24:6ba1245bf049 415 }
jksoft 24:6ba1245bf049 416
jksoft 24:6ba1245bf049 417 /* 3. read the rest of the buffer using a callback to supply the rest of the data */
jksoft 24:6ba1245bf049 418 if (rem_len > 0 && (ipstack.read(readbuf + len, rem_len, timer.left_ms()) != rem_len))
jksoft 24:6ba1245bf049 419 goto exit;
jksoft 24:6ba1245bf049 420
jksoft 24:6ba1245bf049 421 header.byte = readbuf[0];
jksoft 24:6ba1245bf049 422 rc = header.bits.type;
jksoft 24:6ba1245bf049 423 if (this->keepAliveInterval > 0)
jksoft 24:6ba1245bf049 424 last_received.countdown(this->keepAliveInterval); // record the fact that we have successfully received a packet
jksoft 24:6ba1245bf049 425 exit:
jksoft 24:6ba1245bf049 426
jksoft 24:6ba1245bf049 427 #if defined(MQTT_DEBUG)
jksoft 24:6ba1245bf049 428 if (rc >= 0)
jksoft 24:6ba1245bf049 429 {
jksoft 24:6ba1245bf049 430 char printbuf[50];
jksoft 24:6ba1245bf049 431 DEBUG("Rc %d from receiving packet %s\n", rc, MQTTFormat_toClientString(printbuf, sizeof(printbuf), readbuf, len));
jksoft 24:6ba1245bf049 432 }
jksoft 24:6ba1245bf049 433 #endif
jksoft 24:6ba1245bf049 434 return rc;
jksoft 24:6ba1245bf049 435 }
jksoft 24:6ba1245bf049 436
jksoft 24:6ba1245bf049 437
jksoft 24:6ba1245bf049 438 // assume topic filter and name is in correct format
jksoft 24:6ba1245bf049 439 // # can only be at end
jksoft 24:6ba1245bf049 440 // + and # can only be next to separator
jksoft 24:6ba1245bf049 441 template<class Network, class Timer, int a, int b>
jksoft 24:6ba1245bf049 442 bool MQTT::Client<Network, Timer, a, b>::isTopicMatched(char* topicFilter, MQTTString& topicName)
jksoft 24:6ba1245bf049 443 {
jksoft 24:6ba1245bf049 444 char* curf = topicFilter;
jksoft 24:6ba1245bf049 445 char* curn = topicName.lenstring.data;
jksoft 24:6ba1245bf049 446 char* curn_end = curn + topicName.lenstring.len;
jksoft 24:6ba1245bf049 447
jksoft 24:6ba1245bf049 448 while (*curf && curn < curn_end)
jksoft 24:6ba1245bf049 449 {
jksoft 24:6ba1245bf049 450 if (*curn == '/' && *curf != '/')
jksoft 24:6ba1245bf049 451 break;
jksoft 24:6ba1245bf049 452 if (*curf != '+' && *curf != '#' && *curf != *curn)
jksoft 24:6ba1245bf049 453 break;
jksoft 24:6ba1245bf049 454 if (*curf == '+')
jksoft 24:6ba1245bf049 455 { // skip until we meet the next separator, or end of string
jksoft 24:6ba1245bf049 456 char* nextpos = curn + 1;
jksoft 24:6ba1245bf049 457 while (nextpos < curn_end && *nextpos != '/')
jksoft 24:6ba1245bf049 458 nextpos = ++curn + 1;
jksoft 24:6ba1245bf049 459 }
jksoft 24:6ba1245bf049 460 else if (*curf == '#')
jksoft 24:6ba1245bf049 461 curn = curn_end - 1; // skip until end of string
jksoft 24:6ba1245bf049 462 curf++;
jksoft 24:6ba1245bf049 463 curn++;
jksoft 24:6ba1245bf049 464 };
jksoft 24:6ba1245bf049 465
jksoft 24:6ba1245bf049 466 return (curn == curn_end) && (*curf == '\0');
jksoft 24:6ba1245bf049 467 }
jksoft 24:6ba1245bf049 468
jksoft 24:6ba1245bf049 469
jksoft 24:6ba1245bf049 470
jksoft 24:6ba1245bf049 471 template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
jksoft 24:6ba1245bf049 472 int MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::deliverMessage(MQTTString& topicName, Message& message)
jksoft 24:6ba1245bf049 473 {
jksoft 24:6ba1245bf049 474 int rc = FAILURE;
jksoft 24:6ba1245bf049 475
jksoft 24:6ba1245bf049 476 // we have to find the right message handler - indexed by topic
jksoft 24:6ba1245bf049 477 for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
jksoft 24:6ba1245bf049 478 {
jksoft 24:6ba1245bf049 479 if (messageHandlers[i].topicFilter != 0 && (MQTTPacket_equals(&topicName, (char*)messageHandlers[i].topicFilter) ||
jksoft 24:6ba1245bf049 480 isTopicMatched((char*)messageHandlers[i].topicFilter, topicName)))
jksoft 24:6ba1245bf049 481 {
jksoft 24:6ba1245bf049 482 if (messageHandlers[i].fp.attached())
jksoft 24:6ba1245bf049 483 {
jksoft 24:6ba1245bf049 484 MessageData md(topicName, message);
jksoft 24:6ba1245bf049 485 messageHandlers[i].fp(md);
jksoft 24:6ba1245bf049 486 rc = SUCCESS;
jksoft 24:6ba1245bf049 487 }
jksoft 24:6ba1245bf049 488 }
jksoft 24:6ba1245bf049 489 }
jksoft 24:6ba1245bf049 490
jksoft 24:6ba1245bf049 491 if (rc == FAILURE && defaultMessageHandler.attached())
jksoft 24:6ba1245bf049 492 {
jksoft 24:6ba1245bf049 493 MessageData md(topicName, message);
jksoft 24:6ba1245bf049 494 defaultMessageHandler(md);
jksoft 24:6ba1245bf049 495 rc = SUCCESS;
jksoft 24:6ba1245bf049 496 }
jksoft 24:6ba1245bf049 497
jksoft 24:6ba1245bf049 498 return rc;
jksoft 24:6ba1245bf049 499 }
jksoft 24:6ba1245bf049 500
jksoft 24:6ba1245bf049 501
jksoft 24:6ba1245bf049 502
jksoft 24:6ba1245bf049 503 template<class Network, class Timer, int a, int b>
jksoft 24:6ba1245bf049 504 int MQTT::Client<Network, Timer, a, b>::yield(unsigned long timeout_ms)
jksoft 24:6ba1245bf049 505 {
jksoft 24:6ba1245bf049 506 int rc = SUCCESS;
jksoft 24:6ba1245bf049 507 Timer timer = Timer();
jksoft 24:6ba1245bf049 508
jksoft 24:6ba1245bf049 509 timer.countdown_ms(timeout_ms);
jksoft 24:6ba1245bf049 510 while (!timer.expired())
jksoft 24:6ba1245bf049 511 {
jksoft 24:6ba1245bf049 512 if (cycle(timer) < 0)
jksoft 24:6ba1245bf049 513 {
jksoft 24:6ba1245bf049 514 rc = FAILURE;
jksoft 24:6ba1245bf049 515 break;
jksoft 24:6ba1245bf049 516 }
jksoft 24:6ba1245bf049 517 Thread::wait(10);
jksoft 24:6ba1245bf049 518 }
jksoft 24:6ba1245bf049 519
jksoft 24:6ba1245bf049 520 return rc;
jksoft 24:6ba1245bf049 521 }
jksoft 24:6ba1245bf049 522
jksoft 24:6ba1245bf049 523
jksoft 24:6ba1245bf049 524 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 525 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::cycle(Timer& timer)
jksoft 24:6ba1245bf049 526 {
jksoft 24:6ba1245bf049 527 /* get one piece of work off the wire and one pass through */
jksoft 24:6ba1245bf049 528
jksoft 24:6ba1245bf049 529 // read the socket, see what work is due
jksoft 24:6ba1245bf049 530 int packet_type = readPacket(timer);
jksoft 24:6ba1245bf049 531
jksoft 24:6ba1245bf049 532 int len = 0,
jksoft 24:6ba1245bf049 533 rc = SUCCESS;
jksoft 24:6ba1245bf049 534
jksoft 24:6ba1245bf049 535 switch (packet_type)
jksoft 24:6ba1245bf049 536 {
jksoft 24:6ba1245bf049 537 case FAILURE:
jksoft 24:6ba1245bf049 538 case BUFFER_OVERFLOW:
jksoft 24:6ba1245bf049 539 rc = packet_type;
jksoft 24:6ba1245bf049 540 break;
jksoft 24:6ba1245bf049 541 case CONNACK:
jksoft 24:6ba1245bf049 542 case PUBACK:
jksoft 24:6ba1245bf049 543 case SUBACK:
jksoft 24:6ba1245bf049 544 break;
jksoft 24:6ba1245bf049 545 case PUBLISH:
jksoft 24:6ba1245bf049 546 {
jksoft 24:6ba1245bf049 547 MQTTString topicName = MQTTString_initializer;
jksoft 24:6ba1245bf049 548 Message msg;
jksoft 24:6ba1245bf049 549 int intQoS;
jksoft 24:6ba1245bf049 550 if (MQTTDeserialize_publish((unsigned char*)&msg.dup, &intQoS, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicName,
jksoft 24:6ba1245bf049 551 (unsigned char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
jksoft 24:6ba1245bf049 552 goto exit;
jksoft 24:6ba1245bf049 553 msg.qos = (enum QoS)intQoS;
jksoft 24:6ba1245bf049 554 #if MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 555 if (msg.qos != QOS2)
jksoft 24:6ba1245bf049 556 #endif
jksoft 24:6ba1245bf049 557 deliverMessage(topicName, msg);
jksoft 24:6ba1245bf049 558 #if MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 559 else if (isQoS2msgidFree(msg.id))
jksoft 24:6ba1245bf049 560 {
jksoft 24:6ba1245bf049 561 if (useQoS2msgid(msg.id))
jksoft 24:6ba1245bf049 562 deliverMessage(topicName, msg);
jksoft 24:6ba1245bf049 563 else
jksoft 24:6ba1245bf049 564 WARN("Maximum number of incoming QoS2 messages exceeded");
jksoft 24:6ba1245bf049 565 }
jksoft 24:6ba1245bf049 566 #endif
jksoft 24:6ba1245bf049 567 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 568 if (msg.qos != QOS0)
jksoft 24:6ba1245bf049 569 {
jksoft 24:6ba1245bf049 570 if (msg.qos == QOS1)
jksoft 24:6ba1245bf049 571 len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBACK, 0, msg.id);
jksoft 24:6ba1245bf049 572 else if (msg.qos == QOS2)
jksoft 24:6ba1245bf049 573 len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREC, 0, msg.id);
jksoft 24:6ba1245bf049 574 if (len <= 0)
jksoft 24:6ba1245bf049 575 rc = FAILURE;
jksoft 24:6ba1245bf049 576 else
jksoft 24:6ba1245bf049 577 rc = sendPacket(len, timer);
jksoft 24:6ba1245bf049 578 if (rc == FAILURE)
jksoft 24:6ba1245bf049 579 goto exit; // there was a problem
jksoft 24:6ba1245bf049 580 }
jksoft 24:6ba1245bf049 581 break;
jksoft 24:6ba1245bf049 582 #endif
jksoft 24:6ba1245bf049 583 }
jksoft 24:6ba1245bf049 584 #if MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 585 case PUBREC:
jksoft 24:6ba1245bf049 586 case PUBREL:
jksoft 24:6ba1245bf049 587 unsigned short mypacketid;
jksoft 24:6ba1245bf049 588 unsigned char dup, type;
jksoft 24:6ba1245bf049 589 if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
jksoft 24:6ba1245bf049 590 rc = FAILURE;
jksoft 24:6ba1245bf049 591 else if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE,
jksoft 24:6ba1245bf049 592 (packet_type == PUBREC) ? PUBREL : PUBCOMP, 0, mypacketid)) <= 0)
jksoft 24:6ba1245bf049 593 rc = FAILURE;
jksoft 24:6ba1245bf049 594 else if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet
jksoft 24:6ba1245bf049 595 rc = FAILURE; // there was a problem
jksoft 24:6ba1245bf049 596 if (rc == FAILURE)
jksoft 24:6ba1245bf049 597 goto exit; // there was a problem
jksoft 24:6ba1245bf049 598 if (packet_type == PUBREL)
jksoft 24:6ba1245bf049 599 freeQoS2msgid(mypacketid);
jksoft 24:6ba1245bf049 600 break;
jksoft 24:6ba1245bf049 601
jksoft 24:6ba1245bf049 602 case PUBCOMP:
jksoft 24:6ba1245bf049 603 break;
jksoft 24:6ba1245bf049 604 #endif
jksoft 24:6ba1245bf049 605 case PINGRESP:
jksoft 24:6ba1245bf049 606 ping_outstanding = false;
jksoft 24:6ba1245bf049 607 break;
jksoft 24:6ba1245bf049 608 }
jksoft 24:6ba1245bf049 609 keepalive();
jksoft 24:6ba1245bf049 610 exit:
jksoft 24:6ba1245bf049 611 if (rc == SUCCESS)
jksoft 24:6ba1245bf049 612 rc = packet_type;
jksoft 24:6ba1245bf049 613 return rc;
jksoft 24:6ba1245bf049 614 }
jksoft 24:6ba1245bf049 615
jksoft 24:6ba1245bf049 616
jksoft 24:6ba1245bf049 617 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 618 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::keepalive()
jksoft 24:6ba1245bf049 619 {
jksoft 24:6ba1245bf049 620 int rc = FAILURE;
jksoft 24:6ba1245bf049 621
jksoft 24:6ba1245bf049 622 if (keepAliveInterval == 0)
jksoft 24:6ba1245bf049 623 {
jksoft 24:6ba1245bf049 624 rc = SUCCESS;
jksoft 24:6ba1245bf049 625 goto exit;
jksoft 24:6ba1245bf049 626 }
jksoft 24:6ba1245bf049 627
jksoft 24:6ba1245bf049 628 if (last_sent.expired() || last_received.expired())
jksoft 24:6ba1245bf049 629 {
jksoft 24:6ba1245bf049 630 if (!ping_outstanding)
jksoft 24:6ba1245bf049 631 {
jksoft 24:6ba1245bf049 632 Timer timer(1000);
jksoft 24:6ba1245bf049 633 int len = MQTTSerialize_pingreq(sendbuf, MAX_MQTT_PACKET_SIZE);
jksoft 24:6ba1245bf049 634 if (len > 0 && (rc = sendPacket(len, timer)) == SUCCESS) // send the ping packet
jksoft 24:6ba1245bf049 635 ping_outstanding = true;
jksoft 24:6ba1245bf049 636 }
jksoft 24:6ba1245bf049 637 }
jksoft 24:6ba1245bf049 638
jksoft 24:6ba1245bf049 639 exit:
jksoft 24:6ba1245bf049 640 return rc;
jksoft 24:6ba1245bf049 641 }
jksoft 24:6ba1245bf049 642
jksoft 24:6ba1245bf049 643
jksoft 24:6ba1245bf049 644 // only used in single-threaded mode where one command at a time is in process
jksoft 24:6ba1245bf049 645 template<class Network, class Timer, int a, int b>
jksoft 24:6ba1245bf049 646 int MQTT::Client<Network, Timer, a, b>::waitfor(int packet_type, Timer& timer)
jksoft 24:6ba1245bf049 647 {
jksoft 24:6ba1245bf049 648 int rc = FAILURE;
jksoft 24:6ba1245bf049 649
jksoft 24:6ba1245bf049 650 do
jksoft 24:6ba1245bf049 651 {
jksoft 24:6ba1245bf049 652 if (timer.expired())
jksoft 24:6ba1245bf049 653 break; // we timed out
jksoft 24:6ba1245bf049 654 }
jksoft 24:6ba1245bf049 655 while ((rc = cycle(timer)) != packet_type);
jksoft 24:6ba1245bf049 656
jksoft 24:6ba1245bf049 657 return rc;
jksoft 24:6ba1245bf049 658 }
jksoft 24:6ba1245bf049 659
jksoft 24:6ba1245bf049 660
jksoft 24:6ba1245bf049 661 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 662 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect(MQTTPacket_connectData& options)
jksoft 24:6ba1245bf049 663 {
jksoft 24:6ba1245bf049 664 Timer connect_timer(command_timeout_ms);
jksoft 24:6ba1245bf049 665 int rc = FAILURE;
jksoft 24:6ba1245bf049 666 int len = 0;
jksoft 24:6ba1245bf049 667
jksoft 24:6ba1245bf049 668 if (isconnected) // don't send connect packet again if we are already connected
jksoft 24:6ba1245bf049 669 goto exit;
jksoft 24:6ba1245bf049 670
jksoft 24:6ba1245bf049 671 this->keepAliveInterval = options.keepAliveInterval;
jksoft 24:6ba1245bf049 672 this->cleansession = options.cleansession;
jksoft 24:6ba1245bf049 673 if ((len = MQTTSerialize_connect(sendbuf, MAX_MQTT_PACKET_SIZE, &options)) <= 0)
jksoft 24:6ba1245bf049 674 goto exit;
jksoft 24:6ba1245bf049 675 if ((rc = sendPacket(len, connect_timer)) != SUCCESS) // send the connect packet
jksoft 24:6ba1245bf049 676 goto exit; // there was a problem
jksoft 24:6ba1245bf049 677
jksoft 24:6ba1245bf049 678 if (this->keepAliveInterval > 0)
jksoft 24:6ba1245bf049 679 last_received.countdown(this->keepAliveInterval);
jksoft 24:6ba1245bf049 680 // this will be a blocking call, wait for the connack
jksoft 24:6ba1245bf049 681 if (waitfor(CONNACK, connect_timer) == CONNACK)
jksoft 24:6ba1245bf049 682 {
jksoft 24:6ba1245bf049 683 unsigned char connack_rc = 255;
jksoft 24:6ba1245bf049 684 bool sessionPresent = false;
jksoft 24:6ba1245bf049 685 if (MQTTDeserialize_connack((unsigned char*)&sessionPresent, &connack_rc, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
jksoft 24:6ba1245bf049 686 rc = connack_rc;
jksoft 24:6ba1245bf049 687 else
jksoft 24:6ba1245bf049 688 rc = FAILURE;
jksoft 24:6ba1245bf049 689 }
jksoft 24:6ba1245bf049 690 else
jksoft 24:6ba1245bf049 691 rc = FAILURE;
jksoft 24:6ba1245bf049 692
jksoft 24:6ba1245bf049 693 #if MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 694 // resend any inflight publish
jksoft 24:6ba1245bf049 695 if (inflightMsgid > 0 && inflightQoS == QOS2 && pubrel)
jksoft 24:6ba1245bf049 696 {
jksoft 24:6ba1245bf049 697 if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, inflightMsgid)) <= 0)
jksoft 24:6ba1245bf049 698 rc = FAILURE;
jksoft 24:6ba1245bf049 699 else
jksoft 24:6ba1245bf049 700 rc = publish(len, connect_timer, inflightQoS);
jksoft 24:6ba1245bf049 701 }
jksoft 24:6ba1245bf049 702 else
jksoft 24:6ba1245bf049 703 #endif
jksoft 24:6ba1245bf049 704 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 705 if (inflightMsgid > 0)
jksoft 24:6ba1245bf049 706 {
jksoft 24:6ba1245bf049 707 memcpy(sendbuf, pubbuf, MAX_MQTT_PACKET_SIZE);
jksoft 24:6ba1245bf049 708 rc = publish(inflightLen, connect_timer, inflightQoS);
jksoft 24:6ba1245bf049 709 }
jksoft 24:6ba1245bf049 710 #endif
jksoft 24:6ba1245bf049 711
jksoft 24:6ba1245bf049 712 exit:
jksoft 24:6ba1245bf049 713 if (rc == SUCCESS)
jksoft 24:6ba1245bf049 714 isconnected = true;
jksoft 24:6ba1245bf049 715 return rc;
jksoft 24:6ba1245bf049 716 }
jksoft 24:6ba1245bf049 717
jksoft 24:6ba1245bf049 718
jksoft 24:6ba1245bf049 719 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 720 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect()
jksoft 24:6ba1245bf049 721 {
jksoft 24:6ba1245bf049 722 MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;
jksoft 24:6ba1245bf049 723 return connect(default_options);
jksoft 24:6ba1245bf049 724 }
jksoft 24:6ba1245bf049 725
jksoft 24:6ba1245bf049 726
jksoft 24:6ba1245bf049 727 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS>
jksoft 24:6ba1245bf049 728 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::subscribe(const char* topicFilter, enum QoS qos, messageHandler messageHandler)
jksoft 24:6ba1245bf049 729 {
jksoft 24:6ba1245bf049 730 int rc = FAILURE;
jksoft 24:6ba1245bf049 731 Timer timer(command_timeout_ms);
jksoft 24:6ba1245bf049 732 int len = 0;
jksoft 24:6ba1245bf049 733 MQTTString topic = {(char*)topicFilter, {0, 0}};
jksoft 24:6ba1245bf049 734
jksoft 24:6ba1245bf049 735 if (!isconnected)
jksoft 24:6ba1245bf049 736 goto exit;
jksoft 24:6ba1245bf049 737
jksoft 24:6ba1245bf049 738 len = MQTTSerialize_subscribe(sendbuf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos);
jksoft 24:6ba1245bf049 739 if (len <= 0)
jksoft 24:6ba1245bf049 740 goto exit;
jksoft 24:6ba1245bf049 741 if ((rc = sendPacket(len, timer)) != SUCCESS) // send the subscribe packet
jksoft 24:6ba1245bf049 742 goto exit; // there was a problem
jksoft 24:6ba1245bf049 743
jksoft 24:6ba1245bf049 744 if (waitfor(SUBACK, timer) == SUBACK) // wait for suback
jksoft 24:6ba1245bf049 745 {
jksoft 24:6ba1245bf049 746 int count = 0, grantedQoS = -1;
jksoft 24:6ba1245bf049 747 unsigned short mypacketid;
jksoft 24:6ba1245bf049 748 if (MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
jksoft 24:6ba1245bf049 749 rc = grantedQoS; // 0, 1, 2 or 0x80
jksoft 24:6ba1245bf049 750 if (rc != 0x80)
jksoft 24:6ba1245bf049 751 {
jksoft 24:6ba1245bf049 752 for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
jksoft 24:6ba1245bf049 753 {
jksoft 24:6ba1245bf049 754 if (messageHandlers[i].topicFilter == 0)
jksoft 24:6ba1245bf049 755 {
jksoft 24:6ba1245bf049 756 messageHandlers[i].topicFilter = topicFilter;
jksoft 24:6ba1245bf049 757 messageHandlers[i].fp.attach(messageHandler);
jksoft 24:6ba1245bf049 758 rc = 0;
jksoft 24:6ba1245bf049 759 break;
jksoft 24:6ba1245bf049 760 }
jksoft 24:6ba1245bf049 761 }
jksoft 24:6ba1245bf049 762 }
jksoft 24:6ba1245bf049 763 }
jksoft 24:6ba1245bf049 764 else
jksoft 24:6ba1245bf049 765 rc = FAILURE;
jksoft 24:6ba1245bf049 766
jksoft 24:6ba1245bf049 767 exit:
jksoft 24:6ba1245bf049 768 if (rc != SUCCESS)
jksoft 24:6ba1245bf049 769 cleanSession();
jksoft 24:6ba1245bf049 770 return rc;
jksoft 24:6ba1245bf049 771 }
jksoft 24:6ba1245bf049 772
jksoft 24:6ba1245bf049 773
jksoft 24:6ba1245bf049 774 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int MAX_MESSAGE_HANDLERS>
jksoft 24:6ba1245bf049 775 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::unsubscribe(const char* topicFilter)
jksoft 24:6ba1245bf049 776 {
jksoft 24:6ba1245bf049 777 int rc = FAILURE;
jksoft 24:6ba1245bf049 778 Timer timer(command_timeout_ms);
jksoft 24:6ba1245bf049 779 MQTTString topic = {(char*)topicFilter, {0, 0}};
jksoft 24:6ba1245bf049 780 int len = 0;
jksoft 24:6ba1245bf049 781
jksoft 24:6ba1245bf049 782 if (!isconnected)
jksoft 24:6ba1245bf049 783 goto exit;
jksoft 24:6ba1245bf049 784
jksoft 24:6ba1245bf049 785 if ((len = MQTTSerialize_unsubscribe(sendbuf, MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic)) <= 0)
jksoft 24:6ba1245bf049 786 goto exit;
jksoft 24:6ba1245bf049 787 if ((rc = sendPacket(len, timer)) != SUCCESS) // send the unsubscribe packet
jksoft 24:6ba1245bf049 788 goto exit; // there was a problem
jksoft 24:6ba1245bf049 789
jksoft 24:6ba1245bf049 790 if (waitfor(UNSUBACK, timer) == UNSUBACK)
jksoft 24:6ba1245bf049 791 {
jksoft 24:6ba1245bf049 792 unsigned short mypacketid; // should be the same as the packetid above
jksoft 24:6ba1245bf049 793 if (MQTTDeserialize_unsuback(&mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
jksoft 24:6ba1245bf049 794 {
jksoft 24:6ba1245bf049 795 rc = 0;
jksoft 24:6ba1245bf049 796
jksoft 24:6ba1245bf049 797 // remove the subscription message handler associated with this topic, if there is one
jksoft 24:6ba1245bf049 798 for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
jksoft 24:6ba1245bf049 799 {
jksoft 24:6ba1245bf049 800 if (messageHandlers[i].topicFilter && strcmp(messageHandlers[i].topicFilter, topicFilter) == 0)
jksoft 24:6ba1245bf049 801 {
jksoft 24:6ba1245bf049 802 messageHandlers[i].topicFilter = 0;
jksoft 24:6ba1245bf049 803 break;
jksoft 24:6ba1245bf049 804 }
jksoft 24:6ba1245bf049 805 }
jksoft 24:6ba1245bf049 806 }
jksoft 24:6ba1245bf049 807 }
jksoft 24:6ba1245bf049 808 else
jksoft 24:6ba1245bf049 809 rc = FAILURE;
jksoft 24:6ba1245bf049 810
jksoft 24:6ba1245bf049 811 exit:
jksoft 24:6ba1245bf049 812 if (rc != SUCCESS)
jksoft 24:6ba1245bf049 813 cleanSession();
jksoft 24:6ba1245bf049 814 return rc;
jksoft 24:6ba1245bf049 815 }
jksoft 24:6ba1245bf049 816
jksoft 24:6ba1245bf049 817
jksoft 24:6ba1245bf049 818 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 819 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(int len, Timer& timer, enum QoS qos)
jksoft 24:6ba1245bf049 820 {
jksoft 24:6ba1245bf049 821 int rc;
jksoft 24:6ba1245bf049 822
jksoft 24:6ba1245bf049 823 if ((rc = sendPacket(len, timer)) != SUCCESS) // send the publish packet
jksoft 24:6ba1245bf049 824 goto exit; // there was a problem
jksoft 24:6ba1245bf049 825
jksoft 24:6ba1245bf049 826 #if MQTTCLIENT_QOS1
jksoft 24:6ba1245bf049 827 if (qos == QOS1)
jksoft 24:6ba1245bf049 828 {
jksoft 24:6ba1245bf049 829 if (waitfor(PUBACK, timer) == PUBACK)
jksoft 24:6ba1245bf049 830 {
jksoft 24:6ba1245bf049 831 unsigned short mypacketid;
jksoft 24:6ba1245bf049 832 unsigned char dup, type;
jksoft 24:6ba1245bf049 833 if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
jksoft 24:6ba1245bf049 834 rc = FAILURE;
jksoft 24:6ba1245bf049 835 else if (inflightMsgid == mypacketid)
jksoft 24:6ba1245bf049 836 inflightMsgid = 0;
jksoft 24:6ba1245bf049 837 }
jksoft 24:6ba1245bf049 838 else
jksoft 24:6ba1245bf049 839 rc = FAILURE;
jksoft 24:6ba1245bf049 840 }
jksoft 24:6ba1245bf049 841 #elif MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 842 else if (qos == QOS2)
jksoft 24:6ba1245bf049 843 {
jksoft 24:6ba1245bf049 844 if (waitfor(PUBCOMP, timer) == PUBCOMP)
jksoft 24:6ba1245bf049 845 {
jksoft 24:6ba1245bf049 846 unsigned short mypacketid;
jksoft 24:6ba1245bf049 847 unsigned char dup, type;
jksoft 24:6ba1245bf049 848 if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
jksoft 24:6ba1245bf049 849 rc = FAILURE;
jksoft 24:6ba1245bf049 850 else if (inflightMsgid == mypacketid)
jksoft 24:6ba1245bf049 851 inflightMsgid = 0;
jksoft 24:6ba1245bf049 852 }
jksoft 24:6ba1245bf049 853 else
jksoft 24:6ba1245bf049 854 rc = FAILURE;
jksoft 24:6ba1245bf049 855 }
jksoft 24:6ba1245bf049 856 #endif
jksoft 24:6ba1245bf049 857
jksoft 24:6ba1245bf049 858 exit:
jksoft 24:6ba1245bf049 859 if (rc != SUCCESS)
jksoft 24:6ba1245bf049 860 cleanSession();
jksoft 24:6ba1245bf049 861 return rc;
jksoft 24:6ba1245bf049 862 }
jksoft 24:6ba1245bf049 863
jksoft 24:6ba1245bf049 864
jksoft 24:6ba1245bf049 865
jksoft 24:6ba1245bf049 866 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 867 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)
jksoft 24:6ba1245bf049 868 {
jksoft 24:6ba1245bf049 869 int rc = FAILURE;
jksoft 24:6ba1245bf049 870 Timer timer(command_timeout_ms);
jksoft 24:6ba1245bf049 871 MQTTString topicString = MQTTString_initializer;
jksoft 24:6ba1245bf049 872 int len = 0;
jksoft 24:6ba1245bf049 873
jksoft 24:6ba1245bf049 874 if (!isconnected)
jksoft 24:6ba1245bf049 875 goto exit;
jksoft 24:6ba1245bf049 876
jksoft 24:6ba1245bf049 877 topicString.cstring = (char*)topicName;
jksoft 24:6ba1245bf049 878
jksoft 24:6ba1245bf049 879 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 880 if (qos == QOS1 || qos == QOS2)
jksoft 24:6ba1245bf049 881 id = packetid.getNext();
jksoft 24:6ba1245bf049 882 #endif
jksoft 24:6ba1245bf049 883
jksoft 24:6ba1245bf049 884 len = MQTTSerialize_publish(sendbuf, MAX_MQTT_PACKET_SIZE, 0, qos, retained, id,
jksoft 24:6ba1245bf049 885 topicString, (unsigned char*)payload, payloadlen);
jksoft 24:6ba1245bf049 886 if (len <= 0)
jksoft 24:6ba1245bf049 887 goto exit;
jksoft 24:6ba1245bf049 888
jksoft 24:6ba1245bf049 889 #if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 890 if (!cleansession)
jksoft 24:6ba1245bf049 891 {
jksoft 24:6ba1245bf049 892 memcpy(pubbuf, sendbuf, len);
jksoft 24:6ba1245bf049 893 inflightMsgid = id;
jksoft 24:6ba1245bf049 894 inflightLen = len;
jksoft 24:6ba1245bf049 895 inflightQoS = qos;
jksoft 24:6ba1245bf049 896 #if MQTTCLIENT_QOS2
jksoft 24:6ba1245bf049 897 pubrel = false;
jksoft 24:6ba1245bf049 898 #endif
jksoft 24:6ba1245bf049 899 }
jksoft 24:6ba1245bf049 900 #endif
jksoft 24:6ba1245bf049 901
jksoft 24:6ba1245bf049 902 rc = publish(len, timer, qos);
jksoft 24:6ba1245bf049 903 exit:
jksoft 24:6ba1245bf049 904 return rc;
jksoft 24:6ba1245bf049 905 }
jksoft 24:6ba1245bf049 906
jksoft 24:6ba1245bf049 907
jksoft 24:6ba1245bf049 908 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 909 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, void* payload, size_t payloadlen, enum QoS qos, bool retained)
jksoft 24:6ba1245bf049 910 {
jksoft 24:6ba1245bf049 911 unsigned short id = 0; // dummy - not used for anything
jksoft 24:6ba1245bf049 912 return publish(topicName, payload, payloadlen, id, qos, retained);
jksoft 24:6ba1245bf049 913 }
jksoft 24:6ba1245bf049 914
jksoft 24:6ba1245bf049 915
jksoft 24:6ba1245bf049 916 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 917 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, Message& message)
jksoft 24:6ba1245bf049 918 {
jksoft 24:6ba1245bf049 919 return publish(topicName, message.payload, message.payloadlen, message.qos, message.retained);
jksoft 24:6ba1245bf049 920 }
jksoft 24:6ba1245bf049 921
jksoft 24:6ba1245bf049 922
jksoft 24:6ba1245bf049 923 template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
jksoft 24:6ba1245bf049 924 int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::disconnect()
jksoft 24:6ba1245bf049 925 {
jksoft 24:6ba1245bf049 926 int rc = FAILURE;
jksoft 24:6ba1245bf049 927 Timer timer(command_timeout_ms); // we might wait for incomplete incoming publishes to complete
jksoft 24:6ba1245bf049 928 int len = MQTTSerialize_disconnect(sendbuf, MAX_MQTT_PACKET_SIZE);
jksoft 24:6ba1245bf049 929 if (len > 0)
jksoft 24:6ba1245bf049 930 rc = sendPacket(len, timer); // send the disconnect packet
jksoft 24:6ba1245bf049 931
jksoft 24:6ba1245bf049 932 if (cleansession)
jksoft 24:6ba1245bf049 933 cleanSession();
jksoft 24:6ba1245bf049 934 else
jksoft 24:6ba1245bf049 935 isconnected = false;
jksoft 24:6ba1245bf049 936 return rc;
jksoft 24:6ba1245bf049 937 }
jksoft 24:6ba1245bf049 938
jksoft 24:6ba1245bf049 939
jksoft 24:6ba1245bf049 940 #endif