1/** 2 * @file 3 * MQTT client 4 * 5 * @defgroup mqtt MQTT client 6 * @ingroup apps 7 * @verbinclude mqtt_client.txt 8 */ 9 10/* 11 * Copyright (c) 2016 Erik Andersson <erian747@gmail.com> 12 * All rights reserved. 13 * 14 * Redistribution and use in source and binary forms, with or without modification, 15 * are permitted provided that the following conditions are met: 16 * 17 * 1. Redistributions of source code must retain the above copyright notice, 18 * this list of conditions and the following disclaimer. 19 * 2. Redistributions in binary form must reproduce the above copyright notice, 20 * this list of conditions and the following disclaimer in the documentation 21 * and/or other materials provided with the distribution. 22 * 3. The name of the author may not be used to endorse or promote products 23 * derived from this software without specific prior written permission. 24 * 25 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED 26 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF 27 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT 28 * SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 29 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT 30 * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 31 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 32 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING 33 * IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY 34 * OF SUCH DAMAGE. 35 * 36 * This file is part of the lwIP TCP/IP stack 37 * 38 * Author: Erik Andersson <erian747@gmail.com> 39 * 40 * 41 * @todo: 42 * - Handle large outgoing payloads for PUBLISH messages 43 * - Fix restriction of a single topic in each (UN)SUBSCRIBE message (protocol has support for multiple topics) 44 * - Add support for legacy MQTT protocol version 45 * 46 * Please coordinate changes and requests with Erik Andersson 47 * Erik Andersson <erian747@gmail.com> 48 * 49 */ 50#include "lwip/apps/mqtt.h" 51#include "lwip/apps/mqtt_priv.h" 52#include "lwip/timeouts.h" 53#include "lwip/ip_addr.h" 54#include "lwip/mem.h" 55#include "lwip/err.h" 56#include "lwip/pbuf.h" 57#include "lwip/altcp.h" 58#include "lwip/altcp_tcp.h" 59#include "lwip/altcp_tls.h" 60#include <string.h> 61 62#if LWIP_TCP && LWIP_CALLBACK_API 63 64/** 65 * MQTT_DEBUG: Default is off. 66 */ 67#if !defined MQTT_DEBUG || defined __DOXYGEN__ 68#define MQTT_DEBUG LWIP_DBG_OFF 69#endif 70 71#define MQTT_DEBUG_TRACE (MQTT_DEBUG | LWIP_DBG_TRACE) 72#define MQTT_DEBUG_STATE (MQTT_DEBUG | LWIP_DBG_STATE) 73#define MQTT_DEBUG_WARN (MQTT_DEBUG | LWIP_DBG_LEVEL_WARNING) 74#define MQTT_DEBUG_WARN_STATE (MQTT_DEBUG | LWIP_DBG_LEVEL_WARNING | LWIP_DBG_STATE) 75#define MQTT_DEBUG_SERIOUS (MQTT_DEBUG | LWIP_DBG_LEVEL_SERIOUS) 76 77 78 79/** 80 * MQTT client connection states 81 */ 82enum { 83 TCP_DISCONNECTED, 84 TCP_CONNECTING, 85 MQTT_CONNECTING, 86 MQTT_CONNECTED 87}; 88 89/** 90 * MQTT control message types 91 */ 92enum mqtt_message_type { 93 MQTT_MSG_TYPE_CONNECT = 1, 94 MQTT_MSG_TYPE_CONNACK = 2, 95 MQTT_MSG_TYPE_PUBLISH = 3, 96 MQTT_MSG_TYPE_PUBACK = 4, 97 MQTT_MSG_TYPE_PUBREC = 5, 98 MQTT_MSG_TYPE_PUBREL = 6, 99 MQTT_MSG_TYPE_PUBCOMP = 7, 100 MQTT_MSG_TYPE_SUBSCRIBE = 8, 101 MQTT_MSG_TYPE_SUBACK = 9, 102 MQTT_MSG_TYPE_UNSUBSCRIBE = 10, 103 MQTT_MSG_TYPE_UNSUBACK = 11, 104 MQTT_MSG_TYPE_PINGREQ = 12, 105 MQTT_MSG_TYPE_PINGRESP = 13, 106 MQTT_MSG_TYPE_DISCONNECT = 14 107}; 108 109/** Helpers to extract control packet type and qos from first byte in fixed header */ 110#define MQTT_CTL_PACKET_TYPE(fixed_hdr_byte0) ((fixed_hdr_byte0 & 0xf0) >> 4) 111#define MQTT_CTL_PACKET_QOS(fixed_hdr_byte0) ((fixed_hdr_byte0 & 0x6) >> 1) 112 113/** 114 * MQTT connect flags, only used in CONNECT message 115 */ 116enum mqtt_connect_flag { 117 MQTT_CONNECT_FLAG_USERNAME = 1 << 7, 118 MQTT_CONNECT_FLAG_PASSWORD = 1 << 6, 119 MQTT_CONNECT_FLAG_WILL_RETAIN = 1 << 5, 120 MQTT_CONNECT_FLAG_WILL = 1 << 2, 121 MQTT_CONNECT_FLAG_CLEAN_SESSION = 1 << 1 122}; 123 124 125static void mqtt_cyclic_timer(void *arg); 126 127#if defined(LWIP_DEBUG) 128static const char *const mqtt_message_type_str[15] = { 129 "UNDEFINED", 130 "CONNECT", 131 "CONNACK", 132 "PUBLISH", 133 "PUBACK", 134 "PUBREC", 135 "PUBREL", 136 "PUBCOMP", 137 "SUBSCRIBE", 138 "SUBACK", 139 "UNSUBSCRIBE", 140 "UNSUBACK", 141 "PINGREQ", 142 "PINGRESP", 143 "DISCONNECT" 144}; 145 146/** 147 * Message type value to string 148 * @param msg_type see enum mqtt_message_type 149 * 150 * @return Control message type text string 151 */ 152static const char * 153mqtt_msg_type_to_str(u8_t msg_type) 154{ 155 if (msg_type >= LWIP_ARRAYSIZE(mqtt_message_type_str)) { 156 msg_type = 0; 157 } 158 return mqtt_message_type_str[msg_type]; 159} 160 161#endif 162 163 164/** 165 * Generate MQTT packet identifier 166 * @param client MQTT client 167 * @return New packet identifier, range 1 to 65535 168 */ 169static u16_t 170msg_generate_packet_id(mqtt_client_t *client) 171{ 172 client->pkt_id_seq++; 173 if (client->pkt_id_seq == 0) { 174 client->pkt_id_seq++; 175 } 176 return client->pkt_id_seq; 177} 178 179/*--------------------------------------------------------------------------------------------------------------------- */ 180/* Output ring buffer */ 181 182/** Add single item to ring buffer */ 183static void 184mqtt_ringbuf_put(struct mqtt_ringbuf_t *rb, u8_t item) 185{ 186 rb->buf[rb->put] = item; 187 rb->put++; 188 if (rb->put >= MQTT_OUTPUT_RINGBUF_SIZE) { 189 rb->put = 0; 190 } 191} 192 193/** Return pointer to ring buffer get position */ 194static u8_t * 195mqtt_ringbuf_get_ptr(struct mqtt_ringbuf_t *rb) 196{ 197 return &rb->buf[rb->get]; 198} 199 200static void 201mqtt_ringbuf_advance_get_idx(struct mqtt_ringbuf_t *rb, u16_t len) 202{ 203 LWIP_ASSERT("mqtt_ringbuf_advance_get_idx: len < MQTT_OUTPUT_RINGBUF_SIZE", len < MQTT_OUTPUT_RINGBUF_SIZE); 204 205 rb->get += len; 206 if (rb->get >= MQTT_OUTPUT_RINGBUF_SIZE) { 207 rb->get = rb->get - MQTT_OUTPUT_RINGBUF_SIZE; 208 } 209} 210 211/** Return number of bytes in ring buffer */ 212static u16_t 213mqtt_ringbuf_len(struct mqtt_ringbuf_t *rb) 214{ 215 u32_t len = rb->put - rb->get; 216 if (len > 0xFFFF) { 217 len += MQTT_OUTPUT_RINGBUF_SIZE; 218 } 219 return (u16_t)len; 220} 221 222/** Return number of bytes free in ring buffer */ 223#define mqtt_ringbuf_free(rb) (MQTT_OUTPUT_RINGBUF_SIZE - mqtt_ringbuf_len(rb)) 224 225/** Return number of bytes possible to read without wrapping around */ 226#define mqtt_ringbuf_linear_read_length(rb) LWIP_MIN(mqtt_ringbuf_len(rb), (MQTT_OUTPUT_RINGBUF_SIZE - (rb)->get)) 227 228/** 229 * Try send as many bytes as possible from output ring buffer 230 * @param rb Output ring buffer 231 * @param tpcb TCP connection handle 232 */ 233static void 234mqtt_output_send(struct mqtt_ringbuf_t *rb, struct altcp_pcb *tpcb) 235{ 236 err_t err; 237 u8_t wrap = 0; 238 u16_t ringbuf_lin_len = mqtt_ringbuf_linear_read_length(rb); 239 u16_t send_len = altcp_sndbuf(tpcb); 240 LWIP_ASSERT("mqtt_output_send: tpcb != NULL", tpcb != NULL); 241 242 if (send_len == 0 || ringbuf_lin_len == 0) { 243 return; 244 } 245 246 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_output_send: tcp_sndbuf: %d bytes, ringbuf_linear_available: %d, get %d, put %d\n", 247 send_len, ringbuf_lin_len, rb->get, rb->put)); 248 249 if (send_len > ringbuf_lin_len) { 250 /* Space in TCP output buffer is larger than available in ring buffer linear portion */ 251 send_len = ringbuf_lin_len; 252 /* Wrap around if more data in ring buffer after linear portion */ 253 wrap = (mqtt_ringbuf_len(rb) > ringbuf_lin_len); 254 } 255 err = altcp_write(tpcb, mqtt_ringbuf_get_ptr(rb), send_len, TCP_WRITE_FLAG_COPY | (wrap ? TCP_WRITE_FLAG_MORE : 0)); 256 if ((err == ERR_OK) && wrap) { 257 mqtt_ringbuf_advance_get_idx(rb, send_len); 258 /* Use the lesser one of ring buffer linear length and TCP send buffer size */ 259 send_len = LWIP_MIN(altcp_sndbuf(tpcb), mqtt_ringbuf_linear_read_length(rb)); 260 err = altcp_write(tpcb, mqtt_ringbuf_get_ptr(rb), send_len, TCP_WRITE_FLAG_COPY); 261 } 262 263 if (err == ERR_OK) { 264 mqtt_ringbuf_advance_get_idx(rb, send_len); 265 /* Flush */ 266 altcp_output(tpcb); 267 } else { 268 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_output_send: Send failed with err %d (\"%s\")\n", err, lwip_strerr(err))); 269 } 270} 271 272 273 274/*--------------------------------------------------------------------------------------------------------------------- */ 275/* Request queue */ 276 277/** 278 * Create request item 279 * @param r_objs Pointer to request objects 280 * @param r_objs_len Number of array entries 281 * @param pkt_id Packet identifier of request 282 * @param cb Packet callback to call when requests lifetime ends 283 * @param arg Parameter following callback 284 * @return Request or NULL if failed to create 285 */ 286static struct mqtt_request_t * 287mqtt_create_request(struct mqtt_request_t *r_objs, size_t r_objs_len, u16_t pkt_id, mqtt_request_cb_t cb, void *arg) 288{ 289 struct mqtt_request_t *r = NULL; 290 u8_t n; 291 LWIP_ASSERT("mqtt_create_request: r_objs != NULL", r_objs != NULL); 292 for (n = 0; n < r_objs_len; n++) { 293 /* Item point to itself if not in use */ 294 if (r_objs[n].next == &r_objs[n]) { 295 r = &r_objs[n]; 296 r->next = NULL; 297 r->cb = cb; 298 r->arg = arg; 299 r->pkt_id = pkt_id; 300 break; 301 } 302 } 303 return r; 304} 305 306 307/** 308 * Append request to pending request queue 309 * @param tail Pointer to request queue tail pointer 310 * @param r Request to append 311 */ 312static void 313mqtt_append_request(struct mqtt_request_t **tail, struct mqtt_request_t *r) 314{ 315 struct mqtt_request_t *head = NULL; 316 s16_t time_before = 0; 317 struct mqtt_request_t *iter; 318 319 LWIP_ASSERT("mqtt_append_request: tail != NULL", tail != NULL); 320 321 /* Iterate trough queue to find head, and count total timeout time */ 322 for (iter = *tail; iter != NULL; iter = iter->next) { 323 time_before += iter->timeout_diff; 324 head = iter; 325 } 326 327 LWIP_ASSERT("mqtt_append_request: time_before <= MQTT_REQ_TIMEOUT", time_before <= MQTT_REQ_TIMEOUT); 328 r->timeout_diff = MQTT_REQ_TIMEOUT - time_before; 329 if (head == NULL) { 330 *tail = r; 331 } else { 332 head->next = r; 333 } 334} 335 336 337/** 338 * Delete request item 339 * @param r Request item to delete 340 */ 341static void 342mqtt_delete_request(struct mqtt_request_t *r) 343{ 344 if (r != NULL) { 345 r->next = r; 346 } 347} 348 349/** 350 * Remove a request item with a specific packet identifier from request queue 351 * @param tail Pointer to request queue tail pointer 352 * @param pkt_id Packet identifier of request to take 353 * @return Request item if found, NULL if not 354 */ 355static struct mqtt_request_t * 356mqtt_take_request(struct mqtt_request_t **tail, u16_t pkt_id) 357{ 358 struct mqtt_request_t *iter = NULL, *prev = NULL; 359 LWIP_ASSERT("mqtt_take_request: tail != NULL", tail != NULL); 360 /* Search all request for pkt_id */ 361 for (iter = *tail; iter != NULL; iter = iter->next) { 362 if (iter->pkt_id == pkt_id) { 363 break; 364 } 365 prev = iter; 366 } 367 368 /* If request was found */ 369 if (iter != NULL) { 370 /* unchain */ 371 if (prev == NULL) { 372 *tail = iter->next; 373 } else { 374 prev->next = iter->next; 375 } 376 /* If exists, add remaining timeout time for the request to next */ 377 if (iter->next != NULL) { 378 iter->next->timeout_diff += iter->timeout_diff; 379 } 380 iter->next = NULL; 381 } 382 return iter; 383} 384 385/** 386 * Handle requests timeout 387 * @param tail Pointer to request queue tail pointer 388 * @param t Time since last call in seconds 389 */ 390static void 391mqtt_request_time_elapsed(struct mqtt_request_t **tail, u8_t t) 392{ 393 struct mqtt_request_t *r; 394 LWIP_ASSERT("mqtt_request_time_elapsed: tail != NULL", tail != NULL); 395 r = *tail; 396 while (t > 0 && r != NULL) { 397 if (t >= r->timeout_diff) { 398 t -= (u8_t)r->timeout_diff; 399 /* Unchain */ 400 *tail = r->next; 401 /* Notify upper layer about timeout */ 402 if (r->cb != NULL) { 403 r->cb(r->arg, ERR_TIMEOUT); 404 } 405 mqtt_delete_request(r); 406 /* Tail might be be modified in callback, so re-read it in every iteration */ 407 r = *(struct mqtt_request_t *const volatile *)tail; 408 } else { 409 r->timeout_diff -= t; 410 t = 0; 411 } 412 } 413} 414 415/** 416 * Free all request items 417 * @param tail Pointer to request queue tail pointer 418 */ 419static void 420mqtt_clear_requests(struct mqtt_request_t **tail) 421{ 422 struct mqtt_request_t *iter, *next; 423 LWIP_ASSERT("mqtt_clear_requests: tail != NULL", tail != NULL); 424 for (iter = *tail; iter != NULL; iter = next) { 425 next = iter->next; 426 mqtt_delete_request(iter); 427 } 428 *tail = NULL; 429} 430/** 431 * Initialize all request items 432 * @param r_objs Pointer to request objects 433 * @param r_objs_len Number of array entries 434 */ 435static void 436mqtt_init_requests(struct mqtt_request_t *r_objs, size_t r_objs_len) 437{ 438 u8_t n; 439 LWIP_ASSERT("mqtt_init_requests: r_objs != NULL", r_objs != NULL); 440 for (n = 0; n < r_objs_len; n++) { 441 /* Item pointing to itself indicates unused */ 442 r_objs[n].next = &r_objs[n]; 443 } 444} 445 446/*--------------------------------------------------------------------------------------------------------------------- */ 447/* Output message build helpers */ 448 449 450static void 451mqtt_output_append_u8(struct mqtt_ringbuf_t *rb, u8_t value) 452{ 453 mqtt_ringbuf_put(rb, value); 454} 455 456static 457void mqtt_output_append_u16(struct mqtt_ringbuf_t *rb, u16_t value) 458{ 459 mqtt_ringbuf_put(rb, value >> 8); 460 mqtt_ringbuf_put(rb, value & 0xff); 461} 462 463static void 464mqtt_output_append_buf(struct mqtt_ringbuf_t *rb, const void *data, u16_t length) 465{ 466 u16_t n; 467 for (n = 0; n < length; n++) { 468 mqtt_ringbuf_put(rb, ((const u8_t *)data)[n]); 469 } 470} 471 472static void 473mqtt_output_append_string(struct mqtt_ringbuf_t *rb, const char *str, u16_t length) 474{ 475 u16_t n; 476 mqtt_ringbuf_put(rb, length >> 8); 477 mqtt_ringbuf_put(rb, length & 0xff); 478 for (n = 0; n < length; n++) { 479 mqtt_ringbuf_put(rb, str[n]); 480 } 481} 482 483/** 484 * Append fixed header 485 * @param rb Output ring buffer 486 * @param msg_type see enum mqtt_message_type 487 * @param fdup MQTT DUP flag 488 * @param fqos MQTT QoS field 489 * @param fretain MQTT retain flag 490 * @param r_length Remaining length after fixed header 491 */ 492 493static void 494mqtt_output_append_fixed_header(struct mqtt_ringbuf_t *rb, u8_t msg_type, u8_t fdup, 495 u8_t fqos, u8_t fretain, u16_t r_length) 496{ 497 /* Start with control byte */ 498 mqtt_output_append_u8(rb, (((msg_type & 0x0f) << 4) | ((fdup & 1) << 3) | ((fqos & 3) << 1) | (fretain & 1))); 499 /* Encode remaining length field */ 500 do { 501 mqtt_output_append_u8(rb, (r_length & 0x7f) | (r_length >= 128 ? 0x80 : 0)); 502 r_length >>= 7; 503 } while (r_length > 0); 504} 505 506 507/** 508 * Check output buffer space 509 * @param rb Output ring buffer 510 * @param r_length Remaining length after fixed header 511 * @return 1 if message will fit, 0 if not enough buffer space 512 */ 513static u8_t 514mqtt_output_check_space(struct mqtt_ringbuf_t *rb, u16_t r_length) 515{ 516 /* Start with length of type byte + remaining length */ 517 u16_t total_len = 1 + r_length; 518 519 LWIP_ASSERT("mqtt_output_check_space: rb != NULL", rb != NULL); 520 521 /* Calculate number of required bytes to contain the remaining bytes field and add to total*/ 522 do { 523 total_len++; 524 r_length >>= 7; 525 } while (r_length > 0); 526 527 return (total_len <= mqtt_ringbuf_free(rb)); 528} 529 530 531/** 532 * Close connection to server 533 * @param client MQTT client 534 * @param reason Reason for disconnection 535 */ 536static void 537mqtt_close(mqtt_client_t *client, mqtt_connection_status_t reason) 538{ 539 LWIP_ASSERT("mqtt_close: client != NULL", client != NULL); 540 541 /* Bring down TCP connection if not already done */ 542 if (client->conn != NULL) { 543 err_t res; 544 altcp_recv(client->conn, NULL); 545 altcp_err(client->conn, NULL); 546 altcp_sent(client->conn, NULL); 547 res = altcp_close(client->conn); 548 if (res != ERR_OK) { 549 altcp_abort(client->conn); 550 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_close: Close err=%s\n", lwip_strerr(res))); 551 } 552 client->conn = NULL; 553 } 554 555 /* Remove all pending requests */ 556 mqtt_clear_requests(&client->pend_req_queue); 557 /* Stop cyclic timer */ 558 sys_untimeout(mqtt_cyclic_timer, client); 559 560 /* Notify upper layer of disconnection if changed state */ 561 if (client->conn_state != TCP_DISCONNECTED) { 562 563 client->conn_state = TCP_DISCONNECTED; 564 if (client->connect_cb != NULL) { 565 client->connect_cb(client, client->connect_arg, reason); 566 } 567 } 568} 569 570 571/** 572 * Interval timer, called every MQTT_CYCLIC_TIMER_INTERVAL seconds in MQTT_CONNECTING and MQTT_CONNECTED states 573 * @param arg MQTT client 574 */ 575static void 576mqtt_cyclic_timer(void *arg) 577{ 578 u8_t restart_timer = 1; 579 mqtt_client_t *client = (mqtt_client_t *)arg; 580 LWIP_ASSERT("mqtt_cyclic_timer: client != NULL", client != NULL); 581 582 if (client->conn_state == MQTT_CONNECTING) { 583 client->cyclic_tick++; 584 if ((client->cyclic_tick * MQTT_CYCLIC_TIMER_INTERVAL) >= MQTT_CONNECT_TIMOUT) { 585 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_cyclic_timer: CONNECT attempt to server timed out\n")); 586 /* Disconnect TCP */ 587 mqtt_close(client, MQTT_CONNECT_TIMEOUT); 588 restart_timer = 0; 589 } 590 } else if (client->conn_state == MQTT_CONNECTED) { 591 /* Handle timeout for pending requests */ 592 mqtt_request_time_elapsed(&client->pend_req_queue, MQTT_CYCLIC_TIMER_INTERVAL); 593 594 /* keep_alive > 0 means keep alive functionality shall be used */ 595 if (client->keep_alive > 0) { 596 597 client->server_watchdog++; 598 /* If reception from server has been idle for 1.5*keep_alive time, server is considered unresponsive */ 599 if ((client->server_watchdog * MQTT_CYCLIC_TIMER_INTERVAL) > (client->keep_alive + client->keep_alive / 2)) { 600 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_cyclic_timer: Server incoming keep-alive timeout\n")); 601 mqtt_close(client, MQTT_CONNECT_TIMEOUT); 602 restart_timer = 0; 603 } 604 605 /* If time for a keep alive message to be sent, transmission has been idle for keep_alive time */ 606 if ((client->cyclic_tick * MQTT_CYCLIC_TIMER_INTERVAL) >= client->keep_alive) { 607 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_cyclic_timer: Sending keep-alive message to server\n")); 608 if (mqtt_output_check_space(&client->output, 0) != 0) { 609 mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_PINGREQ, 0, 0, 0, 0); 610 client->cyclic_tick = 0; 611 } 612 } else { 613 client->cyclic_tick++; 614 } 615 } 616 } else { 617 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_cyclic_timer: Timer should not be running in state %d\n", client->conn_state)); 618 restart_timer = 0; 619 } 620 if (restart_timer) { 621 sys_timeout(MQTT_CYCLIC_TIMER_INTERVAL * 1000, mqtt_cyclic_timer, arg); 622 } 623} 624 625 626/** 627 * Send PUBACK, PUBREC or PUBREL response message 628 * @param client MQTT client 629 * @param msg PUBACK, PUBREC or PUBREL 630 * @param pkt_id Packet identifier 631 * @param qos QoS value 632 * @return ERR_OK if successful, ERR_MEM if out of memory 633 */ 634static err_t 635pub_ack_rec_rel_response(mqtt_client_t *client, u8_t msg, u16_t pkt_id, u8_t qos) 636{ 637 err_t err = ERR_OK; 638 if (mqtt_output_check_space(&client->output, 2)) { 639 mqtt_output_append_fixed_header(&client->output, msg, 0, qos, 0, 2); 640 mqtt_output_append_u16(&client->output, pkt_id); 641 mqtt_output_send(&client->output, client->conn); 642 } else { 643 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("pub_ack_rec_rel_response: OOM creating response: %s with pkt_id: %d\n", 644 mqtt_msg_type_to_str(msg), pkt_id)); 645 err = ERR_MEM; 646 } 647 return err; 648} 649 650/** 651 * Subscribe response from server 652 * @param r Matching request 653 * @param result Result code from server 654 */ 655static void 656mqtt_incomming_suback(struct mqtt_request_t *r, u8_t result) 657{ 658 if (r->cb != NULL) { 659 r->cb(r->arg, result < 3 ? ERR_OK : ERR_ABRT); 660 } 661} 662 663 664/** 665 * Complete MQTT message received or buffer full 666 * @param client MQTT client 667 * @param fixed_hdr_len length of fixed header 668 * @param length length received part 669 * @param remaining_length Remaining length of complete message 670 */ 671static mqtt_connection_status_t 672mqtt_message_received(mqtt_client_t *client, u8_t fixed_hdr_len, u16_t length, u32_t remaining_length) 673{ 674 mqtt_connection_status_t res = MQTT_CONNECT_ACCEPTED; 675 676 u8_t *var_hdr_payload = client->rx_buffer + fixed_hdr_len; 677 size_t var_hdr_payload_bufsize = sizeof(client->rx_buffer) - fixed_hdr_len; 678 679 /* Control packet type */ 680 u8_t pkt_type = MQTT_CTL_PACKET_TYPE(client->rx_buffer[0]); 681 u16_t pkt_id = 0; 682 683 LWIP_ASSERT("fixed_hdr_len <= client->msg_idx", fixed_hdr_len <= client->msg_idx); 684 LWIP_ERROR("buffer length mismatch", fixed_hdr_len + length <= MQTT_VAR_HEADER_BUFFER_LEN, 685 return MQTT_CONNECT_DISCONNECTED); 686 687 if (pkt_type == MQTT_MSG_TYPE_CONNACK) { 688 if (client->conn_state == MQTT_CONNECTING) { 689 if (length < 2) { 690 LWIP_DEBUGF(MQTT_DEBUG_WARN,( "mqtt_message_received: Received short CONNACK message\n")); 691 goto out_disconnect; 692 } 693 /* Get result code from CONNACK */ 694 res = (mqtt_connection_status_t)var_hdr_payload[1]; 695 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_message_received: Connect response code %d\n", res)); 696 if (res == MQTT_CONNECT_ACCEPTED) { 697 /* Reset cyclic_tick when changing to connected state */ 698 client->cyclic_tick = 0; 699 client->conn_state = MQTT_CONNECTED; 700 /* Notify upper layer */ 701 if (client->connect_cb != 0) { 702 client->connect_cb(client, client->connect_arg, res); 703 } 704 } 705 } else { 706 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_message_received: Received CONNACK in connected state\n")); 707 } 708 } else if (pkt_type == MQTT_MSG_TYPE_PINGRESP) { 709 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ( "mqtt_message_received: Received PINGRESP from server\n")); 710 711 } else if (pkt_type == MQTT_MSG_TYPE_PUBLISH) { 712 u16_t payload_offset = 0; 713 u16_t payload_length = length; 714 u8_t qos = MQTT_CTL_PACKET_QOS(client->rx_buffer[0]); 715 716 if (client->msg_idx <= MQTT_VAR_HEADER_BUFFER_LEN) { 717 /* Should have topic and pkt id*/ 718 u8_t *topic; 719 u16_t after_topic; 720 u8_t bkp; 721 u16_t topic_len; 722 u16_t qos_len = (qos ? 2U : 0U); 723 if (length < 2 + qos_len) { 724 LWIP_DEBUGF(MQTT_DEBUG_WARN,( "mqtt_message_received: Received short PUBLISH packet\n")); 725 goto out_disconnect; 726 } 727 topic_len = var_hdr_payload[0]; 728 topic_len = (topic_len << 8) + (u16_t)(var_hdr_payload[1]); 729 if ((topic_len > length - (2 + qos_len)) || 730 (topic_len > var_hdr_payload_bufsize - (2 + qos_len))) { 731 LWIP_DEBUGF(MQTT_DEBUG_WARN,( "mqtt_message_received: Received short PUBLISH packet (topic)\n")); 732 goto out_disconnect; 733 } 734 735 topic = var_hdr_payload + 2; 736 after_topic = 2 + topic_len; 737 /* Check buffer length, add one byte even for QoS 0 so that zero termination will fit */ 738 if ((after_topic + (qos ? 2U : 1U)) > var_hdr_payload_bufsize) { 739 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_message_received: Receive buffer can not fit topic + pkt_id\n")); 740 goto out_disconnect; 741 } 742 743 /* id for QoS 1 and 2 */ 744 if (qos > 0) { 745 if (length < after_topic + 2U) { 746 LWIP_DEBUGF(MQTT_DEBUG_WARN,( "mqtt_message_received: Received short PUBLISH packet (after_topic)\n")); 747 goto out_disconnect; 748 } 749 client->inpub_pkt_id = ((u16_t)var_hdr_payload[after_topic] << 8) + (u16_t)var_hdr_payload[after_topic + 1]; 750 after_topic += 2; 751 } else { 752 client->inpub_pkt_id = 0; 753 } 754 /* Take backup of byte after topic */ 755 bkp = topic[topic_len]; 756 /* Zero terminate string */ 757 topic[topic_len] = 0; 758 /* Payload data remaining in receive buffer */ 759 payload_length = length - after_topic; 760 payload_offset = after_topic; 761 762 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_incomming_publish: Received message with QoS %d at topic: %s, payload length %"U32_F"\n", 763 qos, topic, remaining_length + payload_length)); 764 if (client->pub_cb != NULL) { 765 client->pub_cb(client->inpub_arg, (const char *)topic, remaining_length + payload_length); 766 } 767 /* Restore byte after topic */ 768 topic[topic_len] = bkp; 769 } 770 if (payload_length > 0 || remaining_length == 0) { 771 if (length < (size_t)(payload_offset + payload_length)) { 772 LWIP_DEBUGF(MQTT_DEBUG_WARN,( "mqtt_message_received: Received short packet (payload)\n")); 773 goto out_disconnect; 774 } 775 if (client->data_cb != NULL) { 776 client->data_cb(client->inpub_arg, var_hdr_payload + payload_offset, payload_length, remaining_length == 0 ? MQTT_DATA_FLAG_LAST : 0); 777 } 778 /* Reply if QoS > 0 */ 779 if (remaining_length == 0 && qos > 0) { 780 /* Send PUBACK for QoS 1 or PUBREC for QoS 2 */ 781 u8_t resp_msg = (qos == 1) ? MQTT_MSG_TYPE_PUBACK : MQTT_MSG_TYPE_PUBREC; 782 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_incomming_publish: Sending publish response: %s with pkt_id: %d\n", 783 mqtt_msg_type_to_str(resp_msg), client->inpub_pkt_id)); 784 pub_ack_rec_rel_response(client, resp_msg, client->inpub_pkt_id, 0); 785 } 786 } 787 } else { 788 /* Get packet identifier */ 789 pkt_id = (u16_t)var_hdr_payload[0] << 8; 790 pkt_id |= (u16_t)var_hdr_payload[1]; 791 if (pkt_id == 0) { 792 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_message_received: Got message with illegal packet identifier: 0\n")); 793 goto out_disconnect; 794 } 795 if (pkt_type == MQTT_MSG_TYPE_PUBREC) { 796 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_message_received: PUBREC, sending PUBREL with pkt_id: %d\n", pkt_id)); 797 pub_ack_rec_rel_response(client, MQTT_MSG_TYPE_PUBREL, pkt_id, 1); 798 799 } else if (pkt_type == MQTT_MSG_TYPE_PUBREL) { 800 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_message_received: PUBREL, sending PUBCOMP response with pkt_id: %d\n", pkt_id)); 801 pub_ack_rec_rel_response(client, MQTT_MSG_TYPE_PUBCOMP, pkt_id, 0); 802 803 } else if (pkt_type == MQTT_MSG_TYPE_SUBACK || pkt_type == MQTT_MSG_TYPE_UNSUBACK || 804 pkt_type == MQTT_MSG_TYPE_PUBCOMP || pkt_type == MQTT_MSG_TYPE_PUBACK) { 805 struct mqtt_request_t *r = mqtt_take_request(&client->pend_req_queue, pkt_id); 806 if (r != NULL) { 807 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_message_received: %s response with id %d\n", mqtt_msg_type_to_str(pkt_type), pkt_id)); 808 if (pkt_type == MQTT_MSG_TYPE_SUBACK) { 809 if (length < 3) { 810 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_message_received: To small SUBACK packet\n")); 811 goto out_disconnect; 812 } else { 813 mqtt_incomming_suback(r, var_hdr_payload[2]); 814 } 815 } else if (r->cb != NULL) { 816 r->cb(r->arg, ERR_OK); 817 } 818 mqtt_delete_request(r); 819 } else { 820 LWIP_DEBUGF(MQTT_DEBUG_WARN, ( "mqtt_message_received: Received %s reply, with wrong pkt_id: %d\n", mqtt_msg_type_to_str(pkt_type), pkt_id)); 821 } 822 } else { 823 LWIP_DEBUGF(MQTT_DEBUG_WARN, ( "mqtt_message_received: Received unknown message type: %d\n", pkt_type)); 824 goto out_disconnect; 825 } 826 } 827 return res; 828out_disconnect: 829 return MQTT_CONNECT_DISCONNECTED; 830} 831 832 833/** 834 * MQTT incoming message parser 835 * @param client MQTT client 836 * @param p PBUF chain of received data 837 * @return Connection status 838 */ 839static mqtt_connection_status_t 840mqtt_parse_incoming(mqtt_client_t *client, struct pbuf *p) 841{ 842 u16_t in_offset = 0; 843 u32_t msg_rem_len = 0; 844 u8_t fixed_hdr_len = 0; 845 u8_t b = 0; 846 847 while (p->tot_len > in_offset) { 848 /* We ALWAYS parse the header here first. Even if the header was not 849 included in this segment, we re-parse it here by buffering it in 850 client->rx_buffer. client->msg_idx keeps track of this. */ 851 if ((fixed_hdr_len < 2) || ((b & 0x80) != 0)) { 852 853 if (fixed_hdr_len < client->msg_idx) { 854 /* parse header from old pbuf (buffered in client->rx_buffer) */ 855 b = client->rx_buffer[fixed_hdr_len]; 856 } else { 857 /* parse header from this pbuf and save it in client->rx_buffer in case 858 it comes in segmented */ 859 b = pbuf_get_at(p, in_offset++); 860 client->rx_buffer[client->msg_idx++] = b; 861 } 862 fixed_hdr_len++; 863 864 if (fixed_hdr_len >= 2) { 865 /* fixed header contains at least 2 bytes but can contain more, depending on 866 'remaining length'. All bytes but the last of this have 0x80 set to 867 indicate more bytes are coming. */ 868 msg_rem_len |= (u32_t)(b & 0x7f) << ((fixed_hdr_len - 2) * 7); 869 if ((b & 0x80) == 0) { 870 /* fixed header is done */ 871 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_parse_incoming: Remaining length after fixed header: %"U32_F"\n", msg_rem_len)); 872 if (msg_rem_len == 0) { 873 /* Complete message with no extra headers of payload received */ 874 mqtt_message_received(client, fixed_hdr_len, 0, 0); 875 client->msg_idx = 0; 876 fixed_hdr_len = 0; 877 } else { 878 /* Bytes remaining in message (changes remaining length if this is 879 not the first segment of this message) */ 880 msg_rem_len = (msg_rem_len + fixed_hdr_len) - client->msg_idx; 881 } 882 } 883 } 884 } else { 885 /* Fixed header has been parsed, parse variable header */ 886 u16_t cpy_len, buffer_space; 887 888 /* Allow to copy the lesser one of available length in input data or bytes remaining in message */ 889 cpy_len = (u16_t)LWIP_MIN((u16_t)(p->tot_len - in_offset), msg_rem_len); 890 891 /* Limit to available space in buffer */ 892 buffer_space = MQTT_VAR_HEADER_BUFFER_LEN - fixed_hdr_len; 893 if (cpy_len > buffer_space) { 894 cpy_len = buffer_space; 895 } 896 pbuf_copy_partial(p, client->rx_buffer + fixed_hdr_len, cpy_len, in_offset); 897 898 /* Advance get and put indexes */ 899 client->msg_idx += cpy_len; 900 in_offset += cpy_len; 901 msg_rem_len -= cpy_len; 902 903 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_parse_incoming: msg_idx: %"U32_F", cpy_len: %"U16_F", remaining %"U32_F"\n", client->msg_idx, cpy_len, msg_rem_len)); 904 if ((msg_rem_len == 0) || (cpy_len == buffer_space)) { 905 /* Whole message received or buffer is full */ 906 mqtt_connection_status_t res = mqtt_message_received(client, fixed_hdr_len, cpy_len, msg_rem_len); 907 if (res != MQTT_CONNECT_ACCEPTED) { 908 return res; 909 } 910 if (msg_rem_len == 0) { 911 /* Reset parser state */ 912 client->msg_idx = 0; 913 /* msg_tot_len = 0; */ 914 fixed_hdr_len = 0; 915 } 916 } 917 } 918 } 919 return MQTT_CONNECT_ACCEPTED; 920} 921 922 923/** 924 * TCP received callback function. @see tcp_recv_fn 925 * @param arg MQTT client 926 * @param p PBUF chain of received data 927 * @param err Passed as return value if not ERR_OK 928 * @return ERR_OK or err passed into callback 929 */ 930static err_t 931mqtt_tcp_recv_cb(void *arg, struct altcp_pcb *pcb, struct pbuf *p, err_t err) 932{ 933 mqtt_client_t *client = (mqtt_client_t *)arg; 934 LWIP_ASSERT("mqtt_tcp_recv_cb: client != NULL", client != NULL); 935 LWIP_ASSERT("mqtt_tcp_recv_cb: client->conn == pcb", client->conn == pcb); 936 937 if (p == NULL) { 938 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_tcp_recv_cb: Recv pbuf=NULL, remote has closed connection\n")); 939 mqtt_close(client, MQTT_CONNECT_DISCONNECTED); 940 } else { 941 mqtt_connection_status_t res; 942 if (err != ERR_OK) { 943 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_tcp_recv_cb: Recv err=%d\n", err)); 944 pbuf_free(p); 945 return err; 946 } 947 948 /* Tell remote that data has been received */ 949 altcp_recved(pcb, p->tot_len); 950 res = mqtt_parse_incoming(client, p); 951 pbuf_free(p); 952 953 if (res != MQTT_CONNECT_ACCEPTED) { 954 mqtt_close(client, res); 955 } 956 /* If keep alive functionality is used */ 957 if (client->keep_alive != 0) { 958 /* Reset server alive watchdog */ 959 client->server_watchdog = 0; 960 } 961 962 } 963 return ERR_OK; 964} 965 966 967/** 968 * TCP data sent callback function. @see tcp_sent_fn 969 * @param arg MQTT client 970 * @param tpcb TCP connection handle 971 * @param len Number of bytes sent 972 * @return ERR_OK 973 */ 974static err_t 975mqtt_tcp_sent_cb(void *arg, struct altcp_pcb *tpcb, u16_t len) 976{ 977 mqtt_client_t *client = (mqtt_client_t *)arg; 978 979 LWIP_UNUSED_ARG(tpcb); 980 LWIP_UNUSED_ARG(len); 981 982 if (client->conn_state == MQTT_CONNECTED) { 983 struct mqtt_request_t *r; 984 985 /* Reset keep-alive send timer and server watchdog */ 986 client->cyclic_tick = 0; 987 client->server_watchdog = 0; 988 /* QoS 0 publish has no response from server, so call its callbacks here */ 989 while ((r = mqtt_take_request(&client->pend_req_queue, 0)) != NULL) { 990 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_tcp_sent_cb: Calling QoS 0 publish complete callback\n")); 991 if (r->cb != NULL) { 992 r->cb(r->arg, ERR_OK); 993 } 994 mqtt_delete_request(r); 995 } 996 /* Try send any remaining buffers from output queue */ 997 mqtt_output_send(&client->output, client->conn); 998 } 999 return ERR_OK; 1000} 1001 1002/** 1003 * TCP error callback function. @see tcp_err_fn 1004 * @param arg MQTT client 1005 * @param err Error encountered 1006 */ 1007static void 1008mqtt_tcp_err_cb(void *arg, err_t err) 1009{ 1010 mqtt_client_t *client = (mqtt_client_t *)arg; 1011 LWIP_UNUSED_ARG(err); /* only used for debug output */ 1012 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_tcp_err_cb: TCP error callback: error %d, arg: %p\n", err, arg)); 1013 LWIP_ASSERT("mqtt_tcp_err_cb: client != NULL", client != NULL); 1014 /* Set conn to null before calling close as pcb is already deallocated*/ 1015 client->conn = 0; 1016 mqtt_close(client, MQTT_CONNECT_DISCONNECTED); 1017} 1018 1019/** 1020 * TCP poll callback function. @see tcp_poll_fn 1021 * @param arg MQTT client 1022 * @param tpcb TCP connection handle 1023 * @return err ERR_OK 1024 */ 1025static err_t 1026mqtt_tcp_poll_cb(void *arg, struct altcp_pcb *tpcb) 1027{ 1028 mqtt_client_t *client = (mqtt_client_t *)arg; 1029 if (client->conn_state == MQTT_CONNECTED) { 1030 /* Try send any remaining buffers from output queue */ 1031 mqtt_output_send(&client->output, tpcb); 1032 } 1033 return ERR_OK; 1034} 1035 1036/** 1037 * TCP connect callback function. @see tcp_connected_fn 1038 * @param arg MQTT client 1039 * @param err Always ERR_OK, mqtt_tcp_err_cb is called in case of error 1040 * @return ERR_OK 1041 */ 1042static err_t 1043mqtt_tcp_connect_cb(void *arg, struct altcp_pcb *tpcb, err_t err) 1044{ 1045 mqtt_client_t *client = (mqtt_client_t *)arg; 1046 1047 if (err != ERR_OK) { 1048 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_tcp_connect_cb: TCP connect error %d\n", err)); 1049 return err; 1050 } 1051 1052 /* Initiate receiver state */ 1053 client->msg_idx = 0; 1054 1055 /* Setup TCP callbacks */ 1056 altcp_recv(tpcb, mqtt_tcp_recv_cb); 1057 altcp_sent(tpcb, mqtt_tcp_sent_cb); 1058 altcp_poll(tpcb, mqtt_tcp_poll_cb, 2); 1059 1060 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_tcp_connect_cb: TCP connection established to server\n")); 1061 /* Enter MQTT connect state */ 1062 client->conn_state = MQTT_CONNECTING; 1063 1064 /* Start cyclic timer */ 1065 sys_timeout(MQTT_CYCLIC_TIMER_INTERVAL * 1000, mqtt_cyclic_timer, client); 1066 client->cyclic_tick = 0; 1067 1068 /* Start transmission from output queue, connect message is the first one out*/ 1069 mqtt_output_send(&client->output, client->conn); 1070 1071 return ERR_OK; 1072} 1073 1074 1075 1076/*---------------------------------------------------------------------------------------------------- */ 1077/* Public API */ 1078 1079 1080/** 1081 * @ingroup mqtt 1082 * MQTT publish function. 1083 * @param client MQTT client 1084 * @param topic Publish topic string 1085 * @param payload Data to publish (NULL is allowed) 1086 * @param payload_length Length of payload (0 is allowed) 1087 * @param qos Quality of service, 0 1 or 2 1088 * @param retain MQTT retain flag 1089 * @param cb Callback to call when publish is complete or has timed out 1090 * @param arg User supplied argument to publish callback 1091 * @return ERR_OK if successful 1092 * ERR_CONN if client is disconnected 1093 * ERR_MEM if short on memory 1094 */ 1095err_t 1096mqtt_publish(mqtt_client_t *client, const char *topic, const void *payload, u16_t payload_length, u8_t qos, u8_t retain, 1097 mqtt_request_cb_t cb, void *arg) 1098{ 1099 struct mqtt_request_t *r; 1100 u16_t pkt_id; 1101 size_t topic_strlen; 1102 size_t total_len; 1103 u16_t topic_len; 1104 u16_t remaining_length; 1105 1106 LWIP_ASSERT_CORE_LOCKED(); 1107 LWIP_ASSERT("mqtt_publish: client != NULL", client); 1108 LWIP_ASSERT("mqtt_publish: topic != NULL", topic); 1109 LWIP_ERROR("mqtt_publish: TCP disconnected", (client->conn_state != TCP_DISCONNECTED), return ERR_CONN); 1110 1111 topic_strlen = strlen(topic); 1112 LWIP_ERROR("mqtt_publish: topic length overflow", (topic_strlen <= (0xFFFF - 2)), return ERR_ARG); 1113 topic_len = (u16_t)topic_strlen; 1114 total_len = 2 + topic_len + payload_length; 1115 1116 if (qos > 0) { 1117 total_len += 2; 1118 /* Generate pkt_id id for QoS1 and 2 */ 1119 pkt_id = msg_generate_packet_id(client); 1120 } else { 1121 /* Use reserved value pkt_id 0 for QoS 0 in request handle */ 1122 pkt_id = 0; 1123 } 1124 LWIP_ERROR("mqtt_publish: total length overflow", (total_len <= 0xFFFF), return ERR_ARG); 1125 remaining_length = (u16_t)total_len; 1126 1127 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_publish: Publish with payload length %d to topic \"%s\"\n", payload_length, topic)); 1128 1129 r = mqtt_create_request(client->req_list, LWIP_ARRAYSIZE(client->req_list), pkt_id, cb, arg); 1130 if (r == NULL) { 1131 return ERR_MEM; 1132 } 1133 1134 if (mqtt_output_check_space(&client->output, remaining_length) == 0) { 1135 mqtt_delete_request(r); 1136 return ERR_MEM; 1137 } 1138 /* Append fixed header */ 1139 mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain, remaining_length); 1140 1141 /* Append Topic */ 1142 mqtt_output_append_string(&client->output, topic, topic_len); 1143 1144 /* Append packet if for QoS 1 and 2*/ 1145 if (qos > 0) { 1146 mqtt_output_append_u16(&client->output, pkt_id); 1147 } 1148 1149 /* Append optional publish payload */ 1150 if ((payload != NULL) && (payload_length > 0)) { 1151 mqtt_output_append_buf(&client->output, payload, payload_length); 1152 } 1153 1154 mqtt_append_request(&client->pend_req_queue, r); 1155 mqtt_output_send(&client->output, client->conn); 1156 return ERR_OK; 1157} 1158 1159 1160/** 1161 * @ingroup mqtt 1162 * MQTT subscribe/unsubscribe function. 1163 * @param client MQTT client 1164 * @param topic topic to subscribe to 1165 * @param qos Quality of service, 0 1 or 2 (only used for subscribe) 1166 * @param cb Callback to call when subscribe/unsubscribe reponse is received 1167 * @param arg User supplied argument to publish callback 1168 * @param sub 1 for subscribe, 0 for unsubscribe 1169 * @return ERR_OK if successful, @see err_t enum for other results 1170 */ 1171err_t 1172mqtt_sub_unsub(mqtt_client_t *client, const char *topic, u8_t qos, mqtt_request_cb_t cb, void *arg, u8_t sub) 1173{ 1174 size_t topic_strlen; 1175 size_t total_len; 1176 u16_t topic_len; 1177 u16_t remaining_length; 1178 u16_t pkt_id; 1179 struct mqtt_request_t *r; 1180 1181 LWIP_ASSERT_CORE_LOCKED(); 1182 LWIP_ASSERT("mqtt_sub_unsub: client != NULL", client); 1183 LWIP_ASSERT("mqtt_sub_unsub: topic != NULL", topic); 1184 1185 topic_strlen = strlen(topic); 1186 LWIP_ERROR("mqtt_sub_unsub: topic length overflow", (topic_strlen <= (0xFFFF - 2)), return ERR_ARG); 1187 topic_len = (u16_t)topic_strlen; 1188 /* Topic string, pkt_id, qos for subscribe */ 1189 total_len = topic_len + 2 + 2 + (sub != 0); 1190 LWIP_ERROR("mqtt_sub_unsub: total length overflow", (total_len <= 0xFFFF), return ERR_ARG); 1191 remaining_length = (u16_t)total_len; 1192 1193 LWIP_ASSERT("mqtt_sub_unsub: qos < 3", qos < 3); 1194 if (client->conn_state == TCP_DISCONNECTED) { 1195 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_sub_unsub: Can not (un)subscribe in disconnected state\n")); 1196 return ERR_CONN; 1197 } 1198 1199 pkt_id = msg_generate_packet_id(client); 1200 r = mqtt_create_request(client->req_list, LWIP_ARRAYSIZE(client->req_list), pkt_id, cb, arg); 1201 if (r == NULL) { 1202 return ERR_MEM; 1203 } 1204 1205 if (mqtt_output_check_space(&client->output, remaining_length) == 0) { 1206 mqtt_delete_request(r); 1207 return ERR_MEM; 1208 } 1209 1210 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_sub_unsub: Client (un)subscribe to topic \"%s\", id: %d\n", topic, pkt_id)); 1211 1212 mqtt_output_append_fixed_header(&client->output, sub ? MQTT_MSG_TYPE_SUBSCRIBE : MQTT_MSG_TYPE_UNSUBSCRIBE, 0, 1, 0, remaining_length); 1213 /* Packet id */ 1214 mqtt_output_append_u16(&client->output, pkt_id); 1215 /* Topic */ 1216 mqtt_output_append_string(&client->output, topic, topic_len); 1217 /* QoS */ 1218 if (sub != 0) { 1219 mqtt_output_append_u8(&client->output, LWIP_MIN(qos, 2)); 1220 } 1221 1222 mqtt_append_request(&client->pend_req_queue, r); 1223 mqtt_output_send(&client->output, client->conn); 1224 return ERR_OK; 1225} 1226 1227 1228/** 1229 * @ingroup mqtt 1230 * Set callback to handle incoming publish requests from server 1231 * @param client MQTT client 1232 * @param pub_cb Callback invoked when publish starts, contain topic and total length of payload 1233 * @param data_cb Callback for each fragment of payload that arrives 1234 * @param arg User supplied argument to both callbacks 1235 */ 1236void 1237mqtt_set_inpub_callback(mqtt_client_t *client, mqtt_incoming_publish_cb_t pub_cb, 1238 mqtt_incoming_data_cb_t data_cb, void *arg) 1239{ 1240 LWIP_ASSERT_CORE_LOCKED(); 1241 LWIP_ASSERT("mqtt_set_inpub_callback: client != NULL", client != NULL); 1242 client->data_cb = data_cb; 1243 client->pub_cb = pub_cb; 1244 client->inpub_arg = arg; 1245} 1246 1247/** 1248 * @ingroup mqtt 1249 * Create a new MQTT client instance 1250 * @return Pointer to instance on success, NULL otherwise 1251 */ 1252mqtt_client_t * 1253mqtt_client_new(void) 1254{ 1255 LWIP_ASSERT_CORE_LOCKED(); 1256 return (mqtt_client_t *)mem_calloc(1, sizeof(mqtt_client_t)); 1257} 1258 1259/** 1260 * @ingroup mqtt 1261 * Free MQTT client instance 1262 * @param client Pointer to instance to be freed 1263 */ 1264void 1265mqtt_client_free(mqtt_client_t *client) 1266{ 1267 mem_free(client); 1268} 1269 1270/** 1271 * @ingroup mqtt 1272 * Connect to MQTT server 1273 * @param client MQTT client 1274 * @param ip_addr Server IP 1275 * @param port Server port 1276 * @param cb Connection state change callback 1277 * @param arg User supplied argument to connection callback 1278 * @param client_info Client identification and connection options 1279 * @return ERR_OK if successful, @see err_t enum for other results 1280 */ 1281err_t 1282mqtt_client_connect(mqtt_client_t *client, const ip_addr_t *ip_addr, u16_t port, mqtt_connection_cb_t cb, void *arg, 1283 const struct mqtt_connect_client_info_t *client_info) 1284{ 1285 err_t err; 1286 size_t len; 1287 u16_t client_id_length; 1288 /* Length is the sum of 2+"MQTT", protocol level, flags and keep alive */ 1289 u16_t remaining_length = 2 + 4 + 1 + 1 + 2; 1290 u8_t flags = 0, will_topic_len = 0, will_msg_len = 0; 1291 u16_t client_user_len = 0, client_pass_len = 0; 1292 1293 LWIP_ASSERT_CORE_LOCKED(); 1294 LWIP_ASSERT("mqtt_client_connect: client != NULL", client != NULL); 1295 LWIP_ASSERT("mqtt_client_connect: ip_addr != NULL", ip_addr != NULL); 1296 LWIP_ASSERT("mqtt_client_connect: client_info != NULL", client_info != NULL); 1297 LWIP_ASSERT("mqtt_client_connect: client_info->client_id != NULL", client_info->client_id != NULL); 1298 1299 if (client->conn_state != TCP_DISCONNECTED) { 1300 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_client_connect: Already connected\n")); 1301 return ERR_ISCONN; 1302 } 1303 1304 /* Wipe clean */ 1305 memset(client, 0, sizeof(mqtt_client_t)); 1306 client->connect_arg = arg; 1307 client->connect_cb = cb; 1308 client->keep_alive = client_info->keep_alive; 1309 mqtt_init_requests(client->req_list, LWIP_ARRAYSIZE(client->req_list)); 1310 1311 /* Build connect message */ 1312 if (client_info->will_topic != NULL && client_info->will_msg != NULL) { 1313 flags |= MQTT_CONNECT_FLAG_WILL; 1314 flags |= (client_info->will_qos & 3) << 3; 1315 if (client_info->will_retain) { 1316 flags |= MQTT_CONNECT_FLAG_WILL_RETAIN; 1317 } 1318 len = strlen(client_info->will_topic); 1319 LWIP_ERROR("mqtt_client_connect: client_info->will_topic length overflow", len <= 0xFF, return ERR_VAL); 1320 LWIP_ERROR("mqtt_client_connect: client_info->will_topic length must be > 0", len > 0, return ERR_VAL); 1321 will_topic_len = (u8_t)len; 1322 len = strlen(client_info->will_msg); 1323 LWIP_ERROR("mqtt_client_connect: client_info->will_msg length overflow", len <= 0xFF, return ERR_VAL); 1324 will_msg_len = (u8_t)len; 1325 len = remaining_length + 2 + will_topic_len + 2 + will_msg_len; 1326 LWIP_ERROR("mqtt_client_connect: remaining_length overflow", len <= 0xFFFF, return ERR_VAL); 1327 remaining_length = (u16_t)len; 1328 } 1329 if (client_info->client_user != NULL) { 1330 flags |= MQTT_CONNECT_FLAG_USERNAME; 1331 len = strlen(client_info->client_user); 1332 LWIP_ERROR("mqtt_client_connect: client_info->client_user length overflow", len <= 0xFFFF, return ERR_VAL); 1333 LWIP_ERROR("mqtt_client_connect: client_info->client_user length must be > 0", len > 0, return ERR_VAL); 1334 client_user_len = (u16_t)len; 1335 len = remaining_length + 2 + client_user_len; 1336 LWIP_ERROR("mqtt_client_connect: remaining_length overflow", len <= 0xFFFF, return ERR_VAL); 1337 remaining_length = (u16_t)len; 1338 } 1339 if (client_info->client_pass != NULL) { 1340 flags |= MQTT_CONNECT_FLAG_PASSWORD; 1341 len = strlen(client_info->client_pass); 1342 LWIP_ERROR("mqtt_client_connect: client_info->client_pass length overflow", len <= 0xFFFF, return ERR_VAL); 1343 LWIP_ERROR("mqtt_client_connect: client_info->client_pass length must be > 0", len > 0, return ERR_VAL); 1344 client_pass_len = (u16_t)len; 1345 len = remaining_length + 2 + client_pass_len; 1346 LWIP_ERROR("mqtt_client_connect: remaining_length overflow", len <= 0xFFFF, return ERR_VAL); 1347 remaining_length = (u16_t)len; 1348 } 1349 1350 /* Don't complicate things, always connect using clean session */ 1351 flags |= MQTT_CONNECT_FLAG_CLEAN_SESSION; 1352 1353 len = strlen(client_info->client_id); 1354 LWIP_ERROR("mqtt_client_connect: client_info->client_id length overflow", len <= 0xFFFF, return ERR_VAL); 1355 client_id_length = (u16_t)len; 1356 len = remaining_length + 2 + client_id_length; 1357 LWIP_ERROR("mqtt_client_connect: remaining_length overflow", len <= 0xFFFF, return ERR_VAL); 1358 remaining_length = (u16_t)len; 1359 1360 if (mqtt_output_check_space(&client->output, remaining_length) == 0) { 1361 return ERR_MEM; 1362 } 1363 1364#if LWIP_ALTCP && LWIP_ALTCP_TLS 1365 if (client_info->tls_config) { 1366 client->conn = altcp_tls_new(client_info->tls_config, IP_GET_TYPE(ip_addr)); 1367 } else 1368#endif 1369 { 1370 client->conn = altcp_tcp_new_ip_type(IP_GET_TYPE(ip_addr)); 1371 } 1372 if (client->conn == NULL) { 1373 return ERR_MEM; 1374 } 1375 1376 /* Set arg pointer for callbacks */ 1377 altcp_arg(client->conn, client); 1378 /* Any local address, pick random local port number */ 1379 err = altcp_bind(client->conn, IP_ADDR_ANY, 0); 1380 if (err != ERR_OK) { 1381 LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_client_connect: Error binding to local ip/port, %d\n", err)); 1382 goto tcp_fail; 1383 } 1384 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_client_connect: Connecting to host: %s at port:%"U16_F"\n", ipaddr_ntoa(ip_addr), port)); 1385 1386 /* Connect to server */ 1387 err = altcp_connect(client->conn, ip_addr, port, mqtt_tcp_connect_cb); 1388 if (err != ERR_OK) { 1389 LWIP_DEBUGF(MQTT_DEBUG_TRACE, ("mqtt_client_connect: Error connecting to remote ip/port, %d\n", err)); 1390 goto tcp_fail; 1391 } 1392 /* Set error callback */ 1393 altcp_err(client->conn, mqtt_tcp_err_cb); 1394 client->conn_state = TCP_CONNECTING; 1395 1396 /* Append fixed header */ 1397 mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_CONNECT, 0, 0, 0, remaining_length); 1398 /* Append Protocol string */ 1399 mqtt_output_append_string(&client->output, "MQTT", 4); 1400 /* Append Protocol level */ 1401 mqtt_output_append_u8(&client->output, 4); 1402 /* Append connect flags */ 1403 mqtt_output_append_u8(&client->output, flags); 1404 /* Append keep-alive */ 1405 mqtt_output_append_u16(&client->output, client_info->keep_alive); 1406 /* Append client id */ 1407 mqtt_output_append_string(&client->output, client_info->client_id, client_id_length); 1408 /* Append will message if used */ 1409 if ((flags & MQTT_CONNECT_FLAG_WILL) != 0) { 1410 mqtt_output_append_string(&client->output, client_info->will_topic, will_topic_len); 1411 mqtt_output_append_string(&client->output, client_info->will_msg, will_msg_len); 1412 } 1413 /* Append user name if given */ 1414 if ((flags & MQTT_CONNECT_FLAG_USERNAME) != 0) { 1415 mqtt_output_append_string(&client->output, client_info->client_user, client_user_len); 1416 } 1417 /* Append password if given */ 1418 if ((flags & MQTT_CONNECT_FLAG_PASSWORD) != 0) { 1419 mqtt_output_append_string(&client->output, client_info->client_pass, client_pass_len); 1420 } 1421 return ERR_OK; 1422 1423tcp_fail: 1424 altcp_abort(client->conn); 1425 client->conn = NULL; 1426 return err; 1427} 1428 1429 1430/** 1431 * @ingroup mqtt 1432 * Disconnect from MQTT server 1433 * @param client MQTT client 1434 */ 1435void 1436mqtt_disconnect(mqtt_client_t *client) 1437{ 1438 LWIP_ASSERT_CORE_LOCKED(); 1439 LWIP_ASSERT("mqtt_disconnect: client != NULL", client); 1440 /* If connection in not already closed */ 1441 if (client->conn_state != TCP_DISCONNECTED) { 1442 /* Set conn_state before calling mqtt_close to prevent callback from being called */ 1443 client->conn_state = TCP_DISCONNECTED; 1444 mqtt_close(client, (mqtt_connection_status_t)0); 1445 } 1446} 1447 1448/** 1449 * @ingroup mqtt 1450 * Check connection with server 1451 * @param client MQTT client 1452 * @return 1 if connected to server, 0 otherwise 1453 */ 1454u8_t 1455mqtt_client_is_connected(mqtt_client_t *client) 1456{ 1457 LWIP_ASSERT_CORE_LOCKED(); 1458 LWIP_ASSERT("mqtt_client_is_connected: client != NULL", client); 1459 return client->conn_state == MQTT_CONNECTED; 1460} 1461 1462#endif /* LWIP_TCP && LWIP_CALLBACK_API */ 1463