diff --git a/Kconfig b/Kconfig new file mode 100644 index 0000000..97e2ca2 --- /dev/null +++ b/Kconfig @@ -0,0 +1,69 @@ +menu "MQTT" + +# This is actually also handled in the ESP32 startup code, not only in FreeRTOS. +config MQTT_PROTOCOL_311 + bool "MQTT Protocol version 3.1.1" + default y + help + If disable, it will use MQTT protocol version 3.1 + +config MQTT_SECURITY_ON + bool "Enable MQTT over SSL" + default y + help + Enable MQTT Over SSL + +config CONFIG_MQTT_LOG_ERROR_ON + bool "Enable MQTT Debug message" + default y + help + Disable it will redurce memory and run faster +config CONFIG_MQTT_LOG_WARN_ON + bool "Enable MQTT Warning message" + default y + help + Disable it will redurce memory and run faster +config CONFIG_MQTT_LOG_INFO_ON + bool "Enable MQTT Info message" + default y + help + Disable it will redurce memory and run faster +config MQTT_RECONNECT_TIMEOUT + int "Reconnect timeout" + range 10 16535 + default 60 + +config MQTT_QUEUE_BUFFER_SIZE_WORD + int "Outbox queue buffer size in word (4 bytes)" + range 256 4096 + default 1024 + +config MQTT_MAX_HOST_LEN + int "Maximum host name len - in byte" + range 32 256 + default 64 + +config MQTT_MAX_CLIENT_LEN + int "Maximum client id len - in byte" + range 4 128 + default 32 + +config MQTT_MAX_USERNAME_LEN + int "Maximum mqtt username len - in byte" + range 4 128 + default 32 + +config MQTT_MAX_PASSWORD_LEN + int "Maximum mqtt password len - in byte" + range 4 128 + default 32 +config MQTT_MAX_LWT_TOPIC + int "Maximum mqtt lwt topic len - in byte" + range 4 128 + default 32 +config MQTT_MAX_LWT_MSG + int "Maximum mqtt lwt message len - in byte" + range 4 128 + default 32 + +endmenu diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..5b642a3 --- /dev/null +++ b/Makefile @@ -0,0 +1,18 @@ +# +# Component Makefile +# +# This Makefile should, at the very least, just include $(SDK_PATH)/make/component.mk. By default, +# this will take the sources in this directory, compile them and link them into +# lib(subdirectory_name).a in the build directory. This behaviour is entirely configurable, +# please read the SDK documents if you need to do this. +# +COMPONENT_ADD_INCLUDEDIRS := include +#COMPONENT_PRIV_INCLUDEDIRS := + +COMPONENT_SRCDIRS := . +#EXTRA_CFLAGS := -DICACHE_RODATA_ATTR +EXTRA_CFLAGS := -Wno-error=implicit-function-declaration -Wno-error=format= -DHAVE_CONFIG_H + + + +include $(IDF_PATH)/make/component.mk diff --git a/README.md b/README.md index 7762949..cd1989f 100644 --- a/README.md +++ b/README.md @@ -1 +1,3 @@ -# espmqtt \ No newline at end of file +# ESP32 MQTT Library + +This is component based on ESP-IDF for ESP32 diff --git a/include/mqtt.h b/include/mqtt.h new file mode 100644 index 0000000..58b478f --- /dev/null +++ b/include/mqtt.h @@ -0,0 +1,41 @@ +#ifndef _MQTT_H_ +#define _MQTT_H_ +#include +#include "mqtt_config.h" +#include "mqtt_msg.h" + +typedef void (* mqtt_callback)(void *); +typedef struct { + mqtt_callback connected_cb; + mqtt_callback disconnected_cb; + mqtt_callback reconnect_cb; + + mqtt_callback subscribe_cb; + mqtt_callback publish_cb; + mqtt_callback data_cb; + + char host[CONFIG_MQTT_MAX_HOST_LEN]; + uint32_t port; + char client_id[CONFIG_MQTT_MAX_CLIENT_LEN]; + char username[CONFIG_MQTT_MAX_USERNAME_LEN]; + char password[CONFIG_MQTT_MAX_PASSWORD_LEN]; + char lwt_topic[CONFIG_MQTT_MAX_LWT_TOPIC]; + char lwt_msg[CONFIG_MQTT_MAX_LWT_MSG]; + uint32_t lwt_qos, + uint32_t lwt_retain; + uint32_t clean_session; + uint32_t keepalive; +} mqtt_settings; + +typedef struct { + int socket; + mqtt_settings *settings; + mqtt_state_t mqtt_state; + mqtt_connect_info_t connect_info; +} mqtt_client; + +void mqtt_task(void *pv); +void mqtt_publish(); +void mqtt_subscribe(); +void mqtt_detroy(); +#endif diff --git a/include/mqtt_config.h b/include/mqtt_config.h new file mode 100644 index 0000000..4f33042 --- /dev/null +++ b/include/mqtt_config.h @@ -0,0 +1,27 @@ +#ifndef _MQTT_CONFIG_H_ +#define _MQTT_CONFIG_H_ +#include "sdkconfig.h" + + + +#ifdef CONFIG_MQTT_LOG_ERROR_ON + #define mqtt_error( format, ... ) printf( "[ERROR] "format, ## __VA_ARGS__ ) +#else + #define mqtt_error( format, ... ) +#endif +#ifdef CONFIG_MQTT_LOG_WARN_ON + #define mqtt_warn( format, ... ) printf( "[WARN] "format, ## __VA_ARGS__ ) +#else + #define mqtt_warn( format, ... ) +#endif +#ifdef CONFIG_MQTT_LOG_INFO_ON + #define mqtt_info( format, ... ) printf( "[INFO] "format, ## __VA_ARGS__ ) +#else + #define mqtt_info( format, ... ) +#endif + +#ifndef CONFIG_MQTT_QUEUE_BUFFER_SIZE_WORD +#define CONFIG_MQTT_QUEUE_BUFFER_SIZE_WORD 1024 +#endif + +#endif diff --git a/include/mqtt_msg.h b/include/mqtt_msg.h new file mode 100644 index 0000000..44da0cc --- /dev/null +++ b/include/mqtt_msg.h @@ -0,0 +1,133 @@ +#ifndef MQTT_MSG_H +#define MQTT_MSG_H +#include "mqtt_config.h" +#ifdef __cplusplus +extern "C" { +#endif + +/* +* Copyright (c) 2014, Stephen Robinson +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions +* are met: +* +* 1. Redistributions of source code must retain the above copyright +* notice, this list of conditions and the following disclaimer. +* 2. Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* 3. Neither the name of the copyright holder nor the names of its +* contributors may be used to endorse or promote products derived +* from this software without specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +* +*/ +/* 7 6 5 4 3 2 1 0 */ +/*| --- Message Type---- | DUP Flag | QoS Level | Retain | */ +/* Remaining Length */ + + +enum mqtt_message_type +{ + MQTT_MSG_TYPE_CONNECT = 1, + MQTT_MSG_TYPE_CONNACK = 2, + MQTT_MSG_TYPE_PUBLISH = 3, + MQTT_MSG_TYPE_PUBACK = 4, + MQTT_MSG_TYPE_PUBREC = 5, + MQTT_MSG_TYPE_PUBREL = 6, + MQTT_MSG_TYPE_PUBCOMP = 7, + MQTT_MSG_TYPE_SUBSCRIBE = 8, + MQTT_MSG_TYPE_SUBACK = 9, + MQTT_MSG_TYPE_UNSUBSCRIBE = 10, + MQTT_MSG_TYPE_UNSUBACK = 11, + MQTT_MSG_TYPE_PINGREQ = 12, + MQTT_MSG_TYPE_PINGRESP = 13, + MQTT_MSG_TYPE_DISCONNECT = 14 +}; + +enum mqtt_connect_return_code +{ + CONNECTION_ACCEPTED = 0, + CONNECTION_REFUSE_PROTOCOL, + CONNECTION_REFUSE_ID_REJECTED, + CONNECTION_REFUSE_SERVER_UNAVAILABLE, + CONNECTION_REFUSE_BAD_USERNAME, + CONNECTION_REFUSE_NOT_AUTHORIZED +}; + +typedef struct mqtt_message +{ + uint8_t* data; + uint16_t length; + +} mqtt_message_t; + +typedef struct mqtt_connection +{ + mqtt_message_t message; + + uint16_t message_id; + uint8_t* buffer; + uint16_t buffer_length; + +} mqtt_connection_t; + +typedef struct mqtt_connect_info +{ + char* client_id; + char* username; + char* password; + char* will_topic; + char* will_message; + int keepalive; + int will_qos; + int will_retain; + int clean_session; + +} mqtt_connect_info_t; + + +static inline int mqtt_get_type(uint8_t* buffer) { return (buffer[0] & 0xf0) >> 4; } +static inline int mqtt_get_connect_return_code(uint8_t* buffer) { return buffer[3]; } +static inline int mqtt_get_dup(uint8_t* buffer) { return (buffer[0] & 0x08) >> 3; } +static inline int mqtt_get_qos(uint8_t* buffer) { return (buffer[0] & 0x06) >> 1; } +static inline int mqtt_get_retain(uint8_t* buffer) { return (buffer[0] & 0x01); } + +void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length); +int mqtt_get_total_length(uint8_t* buffer, uint16_t length); +const char* mqtt_get_publish_topic(uint8_t* buffer, uint16_t* length); +const char* mqtt_get_publish_data(uint8_t* buffer, uint16_t* length); +uint16_t mqtt_get_id(uint8_t* buffer, uint16_t length); + +mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info); +mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topic, const char* data, int data_length, int qos, int retain, uint16_t* message_id); +mqtt_message_t* mqtt_msg_puback(mqtt_connection_t* connection, uint16_t message_id); +mqtt_message_t* mqtt_msg_pubrec(mqtt_connection_t* connection, uint16_t message_id); +mqtt_message_t* mqtt_msg_pubrel(mqtt_connection_t* connection, uint16_t message_id); +mqtt_message_t* mqtt_msg_pubcomp(mqtt_connection_t* connection, uint16_t message_id); +mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* topic, int qos, uint16_t* message_id); +mqtt_message_t* mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* topic, uint16_t* message_id); +mqtt_message_t* mqtt_msg_pingreq(mqtt_connection_t* connection); +mqtt_message_t* mqtt_msg_pingresp(mqtt_connection_t* connection); +mqtt_message_t* mqtt_msg_disconnect(mqtt_connection_t* connection); + + +#ifdef __cplusplus +} +#endif + +#endif /* MQTT_MSG_H */ + diff --git a/mqtt.c b/mqtt.c new file mode 100644 index 0000000..b499084 --- /dev/null +++ b/mqtt.c @@ -0,0 +1,13 @@ +/* +* @Author: Tuan PM +* @Date: 2016-09-10 09:33:06 +* @Last Modified by: Tuan PM +* @Last Modified time: 2016-09-11 09:48:52 +*/ +#include "mqtt.h" + + +void mqtt_task(void *pvParameters) +{ + mqtt_settings *settings = (mqtt_settings *)pvParameters; +} diff --git a/mqtt_msg.c b/mqtt_msg.c new file mode 100644 index 0000000..b88ca4e --- /dev/null +++ b/mqtt_msg.c @@ -0,0 +1,469 @@ +/* +* Copyright (c) 2014, Stephen Robinson +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions +* are met: +* +* 1. Redistributions of source code must retain the above copyright +* notice, this list of conditions and the following disclaimer. +* 2. Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* 3. Neither the name of the copyright holder nor the names of its +* contributors may be used to endorse or promote products derived +* from this software without specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +* +*/ +#include +#include +#include "mqtt_msg.h" +#include "mqtt_config.h" +#define MQTT_MAX_FIXED_HEADER_SIZE 3 + +enum mqtt_connect_flag +{ + MQTT_CONNECT_FLAG_USERNAME = 1 << 7, + MQTT_CONNECT_FLAG_PASSWORD = 1 << 6, + MQTT_CONNECT_FLAG_WILL_RETAIN = 1 << 5, + MQTT_CONNECT_FLAG_WILL = 1 << 2, + MQTT_CONNECT_FLAG_CLEAN_SESSION = 1 << 1 +}; + +struct __attribute((__packed__)) mqtt_connect_variable_header +{ + uint8_t lengthMsb; + uint8_t lengthLsb; +#if defined(CONFIG_MQTT_PROTOCOL_311) + uint8_t magic[4]; +#else + uint8_t magic[6]; +#endif + uint8_t version; + uint8_t flags; + uint8_t keepaliveMsb; + uint8_t keepaliveLsb; +}; + +static int append_string(mqtt_connection_t* connection, const char* string, int len) +{ + if (connection->message.length + len + 2 > connection->buffer_length) + return -1; + + connection->buffer[connection->message.length++] = len >> 8; + connection->buffer[connection->message.length++] = len & 0xff; + memcpy(connection->buffer + connection->message.length, string, len); + connection->message.length += len; + + return len + 2; +} + +static uint16_t append_message_id(mqtt_connection_t* connection, uint16_t message_id) +{ + // If message_id is zero then we should assign one, otherwise + // we'll use the one supplied by the caller + while (message_id == 0) + message_id = ++connection->message_id; + + if (connection->message.length + 2 > connection->buffer_length) + return 0; + + connection->buffer[connection->message.length++] = message_id >> 8; + connection->buffer[connection->message.length++] = message_id & 0xff; + + return message_id; +} + +static int init_message(mqtt_connection_t* connection) +{ + connection->message.length = MQTT_MAX_FIXED_HEADER_SIZE; + return MQTT_MAX_FIXED_HEADER_SIZE; +} + +static mqtt_message_t* fail_message(mqtt_connection_t* connection) +{ + connection->message.data = connection->buffer; + connection->message.length = 0; + return &connection->message; +} + +static mqtt_message_t* fini_message(mqtt_connection_t* connection, int type, int dup, int qos, int retain) +{ + int remaining_length = connection->message.length - MQTT_MAX_FIXED_HEADER_SIZE; + + if (remaining_length > 127) + { + connection->buffer[0] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1); + connection->buffer[1] = 0x80 | (remaining_length % 128); + connection->buffer[2] = remaining_length / 128; + connection->message.length = remaining_length + 3; + connection->message.data = connection->buffer; + } + else + { + connection->buffer[1] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1); + connection->buffer[2] = remaining_length; + connection->message.length = remaining_length + 2; + connection->message.data = connection->buffer + 1; + } + + return &connection->message; +} + +void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length) +{ + memset(connection, 0, sizeof(mqtt_connection_t)); + connection->buffer = buffer; + connection->buffer_length = buffer_length; +} + +int mqtt_get_total_length(uint8_t* buffer, uint16_t length) +{ + int i; + int totlen = 0; + + for (i = 1; i < length; ++i) + { + totlen += (buffer[i] & 0x7f) << (7 * (i - 1)); + if ((buffer[i] & 0x80) == 0) + { + ++i; + break; + } + } + totlen += i; + + return totlen; +} + +const char* mqtt_get_publish_topic(uint8_t* buffer, uint16_t* length) +{ + int i; + int totlen = 0; + int topiclen; + + for (i = 1; i < *length; ++i) + { + totlen += (buffer[i] & 0x7f) << (7 * (i - 1)); + if ((buffer[i] & 0x80) == 0) + { + ++i; + break; + } + } + totlen += i; + + if (i + 2 >= *length) + return NULL; + topiclen = buffer[i++] << 8; + topiclen |= buffer[i++]; + + if (i + topiclen > *length) + return NULL; + + *length = topiclen; + return (const char*)(buffer + i); +} + +const char* mqtt_get_publish_data(uint8_t* buffer, uint16_t* length) +{ + int i; + int totlen = 0; + int topiclen; + int blength = *length; + *length = 0; + + for (i = 1; i < blength; ++i) + { + totlen += (buffer[i] & 0x7f) << (7 * (i - 1)); + if ((buffer[i] & 0x80) == 0) + { + ++i; + break; + } + } + totlen += i; + + if (i + 2 >= blength) + return NULL; + topiclen = buffer[i++] << 8; + topiclen |= buffer[i++]; + + if (i + topiclen >= blength) + return NULL; + + i += topiclen; + + if (mqtt_get_qos(buffer) > 0) + { + if (i + 2 >= blength) + return NULL; + i += 2; + } + + if (totlen < i) + return NULL; + + if (totlen <= blength) + *length = totlen - i; + else + *length = blength - i; + return (const char*)(buffer + i); +} + +uint16_t mqtt_get_id(uint8_t* buffer, uint16_t length) +{ + if (length < 1) + return 0; + + switch (mqtt_get_type(buffer)) + { + case MQTT_MSG_TYPE_PUBLISH: + { + int i; + int topiclen; + + for (i = 1; i < length; ++i) + { + if ((buffer[i] & 0x80) == 0) + { + ++i; + break; + } + } + + if (i + 2 >= length) + return 0; + topiclen = buffer[i++] << 8; + topiclen |= buffer[i++]; + + if (i + topiclen >= length) + return 0; + i += topiclen; + + if (mqtt_get_qos(buffer) > 0) + { + if (i + 2 >= length) + return 0; + //i += 2; + } else { + return 0; + } + + return (buffer[i] << 8) | buffer[i + 1]; + } + case MQTT_MSG_TYPE_PUBACK: + case MQTT_MSG_TYPE_PUBREC: + case MQTT_MSG_TYPE_PUBREL: + case MQTT_MSG_TYPE_PUBCOMP: + case MQTT_MSG_TYPE_SUBACK: + case MQTT_MSG_TYPE_UNSUBACK: + case MQTT_MSG_TYPE_SUBSCRIBE: + { + // This requires the remaining length to be encoded in 1 byte, + // which it should be. + if (length >= 4 && (buffer[1] & 0x80) == 0) + return (buffer[2] << 8) | buffer[3]; + else + return 0; + } + + default: + return 0; + } +} + +mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info) +{ + struct mqtt_connect_variable_header* variable_header; + + init_message(connection); + + if (connection->message.length + sizeof(*variable_header) > connection->buffer_length) + return fail_message(connection); + variable_header = (void*)(connection->buffer + connection->message.length); + connection->message.length += sizeof(*variable_header); + + variable_header->lengthMsb = 0; +#if defined(CONFIG_MQTT_PROTOCOL_311) + variable_header->lengthLsb = 4; + memcpy(variable_header->magic, "MQTT", 4); + variable_header->version = 4; +#else + variable_header->lengthLsb = 6; + memcpy(variable_header->magic, "MQIsdp", 6); + variable_header->version = 3; +#endif + + variable_header->flags = 0; + variable_header->keepaliveMsb = info->keepalive >> 8; + variable_header->keepaliveLsb = info->keepalive & 0xff; + + if (info->clean_session) + variable_header->flags |= MQTT_CONNECT_FLAG_CLEAN_SESSION; + + if (info->client_id != NULL && info->client_id[0] != '\0') + { + if (append_string(connection, info->client_id, strlen(info->client_id)) < 0) + return fail_message(connection); + } + else + return fail_message(connection); + + if (info->will_topic != NULL && info->will_topic[0] != '\0') + { + if (append_string(connection, info->will_topic, strlen(info->will_topic)) < 0) + return fail_message(connection); + + if (append_string(connection, info->will_message, strlen(info->will_message)) < 0) + return fail_message(connection); + + variable_header->flags |= MQTT_CONNECT_FLAG_WILL; + if (info->will_retain) + variable_header->flags |= MQTT_CONNECT_FLAG_WILL_RETAIN; + variable_header->flags |= (info->will_qos & 3) << 3; + } + + if (info->username != NULL && info->username[0] != '\0') + { + if (append_string(connection, info->username, strlen(info->username)) < 0) + return fail_message(connection); + + variable_header->flags |= MQTT_CONNECT_FLAG_USERNAME; + } + + if (info->password != NULL && info->password[0] != '\0') + { + if (append_string(connection, info->password, strlen(info->password)) < 0) + return fail_message(connection); + + variable_header->flags |= MQTT_CONNECT_FLAG_PASSWORD; + } + + return fini_message(connection, MQTT_MSG_TYPE_CONNECT, 0, 0, 0); +} + +mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topic, const char* data, int data_length, int qos, int retain, uint16_t* message_id) +{ + init_message(connection); + + if (topic == NULL || topic[0] == '\0') + return fail_message(connection); + + if (append_string(connection, topic, strlen(topic)) < 0) + return fail_message(connection); + + if (qos > 0) + { + if ((*message_id = append_message_id(connection, 0)) == 0) + return fail_message(connection); + } + else + *message_id = 0; + + if (connection->message.length + data_length > connection->buffer_length) + return fail_message(connection); + memcpy(connection->buffer + connection->message.length, data, data_length); + connection->message.length += data_length; + + return fini_message(connection, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain); +} + +mqtt_message_t* mqtt_msg_puback(mqtt_connection_t* connection, uint16_t message_id) +{ + init_message(connection); + if (append_message_id(connection, message_id) == 0) + return fail_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PUBACK, 0, 0, 0); +} + +mqtt_message_t* mqtt_msg_pubrec(mqtt_connection_t* connection, uint16_t message_id) +{ + init_message(connection); + if (append_message_id(connection, message_id) == 0) + return fail_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PUBREC, 0, 0, 0); +} + +mqtt_message_t* mqtt_msg_pubrel(mqtt_connection_t* connection, uint16_t message_id) +{ + init_message(connection); + if (append_message_id(connection, message_id) == 0) + return fail_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PUBREL, 0, 1, 0); +} + +mqtt_message_t* mqtt_msg_pubcomp(mqtt_connection_t* connection, uint16_t message_id) +{ + init_message(connection); + if (append_message_id(connection, message_id) == 0) + return fail_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PUBCOMP, 0, 0, 0); +} + +mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* topic, int qos, uint16_t* message_id) +{ + init_message(connection); + + if (topic == NULL || topic[0] == '\0') + return fail_message(connection); + + if ((*message_id = append_message_id(connection, 0)) == 0) + return fail_message(connection); + + if (append_string(connection, topic, strlen(topic)) < 0) + return fail_message(connection); + + if (connection->message.length + 1 > connection->buffer_length) + return fail_message(connection); + connection->buffer[connection->message.length++] = qos; + + return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0); +} + +mqtt_message_t* mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* topic, uint16_t* message_id) +{ + init_message(connection); + + if (topic == NULL || topic[0] == '\0') + return fail_message(connection); + + if ((*message_id = append_message_id(connection, 0)) == 0) + return fail_message(connection); + + if (append_string(connection, topic, strlen(topic)) < 0) + return fail_message(connection); + + return fini_message(connection, MQTT_MSG_TYPE_UNSUBSCRIBE, 0, 1, 0); +} + +mqtt_message_t* mqtt_msg_pingreq(mqtt_connection_t* connection) +{ + init_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PINGREQ, 0, 0, 0); +} + +mqtt_message_t* mqtt_msg_pingresp(mqtt_connection_t* connection) +{ + init_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PINGRESP, 0, 0, 0); +} + +mqtt_message_t* mqtt_msg_disconnect(mqtt_connection_t* connection) +{ + init_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_DISCONNECT, 0, 0, 0); +}