Added function to return network interface.
MQTTAsync.h@65:ff8eaf67f510, 2020-03-21 (annotated)
- Committer:
- lamell
- Date:
- Sat Mar 21 16:17:57 2020 -0400
- Revision:
- 65:ff8eaf67f510
- Parent:
- 61:feeb4b0a918f
File MQTTClient.h:
Changed MAX_MQTT_PACKET_SIZE = 2048, original size is limited to 100 bytes.
If the message is larher than this limit, then no message will be identified in the server.
Who changed what in which revision?
User | Revision | Line number | New contents of line |
---|---|---|---|
icraggs | 20:cad3d54d7ecf | 1 | /******************************************************************************* |
icraggs | 20:cad3d54d7ecf | 2 | * Copyright (c) 2014 IBM Corp. |
icraggs | 20:cad3d54d7ecf | 3 | * |
icraggs | 20:cad3d54d7ecf | 4 | * All rights reserved. This program and the accompanying materials |
icraggs | 20:cad3d54d7ecf | 5 | * are made available under the terms of the Eclipse Public License v1.0 |
icraggs | 20:cad3d54d7ecf | 6 | * and Eclipse Distribution License v1.0 which accompany this distribution. |
icraggs | 20:cad3d54d7ecf | 7 | * |
icraggs | 20:cad3d54d7ecf | 8 | * The Eclipse Public License is available at |
icraggs | 20:cad3d54d7ecf | 9 | * http://www.eclipse.org/legal/epl-v10.html |
icraggs | 20:cad3d54d7ecf | 10 | * and the Eclipse Distribution License is available at |
icraggs | 20:cad3d54d7ecf | 11 | * http://www.eclipse.org/org/documents/edl-v10.php. |
icraggs | 20:cad3d54d7ecf | 12 | * |
icraggs | 20:cad3d54d7ecf | 13 | * Contributors: |
icraggs | 20:cad3d54d7ecf | 14 | * Ian Craggs - initial API and implementation and/or initial documentation |
icraggs | 20:cad3d54d7ecf | 15 | *******************************************************************************/ |
icraggs | 20:cad3d54d7ecf | 16 | |
icraggs | 21:e918525e529d | 17 | #if !defined(MQTTASYNC_H) |
icraggs | 21:e918525e529d | 18 | #define MQTTASYNC_H |
icraggs | 20:cad3d54d7ecf | 19 | |
icraggs | 20:cad3d54d7ecf | 20 | #include "FP.h" |
icraggs | 20:cad3d54d7ecf | 21 | #include "MQTTPacket.h" |
icraggs | 20:cad3d54d7ecf | 22 | #include "stdio.h" |
icraggs | 20:cad3d54d7ecf | 23 | |
icraggs | 20:cad3d54d7ecf | 24 | namespace MQTT |
icraggs | 20:cad3d54d7ecf | 25 | { |
icraggs | 20:cad3d54d7ecf | 26 | |
icraggs | 20:cad3d54d7ecf | 27 | |
icraggs | 20:cad3d54d7ecf | 28 | enum QoS { QOS0, QOS1, QOS2 }; |
icraggs | 20:cad3d54d7ecf | 29 | |
icraggs | 20:cad3d54d7ecf | 30 | |
icraggs | 20:cad3d54d7ecf | 31 | struct Message |
icraggs | 20:cad3d54d7ecf | 32 | { |
icraggs | 20:cad3d54d7ecf | 33 | enum QoS qos; |
icraggs | 20:cad3d54d7ecf | 34 | bool retained; |
icraggs | 20:cad3d54d7ecf | 35 | bool dup; |
icraggs | 20:cad3d54d7ecf | 36 | unsigned short id; |
icraggs | 20:cad3d54d7ecf | 37 | void *payload; |
icraggs | 20:cad3d54d7ecf | 38 | size_t payloadlen; |
icraggs | 20:cad3d54d7ecf | 39 | }; |
icraggs | 20:cad3d54d7ecf | 40 | |
icraggs | 20:cad3d54d7ecf | 41 | |
icraggs | 20:cad3d54d7ecf | 42 | class PacketId |
icraggs | 20:cad3d54d7ecf | 43 | { |
icraggs | 20:cad3d54d7ecf | 44 | public: |
icraggs | 20:cad3d54d7ecf | 45 | PacketId(); |
icraggs | 20:cad3d54d7ecf | 46 | |
icraggs | 20:cad3d54d7ecf | 47 | int getNext(); |
icraggs | 20:cad3d54d7ecf | 48 | |
icraggs | 20:cad3d54d7ecf | 49 | private: |
icraggs | 20:cad3d54d7ecf | 50 | static const int MAX_PACKET_ID = 65535; |
icraggs | 20:cad3d54d7ecf | 51 | int next; |
icraggs | 20:cad3d54d7ecf | 52 | }; |
icraggs | 20:cad3d54d7ecf | 53 | |
icraggs | 20:cad3d54d7ecf | 54 | typedef void (*messageHandler)(Message*); |
icraggs | 20:cad3d54d7ecf | 55 | |
icraggs | 20:cad3d54d7ecf | 56 | typedef struct limits |
icraggs | 20:cad3d54d7ecf | 57 | { |
icraggs | 20:cad3d54d7ecf | 58 | int MAX_MQTT_PACKET_SIZE; // |
icraggs | 20:cad3d54d7ecf | 59 | int MAX_MESSAGE_HANDLERS; // each subscription requires a message handler |
icraggs | 20:cad3d54d7ecf | 60 | int MAX_CONCURRENT_OPERATIONS; // each command which runs concurrently can have a result handler, when we are in multi-threaded mode |
icraggs | 21:e918525e529d | 61 | int command_timeout_ms; |
icraggs | 20:cad3d54d7ecf | 62 | |
icraggs | 20:cad3d54d7ecf | 63 | limits() |
icraggs | 20:cad3d54d7ecf | 64 | { |
lamell | 61:feeb4b0a918f | 65 | MAX_MQTT_PACKET_SIZE = 1024; |
icraggs | 20:cad3d54d7ecf | 66 | MAX_MESSAGE_HANDLERS = 5; |
icraggs | 20:cad3d54d7ecf | 67 | MAX_CONCURRENT_OPERATIONS = 1; // 1 indicates single-threaded mode - set to >1 for multithreaded mode |
icraggs | 21:e918525e529d | 68 | command_timeout_ms = 30000; |
icraggs | 20:cad3d54d7ecf | 69 | } |
icraggs | 20:cad3d54d7ecf | 70 | } Limits; |
icraggs | 20:cad3d54d7ecf | 71 | |
icraggs | 22:aadb79d29330 | 72 | |
icraggs | 22:aadb79d29330 | 73 | /** |
icraggs | 22:aadb79d29330 | 74 | * @class Async |
icraggs | 22:aadb79d29330 | 75 | * @brief non-blocking, threaded MQTT client API |
icraggs | 22:aadb79d29330 | 76 | * @param Network a network class which supports send, receive |
icraggs | 22:aadb79d29330 | 77 | * @param Timer a timer class with the methods: |
icraggs | 22:aadb79d29330 | 78 | */ |
icraggs | 21:e918525e529d | 79 | template<class Network, class Timer, class Thread, class Mutex> class Async |
icraggs | 20:cad3d54d7ecf | 80 | { |
icraggs | 20:cad3d54d7ecf | 81 | |
icraggs | 20:cad3d54d7ecf | 82 | public: |
icraggs | 20:cad3d54d7ecf | 83 | |
icraggs | 20:cad3d54d7ecf | 84 | struct Result |
icraggs | 20:cad3d54d7ecf | 85 | { |
icraggs | 20:cad3d54d7ecf | 86 | /* success or failure result data */ |
icraggs | 21:e918525e529d | 87 | Async<Network, Timer, Thread, Mutex>* client; |
icraggs | 21:e918525e529d | 88 | int rc; |
icraggs | 20:cad3d54d7ecf | 89 | }; |
icraggs | 20:cad3d54d7ecf | 90 | |
icraggs | 20:cad3d54d7ecf | 91 | typedef void (*resultHandler)(Result*); |
icraggs | 20:cad3d54d7ecf | 92 | |
icraggs | 21:e918525e529d | 93 | Async(Network* network, const Limits limits = Limits()); |
icraggs | 21:e918525e529d | 94 | |
icraggs | 21:e918525e529d | 95 | typedef struct |
icraggs | 21:e918525e529d | 96 | { |
icraggs | 21:e918525e529d | 97 | Async* client; |
icraggs | 21:e918525e529d | 98 | Network* network; |
icraggs | 21:e918525e529d | 99 | } connectionLostInfo; |
icraggs | 21:e918525e529d | 100 | |
icraggs | 21:e918525e529d | 101 | typedef int (*connectionLostHandlers)(connectionLostInfo*); |
icraggs | 21:e918525e529d | 102 | |
icraggs | 21:e918525e529d | 103 | /** Set the connection lost callback - called whenever the connection is lost and we should be connected |
icraggs | 21:e918525e529d | 104 | * @param clh - pointer to the callback function |
icraggs | 21:e918525e529d | 105 | */ |
icraggs | 21:e918525e529d | 106 | void setConnectionLostHandler(connectionLostHandlers clh) |
icraggs | 21:e918525e529d | 107 | { |
icraggs | 21:e918525e529d | 108 | connectionLostHandler.attach(clh); |
icraggs | 21:e918525e529d | 109 | } |
icraggs | 21:e918525e529d | 110 | |
icraggs | 21:e918525e529d | 111 | /** Set the default message handling callback - used for any message which does not match a subscription message handler |
icraggs | 21:e918525e529d | 112 | * @param mh - pointer to the callback function |
icraggs | 21:e918525e529d | 113 | */ |
icraggs | 21:e918525e529d | 114 | void setDefaultMessageHandler(messageHandler mh) |
icraggs | 21:e918525e529d | 115 | { |
icraggs | 21:e918525e529d | 116 | defaultMessageHandler.attach(mh); |
icraggs | 21:e918525e529d | 117 | } |
icraggs | 20:cad3d54d7ecf | 118 | |
icraggs | 21:e918525e529d | 119 | int connect(resultHandler fn, MQTTPacket_connectData* options = 0); |
icraggs | 20:cad3d54d7ecf | 120 | |
icraggs | 20:cad3d54d7ecf | 121 | template<class T> |
icraggs | 21:e918525e529d | 122 | int connect(void(T::*method)(Result *), MQTTPacket_connectData* options = 0, T *item = 0); // alternative to pass in pointer to member function |
icraggs | 20:cad3d54d7ecf | 123 | |
icraggs | 21:e918525e529d | 124 | int publish(resultHandler rh, const char* topic, Message* message); |
icraggs | 20:cad3d54d7ecf | 125 | |
icraggs | 21:e918525e529d | 126 | int subscribe(resultHandler rh, const char* topicFilter, enum QoS qos, messageHandler mh); |
icraggs | 20:cad3d54d7ecf | 127 | |
icraggs | 21:e918525e529d | 128 | int unsubscribe(resultHandler rh, const char* topicFilter); |
icraggs | 20:cad3d54d7ecf | 129 | |
icraggs | 21:e918525e529d | 130 | int disconnect(resultHandler rh); |
icraggs | 20:cad3d54d7ecf | 131 | |
icraggs | 20:cad3d54d7ecf | 132 | private: |
icraggs | 20:cad3d54d7ecf | 133 | |
icraggs | 20:cad3d54d7ecf | 134 | void run(void const *argument); |
icraggs | 20:cad3d54d7ecf | 135 | int cycle(int timeout); |
icraggs | 20:cad3d54d7ecf | 136 | int waitfor(int packet_type, Timer& atimer); |
icraggs | 20:cad3d54d7ecf | 137 | int keepalive(); |
icraggs | 20:cad3d54d7ecf | 138 | int findFreeOperation(); |
icraggs | 20:cad3d54d7ecf | 139 | |
icraggs | 20:cad3d54d7ecf | 140 | int decodePacket(int* value, int timeout); |
icraggs | 20:cad3d54d7ecf | 141 | int readPacket(int timeout); |
icraggs | 20:cad3d54d7ecf | 142 | int sendPacket(int length, int timeout); |
icraggs | 20:cad3d54d7ecf | 143 | int deliverMessage(MQTTString* topic, Message* message); |
icraggs | 20:cad3d54d7ecf | 144 | |
icraggs | 20:cad3d54d7ecf | 145 | Thread* thread; |
icraggs | 20:cad3d54d7ecf | 146 | Network* ipstack; |
icraggs | 20:cad3d54d7ecf | 147 | |
icraggs | 20:cad3d54d7ecf | 148 | Limits limits; |
icraggs | 20:cad3d54d7ecf | 149 | |
icraggs | 20:cad3d54d7ecf | 150 | char* buf; |
icraggs | 20:cad3d54d7ecf | 151 | char* readbuf; |
icraggs | 20:cad3d54d7ecf | 152 | |
icraggs | 20:cad3d54d7ecf | 153 | Timer ping_timer, connect_timer; |
icraggs | 20:cad3d54d7ecf | 154 | unsigned int keepAliveInterval; |
icraggs | 20:cad3d54d7ecf | 155 | bool ping_outstanding; |
icraggs | 20:cad3d54d7ecf | 156 | |
icraggs | 20:cad3d54d7ecf | 157 | PacketId packetid; |
icraggs | 20:cad3d54d7ecf | 158 | |
icraggs | 20:cad3d54d7ecf | 159 | typedef FP<void, Result*> resultHandlerFP; |
icraggs | 20:cad3d54d7ecf | 160 | resultHandlerFP connectHandler; |
icraggs | 20:cad3d54d7ecf | 161 | |
icraggs | 20:cad3d54d7ecf | 162 | typedef FP<void, Message*> messageHandlerFP; |
icraggs | 20:cad3d54d7ecf | 163 | struct MessageHandlers |
icraggs | 20:cad3d54d7ecf | 164 | { |
icraggs | 20:cad3d54d7ecf | 165 | const char* topic; |
icraggs | 20:cad3d54d7ecf | 166 | messageHandlerFP fp; |
icraggs | 20:cad3d54d7ecf | 167 | } *messageHandlers; // Message handlers are indexed by subscription topic |
icraggs | 20:cad3d54d7ecf | 168 | |
icraggs | 20:cad3d54d7ecf | 169 | // how many concurrent operations should we allow? Each one will require a function pointer |
icraggs | 20:cad3d54d7ecf | 170 | struct Operations |
icraggs | 20:cad3d54d7ecf | 171 | { |
icraggs | 20:cad3d54d7ecf | 172 | unsigned short id; |
icraggs | 20:cad3d54d7ecf | 173 | resultHandlerFP fp; |
icraggs | 20:cad3d54d7ecf | 174 | const char* topic; // if this is a publish, store topic name in case republishing is required |
icraggs | 20:cad3d54d7ecf | 175 | Message* message; // for publish, |
icraggs | 20:cad3d54d7ecf | 176 | Timer timer; // to check if the command has timed out |
icraggs | 20:cad3d54d7ecf | 177 | } *operations; // result handlers are indexed by packet ids |
icraggs | 20:cad3d54d7ecf | 178 | |
icraggs | 20:cad3d54d7ecf | 179 | static void threadfn(void* arg); |
icraggs | 21:e918525e529d | 180 | |
icraggs | 21:e918525e529d | 181 | messageHandlerFP defaultMessageHandler; |
icraggs | 21:e918525e529d | 182 | |
icraggs | 21:e918525e529d | 183 | typedef FP<int, connectionLostInfo*> connectionLostFP; |
icraggs | 21:e918525e529d | 184 | |
icraggs | 21:e918525e529d | 185 | connectionLostFP connectionLostHandler; |
icraggs | 20:cad3d54d7ecf | 186 | |
icraggs | 20:cad3d54d7ecf | 187 | }; |
icraggs | 20:cad3d54d7ecf | 188 | |
icraggs | 20:cad3d54d7ecf | 189 | } |
icraggs | 20:cad3d54d7ecf | 190 | |
icraggs | 20:cad3d54d7ecf | 191 | |
icraggs | 21:e918525e529d | 192 | template<class Network, class Timer, class Thread, class Mutex> void MQTT::Async<Network, Timer, Thread, Mutex>::threadfn(void* arg) |
icraggs | 20:cad3d54d7ecf | 193 | { |
icraggs | 21:e918525e529d | 194 | ((Async<Network, Timer, Thread, Mutex>*) arg)->run(NULL); |
icraggs | 20:cad3d54d7ecf | 195 | } |
icraggs | 20:cad3d54d7ecf | 196 | |
icraggs | 20:cad3d54d7ecf | 197 | |
icraggs | 21:e918525e529d | 198 | template<class Network, class Timer, class Thread, class Mutex> MQTT::Async<Network, Timer, Thread, Mutex>::Async(Network* network, Limits limits) : limits(limits), packetid() |
icraggs | 20:cad3d54d7ecf | 199 | { |
icraggs | 20:cad3d54d7ecf | 200 | this->thread = 0; |
icraggs | 20:cad3d54d7ecf | 201 | this->ipstack = network; |
icraggs | 20:cad3d54d7ecf | 202 | this->ping_timer = Timer(); |
icraggs | 20:cad3d54d7ecf | 203 | this->ping_outstanding = 0; |
icraggs | 20:cad3d54d7ecf | 204 | |
icraggs | 20:cad3d54d7ecf | 205 | // How to make these memory allocations portable? I was hoping to avoid the heap |
icraggs | 20:cad3d54d7ecf | 206 | buf = new char[limits.MAX_MQTT_PACKET_SIZE]; |
icraggs | 20:cad3d54d7ecf | 207 | readbuf = new char[limits.MAX_MQTT_PACKET_SIZE]; |
icraggs | 20:cad3d54d7ecf | 208 | this->operations = new struct Operations[limits.MAX_CONCURRENT_OPERATIONS]; |
icraggs | 20:cad3d54d7ecf | 209 | for (int i = 0; i < limits.MAX_CONCURRENT_OPERATIONS; ++i) |
icraggs | 20:cad3d54d7ecf | 210 | operations[i].id = 0; |
icraggs | 20:cad3d54d7ecf | 211 | this->messageHandlers = new struct MessageHandlers[limits.MAX_MESSAGE_HANDLERS]; |
icraggs | 20:cad3d54d7ecf | 212 | for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i) |
icraggs | 20:cad3d54d7ecf | 213 | messageHandlers[i].topic = 0; |
icraggs | 20:cad3d54d7ecf | 214 | } |
icraggs | 20:cad3d54d7ecf | 215 | |
icraggs | 20:cad3d54d7ecf | 216 | |
icraggs | 21:e918525e529d | 217 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::sendPacket(int length, int timeout) |
icraggs | 20:cad3d54d7ecf | 218 | { |
icraggs | 20:cad3d54d7ecf | 219 | int sent = 0; |
icraggs | 20:cad3d54d7ecf | 220 | |
icraggs | 20:cad3d54d7ecf | 221 | while (sent < length) |
icraggs | 20:cad3d54d7ecf | 222 | sent += ipstack->write(&buf[sent], length, timeout); |
icraggs | 20:cad3d54d7ecf | 223 | if (sent == length) |
icraggs | 20:cad3d54d7ecf | 224 | ping_timer.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet |
icraggs | 20:cad3d54d7ecf | 225 | return sent; |
icraggs | 20:cad3d54d7ecf | 226 | } |
icraggs | 20:cad3d54d7ecf | 227 | |
icraggs | 20:cad3d54d7ecf | 228 | |
icraggs | 21:e918525e529d | 229 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::decodePacket(int* value, int timeout) |
icraggs | 20:cad3d54d7ecf | 230 | { |
icraggs | 20:cad3d54d7ecf | 231 | char c; |
icraggs | 20:cad3d54d7ecf | 232 | int multiplier = 1; |
icraggs | 20:cad3d54d7ecf | 233 | int len = 0; |
icraggs | 20:cad3d54d7ecf | 234 | const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4; |
icraggs | 20:cad3d54d7ecf | 235 | |
icraggs | 20:cad3d54d7ecf | 236 | *value = 0; |
icraggs | 20:cad3d54d7ecf | 237 | do |
icraggs | 20:cad3d54d7ecf | 238 | { |
icraggs | 20:cad3d54d7ecf | 239 | int rc = MQTTPACKET_READ_ERROR; |
icraggs | 20:cad3d54d7ecf | 240 | |
icraggs | 20:cad3d54d7ecf | 241 | if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES) |
icraggs | 20:cad3d54d7ecf | 242 | { |
icraggs | 20:cad3d54d7ecf | 243 | rc = MQTTPACKET_READ_ERROR; /* bad data */ |
icraggs | 20:cad3d54d7ecf | 244 | goto exit; |
icraggs | 20:cad3d54d7ecf | 245 | } |
icraggs | 20:cad3d54d7ecf | 246 | rc = ipstack->read(&c, 1, timeout); |
icraggs | 20:cad3d54d7ecf | 247 | if (rc != 1) |
icraggs | 20:cad3d54d7ecf | 248 | goto exit; |
icraggs | 20:cad3d54d7ecf | 249 | *value += (c & 127) * multiplier; |
icraggs | 20:cad3d54d7ecf | 250 | multiplier *= 128; |
icraggs | 20:cad3d54d7ecf | 251 | } while ((c & 128) != 0); |
icraggs | 20:cad3d54d7ecf | 252 | exit: |
icraggs | 20:cad3d54d7ecf | 253 | return len; |
icraggs | 20:cad3d54d7ecf | 254 | } |
icraggs | 20:cad3d54d7ecf | 255 | |
icraggs | 20:cad3d54d7ecf | 256 | |
icraggs | 20:cad3d54d7ecf | 257 | /** |
icraggs | 20:cad3d54d7ecf | 258 | * If any read fails in this method, then we should disconnect from the network, as on reconnect |
icraggs | 20:cad3d54d7ecf | 259 | * the packets can be retried. |
icraggs | 20:cad3d54d7ecf | 260 | * @param timeout the max time to wait for the packet read to complete, in milliseconds |
icraggs | 20:cad3d54d7ecf | 261 | * @return the MQTT packet type, or -1 if none |
icraggs | 20:cad3d54d7ecf | 262 | */ |
icraggs | 21:e918525e529d | 263 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::readPacket(int timeout) |
icraggs | 20:cad3d54d7ecf | 264 | { |
icraggs | 20:cad3d54d7ecf | 265 | int rc = -1; |
icraggs | 20:cad3d54d7ecf | 266 | MQTTHeader header = {0}; |
icraggs | 20:cad3d54d7ecf | 267 | int len = 0; |
icraggs | 20:cad3d54d7ecf | 268 | int rem_len = 0; |
icraggs | 20:cad3d54d7ecf | 269 | |
icraggs | 20:cad3d54d7ecf | 270 | /* 1. read the header byte. This has the packet type in it */ |
icraggs | 20:cad3d54d7ecf | 271 | if (ipstack->read(readbuf, 1, timeout) != 1) |
icraggs | 20:cad3d54d7ecf | 272 | goto exit; |
icraggs | 20:cad3d54d7ecf | 273 | |
icraggs | 20:cad3d54d7ecf | 274 | len = 1; |
icraggs | 20:cad3d54d7ecf | 275 | /* 2. read the remaining length. This is variable in itself */ |
icraggs | 20:cad3d54d7ecf | 276 | decodePacket(&rem_len, timeout); |
icraggs | 20:cad3d54d7ecf | 277 | len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length back into the buffer */ |
icraggs | 20:cad3d54d7ecf | 278 | |
icraggs | 20:cad3d54d7ecf | 279 | /* 3. read the rest of the buffer using a callback to supply the rest of the data */ |
icraggs | 20:cad3d54d7ecf | 280 | if (ipstack->read(readbuf + len, rem_len, timeout) != rem_len) |
icraggs | 20:cad3d54d7ecf | 281 | goto exit; |
icraggs | 20:cad3d54d7ecf | 282 | |
icraggs | 20:cad3d54d7ecf | 283 | header.byte = readbuf[0]; |
icraggs | 20:cad3d54d7ecf | 284 | rc = header.bits.type; |
icraggs | 20:cad3d54d7ecf | 285 | exit: |
icraggs | 20:cad3d54d7ecf | 286 | return rc; |
icraggs | 20:cad3d54d7ecf | 287 | } |
icraggs | 20:cad3d54d7ecf | 288 | |
icraggs | 20:cad3d54d7ecf | 289 | |
icraggs | 21:e918525e529d | 290 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::deliverMessage(MQTTString* topic, Message* message) |
icraggs | 20:cad3d54d7ecf | 291 | { |
icraggs | 20:cad3d54d7ecf | 292 | int rc = -1; |
icraggs | 20:cad3d54d7ecf | 293 | |
icraggs | 20:cad3d54d7ecf | 294 | // we have to find the right message handler - indexed by topic |
icraggs | 20:cad3d54d7ecf | 295 | for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i) |
icraggs | 20:cad3d54d7ecf | 296 | { |
icraggs | 20:cad3d54d7ecf | 297 | if (messageHandlers[i].topic != 0 && MQTTPacket_equals(topic, (char*)messageHandlers[i].topic)) |
icraggs | 20:cad3d54d7ecf | 298 | { |
icraggs | 20:cad3d54d7ecf | 299 | messageHandlers[i].fp(message); |
icraggs | 20:cad3d54d7ecf | 300 | rc = 0; |
icraggs | 20:cad3d54d7ecf | 301 | break; |
icraggs | 20:cad3d54d7ecf | 302 | } |
icraggs | 20:cad3d54d7ecf | 303 | } |
icraggs | 20:cad3d54d7ecf | 304 | |
icraggs | 20:cad3d54d7ecf | 305 | return rc; |
icraggs | 20:cad3d54d7ecf | 306 | } |
icraggs | 20:cad3d54d7ecf | 307 | |
icraggs | 20:cad3d54d7ecf | 308 | |
icraggs | 20:cad3d54d7ecf | 309 | |
icraggs | 21:e918525e529d | 310 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::cycle(int timeout) |
icraggs | 20:cad3d54d7ecf | 311 | { |
icraggs | 20:cad3d54d7ecf | 312 | /* get one piece of work off the wire and one pass through */ |
icraggs | 20:cad3d54d7ecf | 313 | |
icraggs | 20:cad3d54d7ecf | 314 | // read the socket, see what work is due |
icraggs | 20:cad3d54d7ecf | 315 | int packet_type = readPacket(timeout); |
icraggs | 20:cad3d54d7ecf | 316 | |
icraggs | 20:cad3d54d7ecf | 317 | int len, rc; |
icraggs | 20:cad3d54d7ecf | 318 | switch (packet_type) |
icraggs | 20:cad3d54d7ecf | 319 | { |
icraggs | 20:cad3d54d7ecf | 320 | case CONNACK: |
icraggs | 20:cad3d54d7ecf | 321 | if (this->thread) |
icraggs | 20:cad3d54d7ecf | 322 | { |
icraggs | 20:cad3d54d7ecf | 323 | Result res = {this, 0}; |
icraggs | 21:e918525e529d | 324 | if (MQTTDeserialize_connack(&res.rc, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1) |
icraggs | 20:cad3d54d7ecf | 325 | ; |
icraggs | 20:cad3d54d7ecf | 326 | connectHandler(&res); |
icraggs | 20:cad3d54d7ecf | 327 | connectHandler.detach(); // only invoke the callback once |
icraggs | 20:cad3d54d7ecf | 328 | } |
icraggs | 20:cad3d54d7ecf | 329 | break; |
icraggs | 20:cad3d54d7ecf | 330 | case PUBACK: |
icraggs | 20:cad3d54d7ecf | 331 | if (this->thread) |
icraggs | 20:cad3d54d7ecf | 332 | ; //call resultHandler |
icraggs | 20:cad3d54d7ecf | 333 | case SUBACK: |
icraggs | 20:cad3d54d7ecf | 334 | break; |
icraggs | 20:cad3d54d7ecf | 335 | case PUBLISH: |
icraggs | 20:cad3d54d7ecf | 336 | MQTTString topicName; |
icraggs | 20:cad3d54d7ecf | 337 | Message msg; |
icraggs | 20:cad3d54d7ecf | 338 | rc = MQTTDeserialize_publish((int*)&msg.dup, (int*)&msg.qos, (int*)&msg.retained, (int*)&msg.id, &topicName, |
icraggs | 20:cad3d54d7ecf | 339 | (char**)&msg.payload, (int*)&msg.payloadlen, readbuf, limits.MAX_MQTT_PACKET_SIZE);; |
icraggs | 20:cad3d54d7ecf | 340 | if (msg.qos == QOS0) |
icraggs | 20:cad3d54d7ecf | 341 | deliverMessage(&topicName, &msg); |
icraggs | 20:cad3d54d7ecf | 342 | break; |
icraggs | 20:cad3d54d7ecf | 343 | case PUBREC: |
icraggs | 20:cad3d54d7ecf | 344 | int type, dup, mypacketid; |
icraggs | 20:cad3d54d7ecf | 345 | if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1) |
icraggs | 20:cad3d54d7ecf | 346 | ; |
icraggs | 20:cad3d54d7ecf | 347 | // must lock this access against the application thread, if we are multi-threaded |
icraggs | 20:cad3d54d7ecf | 348 | len = MQTTSerialize_ack(buf, limits.MAX_MQTT_PACKET_SIZE, PUBREL, 0, mypacketid); |
icraggs | 20:cad3d54d7ecf | 349 | rc = sendPacket(len, timeout); // send the PUBREL packet |
icraggs | 20:cad3d54d7ecf | 350 | if (rc != len) |
icraggs | 20:cad3d54d7ecf | 351 | goto exit; // there was a problem |
icraggs | 20:cad3d54d7ecf | 352 | |
icraggs | 20:cad3d54d7ecf | 353 | break; |
icraggs | 20:cad3d54d7ecf | 354 | case PUBCOMP: |
icraggs | 20:cad3d54d7ecf | 355 | break; |
icraggs | 20:cad3d54d7ecf | 356 | case PINGRESP: |
icraggs | 20:cad3d54d7ecf | 357 | ping_outstanding = false; |
icraggs | 20:cad3d54d7ecf | 358 | break; |
icraggs | 20:cad3d54d7ecf | 359 | } |
icraggs | 20:cad3d54d7ecf | 360 | keepalive(); |
icraggs | 20:cad3d54d7ecf | 361 | exit: |
icraggs | 20:cad3d54d7ecf | 362 | return packet_type; |
icraggs | 20:cad3d54d7ecf | 363 | } |
icraggs | 20:cad3d54d7ecf | 364 | |
icraggs | 20:cad3d54d7ecf | 365 | |
icraggs | 21:e918525e529d | 366 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::keepalive() |
icraggs | 20:cad3d54d7ecf | 367 | { |
icraggs | 20:cad3d54d7ecf | 368 | int rc = 0; |
icraggs | 20:cad3d54d7ecf | 369 | |
icraggs | 20:cad3d54d7ecf | 370 | if (keepAliveInterval == 0) |
icraggs | 20:cad3d54d7ecf | 371 | goto exit; |
icraggs | 20:cad3d54d7ecf | 372 | |
icraggs | 20:cad3d54d7ecf | 373 | if (ping_timer.expired()) |
icraggs | 20:cad3d54d7ecf | 374 | { |
icraggs | 20:cad3d54d7ecf | 375 | if (ping_outstanding) |
icraggs | 20:cad3d54d7ecf | 376 | rc = -1; |
icraggs | 20:cad3d54d7ecf | 377 | else |
icraggs | 20:cad3d54d7ecf | 378 | { |
icraggs | 20:cad3d54d7ecf | 379 | int len = MQTTSerialize_pingreq(buf, limits.MAX_MQTT_PACKET_SIZE); |
icraggs | 20:cad3d54d7ecf | 380 | rc = sendPacket(len, 1000); // send the ping packet |
icraggs | 20:cad3d54d7ecf | 381 | if (rc != len) |
icraggs | 20:cad3d54d7ecf | 382 | rc = -1; // indicate there's a problem |
icraggs | 20:cad3d54d7ecf | 383 | else |
icraggs | 20:cad3d54d7ecf | 384 | ping_outstanding = true; |
icraggs | 20:cad3d54d7ecf | 385 | } |
icraggs | 20:cad3d54d7ecf | 386 | } |
icraggs | 20:cad3d54d7ecf | 387 | |
icraggs | 20:cad3d54d7ecf | 388 | exit: |
icraggs | 20:cad3d54d7ecf | 389 | return rc; |
icraggs | 20:cad3d54d7ecf | 390 | } |
icraggs | 20:cad3d54d7ecf | 391 | |
icraggs | 20:cad3d54d7ecf | 392 | |
icraggs | 21:e918525e529d | 393 | template<class Network, class Timer, class Thread, class Mutex> void MQTT::Async<Network, Timer, Thread, Mutex>::run(void const *argument) |
icraggs | 20:cad3d54d7ecf | 394 | { |
icraggs | 20:cad3d54d7ecf | 395 | while (true) |
icraggs | 20:cad3d54d7ecf | 396 | cycle(ping_timer.left_ms()); |
icraggs | 20:cad3d54d7ecf | 397 | } |
icraggs | 20:cad3d54d7ecf | 398 | |
icraggs | 20:cad3d54d7ecf | 399 | |
icraggs | 20:cad3d54d7ecf | 400 | // only used in single-threaded mode where one command at a time is in process |
icraggs | 21:e918525e529d | 401 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::waitfor(int packet_type, Timer& atimer) |
icraggs | 20:cad3d54d7ecf | 402 | { |
icraggs | 20:cad3d54d7ecf | 403 | int rc = -1; |
icraggs | 20:cad3d54d7ecf | 404 | |
icraggs | 20:cad3d54d7ecf | 405 | do |
icraggs | 20:cad3d54d7ecf | 406 | { |
icraggs | 20:cad3d54d7ecf | 407 | if (atimer.expired()) |
icraggs | 20:cad3d54d7ecf | 408 | break; // we timed out |
icraggs | 20:cad3d54d7ecf | 409 | } |
icraggs | 20:cad3d54d7ecf | 410 | while ((rc = cycle(atimer.left_ms())) != packet_type); |
icraggs | 20:cad3d54d7ecf | 411 | |
icraggs | 20:cad3d54d7ecf | 412 | return rc; |
icraggs | 20:cad3d54d7ecf | 413 | } |
icraggs | 20:cad3d54d7ecf | 414 | |
icraggs | 20:cad3d54d7ecf | 415 | |
icraggs | 21:e918525e529d | 416 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::connect(resultHandler resultHandler, MQTTPacket_connectData* options) |
icraggs | 20:cad3d54d7ecf | 417 | { |
icraggs | 21:e918525e529d | 418 | connect_timer.countdown(limits.command_timeout_ms); |
icraggs | 20:cad3d54d7ecf | 419 | |
icraggs | 20:cad3d54d7ecf | 420 | MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer; |
icraggs | 20:cad3d54d7ecf | 421 | if (options == 0) |
icraggs | 20:cad3d54d7ecf | 422 | options = &default_options; // set default options if none were supplied |
icraggs | 20:cad3d54d7ecf | 423 | |
icraggs | 20:cad3d54d7ecf | 424 | this->keepAliveInterval = options->keepAliveInterval; |
icraggs | 20:cad3d54d7ecf | 425 | ping_timer.countdown(this->keepAliveInterval); |
icraggs | 20:cad3d54d7ecf | 426 | int len = MQTTSerialize_connect(buf, limits.MAX_MQTT_PACKET_SIZE, options); |
icraggs | 20:cad3d54d7ecf | 427 | int rc = sendPacket(len, connect_timer.left_ms()); // send the connect packet |
icraggs | 20:cad3d54d7ecf | 428 | if (rc != len) |
icraggs | 20:cad3d54d7ecf | 429 | goto exit; // there was a problem |
icraggs | 20:cad3d54d7ecf | 430 | |
icraggs | 20:cad3d54d7ecf | 431 | if (resultHandler == 0) // wait until the connack is received |
icraggs | 20:cad3d54d7ecf | 432 | { |
icraggs | 20:cad3d54d7ecf | 433 | // this will be a blocking call, wait for the connack |
icraggs | 20:cad3d54d7ecf | 434 | if (waitfor(CONNACK, connect_timer) == CONNACK) |
icraggs | 20:cad3d54d7ecf | 435 | { |
icraggs | 20:cad3d54d7ecf | 436 | int connack_rc = -1; |
icraggs | 20:cad3d54d7ecf | 437 | if (MQTTDeserialize_connack(&connack_rc, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1) |
icraggs | 20:cad3d54d7ecf | 438 | rc = connack_rc; |
icraggs | 20:cad3d54d7ecf | 439 | } |
icraggs | 20:cad3d54d7ecf | 440 | } |
icraggs | 20:cad3d54d7ecf | 441 | else |
icraggs | 20:cad3d54d7ecf | 442 | { |
icraggs | 20:cad3d54d7ecf | 443 | // set connect response callback function |
icraggs | 20:cad3d54d7ecf | 444 | connectHandler.attach(resultHandler); |
icraggs | 20:cad3d54d7ecf | 445 | |
icraggs | 20:cad3d54d7ecf | 446 | // start background thread |
icraggs | 21:e918525e529d | 447 | this->thread = new Thread((void (*)(void const *argument))&MQTT::Async<Network, Timer, Thread, Mutex>::threadfn, (void*)this); |
icraggs | 20:cad3d54d7ecf | 448 | } |
icraggs | 20:cad3d54d7ecf | 449 | |
icraggs | 20:cad3d54d7ecf | 450 | exit: |
icraggs | 20:cad3d54d7ecf | 451 | return rc; |
icraggs | 20:cad3d54d7ecf | 452 | } |
icraggs | 20:cad3d54d7ecf | 453 | |
icraggs | 20:cad3d54d7ecf | 454 | |
icraggs | 21:e918525e529d | 455 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::findFreeOperation() |
icraggs | 20:cad3d54d7ecf | 456 | { |
icraggs | 20:cad3d54d7ecf | 457 | int found = -1; |
icraggs | 20:cad3d54d7ecf | 458 | for (int i = 0; i < limits.MAX_CONCURRENT_OPERATIONS; ++i) |
icraggs | 20:cad3d54d7ecf | 459 | { |
icraggs | 20:cad3d54d7ecf | 460 | if (operations[i].id == 0) |
icraggs | 20:cad3d54d7ecf | 461 | { |
icraggs | 20:cad3d54d7ecf | 462 | found = i; |
icraggs | 20:cad3d54d7ecf | 463 | break; |
icraggs | 20:cad3d54d7ecf | 464 | } |
icraggs | 20:cad3d54d7ecf | 465 | } |
icraggs | 20:cad3d54d7ecf | 466 | return found; |
icraggs | 20:cad3d54d7ecf | 467 | } |
icraggs | 20:cad3d54d7ecf | 468 | |
icraggs | 20:cad3d54d7ecf | 469 | |
icraggs | 21:e918525e529d | 470 | 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) |
icraggs | 20:cad3d54d7ecf | 471 | { |
icraggs | 20:cad3d54d7ecf | 472 | int index = 0; |
icraggs | 20:cad3d54d7ecf | 473 | if (this->thread) |
icraggs | 20:cad3d54d7ecf | 474 | index = findFreeOperation(); |
icraggs | 20:cad3d54d7ecf | 475 | Timer& atimer = operations[index].timer; |
icraggs | 20:cad3d54d7ecf | 476 | |
icraggs | 21:e918525e529d | 477 | atimer.countdown(limits.command_timeout_ms); |
icraggs | 20:cad3d54d7ecf | 478 | MQTTString topic = {(char*)topicFilter, 0, 0}; |
icraggs | 20:cad3d54d7ecf | 479 | |
icraggs | 20:cad3d54d7ecf | 480 | int len = MQTTSerialize_subscribe(buf, limits.MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic, (int*)&qos); |
icraggs | 20:cad3d54d7ecf | 481 | int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet |
icraggs | 20:cad3d54d7ecf | 482 | if (rc != len) |
icraggs | 20:cad3d54d7ecf | 483 | goto exit; // there was a problem |
icraggs | 20:cad3d54d7ecf | 484 | |
icraggs | 20:cad3d54d7ecf | 485 | /* wait for suback */ |
icraggs | 20:cad3d54d7ecf | 486 | if (resultHandler == 0) |
icraggs | 20:cad3d54d7ecf | 487 | { |
icraggs | 20:cad3d54d7ecf | 488 | // this will block |
icraggs | 20:cad3d54d7ecf | 489 | if (waitfor(SUBACK, atimer) == SUBACK) |
icraggs | 20:cad3d54d7ecf | 490 | { |
icraggs | 20:cad3d54d7ecf | 491 | int count = 0, grantedQoS = -1, mypacketid; |
icraggs | 20:cad3d54d7ecf | 492 | if (MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1) |
icraggs | 20:cad3d54d7ecf | 493 | rc = grantedQoS; // 0, 1, 2 or 0x80 |
icraggs | 20:cad3d54d7ecf | 494 | if (rc != 0x80) |
icraggs | 20:cad3d54d7ecf | 495 | { |
icraggs | 20:cad3d54d7ecf | 496 | for (int i = 0; i < limits.MAX_MESSAGE_HANDLERS; ++i) |
icraggs | 20:cad3d54d7ecf | 497 | { |
icraggs | 20:cad3d54d7ecf | 498 | if (messageHandlers[i].topic == 0) |
icraggs | 20:cad3d54d7ecf | 499 | { |
icraggs | 20:cad3d54d7ecf | 500 | messageHandlers[i].topic = topicFilter; |
icraggs | 20:cad3d54d7ecf | 501 | messageHandlers[i].fp.attach(messageHandler); |
icraggs | 20:cad3d54d7ecf | 502 | rc = 0; |
icraggs | 20:cad3d54d7ecf | 503 | break; |
icraggs | 20:cad3d54d7ecf | 504 | } |
icraggs | 20:cad3d54d7ecf | 505 | } |
icraggs | 20:cad3d54d7ecf | 506 | } |
icraggs | 20:cad3d54d7ecf | 507 | } |
icraggs | 20:cad3d54d7ecf | 508 | } |
icraggs | 20:cad3d54d7ecf | 509 | else |
icraggs | 20:cad3d54d7ecf | 510 | { |
icraggs | 20:cad3d54d7ecf | 511 | // set subscribe response callback function |
icraggs | 20:cad3d54d7ecf | 512 | |
icraggs | 20:cad3d54d7ecf | 513 | } |
icraggs | 20:cad3d54d7ecf | 514 | |
icraggs | 20:cad3d54d7ecf | 515 | exit: |
icraggs | 20:cad3d54d7ecf | 516 | return rc; |
icraggs | 20:cad3d54d7ecf | 517 | } |
icraggs | 20:cad3d54d7ecf | 518 | |
icraggs | 20:cad3d54d7ecf | 519 | |
icraggs | 21:e918525e529d | 520 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::unsubscribe(resultHandler resultHandler, const char* topicFilter) |
icraggs | 20:cad3d54d7ecf | 521 | { |
icraggs | 20:cad3d54d7ecf | 522 | int index = 0; |
icraggs | 20:cad3d54d7ecf | 523 | if (this->thread) |
icraggs | 20:cad3d54d7ecf | 524 | index = findFreeOperation(); |
icraggs | 20:cad3d54d7ecf | 525 | Timer& atimer = operations[index].timer; |
icraggs | 20:cad3d54d7ecf | 526 | |
icraggs | 21:e918525e529d | 527 | atimer.countdown(limits.command_timeout_ms); |
icraggs | 20:cad3d54d7ecf | 528 | MQTTString topic = {(char*)topicFilter, 0, 0}; |
icraggs | 20:cad3d54d7ecf | 529 | |
icraggs | 20:cad3d54d7ecf | 530 | int len = MQTTSerialize_unsubscribe(buf, limits.MAX_MQTT_PACKET_SIZE, 0, packetid.getNext(), 1, &topic); |
icraggs | 20:cad3d54d7ecf | 531 | int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet |
icraggs | 20:cad3d54d7ecf | 532 | if (rc != len) |
icraggs | 20:cad3d54d7ecf | 533 | goto exit; // there was a problem |
icraggs | 20:cad3d54d7ecf | 534 | |
icraggs | 21:e918525e529d | 535 | // set unsubscribe response callback function |
icraggs | 20:cad3d54d7ecf | 536 | |
icraggs | 20:cad3d54d7ecf | 537 | |
icraggs | 20:cad3d54d7ecf | 538 | exit: |
icraggs | 20:cad3d54d7ecf | 539 | return rc; |
icraggs | 20:cad3d54d7ecf | 540 | } |
icraggs | 20:cad3d54d7ecf | 541 | |
icraggs | 20:cad3d54d7ecf | 542 | |
icraggs | 20:cad3d54d7ecf | 543 | |
icraggs | 21:e918525e529d | 544 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::publish(resultHandler resultHandler, const char* topicName, Message* message) |
icraggs | 20:cad3d54d7ecf | 545 | { |
icraggs | 20:cad3d54d7ecf | 546 | int index = 0; |
icraggs | 20:cad3d54d7ecf | 547 | if (this->thread) |
icraggs | 20:cad3d54d7ecf | 548 | index = findFreeOperation(); |
icraggs | 20:cad3d54d7ecf | 549 | Timer& atimer = operations[index].timer; |
icraggs | 20:cad3d54d7ecf | 550 | |
icraggs | 21:e918525e529d | 551 | atimer.countdown(limits.command_timeout_ms); |
icraggs | 20:cad3d54d7ecf | 552 | MQTTString topic = {(char*)topicName, 0, 0}; |
icraggs | 20:cad3d54d7ecf | 553 | |
icraggs | 20:cad3d54d7ecf | 554 | if (message->qos == QOS1 || message->qos == QOS2) |
icraggs | 20:cad3d54d7ecf | 555 | message->id = packetid.getNext(); |
icraggs | 20:cad3d54d7ecf | 556 | |
icraggs | 20:cad3d54d7ecf | 557 | int len = MQTTSerialize_publish(buf, limits.MAX_MQTT_PACKET_SIZE, 0, message->qos, message->retained, message->id, topic, (char*)message->payload, message->payloadlen); |
icraggs | 20:cad3d54d7ecf | 558 | int rc = sendPacket(len, atimer.left_ms()); // send the subscribe packet |
icraggs | 20:cad3d54d7ecf | 559 | if (rc != len) |
icraggs | 20:cad3d54d7ecf | 560 | goto exit; // there was a problem |
icraggs | 20:cad3d54d7ecf | 561 | |
icraggs | 20:cad3d54d7ecf | 562 | /* wait for acks */ |
icraggs | 20:cad3d54d7ecf | 563 | if (resultHandler == 0) |
icraggs | 20:cad3d54d7ecf | 564 | { |
icraggs | 20:cad3d54d7ecf | 565 | if (message->qos == QOS1) |
icraggs | 20:cad3d54d7ecf | 566 | { |
icraggs | 20:cad3d54d7ecf | 567 | if (waitfor(PUBACK, atimer) == PUBACK) |
icraggs | 20:cad3d54d7ecf | 568 | { |
icraggs | 20:cad3d54d7ecf | 569 | int type, dup, mypacketid; |
icraggs | 20:cad3d54d7ecf | 570 | if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1) |
icraggs | 20:cad3d54d7ecf | 571 | rc = 0; |
icraggs | 20:cad3d54d7ecf | 572 | } |
icraggs | 20:cad3d54d7ecf | 573 | } |
icraggs | 20:cad3d54d7ecf | 574 | else if (message->qos == QOS2) |
icraggs | 20:cad3d54d7ecf | 575 | { |
icraggs | 20:cad3d54d7ecf | 576 | if (waitfor(PUBCOMP, atimer) == PUBCOMP) |
icraggs | 20:cad3d54d7ecf | 577 | { |
icraggs | 20:cad3d54d7ecf | 578 | int type, dup, mypacketid; |
icraggs | 20:cad3d54d7ecf | 579 | if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, limits.MAX_MQTT_PACKET_SIZE) == 1) |
icraggs | 20:cad3d54d7ecf | 580 | rc = 0; |
icraggs | 20:cad3d54d7ecf | 581 | } |
icraggs | 20:cad3d54d7ecf | 582 | |
icraggs | 20:cad3d54d7ecf | 583 | } |
icraggs | 20:cad3d54d7ecf | 584 | } |
icraggs | 20:cad3d54d7ecf | 585 | else |
icraggs | 20:cad3d54d7ecf | 586 | { |
icraggs | 20:cad3d54d7ecf | 587 | // set publish response callback function |
icraggs | 20:cad3d54d7ecf | 588 | |
icraggs | 20:cad3d54d7ecf | 589 | } |
icraggs | 20:cad3d54d7ecf | 590 | |
icraggs | 20:cad3d54d7ecf | 591 | exit: |
icraggs | 20:cad3d54d7ecf | 592 | return rc; |
icraggs | 20:cad3d54d7ecf | 593 | } |
icraggs | 20:cad3d54d7ecf | 594 | |
icraggs | 20:cad3d54d7ecf | 595 | |
icraggs | 21:e918525e529d | 596 | template<class Network, class Timer, class Thread, class Mutex> int MQTT::Async<Network, Timer, Thread, Mutex>::disconnect(resultHandler resultHandler) |
icraggs | 21:e918525e529d | 597 | { |
icraggs | 21:e918525e529d | 598 | Timer timer = Timer(limits.command_timeout_ms); // we might wait for incomplete incoming publishes to complete |
icraggs | 21:e918525e529d | 599 | int len = MQTTSerialize_disconnect(buf, limits.MAX_MQTT_PACKET_SIZE); |
icraggs | 21:e918525e529d | 600 | int rc = sendPacket(len, timer.left_ms()); // send the disconnect packet |
icraggs | 21:e918525e529d | 601 | |
icraggs | 21:e918525e529d | 602 | return (rc == len) ? 0 : -1; |
icraggs | 21:e918525e529d | 603 | } |
icraggs | 21:e918525e529d | 604 | |
icraggs | 21:e918525e529d | 605 | |
icraggs | 21:e918525e529d | 606 | |
icraggs | 23:05fc7de97d4a | 607 | #endif |