[209] | 1 | /*
|
---|
| 2 | PubSubClient.cpp - A simple client for MQTT.
|
---|
| 3 | Nick O'Leary
|
---|
| 4 | http://knolleary.net
|
---|
| 5 | */
|
---|
| 6 |
|
---|
| 7 | #include "PubSubClient.h"
|
---|
| 8 | #include "Arduino.h"
|
---|
| 9 |
|
---|
| 10 | PubSubClient::PubSubClient() {
|
---|
| 11 | this->_state = MQTT_DISCONNECTED;
|
---|
| 12 | this->_client = NULL;
|
---|
| 13 | this->stream = NULL;
|
---|
| 14 | setCallback(NULL);
|
---|
| 15 | }
|
---|
| 16 |
|
---|
| 17 | PubSubClient::PubSubClient(Client& client) {
|
---|
| 18 | this->_state = MQTT_DISCONNECTED;
|
---|
| 19 | setClient(client);
|
---|
| 20 | this->stream = NULL;
|
---|
| 21 | }
|
---|
| 22 |
|
---|
| 23 | PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) {
|
---|
| 24 | this->_state = MQTT_DISCONNECTED;
|
---|
| 25 | setServer(addr, port);
|
---|
| 26 | setClient(client);
|
---|
| 27 | this->stream = NULL;
|
---|
| 28 | }
|
---|
| 29 | PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) {
|
---|
| 30 | this->_state = MQTT_DISCONNECTED;
|
---|
| 31 | setServer(addr,port);
|
---|
| 32 | setClient(client);
|
---|
| 33 | setStream(stream);
|
---|
| 34 | }
|
---|
| 35 | PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
|
---|
| 36 | this->_state = MQTT_DISCONNECTED;
|
---|
| 37 | setServer(addr, port);
|
---|
| 38 | setCallback(callback);
|
---|
| 39 | setClient(client);
|
---|
| 40 | this->stream = NULL;
|
---|
| 41 | }
|
---|
| 42 | PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
|
---|
| 43 | this->_state = MQTT_DISCONNECTED;
|
---|
| 44 | setServer(addr,port);
|
---|
| 45 | setCallback(callback);
|
---|
| 46 | setClient(client);
|
---|
| 47 | setStream(stream);
|
---|
| 48 | }
|
---|
| 49 |
|
---|
| 50 | PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) {
|
---|
| 51 | this->_state = MQTT_DISCONNECTED;
|
---|
| 52 | setServer(ip, port);
|
---|
| 53 | setClient(client);
|
---|
| 54 | this->stream = NULL;
|
---|
| 55 | }
|
---|
| 56 | PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) {
|
---|
| 57 | this->_state = MQTT_DISCONNECTED;
|
---|
| 58 | setServer(ip,port);
|
---|
| 59 | setClient(client);
|
---|
| 60 | setStream(stream);
|
---|
| 61 | }
|
---|
| 62 | PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
|
---|
| 63 | this->_state = MQTT_DISCONNECTED;
|
---|
| 64 | setServer(ip, port);
|
---|
| 65 | setCallback(callback);
|
---|
| 66 | setClient(client);
|
---|
| 67 | this->stream = NULL;
|
---|
| 68 | }
|
---|
| 69 | PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
|
---|
| 70 | this->_state = MQTT_DISCONNECTED;
|
---|
| 71 | setServer(ip,port);
|
---|
| 72 | setCallback(callback);
|
---|
| 73 | setClient(client);
|
---|
| 74 | setStream(stream);
|
---|
| 75 | }
|
---|
| 76 |
|
---|
| 77 | PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) {
|
---|
| 78 | this->_state = MQTT_DISCONNECTED;
|
---|
| 79 | setServer(domain,port);
|
---|
| 80 | setClient(client);
|
---|
| 81 | this->stream = NULL;
|
---|
| 82 | }
|
---|
| 83 | PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) {
|
---|
| 84 | this->_state = MQTT_DISCONNECTED;
|
---|
| 85 | setServer(domain,port);
|
---|
| 86 | setClient(client);
|
---|
| 87 | setStream(stream);
|
---|
| 88 | }
|
---|
| 89 | PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
|
---|
| 90 | this->_state = MQTT_DISCONNECTED;
|
---|
| 91 | setServer(domain,port);
|
---|
| 92 | setCallback(callback);
|
---|
| 93 | setClient(client);
|
---|
| 94 | this->stream = NULL;
|
---|
| 95 | }
|
---|
| 96 | PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
|
---|
| 97 | this->_state = MQTT_DISCONNECTED;
|
---|
| 98 | setServer(domain,port);
|
---|
| 99 | setCallback(callback);
|
---|
| 100 | setClient(client);
|
---|
| 101 | setStream(stream);
|
---|
| 102 | }
|
---|
| 103 |
|
---|
| 104 | boolean PubSubClient::connect(const char *id) {
|
---|
| 105 | return connect(id,NULL,NULL,0,0,0,0);
|
---|
| 106 | }
|
---|
| 107 |
|
---|
| 108 | boolean PubSubClient::connect(const char *id, const char *user, const char *pass) {
|
---|
| 109 | return connect(id,user,pass,0,0,0,0);
|
---|
| 110 | }
|
---|
| 111 |
|
---|
| 112 | boolean PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) {
|
---|
| 113 | return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage);
|
---|
| 114 | }
|
---|
| 115 |
|
---|
| 116 | boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) {
|
---|
| 117 | if (!connected()) {
|
---|
| 118 | int result = 0;
|
---|
| 119 |
|
---|
| 120 | if (domain != NULL) {
|
---|
| 121 | result = _client->connect(this->domain, this->port);
|
---|
| 122 | } else {
|
---|
| 123 | result = _client->connect(this->ip, this->port);
|
---|
| 124 | }
|
---|
| 125 | if (result == 1) {
|
---|
| 126 | nextMsgId = 1;
|
---|
| 127 | // Leave room in the buffer for header and variable length field
|
---|
| 128 | uint16_t length = 5;
|
---|
| 129 | unsigned int j;
|
---|
| 130 |
|
---|
| 131 | #if MQTT_VERSION == MQTT_VERSION_3_1
|
---|
| 132 | uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p', MQTT_VERSION};
|
---|
| 133 | #define MQTT_HEADER_VERSION_LENGTH 9
|
---|
| 134 | #elif MQTT_VERSION == MQTT_VERSION_3_1_1
|
---|
| 135 | uint8_t d[7] = {0x00,0x04,'M','Q','T','T',MQTT_VERSION};
|
---|
| 136 | #define MQTT_HEADER_VERSION_LENGTH 7
|
---|
| 137 | #endif
|
---|
| 138 | for (j = 0;j<MQTT_HEADER_VERSION_LENGTH;j++) {
|
---|
| 139 | buffer[length++] = d[j];
|
---|
| 140 | }
|
---|
| 141 |
|
---|
| 142 | uint8_t v;
|
---|
| 143 | if (willTopic) {
|
---|
| 144 | v = 0x06|(willQos<<3)|(willRetain<<5);
|
---|
| 145 | } else {
|
---|
| 146 | v = 0x02;
|
---|
| 147 | }
|
---|
| 148 |
|
---|
| 149 | if(user != NULL) {
|
---|
| 150 | v = v|0x80;
|
---|
| 151 |
|
---|
| 152 | if(pass != NULL) {
|
---|
| 153 | v = v|(0x80>>1);
|
---|
| 154 | }
|
---|
| 155 | }
|
---|
| 156 |
|
---|
| 157 | buffer[length++] = v;
|
---|
| 158 |
|
---|
| 159 | buffer[length++] = ((MQTT_KEEPALIVE) >> 8);
|
---|
| 160 | buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF);
|
---|
| 161 | length = writeString(id,buffer,length);
|
---|
| 162 | if (willTopic) {
|
---|
| 163 | length = writeString(willTopic,buffer,length);
|
---|
| 164 | length = writeString(willMessage,buffer,length);
|
---|
| 165 | }
|
---|
| 166 |
|
---|
| 167 | if(user != NULL) {
|
---|
| 168 | length = writeString(user,buffer,length);
|
---|
| 169 | if(pass != NULL) {
|
---|
| 170 | length = writeString(pass,buffer,length);
|
---|
| 171 | }
|
---|
| 172 | }
|
---|
| 173 |
|
---|
| 174 | write(MQTTCONNECT,buffer,length-5);
|
---|
| 175 |
|
---|
| 176 | lastInActivity = lastOutActivity = millis();
|
---|
| 177 |
|
---|
| 178 | while (!_client->available()) {
|
---|
| 179 | unsigned long t = millis();
|
---|
| 180 | if (t-lastInActivity >= ((int32_t) MQTT_SOCKET_TIMEOUT*1000UL)) {
|
---|
| 181 | _state = MQTT_CONNECTION_TIMEOUT;
|
---|
| 182 | _client->stop();
|
---|
| 183 | return false;
|
---|
| 184 | }
|
---|
| 185 | }
|
---|
| 186 | uint8_t llen;
|
---|
| 187 | uint16_t len = readPacket(&llen);
|
---|
| 188 |
|
---|
| 189 | if (len == 4) {
|
---|
| 190 | if (buffer[3] == 0) {
|
---|
| 191 | lastInActivity = millis();
|
---|
| 192 | pingOutstanding = false;
|
---|
| 193 | _state = MQTT_CONNECTED;
|
---|
| 194 | return true;
|
---|
| 195 | } else {
|
---|
| 196 | _state = buffer[3];
|
---|
| 197 | }
|
---|
| 198 | }
|
---|
| 199 | _client->stop();
|
---|
| 200 | } else {
|
---|
| 201 | _state = MQTT_CONNECT_FAILED;
|
---|
| 202 | }
|
---|
| 203 | return false;
|
---|
| 204 | }
|
---|
| 205 | return true;
|
---|
| 206 | }
|
---|
| 207 |
|
---|
| 208 | // reads a byte into result
|
---|
| 209 | boolean PubSubClient::readByte(uint8_t * result) {
|
---|
| 210 | uint32_t previousMillis = millis();
|
---|
| 211 | while(!_client->available()) {
|
---|
| 212 | uint32_t currentMillis = millis();
|
---|
| 213 | if(currentMillis - previousMillis >= ((int32_t) MQTT_SOCKET_TIMEOUT * 1000)){
|
---|
| 214 | return false;
|
---|
| 215 | }
|
---|
| 216 | }
|
---|
| 217 | *result = _client->read();
|
---|
| 218 | return true;
|
---|
| 219 | }
|
---|
| 220 |
|
---|
| 221 | // reads a byte into result[*index] and increments index
|
---|
| 222 | boolean PubSubClient::readByte(uint8_t * result, uint16_t * index){
|
---|
| 223 | uint16_t current_index = *index;
|
---|
| 224 | uint8_t * write_address = &(result[current_index]);
|
---|
| 225 | if(readByte(write_address)){
|
---|
| 226 | *index = current_index + 1;
|
---|
| 227 | return true;
|
---|
| 228 | }
|
---|
| 229 | return false;
|
---|
| 230 | }
|
---|
| 231 |
|
---|
| 232 | uint16_t PubSubClient::readPacket(uint8_t* lengthLength) {
|
---|
| 233 | uint16_t len = 0;
|
---|
| 234 | if(!readByte(buffer, &len)) return 0;
|
---|
| 235 | bool isPublish = (buffer[0]&0xF0) == MQTTPUBLISH;
|
---|
| 236 | uint32_t multiplier = 1;
|
---|
| 237 | uint16_t length = 0;
|
---|
| 238 | uint8_t digit = 0;
|
---|
| 239 | uint16_t skip = 0;
|
---|
| 240 | uint8_t start = 0;
|
---|
| 241 |
|
---|
| 242 | do {
|
---|
| 243 | if(!readByte(&digit)) return 0;
|
---|
| 244 | buffer[len++] = digit;
|
---|
| 245 | length += (digit & 127) * multiplier;
|
---|
| 246 | multiplier *= 128;
|
---|
| 247 | } while ((digit & 128) != 0);
|
---|
| 248 | *lengthLength = len-1;
|
---|
| 249 |
|
---|
| 250 | if (isPublish) {
|
---|
| 251 | // Read in topic length to calculate bytes to skip over for Stream writing
|
---|
| 252 | if(!readByte(buffer, &len)) return 0;
|
---|
| 253 | if(!readByte(buffer, &len)) return 0;
|
---|
| 254 | skip = (buffer[*lengthLength+1]<<8)+buffer[*lengthLength+2];
|
---|
| 255 | start = 2;
|
---|
| 256 | if (buffer[0]&MQTTQOS1) {
|
---|
| 257 | // skip message id
|
---|
| 258 | skip += 2;
|
---|
| 259 | }
|
---|
| 260 | }
|
---|
| 261 |
|
---|
| 262 | for (uint16_t i = start;i<length;i++) {
|
---|
| 263 | if(!readByte(&digit)) return 0;
|
---|
| 264 | if (this->stream) {
|
---|
| 265 | if (isPublish && len-*lengthLength-2>skip) {
|
---|
| 266 | this->stream->write(digit);
|
---|
| 267 | }
|
---|
| 268 | }
|
---|
| 269 | if (len < MQTT_MAX_PACKET_SIZE) {
|
---|
| 270 | buffer[len] = digit;
|
---|
| 271 | }
|
---|
| 272 | len++;
|
---|
| 273 | }
|
---|
| 274 |
|
---|
| 275 | if (!this->stream && len > MQTT_MAX_PACKET_SIZE) {
|
---|
| 276 | len = 0; // This will cause the packet to be ignored.
|
---|
| 277 | }
|
---|
| 278 |
|
---|
| 279 | return len;
|
---|
| 280 | }
|
---|
| 281 |
|
---|
| 282 | boolean PubSubClient::loop() {
|
---|
| 283 | if (connected()) {
|
---|
| 284 | unsigned long t = millis();
|
---|
| 285 | if ((t - lastInActivity > MQTT_KEEPALIVE*1000UL) || (t - lastOutActivity > MQTT_KEEPALIVE*1000UL)) {
|
---|
| 286 | if (pingOutstanding) {
|
---|
| 287 | this->_state = MQTT_CONNECTION_TIMEOUT;
|
---|
| 288 | _client->stop();
|
---|
| 289 | return false;
|
---|
| 290 | } else {
|
---|
| 291 | buffer[0] = MQTTPINGREQ;
|
---|
| 292 | buffer[1] = 0;
|
---|
| 293 | _client->write(buffer,2);
|
---|
| 294 | lastOutActivity = t;
|
---|
| 295 | lastInActivity = t;
|
---|
| 296 | pingOutstanding = true;
|
---|
| 297 | }
|
---|
| 298 | }
|
---|
| 299 | if (_client->available()) {
|
---|
| 300 | uint8_t llen;
|
---|
| 301 | uint16_t len = readPacket(&llen);
|
---|
| 302 | uint16_t msgId = 0;
|
---|
| 303 | uint8_t *payload;
|
---|
| 304 | if (len > 0) {
|
---|
| 305 | lastInActivity = t;
|
---|
| 306 | uint8_t type = buffer[0]&0xF0;
|
---|
| 307 | if (type == MQTTPUBLISH) {
|
---|
| 308 | if (callback) {
|
---|
| 309 | uint16_t tl = (buffer[llen+1]<<8)+buffer[llen+2];
|
---|
| 310 | char topic[tl+1];
|
---|
| 311 | for (uint16_t i=0;i<tl;i++) {
|
---|
| 312 | topic[i] = buffer[llen+3+i];
|
---|
| 313 | }
|
---|
| 314 | topic[tl] = 0;
|
---|
| 315 | // msgId only present for QOS>0
|
---|
| 316 | if ((buffer[0]&0x06) == MQTTQOS1) {
|
---|
| 317 | msgId = (buffer[llen+3+tl]<<8)+buffer[llen+3+tl+1];
|
---|
| 318 | payload = buffer+llen+3+tl+2;
|
---|
| 319 | callback(topic,payload,len-llen-3-tl-2);
|
---|
| 320 |
|
---|
| 321 | buffer[0] = MQTTPUBACK;
|
---|
| 322 | buffer[1] = 2;
|
---|
| 323 | buffer[2] = (msgId >> 8);
|
---|
| 324 | buffer[3] = (msgId & 0xFF);
|
---|
| 325 | _client->write(buffer,4);
|
---|
| 326 | lastOutActivity = t;
|
---|
| 327 |
|
---|
| 328 | } else {
|
---|
| 329 | payload = buffer+llen+3+tl;
|
---|
| 330 | callback(topic,payload,len-llen-3-tl);
|
---|
| 331 | }
|
---|
| 332 | }
|
---|
| 333 | } else if (type == MQTTPINGREQ) {
|
---|
| 334 | buffer[0] = MQTTPINGRESP;
|
---|
| 335 | buffer[1] = 0;
|
---|
| 336 | _client->write(buffer,2);
|
---|
| 337 | } else if (type == MQTTPINGRESP) {
|
---|
| 338 | pingOutstanding = false;
|
---|
| 339 | }
|
---|
| 340 | }
|
---|
| 341 | }
|
---|
| 342 | return true;
|
---|
| 343 | }
|
---|
| 344 | return false;
|
---|
| 345 | }
|
---|
| 346 |
|
---|
| 347 | boolean PubSubClient::publish(const char* topic, const char* payload) {
|
---|
| 348 | return publish(topic,(const uint8_t*)payload,strlen(payload),false);
|
---|
| 349 | }
|
---|
| 350 |
|
---|
| 351 | boolean PubSubClient::publish(const char* topic, const char* payload, boolean retained) {
|
---|
| 352 | return publish(topic,(const uint8_t*)payload,strlen(payload),retained);
|
---|
| 353 | }
|
---|
| 354 |
|
---|
| 355 | boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength) {
|
---|
| 356 | return publish(topic, payload, plength, false);
|
---|
| 357 | }
|
---|
| 358 |
|
---|
| 359 | boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
|
---|
| 360 | if (connected()) {
|
---|
| 361 | if (MQTT_MAX_PACKET_SIZE < 5 + 2+strlen(topic) + plength) {
|
---|
| 362 | // Too long
|
---|
| 363 | return false;
|
---|
| 364 | }
|
---|
| 365 | // Leave room in the buffer for header and variable length field
|
---|
| 366 | uint16_t length = 5;
|
---|
| 367 | length = writeString(topic,buffer,length);
|
---|
| 368 | uint16_t i;
|
---|
| 369 | for (i=0;i<plength;i++) {
|
---|
| 370 | buffer[length++] = payload[i];
|
---|
| 371 | }
|
---|
| 372 | uint8_t header = MQTTPUBLISH;
|
---|
| 373 | if (retained) {
|
---|
| 374 | header |= 1;
|
---|
| 375 | }
|
---|
| 376 | return write(header,buffer,length-5);
|
---|
| 377 | }
|
---|
| 378 | return false;
|
---|
| 379 | }
|
---|
| 380 |
|
---|
| 381 | boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
|
---|
| 382 | uint8_t llen = 0;
|
---|
| 383 | uint8_t digit;
|
---|
| 384 | unsigned int rc = 0;
|
---|
| 385 | uint16_t tlen;
|
---|
| 386 | unsigned int pos = 0;
|
---|
| 387 | unsigned int i;
|
---|
| 388 | uint8_t header;
|
---|
| 389 | unsigned int len;
|
---|
| 390 |
|
---|
| 391 | if (!connected()) {
|
---|
| 392 | return false;
|
---|
| 393 | }
|
---|
| 394 |
|
---|
| 395 | tlen = strlen(topic);
|
---|
| 396 |
|
---|
| 397 | header = MQTTPUBLISH;
|
---|
| 398 | if (retained) {
|
---|
| 399 | header |= 1;
|
---|
| 400 | }
|
---|
| 401 | buffer[pos++] = header;
|
---|
| 402 | len = plength + 2 + tlen;
|
---|
| 403 | do {
|
---|
| 404 | digit = len % 128;
|
---|
| 405 | len = len / 128;
|
---|
| 406 | if (len > 0) {
|
---|
| 407 | digit |= 0x80;
|
---|
| 408 | }
|
---|
| 409 | buffer[pos++] = digit;
|
---|
| 410 | llen++;
|
---|
| 411 | } while(len>0);
|
---|
| 412 |
|
---|
| 413 | pos = writeString(topic,buffer,pos);
|
---|
| 414 |
|
---|
| 415 | rc += _client->write(buffer,pos);
|
---|
| 416 |
|
---|
| 417 | for (i=0;i<plength;i++) {
|
---|
| 418 | rc += _client->write((char)pgm_read_byte_near(payload + i));
|
---|
| 419 | }
|
---|
| 420 |
|
---|
| 421 | lastOutActivity = millis();
|
---|
| 422 |
|
---|
| 423 | return rc == tlen + 4 + plength;
|
---|
| 424 | }
|
---|
| 425 |
|
---|
| 426 | boolean PubSubClient::write(uint8_t header, uint8_t* buf, uint16_t length) {
|
---|
| 427 | uint8_t lenBuf[4];
|
---|
| 428 | uint8_t llen = 0;
|
---|
| 429 | uint8_t digit;
|
---|
| 430 | uint8_t pos = 0;
|
---|
| 431 | uint16_t rc;
|
---|
| 432 | uint16_t len = length;
|
---|
| 433 | do {
|
---|
| 434 | digit = len % 128;
|
---|
| 435 | len = len / 128;
|
---|
| 436 | if (len > 0) {
|
---|
| 437 | digit |= 0x80;
|
---|
| 438 | }
|
---|
| 439 | lenBuf[pos++] = digit;
|
---|
| 440 | llen++;
|
---|
| 441 | } while(len>0);
|
---|
| 442 |
|
---|
| 443 | buf[4-llen] = header;
|
---|
| 444 | for (int i=0;i<llen;i++) {
|
---|
| 445 | buf[5-llen+i] = lenBuf[i];
|
---|
| 446 | }
|
---|
| 447 |
|
---|
| 448 | #ifdef MQTT_MAX_TRANSFER_SIZE
|
---|
| 449 | uint8_t* writeBuf = buf+(4-llen);
|
---|
| 450 | uint16_t bytesRemaining = length+1+llen; //Match the length type
|
---|
| 451 | uint8_t bytesToWrite;
|
---|
| 452 | boolean result = true;
|
---|
| 453 | while((bytesRemaining > 0) && result) {
|
---|
| 454 | bytesToWrite = (bytesRemaining > MQTT_MAX_TRANSFER_SIZE)?MQTT_MAX_TRANSFER_SIZE:bytesRemaining;
|
---|
| 455 | rc = _client->write(writeBuf,bytesToWrite);
|
---|
| 456 | result = (rc == bytesToWrite);
|
---|
| 457 | bytesRemaining -= rc;
|
---|
| 458 | writeBuf += rc;
|
---|
| 459 | }
|
---|
| 460 | return result;
|
---|
| 461 | #else
|
---|
| 462 | rc = _client->write(buf+(4-llen),length+1+llen);
|
---|
| 463 | lastOutActivity = millis();
|
---|
| 464 | return (rc == 1+llen+length);
|
---|
| 465 | #endif
|
---|
| 466 | }
|
---|
| 467 |
|
---|
| 468 | boolean PubSubClient::subscribe(const char* topic) {
|
---|
| 469 | return subscribe(topic, 0);
|
---|
| 470 | }
|
---|
| 471 |
|
---|
| 472 | boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
|
---|
| 473 | if (qos < 0 || qos > 1) {
|
---|
| 474 | return false;
|
---|
| 475 | }
|
---|
| 476 | if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) {
|
---|
| 477 | // Too long
|
---|
| 478 | return false;
|
---|
| 479 | }
|
---|
| 480 | if (connected()) {
|
---|
| 481 | // Leave room in the buffer for header and variable length field
|
---|
| 482 | uint16_t length = 5;
|
---|
| 483 | nextMsgId++;
|
---|
| 484 | if (nextMsgId == 0) {
|
---|
| 485 | nextMsgId = 1;
|
---|
| 486 | }
|
---|
| 487 | buffer[length++] = (nextMsgId >> 8);
|
---|
| 488 | buffer[length++] = (nextMsgId & 0xFF);
|
---|
| 489 | length = writeString((char*)topic, buffer,length);
|
---|
| 490 | buffer[length++] = qos;
|
---|
| 491 | return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-5);
|
---|
| 492 | }
|
---|
| 493 | return false;
|
---|
| 494 | }
|
---|
| 495 |
|
---|
| 496 | boolean PubSubClient::unsubscribe(const char* topic) {
|
---|
| 497 | if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) {
|
---|
| 498 | // Too long
|
---|
| 499 | return false;
|
---|
| 500 | }
|
---|
| 501 | if (connected()) {
|
---|
| 502 | uint16_t length = 5;
|
---|
| 503 | nextMsgId++;
|
---|
| 504 | if (nextMsgId == 0) {
|
---|
| 505 | nextMsgId = 1;
|
---|
| 506 | }
|
---|
| 507 | buffer[length++] = (nextMsgId >> 8);
|
---|
| 508 | buffer[length++] = (nextMsgId & 0xFF);
|
---|
| 509 | length = writeString(topic, buffer,length);
|
---|
| 510 | return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-5);
|
---|
| 511 | }
|
---|
| 512 | return false;
|
---|
| 513 | }
|
---|
| 514 |
|
---|
| 515 | void PubSubClient::disconnect() {
|
---|
| 516 | buffer[0] = MQTTDISCONNECT;
|
---|
| 517 | buffer[1] = 0;
|
---|
| 518 | _client->write(buffer,2);
|
---|
| 519 | _state = MQTT_DISCONNECTED;
|
---|
| 520 | _client->stop();
|
---|
| 521 | lastInActivity = lastOutActivity = millis();
|
---|
| 522 | }
|
---|
| 523 |
|
---|
| 524 | uint16_t PubSubClient::writeString(const char* string, uint8_t* buf, uint16_t pos) {
|
---|
| 525 | const char* idp = string;
|
---|
| 526 | uint16_t i = 0;
|
---|
| 527 | pos += 2;
|
---|
| 528 | while (*idp) {
|
---|
| 529 | buf[pos++] = *idp++;
|
---|
| 530 | i++;
|
---|
| 531 | }
|
---|
| 532 | buf[pos-i-2] = (i >> 8);
|
---|
| 533 | buf[pos-i-1] = (i & 0xFF);
|
---|
| 534 | return pos;
|
---|
| 535 | }
|
---|
| 536 |
|
---|
| 537 |
|
---|
| 538 | boolean PubSubClient::connected() {
|
---|
| 539 | boolean rc;
|
---|
| 540 | if (_client == NULL ) {
|
---|
| 541 | rc = false;
|
---|
| 542 | } else {
|
---|
| 543 | rc = (int)_client->connected();
|
---|
| 544 | if (!rc) {
|
---|
| 545 | if (this->_state == MQTT_CONNECTED) {
|
---|
| 546 | this->_state = MQTT_CONNECTION_LOST;
|
---|
| 547 | _client->flush();
|
---|
| 548 | _client->stop();
|
---|
| 549 | }
|
---|
| 550 | }
|
---|
| 551 | }
|
---|
| 552 | return rc;
|
---|
| 553 | }
|
---|
| 554 |
|
---|
| 555 | PubSubClient& PubSubClient::setServer(uint8_t * ip, uint16_t port) {
|
---|
| 556 | IPAddress addr(ip[0],ip[1],ip[2],ip[3]);
|
---|
| 557 | return setServer(addr,port);
|
---|
| 558 | }
|
---|
| 559 |
|
---|
| 560 | PubSubClient& PubSubClient::setServer(IPAddress ip, uint16_t port) {
|
---|
| 561 | this->ip = ip;
|
---|
| 562 | this->port = port;
|
---|
| 563 | this->domain = NULL;
|
---|
| 564 | return *this;
|
---|
| 565 | }
|
---|
| 566 |
|
---|
| 567 | PubSubClient& PubSubClient::setServer(const char * domain, uint16_t port) {
|
---|
| 568 | this->domain = domain;
|
---|
| 569 | this->port = port;
|
---|
| 570 | return *this;
|
---|
| 571 | }
|
---|
| 572 |
|
---|
| 573 | PubSubClient& PubSubClient::setCallback(MQTT_CALLBACK_SIGNATURE) {
|
---|
| 574 | this->callback = callback;
|
---|
| 575 | return *this;
|
---|
| 576 | }
|
---|
| 577 |
|
---|
| 578 | PubSubClient& PubSubClient::setClient(Client& client){
|
---|
| 579 | this->_client = &client;
|
---|
| 580 | return *this;
|
---|
| 581 | }
|
---|
| 582 |
|
---|
| 583 | PubSubClient& PubSubClient::setStream(Stream& stream){
|
---|
| 584 | this->stream = &stream;
|
---|
| 585 | return *this;
|
---|
| 586 | }
|
---|
| 587 |
|
---|
| 588 | int PubSubClient::state() {
|
---|
| 589 | return this->_state;
|
---|
| 590 | }
|
---|