lwip_mqtt.c

00001 /**
00002  * @file
00003  * MQTT client
00004  *
00005  * @defgroup mqtt MQTT client
00006  * @ingroup apps
00007  * @verbinclude mqtt_client.txt
00008  */
00009 
00010 /*
00011  * Copyright (c) 2016 Erik Andersson <erian747@gmail.com>
00012  * All rights reserved.
00013  *
00014  * Redistribution and use in source and binary forms, with or without modification,
00015  * are permitted provided that the following conditions are met:
00016  *
00017  * 1. Redistributions of source code must retain the above copyright notice,
00018  *    this list of conditions and the following disclaimer.
00019  * 2. Redistributions in binary form must reproduce the above copyright notice,
00020  *    this list of conditions and the following disclaimer in the documentation
00021  *    and/or other materials provided with the distribution.
00022  * 3. The name of the author may not be used to endorse or promote products
00023  *    derived from this software without specific prior written permission.
00024  *
00025  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
00026  * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
00027  * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
00028  * SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
00029  * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
00030  * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
00031  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
00032  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING
00033  * IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
00034  * OF SUCH DAMAGE.
00035  *
00036  * This file is part of the lwIP TCP/IP stack
00037  *
00038  * Author: Erik Andersson <erian747@gmail.com>
00039  *
00040  *
00041  * @todo:
00042  * - Handle large outgoing payloads for PUBLISH messages
00043  * - Fix restriction of a single topic in each (UN)SUBSCRIBE message (protocol has support for multiple topics)
00044  * - Add support for legacy MQTT protocol version
00045  *
00046  * Please coordinate changes and requests with Erik Andersson
00047  * Erik Andersson <erian747@gmail.com>
00048  *
00049  */
00050 #include "lwip/apps/mqtt.h"
00051 #include "lwip/timeouts.h"
00052 #include "lwip/ip_addr.h"
00053 #include "lwip/mem.h"
00054 #include "lwip/err.h"
00055 #include "lwip/pbuf.h"
00056 #include "lwip/tcp.h"
00057 #include <string.h>
00058 
00059 #if LWIP_TCP && LWIP_CALLBACK_API
00060 
00061 /**
00062  * MQTT_DEBUG: Default is off.
00063  */
00064 #if !defined MQTT_DEBUG || defined __DOXYGEN__
00065 #define MQTT_DEBUG                  LWIP_DBG_OFF
00066 #endif
00067 
00068 #define MQTT_DEBUG_TRACE        (MQTT_DEBUG | LWIP_DBG_TRACE)
00069 #define MQTT_DEBUG_STATE        (MQTT_DEBUG | LWIP_DBG_STATE)
00070 #define MQTT_DEBUG_WARN         (MQTT_DEBUG | LWIP_DBG_LEVEL_WARNING)
00071 #define MQTT_DEBUG_WARN_STATE   (MQTT_DEBUG | LWIP_DBG_LEVEL_WARNING | LWIP_DBG_STATE)
00072 #define MQTT_DEBUG_SERIOUS      (MQTT_DEBUG | LWIP_DBG_LEVEL_SERIOUS)
00073 
00074 static void mqtt_cyclic_timer(void *arg);
00075 
00076 /**
00077  * MQTT client connection states
00078  */
00079 enum {
00080   TCP_DISCONNECTED,
00081   TCP_CONNECTING,
00082   MQTT_CONNECTING,
00083   MQTT_CONNECTED
00084 };
00085 
00086 /**
00087  * MQTT control message types
00088  */
00089 enum mqtt_message_type {
00090   MQTT_MSG_TYPE_CONNECT = 1,
00091   MQTT_MSG_TYPE_CONNACK = 2,
00092   MQTT_MSG_TYPE_PUBLISH = 3,
00093   MQTT_MSG_TYPE_PUBACK = 4,
00094   MQTT_MSG_TYPE_PUBREC = 5,
00095   MQTT_MSG_TYPE_PUBREL = 6,
00096   MQTT_MSG_TYPE_PUBCOMP = 7,
00097   MQTT_MSG_TYPE_SUBSCRIBE = 8,
00098   MQTT_MSG_TYPE_SUBACK = 9,
00099   MQTT_MSG_TYPE_UNSUBSCRIBE = 10,
00100   MQTT_MSG_TYPE_UNSUBACK = 11,
00101   MQTT_MSG_TYPE_PINGREQ = 12,
00102   MQTT_MSG_TYPE_PINGRESP = 13,
00103   MQTT_MSG_TYPE_DISCONNECT = 14
00104 };
00105 
00106 /** Helpers to extract control packet type and qos from first byte in fixed header */
00107 #define MQTT_CTL_PACKET_TYPE(fixed_hdr_byte0) ((fixed_hdr_byte0 & 0xf0) >> 4)
00108 #define MQTT_CTL_PACKET_QOS(fixed_hdr_byte0) ((fixed_hdr_byte0 & 0x6) >> 1)
00109 
00110 /**
00111  * MQTT connect flags, only used in CONNECT message
00112  */
00113 enum mqtt_connect_flag {
00114   MQTT_CONNECT_FLAG_USERNAME = 1 << 7,
00115   MQTT_CONNECT_FLAG_PASSWORD = 1 << 6,
00116   MQTT_CONNECT_FLAG_WILL_RETAIN = 1 << 5,
00117   MQTT_CONNECT_FLAG_WILL = 1 << 2,
00118   MQTT_CONNECT_FLAG_CLEAN_SESSION = 1 << 1
00119 };
00120 
00121 
00122 #if defined(LWIP_DEBUG)
00123 static const char * const mqtt_message_type_str[15] =
00124 {
00125   "UNDEFINED",
00126   "CONNECT",
00127   "CONNACK",
00128   "PUBLISH",
00129   "PUBACK",
00130   "PUBREC",
00131   "PUBREL",
00132   "PUBCOMP",
00133   "SUBSCRIBE",
00134   "SUBACK",
00135   "UNSUBSCRIBE",
00136   "UNSUBACK",
00137   "PINGREQ",
00138   "PINGRESP",
00139   "DISCONNECT"
00140 };
00141 
00142 /**
00143  * Message type value to string
00144  * @param msg_type see enum mqtt_message_type
00145  * 
00146  * @return Control message type text string
00147  */
00148 static const char *
00149 mqtt_msg_type_to_str(u8_t msg_type)
00150 {
00151   if (msg_type >= LWIP_ARRAYSIZE(mqtt_message_type_str)) {
00152     msg_type = 0;
00153   }
00154   return mqtt_message_type_str[msg_type];
00155 }
00156 
00157 #endif
00158 
00159 
00160 /**
00161  * Generate MQTT packet identifier
00162  * @param client MQTT client
00163  * @return New packet identifier, range 1 to 65535
00164  */
00165 static u16_t
00166 msg_generate_packet_id(mqtt_client_t *client)
00167 {
00168   client->pkt_id_seq++;
00169   if (client->pkt_id_seq == 0) {
00170     client->pkt_id_seq++;
00171   }
00172   return client->pkt_id_seq;
00173 }
00174 
00175 /*--------------------------------------------------------------------------------------------------------------------- */
00176 /* Output ring buffer */
00177 
00178 
00179 #define MQTT_RINGBUF_IDX_MASK ((MQTT_OUTPUT_RINGBUF_SIZE) - 1)
00180 
00181 /** Add single item to ring buffer */
00182 #define mqtt_ringbuf_put(rb, item) ((rb)->buf)[(rb)->put++ & MQTT_RINGBUF_IDX_MASK] = (item)
00183 
00184 /** Return number of bytes in ring buffer */
00185 #define mqtt_ringbuf_len(rb) ((u16_t)((rb)->put - (rb)->get))
00186 
00187 /** Return number of bytes free in ring buffer */
00188 #define mqtt_ringbuf_free(rb) (MQTT_OUTPUT_RINGBUF_SIZE - mqtt_ringbuf_len(rb))
00189 
00190 /** Return number of bytes possible to read without wrapping around */
00191 #define mqtt_ringbuf_linear_read_length(rb) LWIP_MIN(mqtt_ringbuf_len(rb), (MQTT_OUTPUT_RINGBUF_SIZE - ((rb)->get & MQTT_RINGBUF_IDX_MASK)))
00192 
00193 /** Return pointer to ring buffer get position */
00194 #define mqtt_ringbuf_get_ptr(rb) (&(rb)->buf[(rb)->get & MQTT_RINGBUF_IDX_MASK])
00195 
00196 #define mqtt_ringbuf_advance_get_idx(rb, len) ((rb)->get += (len))
00197 
00198 
00199 /**
00200  * Try send as many bytes as possible from output ring buffer
00201  * @param rb Output ring buffer
00202  * @param tpcb TCP connection handle
00203  */
00204 static void
00205 mqtt_output_send(struct mqtt_ringbuf_t *rb, struct tcp_pcb *tpcb)
00206 {
00207   err_t err;
00208   u8_t wrap = 0;
00209   u16_t ringbuf_lin_len = mqtt_ringbuf_linear_read_length(rb);
00210   u16_t send_len = tcp_sndbuf(tpcb);
00211   LWIP_ASSERT("mqtt_output_send: tpcb != NULL", tpcb != NULL);
00212 
00213   if (send_len == 0 || ringbuf_lin_len == 0) {
00214     return;
00215   }
00216 
00217   LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_output_send: tcp_sndbuf: %d bytes, ringbuf_linear_available: %d, get %d, put %d\n",
00218                                 send_len, ringbuf_lin_len, ((rb)->get & MQTT_RINGBUF_IDX_MASK), ((rb)->put & MQTT_RINGBUF_IDX_MASK)));
00219 
00220   if (send_len > ringbuf_lin_len) {
00221     /* Space in TCP output buffer is larger than available in ring buffer linear portion */
00222     send_len = ringbuf_lin_len;
00223     /* Wrap around if more data in ring buffer after linear portion */
00224     wrap = (mqtt_ringbuf_len(rb) > ringbuf_lin_len);
00225   }
00226   err = tcp_write(tpcb, mqtt_ringbuf_get_ptr(rb), send_len, TCP_WRITE_FLAG_COPY | (wrap ? TCP_WRITE_FLAG_MORE : 0));
00227   if ((err == ERR_OK) && wrap) {
00228     mqtt_ringbuf_advance_get_idx(rb, send_len);
00229     /* Use the lesser one of ring buffer linear length and TCP send buffer size */
00230     send_len = LWIP_MIN(tcp_sndbuf(tpcb), mqtt_ringbuf_linear_read_length(rb));
00231     err = tcp_write(tpcb, mqtt_ringbuf_get_ptr(rb), send_len, TCP_WRITE_FLAG_COPY);
00232   }
00233 
00234   if (err == ERR_OK) {
00235     mqtt_ringbuf_advance_get_idx(rb, send_len);
00236     /* Flush */
00237     tcp_output(tpcb);
00238   } else {
00239     LWIP_DEBUGF(MQTT_DEBUG_WARN, ("mqtt_output_send: Send failed with err %d (\"%s\")\n", err, lwip_strerr(err)));
00240   }
00241 }
00242 
00243 
00244 
00245 /*--------------------------------------------------------------------------------------------------------------------- */
00246 /* Request queue */
00247 
00248 /**
00249  * Create request item
00250  * @param r_objs Pointer to request objects
00251  * @param pkt_id Packet identifier of request
00252  * @param cb Packet callback to call when requests lifetime ends
00253  * @param arg Parameter following callback
00254  * @return Request or NULL if failed to create
00255  */
00256 static struct mqtt_request_t *
00257 mqtt_create_request(struct mqtt_request_t *r_objs, u16_t pkt_id, mqtt_request_cb_t cb, void *arg)
00258 {
00259   struct mqtt_request_t *r = NULL;
00260   u8_t n;
00261   LWIP_ASSERT("mqtt_create_request: r_objs != NULL", r_objs != NULL);
00262   for (n = 0; n < MQTT_REQ_MAX_IN_FLIGHT; n++) {
00263     /* Item point to itself if not in use */
00264     if (r_objs[n].next == &r_objs[n]) {
00265       r = &r_objs[n];
00266       r->next = NULL;
00267       r->cb = cb;
00268       r->arg = arg;
00269       r->pkt_id = pkt_id;
00270       break;
00271     }
00272   }
00273   return r;
00274 }
00275 
00276 
00277 /**
00278  * Append request to pending request queue
00279  * @param tail Pointer to request queue tail pointer
00280  * @param r Request to append
00281  */
00282 static void
00283 mqtt_append_request(struct mqtt_request_t **tail, struct mqtt_request_t *r)
00284 {
00285   struct mqtt_request_t *head = NULL;
00286   s16_t time_before = 0;
00287   struct mqtt_request_t *iter;
00288 
00289   LWIP_ASSERT("mqtt_append_request: tail != NULL", tail != NULL);
00290 
00291   /* Iterate trough queue to find head, and count total timeout time */
00292   for (iter = *tail; iter != NULL; iter = iter->next) {
00293     time_before += iter->timeout_diff;
00294     head = iter;
00295   }
00296 
00297   LWIP_ASSERT("mqtt_append_request: time_before <= MQTT_REQ_TIMEOUT", time_before <= MQTT_REQ_TIMEOUT);
00298   r->timeout_diff = MQTT_REQ_TIMEOUT - time_before;
00299   if (head == NULL) {
00300     *tail = r;
00301   } else {
00302     head->next = r;
00303   }
00304 }
00305 
00306 
00307 /**
00308  * Delete request item
00309  * @param r Request item to delete
00310  */
00311 static void
00312 mqtt_delete_request(struct mqtt_request_t *r)
00313 {
00314   if (r != NULL) {
00315     r->next = r;
00316   }
00317 }
00318 
00319 /**
00320  * Remove a request item with a specific packet identifier from request queue
00321  * @param tail Pointer to request queue tail pointer
00322  * @param pkt_id Packet identifier of request to take
00323  * @return Request item if found, NULL if not
00324  */
00325 static struct mqtt_request_t *
00326 mqtt_take_request(struct mqtt_request_t **tail, u16_t pkt_id)
00327 {
00328   struct mqtt_request_t *iter = NULL, *prev = NULL;
00329   LWIP_ASSERT("mqtt_take_request: tail != NULL", tail != NULL);
00330   /* Search all request for pkt_id */
00331   for (iter = *tail; iter != NULL; iter = iter->next) {
00332     if (iter->pkt_id == pkt_id) {
00333       break;
00334     }
00335     prev = iter;
00336   }
00337 
00338   /* If request was found */
00339   if (iter != NULL) {
00340     /* unchain */
00341     if (prev == NULL) {
00342       *tail= iter->next;
00343     } else {
00344       prev->next = iter->next;
00345     }
00346     /* If exists, add remaining timeout time for the request to next */
00347     if (iter->next != NULL) {
00348       iter->next->timeout_diff += iter->timeout_diff;
00349     }
00350     iter->next = NULL;
00351   }
00352   return iter;
00353 }
00354 
00355 /**
00356  * Handle requests timeout
00357  * @param tail Pointer to request queue tail pointer
00358  * @param t Time since last call in seconds
00359  */
00360 static void
00361 mqtt_request_time_elapsed(struct mqtt_request_t **tail, u8_t t)
00362 {
00363   struct mqtt_request_t *r = *tail;
00364   LWIP_ASSERT("mqtt_request_time_elapsed: tail != NULL", tail != NULL);
00365   while (t > 0 && r != NULL) {
00366     if (t >= r->timeout_diff) {
00367       t -= (u8_t)r->timeout_diff;
00368       /* Unchain */
00369       *tail = r->next;
00370       /* Notify upper layer about timeout */
00371       if (r->cb != NULL) {
00372         r->cb(r->arg, ERR_TIMEOUT);
00373       }
00374       mqtt_delete_request(r);
00375       /* Tail might be be modified in callback, so re-read it in every iteration */
00376       r = *(struct mqtt_request_t * const volatile *)tail;
00377     } else {
00378       r->timeout_diff -= t;
00379       t = 0;
00380     }
00381   }
00382 }
00383 
00384 /**
00385  * Free all request items
00386  * @param tail Pointer to request queue tail pointer
00387  */
00388 static void
00389 mqtt_clear_requests(struct mqtt_request_t **tail)
00390 {
00391   struct mqtt_request_t *iter, *next;
00392   LWIP_ASSERT("mqtt_clear_requests: tail != NULL", tail != NULL);
00393   for (iter = *tail; iter != NULL; iter = next) {
00394     next = iter->next;
00395     mqtt_delete_request(iter);
00396   }
00397   *tail = NULL;
00398 }
00399 /**
00400  * Initialize all request items
00401  * @param r_objs Pointer to request objects
00402  */
00403 static void
00404 mqtt_init_requests(struct mqtt_request_t *r_objs)
00405 {
00406   u8_t n;
00407   LWIP_ASSERT("mqtt_init_requests: r_objs != NULL", r_objs != NULL);
00408   for (n = 0; n < MQTT_REQ_MAX_IN_FLIGHT; n++) {
00409     /* Item pointing to itself indicates unused */
00410     r_objs[n].next = &r_objs[n];
00411   }
00412 }
00413 
00414 /*--------------------------------------------------------------------------------------------------------------------- */
00415 /* Output message build helpers */
00416 
00417 
00418 static void
00419 mqtt_output_append_u8(struct mqtt_ringbuf_t *rb, u8_t value)
00420 {
00421   mqtt_ringbuf_put(rb, value);
00422 }
00423 
00424 static
00425 void mqtt_output_append_u16(struct mqtt_ringbuf_t *rb, u16_t value)
00426 {
00427   mqtt_ringbuf_put(rb, value >> 8);
00428   mqtt_ringbuf_put(rb, value & 0xff);
00429 }
00430 
00431 static void
00432 mqtt_output_append_buf(struct mqtt_ringbuf_t *rb, const void *data, u16_t length)
00433 {
00434   u16_t n;
00435   for (n = 0; n < length; n++) {
00436     mqtt_ringbuf_put(rb, ((const u8_t *)data)[n]);
00437   }
00438 }
00439 
00440 static void
00441 mqtt_output_append_string(struct mqtt_ringbuf_t *rb, const char *str, u16_t length)
00442 {
00443   u16_t n;
00444   mqtt_ringbuf_put(rb, length >> 8);
00445   mqtt_ringbuf_put(rb, length & 0xff);
00446   for (n = 0; n < length; n++) {
00447     mqtt_ringbuf_put(rb, str[n]);
00448   }
00449 }
00450 
00451 /**
00452  * Append fixed header
00453  * @param rb Output ring buffer
00454  * @param msg_type see enum mqtt_message_type
00455  * @param dup MQTT DUP flag
00456  * @param qos MQTT QoS field
00457  * @param retain MQTT retain flag
00458  * @param r_length Remaining length after fixed header
00459  */
00460 
00461 static void
00462 mqtt_output_append_fixed_header(struct mqtt_ringbuf_t *rb, u8_t msg_type, u8_t dup,
00463                  u8_t qos, u8_t retain, u16_t r_length)
00464 {
00465   /* Start with control byte */
00466   mqtt_output_append_u8(rb, (((msg_type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1)));
00467   /* Encode remaining length field */
00468   do {
00469     mqtt_output_append_u8(rb, (r_length & 0x7f) | (r_length >= 128 ? 0x80 : 0));
00470     r_length >>= 7;
00471   } while (r_length > 0);
00472 }
00473 
00474 
00475 /**
00476  * Check output buffer space
00477  * @param rb Output ring buffer
00478  * @param r_length Remaining length after fixed header
00479  * @return 1 if message will fit, 0 if not enough buffer space
00480  */
00481 static u8_t
00482 mqtt_output_check_space(struct mqtt_ringbuf_t *rb, u16_t r_length)
00483 {
00484   /* Start with length of type byte + remaining length */
00485   u16_t total_len = 1 + r_length;
00486 
00487   LWIP_ASSERT("mqtt_output_check_space: rb != NULL", rb != NULL);
00488 
00489  /* Calculate number of required bytes to contain the remaining bytes field and add to total*/
00490   do {
00491     total_len++;
00492     r_length >>= 7;
00493   } while (r_length > 0);
00494 
00495   return (total_len <= mqtt_ringbuf_free(rb));
00496 }
00497 
00498 
00499 /**
00500  * Close connection to server
00501  * @param client MQTT client
00502  * @param reason Reason for disconnection
00503  */
00504 static void
00505 mqtt_close(mqtt_client_t *client, mqtt_connection_status_t reason)
00506 {
00507   LWIP_ASSERT("mqtt_close: client != NULL", client != NULL);
00508 
00509   /* Bring down TCP connection if not already done */
00510   if (client->conn != NULL) {
00511     err_t res;
00512     tcp_recv(client->conn, NULL);
00513     tcp_err(client->conn,  NULL);
00514     tcp_sent(client->conn, NULL);
00515     res = tcp_close(client->conn);
00516     if (res != ERR_OK) {
00517       tcp_abort(client->conn);
00518       LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_close: Close err=%s\n", lwip_strerr(res)));
00519     }
00520     client->conn = NULL;
00521   }
00522 
00523   /* Remove all pending requests */
00524   mqtt_clear_requests(&client->pend_req_queue);
00525   /* Stop cyclic timer */
00526   sys_untimeout(mqtt_cyclic_timer, client);
00527 
00528   /* Notify upper layer of disconnection if changed state */
00529   if (client->conn_state != TCP_DISCONNECTED) {
00530 
00531     client->conn_state = TCP_DISCONNECTED;
00532     if (client->connect_cb != NULL) {
00533       client->connect_cb(client, client->connect_arg, reason);
00534     }
00535   }
00536 }
00537 
00538 
00539 /**
00540  * Interval timer, called every MQTT_CYCLIC_TIMER_INTERVAL seconds in MQTT_CONNECTING and MQTT_CONNECTED states
00541  * @param arg MQTT client
00542  */
00543 static void
00544 mqtt_cyclic_timer(void *arg)
00545 {
00546   u8_t restart_timer = 1;
00547   mqtt_client_t *client = (mqtt_client_t *)arg;
00548   LWIP_ASSERT("mqtt_cyclic_timer: client != NULL", client != NULL);
00549 
00550   if (client->conn_state == MQTT_CONNECTING) {
00551     client->cyclic_tick++;
00552     if ((client->cyclic_tick * MQTT_CYCLIC_TIMER_INTERVAL) >= MQTT_CONNECT_TIMOUT) {
00553       LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_cyclic_timer: CONNECT attempt to server timed out\n"));
00554       /* Disconnect TCP */
00555       mqtt_close(client, MQTT_CONNECT_TIMEOUT);
00556       restart_timer = 0;
00557     }
00558   } else if (client->conn_state == MQTT_CONNECTED) {
00559     /* Handle timeout for pending requests */
00560     mqtt_request_time_elapsed(&client->pend_req_queue, MQTT_CYCLIC_TIMER_INTERVAL);
00561 
00562     /* keep_alive > 0 means keep alive functionality shall be used */
00563     if (client->keep_alive > 0) {
00564 
00565       client->server_watchdog++;
00566       /* If reception from server has been idle for 1.5*keep_alive time, server is considered unresponsive */
00567       if ((client->server_watchdog * MQTT_CYCLIC_TIMER_INTERVAL) > (client->keep_alive + client->keep_alive/2)) {
00568         LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_cyclic_timer: Server incoming keep-alive timeout\n"));
00569         mqtt_close(client, MQTT_CONNECT_TIMEOUT);
00570         restart_timer = 0;
00571       }
00572 
00573       /* If time for a keep alive message to be sent, transmission has been idle for keep_alive time */
00574       if ((client->cyclic_tick * MQTT_CYCLIC_TIMER_INTERVAL) >= client->keep_alive) {
00575         LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_cyclic_timer: Sending keep-alive message to server\n"));
00576         if (mqtt_output_check_space(&client->output, 0) != 0) {
00577           mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_PINGREQ, 0, 0, 0, 0);
00578           client->cyclic_tick = 0;
00579         }
00580       } else {
00581         client->cyclic_tick++;
00582       }
00583     }
00584   } else {
00585     LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_cyclic_timer: Timer should not be running in state %d\n", client->conn_state));
00586     restart_timer = 0;
00587   }
00588   if (restart_timer) {
00589     sys_timeout(MQTT_CYCLIC_TIMER_INTERVAL*1000, mqtt_cyclic_timer, arg);
00590   }
00591 }
00592 
00593 
00594 /**
00595  * Send PUBACK, PUBREC or PUBREL response message
00596  * @param client MQTT client
00597  * @param msg PUBACK, PUBREC or PUBREL
00598  * @param pkt_id Packet identifier
00599  * @param qos QoS value
00600  * @return ERR_OK if successful, ERR_MEM if out of memory
00601  */
00602 static err_t
00603 pub_ack_rec_rel_response(mqtt_client_t *client, u8_t msg, u16_t pkt_id, u8_t qos)
00604 {
00605   err_t err = ERR_OK;
00606   if (mqtt_output_check_space(&client->output, 2)) {
00607     mqtt_output_append_fixed_header(&client->output, msg, 0, qos, 0, 2);
00608     mqtt_output_append_u16(&client->output, pkt_id);
00609     mqtt_output_send(&client->output, client->conn);
00610   } else {
00611     LWIP_DEBUGF(MQTT_DEBUG_TRACE,("pub_ack_rec_rel_response: OOM creating response: %s with pkt_id: %d\n",
00612                                   mqtt_msg_type_to_str(msg), pkt_id));
00613     err = ERR_MEM;
00614   }
00615   return err;
00616 }
00617 
00618 /**
00619  * Subscribe response from server
00620  * @param r Matching request
00621  * @param result Result code from server
00622  */
00623 static void
00624 mqtt_incomming_suback(struct mqtt_request_t *r, u8_t result)
00625 {
00626   if (r->cb != NULL) {
00627     r->cb(r->arg, result < 3 ? ERR_OK : ERR_ABRT);
00628   }
00629 }
00630 
00631 
00632 /**
00633  * Complete MQTT message received or buffer full
00634  * @param client MQTT client
00635  * @param fixed_hdr_idx header index
00636  * @param length length received part
00637  * @param remaining_length Remaining length of complete message
00638  */
00639 static mqtt_connection_status_t
00640   mqtt_message_received(mqtt_client_t *client, u8_t fixed_hdr_idx, u16_t length, u32_t remaining_length)
00641 {
00642   mqtt_connection_status_t res = MQTT_CONNECT_ACCEPTED;
00643 
00644   u8_t *var_hdr_payload = client->rx_buffer + fixed_hdr_idx;
00645 
00646   /* Control packet type */
00647   u8_t pkt_type = MQTT_CTL_PACKET_TYPE(client->rx_buffer[0]);
00648   u16_t pkt_id = 0;
00649 
00650   if (pkt_type == MQTT_MSG_TYPE_CONNACK) {
00651     if (client->conn_state == MQTT_CONNECTING) {
00652       /* Get result code from CONNACK */
00653       res = (mqtt_connection_status_t)var_hdr_payload[1];
00654       LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_message_received: Connect response code %d\n", res));
00655       if (res == MQTT_CONNECT_ACCEPTED) {
00656         /* Reset cyclic_tick when changing to connected state */
00657         client->cyclic_tick = 0;
00658         client->conn_state = MQTT_CONNECTED;
00659         /* Notify upper layer */
00660         if (client->connect_cb != 0) {
00661           client->connect_cb(client, client->connect_arg, res);
00662         }
00663       }
00664     } else {
00665       LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_message_received: Received CONNACK in connected state\n"));
00666     }
00667   } else if (pkt_type == MQTT_MSG_TYPE_PINGRESP) {
00668     LWIP_DEBUGF(MQTT_DEBUG_TRACE,( "mqtt_message_received: Received PINGRESP from server\n"));
00669 
00670   } else if (pkt_type == MQTT_MSG_TYPE_PUBLISH) {
00671     u16_t payload_offset = 0;
00672     u16_t payload_length = length;
00673     u8_t qos = MQTT_CTL_PACKET_QOS(client->rx_buffer[0]);
00674 
00675     if (client->msg_idx <= MQTT_VAR_HEADER_BUFFER_LEN) {
00676       /* Should have topic and pkt id*/
00677       uint8_t *topic;
00678       uint16_t after_topic;
00679       u8_t bkp;
00680       u16_t topic_len = var_hdr_payload[0];
00681       topic_len = (topic_len << 8) + (u16_t)(var_hdr_payload[1]);
00682 
00683       topic = var_hdr_payload + 2;
00684       after_topic = 2 + topic_len;
00685       /* Check length, add one byte even for QoS 0 so that zero termination will fit */
00686       if ((after_topic + (qos? 2 : 1)) > length) {
00687         LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_message_received: Receive buffer can not fit topic + pkt_id\n"));
00688         goto out_disconnect;
00689       }
00690 
00691       /* id for QoS 1 and 2 */
00692       if (qos > 0) {
00693         client->inpub_pkt_id = ((u16_t)var_hdr_payload[after_topic] << 8) + (u16_t)var_hdr_payload[after_topic + 1];
00694         after_topic += 2;
00695       } else {
00696         client->inpub_pkt_id = 0;
00697       }
00698       /* Take backup of byte after topic */
00699       bkp = topic[topic_len];
00700       /* Zero terminate string */
00701       topic[topic_len] = 0;
00702       /* Payload data remaining in receive buffer */
00703       payload_length = length - after_topic;
00704       payload_offset = after_topic;
00705 
00706       LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_incomming_publish: Received message with QoS %d at topic: %s, payload length %d\n",
00707                                     qos, topic, remaining_length + payload_length));
00708       if (client->pub_cb != NULL) {
00709         client->pub_cb(client->inpub_arg, (const char *)topic, remaining_length + payload_length);
00710       }
00711       /* Restore byte after topic */
00712       topic[topic_len] = bkp;
00713     }
00714     if (payload_length > 0 || remaining_length == 0) {
00715       client->data_cb(client->inpub_arg, var_hdr_payload + payload_offset, payload_length, remaining_length == 0 ? MQTT_DATA_FLAG_LAST : 0);
00716       /* Reply if QoS > 0 */
00717       if (remaining_length == 0 && qos > 0) {
00718         /* Send PUBACK for QoS 1 or PUBREC for QoS 2 */
00719         u8_t resp_msg = (qos == 1) ? MQTT_MSG_TYPE_PUBACK : MQTT_MSG_TYPE_PUBREC;
00720         LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_incomming_publish: Sending publish response: %s with pkt_id: %d\n",
00721                                       mqtt_msg_type_to_str(resp_msg), client->inpub_pkt_id));
00722         pub_ack_rec_rel_response(client, resp_msg, client->inpub_pkt_id, 0);
00723       }
00724     }
00725   } else {
00726     /* Get packet identifier */
00727     pkt_id = (u16_t)var_hdr_payload[0] << 8;
00728     pkt_id |= (u16_t)var_hdr_payload[1];
00729     if (pkt_id == 0) {
00730       LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_message_received: Got message with illegal packet identifier: 0\n"));
00731       goto out_disconnect;
00732     }
00733     if (pkt_type == MQTT_MSG_TYPE_PUBREC) {
00734       LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_message_received: PUBREC, sending PUBREL with pkt_id: %d\n",pkt_id));
00735       pub_ack_rec_rel_response(client, MQTT_MSG_TYPE_PUBREL, pkt_id, 1);
00736 
00737     } else if (pkt_type == MQTT_MSG_TYPE_PUBREL) {
00738       LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_message_received: PUBREL, sending PUBCOMP response with pkt_id: %d\n",pkt_id));
00739       pub_ack_rec_rel_response(client, MQTT_MSG_TYPE_PUBCOMP, pkt_id, 0);
00740 
00741     } else if (pkt_type == MQTT_MSG_TYPE_SUBACK || pkt_type == MQTT_MSG_TYPE_UNSUBACK ||
00742               pkt_type == MQTT_MSG_TYPE_PUBCOMP || pkt_type == MQTT_MSG_TYPE_PUBACK) {
00743       struct mqtt_request_t *r = mqtt_take_request(&client->pend_req_queue, pkt_id);
00744       if (r != NULL) {
00745         LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_message_received: %s response with id %d\n", mqtt_msg_type_to_str(pkt_type), pkt_id));
00746         if (pkt_type == MQTT_MSG_TYPE_SUBACK) {
00747           if (length < 3) {
00748             LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_message_received: To small SUBACK packet\n"));
00749             goto out_disconnect;
00750           } else {
00751             mqtt_incomming_suback(r, var_hdr_payload[2]);
00752           }
00753         } else if (r->cb != NULL) {
00754           r->cb(r->arg, ERR_OK);
00755         }
00756         mqtt_delete_request(r);
00757       } else {
00758         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));
00759       }
00760     } else {
00761       LWIP_DEBUGF(MQTT_DEBUG_WARN,( "mqtt_message_received: Received unknown message type: %d\n", pkt_type));
00762       goto out_disconnect;
00763     }
00764   }
00765   return res;
00766 out_disconnect:
00767   return MQTT_CONNECT_DISCONNECTED;
00768 }
00769 
00770 
00771 /**
00772  * MQTT incoming message parser
00773  * @param client MQTT client
00774  * @param p PBUF chain of received data
00775  * @return Connection status
00776  */
00777 static mqtt_connection_status_t
00778 mqtt_parse_incoming(mqtt_client_t *client, struct pbuf *p)
00779 {
00780   u16_t in_offset = 0;
00781   u32_t msg_rem_len = 0;
00782   u8_t fixed_hdr_idx = 0;
00783   u8_t b = 0;
00784 
00785   while (p->tot_len > in_offset) {
00786     if ((fixed_hdr_idx < 2) || ((b & 0x80) != 0)) {
00787 
00788       if (fixed_hdr_idx < client->msg_idx) {
00789         b = client->rx_buffer[fixed_hdr_idx];
00790       } else {
00791         b = pbuf_get_at(p, in_offset++);
00792         client->rx_buffer[client->msg_idx++] = b;
00793       }
00794       fixed_hdr_idx++;
00795 
00796       if (fixed_hdr_idx >= 2) {
00797         msg_rem_len |= (u32_t)(b & 0x7f) << ((fixed_hdr_idx - 2) * 7);
00798         if ((b & 0x80) == 0) {
00799           LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_parse_incoming: Remaining length after fixed header: %d\n", msg_rem_len));
00800           if (msg_rem_len == 0) {
00801             /* Complete message with no extra headers of payload received */
00802             mqtt_message_received(client, fixed_hdr_idx, 0, 0);
00803             client->msg_idx = 0;
00804             fixed_hdr_idx = 0;
00805           } else {
00806             /* Bytes remaining in message */
00807             msg_rem_len = (msg_rem_len + fixed_hdr_idx) - client->msg_idx;
00808           }
00809         }
00810       }
00811     } else {
00812       u16_t cpy_len, cpy_start, buffer_space;
00813 
00814       cpy_start = (client->msg_idx - fixed_hdr_idx) % (MQTT_VAR_HEADER_BUFFER_LEN - fixed_hdr_idx) + fixed_hdr_idx;
00815 
00816       /* Allow to copy the lesser one of available length in input data or bytes remaining in message */
00817       cpy_len = (u16_t)LWIP_MIN((u16_t)(p->tot_len - in_offset), msg_rem_len);
00818 
00819       /* Limit to available space in buffer */
00820       buffer_space = MQTT_VAR_HEADER_BUFFER_LEN - cpy_start;
00821       if (cpy_len > buffer_space) {
00822         cpy_len = buffer_space;
00823       }
00824       pbuf_copy_partial(p, client->rx_buffer+cpy_start, cpy_len, in_offset);
00825 
00826       /* Advance get and put indexes  */
00827       client->msg_idx += cpy_len;
00828       in_offset += cpy_len;
00829       msg_rem_len -= cpy_len;
00830 
00831       LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_parse_incoming: msg_idx: %d, cpy_len: %d, remaining %d\n", client->msg_idx, cpy_len, msg_rem_len));
00832       if (msg_rem_len == 0 || cpy_len == buffer_space) {
00833         /* Whole message received or buffer is full */
00834         mqtt_connection_status_t res = mqtt_message_received(client, fixed_hdr_idx, (cpy_start + cpy_len) - fixed_hdr_idx, msg_rem_len);
00835         if (res != MQTT_CONNECT_ACCEPTED) {
00836           return res;
00837         }
00838         if (msg_rem_len == 0) {
00839           /* Reset parser state */
00840           client->msg_idx = 0;
00841           /* msg_tot_len = 0; */
00842           fixed_hdr_idx = 0;
00843         }
00844       }
00845     }
00846   }
00847   return MQTT_CONNECT_ACCEPTED;
00848 }
00849 
00850 
00851 /**
00852  * TCP received callback function. @see tcp_recv_fn
00853  * @param arg MQTT client
00854  * @param p PBUF chain of received data
00855  * @param err Passed as return value if not ERR_OK
00856  * @return ERR_OK or err passed into callback
00857  */
00858 static err_t
00859 mqtt_tcp_recv_cb(void *arg, struct tcp_pcb *pcb, struct pbuf *p, err_t err)
00860 {
00861   mqtt_client_t *client = (mqtt_client_t *)arg;
00862   LWIP_ASSERT("mqtt_tcp_recv_cb: client != NULL", client != NULL);
00863   LWIP_ASSERT("mqtt_tcp_recv_cb: client->conn == pcb", client->conn == pcb);
00864 
00865   if (p == NULL) {
00866     LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_tcp_recv_cb: Recv pbuf=NULL, remote has closed connection\n"));
00867     mqtt_close(client, MQTT_CONNECT_DISCONNECTED);
00868   } else {
00869     mqtt_connection_status_t res;
00870     if (err != ERR_OK) {
00871       LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_tcp_recv_cb: Recv err=%d\n", err));
00872       pbuf_free(p);
00873       return err;
00874     }
00875 
00876     /* Tell remote that data has been received */
00877     tcp_recved(pcb, p->tot_len);
00878     res = mqtt_parse_incoming(client, p);
00879     pbuf_free(p);
00880 
00881     if (res != MQTT_CONNECT_ACCEPTED) {
00882       mqtt_close(client, res);
00883     }
00884     /* If keep alive functionality is used */
00885     if (client->keep_alive != 0) {
00886       /* Reset server alive watchdog */
00887       client->server_watchdog = 0;
00888     }
00889 
00890   }
00891   return ERR_OK;
00892 }
00893 
00894 
00895 /**
00896  * TCP data sent callback function. @see tcp_sent_fn
00897  * @param arg MQTT client
00898  * @param tpcb TCP connection handle
00899  * @param len Number of bytes sent
00900  * @return ERR_OK
00901  */
00902 static err_t
00903 mqtt_tcp_sent_cb(void *arg, struct tcp_pcb *tpcb, u16_t len)
00904 {
00905   mqtt_client_t *client = (mqtt_client_t *)arg;
00906 
00907   LWIP_UNUSED_ARG(tpcb);
00908   LWIP_UNUSED_ARG(len);
00909 
00910   if (client->conn_state == MQTT_CONNECTED) {
00911     struct mqtt_request_t *r;
00912 
00913     /* Reset keep-alive send timer and server watchdog */
00914     client->cyclic_tick = 0;
00915     client->server_watchdog = 0;
00916     /* QoS 0 publish has no response from server, so call its callbacks here */
00917     while ((r = mqtt_take_request(&client->pend_req_queue, 0)) != NULL) {
00918       LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_tcp_sent_cb: Calling QoS 0 publish complete callback\n"));
00919       if (r->cb != NULL) {
00920         r->cb(r->arg, ERR_OK);
00921       }
00922       mqtt_delete_request(r);
00923     }
00924     /* Try send any remaining buffers from output queue */
00925     mqtt_output_send(&client->output, client->conn);
00926   }
00927   return ERR_OK;
00928 }
00929 
00930 /**
00931  * TCP error callback function. @see tcp_err_fn
00932  * @param arg MQTT client
00933  * @param err Error encountered
00934  */
00935 static void
00936 mqtt_tcp_err_cb(void *arg, err_t err)
00937 {
00938   mqtt_client_t *client = (mqtt_client_t *)arg;
00939   LWIP_UNUSED_ARG(err); /* only used for debug output */
00940   LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_tcp_err_cb: TCP error callback: error %d, arg: %p\n", err, arg));
00941   LWIP_ASSERT("mqtt_tcp_err_cb: client != NULL", client != NULL);
00942   /* Set conn to null before calling close as pcb is already deallocated*/
00943   client->conn = 0;
00944   mqtt_close(client, MQTT_CONNECT_DISCONNECTED);
00945 }
00946 
00947 /**
00948  * TCP poll callback function. @see tcp_poll_fn
00949  * @param arg MQTT client
00950  * @param tpcb TCP connection handle
00951  * @return err ERR_OK
00952  */
00953 static err_t
00954 mqtt_tcp_poll_cb(void *arg, struct tcp_pcb *tpcb)
00955 {
00956   mqtt_client_t *client = (mqtt_client_t *)arg;
00957   if (client->conn_state == MQTT_CONNECTED) {
00958     /* Try send any remaining buffers from output queue */
00959     mqtt_output_send(&client->output, tpcb);
00960   }
00961   return ERR_OK;
00962 }
00963 
00964 /**
00965  * TCP connect callback function. @see tcp_connected_fn
00966  * @param arg MQTT client
00967  * @param err Always ERR_OK, mqtt_tcp_err_cb is called in case of error
00968  * @return ERR_OK
00969  */
00970 static err_t
00971 mqtt_tcp_connect_cb(void *arg, struct tcp_pcb *tpcb, err_t err)
00972 {
00973   mqtt_client_t* client = (mqtt_client_t *)arg;
00974 
00975   if (err != ERR_OK) {
00976     LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_tcp_connect_cb: TCP connect error %d\n", err));
00977     return err;
00978   }
00979 
00980   /* Initiate receiver state */
00981   client->msg_idx = 0;
00982 
00983   /* Setup TCP callbacks */
00984   tcp_recv(tpcb, mqtt_tcp_recv_cb);
00985   tcp_sent(tpcb, mqtt_tcp_sent_cb);
00986   tcp_poll(tpcb, mqtt_tcp_poll_cb, 2);
00987 
00988   LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_tcp_connect_cb: TCP connection established to server\n"));
00989   /* Enter MQTT connect state */
00990   client->conn_state = MQTT_CONNECTING;
00991 
00992   /* Start cyclic timer */
00993   sys_timeout(MQTT_CYCLIC_TIMER_INTERVAL*1000, mqtt_cyclic_timer, client);
00994   client->cyclic_tick = 0;
00995 
00996   /* Start transmission from output queue, connect message is the first one out*/
00997   mqtt_output_send(&client->output, client->conn);
00998 
00999   return ERR_OK;
01000 }
01001 
01002 
01003 
01004 /*---------------------------------------------------------------------------------------------------- */
01005 /* Public API */
01006 
01007 
01008 /**
01009  * @ingroup mqtt
01010  * MQTT publish function.
01011  * @param client MQTT client
01012  * @param topic Publish topic string
01013  * @param payload Data to publish (NULL is allowed)
01014  * @param payload_length: Length of payload (0 is allowed)
01015  * @param qos Quality of service, 0 1 or 2
01016  * @param retain MQTT retain flag
01017  * @param cb Callback to call when publish is complete or has timed out
01018  * @param arg User supplied argument to publish callback
01019  * @return ERR_OK if successful
01020  *         ERR_CONN if client is disconnected
01021  *         ERR_MEM if short on memory
01022  */
01023 err_t
01024 mqtt_publish(mqtt_client_t *client, const char *topic, const void *payload, u16_t payload_length, u8_t qos, u8_t retain,
01025              mqtt_request_cb_t cb, void *arg)
01026 {
01027   struct mqtt_request_t *r;
01028   u16_t pkt_id;
01029   size_t topic_strlen;
01030   size_t total_len;
01031   u16_t topic_len;
01032   u16_t remaining_length;
01033 
01034   LWIP_ASSERT("mqtt_publish: client != NULL", client);
01035   LWIP_ASSERT("mqtt_publish: topic != NULL", topic);
01036   LWIP_ERROR("mqtt_publish: TCP disconnected", (client->conn_state != TCP_DISCONNECTED), return ERR_CONN);
01037 
01038   topic_strlen = strlen(topic);
01039   LWIP_ERROR("mqtt_publish: topic length overflow", (topic_strlen <= (0xFFFF - 2)), return ERR_ARG);
01040   topic_len = (u16_t)topic_strlen;
01041   total_len = 2 + topic_len + payload_length;
01042   LWIP_ERROR("mqtt_publish: total length overflow", (total_len <= 0xFFFF), return ERR_ARG);
01043   remaining_length = (u16_t)total_len;
01044 
01045   LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_publish: Publish with payload length %d to topic \"%s\"\n", payload_length, topic));
01046 
01047   if (qos > 0) {
01048     remaining_length += 2;
01049     /* Generate pkt_id id for QoS1 and 2 */
01050     pkt_id = msg_generate_packet_id(client);
01051   } else {
01052     /* Use reserved value pkt_id 0 for QoS 0 in request handle */
01053     pkt_id = 0;
01054   }
01055 
01056   r = mqtt_create_request(client->req_list, pkt_id, cb, arg);
01057   if (r == NULL) {
01058     return ERR_MEM;
01059   }
01060 
01061   if (mqtt_output_check_space(&client->output, remaining_length) == 0) {
01062     mqtt_delete_request(r);
01063     return ERR_MEM;
01064   }
01065   /* Append fixed header */
01066   mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain, remaining_length);
01067 
01068   /* Append Topic */
01069   mqtt_output_append_string(&client->output, topic, topic_len);
01070 
01071   /* Append packet if for QoS 1 and 2*/
01072   if (qos > 0) {
01073     mqtt_output_append_u16(&client->output, pkt_id);
01074   }
01075 
01076   /* Append optional publish payload */
01077   if ((payload != NULL) && (payload_length > 0)) {
01078     mqtt_output_append_buf(&client->output, payload, payload_length);
01079   }
01080 
01081   mqtt_append_request(&client->pend_req_queue, r);
01082   mqtt_output_send(&client->output, client->conn);
01083   return ERR_OK;
01084 }
01085 
01086 
01087 /**
01088  * @ingroup mqtt
01089  * MQTT subscribe/unsubscribe function.
01090  * @param client MQTT client
01091  * @param topic topic to subscribe to
01092  * @param qos Quality of service, 0 1 or 2 (only used for subscribe)
01093  * @param cb Callback to call when subscribe/unsubscribe reponse is received
01094  * @param arg User supplied argument to publish callback
01095  * @param sub 1 for subscribe, 0 for unsubscribe
01096  * @return ERR_OK if successful, @see err_t enum for other results
01097  */
01098 err_t
01099 mqtt_sub_unsub(mqtt_client_t *client, const char *topic, u8_t qos, mqtt_request_cb_t cb, void *arg, u8_t sub)
01100 {
01101   size_t topic_strlen;
01102   size_t total_len;
01103   u16_t topic_len;
01104   u16_t remaining_length;
01105   u16_t pkt_id;
01106   struct mqtt_request_t *r;
01107 
01108   LWIP_ASSERT("mqtt_sub_unsub: client != NULL", client);
01109   LWIP_ASSERT("mqtt_sub_unsub: topic != NULL", topic);
01110 
01111   topic_strlen = strlen(topic);
01112   LWIP_ERROR("mqtt_sub_unsub: topic length overflow", (topic_strlen <= (0xFFFF - 2)), return ERR_ARG);
01113   topic_len = (u16_t)topic_strlen;
01114   /* Topic string, pkt_id, qos for subscribe */
01115   total_len =  topic_len + 2 + 2 + (sub != 0);
01116   LWIP_ERROR("mqtt_sub_unsub: total length overflow", (total_len <= 0xFFFF), return ERR_ARG);
01117   remaining_length = (u16_t)total_len;
01118 
01119   LWIP_ASSERT("mqtt_sub_unsub: qos < 3", qos < 3);
01120   if (client->conn_state == TCP_DISCONNECTED) {
01121     LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_sub_unsub: Can not (un)subscribe in disconnected state\n"));
01122     return ERR_CONN;
01123   }
01124 
01125   pkt_id = msg_generate_packet_id(client);
01126   r = mqtt_create_request(client->req_list, pkt_id, cb, arg);
01127   if (r == NULL) {
01128     return ERR_MEM;
01129   }
01130 
01131   if (mqtt_output_check_space(&client->output, remaining_length) == 0) {
01132     mqtt_delete_request(r);
01133     return ERR_MEM;
01134   }
01135 
01136   LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_sub_unsub: Client (un)subscribe to topic \"%s\", id: %d\n", topic, pkt_id));
01137 
01138   mqtt_output_append_fixed_header(&client->output, sub ? MQTT_MSG_TYPE_SUBSCRIBE : MQTT_MSG_TYPE_UNSUBSCRIBE, 0, 1, 0, remaining_length);
01139   /* Packet id */
01140   mqtt_output_append_u16(&client->output, pkt_id);
01141   /* Topic */
01142   mqtt_output_append_string(&client->output, topic, topic_len);
01143   /* QoS */
01144   if (sub != 0) {
01145     mqtt_output_append_u8(&client->output, LWIP_MIN(qos, 2));
01146   }
01147 
01148   mqtt_append_request(&client->pend_req_queue, r);
01149   mqtt_output_send(&client->output, client->conn);
01150   return ERR_OK;
01151 }
01152 
01153 
01154 /**
01155  * @ingroup mqtt
01156  * Set callback to handle incoming publish requests from server
01157  * @param client MQTT client
01158  * @param pub_cb Callback invoked when publish starts, contain topic and total length of payload
01159  * @param data_cb Callback for each fragment of payload that arrives
01160  * @param arg User supplied argument to both callbacks
01161  */
01162 void
01163 mqtt_set_inpub_callback(mqtt_client_t *client, mqtt_incoming_publish_cb_t pub_cb,
01164                              mqtt_incoming_data_cb_t data_cb, void *arg)
01165 {
01166   LWIP_ASSERT("mqtt_set_inpub_callback: client != NULL", client != NULL);
01167   client->data_cb = data_cb;
01168   client->pub_cb = pub_cb;
01169   client->inpub_arg = arg;
01170 }
01171 
01172 /**
01173  * @ingroup mqtt
01174  * Create a new MQTT client instance
01175  * @return Pointer to instance on success, NULL otherwise
01176  */
01177 mqtt_client_t *
01178 mqtt_client_new(void)
01179 {
01180   mqtt_client_t *client = (mqtt_client_t *)mem_malloc(sizeof(mqtt_client_t));
01181   if (client != NULL) {
01182     memset(client, 0, sizeof(mqtt_client_t));
01183   }
01184   return client;
01185 }
01186 
01187 
01188 /**
01189  * @ingroup mqtt
01190  * Connect to MQTT server
01191  * @param client MQTT client
01192  * @param ip_addr Server IP
01193  * @param port Server port
01194  * @param cb Connection state change callback
01195  * @param arg User supplied argument to connection callback
01196  * @param client_info Client identification and connection options
01197  * @return ERR_OK if successful, @see err_t enum for other results
01198  */
01199 err_t
01200 mqtt_client_connect(mqtt_client_t *client, const ip_addr_t *ip_addr, u16_t port, mqtt_connection_cb_t cb, void *arg,
01201                     const struct mqtt_connect_client_info_t *client_info)
01202 {
01203   err_t err;
01204   size_t len;
01205   u16_t client_id_length;
01206   /* Length is the sum of 2+"MQTT", protocol level, flags and keep alive */
01207   u16_t remaining_length = 2 + 4 + 1 + 1 + 2;
01208   u8_t flags = 0, will_topic_len = 0, will_msg_len = 0;
01209 
01210   LWIP_ASSERT("mqtt_client_connect: client != NULL", client != NULL);
01211   LWIP_ASSERT("mqtt_client_connect: ip_addr != NULL", ip_addr != NULL);
01212   LWIP_ASSERT("mqtt_client_connect: client_info != NULL", client_info != NULL);
01213   LWIP_ASSERT("mqtt_client_connect: client_info->client_id != NULL", client_info->client_id != NULL);
01214 
01215   if (client->conn_state != TCP_DISCONNECTED) {
01216     LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_client_connect: Already connected\n"));
01217     return ERR_ISCONN;
01218   }
01219 
01220   /* Wipe clean */
01221   memset(client, 0, sizeof(mqtt_client_t));
01222   client->connect_arg = arg;
01223   client->connect_cb = cb;
01224   client->keep_alive = client_info->keep_alive;
01225   mqtt_init_requests(client->req_list);
01226 
01227   /* Build connect message */
01228   if (client_info->will_topic != NULL && client_info->will_msg != NULL) {
01229     flags |= MQTT_CONNECT_FLAG_WILL;
01230     flags |= (client_info->will_qos & 3) << 3;
01231     if (client_info->will_retain) {
01232       flags |= MQTT_CONNECT_FLAG_WILL_RETAIN;
01233     }
01234     len = strlen(client_info->will_topic);
01235     LWIP_ERROR("mqtt_client_connect: client_info->will_topic length overflow", len <= 0xFF, return ERR_VAL);
01236     LWIP_ERROR("mqtt_client_connect: client_info->will_topic length must be > 0", len > 0, return ERR_VAL);
01237     will_topic_len = (u8_t)len;
01238     len = strlen(client_info->will_msg);
01239     LWIP_ERROR("mqtt_client_connect: client_info->will_msg length overflow", len <= 0xFF, return ERR_VAL);
01240     will_msg_len = (u8_t)len;
01241     len = remaining_length + 2 + will_topic_len + 2 + will_msg_len;
01242     LWIP_ERROR("mqtt_client_connect: remaining_length overflow", len <= 0xFFFF, return ERR_VAL);
01243     remaining_length = (u16_t)len;
01244   }
01245 
01246   /* Don't complicate things, always connect using clean session */
01247   flags |= MQTT_CONNECT_FLAG_CLEAN_SESSION;
01248 
01249   len = strlen(client_info->client_id);
01250   LWIP_ERROR("mqtt_client_connect: client_info->client_id length overflow", len <= 0xFFFF, return ERR_VAL);
01251   client_id_length = (u16_t)len;
01252   len = remaining_length + 2 + client_id_length;
01253   LWIP_ERROR("mqtt_client_connect: remaining_length overflow", len <= 0xFFFF, return ERR_VAL);
01254   remaining_length = (u16_t)len;
01255 
01256   if (mqtt_output_check_space(&client->output, remaining_length) == 0) {
01257     return ERR_MEM;
01258   }
01259 
01260   client->conn = tcp_new();
01261   if (client->conn == NULL) {
01262     return ERR_MEM;
01263   }
01264 
01265   /* Set arg pointer for callbacks */
01266   tcp_arg(client->conn, client);
01267   /* Any local address, pick random local port number */
01268   err = tcp_bind(client->conn, IP_ADDR_ANY, 0);
01269   if (err != ERR_OK) {
01270     LWIP_DEBUGF(MQTT_DEBUG_WARN,("mqtt_client_connect: Error binding to local ip/port, %d\n", err));
01271     goto tcp_fail;
01272   }
01273   LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_client_connect: Connecting to host: %s at port:%"U16_F"\n", ipaddr_ntoa(ip_addr), port));
01274 
01275   /* Connect to server */
01276   err = tcp_connect(client->conn, ip_addr, port, mqtt_tcp_connect_cb);
01277   if (err != ERR_OK) {
01278     LWIP_DEBUGF(MQTT_DEBUG_TRACE,("mqtt_client_connect: Error connecting to remote ip/port, %d\n", err));
01279     goto tcp_fail;
01280   }
01281   /* Set error callback */
01282   tcp_err(client->conn, mqtt_tcp_err_cb);
01283   client->conn_state = TCP_CONNECTING;
01284 
01285   /* Append fixed header */
01286   mqtt_output_append_fixed_header(&client->output, MQTT_MSG_TYPE_CONNECT, 0, 0, 0, remaining_length);
01287   /* Append Protocol string */
01288   mqtt_output_append_string(&client->output, "MQTT", 4);
01289   /* Append Protocol level */
01290   mqtt_output_append_u8(&client->output, 4);
01291   /* Append connect flags */
01292   mqtt_output_append_u8(&client->output, flags);
01293   /* Append keep-alive */
01294   mqtt_output_append_u16(&client->output, client_info->keep_alive);
01295   /* Append client id */
01296   mqtt_output_append_string(&client->output, client_info->client_id, client_id_length);
01297   /* Append will message if used */
01298   if ((flags & MQTT_CONNECT_FLAG_WILL) != 0) {
01299     mqtt_output_append_string(&client->output, client_info->will_topic, will_topic_len);
01300     mqtt_output_append_string(&client->output, client_info->will_msg, will_msg_len);
01301   }
01302   return ERR_OK;
01303 
01304 tcp_fail:
01305   tcp_abort(client->conn);
01306   client->conn = NULL;
01307   return err;
01308 }
01309 
01310 
01311 /**
01312  * @ingroup mqtt
01313  * Disconnect from MQTT server
01314  * @param client MQTT client
01315  */
01316 void
01317 mqtt_disconnect(mqtt_client_t *client)
01318 {
01319   LWIP_ASSERT("mqtt_disconnect: client != NULL", client);
01320   /* If connection in not already closed */
01321   if (client->conn_state != TCP_DISCONNECTED) {
01322     /* Set conn_state before calling mqtt_close to prevent callback from being called */
01323     client->conn_state = TCP_DISCONNECTED;
01324     mqtt_close(client, (mqtt_connection_status_t)0);
01325   }
01326 }
01327 
01328 /**
01329  * @ingroup mqtt
01330  * Check connection with server
01331  * @param client MQTT client
01332  * @return 1 if connected to server, 0 otherwise
01333  */
01334 u8_t
01335 mqtt_client_is_connected(mqtt_client_t *client)
01336 {
01337   LWIP_ASSERT("mqtt_client_is_connected: client != NULL", client);
01338   return client->conn_state == MQTT_CONNECTED;
01339 }
01340 
01341 #endif /* LWIP_TCP && LWIP_CALLBACK_API */