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