Contiki-NG
Loading...
Searching...
No Matches
mqtt.c
Go to the documentation of this file.
1/*
2 * Copyright (c) 2015, Texas Instruments Incorporated - http://www.ti.com/
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
7 * are met:
8 * 1. Redistributions of source code must retain the above copyright
9 * notice, this list of conditions and the following disclaimer.
10 * 2. Redistributions in binary form must reproduce the above copyright
11 * notice, this list of conditions and the following disclaimer in the
12 * documentation and/or other materials provided with the distribution.
13 * 3. Neither the name of the copyright holder nor the names of its
14 * contributors may be used to endorse or promote products derived
15 * from this software without specific prior written permission.
16 *
17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
18 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
19 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
20 * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
21 * COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
22 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
23 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
24 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
25 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
26 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
27 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
28 * OF THE POSSIBILITY OF SUCH DAMAGE.
29 */
30/*---------------------------------------------------------------------------*/
31/**
32 * \addtogroup mqtt-engine
33 * @{
34 */
35/**
36 * \file
37 * Implementation of the Contiki MQTT engine
38 *
39 * \author
40 * Texas Instruments
41 */
42/*---------------------------------------------------------------------------*/
43#include "mqtt.h"
44#include "mqtt-prop.h"
45#include "contiki.h"
46#include "contiki-net.h"
47#include "contiki-lib.h"
48#include "lib/random.h"
49#include "sys/ctimer.h"
50#include "sys/etimer.h"
51#include "sys/pt.h"
52#include "net/ipv6/uip.h"
53#include "net/ipv6/uip-ds6.h"
54#include "dev/leds.h"
55
56#include "tcp-socket.h"
57
58#include "lib/assert.h"
59#include "lib/list.h"
60#include "sys/cc.h"
61
62#include <stdlib.h>
63#include <stdio.h>
64#include <string.h>
65#include <stdarg.h>
66/*---------------------------------------------------------------------------*/
67#define DEBUG 0
68#if DEBUG
69#define PRINTF(...) printf(__VA_ARGS__)
70#else
71#define PRINTF(...)
72#endif
73/*---------------------------------------------------------------------------*/
74typedef enum {
75 MQTT_FHDR_DUP_FLAG = 0x08,
76
77 MQTT_FHDR_QOS_LEVEL_0 = 0x00,
78 MQTT_FHDR_QOS_LEVEL_1 = 0x02,
79 MQTT_FHDR_QOS_LEVEL_2 = 0x04,
80
81 MQTT_FHDR_RETAIN_FLAG = 0x01,
82} mqtt_fhdr_fields_t;
83/*---------------------------------------------------------------------------*/
84typedef enum {
85 MQTT_VHDR_USERNAME_FLAG = 0x80,
86 MQTT_VHDR_PASSWORD_FLAG = 0x40,
87
88 MQTT_VHDR_WILL_RETAIN_FLAG = 0x20,
89 MQTT_VHDR_WILL_QOS_LEVEL_0 = 0x00,
90 MQTT_VHDR_WILL_QOS_LEVEL_1 = 0x08,
91 MQTT_VHDR_WILL_QOS_LEVEL_2 = 0x10,
92
93 MQTT_VHDR_WILL_FLAG = 0x04,
94 MQTT_VHDR_CLEAN_SESSION_FLAG = 0x02, /* called Clean Start in MQTTv5.0 */
95} mqtt_vhdr_conn_fields_t;
96/*---------------------------------------------------------------------------*/
97typedef enum {
98 MQTT_VHDR_CONN_ACCEPTED,
99 MQTT_VHDR_CONN_REJECTED_PROTOCOL,
100 MQTT_VHDR_CONN_REJECTED_IDENTIFIER,
101 MQTT_VHDR_CONN_REJECTED_UNAVAILABLE,
102 MQTT_VHDR_CONN_REJECTED_BAD_USER_PASS,
103 MQTT_VHDR_CONN_REJECTED_UNAUTHORIZED,
104} mqtt_vhdr_connack_ret_code_t;
105
106typedef enum {
107 MQTT_VHDR_CONNACK_SESSION_PRESENT = 0x1
108} mqtt_vhdr_connack_flags_t;
109
110/*---------------------------------------------------------------------------*/
111typedef enum {
112 MQTT_SUBACK_RET_QOS_0 = 0x00,
113 MQTT_SUBACK_RET_QOS_1 = 0x01,
114 MQTT_SUBACK_RET_QOS_2 = 0x02,
115 MQTT_SUBACK_RET_FAIL = 0x08,
116} mqtt_suback_ret_code_t;
117/*---------------------------------------------------------------------------*/
118/* MQTTv5.0 Reason Codes */
119typedef enum {
120 MQTT_VHDR_RC_SUCCES_OR_NORMAL = 0x00,
121 MQTT_VHDR_RC_QOS_0 = 0x01,
122 MQTT_VHDR_RC_QOS_1 = 0x02,
123 MQTT_VHDR_RC_QOS_2 = 0x03,
124 MQTT_VHDR_RC_DISC_WITH_WILL = 0x04,
125 MQTT_VHDR_RC_NO_MATCH_SUB = 0x10,
126 MQTT_VHDR_RC_NO_SUB_EXISTED = 0x11,
127 MQTT_VHDR_RC_CONTINUE_AUTH = 0x18,
128 MQTT_VHDR_RC_REAUTH = 0x19,
129 MQTT_VHDR_RC_UNSPEC_ERR = 0x80,
130 MQTT_VHDR_RC_MALFORMED_PKT = 0x81,
131 MQTT_VHDR_RC_PROTOCOL_ERR = 0x82,
132 MQTT_VHDR_RC_IMPL_SPEC_ERR = 0x83,
133 MQTT_VHDR_RC_PROT_VER_UNUSUPPORTED = 0x84,
134 MQTT_VHDR_RC_CLIENT_ID_INVALID = 0x85,
135 MQTT_VHDR_RC_BAD_USER_PASS = 0x86,
136 MQTT_VHDR_RC_NOT_AUTH = 0x87,
137 MQTT_VHDR_RC_SRV_UNAVAIL = 0x88,
138 MQTT_VHDR_RC_SRV_BUSY = 0x89,
139 MQTT_VHDR_RC_BANNED = 0x8A,
140 MQTT_VHDR_RC_SRV_SHUTDOWN = 0x8B,
141 MQTT_VHDR_RC_BAD_AUTH_METHOD = 0x8C,
142 MQTT_VHDR_RC_KEEP_ALIVE_TIMEOUT = 0x8D,
143 MQTT_VHDR_RC_SESS_TAKEN_OVER = 0x8E,
144 MQTT_VHDR_RC_TOPIC_FILT_INVAL = 0x8F,
145 MQTT_VHDR_RC_TOPIC_NAME_INVAL = 0x90,
146 MQTT_VHDR_RC_PKT_ID_IN_USE = 0x91,
147 MQTT_VHDR_RC_PKT_ID_NOT_FOUND = 0x92,
148 MQTT_VHDR_RC_RECV_MAX_EXCEEDED = 0x93,
149 MQTT_VHDR_RC_TOPIC_ALIAS_INVAL = 0x94,
150 MQTT_VHDR_RC_PKT_TOO_LARGE = 0x95,
151 MQTT_VHDR_RC_MSG_RATE_TOO_HIGH = 0x96,
152 MQTT_VHDR_RC_QUOTA_EXCEEDED = 0x97,
153 MQTT_VHDR_RC_ADMIN_ACTION = 0x98,
154 MQTT_VHDR_RC_PAYLD_FMT_INVAL = 0x99,
155 MQTT_VHDR_RC_RETAIN_UNSUPPORTED = 0x9A,
156 MQTT_VHDR_RC_QOS_UNSUPPORTED = 0x9B,
157 MQTT_VHDR_RC_USE_ANOTHER_SRV = 0x9C,
158 MQTT_VHDR_RC_SRV_MOVED = 0x9D,
159 MQTT_VHDR_RC_SHARED_SUB_UNSUPPORTED = 0x9E,
160 MQTT_VHDR_RC_CONN_RATE_EXCEEDED = 0x9F,
161 MQTT_VHDR_RC_MAX_CONN_TIME = 0xA0,
162 MQTT_VHDR_RC_SUB_ID_UNSUPPORTED = 0xA1,
163 MQTT_VHDR_RC_WILD_SUB_UNSUPPORTED = 0xA2,
164} mqtt_reason_code_t;
165/*---------------------------------------------------------------------------*/
166#define RESPONSE_WAIT_TIMEOUT (CLOCK_SECOND * 10)
167/*---------------------------------------------------------------------------*/
168#define INCREMENT_MID(conn) (conn)->mid_counter += 2
169#define MQTT_STRING_LENGTH(s) (((s)->length) == 0 ? 0 : (MQTT_STRING_LEN_SIZE + (s)->length))
170/*---------------------------------------------------------------------------*/
171/* Protothread send macros */
172#define PT_MQTT_WRITE_BYTES(conn, data, len) \
173 conn->out_write_pos = 0; \
174 while(write_bytes(conn, data, len)) { \
175 PT_WAIT_UNTIL(pt, (conn)->out_buffer_sent); \
176 }
177
178#define PT_MQTT_WRITE_BYTE(conn, data) \
179 while(write_byte(conn, data)) { \
180 PT_WAIT_UNTIL(pt, (conn)->out_buffer_sent); \
181 }
182/*---------------------------------------------------------------------------*/
183/*
184 * Sends the continue send event and wait for that event.
185 *
186 * The reason we cannot use PROCESS_PAUSE() is since we would risk loosing any
187 * events posted during the sending process.
188 */
189#define PT_MQTT_WAIT_SEND() \
190 do { \
191 if (PROCESS_ERR_OK == \
192 process_post(PROCESS_CURRENT(), mqtt_continue_send_event, NULL)) { \
193 do { \
194 PROCESS_WAIT_EVENT(); \
195 if(ev == mqtt_abort_now_event) { \
196 conn->state = MQTT_CONN_STATE_ABORT_IMMEDIATE; \
197 PT_INIT(&conn->out_proto_thread); \
198 process_post(PROCESS_CURRENT(), ev, data); \
199 } else if(ev >= mqtt_event_min && ev <= mqtt_event_max) { \
200 process_post(PROCESS_CURRENT(), ev, data); \
201 } \
202 } while (ev != mqtt_continue_send_event); \
203 } \
204 } while(0)
205/*---------------------------------------------------------------------------*/
206static process_event_t mqtt_do_connect_tcp_event;
207static process_event_t mqtt_do_connect_mqtt_event;
208static process_event_t mqtt_do_disconnect_mqtt_event;
209static process_event_t mqtt_do_subscribe_event;
210static process_event_t mqtt_do_unsubscribe_event;
211static process_event_t mqtt_do_publish_event;
212static process_event_t mqtt_do_pingreq_event;
213static process_event_t mqtt_continue_send_event;
214static process_event_t mqtt_abort_now_event;
215static process_event_t mqtt_do_auth_event;
216process_event_t mqtt_update_event;
217
218/*
219 * Min and Max event numbers we want to acknowledge while we're in the process
220 * of doing something else. continue_send does not count, therefore must be
221 * allocated last
222 */
223static process_event_t mqtt_event_min;
224static process_event_t mqtt_event_max;
225/*---------------------------------------------------------------------------*/
226/* Prototypes */
227static int
228tcp_input(struct tcp_socket *s, void *ptr, const uint8_t *input_data_ptr,
229 int input_data_len);
230
231static void tcp_event(struct tcp_socket *s, void *ptr,
232 tcp_socket_event_t event);
233
234static void reset_packet(struct mqtt_in_packet *packet);
235/*---------------------------------------------------------------------------*/
236LIST(mqtt_conn_list);
237/*---------------------------------------------------------------------------*/
238PROCESS(mqtt_process, "MQTT process");
239/*---------------------------------------------------------------------------*/
240static void
241call_event(struct mqtt_connection *conn,
242 mqtt_event_t event,
243 void *data)
244{
245 conn->event_callback(conn, event, data);
246 process_post(conn->app_process, mqtt_update_event, NULL);
247}
248/*---------------------------------------------------------------------------*/
249static void
250reset_defaults(struct mqtt_connection *conn)
251{
252 conn->mid_counter = 1;
253 PT_INIT(&conn->out_proto_thread);
254 conn->waiting_for_pingresp = 0;
255
256 reset_packet(&conn->in_packet);
257 conn->out_buffer_sent = 0;
258}
259/*---------------------------------------------------------------------------*/
260static void
261abort_connection(struct mqtt_connection *conn)
262{
263 conn->out_buffer_ptr = conn->out_buffer;
264 conn->out_queue_full = 0;
265
266 /* Reset outgoing packet */
267 memset(&conn->out_packet, 0, sizeof(conn->out_packet));
268
269 tcp_socket_close(&conn->socket);
270 tcp_socket_unregister(&conn->socket);
271
272 memset(&conn->socket, 0, sizeof(conn->socket));
273
274 conn->state = MQTT_CONN_STATE_NOT_CONNECTED;
275}
276/*---------------------------------------------------------------------------*/
277static void
278connect_tcp(struct mqtt_connection *conn)
279{
280 conn->state = MQTT_CONN_STATE_TCP_CONNECTING;
281
282 reset_defaults(conn);
283 tcp_socket_register(&(conn->socket),
284 conn,
285 conn->in_buffer,
286 MQTT_TCP_INPUT_BUFF_SIZE,
287 conn->out_buffer,
288 MQTT_TCP_OUTPUT_BUFF_SIZE,
289 tcp_input,
290 tcp_event);
291 tcp_socket_connect(&(conn->socket), &(conn->server_ip), conn->server_port);
292}
293/*---------------------------------------------------------------------------*/
294static void
295disconnect_tcp(struct mqtt_connection *conn)
296{
297 conn->state = MQTT_CONN_STATE_DISCONNECTING;
298 tcp_socket_close(&(conn->socket));
299 tcp_socket_unregister(&conn->socket);
300
301 memset(&conn->socket, 0, sizeof(conn->socket));
302}
303/*---------------------------------------------------------------------------*/
304static void
305send_out_buffer(struct mqtt_connection *conn)
306{
307 if(conn->out_buffer_ptr - conn->out_buffer == 0) {
308 conn->out_buffer_sent = 1;
309 return;
310 }
311 conn->out_buffer_sent = 0;
312
313 DBG("MQTT - (send_out_buffer) Space used in buffer: %i\n",
314 conn->out_buffer_ptr - conn->out_buffer);
315
316 tcp_socket_send(&conn->socket, conn->out_buffer,
317 conn->out_buffer_ptr - conn->out_buffer);
318}
319/*---------------------------------------------------------------------------*/
320static void
321string_to_mqtt_string(struct mqtt_string *mqtt_string, char *string)
322{
323 if(mqtt_string == NULL) {
324 return;
325 }
326 mqtt_string->string = string;
327
328 if(string != NULL) {
329 mqtt_string->length = strlen(string);
330 } else {
331 mqtt_string->length = 0;
332 }
333}
334/*---------------------------------------------------------------------------*/
335static int
336write_byte(struct mqtt_connection *conn, uint8_t data)
337{
338 DBG("MQTT - (write_byte) buff_size: %i write: '%02X'\n",
339 &conn->out_buffer[MQTT_TCP_OUTPUT_BUFF_SIZE] - conn->out_buffer_ptr,
340 data);
341
342 if(&conn->out_buffer[MQTT_TCP_OUTPUT_BUFF_SIZE] - conn->out_buffer_ptr == 0) {
343 send_out_buffer(conn);
344 return 1;
345 }
346
347 *conn->out_buffer_ptr = data;
348 conn->out_buffer_ptr++;
349 return 0;
350}
351/*---------------------------------------------------------------------------*/
352static int
353write_bytes(struct mqtt_connection *conn, uint8_t *data, uint16_t len)
354{
355 uint16_t write_bytes;
356 write_bytes =
357 MIN(&conn->out_buffer[MQTT_TCP_OUTPUT_BUFF_SIZE] - conn->out_buffer_ptr,
358 len - conn->out_write_pos);
359
360 memcpy(conn->out_buffer_ptr, &data[conn->out_write_pos], write_bytes);
361 conn->out_write_pos += write_bytes;
362 conn->out_buffer_ptr += write_bytes;
363
364 DBG("MQTT - (write_bytes) len: %u write_pos: %i\n", len,
365 conn->out_write_pos);
366
367 if(len - conn->out_write_pos == 0) {
368 conn->out_write_pos = 0;
369 return 0;
370 } else {
371 send_out_buffer(conn);
372 return len - conn->out_write_pos;
373 }
374}
375/*---------------------------------------------------------------------------*/
376uint8_t
377mqtt_decode_var_byte_int(const uint8_t *input_data_ptr,
378 int input_data_len,
379 uint32_t *input_pos,
380 uint32_t *pkt_byte_count,
381 uint16_t *dest)
382{
383 uint8_t read_bytes = 0;
384 uint8_t byte_in;
385 uint32_t multiplier = 1;
386 uint32_t input_pos_0 = 0;
387
388 if(input_pos == NULL) {
389 input_pos = &input_pos_0;
390 }
391
392 *dest = 0;
393
394 do {
395 if(*input_pos >= input_data_len) {
396 return 0;
397 }
398
399 byte_in = input_data_ptr[*input_pos];
400 (*input_pos)++;
401 if(pkt_byte_count) {
402 (*pkt_byte_count)++;
403 }
404 read_bytes++;
405 DBG("MQTT - Read Variable Byte Integer byte %i\n", byte_in);
406
407 if(read_bytes > 4) {
408 DBG("Received more than 4 byte 'Variable Byte Integer'.");
409 return 0;
410 }
411
412 *dest += (byte_in & 127) * multiplier;
413 multiplier *= 128;
414 } while((byte_in & 128) != 0);
415
416 return read_bytes;
417}
418/*---------------------------------------------------------------------------*/
419void
420mqtt_encode_var_byte_int(uint8_t *vbi_out,
421 uint8_t *vbi_bytes,
422 uint32_t val)
423{
424 uint8_t digit;
425
426 DBG("MQTT - Encoding Variable Byte Integer %u\n", val);
427
428 *vbi_bytes = 0;
429 do {
430 digit = val % 128;
431 val = val / 128;
432 if(val > 0) {
433 digit = digit | 0x80;
434 }
435
436 vbi_out[*vbi_bytes] = digit;
437 (*vbi_bytes)++;
438 DBG("MQTT - Encode VBI digit '%u' length '%i'\n", digit, val);
439 } while(val > 0 && *vbi_bytes < 5);
440 DBG("MQTT - var_byte_int bytes %u\n", *vbi_bytes);
441}
442/*---------------------------------------------------------------------------*/
443static void
444keep_alive_callback(void *ptr)
445{
446 struct mqtt_connection *conn = ptr;
447
448 DBG("MQTT - (keep_alive_callback) Called!\n");
449
450 /* The flag is set when the PINGREQ has been sent */
451 if(conn->waiting_for_pingresp) {
452 PRINTF("MQTT - Disconnect due to no PINGRESP from broker.\n");
453 disconnect_tcp(conn);
454 return;
455 }
456
457 process_post(&mqtt_process, mqtt_do_pingreq_event, conn);
458}
459/*---------------------------------------------------------------------------*/
460static void
461reset_packet(struct mqtt_in_packet *packet)
462{
463 memset(packet, 0, sizeof(struct mqtt_in_packet));
464}
465/*---------------------------------------------------------------------------*/
466#if MQTT_5
467static
468PT_THREAD(write_out_props(struct pt *pt, struct mqtt_connection *conn,
469 struct mqtt_prop_list *prop_list))
470{
471 PT_BEGIN(pt);
472
473 static struct mqtt_prop_out_property *prop;
474
475 if(prop_list) {
476 DBG("MQTT - Writing %i property bytes\n", prop_list->properties_len + prop_list->properties_len_enc_bytes);
477 /* Write total length of properties */
478 PT_MQTT_WRITE_BYTES(conn,
479 prop_list->properties_len_enc,
480 prop_list->properties_len_enc_bytes);
481
482 prop = (struct mqtt_prop_out_property *)list_head(prop_list->props);
483 do {
484 if(prop != NULL) {
485 DBG("MQTT - Property ID %i len %i\n", prop->id, prop->property_len);
486 PT_MQTT_WRITE_BYTE(conn, prop->id);
487 PT_MQTT_WRITE_BYTES(conn,
488 prop->val,
489 prop->property_len);
490 }
491 prop = (struct mqtt_prop_out_property *)list_item_next(prop);
492 } while(prop != NULL);
493 } else {
494 /* Write Property Length */
495 DBG("MQTT - No properties to write\n");
496 PT_MQTT_WRITE_BYTE(conn, 0);
497 }
498
499 PT_END(pt);
500}
501#endif
502/*---------------------------------------------------------------------------*/
503static
504PT_THREAD(connect_pt(struct pt *pt, struct mqtt_connection *conn))
505{
506 PT_BEGIN(pt);
507
508#if MQTT_5
509 static struct mqtt_prop_list *will_props = MQTT_PROP_LIST_NONE;
510 if(conn->will.properties) {
511 will_props = (struct mqtt_prop_list *)list_head(conn->will.properties);
512 }
513#endif
514
515 DBG("MQTT - Sending CONNECT message...\n");
516
517 /* Set up FHDR */
518 conn->out_packet.fhdr = MQTT_FHDR_MSG_TYPE_CONNECT;
519 conn->out_packet.remaining_length = 0;
520 conn->out_packet.remaining_length += MQTT_CONNECT_VHDR_SIZE;
521 conn->out_packet.remaining_length += MQTT_STRING_LENGTH(&conn->client_id);
522#if (MQTT_PROTOCOL_VERSION > MQTT_PROTOCOL_VERSION_3_1) && MQTT_SRV_SUPPORTS_EMPTY_CLIENT_ID
523 /* Ensure we leave space for the 2 length bytes (which will encode 0) */
524 if(MQTT_STRING_LENGTH(&conn->client_id) == 0) {
525 conn->out_packet.remaining_length += 2;
526 }
527#endif
528 conn->out_packet.remaining_length += MQTT_STRING_LENGTH(&conn->credentials.username);
529 conn->out_packet.remaining_length += MQTT_STRING_LENGTH(&conn->credentials.password);
530 conn->out_packet.remaining_length += MQTT_STRING_LENGTH(&conn->will.topic);
531 conn->out_packet.remaining_length += MQTT_STRING_LENGTH(&conn->will.message);
532
533#if MQTT_5
534 /* For connect properties */
535 conn->out_packet.remaining_length +=
536 conn->out_props ? (conn->out_props->properties_len + conn->out_props->properties_len_enc_bytes)
537 : 1;
538
539 /* For will properties */
540 if(conn->connect_vhdr_flags & MQTT_VHDR_WILL_FLAG) {
541 conn->out_packet.remaining_length +=
542 will_props ? will_props->properties_len + will_props->properties_len_enc_bytes
543 : 1;
544 }
545#endif
546
547 mqtt_encode_var_byte_int(conn->out_packet.remaining_length_enc,
548 &conn->out_packet.remaining_length_enc_bytes,
549 conn->out_packet.remaining_length);
550 if(conn->out_packet.remaining_length_enc_bytes > 4) {
551 call_event(conn, MQTT_EVENT_PROTOCOL_ERROR, NULL);
552 PRINTF("MQTT - Error, remaining length > 4 bytes\n");
553 PT_EXIT(pt);
554 }
555
556 /* Write Fixed Header */
557 PT_MQTT_WRITE_BYTE(conn, conn->out_packet.fhdr);
558 PT_MQTT_WRITE_BYTES(conn,
559 conn->out_packet.remaining_length_enc,
560 conn->out_packet.remaining_length_enc_bytes);
561 PT_MQTT_WRITE_BYTE(conn, 0);
562 PT_MQTT_WRITE_BYTE(conn, strlen(MQTT_PROTOCOL_NAME));
563 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)MQTT_PROTOCOL_NAME, strlen(MQTT_PROTOCOL_NAME));
564 PT_MQTT_WRITE_BYTE(conn, MQTT_PROTOCOL_VERSION);
565 PT_MQTT_WRITE_BYTE(conn, conn->connect_vhdr_flags);
566 PT_MQTT_WRITE_BYTE(conn, (conn->keep_alive >> 8));
567 PT_MQTT_WRITE_BYTE(conn, (conn->keep_alive & 0x00FF));
568
569#if MQTT_5
570 /* Write Properties */
571 write_out_props(pt, conn, conn->out_props);
572#endif
573
574 /* Write Payload */
575 PT_MQTT_WRITE_BYTE(conn, conn->client_id.length >> 8);
576 PT_MQTT_WRITE_BYTE(conn, conn->client_id.length & 0x00FF);
577 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->client_id.string,
578 conn->client_id.length);
579
580 if(conn->connect_vhdr_flags & MQTT_VHDR_WILL_FLAG) {
581#if MQTT_5
582 /* Write Will Properties */
583 DBG("MQTT - Writing will properties\n");
584 write_out_props(pt, conn, will_props);
585#endif
586 PT_MQTT_WRITE_BYTE(conn, conn->will.topic.length >> 8);
587 PT_MQTT_WRITE_BYTE(conn, conn->will.topic.length & 0x00FF);
588 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->will.topic.string,
589 conn->will.topic.length);
590 PT_MQTT_WRITE_BYTE(conn, conn->will.message.length >> 8);
591 PT_MQTT_WRITE_BYTE(conn, conn->will.message.length & 0x00FF);
592 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->will.message.string,
593 conn->will.message.length);
594 DBG("MQTT - Setting will topic to '%s' %u bytes and message to '%s' %u bytes\n",
595 conn->will.topic.string,
596 conn->will.topic.length,
597 conn->will.message.string,
598 conn->will.message.length);
599 }
600 if(conn->connect_vhdr_flags & MQTT_VHDR_USERNAME_FLAG) {
601 PT_MQTT_WRITE_BYTE(conn, conn->credentials.username.length >> 8);
602 PT_MQTT_WRITE_BYTE(conn, conn->credentials.username.length & 0x00FF);
603 PT_MQTT_WRITE_BYTES(conn,
604 (uint8_t *)conn->credentials.username.string,
605 conn->credentials.username.length);
606 }
607 if(conn->connect_vhdr_flags & MQTT_VHDR_PASSWORD_FLAG) {
608 PT_MQTT_WRITE_BYTE(conn, conn->credentials.password.length >> 8);
609 PT_MQTT_WRITE_BYTE(conn, conn->credentials.password.length & 0x00FF);
610 PT_MQTT_WRITE_BYTES(conn,
611 (uint8_t *)conn->credentials.password.string,
612 conn->credentials.password.length);
613 }
614
615 /* Send out buffer */
616 send_out_buffer(conn);
617 conn->state = MQTT_CONN_STATE_CONNECTING_TO_BROKER;
618
619 timer_set(&conn->t, RESPONSE_WAIT_TIMEOUT);
620
621 /* Wait for CONNACK */
622 reset_packet(&conn->in_packet);
623 PT_WAIT_UNTIL(pt, conn->out_packet.qos_state == MQTT_QOS_STATE_GOT_ACK ||
624 timer_expired(&conn->t));
625 if(timer_expired(&conn->t)) {
626 DBG("Timeout waiting for CONNACK\n");
627 /* We stick to the letter of the spec here: Tear the connection down */
628#if MQTT_5
629 mqtt_disconnect(conn, MQTT_PROP_LIST_NONE);
630#else
631 mqtt_disconnect(conn);
632#endif
633 }
634 reset_packet(&conn->in_packet);
635
636 DBG("MQTT - Done sending CONNECT\n");
637
638#if DEBUG_MQTT == 1
639 DBG("MQTT - CONNECT message sent: \n");
640 uint16_t i;
641 for(i = 0; i < (conn->out_buffer_ptr - conn->out_buffer); i++) {
642 DBG("%02X ", conn->out_buffer[i]);
643 }
644 DBG("\n");
645#endif
646
647 PT_END(pt);
648}
649/*---------------------------------------------------------------------------*/
650static
651PT_THREAD(disconnect_pt(struct pt *pt, struct mqtt_connection *conn))
652{
653 PT_BEGIN(pt);
654
655 PT_MQTT_WRITE_BYTE(conn, MQTT_FHDR_MSG_TYPE_DISCONNECT);
656 PT_MQTT_WRITE_BYTE(conn, 0);
657
658#if MQTT_5
659/* Write Properties */
660 write_out_props(pt, conn, conn->out_props);
661#endif
662
663 send_out_buffer(conn);
664
665 /*
666 * Wait a couple of seconds for a TCP ACK. We don't really need the ACK,
667 * we do want the TCP/IP stack to actually send this disconnect before we
668 * tear down the session.
669 */
670 timer_set(&conn->t, (CLOCK_SECOND * 2));
671 PT_WAIT_UNTIL(pt, conn->out_buffer_sent || timer_expired(&conn->t));
672
673 PT_END(pt);
674}
675/*---------------------------------------------------------------------------*/
676static
677PT_THREAD(subscribe_pt(struct pt *pt, struct mqtt_connection *conn))
678{
679 PT_BEGIN(pt);
680
681 DBG("MQTT - Sending subscribe message! topic %s topic_length %i\n",
682 conn->out_packet.topic,
683 conn->out_packet.topic_length);
684 DBG("MQTT - Buffer space is %i \n",
685 &conn->out_buffer[MQTT_TCP_OUTPUT_BUFF_SIZE] - conn->out_buffer_ptr);
686
687 /* Set up FHDR */
688 conn->out_packet.fhdr = MQTT_FHDR_MSG_TYPE_SUBSCRIBE | MQTT_FHDR_QOS_LEVEL_1;
689 conn->out_packet.remaining_length = MQTT_MID_SIZE +
690 MQTT_STRING_LEN_SIZE +
691 conn->out_packet.topic_length +
692 MQTT_QOS_SIZE;
693
694#if MQTT_5
695 conn->out_packet.remaining_length +=
696 conn->out_props ? (conn->out_props->properties_len + conn->out_props->properties_len_enc_bytes)
697 : 1;
698#endif
699
700 mqtt_encode_var_byte_int(conn->out_packet.remaining_length_enc,
701 &conn->out_packet.remaining_length_enc_bytes,
702 conn->out_packet.remaining_length);
703 if(conn->out_packet.remaining_length_enc_bytes > 4) {
704 call_event(conn, MQTT_EVENT_PROTOCOL_ERROR, NULL);
705 PRINTF("MQTT - Error, remaining length > 4 bytes\n");
706 PT_EXIT(pt);
707 }
708
709 /* Write Fixed Header */
710 PT_MQTT_WRITE_BYTE(conn, conn->out_packet.fhdr);
711 PT_MQTT_WRITE_BYTES(conn,
712 conn->out_packet.remaining_length_enc,
713 conn->out_packet.remaining_length_enc_bytes);
714 /* Write Variable Header */
715 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.mid >> 8));
716 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.mid & 0x00FF));
717
718#if MQTT_5
719 /* Write Properties */
720 write_out_props(pt, conn, conn->out_props);
721#endif
722
723 /* Write Payload */
724 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.topic_length >> 8));
725 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.topic_length & 0x00FF));
726 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->out_packet.topic,
727 conn->out_packet.topic_length);
728
729#if MQTT_5
730 PT_MQTT_WRITE_BYTE(conn, conn->out_packet.sub_options);
731#else
732 PT_MQTT_WRITE_BYTE(conn, conn->out_packet.qos);
733#endif
734
735 /* Send out buffer */
736 send_out_buffer(conn);
737 timer_set(&conn->t, RESPONSE_WAIT_TIMEOUT);
738
739 /* Wait for SUBACK. */
740 reset_packet(&conn->in_packet);
741 PT_WAIT_UNTIL(pt, conn->out_packet.qos_state == MQTT_QOS_STATE_GOT_ACK ||
742 timer_expired(&conn->t));
743
744 if(timer_expired(&conn->t)) {
745 DBG("Timeout waiting for SUBACK\n");
746 }
747 reset_packet(&conn->in_packet);
748
749 /* This is clear after the entire transaction is complete */
750 conn->out_queue_full = 0;
751
752 DBG("MQTT - Done in send_subscribe!\n");
753
754 PT_END(pt);
755}
756/*---------------------------------------------------------------------------*/
757static
758PT_THREAD(unsubscribe_pt(struct pt *pt, struct mqtt_connection *conn))
759{
760 PT_BEGIN(pt);
761
762 DBG("MQTT - Sending unsubscribe message on topic %s topic_length %i\n",
763 conn->out_packet.topic,
764 conn->out_packet.topic_length);
765 DBG("MQTT - Buffer space is %i \n",
766 &conn->out_buffer[MQTT_TCP_OUTPUT_BUFF_SIZE] - conn->out_buffer_ptr);
767
768 /* Set up FHDR */
769 conn->out_packet.fhdr = MQTT_FHDR_MSG_TYPE_UNSUBSCRIBE |
770 MQTT_FHDR_QOS_LEVEL_1;
771 conn->out_packet.remaining_length = MQTT_MID_SIZE +
772 MQTT_STRING_LEN_SIZE +
773 conn->out_packet.topic_length;
774
775#if MQTT_5
776 conn->out_packet.remaining_length +=
777 conn->out_props ? (conn->out_props->properties_len + conn->out_props->properties_len_enc_bytes)
778 : 1;
779#endif
780
781 mqtt_encode_var_byte_int(conn->out_packet.remaining_length_enc,
782 &conn->out_packet.remaining_length_enc_bytes,
783 conn->out_packet.remaining_length);
784 if(conn->out_packet.remaining_length_enc_bytes > 4) {
785 call_event(conn, MQTT_EVENT_PROTOCOL_ERROR, NULL);
786 PRINTF("MQTT - Error, remaining length > 4 bytes\n");
787 PT_EXIT(pt);
788 }
789
790 /* Write Fixed Header */
791 PT_MQTT_WRITE_BYTE(conn, conn->out_packet.fhdr);
792 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->out_packet.remaining_length_enc,
793 conn->out_packet.remaining_length_enc_bytes);
794
795 /* Write Variable Header */
796 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.mid >> 8));
797 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.mid & 0x00FF));
798#if MQTT_5
799 /* Write Properties */
800 write_out_props(pt, conn, conn->out_props);
801#endif
802
803 /* Write Payload */
804 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.topic_length >> 8));
805 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.topic_length & 0x00FF));
806 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->out_packet.topic,
807 conn->out_packet.topic_length);
808
809 /* Send out buffer */
810 send_out_buffer(conn);
811 timer_set(&conn->t, RESPONSE_WAIT_TIMEOUT);
812
813 /* Wait for UNSUBACK */
814 reset_packet(&conn->in_packet);
815 PT_WAIT_UNTIL(pt, conn->out_packet.qos_state == MQTT_QOS_STATE_GOT_ACK ||
816 timer_expired(&conn->t));
817
818 if(timer_expired(&conn->t)) {
819 DBG("Timeout waiting for UNSUBACK\n");
820 }
821
822 reset_packet(&conn->in_packet);
823
824 /* This is clear after the entire transaction is complete */
825 conn->out_queue_full = 0;
826
827 DBG("MQTT - Done writing subscribe message to out buffer!\n");
828
829 PT_END(pt);
830}
831/*---------------------------------------------------------------------------*/
832static
833PT_THREAD(publish_pt(struct pt *pt, struct mqtt_connection *conn))
834{
835 PT_BEGIN(pt);
836
837 DBG("MQTT - Sending publish message! topic %s topic_length %i\n",
838 conn->out_packet.topic,
839 conn->out_packet.topic_length);
840 DBG("MQTT - Buffer space is %i \n",
841 &conn->out_buffer[MQTT_TCP_OUTPUT_BUFF_SIZE] - conn->out_buffer_ptr);
842
843 /* Set up FHDR */
844 conn->out_packet.fhdr = MQTT_FHDR_MSG_TYPE_PUBLISH |
845 conn->out_packet.qos << 1;
846 if(conn->out_packet.retain == MQTT_RETAIN_ON) {
847 conn->out_packet.fhdr |= MQTT_FHDR_RETAIN_FLAG;
848 }
849 conn->out_packet.remaining_length = MQTT_STRING_LEN_SIZE +
850 conn->out_packet.topic_length +
851 conn->out_packet.payload_size;
852 if(conn->out_packet.qos > MQTT_QOS_LEVEL_0) {
853 conn->out_packet.remaining_length += MQTT_MID_SIZE;
854 }
855
856#if MQTT_5
857 conn->out_packet.remaining_length +=
858 conn->out_props ? (conn->out_props->properties_len + conn->out_props->properties_len_enc_bytes)
859 : 1;
860#endif
861
862 mqtt_encode_var_byte_int(conn->out_packet.remaining_length_enc,
863 &conn->out_packet.remaining_length_enc_bytes,
864 conn->out_packet.remaining_length);
865 if(conn->out_packet.remaining_length_enc_bytes > 4) {
866 call_event(conn, MQTT_EVENT_PROTOCOL_ERROR, NULL);
867 PRINTF("MQTT - Error, remaining length > 4 bytes\n");
868 PT_EXIT(pt);
869 }
870
871 /* The DUP flag MUST be set to 0 for all QoS 0 messages */
872 if(conn->out_packet.qos == MQTT_QOS_LEVEL_0) {
873 conn->out_packet.fhdr &= ~MQTT_FHDR_DUP_FLAG;
874 }
875
876 /* Write Fixed Header */
877 PT_MQTT_WRITE_BYTE(conn, conn->out_packet.fhdr);
878 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->out_packet.remaining_length_enc,
879 conn->out_packet.remaining_length_enc_bytes);
880 /* Write Variable Header */
881 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.topic_length >> 8));
882 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.topic_length & 0x00FF));
883 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->out_packet.topic,
884 conn->out_packet.topic_length);
885 if(conn->out_packet.qos > MQTT_QOS_LEVEL_0) {
886 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.mid >> 8));
887 PT_MQTT_WRITE_BYTE(conn, (conn->out_packet.mid & 0x00FF));
888 }
889
890#if MQTT_5
891 /* Write Properties */
892 write_out_props(pt, conn, conn->out_props);
893#endif
894
895 /* Write Payload */
896 PT_MQTT_WRITE_BYTES(conn,
897 conn->out_packet.payload,
898 conn->out_packet.payload_size);
899
900 send_out_buffer(conn);
901 timer_set(&conn->t, RESPONSE_WAIT_TIMEOUT);
902
903 /*
904 * If QoS is zero then wait until the message has been sent, since there is
905 * no ACK to wait for.
906 *
907 * Also notify the app will not be notified via PUBACK or PUBCOMP
908 */
909 if(conn->out_packet.qos == 0) {
910 process_post(conn->app_process, mqtt_update_event, NULL);
911 } else if(conn->out_packet.qos == 1) {
912 /* Wait for PUBACK */
913 reset_packet(&conn->in_packet);
914 PT_WAIT_UNTIL(pt, conn->out_packet.qos_state == MQTT_QOS_STATE_GOT_ACK ||
915 timer_expired(&conn->t));
916 if(timer_expired(&conn->t)) {
917 DBG("Timeout waiting for PUBACK\n");
918 }
919 if(conn->in_packet.mid != conn->out_packet.mid) {
920 DBG("MQTT - Warning, got PUBACK with none matching MID. Currently there "
921 "is no support for several concurrent PUBLISH messages.\n");
922 }
923 } else if(conn->out_packet.qos == 2) {
924 DBG("MQTT - QoS not implemented yet.\n");
925 /* Should wait for PUBREC, send PUBREL and then wait for PUBCOMP */
926 }
927
928 reset_packet(&conn->in_packet);
929
930 /* This is clear after the entire transaction is complete */
931 conn->out_queue_full = 0;
932
933 DBG("MQTT - Publish Enqueued\n");
934
935 PT_END(pt);
936}
937/*---------------------------------------------------------------------------*/
938static
939PT_THREAD(pingreq_pt(struct pt *pt, struct mqtt_connection *conn))
940{
941 PT_BEGIN(pt);
942
943 DBG("MQTT - Sending PINGREQ\n");
944
945 /* Write Fixed Header */
946 PT_MQTT_WRITE_BYTE(conn, MQTT_FHDR_MSG_TYPE_PINGREQ);
947 PT_MQTT_WRITE_BYTE(conn, 0);
948
949 send_out_buffer(conn);
950
951 /* Start timeout for reply. */
952 conn->waiting_for_pingresp = 1;
953
954 /* Wait for PINGRESP or timeout */
955 reset_packet(&conn->in_packet);
956 timer_set(&conn->t, RESPONSE_WAIT_TIMEOUT);
957
958 PT_WAIT_UNTIL(pt, conn->in_packet.packet_received || timer_expired(&conn->t));
959
960 reset_packet(&conn->in_packet);
961
962 conn->waiting_for_pingresp = 0;
963
964 PT_END(pt);
965}
966/*---------------------------------------------------------------------------*/
967#if MQTT_5
968static
969PT_THREAD(auth_pt(struct pt *pt, struct mqtt_connection *conn))
970{
971 PT_BEGIN(pt);
972
973 conn->out_packet.remaining_length +=
974 conn->out_props ? (conn->out_props->properties_len + conn->out_props->properties_len_enc_bytes)
975 : 1;
976
977 mqtt_encode_var_byte_int(conn->out_packet.remaining_length_enc,
978 &conn->out_packet.remaining_length_enc_bytes,
979 conn->out_packet.remaining_length);
980
981 if(conn->out_packet.remaining_length_enc_bytes > 4) {
982 call_event(conn, MQTT_EVENT_PROTOCOL_ERROR, NULL);
983 PRINTF("MQTT - Error, remaining length > 4 bytes\n");
984 PT_EXIT(pt);
985 }
986
987 /* Write Fixed Header */
988 PT_MQTT_WRITE_BYTE(conn, conn->out_packet.fhdr);
989 PT_MQTT_WRITE_BYTES(conn, (uint8_t *)conn->out_packet.remaining_length_enc,
990 conn->out_packet.remaining_length_enc_bytes);
991
992 /* Write Variable Header */
993 PT_MQTT_WRITE_BYTE(conn, conn->out_packet.auth_reason_code);
994
995 /* Write Properties */
996 write_out_props(pt, conn, conn->out_props);
997
998 /* No Payload */
999 send_out_buffer(conn);
1000
1001 PT_WAIT_UNTIL(pt, conn->out_buffer_sent);
1002
1003 PT_END(pt);
1004}
1005#endif
1006/*---------------------------------------------------------------------------*/
1007static void
1008handle_connack(struct mqtt_connection *conn)
1009{
1010 struct mqtt_connack_event connack_event;
1011
1012 DBG("MQTT - Got CONNACK\n");
1013
1014#if MQTT_PROTOCOL_VERSION <= MQTT_PROTOCOL_VERSION_3_1_1
1015 if(conn->in_packet.remaining_length != 2) {
1016 PRINTF("MQTT - CONNACK VHDR remaining length %i incorrect\n",
1017 conn->in_packet.remaining_length);
1018 call_event(conn,
1019 MQTT_EVENT_ERROR,
1020 NULL);
1021 abort_connection(conn);
1022 return;
1023 }
1024
1025 if(conn->in_packet.payload[1] != 0) {
1026 PRINTF("MQTT - Connection refused with Return Code %i\n",
1027 conn->in_packet.payload[1]);
1028 call_event(conn,
1029 MQTT_EVENT_CONNECTION_REFUSED_ERROR,
1030 &conn->in_packet.payload[1]);
1031 abort_connection(conn);
1032 return;
1033 }
1034#endif
1035
1036#if MQTT_PROTOCOL_VERSION >= MQTT_PROTOCOL_VERSION_5
1037 /* The CONNACK VHDR must contain:
1038 * 0: Connect Acknowledge Flags
1039 * 1: Connect Reason Code
1040 * 2: Properties (whose Length field must be set even if no properties are present)
1041 */
1042 if(conn->in_packet.remaining_length < 3) {
1043 PRINTF("MQTT - CONNACK VHDR remaining length %i incorrect\n",
1044 conn->in_packet.remaining_length);
1045 call_event(conn,
1046 MQTT_EVENT_ERROR,
1047 NULL);
1048 abort_connection(conn);
1049 return;
1050 }
1051#endif
1052
1053 conn->out_packet.qos_state = MQTT_QOS_STATE_GOT_ACK;
1054
1055#if MQTT_PROTOCOL_VERSION >= MQTT_PROTOCOL_VERSION_3_1_1
1056 connack_event.session_present = conn->in_packet.payload[0] & MQTT_VHDR_CONNACK_SESSION_PRESENT;
1057#endif
1058
1059#if MQTT_PROTOCOL_VERSION >= MQTT_PROTOCOL_VERSION_5
1060 mqtt_prop_parse_connack_props(conn);
1061#endif
1062
1063 ctimer_set(&conn->keep_alive_timer, conn->keep_alive * CLOCK_SECOND,
1064 keep_alive_callback, conn);
1065
1066 /* Always reset packet before callback since it might be used directly */
1067 conn->state = MQTT_CONN_STATE_CONNECTED_TO_BROKER;
1068 call_event(conn, MQTT_EVENT_CONNECTED, &connack_event);
1069}
1070/*---------------------------------------------------------------------------*/
1071static void
1072handle_pingresp(struct mqtt_connection *conn)
1073{
1074 DBG("MQTT - Got PINGRESP\n");
1075}
1076/*---------------------------------------------------------------------------*/
1077static void
1078handle_suback(struct mqtt_connection *conn)
1079{
1080 struct mqtt_suback_event suback_event;
1081
1082 DBG("MQTT - Got SUBACK\n");
1083
1084 /* Only accept SUBACKS with X topic QoS response, assume 1 */
1085#if MQTT_5
1086 if(conn->in_packet.remaining_length > MQTT_MID_SIZE +
1087 MQTT_MAX_TOPICS_PER_SUBSCRIBE * MQTT_QOS_SIZE +
1088 conn->in_packet.properties_len + conn->in_packet.properties_enc_len) {
1089#else
1090 if(conn->in_packet.remaining_length > MQTT_MID_SIZE +
1091 MQTT_MAX_TOPICS_PER_SUBSCRIBE * MQTT_QOS_SIZE) {
1092#endif
1093 DBG("MQTT - Error, SUBACK with > 1 topic, not supported.\n");
1094 }
1095
1096 conn->out_packet.qos_state = MQTT_QOS_STATE_GOT_ACK;
1097
1098 suback_event.mid = conn->in_packet.mid;
1099
1100#if !MQTT_31
1101 suback_event.success = 0;
1102
1103 switch(conn->in_packet.payload_start[0]) {
1104 case MQTT_SUBACK_RET_FAIL:
1105 PRINTF("MQTT - Error, SUBSCRIBE failed with SUBACK return code '%x'", conn->in_packet.payload_start[0]);
1106 break;
1107
1108 case MQTT_SUBACK_RET_QOS_0:
1109 case MQTT_SUBACK_RET_QOS_1:
1110 case MQTT_SUBACK_RET_QOS_2:
1111 suback_event.qos_level = conn->in_packet.payload_start[0] & 0x03;
1112 suback_event.success = 1;
1113 break;
1114
1115 default:
1116 PRINTF("MQTT - Error, Unrecognised SUBACK return code '%x'", conn->in_packet.payload_start[0]);
1117 break;
1118 }
1119
1120 suback_event.return_code = conn->in_packet.payload_start[0];
1121#else
1122 suback_event.qos_level = conn->in_packet.payload_start[0];
1123#endif
1124
1125 if(conn->in_packet.mid != conn->out_packet.mid) {
1126 DBG("MQTT - Warning, got SUBACK with none matching MID. Currently there is"
1127 "no support for several concurrent SUBSCRIBE messages.\n");
1128 }
1129
1130 /* Always reset packet before callback since it might be used directly */
1131 call_event(conn, MQTT_EVENT_SUBACK, &suback_event);
1132}
1133/*---------------------------------------------------------------------------*/
1134static void
1135handle_unsuback(struct mqtt_connection *conn)
1136{
1137 DBG("MQTT - Got UNSUBACK\n");
1138
1139 conn->out_packet.qos_state = MQTT_QOS_STATE_GOT_ACK;
1140
1141 if(conn->in_packet.mid != conn->out_packet.mid) {
1142 DBG("MQTT - Warning, got UNSUBACK with none matching MID. Currently there is"
1143 "no support for several concurrent UNSUBSCRIBE messages.\n");
1144 }
1145
1146 call_event(conn, MQTT_EVENT_UNSUBACK, &conn->in_packet.mid);
1147}
1148/*---------------------------------------------------------------------------*/
1149static void
1150handle_puback(struct mqtt_connection *conn)
1151{
1152 DBG("MQTT - Got PUBACK\n");
1153
1154 conn->out_packet.qos_state = MQTT_QOS_STATE_GOT_ACK;
1155
1156 call_event(conn, MQTT_EVENT_PUBACK, &conn->in_packet.mid);
1157}
1158/*---------------------------------------------------------------------------*/
1159static mqtt_pub_status_t
1160handle_publish(struct mqtt_connection *conn)
1161{
1162 DBG("MQTT - Got PUBLISH, called once per manageable chunk of message.\n");
1163 DBG("MQTT - Handling publish on topic '%s'\n", conn->in_publish_msg.topic);
1164
1165#if MQTT_PROTOCOL_VERSION >= MQTT_PROTOCOL_VERSION_3_1_1
1166 if(strlen(conn->in_publish_msg.topic) < conn->in_packet.topic_len) {
1167 DBG("NULL detected in received PUBLISH topic\n");
1168#if MQTT_5
1169 mqtt_disconnect(conn, MQTT_PROP_LIST_NONE);
1170#else
1171 mqtt_disconnect(conn);
1172#endif
1173 return MQTT_PUBLISH_ERR;
1174 }
1175#endif
1176
1177 DBG("MQTT - This chunk is %i bytes\n", conn->in_publish_msg.payload_chunk_length);
1178
1179 if(((conn->in_packet.fhdr & 0x09) >> 1) != 0) {
1180 PRINTF("MQTT - Error, got incoming PUBLISH with QoS > 0, not supported atm!\n");
1181 }
1182
1183 call_event(conn, MQTT_EVENT_PUBLISH, &conn->in_publish_msg);
1184
1185 if(conn->in_publish_msg.first_chunk == 1) {
1186 conn->in_publish_msg.first_chunk = 0;
1187 }
1188
1189 /* If this is the last time handle_publish will be called, reset packet. */
1190 if(conn->in_publish_msg.payload_left == 0) {
1191
1192 /* Check for QoS and initiate the reply, do not rely on the data in the
1193 * in_packet being untouched. */
1194
1195 DBG("MQTT - (handle_publish) resetting packet.\n");
1196 reset_packet(&conn->in_packet);
1197 }
1198
1199 return MQTT_PUBLISH_OK;
1200}
1201/*---------------------------------------------------------------------------*/
1202static int
1203parse_publish_vhdr(struct mqtt_connection *conn,
1204 uint32_t *pos,
1205 const uint8_t *input_data_ptr,
1206 int input_data_len)
1207{
1208 uint16_t copy_bytes;
1209
1210 /* Read out topic length */
1211 if(conn->in_packet.topic_len_received == 0) {
1212 if(!conn->in_packet.topic_len_msb_received) {
1213 conn->in_packet.topic_pos = 0;
1214 conn->in_packet.topic_len = (input_data_ptr[(*pos)++] << 8);
1215 conn->in_packet.byte_counter++;
1216 conn->in_packet.topic_len_msb_received = 1;
1217 if(*pos >= input_data_len) {
1218 return 0;
1219 }
1220 }
1221 conn->in_packet.topic_len |= input_data_ptr[(*pos)++];
1222 conn->in_packet.byte_counter++;
1223 conn->in_packet.topic_len_received = 1;
1224 /* Reject malformed packets where the advertised topic length does
1225 * not leave room for the topic itself inside remaining_length. */
1226 if((uint32_t)conn->in_packet.topic_len + 2 >
1227 conn->in_packet.remaining_length) {
1228 PRINTF("MQTT - PUBLISH topic_len %u exceeds remaining_length %u\n",
1229 conn->in_packet.topic_len, conn->in_packet.remaining_length);
1230 call_event(conn, MQTT_EVENT_ERROR, NULL);
1231 abort_connection(conn);
1232 return -1;
1233 }
1234 /* Abort if topic is longer than our topic buffer */
1235 if(conn->in_packet.topic_len > MQTT_MAX_TOPIC_LENGTH) {
1236 PRINTF("MQTT - PUBLISH topic too long %u/%u, aborting\n",
1237 conn->in_packet.topic_len, MQTT_MAX_TOPIC_LENGTH);
1238 call_event(conn, MQTT_EVENT_ERROR, NULL);
1239 abort_connection(conn);
1240 return -1;
1241 }
1242 DBG("MQTT - Read PUBLISH topic len %i\n", conn->in_packet.topic_len);
1243 }
1244
1245 /* Read out topic */
1246 if(conn->in_packet.topic_len_received == 1 &&
1247 conn->in_packet.topic_received == 0) {
1248 copy_bytes = MIN(conn->in_packet.topic_len - conn->in_packet.topic_pos,
1249 input_data_len - *pos);
1250 DBG("MQTT - topic_pos: %i copy_bytes: %i\n", conn->in_packet.topic_pos,
1251 copy_bytes);
1252 memcpy(&conn->in_publish_msg.topic[conn->in_packet.topic_pos],
1253 &input_data_ptr[*pos],
1254 copy_bytes);
1255 (*pos) += copy_bytes;
1256 conn->in_packet.byte_counter += copy_bytes;
1257 conn->in_packet.topic_pos += copy_bytes;
1258
1259 if(conn->in_packet.topic_len - conn->in_packet.topic_pos == 0) {
1260 DBG("MQTT - Got topic '%s'", conn->in_publish_msg.topic);
1261 conn->in_packet.topic_received = 1;
1262 conn->in_publish_msg.topic[conn->in_packet.topic_pos] = '\0';
1263 conn->in_publish_msg.payload_length =
1264 conn->in_packet.remaining_length - conn->in_packet.topic_len - 2;
1265 conn->in_publish_msg.payload_left = conn->in_publish_msg.payload_length;
1266 }
1267
1268 /* Set this once per incomming publish message */
1269 conn->in_publish_msg.first_chunk = 1;
1270 }
1271
1272 return 0;
1273}
1274/*---------------------------------------------------------------------------*/
1275/* MQTTv5 only */
1276#if MQTT_5
1277static void
1278handle_disconnect(struct mqtt_connection *conn)
1279{
1280 DBG("MQTT - (handle_disconnect) Got DISCONNECT.\n");
1281 call_event(conn, MQTT_EVENT_DISCONNECTED, NULL);
1282 abort_connection(conn);
1283}
1284/*---------------------------------------------------------------------------*/
1285static void
1286handle_auth(struct mqtt_connection *conn)
1287{
1288 struct mqtt_prop_auth_event event;
1289
1290 DBG("MQTT - (handle_auth) Got AUTH.\n");
1291
1292 if((conn->in_packet.fhdr & 0x0F) != 0x0) {
1293 call_event(conn,
1294 MQTT_EVENT_ERROR,
1295 NULL);
1296 abort_connection(conn);
1297 return;
1298 }
1299
1300 /* AUTH messages from the server */
1301 if(conn->state == MQTT_CONN_STATE_CONNECTING_TO_BROKER &&
1302 (!conn->in_packet.has_reason_code ||
1303 conn->in_packet.reason_code != MQTT_VHDR_RC_CONTINUE_AUTH)) {
1304 DBG("MQTT - (handle_auth) Not reauth - Reason Code 0x18 expected!\n");
1305 }
1306
1307 mqtt_prop_parse_auth_props(conn, &event);
1308 call_event(conn, MQTT_EVENT_AUTH, &event);
1309}
1310#endif
1311/*---------------------------------------------------------------------------*/
1312static void
1313parse_vhdr(struct mqtt_connection *conn)
1314{
1315 conn->in_packet.payload_start = conn->in_packet.payload;
1316
1317 /* Some message types include a packet identifier */
1318 switch(conn->in_packet.fhdr & 0xF0) {
1319 case MQTT_FHDR_MSG_TYPE_PUBACK:
1320 case MQTT_FHDR_MSG_TYPE_SUBACK:
1321 case MQTT_FHDR_MSG_TYPE_UNSUBACK:
1322 conn->in_packet.mid = (conn->in_packet.payload[0] << 8) |
1323 (conn->in_packet.payload[1]);
1324 conn->in_packet.payload_start += 2;
1325 break;
1326
1327 /* Other message types have a 0-length VHDR */
1328 /* PUBLISH has a VHDR for QoS > 0, which is currently unsupported */
1329 default:
1330 break;
1331 }
1332
1333#if MQTT_5
1334 /* CONNACK, PUBACK, PUBREC, PUBREL, PUBCOMP, DISCONNECT and AUTH have a single
1335 * Reason Code as part of the Variable Header.
1336 * SUBACK and UNSUBACK contain a list of one or more Reason Codes in the Payload.
1337 */
1338 switch(conn->in_packet.fhdr & 0xF0) {
1339 case MQTT_FHDR_MSG_TYPE_CONNACK:
1340 case MQTT_FHDR_MSG_TYPE_PUBACK:
1341 case MQTT_FHDR_MSG_TYPE_PUBREC:
1342 case MQTT_FHDR_MSG_TYPE_PUBREL:
1343 case MQTT_FHDR_MSG_TYPE_PUBCOMP:
1344 case MQTT_FHDR_MSG_TYPE_DISCONNECT:
1345 case MQTT_FHDR_MSG_TYPE_AUTH:
1346 conn->in_packet.reason_code = conn->in_packet.payload_start[0];
1347 conn->in_packet.has_reason_code = 1;
1348 conn->in_packet.payload_start += 1;
1349 break;
1350
1351 default:
1352 conn->in_packet.has_reason_code = 0;
1353 break;
1354 }
1355
1356 if(!conn->in_packet.has_props) {
1357 mqtt_prop_decode_input_props(conn);
1358 }
1359#endif
1360}
1361/*---------------------------------------------------------------------------*/
1362#if MQTT_5
1363static int
1364trim_publish_props(struct mqtt_connection *conn)
1365{
1366 uint32_t prop_total = (uint32_t)conn->in_packet.properties_len +
1367 conn->in_packet.properties_enc_len;
1368
1369 if(prop_total > conn->in_publish_msg.payload_chunk_length) {
1370 PRINTF("MQTT - Error, PUBLISH properties length exceeds buffered chunk\n");
1371 call_event(conn, MQTT_EVENT_ERROR, NULL);
1372 abort_connection(conn);
1373 return -1;
1374 }
1375
1376 conn->in_publish_msg.payload_chunk_length -= prop_total;
1377 conn->in_publish_msg.payload_chunk += prop_total;
1378 return 0;
1379}
1380#endif
1381/*---------------------------------------------------------------------------*/
1382static int
1383tcp_input(struct tcp_socket *s,
1384 void *ptr,
1385 const uint8_t *input_data_ptr,
1386 int input_data_len)
1387{
1388 struct mqtt_connection *conn = ptr;
1389 uint32_t pos = 0;
1390 uint32_t copy_bytes = 0;
1391 mqtt_pub_status_t pub_status;
1392 uint8_t remaining_length_bytes;
1393
1394 if(input_data_len == 0) {
1395 return 0;
1396 }
1397
1398 DBG("tcp_input with %i bytes of data:\n", input_data_len);
1399
1400parse_next:
1401 if(conn->in_packet.packet_received) {
1402 reset_packet(&conn->in_packet);
1403 }
1404
1405 /* Read the fixed header field, if we do not have it */
1406 if(!conn->in_packet.fhdr) {
1407 conn->in_packet.fhdr = input_data_ptr[pos++];
1408 conn->in_packet.byte_counter++;
1409
1410 DBG("MQTT - Read VHDR '%02X'\n", conn->in_packet.fhdr);
1411
1412 if(pos >= input_data_len) {
1413 return 0;
1414 }
1415 }
1416
1417 /*
1418 * Read the Remaining Length field, if we do not have it.
1419 * byte_counter intentionally does not include the VBI bytes: the inner
1420 * read-loop target is MQTT_FHDR_SIZE + remaining_length, which only
1421 * accounts for the message-type byte and the remaining payload.
1422 */
1423 if(!conn->in_packet.has_remaining_length) {
1424 remaining_length_bytes =
1425 mqtt_decode_var_byte_int(input_data_ptr, input_data_len, &pos,
1426 NULL,
1427 &conn->in_packet.remaining_length);
1428
1429 if(remaining_length_bytes == 0) {
1430 call_event(conn, MQTT_EVENT_ERROR, NULL);
1431 return 0;
1432 }
1433
1434 DBG("MQTT - Finished reading remaining length byte\n");
1435 conn->in_packet.has_remaining_length = 1;
1436 }
1437
1438 /*
1439 * Check for unsupported payload length. Will read all incoming data from the
1440 * server in any case and then reset the packet.
1441 *
1442 * TODO: Decide if we, for example, want to disconnect instead.
1443 */
1444 if((conn->in_packet.remaining_length > MQTT_INPUT_BUFF_SIZE) &&
1445 (conn->in_packet.fhdr & 0xF0) != MQTT_FHDR_MSG_TYPE_PUBLISH) {
1446 uint32_t pkt_total = MQTT_FHDR_SIZE + conn->in_packet.remaining_length;
1447 uint32_t drain = MIN((uint32_t)input_data_len - pos,
1448 pkt_total - conn->in_packet.byte_counter);
1449
1450 PRINTF("MQTT - Error, unsupported payload size for non-PUBLISH message\n");
1451
1452 pos += drain;
1453 conn->in_packet.byte_counter += drain;
1454 if(conn->in_packet.byte_counter >= pkt_total) {
1455 conn->in_packet.packet_received = 1;
1456 if(pos < input_data_len) {
1457 goto parse_next;
1458 }
1459 }
1460 return 0;
1461 }
1462
1463 /*
1464 * Supported payload, reads out both VHDR and Payload of all packets.
1465 *
1466 * Note: There will always be at least one byte left to read when we enter
1467 * this loop.
1468 */
1469 while(conn->in_packet.byte_counter <
1470 (MQTT_FHDR_SIZE + conn->in_packet.remaining_length)) {
1471
1472 if((conn->in_packet.fhdr & 0xF0) == MQTT_FHDR_MSG_TYPE_PUBLISH &&
1473 conn->in_packet.topic_received == 0) {
1474 if(parse_publish_vhdr(conn, &pos, input_data_ptr, input_data_len) < 0) {
1475 return 0;
1476 }
1477 }
1478
1479 /*
1480 * Read in as much as we can into the packet payload, but not past
1481 * the end of the current MQTT packet on the wire. Without this last
1482 * bound, a TCP segment that spans the end of one packet and the
1483 * start of the next would slurp the next packet's bytes into
1484 * in_packet.payload[].
1485 */
1486 copy_bytes = MIN(input_data_len - pos,
1487 MQTT_INPUT_BUFF_SIZE - conn->in_packet.payload_pos);
1488 copy_bytes = MIN(copy_bytes,
1489 (MQTT_FHDR_SIZE + conn->in_packet.remaining_length) -
1490 conn->in_packet.byte_counter);
1491 DBG("- Copied %i payload bytes\n", copy_bytes);
1492 memcpy(&conn->in_packet.payload[conn->in_packet.payload_pos],
1493 &input_data_ptr[pos],
1494 copy_bytes);
1495 conn->in_packet.byte_counter += copy_bytes;
1496 conn->in_packet.payload_pos += copy_bytes;
1497 pos += copy_bytes;
1498
1499#if DEBUG_MQTT == 1
1500 uint32_t i;
1501 DBG("MQTT - Copied bytes: \n");
1502 for(i = 0; i < copy_bytes; i++) {
1503 DBG("%02X ", conn->in_packet.payload[i]);
1504 }
1505 DBG("\n");
1506#endif
1507
1508 /* Full buffer, shall only happen to PUBLISH messages. */
1509 if(MQTT_INPUT_BUFF_SIZE - conn->in_packet.payload_pos == 0) {
1510 conn->in_publish_msg.payload_chunk = conn->in_packet.payload;
1511 conn->in_publish_msg.payload_chunk_length = MQTT_INPUT_BUFF_SIZE;
1512 conn->in_publish_msg.payload_left -= MQTT_INPUT_BUFF_SIZE;
1513
1514#if MQTT_5
1515 if(!conn->in_packet.has_props) {
1516 mqtt_prop_decode_input_props(conn);
1517 }
1518
1519 if(conn->in_publish_msg.first_chunk) {
1520 if(trim_publish_props(conn) < 0) {
1521 return 0;
1522 }
1523 }
1524#endif
1525
1526 pub_status = handle_publish(conn);
1527
1528 conn->in_publish_msg.payload_chunk = conn->in_packet.payload;
1529 conn->in_packet.payload_pos = 0;
1530
1531 if(pub_status != MQTT_PUBLISH_OK) {
1532 return 0;
1533 }
1534 }
1535
1536 if(pos >= input_data_len &&
1537 (conn->in_packet.byte_counter < (MQTT_FHDR_SIZE + conn->in_packet.remaining_length))) {
1538 return 0;
1539 }
1540 }
1541
1542 parse_vhdr(conn);
1543
1544 /* Debug information */
1545 DBG("\n");
1546 /* Take care of input */
1547 DBG("MQTT - Finished reading packet!\n");
1548 /* What to return? */
1549 DBG("MQTT - total data was %i bytes of data. \n",
1550 (MQTT_FHDR_SIZE + conn->in_packet.remaining_length));
1551
1552#if MQTT_5
1553 if(conn->in_packet.has_reason_code &&
1554 conn->in_packet.reason_code >= MQTT_VHDR_RC_UNSPEC_ERR) {
1555 PRINTF("MQTT - Reason Code indicated error %i\n",
1556 conn->in_packet.reason_code);
1557 call_event(conn,
1558 MQTT_EVENT_ERROR,
1559 NULL);
1560 abort_connection(conn);
1561 return 0;
1562 }
1563#endif
1564
1565 /* Handle packet here. */
1566 switch(conn->in_packet.fhdr & 0xF0) {
1567 case MQTT_FHDR_MSG_TYPE_CONNACK:
1568 handle_connack(conn);
1569 break;
1570 case MQTT_FHDR_MSG_TYPE_PUBLISH:
1571 /* This is the only or the last chunk of publish payload */
1572 conn->in_publish_msg.payload_chunk = conn->in_packet.payload;
1573 conn->in_publish_msg.payload_chunk_length = conn->in_packet.payload_pos;
1574 conn->in_publish_msg.payload_left = 0;
1575
1576 DBG("MQTT - First chunk? %i\n", conn->in_publish_msg.first_chunk);
1577#if MQTT_5
1578 if(conn->in_publish_msg.first_chunk) {
1579 if(trim_publish_props(conn) < 0) {
1580 return 0;
1581 }
1582 }
1583#endif
1584 (void)handle_publish(conn);
1585 break;
1586 case MQTT_FHDR_MSG_TYPE_PUBACK:
1587 handle_puback(conn);
1588 break;
1589 case MQTT_FHDR_MSG_TYPE_SUBACK:
1590 handle_suback(conn);
1591 break;
1592 case MQTT_FHDR_MSG_TYPE_UNSUBACK:
1593 handle_unsuback(conn);
1594 break;
1595 case MQTT_FHDR_MSG_TYPE_PINGRESP:
1596 handle_pingresp(conn);
1597 break;
1598
1599 /* QoS 2 not implemented yet */
1600 case MQTT_FHDR_MSG_TYPE_PUBREC:
1601 case MQTT_FHDR_MSG_TYPE_PUBREL:
1602 case MQTT_FHDR_MSG_TYPE_PUBCOMP:
1603 call_event(conn, MQTT_EVENT_NOT_IMPLEMENTED_ERROR, NULL);
1604 PRINTF("MQTT - Got unhandled MQTT Message Type '%i'",
1605 (conn->in_packet.fhdr & 0xF0));
1606 break;
1607
1608#if MQTT_PROTOCOL_VERSION >= MQTT_PROTOCOL_VERSION_5
1609 case MQTT_FHDR_MSG_TYPE_DISCONNECT:
1610 handle_disconnect(conn);
1611 break;
1612
1613 case MQTT_FHDR_MSG_TYPE_AUTH:
1614 handle_auth(conn);
1615 break;
1616#endif
1617
1618 default:
1619 /* All server-only message */
1620 PRINTF("MQTT - Got MQTT Message Type '%i'", (conn->in_packet.fhdr & 0xF0));
1621 break;
1622 }
1623
1624 conn->in_packet.packet_received = 1;
1625
1626 /*
1627 * A handler above may have torn down the connection (e.g. an aborted
1628 * CONNACK, a v5 DISCONNECT, or an application callback that disconnected).
1629 * In that case the socket has been closed and unregistered, so we must not
1630 * keep parsing trailing bytes of the segment into the dead connection.
1631 */
1632 if(conn->state == MQTT_CONN_STATE_NOT_CONNECTED) {
1633 return 0;
1634 }
1635
1636 if(pos < input_data_len) {
1637 goto parse_next;
1638 }
1639
1640 return 0;
1641}
1642/*---------------------------------------------------------------------------*/
1643/*
1644 * Handles TCP events from Simple TCP
1645 */
1646static void
1647tcp_event(struct tcp_socket *s, void *ptr, tcp_socket_event_t event)
1648{
1649 struct mqtt_connection *conn = ptr;
1650
1651 /* Take care of event */
1652 switch(event) {
1653
1654 /* Fall through to manage different disconnect event the same way. */
1655 case TCP_SOCKET_CLOSED:
1656 case TCP_SOCKET_TIMEDOUT:
1657 case TCP_SOCKET_ABORTED: {
1658
1659 DBG("MQTT - Disconnected by tcp event %d\n", event);
1660 process_post(&mqtt_process, mqtt_abort_now_event, conn);
1661 conn->state = MQTT_CONN_STATE_NOT_CONNECTED;
1662 ctimer_stop(&conn->keep_alive_timer);
1663 call_event(conn, MQTT_EVENT_DISCONNECTED, &event);
1664 abort_connection(conn);
1665
1666 /* If connecting retry */
1667 if(conn->auto_reconnect == 1) {
1668 connect_tcp(conn);
1669 }
1670 break;
1671 }
1672 case TCP_SOCKET_CONNECTED: {
1673 conn->state = MQTT_CONN_STATE_TCP_CONNECTED;
1674 conn->out_buffer_sent = 1;
1675
1676 process_post(&mqtt_process, mqtt_do_connect_mqtt_event, conn);
1677 break;
1678 }
1679 case TCP_SOCKET_DATA_SENT: {
1680 DBG("MQTT - Got TCP_DATA_SENT\n");
1681
1682 if(conn->socket.output_data_len == 0) {
1683 conn->out_buffer_sent = 1;
1684 conn->out_buffer_ptr = conn->out_buffer;
1685 }
1686
1687 ctimer_restart(&conn->keep_alive_timer);
1688 break;
1689 }
1690
1691 default: {
1692 DBG("MQTT - TCP Event %d is currently not managed by the tcp event callback\n",
1693 event);
1694 }
1695 }
1696}
1697/*---------------------------------------------------------------------------*/
1698PROCESS_THREAD(mqtt_process, ev, data)
1699{
1700 static struct mqtt_connection *conn;
1701
1702 PROCESS_BEGIN();
1703
1704 while(1) {
1705 PROCESS_WAIT_EVENT();
1706
1707 if(ev == mqtt_abort_now_event) {
1708 DBG("MQTT - Abort\n");
1709 conn = data;
1710 conn->state = MQTT_CONN_STATE_ABORT_IMMEDIATE;
1711
1712 abort_connection(conn);
1713 }
1714 if(ev == mqtt_do_connect_tcp_event) {
1715 conn = data;
1716 DBG("MQTT - Got mqtt_do_connect_tcp_event!\n");
1717 connect_tcp(conn);
1718 }
1719 if(ev == mqtt_do_connect_mqtt_event) {
1720 conn = data;
1721 conn->socket.output_data_max_seg = conn->max_segment_size;
1722 DBG("MQTT - Got mqtt_do_connect_mqtt_event!\n");
1723
1724 if(conn->out_buffer_sent == 1) {
1725 PT_INIT(&conn->out_proto_thread);
1726 while(connect_pt(&conn->out_proto_thread, conn) < PT_EXITED &&
1727 conn->state != MQTT_CONN_STATE_ABORT_IMMEDIATE) {
1728 PT_MQTT_WAIT_SEND();
1729 }
1730 }
1731 }
1732 if(ev == mqtt_do_disconnect_mqtt_event) {
1733 conn = data;
1734 DBG("MQTT - Got mqtt_do_disconnect_mqtt_event!\n");
1735
1736 /* Send MQTT Disconnect if we are connected */
1737 if(conn->state == MQTT_CONN_STATE_SENDING_MQTT_DISCONNECT) {
1738 if(conn->out_buffer_sent == 1) {
1739 PT_INIT(&conn->out_proto_thread);
1740 while(conn->state != MQTT_CONN_STATE_ABORT_IMMEDIATE &&
1741 disconnect_pt(&conn->out_proto_thread, conn) < PT_EXITED) {
1742 PT_MQTT_WAIT_SEND();
1743 }
1744 abort_connection(conn);
1745 call_event(conn, MQTT_EVENT_DISCONNECTED, &ev);
1746 } else {
1747 process_post(&mqtt_process, mqtt_do_disconnect_mqtt_event, conn);
1748 }
1749 }
1750 }
1751 if(ev == mqtt_do_pingreq_event) {
1752 conn = data;
1753 DBG("MQTT - Got mqtt_do_pingreq_event!\n");
1754
1755 if(conn->out_buffer_sent == 1 &&
1756 conn->state == MQTT_CONN_STATE_CONNECTED_TO_BROKER) {
1757 PT_INIT(&conn->out_proto_thread);
1758 while(conn->state == MQTT_CONN_STATE_CONNECTED_TO_BROKER &&
1759 pingreq_pt(&conn->out_proto_thread, conn) < PT_EXITED) {
1760 PT_MQTT_WAIT_SEND();
1761 }
1762 }
1763 }
1764 if(ev == mqtt_do_subscribe_event) {
1765 conn = data;
1766 DBG("MQTT - Got mqtt_do_subscribe_mqtt_event!\n");
1767
1768 if(conn->out_buffer_sent == 1 &&
1769 conn->state == MQTT_CONN_STATE_CONNECTED_TO_BROKER) {
1770 PT_INIT(&conn->out_proto_thread);
1771 while(conn->state == MQTT_CONN_STATE_CONNECTED_TO_BROKER &&
1772 subscribe_pt(&conn->out_proto_thread, conn) < PT_EXITED) {
1773 PT_MQTT_WAIT_SEND();
1774 }
1775 }
1776 }
1777 if(ev == mqtt_do_unsubscribe_event) {
1778 conn = data;
1779 DBG("MQTT - Got mqtt_do_unsubscribe_mqtt_event!\n");
1780
1781 if(conn->out_buffer_sent == 1 &&
1782 conn->state == MQTT_CONN_STATE_CONNECTED_TO_BROKER) {
1783 PT_INIT(&conn->out_proto_thread);
1784 while(conn->state == MQTT_CONN_STATE_CONNECTED_TO_BROKER &&
1785 unsubscribe_pt(&conn->out_proto_thread, conn) < PT_EXITED) {
1786 PT_MQTT_WAIT_SEND();
1787 }
1788 }
1789 }
1790 if(ev == mqtt_do_publish_event) {
1791 conn = data;
1792 DBG("MQTT - Got mqtt_do_publish_mqtt_event!\n");
1793
1794 if(conn->out_buffer_sent == 1 &&
1795 conn->state == MQTT_CONN_STATE_CONNECTED_TO_BROKER) {
1796 PT_INIT(&conn->out_proto_thread);
1797 while(conn->state == MQTT_CONN_STATE_CONNECTED_TO_BROKER &&
1798 publish_pt(&conn->out_proto_thread, conn) < PT_EXITED) {
1799 PT_MQTT_WAIT_SEND();
1800 }
1801 }
1802 }
1803#if MQTT_5
1804 if(ev == mqtt_do_auth_event) {
1805 conn = data;
1806 DBG("MQTT - Got mqtt_do_auth_event!\n");
1807
1808 if(conn->out_buffer_sent == 1) {
1809 PT_INIT(&conn->out_proto_thread);
1810 while(auth_pt(&conn->out_proto_thread, conn) < PT_EXITED) {
1811 PT_MQTT_WAIT_SEND();
1812 }
1813 }
1814 }
1815 /* clear output properties; the next message sent should overwrite them */
1816 conn->out_props = NULL;
1817#endif
1818 }
1819 PROCESS_END();
1820}
1821/*---------------------------------------------------------------------------*/
1822void
1823mqtt_init(void)
1824{
1825 static uint8_t inited = 0;
1826 if(!inited) {
1827 mqtt_do_connect_tcp_event = process_alloc_event();
1828 mqtt_event_min = mqtt_do_connect_tcp_event;
1829
1830 mqtt_do_connect_mqtt_event = process_alloc_event();
1831 mqtt_do_disconnect_mqtt_event = process_alloc_event();
1832 mqtt_do_subscribe_event = process_alloc_event();
1833 mqtt_do_unsubscribe_event = process_alloc_event();
1834 mqtt_do_publish_event = process_alloc_event();
1835 mqtt_do_pingreq_event = process_alloc_event();
1836 mqtt_update_event = process_alloc_event();
1837 mqtt_abort_now_event = process_alloc_event();
1838 mqtt_event_max = mqtt_abort_now_event;
1839
1840 mqtt_continue_send_event = process_alloc_event();
1841 mqtt_do_auth_event = process_alloc_event();
1842
1843 list_init(mqtt_conn_list);
1844
1845 process_start(&mqtt_process, NULL);
1846 inited = 1;
1847 }
1848}
1849/*---------------------------------------------------------------------------*/
1850mqtt_status_t
1851mqtt_register(struct mqtt_connection *conn, struct process *app_process,
1852 char *client_id, mqtt_event_callback_t event_callback,
1853 uint16_t max_segment_size)
1854{
1855#if MQTT_31 || !MQTT_SRV_SUPPORTS_EMPTY_CLIENT_ID
1856 if(strlen(client_id) < 1) {
1857 return MQTT_STATUS_INVALID_ARGS_ERROR;
1858 }
1859#endif
1860
1861 /* Set defaults - Set all to zero to begin with */
1862 memset(conn, 0, sizeof(struct mqtt_connection));
1863#if MQTT_5
1864 /* Server capabilities have non-zero defaults */
1865 conn->srv_feature_en = -1;
1866#endif
1867 string_to_mqtt_string(&conn->client_id, client_id);
1868 conn->event_callback = event_callback;
1869 conn->app_process = app_process;
1870 conn->auto_reconnect = 1;
1871 conn->max_segment_size = max_segment_size;
1872
1873 reset_defaults(conn);
1874
1875 mqtt_init();
1876
1877 list_add(mqtt_conn_list, conn);
1878
1879 DBG("MQTT - Registered successfully\n");
1880
1881 return MQTT_STATUS_OK;
1882}
1883/*---------------------------------------------------------------------------*/
1884/*
1885 * Connect to MQTT broker.
1886 *
1887 * N.B. Non-blocking call.
1888 */
1889mqtt_status_t
1890mqtt_connect(struct mqtt_connection *conn, char *host, uint16_t port,
1891 uint16_t keep_alive,
1892#if MQTT_5
1893 uint8_t clean_session,
1894 struct mqtt_prop_list *prop_list)
1895#else
1896 uint8_t clean_session)
1897#endif
1898{
1899 uip_ip6addr_t ip6addr;
1900 uip_ipaddr_t *ipaddr;
1901 ipaddr = &ip6addr;
1902
1903 /* Check if we are already trying to connect */
1904 if(conn->state > MQTT_CONN_STATE_NOT_CONNECTED) {
1905 return MQTT_STATUS_OK;
1906 }
1907
1908 conn->server_host = host;
1909 conn->keep_alive = keep_alive;
1910 conn->server_port = port;
1911 conn->out_buffer_ptr = conn->out_buffer;
1912 conn->out_packet.qos_state = MQTT_QOS_STATE_NO_ACK;
1913
1914 /* If the Client supplies a zero-byte ClientId, the Client MUST also set CleanSession to 1 */
1915 if(clean_session || (conn->client_id.length == 0)) {
1916 conn->connect_vhdr_flags |= MQTT_VHDR_CLEAN_SESSION_FLAG;
1917 }
1918
1919 /* convert the string IPv6 address to a numeric IPv6 address */
1920 if(uiplib_ip6addrconv(host, &ip6addr) == 0) {
1921 return MQTT_STATUS_ERROR;
1922 }
1923
1924 uip_ipaddr_copy(&(conn->server_ip), ipaddr);
1925
1926 /*
1927 * Initiate the connection if the IP could be resolved. Otherwise the
1928 * connection will be initiated when the DNS lookup is finished, in the main
1929 * event loop.
1930 */
1931#if MQTT_5
1932 conn->out_props = prop_list;
1933#endif
1934
1935 process_post(&mqtt_process, mqtt_do_connect_tcp_event, conn);
1936
1937 return MQTT_STATUS_OK;
1938}
1939/*----------------------------------------------------------------------------*/
1940void
1941#if MQTT_5
1942mqtt_disconnect(struct mqtt_connection *conn,
1943 struct mqtt_prop_list *prop_list)
1944#else
1945mqtt_disconnect(struct mqtt_connection *conn)
1946#endif
1947{
1948 if(conn->state != MQTT_CONN_STATE_CONNECTED_TO_BROKER) {
1949 return;
1950 }
1951
1952 conn->state = MQTT_CONN_STATE_SENDING_MQTT_DISCONNECT;
1953
1954#if MQTT_5
1955 conn->out_props = prop_list;
1956#endif
1957
1958 process_post(&mqtt_process, mqtt_do_disconnect_mqtt_event, conn);
1959}
1960/*----------------------------------------------------------------------------*/
1961mqtt_status_t
1962mqtt_subscribe(struct mqtt_connection *conn, uint16_t *mid, char *topic,
1963#if MQTT_5
1964 mqtt_qos_level_t qos_level,
1965 mqtt_nl_en_t nl, mqtt_rap_en_t rap,
1966 mqtt_retain_handling_t ret_handling,
1967 struct mqtt_prop_list *prop_list)
1968#else
1969 mqtt_qos_level_t qos_level)
1970#endif
1971{
1972 if(conn->state != MQTT_CONN_STATE_CONNECTED_TO_BROKER) {
1973 return MQTT_STATUS_NOT_CONNECTED_ERROR;
1974 }
1975
1976 DBG("MQTT - Call to mqtt_subscribe...\n");
1977
1978 /* Currently don't have a queue, so only one item at a time */
1979 if(conn->out_queue_full) {
1980 DBG("MQTT - Not accepted!\n");
1981 return MQTT_STATUS_OUT_QUEUE_FULL;
1982 }
1983 conn->out_queue_full = 1;
1984 DBG("MQTT - Accepted!\n");
1985
1986 conn->out_packet.mid = INCREMENT_MID(conn);
1987 conn->out_packet.topic = topic;
1988 conn->out_packet.topic_length = strlen(topic);
1989 conn->out_packet.qos_state = MQTT_QOS_STATE_NO_ACK;
1990
1991 if(mid) {
1992 *mid = conn->out_packet.mid;
1993 }
1994
1995#if MQTT_5
1996 conn->out_packet.sub_options = 0x00;
1997 conn->out_packet.sub_options |= qos_level & MQTT_SUB_OPTION_QOS;
1998 conn->out_packet.sub_options |= nl & MQTT_SUB_OPTION_NL;
1999 conn->out_packet.sub_options |= rap & MQTT_SUB_OPTION_RAP;
2000 conn->out_packet.sub_options |= ret_handling & MQTT_SUB_OPTION_RETAIN_HANDLING;
2001#else
2002 conn->out_packet.qos = qos_level;
2003#endif
2004
2005#if MQTT_5
2006 conn->out_props = prop_list;
2007#endif
2008
2009 process_post(&mqtt_process, mqtt_do_subscribe_event, conn);
2010 return MQTT_STATUS_OK;
2011}
2012/*----------------------------------------------------------------------------*/
2013mqtt_status_t
2014mqtt_unsubscribe(struct mqtt_connection *conn, uint16_t *mid,
2015#if MQTT_5
2016 char *topic,
2017 struct mqtt_prop_list *prop_list)
2018#else
2019 char *topic)
2020#endif
2021{
2022 if(conn->state != MQTT_CONN_STATE_CONNECTED_TO_BROKER) {
2023 return MQTT_STATUS_NOT_CONNECTED_ERROR;
2024 }
2025
2026 DBG("MQTT - Call to mqtt_unsubscribe...\n");
2027 /* Currently don't have a queue, so only one item at a time */
2028 if(conn->out_queue_full) {
2029 DBG("MQTT - Not accepted!\n");
2030 return MQTT_STATUS_OUT_QUEUE_FULL;
2031 }
2032 conn->out_queue_full = 1;
2033 DBG("MQTT - Accepted!\n");
2034
2035 conn->out_packet.mid = INCREMENT_MID(conn);
2036 conn->out_packet.topic = topic;
2037 conn->out_packet.topic_length = strlen(topic);
2038 conn->out_packet.qos_state = MQTT_QOS_STATE_NO_ACK;
2039
2040 if(mid) {
2041 *mid = conn->out_packet.mid;
2042 }
2043
2044#if MQTT_5
2045 conn->out_props = prop_list;
2046#endif
2047
2048 process_post(&mqtt_process, mqtt_do_unsubscribe_event, conn);
2049 return MQTT_STATUS_OK;
2050}
2051/*----------------------------------------------------------------------------*/
2052mqtt_status_t
2053mqtt_publish(struct mqtt_connection *conn, uint16_t *mid, char *topic,
2054 uint8_t *payload, uint32_t payload_size,
2055 mqtt_qos_level_t qos_level,
2056#if MQTT_5
2057 mqtt_retain_t retain,
2058 uint8_t topic_alias, mqtt_topic_alias_en_t topic_alias_en,
2059 struct mqtt_prop_list *prop_list)
2060#else
2061 mqtt_retain_t retain)
2062#endif
2063{
2064 if(conn->state != MQTT_CONN_STATE_CONNECTED_TO_BROKER) {
2065 return MQTT_STATUS_NOT_CONNECTED_ERROR;
2066 }
2067
2068 DBG("MQTT - Call to mqtt_publish...\n");
2069
2070 /* Currently don't have a queue, so only one item at a time */
2071 if(conn->out_queue_full) {
2072 DBG("MQTT - Not accepted!\n");
2073 return MQTT_STATUS_OUT_QUEUE_FULL;
2074 }
2075 conn->out_queue_full = 1;
2076 DBG("MQTT - Accepted!\n");
2077
2078 conn->out_packet.mid = INCREMENT_MID(conn);
2079 conn->out_packet.retain = retain;
2080#if MQTT_5
2081 if(topic_alias_en == MQTT_TOPIC_ALIAS_ON) {
2082 conn->out_packet.topic = "";
2083 conn->out_packet.topic_length = 0;
2084 conn->out_packet.topic_alias = topic_alias;
2085 if(topic_alias == 0) {
2086 DBG("MQTT - Error, a topic alias of 0 is not permitted! It won't be sent.\n");
2087 }
2088 } else {
2089 conn->out_packet.topic = topic;
2090 conn->out_packet.topic_length = strlen(topic);
2091 conn->out_packet.topic_alias = 0;
2092 }
2093#else
2094 conn->out_packet.topic = topic;
2095 conn->out_packet.topic_length = strlen(topic);
2096#endif
2097 conn->out_packet.payload = payload;
2098 conn->out_packet.payload_size = payload_size;
2099 conn->out_packet.qos = qos_level;
2100 conn->out_packet.qos_state = MQTT_QOS_STATE_NO_ACK;
2101
2102 if(mid) {
2103 *mid = conn->out_packet.mid;
2104 }
2105
2106#if MQTT_5
2107 conn->out_props = prop_list;
2108#endif
2109
2110 process_post(&mqtt_process, mqtt_do_publish_event, conn);
2111 return MQTT_STATUS_OK;
2112}
2113/*----------------------------------------------------------------------------*/
2114void
2115mqtt_set_username_password(struct mqtt_connection *conn, char *username,
2116 char *password)
2117{
2118 /* Set strings, NULL string will simply set length to zero */
2119 string_to_mqtt_string(&conn->credentials.username, username);
2120 string_to_mqtt_string(&conn->credentials.password, password);
2121
2122 /* Set CONNECT VHDR flags */
2123 if(username != NULL) {
2124 conn->connect_vhdr_flags |= MQTT_VHDR_USERNAME_FLAG;
2125 } else {
2126 conn->connect_vhdr_flags &= ~MQTT_VHDR_USERNAME_FLAG;
2127 }
2128 if(password != NULL) {
2129 conn->connect_vhdr_flags |= MQTT_VHDR_PASSWORD_FLAG;
2130 } else {
2131 conn->connect_vhdr_flags &= ~MQTT_VHDR_PASSWORD_FLAG;
2132 }
2133}
2134/*----------------------------------------------------------------------------*/
2135void
2136mqtt_set_last_will(struct mqtt_connection *conn, char *topic, char *message,
2137#if MQTT_5
2138 mqtt_qos_level_t qos, struct mqtt_prop_list *will_props)
2139#else
2140 mqtt_qos_level_t qos)
2141#endif
2142{
2143 /* Set strings, NULL string will simply set length to zero */
2144 string_to_mqtt_string(&conn->will.topic, topic);
2145 string_to_mqtt_string(&conn->will.message, message);
2146
2147 /* Currently not used! */
2148 conn->will.qos = qos;
2149
2150 if(topic != NULL) {
2151 conn->connect_vhdr_flags |= MQTT_VHDR_WILL_FLAG |
2152 MQTT_VHDR_WILL_RETAIN_FLAG;
2153 }
2154#if MQTT_5
2155 conn->will.properties = (list_t)will_props;
2156#endif
2157}
2158/*---------------------------------------------------------------------------*/
2159#if MQTT_5
2160/*----------------------------------------------------------------------------*/
2161/* MQTTv5-specific functions */
2162/*----------------------------------------------------------------------------*/
2163/*
2164 * Send authentication data to broker.
2165 *
2166 * N.B. Non-blocking call.
2167 */
2168mqtt_status_t
2169mqtt_auth(struct mqtt_connection *conn,
2170 mqtt_auth_type_t auth_type,
2171 struct mqtt_prop_list *prop_list)
2172{
2173 DBG("MQTT - Call to mqtt_auth...\n");
2174
2175 conn->out_packet.fhdr = MQTT_FHDR_MSG_TYPE_AUTH;
2176 conn->out_packet.remaining_length = 1; /* for the auth reason code */
2177 conn->out_packet.auth_reason_code = MQTT_VHDR_RC_CONTINUE_AUTH + auth_type;
2178
2179 conn->out_props = prop_list;
2180
2181 process_post(&mqtt_process, mqtt_do_auth_event, conn);
2182 return MQTT_STATUS_OK;
2183}
2184#endif
2185/*----------------------------------------------------------------------------*/
2186/** @} */
cc.h
Default definitions of C compiler quirk work-arounds.
ctimer.h
Header file for the callback timer.
etimer.h
Event timer header file.
CLOCK_SECOND
#define CLOCK_SECOND
A second, measured in system clock time.
Definition clock.h:105
ctimer_stop
void ctimer_stop(struct ctimer *c)
Stop a pending callback timer.
Definition ctimer.c:150
ctimer_set
static void ctimer_set(struct ctimer *c, clock_time_t t, void(*f)(void *), void *ptr)
Set a callback timer.
Definition ctimer.h:150
ctimer_restart
void ctimer_restart(struct ctimer *c)
Restart a callback timer from the current point in time.
Definition ctimer.c:138
list_init
static void list_init(list_t list)
Initialize a list.
Definition list.h:152
LIST
#define LIST(name)
Declare a linked list.
Definition list.h:90
list_item_next
static void * list_item_next(const void *item)
Get the next item following this item.
Definition list.h:294
list_add
void list_add(list_t list, void *item)
Add an item at the end of a list.
Definition list.c:71
list_t
void ** list_t
The linked list type.
Definition list.h:136
list_head
static void * list_head(const_list_t list)
Get a pointer to the first element of a list.
Definition list.h:169
mqtt_auth
mqtt_status_t mqtt_auth(struct mqtt_connection *conn, mqtt_auth_type_t auth_type, struct mqtt_prop_list *prop_list)
Send authentication message (MQTTv5-only).
Definition mqtt.c:2169
mqtt_connect
mqtt_status_t mqtt_connect(struct mqtt_connection *conn, char *host, uint16_t port, uint16_t keep_alive, uint8_t clean_session, struct mqtt_prop_list *prop_list)
Connects to a MQTT broker.
Definition mqtt.c:1890
mqtt_register
mqtt_status_t mqtt_register(struct mqtt_connection *conn, struct process *app_process, char *client_id, mqtt_event_callback_t event_callback, uint16_t max_segment_size)
Initializes the MQTT engine.
Definition mqtt.c:1851
mqtt_unsubscribe
mqtt_status_t mqtt_unsubscribe(struct mqtt_connection *conn, uint16_t *mid, char *topic, struct mqtt_prop_list *prop_list)
Unsubscribes from a MQTT topic.
Definition mqtt.c:2014
mqtt_event_callback_t
void(* mqtt_event_callback_t)(struct mqtt_connection *m, mqtt_event_t event, void *data)
MQTT event callback function.
Definition mqtt.h:500
mqtt_subscribe
mqtt_status_t mqtt_subscribe(struct mqtt_connection *conn, uint16_t *mid, char *topic, mqtt_qos_level_t qos_level, mqtt_nl_en_t nl, mqtt_rap_en_t rap, mqtt_retain_handling_t ret_handling, struct mqtt_prop_list *prop_list)
Subscribes to a MQTT topic.
Definition mqtt.c:1962
mqtt_disconnect
void mqtt_disconnect(struct mqtt_connection *conn, struct mqtt_prop_list *prop_list)
Disconnects from a MQTT broker.
Definition mqtt.c:1942
mqtt_publish
mqtt_status_t mqtt_publish(struct mqtt_connection *conn, uint16_t *mid, char *topic, uint8_t *payload, uint32_t payload_size, mqtt_qos_level_t qos_level, mqtt_retain_t retain, uint8_t topic_alias, mqtt_topic_alias_en_t topic_alias_en, struct mqtt_prop_list *prop_list)
Publish to a MQTT topic.
Definition mqtt.c:2053
mqtt_event_t
mqtt_event_t
MQTT engine events.
Definition mqtt.h:218
mqtt_set_username_password
void mqtt_set_username_password(struct mqtt_connection *conn, char *username, char *password)
Set the user name and password for a MQTT client.
Definition mqtt.c:2115
mqtt_set_last_will
void mqtt_set_last_will(struct mqtt_connection *conn, char *topic, char *message, mqtt_qos_level_t qos, struct mqtt_prop_list *will_props)
Set the last will topic and message for a MQTT client.
Definition mqtt.c:2136
PROCESS
#define PROCESS(name, strname)
Declare a process.
Definition process.h:309
PROCESS_WAIT_EVENT
#define PROCESS_WAIT_EVENT()
Wait for an event to be posted to the process.
Definition process.h:143
process_post
int process_post(struct process *p, process_event_t ev, process_data_t data)
Post an asynchronous event.
Definition process.c:325
process_alloc_event
process_event_t process_alloc_event(void)
Allocate a global event number.
Definition process.c:111
PROCESS_BEGIN
#define PROCESS_BEGIN()
Define the beginning of a process.
Definition process.h:122
PROCESS_END
#define PROCESS_END()
Define the end of a process.
Definition process.h:133
process_start
void process_start(struct process *p, process_data_t data)
Start a process.
Definition process.c:121
PROCESS_THREAD
#define PROCESS_THREAD(name, ev, data)
Define the body of a process.
Definition process.h:275
PT_BEGIN
#define PT_BEGIN(pt)
Declare the start of a protothread inside the C function implementing the protothread.
Definition pt.h:280
PT_THREAD
#define PT_THREAD(name_args)
Declaration of a protothread.
Definition pt.h:265
PT_END
#define PT_END(pt)
Declare the end of a protothread.
Definition pt.h:292
PT_EXIT
#define PT_EXIT(pt)
Exit the protothread.
Definition pt.h:411
PT_WAIT_UNTIL
#define PT_WAIT_UNTIL(pt, condition)
Block and wait until condition is true.
Definition pt.h:313
PT_INIT
#define PT_INIT(pt)
Initialize a protothread.
Definition pt.h:245
timer_set
void timer_set(struct timer *t, clock_time_t interval)
Set a timer.
Definition timer.c:64
timer_expired
bool timer_expired(struct timer *t)
Check if a timer has expired.
Definition timer.c:123
uip_ipaddr_copy
#define uip_ipaddr_copy(dest, src)
Copy an IP address from one place to another.
Definition uip.h:969
leds.h
Header file for the LED HAL.
list.h
Linked list manipulation routines.
mqtt.h
Header file for the Contiki MQTT engine.
pt.h
Protothreads implementation.
random.h
Header file for generating non-cryptographic random numbers.
uip-ds6.h
Header file for IPv6-related data structures.
ipaddr
static uip_ipaddr_t ipaddr
Pointer to prefix information option in uip_buf.
Definition uip-nd6.c:116
uip.h
Header file for the uIP TCP/IP stack.