feedc0de/esp-mqtt
1
0
Fork 0
forked from espressif/esp-mqtt
Code Pull Requests Activity

Seperate MQTT5 from MQTT 3.1.1

Browse Source
This commit is contained in:
yuanjm
2022-05-20 11:55:28 +08:00
committed by David Čermák
parent 3d275f42b9
commit fdf2aeb36f
8 changed files with 1242 additions and 1170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

280
include/mqtt5_client.h Normal file
View File

@ -0,0 +1,280 @@
/*
* SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#ifndef _MQTT5_CLIENT_H_
#define _MQTT5_CLIENT_H_
#include "mqtt_client.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct esp_mqtt_client *esp_mqtt5_client_handle_t;
/**
* MQTT5 protocol error reason code, more details refer to MQTT5 protocol document section 2.4
*/
enum mqtt5_error_reason_code {
MQTT5_UNSPECIFIED_ERROR = 0x80,
MQTT5_MALFORMED_PACKET = 0x81,
MQTT5_PROTOCOL_ERROR = 0x82,
MQTT5_IMPLEMENT_SPECIFIC_ERROR = 0x83,
MQTT5_UNSUPPORTED_PROTOCOL_VER = 0x84,
MQTT5_INVAILD_CLIENT_ID = 0x85,
MQTT5_BAD_USERNAME_OR_PWD = 0x86,
MQTT5_NOT_AUTHORIZED = 0x87,
MQTT5_SERVER_UNAVAILABLE = 0x88,
MQTT5_SERVER_BUSY = 0x89,
MQTT5_BANNED = 0x8A,
MQTT5_SERVER_SHUTTING_DOWN = 0x8B,
MQTT5_BAD_AUTH_METHOD = 0x8C,
MQTT5_KEEP_ALIVE_TIMEOUT = 0x8D,
MQTT5_SESSION_TAKEN_OVER = 0x8E,
MQTT5_TOPIC_FILTER_INVAILD = 0x8F,
MQTT5_TOPIC_NAME_INVAILD = 0x90,
MQTT5_PACKET_IDENTIFIER_IN_USE = 0x91,
MQTT5_PACKET_IDENTIFIER_NOT_FOUND = 0x92,
MQTT5_RECEIVE_MAXIMUM_EXCEEDED = 0x93,
MQTT5_TOPIC_ALIAS_INVAILD = 0x94,
MQTT5_PACKET_TOO_LARGE = 0x95,
MQTT5_MESSAGE_RATE_TOO_HIGH = 0x96,
MQTT5_QUOTA_EXCEEDED = 0x97,
MQTT5_ADMINISTRATIVE_ACTION = 0x98,
MQTT5_PAYLOAD_FORMAT_INVAILD = 0x99,
MQTT5_RETAIN_NOT_SUPPORT = 0x9A,
MQTT5_QOS_NOT_SUPPORT = 0x9B,
MQTT5_USE_ANOTHER_SERVER = 0x9C,
MQTT5_SERVER_MOVED = 0x9D,
MQTT5_SHARED_SUBSCR_NOT_SUPPORTED = 0x9E,
MQTT5_CONNECTION_RATE_EXCEEDED = 0x9F,
MQTT5_MAXIMUM_CONNECT_TIME = 0xA0,
MQTT5_SUBSCRIBE_IDENTIFIER_NOT_SUPPORT = 0xA1,
MQTT5_WILDCARD_SUBSCRIBE_NOT_SUPPORT = 0xA2,
};
/**
* MQTT5 user property handle
*/
typedef struct mqtt5_user_property_list_t *mqtt5_user_property_handle_t;
/**
* MQTT5 protocol connect properties and will properties configuration, more details refer to MQTT5 protocol document section 3.1.2.11 and 3.3.2.3
*/
typedef struct {
uint32_t session_expiry_interval; /*!< The interval time of session expiry */
uint32_t maximum_packet_size; /*!< The maximum packet size that we can receive */
uint16_t receive_maximum; /*!< The maximum pakcket count that we process concurrently */
uint16_t topic_alias_maximum; /*!< The maximum topic alias that we support */
bool request_resp_info; /*!< This value to request Server to return Response information */
bool request_problem_info; /*!< This value to indicate whether the reason string or user properties are sent in case of failures */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
uint32_t will_delay_interval; /*!< The time interval that server delays publishing will message */
uint32_t message_expiry_interval; /*!< The time interval that message expiry */
bool payload_format_indicator; /*!< This value is to indicator will message payload format */
const char *content_type; /*!< This value is to indicator will message content type, use a MIME content type string */
const char *response_topic; /*!< Topic name for a response message */
const char *correlation_data; /*!< Binary data for receiver to match the response message */
uint16_t correlation_data_len; /*!< The length of correlation data */
mqtt5_user_property_handle_t will_user_property; /*!< The handle for will message user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_connection_property_config_t;
/**
* MQTT5 protocol publish properties configuration, more details refer to MQTT5 protocol document section 3.3.2.3
*/
typedef struct {
bool payload_format_indicator; /*!< This value is to indicator publish message payload format */
uint32_t message_expiry_interval; /*!< The time interval that message expiry */
uint16_t topic_alias; /*!< An interger value to identify the topic instead of using topic name string */
const char *response_topic; /*!< Topic name for a response message */
const char *correlation_data; /*!< Binary data for receiver to match the response message */
uint16_t correlation_data_len; /*!< The length of correlation data */
const char *content_type; /*!< This value is to indicator publish message content type, use a MIME content type string */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_publish_property_config_t;
/**
* MQTT5 protocol subscribe properties configuration, more details refer to MQTT5 protocol document section 3.8.2.1
*/
typedef struct {
uint16_t subscribe_id; /*!< A variable byte represents the identifier of the subscription */
bool no_local_flag; /*!< Subscription Option to allow that server publish message that client sent */
bool retain_as_published_flag; /*!< Subscription Option to keep the retain flag as published option */
uint8_t retain_handle; /*!< Subscription Option to handle retain option */
bool is_share_subscribe; /*!< Whether subscribe is a shared subscription */
const char *share_name; /*!< The name of shared subscription which is a part of $share/{share_name}/{topic} */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_subscribe_property_config_t;
/**
* MQTT5 protocol unsubscribe properties configuration, more details refer to MQTT5 protocol document section 3.10.2.1
*/
typedef struct {
bool is_share_subscribe; /*!< Whether subscribe is a shared subscription */
const char *share_name; /*!< The name of shared subscription which is a part of $share/{share_name}/{topic} */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_unsubscribe_property_config_t;
/**
* MQTT5 protocol disconnect properties configuration, more details refer to MQTT5 protocol document section 3.14.2.2
*/
typedef struct {
uint32_t session_expiry_interval; /*!< The interval time of session expiry */
uint8_t disconnect_reason; /*!< The reason that connection disconnet, refer to mqtt5_error_reason_code */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_disconnect_property_config_t;
/**
* MQTT5 protocol for event properties
*/
typedef struct {
bool payload_format_indicator; /*!< Payload format of the message */
char *response_topic; /*!< Response topic of the message */
int response_topic_len; /*!< Response topic length of the message */
char *correlation_data; /*!< Correlation data of the message */
uint16_t correlation_data_len; /*!< Correlation data length of the message */
char *content_type; /*!< Content type of the message */
int content_type_len; /*!< Content type length of the message */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_delete_user_property to free the memory */
} esp_mqtt5_event_property_t;
/**
* MQTT5 protocol for user property
*/
typedef struct {
const char *key; /*!< Item key name */
const char *value; /*!< Item value string */
} esp_mqtt5_user_property_item_t;
/**
* @brief Set MQTT5 client connect property configuration
*
* @param client mqtt client handle
* @param connect_property connect property
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_connect_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_connection_property_config_t *connect_property);
/**
* @brief Set MQTT5 client publish property configuration
*
* This API will not store the publish property, it is one-time configuration.
* Before call `esp_mqtt_client_publish` to publish data, call this API to set publish property if have
*
* @param client mqtt client handle
* @param property publish property
*
* @return ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_publish_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_publish_property_config_t *property);
/**
* @brief Set MQTT5 client subscribe property configuration
*
* This API will not store the subscribe property, it is one-time configuration.
* Before call `esp_mqtt_client_subscribe` to subscribe topic, call this API to set subscribe property if have
*
* @param client mqtt client handle
* @param property subscribe property
*
* @return ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_subscribe_property_config_t *property);
/**
* @brief Set MQTT5 client unsubscribe property configuration
*
* This API will not store the unsubscribe property, it is one-time configuration.
* Before call `esp_mqtt_client_unsubscribe` to unsubscribe topic, call this API to set unsubscribe property if have
*
* @param client mqtt client handle
* @param property unsubscribe property
*
* @return ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_unsubscribe_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_unsubscribe_property_config_t *property);
/**
* @brief Set MQTT5 client disconnect property configuration
*
* This API will not store the disconnect property, it is one-time configuration.
* Before call `esp_mqtt_client_disconnect` to disconnect connection, call this API to set disconnect property if have
*
* @param client mqtt client handle
* @param property disconnect property
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_disconnect_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_disconnect_property_config_t *property);
/**
* @brief Set MQTT5 client user property configuration
*
* This API will allocate memory for user_property, please DO NOT forget `call esp_mqtt5_client_delete_user_property`
* after you use it.
* Before publish data, subscribe topic, unsubscribe, etc, call this API to set user property if have
*
* @param user_property user_property handle
* @param item array of user property data (eg. {{"var","val"},{"other","2"}})
* @param item_num number of items in user property data
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_user_property(mqtt5_user_property_handle_t *user_property, esp_mqtt5_user_property_item_t item[], uint8_t item_num);
/**
* @brief Get MQTT5 client user property
*
* @param user_property user_property handle
* @param item point that store user property data
* @param item_num number of items in user property data
*
* This API can use with `esp_mqtt5_client_get_user_property_count` to get list count of user property.
* And malloc number of count item array memory to store the user property data.
* Please DO NOT forget the item memory, key and value point in item memory when get user property data successfully.
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_get_user_property(mqtt5_user_property_handle_t user_property, esp_mqtt5_user_property_item_t *item, uint8_t *item_num);
/**
* @brief Get MQTT5 client user property list count
*
* @param user_property user_property handle
* @return user property list count
*/
uint8_t esp_mqtt5_client_get_user_property_count(mqtt5_user_property_handle_t user_property);
/**
* @brief Free the user property list
*
* @param user_property user_property handle
*
* This API will free the memory in user property list and free user_property itself
*/
void esp_mqtt5_client_delete_user_property(mqtt5_user_property_handle_t user_property);
#ifdef __cplusplus
}
#endif //__cplusplus
#endif

266
include/mqtt_client.h
View File

@ -12,6 +12,9 @@
#include <string.h>
#include "esp_err.h"
#include "esp_event.h"
#ifdef CONFIG_MQTT_PROTOCOL_5
#include "mqtt5_client.h"
#endif
#ifdef __cplusplus
extern "C" {
@ -118,142 +121,6 @@ typedef enum esp_mqtt_protocol_ver_t {
MQTT_PROTOCOL_V_5,
} esp_mqtt_protocol_ver_t;
#ifdef CONFIG_MQTT_PROTOCOL_5
/**
* MQTT5 protocol error reason code, more details refer to MQTT5 protocol document section 2.4
*/
enum mqtt5_error_reason_code {
MQTT5_UNSPECIFIED_ERROR = 0x80,
MQTT5_MALFORMED_PACKET = 0x81,
MQTT5_PROTOCOL_ERROR = 0x82,
MQTT5_IMPLEMENT_SPECIFIC_ERROR = 0x83,
MQTT5_UNSUPPORTED_PROTOCOL_VER = 0x84,
MQTT5_INVAILD_CLIENT_ID = 0x85,
MQTT5_BAD_USERNAME_OR_PWD = 0x86,
MQTT5_NOT_AUTHORIZED = 0x87,
MQTT5_SERVER_UNAVAILABLE = 0x88,
MQTT5_SERVER_BUSY = 0x89,
MQTT5_BANNED = 0x8A,
MQTT5_SERVER_SHUTTING_DOWN = 0x8B,
MQTT5_BAD_AUTH_METHOD = 0x8C,
MQTT5_KEEP_ALIVE_TIMEOUT = 0x8D,
MQTT5_SESSION_TAKEN_OVER = 0x8E,
MQTT5_TOPIC_FILTER_INVAILD = 0x8F,
MQTT5_TOPIC_NAME_INVAILD = 0x90,
MQTT5_PACKET_IDENTIFIER_IN_USE = 0x91,
MQTT5_PACKET_IDENTIFIER_NOT_FOUND = 0x92,
MQTT5_RECEIVE_MAXIMUM_EXCEEDED = 0x93,
MQTT5_TOPIC_ALIAS_INVAILD = 0x94,
MQTT5_PACKET_TOO_LARGE = 0x95,
MQTT5_MESSAGE_RATE_TOO_HIGH = 0x96,
MQTT5_QUOTA_EXCEEDED = 0x97,
MQTT5_ADMINISTRATIVE_ACTION = 0x98,
MQTT5_PAYLOAD_FORMAT_INVAILD = 0x99,
MQTT5_RETAIN_NOT_SUPPORT = 0x9A,
MQTT5_QOS_NOT_SUPPORT = 0x9B,
MQTT5_USE_ANOTHER_SERVER = 0x9C,
MQTT5_SERVER_MOVED = 0x9D,
MQTT5_SHARED_SUBSCR_NOT_SUPPORTED = 0x9E,
MQTT5_CONNECTION_RATE_EXCEEDED = 0x9F,
MQTT5_MAXIMUM_CONNECT_TIME = 0xA0,
MQTT5_SUBSCRIBE_IDENTIFIER_NOT_SUPPORT = 0xA1,
MQTT5_WILDCARD_SUBSCRIBE_NOT_SUPPORT = 0xA2,
};
/**
* MQTT5 user property handle
*/
typedef struct mqtt5_user_property_list_t *mqtt5_user_property_handle_t;
/**
* MQTT5 protocol connect properties and will properties configuration, more details refer to MQTT5 protocol document section 3.1.2.11 and 3.3.2.3
*/
typedef struct {
uint32_t session_expiry_interval; /*!< The interval time of session expiry */
uint32_t maximum_packet_size; /*!< The maximum packet size that we can receive */
uint16_t receive_maximum; /*!< The maximum pakcket count that we process concurrently */
uint16_t topic_alias_maximum; /*!< The maximum topic alias that we support */
bool request_resp_info; /*!< This value to request Server to return Response information */
bool request_problem_info; /*!< This value to indicate whether the reason string or user properties are sent in case of failures */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
uint32_t will_delay_interval; /*!< The time interval that server delays publishing will message */
uint32_t message_expiry_interval; /*!< The time interval that message expiry */
bool payload_format_indicator; /*!< This value is to indicator will message payload format */
const char *content_type; /*!< This value is to indicator will message content type, use a MIME content type string */
const char *response_topic; /*!< Topic name for a response message */
const char *correlation_data; /*!< Binary data for receiver to match the response message */
uint16_t correlation_data_len; /*!< The length of correlation data */
mqtt5_user_property_handle_t will_user_property; /*!< The handle for will message user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_connection_property_config_t;
/**
* MQTT5 protocol publish properties configuration, more details refer to MQTT5 protocol document section 3.3.2.3
*/
typedef struct {
bool payload_format_indicator; /*!< This value is to indicator publish message payload format */
uint32_t message_expiry_interval; /*!< The time interval that message expiry */
uint16_t topic_alias; /*!< An interger value to identify the topic instead of using topic name string */
const char *response_topic; /*!< Topic name for a response message */
const char *correlation_data; /*!< Binary data for receiver to match the response message */
uint16_t correlation_data_len; /*!< The length of correlation data */
const char *content_type; /*!< This value is to indicator publish message content type, use a MIME content type string */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_publish_property_config_t;
/**
* MQTT5 protocol subscribe properties configuration, more details refer to MQTT5 protocol document section 3.8.2.1
*/
typedef struct {
uint16_t subscribe_id; /*!< A variable byte represents the identifier of the subscription */
bool no_local_flag; /*!< Subscription Option to allow that server publish message that client sent */
bool retain_as_published_flag; /*!< Subscription Option to keep the retain flag as published option */
uint8_t retain_handle; /*!< Subscription Option to handle retain option */
bool is_share_subscribe; /*!< Whether subscribe is a shared subscription */
const char *share_name; /*!< The name of shared subscription which is a part of $share/{share_name}/{topic} */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_subscribe_property_config_t;
/**
* MQTT5 protocol unsubscribe properties configuration, more details refer to MQTT5 protocol document section 3.10.2.1
*/
typedef struct {
bool is_share_subscribe; /*!< Whether subscribe is a shared subscription */
const char *share_name; /*!< The name of shared subscription which is a part of $share/{share_name}/{topic} */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_unsubscribe_property_config_t;
/**
* MQTT5 protocol disconnect properties configuration, more details refer to MQTT5 protocol document section 3.14.2.2
*/
typedef struct {
uint32_t session_expiry_interval; /*!< The interval time of session expiry */
uint8_t disconnect_reason; /*!< The reason that connection disconnet, refer to mqtt5_error_reason_code */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_set_user_property to set it */
} esp_mqtt5_disconnect_property_config_t;
/**
* MQTT5 protocol for event properties
*/
typedef struct {
bool payload_format_indicator; /*!< Payload format of the message */
char *response_topic; /*!< Response topic of the message */
int response_topic_len; /*!< Response topic length of the message */
char *correlation_data; /*!< Correlation data of the message */
uint16_t correlation_data_len; /*!< Correlation data length of the message */
char *content_type; /*!< Content type of the message */
int content_type_len; /*!< Content type length of the message */
mqtt5_user_property_handle_t user_property; /*!< The handle for user property, call function esp_mqtt5_client_delete_user_property to free the memory */
} esp_mqtt5_event_property_t;
/**
* MQTT5 protocol for user property
*/
typedef struct {
const char *key; /*!< Item key name */
const char *value; /*!< Item value string */
} esp_mqtt5_user_property_item_t;
#endif
/**
* @brief MQTT error code structure to be passed as a contextual information into ERROR event
*
@ -561,133 +428,6 @@ esp_err_t esp_mqtt_client_register_event(esp_mqtt_client_handle_t client, esp_mq
* 0 on wrong initialization
*/
int esp_mqtt_client_get_outbox_size(esp_mqtt_client_handle_t client);
#ifdef CONFIG_MQTT_PROTOCOL_5
/**
* @brief Set MQTT5 client connect property configuration
*
* @param client mqtt client handle
* @param connect_property connect property
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_connect_property(esp_mqtt_client_handle_t client, const esp_mqtt5_connection_property_config_t *connect_property);
/**
* @brief Set MQTT5 client publish property configuration
*
* This API will not store the publish property, it is one-time configuration.
* Before call `esp_mqtt_client_publish` to publish data, call this API to set publish property if have
*
* @param client mqtt client handle
* @param property publish property
*
* @return ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_publish_property(esp_mqtt_client_handle_t client, const esp_mqtt5_publish_property_config_t *property);
/**
* @brief Set MQTT5 client subscribe property configuration
*
* This API will not store the subscribe property, it is one-time configuration.
* Before call `esp_mqtt_client_subscribe` to subscribe topic, call this API to set subscribe property if have
*
* @param client mqtt client handle
* @param property subscribe property
*
* @return ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt_client_handle_t client, const esp_mqtt5_subscribe_property_config_t *property);
/**
* @brief Set MQTT5 client unsubscribe property configuration
*
* This API will not store the unsubscribe property, it is one-time configuration.
* Before call `esp_mqtt_client_unsubscribe` to unsubscribe topic, call this API to set unsubscribe property if have
*
* @param client mqtt client handle
* @param property unsubscribe property
*
* @return ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_unsubscribe_property(esp_mqtt_client_handle_t client, const esp_mqtt5_unsubscribe_property_config_t *property);
/**
* @brief Set MQTT5 client disconnect property configuration
*
* This API will not store the disconnect property, it is one-time configuration.
* Before call `esp_mqtt_client_disconnect` to disconnect connection, call this API to set disconnect property if have
*
* @param client mqtt client handle
* @param property disconnect property
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_disconnect_property(esp_mqtt_client_handle_t client, const esp_mqtt5_disconnect_property_config_t *property);
/**
* @brief Set MQTT5 client user property configuration
*
* This API will allocate memory for user_property, please DO NOT forget `call esp_mqtt5_client_delete_user_property`
* after you use it.
* Before publish data, subscribe topic, unsubscribe, etc, call this API to set user property if have
*
* @param user_property user_property handle
* @param item array of user property data (eg. {{"var","val"},{"other","2"}})
* @param item_num number of items in user property data
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_set_user_property(mqtt5_user_property_handle_t *user_property, esp_mqtt5_user_property_item_t item[], uint8_t item_num);
/**
* @brief Get MQTT5 client user property
*
* @param user_property user_property handle
* @param item point that store user property data
* @param item_num number of items in user property data
*
* This API can use with `esp_mqtt5_client_get_user_property_count` to get list count of user property.
* And malloc number of count item array memory to store the user property data.
* Please DO NOT forget the item memory, key and value point in item memory when get user property data successfully.
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_FAIL on fail
* ESP_OK on success
*/
esp_err_t esp_mqtt5_client_get_user_property(mqtt5_user_property_handle_t user_property, esp_mqtt5_user_property_item_t *item, uint8_t *item_num);
/**
* @brief Get MQTT5 client user property list count
*
* @param user_property user_property handle
* @return user property list count
*/
uint8_t esp_mqtt5_client_get_user_property_count(mqtt5_user_property_handle_t user_property);
/**
* @brief Free the user property list
*
* @param user_property user_property handle
*
* This API will free the memory in user property list and free user_property itself
*/
void esp_mqtt5_client_delete_user_property(mqtt5_user_property_handle_t user_property);
#endif
#ifdef __cplusplus
}
#endif //__cplusplus

54
lib/include/mqtt5_client_priv.h Normal file
View File

@ -0,0 +1,54 @@
/*
* SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#ifndef _MQTT5_CLIENT_PRIV_H_
#define _MQTT5_CLIENT_PRIV_H_
#include "mqtt5_client.h"
#include "mqtt_client_priv.h"
#include "mqtt5_msg.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct mqtt5_topic_alias {
char *topic;
uint16_t topic_len;
uint16_t topic_alias;
STAILQ_ENTRY(mqtt5_topic_alias) next;
} mqtt5_topic_alias_t;
STAILQ_HEAD(mqtt5_topic_alias_list_t, mqtt5_topic_alias);
typedef struct mqtt5_topic_alias_list_t *mqtt5_topic_alias_handle_t;
typedef struct mqtt5_topic_alias *mqtt5_topic_alias_item_t;
typedef struct {
esp_mqtt5_connection_property_storage_t connect_property_info;
esp_mqtt5_connection_will_property_storage_t will_property_info;
esp_mqtt5_connection_server_resp_property_t server_resp_property_info;
esp_mqtt5_disconnect_property_config_t disconnect_property_info;
const esp_mqtt5_publish_property_config_t *publish_property_info;
const esp_mqtt5_subscribe_property_config_t *subscribe_property_info;
const esp_mqtt5_unsubscribe_property_config_t *unsubscribe_property_info;
mqtt5_topic_alias_handle_t peer_topic_alias;
} mqtt5_config_storage_t;
void esp_mqtt5_flow_control(esp_mqtt5_client_handle_t client);
void esp_mqtt5_parse_pubcomp(esp_mqtt5_client_handle_t client);
void esp_mqtt5_parse_puback(esp_mqtt5_client_handle_t client);
void esp_mqtt5_parse_unsuback(esp_mqtt5_client_handle_t client);
void esp_mqtt5_parse_suback(esp_mqtt5_client_handle_t client);
esp_err_t esp_mqtt5_parse_connack(esp_mqtt5_client_handle_t client, int *connect_rsp_code);
void esp_mqtt5_client_destory(esp_mqtt5_client_handle_t client);
esp_err_t esp_mqtt5_client_publish_check(esp_mqtt5_client_handle_t client, int qos, int retain);
esp_err_t esp_mqtt5_client_subscribe_check(esp_mqtt5_client_handle_t client, int qos);
esp_err_t esp_mqtt5_create_default_config(esp_mqtt5_client_handle_t client);
esp_err_t esp_mqtt5_get_publish_data(esp_mqtt5_client_handle_t client, uint8_t *msg_buf, size_t msg_read_len, char **msg_topic, size_t *msg_topic_len, char **msg_data, size_t *msg_data_len);
#ifdef __cplusplus
}
#endif //__cplusplus
#endif

2
lib/include/mqtt5_msg.h
View File

@ -2,7 +2,7 @@
#define MQTT5_MSG_H
#include <stdint.h>
#include <stdbool.h>
#include "sys/queue.h"
#include "mqtt_config.h"
#include "mqtt_msg.h"
#include "mqtt_client.h"

136
lib/include/mqtt_client_priv.h Normal file
View File

@ -0,0 +1,136 @@
/*
* SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#ifndef _MQTT_CLIENT_PRIV_H_
#define _MQTT_CLIENT_PRIV_H_
#include <stdio.h>
#include <stdlib.h>
#include "esp_err.h"
#include "platform.h"
#include "esp_event.h"
#include "mqtt_client.h"
#include "mqtt_msg.h"
#ifdef MQTT_PROTOCOL_5
#include "mqtt5_client_priv.h"
#endif
#include "esp_transport.h"
#include "esp_transport_tcp.h"
#include "esp_transport_ssl.h"
#include "esp_transport_ws.h"
#include "esp_log.h"
#include "mqtt_outbox.h"
#include "freertos/event_groups.h"
#include <errno.h>
#include <string.h>
#include "mqtt_supported_features.h"
/* using uri parser */
#include "http_parser.h"
#ifdef __cplusplus
extern "C" {
#endif
#ifdef MQTT_DISABLE_API_LOCKS
# define MQTT_API_LOCK(c)
# define MQTT_API_UNLOCK(c)
#else
# define MQTT_API_LOCK(c) xSemaphoreTakeRecursive(c->api_lock, portMAX_DELAY)
# define MQTT_API_UNLOCK(c) xSemaphoreGiveRecursive(c->api_lock)
#endif /* MQTT_USE_API_LOCKS */
typedef struct mqtt_state {
uint8_t *in_buffer;
uint8_t *out_buffer;
int in_buffer_length;
int out_buffer_length;
size_t message_length;
size_t in_buffer_read_len;
mqtt_message_t *outbound_message;
mqtt_connection_t mqtt_connection;
uint16_t pending_msg_id;
int pending_msg_type;
int pending_publish_qos;
int pending_msg_count;
} mqtt_state_t;
typedef struct {
mqtt_event_callback_t event_handle;
esp_event_loop_handle_t event_loop_handle;
int task_stack;
int task_prio;
char *uri;
char *host;
char *path;
char *scheme;
int port;
bool auto_reconnect;
void *user_context;
int network_timeout_ms;
int refresh_connection_after_ms;
int reconnect_timeout_ms;
char **alpn_protos;
int num_alpn_protos;
char *clientkey_password;
int clientkey_password_len;
bool use_global_ca_store;
esp_err_t ((*crt_bundle_attach)(void *conf));
const char *cacert_buf;
size_t cacert_bytes;
const char *clientcert_buf;
size_t clientcert_bytes;
const char *clientkey_buf;
size_t clientkey_bytes;
const struct psk_key_hint *psk_hint_key;
bool skip_cert_common_name_check;
bool use_secure_element;
void *ds_data;
int message_retransmit_timeout;
} mqtt_config_storage_t;
typedef enum {
MQTT_STATE_INIT = 0,
MQTT_STATE_DISCONNECTED,
MQTT_STATE_CONNECTED,
MQTT_STATE_WAIT_RECONNECT,
} mqtt_client_state_t;
struct esp_mqtt_client {
esp_transport_list_handle_t transport_list;
esp_transport_handle_t transport;
mqtt_config_storage_t *config;
mqtt_state_t mqtt_state;
mqtt_connect_info_t connect_info;
mqtt_client_state_t state;
uint64_t refresh_connection_tick;
int64_t keepalive_tick;
uint64_t reconnect_tick;
#ifdef MQTT_PROTOCOL_5
mqtt5_config_storage_t *mqtt5_config;
uint16_t send_publish_packet_count; // This is for MQTT v5.0 flow control
#endif
int wait_timeout_ms;
int auto_reconnect;
esp_mqtt_event_t event;
bool run;
bool wait_for_ping_resp;
outbox_handle_t outbox;
EventGroupHandle_t status_bits;
SemaphoreHandle_t api_lock;
TaskHandle_t task_handle;
};
bool esp_mqtt_set_if_config(char const *const new_config, char **old_config);
void esp_mqtt_destroy_config(esp_mqtt_client_handle_t client);
#ifdef __cplusplus
}
#endif //__cplusplus
#endif

2
lib/mqtt5_msg.c
View File

@ -679,7 +679,7 @@ esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, m
continue;
case MQTT5_PROPERTY_SERVER_KEEP_ALIVE:
MQTT5_CONVERT_ONE_BYTE_TO_TWO(connection_info->keepalive, property[property_offset ++], property[property_offset ++])
ESP_LOGD(TAG, "MQTT5_PROPERTY_SERVER_KEEP_ALIVE %d", connection_info->keepalive);
ESP_LOGD(TAG, "MQTT5_PROPERTY_SERVER_KEEP_ALIVE %lld", connection_info->keepalive);
continue;
case MQTT5_PROPERTY_RESP_INFO:
if (resp_property->response_info) {

757
mqtt5_client.c Normal file
View File

@ -0,0 +1,757 @@
/*
* SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
#include "mqtt_client_priv.h"
#include "esp_log.h"
#include <string.h>
static const char *TAG = "mqtt5_client";
static void esp_mqtt5_print_error_code(esp_mqtt5_client_handle_t client, int code);
static esp_err_t esp_mqtt5_client_update_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle, uint16_t topic_alias, char *topic, size_t topic_len);
static char *esp_mqtt5_client_get_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle, uint16_t topic_alias, size_t *topic_length);
static void esp_mqtt5_client_delete_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle);
static esp_err_t esp_mqtt5_user_property_copy(mqtt5_user_property_handle_t user_property_new, const mqtt5_user_property_handle_t user_property_old);
void esp_mqtt5_flow_control(esp_mqtt5_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
int msg_type = mqtt5_get_type(client->mqtt_state.outbound_message->data);
if (msg_type == MQTT_MSG_TYPE_PUBLISH) {
int msg_qos = mqtt5_get_qos(client->mqtt_state.outbound_message->data);
if (msg_qos > 0) {
client->send_publish_packet_count ++;
ESP_LOGD(TAG, "Sent (%d) qos > 0 publish packet without ack", client->send_publish_packet_count);
}
}
}
}
void esp_mqtt5_parse_pubcomp(esp_mqtt5_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBCOMP return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.data = mqtt5_get_pubcomp_data(client->mqtt_state.in_buffer, &msg_data_len, &client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
client->event.current_data_offset = 0;
}
}
void esp_mqtt5_parse_puback(esp_mqtt5_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.data = mqtt5_get_puback_data(client->mqtt_state.in_buffer, &msg_data_len, &client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
client->event.current_data_offset = 0;
client->send_publish_packet_count --;
}
}
void esp_mqtt5_parse_unsuback(esp_mqtt5_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGI(TAG, "MQTT_MSG_TYPE_UNSUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.data = mqtt5_get_unsuback_data(client->mqtt_state.in_buffer, &msg_data_len, &client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
client->event.current_data_offset = 0;
}
}
void esp_mqtt5_parse_suback(esp_mqtt5_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGI(TAG, "MQTT_MSG_TYPE_SUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
}
}
esp_err_t esp_mqtt5_parse_connack(esp_mqtt5_client_handle_t client, int *connect_rsp_code)
{
size_t len = client->mqtt_state.in_buffer_read_len;
client->mqtt_state.in_buffer_read_len = 0;
uint8_t ack_flag = 0;
if (mqtt5_msg_parse_connack_property(client->mqtt_state.in_buffer, len, &client->connect_info, &client->mqtt5_config->connect_property_info, &client->mqtt5_config->server_resp_property_info, connect_rsp_code, &ack_flag, &client->event.property->user_property) != ESP_OK) {
ESP_LOGE(TAG, "Failed to parse CONNACK packet");
return ESP_FAIL;
}
if (*connect_rsp_code == MQTT_CONNECTION_ACCEPTED) {
ESP_LOGD(TAG, "Connected");
client->event.session_present = ack_flag & 0x01;
return ESP_OK;
}
esp_mqtt5_print_error_code(client, *connect_rsp_code);
return ESP_FAIL;
}
esp_err_t esp_mqtt5_get_publish_data(esp_mqtt5_client_handle_t client, uint8_t *msg_buf, size_t msg_read_len, char **msg_topic, size_t *msg_topic_len, char **msg_data, size_t *msg_data_len)
{
// get property
uint16_t property_len = 0;
esp_mqtt5_publish_resp_property_t property = {0};
*msg_data = mqtt5_get_publish_property_payload(msg_buf, msg_read_len, msg_topic, msg_topic_len, &property, &property_len, msg_data_len, &client->event.property->user_property);
if (*msg_data_len == 0 || *msg_data == NULL) {
ESP_LOGE(TAG, "%s: mqtt5_get_publish_property_payload() failed", __func__);
return ESP_FAIL;
}
if (property.topic_alias > client->mqtt5_config->connect_property_info.topic_alias_maximum) {
ESP_LOGE(TAG, "%s: Broker response topic alias %d is over the max topic alias %d", __func__, property.topic_alias, client->mqtt5_config->connect_property_info.topic_alias_maximum);
return ESP_FAIL;
}
if (property.topic_alias) {
if (*msg_topic_len == 0) {
ESP_LOGI(TAG, "Publish topic is empty, use topic alias");
*msg_topic = esp_mqtt5_client_get_topic_alias(client->mqtt5_config->peer_topic_alias, property.topic_alias, msg_topic_len);
if (!*msg_topic) {
ESP_LOGE(TAG, "%s: esp_mqtt5_client_get_topic_alias() failed", __func__);
return ESP_FAIL;
}
} else {
if (esp_mqtt5_client_update_topic_alias(client->mqtt5_config->peer_topic_alias, property.topic_alias, *msg_topic, *msg_topic_len) != ESP_OK) {
ESP_LOGE(TAG, "%s: esp_mqtt5_client_update_topic_alias() failed", __func__);
return ESP_FAIL;
}
}
}
client->event.property->payload_format_indicator = property.payload_format_indicator;
client->event.property->response_topic = property.response_topic;
client->event.property->response_topic_len = property.response_topic_len;
client->event.property->correlation_data = property.correlation_data;
client->event.property->correlation_data_len = property.correlation_data_len;
client->event.property->content_type = property.content_type;
client->event.property->content_type_len = property.content_type_len;
return ESP_OK;
}
esp_err_t esp_mqtt5_create_default_config(esp_mqtt5_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
client->event.property = calloc(1, sizeof(esp_mqtt5_event_property_t));
ESP_MEM_CHECK(TAG, client->event.property, return ESP_FAIL)
client->mqtt5_config = calloc(1, sizeof(mqtt5_config_storage_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config, return ESP_FAIL)
client->mqtt5_config->server_resp_property_info.max_qos = 2;
client->mqtt5_config->server_resp_property_info.retain_available = true;
client->mqtt5_config->server_resp_property_info.wildcard_subscribe_available = true;
client->mqtt5_config->server_resp_property_info.subscribe_identifiers_available = true;
client->mqtt5_config->server_resp_property_info.shared_subscribe_available = true;
client->mqtt5_config->server_resp_property_info.receive_maximum = 65535;
}
return ESP_OK;
}
static void esp_mqtt5_print_error_code(esp_mqtt5_client_handle_t client, int code)
{
switch (code) {
case MQTT5_UNSPECIFIED_ERROR:
ESP_LOGW(TAG, "Unspecified error");
break;
case MQTT5_MALFORMED_PACKET:
ESP_LOGW(TAG, "Malformed Packet");
break;
case MQTT5_PROTOCOL_ERROR:
ESP_LOGW(TAG, "Protocol Error");
break;
case MQTT5_IMPLEMENT_SPECIFIC_ERROR:
ESP_LOGW(TAG, "Implementation specific error");
break;
case MQTT5_UNSUPPORTED_PROTOCOL_VER:
ESP_LOGW(TAG, "Unsupported Protocol Version");
break;
case MQTT5_INVAILD_CLIENT_ID:
ESP_LOGW(TAG, "Client Identifier not valid");
break;
case MQTT5_BAD_USERNAME_OR_PWD:
ESP_LOGW(TAG, "Bad User Name or Password");
break;
case MQTT5_NOT_AUTHORIZED:
ESP_LOGW(TAG, "Not authorized");
break;
case MQTT5_SERVER_UNAVAILABLE:
ESP_LOGW(TAG, "Server unavailable");
break;
case MQTT5_SERVER_BUSY:
ESP_LOGW(TAG, "Server busy");
break;
case MQTT5_BANNED:
ESP_LOGW(TAG, "Banned");
break;
case MQTT5_SERVER_SHUTTING_DOWN:
ESP_LOGW(TAG, "Server shutting down");
break;
case MQTT5_BAD_AUTH_METHOD:
ESP_LOGW(TAG, "Bad authentication method");
break;
case MQTT5_KEEP_ALIVE_TIMEOUT:
ESP_LOGW(TAG, "Keep Alive timeout");
break;
case MQTT5_SESSION_TAKEN_OVER:
ESP_LOGW(TAG, "Session taken over");
break;
case MQTT5_TOPIC_FILTER_INVAILD:
ESP_LOGW(TAG, "Topic Filter invalid");
break;
case MQTT5_TOPIC_NAME_INVAILD:
ESP_LOGW(TAG, "Topic Name invalid");
break;
case MQTT5_PACKET_IDENTIFIER_IN_USE:
ESP_LOGW(TAG, "Packet Identifier in use");
break;
case MQTT5_PACKET_IDENTIFIER_NOT_FOUND:
ESP_LOGW(TAG, "Packet Identifier not found");
break;
case MQTT5_RECEIVE_MAXIMUM_EXCEEDED:
ESP_LOGW(TAG, "Receive Maximum exceeded");
break;
case MQTT5_TOPIC_ALIAS_INVAILD:
ESP_LOGW(TAG, "Topic Alias invalid");
break;
case MQTT5_PACKET_TOO_LARGE:
ESP_LOGW(TAG, "Packet too large");
break;
case MQTT5_MESSAGE_RATE_TOO_HIGH:
ESP_LOGW(TAG, "Message rate too high");
break;
case MQTT5_QUOTA_EXCEEDED:
ESP_LOGW(TAG, "Quota exceeded");
break;
case MQTT5_ADMINISTRATIVE_ACTION:
ESP_LOGW(TAG, "Administrative action");
break;
case MQTT5_PAYLOAD_FORMAT_INVAILD:
ESP_LOGW(TAG, "Payload format invalid");
break;
case MQTT5_RETAIN_NOT_SUPPORT:
ESP_LOGW(TAG, "Retain not supported");
break;
case MQTT5_QOS_NOT_SUPPORT:
ESP_LOGW(TAG, "QoS not supported");
break;
case MQTT5_USE_ANOTHER_SERVER:
ESP_LOGW(TAG, "Use another server");
break;
case MQTT5_SERVER_MOVED:
ESP_LOGW(TAG, "Server moved");
break;
case MQTT5_SHARED_SUBSCR_NOT_SUPPORTED:
ESP_LOGW(TAG, "Shared Subscriptions not supported");
break;
case MQTT5_CONNECTION_RATE_EXCEEDED:
ESP_LOGW(TAG, "Connection rate exceeded");
break;
case MQTT5_MAXIMUM_CONNECT_TIME:
ESP_LOGW(TAG, "Maximum connect time");
break;
case MQTT5_SUBSCRIBE_IDENTIFIER_NOT_SUPPORT:
ESP_LOGW(TAG, "Subscription Identifiers not supported");
break;
case MQTT5_WILDCARD_SUBSCRIBE_NOT_SUPPORT:
ESP_LOGW(TAG, "Wildcard Subscriptions not supported");
break;
default:
ESP_LOGW(TAG, "Connection refused, Unknow reason");
break;
}
}
esp_err_t esp_mqtt5_client_subscribe_check(esp_mqtt5_client_handle_t client, int qos)
{
/* Check Server support QoS level */
if (client->mqtt5_config->server_resp_property_info.max_qos < qos) {
ESP_LOGE(TAG, "Server only support max QoS level %d", client->mqtt5_config->server_resp_property_info.max_qos);
return ESP_FAIL;
}
return ESP_OK;
}
esp_err_t esp_mqtt5_client_publish_check(esp_mqtt5_client_handle_t client, int qos, int retain)
{
/* Check Server support QoS level */
if (client->mqtt5_config->server_resp_property_info.max_qos < qos) {
ESP_LOGE(TAG, "Server only support max QoS level %d", client->mqtt5_config->server_resp_property_info.max_qos);
return ESP_FAIL;
}
/* Check Server support RETAIN */
if (!client->mqtt5_config->server_resp_property_info.retain_available && retain) {
ESP_LOGE(TAG, "Server not support retain");
return ESP_FAIL;
}
/* Flow control to check PUBLISH(No PUBACK or PUBCOMP received) packet sent count(Only record QoS1 and QoS2)*/
if (client->send_publish_packet_count >= client->mqtt5_config->server_resp_property_info.receive_maximum) {
ESP_LOGE(TAG, "Client send more than %d QoS1 and QoS2 PUBLISH packet without no ack", client->mqtt5_config->server_resp_property_info.receive_maximum);
return ESP_FAIL;
}
return ESP_OK;
}
void esp_mqtt5_client_destory(esp_mqtt5_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
if (client->mqtt5_config) {
free(client->mqtt5_config->will_property_info.content_type);
free(client->mqtt5_config->will_property_info.response_topic);
free(client->mqtt5_config->will_property_info.correlation_data);
free(client->mqtt5_config->server_resp_property_info.response_info);
esp_mqtt5_client_delete_topic_alias(client->mqtt5_config->peer_topic_alias);
esp_mqtt5_client_delete_user_property(client->mqtt5_config->connect_property_info.user_property);
esp_mqtt5_client_delete_user_property(client->mqtt5_config->will_property_info.user_property);
esp_mqtt5_client_delete_user_property(client->mqtt5_config->disconnect_property_info.user_property);
free(client->mqtt5_config);
}
free(client->event.property);
}
}
static void esp_mqtt5_client_delete_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle)
{
if (topic_alias_handle) {
mqtt5_topic_alias_item_t item, tmp;
STAILQ_FOREACH_SAFE(item, topic_alias_handle, next, tmp) {
STAILQ_REMOVE(topic_alias_handle, item, mqtt5_topic_alias, next);
free(item->topic);
free(item);
}
free(topic_alias_handle);
}
}
static esp_err_t esp_mqtt5_client_update_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle, uint16_t topic_alias, char *topic, size_t topic_len)
{
mqtt5_topic_alias_item_t item;
bool found = false;
STAILQ_FOREACH(item, topic_alias_handle, next) {
if (item->topic_alias == topic_alias) {
found = true;
break;
}
}
if (found) {
if ((item->topic_len != topic_len) || strncmp(topic, item->topic, topic_len)) {
free(item->topic);
item->topic = calloc(1, topic_len);
ESP_MEM_CHECK(TAG, item->topic, return ESP_FAIL);
memcpy(item->topic, topic, topic_len);
item->topic_len = topic_len;
}
} else {
item = calloc(1, sizeof(mqtt5_topic_alias_t));
ESP_MEM_CHECK(TAG, item, return ESP_FAIL);
item->topic_alias = topic_alias;
item->topic_len = topic_len;
item->topic = calloc(1, topic_len);
ESP_MEM_CHECK(TAG, item->topic, {
free(item);
return ESP_FAIL;
});
memcpy(item->topic, topic, topic_len);
STAILQ_INSERT_TAIL(topic_alias_handle, item, next);
}
return ESP_OK;
}
static char *esp_mqtt5_client_get_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle, uint16_t topic_alias, size_t *topic_length)
{
mqtt5_topic_alias_item_t item;
STAILQ_FOREACH(item, topic_alias_handle, next) {
if (item->topic_alias == topic_alias) {
*topic_length = item->topic_len;
return item->topic;
}
}
*topic_length = 0;
return NULL;
}
static esp_err_t esp_mqtt5_user_property_copy(mqtt5_user_property_handle_t user_property_new, const mqtt5_user_property_handle_t user_property_old)
{
if (!user_property_new || !user_property_old) {
ESP_LOGE(TAG, "Input is NULL");
return ESP_FAIL;
}
mqtt5_user_property_item_t old_item, new_item;
STAILQ_FOREACH(old_item, user_property_old, next) {
new_item = calloc(1, sizeof(mqtt5_user_property_t));
ESP_MEM_CHECK(TAG, new_item, return ESP_FAIL);
new_item->key = strdup(old_item->key);
ESP_MEM_CHECK(TAG, new_item->key, {
free(new_item);
return ESP_FAIL;
});
new_item->value = strdup(old_item->value);
ESP_MEM_CHECK(TAG, new_item->value, {
free(new_item->key);
free(new_item);
return ESP_FAIL;
});
STAILQ_INSERT_TAIL(user_property_new, new_item, next);
}
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_publish_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_publish_property_config_t *property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if(client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
/* Check topic alias less than server maximum topic alias */
if (property->topic_alias > client->mqtt5_config->server_resp_property_info.topic_alias_maximum) {
ESP_LOGE(TAG, "Topic alias %d is bigger than server support %d", property->topic_alias, client->mqtt5_config->server_resp_property_info.topic_alias_maximum);
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
client->mqtt5_config->publish_property_info = property;
MQTT_API_UNLOCK(client);
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_subscribe_property_config_t *property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
if (property->retain_handle > 2) {
ESP_LOGE(TAG, "retain_handle only support 0, 1, 2");
return -1;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (property->is_share_subscribe) {
if (property->no_local_flag) {
// MQTT-3.8.3-4 not allow that No Local bit to 1 on a Shared Subscription
ESP_LOGE(TAG, "Protocol error that no local flag set on shared subscription");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (!client->mqtt5_config->server_resp_property_info.shared_subscribe_available) {
ESP_LOGE(TAG, "MQTT broker not support shared subscribe");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (!property->share_name || !strlen(property->share_name)) {
ESP_LOGE(TAG, "Share name can't be empty for shared subscribe");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
}
client->mqtt5_config->subscribe_property_info = property;
MQTT_API_UNLOCK(client);
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_unsubscribe_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_unsubscribe_property_config_t *property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (property->is_share_subscribe) {
if (!client->mqtt5_config->server_resp_property_info.shared_subscribe_available) {
ESP_LOGE(TAG, "MQTT broker not support shared subscribe");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (!property->share_name || !strlen(property->share_name)) {
ESP_LOGE(TAG, "Share name can't be empty for shared subscribe");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
}
client->mqtt5_config->unsubscribe_property_info = property;
MQTT_API_UNLOCK(client);
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_disconnect_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_disconnect_property_config_t *property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (property) {
if (property->session_expiry_interval) {
client->mqtt5_config->disconnect_property_info.session_expiry_interval = property->session_expiry_interval;
}
if (property->disconnect_reason) {
client->mqtt5_config->disconnect_property_info.disconnect_reason = property->disconnect_reason;
}
if (property->user_property) {
esp_mqtt5_client_delete_user_property(client->mqtt5_config->disconnect_property_info.user_property);
client->mqtt5_config->disconnect_property_info.user_property = calloc(1, sizeof(struct mqtt5_user_property_list_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config->disconnect_property_info.user_property, {
MQTT_API_UNLOCK(client);
return ESP_ERR_NO_MEM;
});
STAILQ_INIT(client->mqtt5_config->disconnect_property_info.user_property);
if (esp_mqtt5_user_property_copy(client->mqtt5_config->disconnect_property_info.user_property, property->user_property) != ESP_OK) {
ESP_LOGE(TAG, "esp_mqtt5_user_property_copy fail");
free(client->mqtt5_config->disconnect_property_info.user_property);
client->mqtt5_config->disconnect_property_info.user_property = NULL;
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
}
}
MQTT_API_UNLOCK(client);
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_connect_property(esp_mqtt5_client_handle_t client, const esp_mqtt5_connection_property_config_t *connect_property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (connect_property) {
if (connect_property->session_expiry_interval) {
client->mqtt5_config->connect_property_info.session_expiry_interval = connect_property->session_expiry_interval;
}
if (connect_property->maximum_packet_size) {
if (connect_property->maximum_packet_size > client->mqtt_state.in_buffer_length) {
ESP_LOGW(TAG, "Connect maximum_packet_size property is over buffer_size(%d), Please first change it", client->mqtt_state.in_buffer_length);
MQTT_API_UNLOCK(client);
return ESP_FAIL;
} else {
client->mqtt5_config->connect_property_info.maximum_packet_size = connect_property->maximum_packet_size;
}
} else {
client->mqtt5_config->connect_property_info.maximum_packet_size = client->mqtt_state.in_buffer_length;
}
if (connect_property->receive_maximum) {
client->mqtt5_config->connect_property_info.receive_maximum = connect_property->receive_maximum;
}
if (connect_property->topic_alias_maximum) {
client->mqtt5_config->connect_property_info.topic_alias_maximum = connect_property->topic_alias_maximum;
if (!client->mqtt5_config->peer_topic_alias) {
client->mqtt5_config->peer_topic_alias = calloc(1, sizeof(struct mqtt5_topic_alias_list_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config->peer_topic_alias, goto _mqtt_set_config_failed);
STAILQ_INIT(client->mqtt5_config->peer_topic_alias);
}
}
if (connect_property->request_resp_info) {
client->mqtt5_config->connect_property_info.request_resp_info = connect_property->request_resp_info;
}
if (connect_property->request_problem_info) {
client->mqtt5_config->connect_property_info.request_problem_info = connect_property->request_problem_info;
}
if (connect_property->user_property) {
esp_mqtt5_client_delete_user_property(client->mqtt5_config->connect_property_info.user_property);
client->mqtt5_config->connect_property_info.user_property = calloc(1, sizeof(struct mqtt5_user_property_list_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config->connect_property_info.user_property, goto _mqtt_set_config_failed);
STAILQ_INIT(client->mqtt5_config->connect_property_info.user_property);
if (esp_mqtt5_user_property_copy(client->mqtt5_config->connect_property_info.user_property, connect_property->user_property) != ESP_OK) {
ESP_LOGE(TAG, "esp_mqtt5_user_property_copy fail");
goto _mqtt_set_config_failed;
}
}
if (connect_property->payload_format_indicator) {
client->mqtt5_config->will_property_info.payload_format_indicator = connect_property->payload_format_indicator;
}
if (connect_property->will_delay_interval) {
client->mqtt5_config->will_property_info.will_delay_interval = connect_property->will_delay_interval;
}
if (connect_property->message_expiry_interval) {
client->mqtt5_config->will_property_info.message_expiry_interval = connect_property->message_expiry_interval;
}
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(connect_property->content_type, &client->mqtt5_config->will_property_info.content_type), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(connect_property->response_topic, &client->mqtt5_config->will_property_info.response_topic), goto _mqtt_set_config_failed);
if (connect_property->correlation_data && connect_property->correlation_data_len) {
free(client->mqtt5_config->will_property_info.correlation_data);
client->mqtt5_config->will_property_info.correlation_data = malloc(connect_property->correlation_data_len);
ESP_MEM_CHECK(TAG, client->mqtt5_config->will_property_info.correlation_data, goto _mqtt_set_config_failed);
memcpy(client->mqtt5_config->will_property_info.correlation_data, connect_property->correlation_data, connect_property->correlation_data_len);
client->mqtt5_config->will_property_info.correlation_data_len = connect_property->correlation_data_len;
}
if (connect_property->will_user_property) {
esp_mqtt5_client_delete_user_property(client->mqtt5_config->will_property_info.user_property);
client->mqtt5_config->will_property_info.user_property = calloc(1, sizeof(struct mqtt5_user_property_list_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config->will_property_info.user_property, goto _mqtt_set_config_failed);
STAILQ_INIT(client->mqtt5_config->will_property_info.user_property);
if (esp_mqtt5_user_property_copy(client->mqtt5_config->will_property_info.user_property, connect_property->will_user_property) != ESP_OK) {
ESP_LOGE(TAG, "esp_mqtt5_user_property_copy fail");
goto _mqtt_set_config_failed;
}
}
}
MQTT_API_UNLOCK(client);
return ESP_OK;
_mqtt_set_config_failed:
esp_mqtt_destroy_config(client);
MQTT_API_UNLOCK(client);
return ESP_ERR_NO_MEM;
}
esp_err_t esp_mqtt5_client_set_user_property(mqtt5_user_property_handle_t *user_property, esp_mqtt5_user_property_item_t item[], uint8_t item_num)
{
if (!item_num || !item) {
ESP_LOGE(TAG, "Input value is NULL");
return ESP_FAIL;
}
if (!*user_property) {
*user_property = calloc(1, sizeof(struct mqtt5_user_property_list_t));
ESP_MEM_CHECK(TAG, *user_property, return ESP_ERR_NO_MEM);
STAILQ_INIT(*user_property);
}
for (int i = 0; i < item_num; i ++) {
if (item[i].key && item[i].value) {
mqtt5_user_property_item_t user_property_item = calloc(1, sizeof(mqtt5_user_property_t));
ESP_MEM_CHECK(TAG, user_property_item, goto err);
size_t key_len = strlen(item[i].key);
size_t value_len = strlen(item[i].value);
user_property_item->key = calloc(1, key_len + 1);
ESP_MEM_CHECK(TAG, user_property_item->key, {
free(user_property_item);
goto err;
});
memcpy(user_property_item->key, item[i].key, key_len);
user_property_item->key[key_len] = '\0';
user_property_item->value = calloc(1, value_len + 1);
ESP_MEM_CHECK(TAG, user_property_item->value, {
free(user_property_item->key);
free(user_property_item);
goto err;
});
memcpy(user_property_item->value, item[i].value, value_len);
user_property_item->value[value_len] = '\0';
STAILQ_INSERT_TAIL(*user_property, user_property_item, next);
}
}
return ESP_OK;
err:
esp_mqtt5_client_delete_user_property(*user_property);
*user_property = NULL;
return ESP_ERR_NO_MEM;
}
esp_err_t esp_mqtt5_client_get_user_property(mqtt5_user_property_handle_t user_property, esp_mqtt5_user_property_item_t *item, uint8_t *item_num)
{
int i = 0, j = 0;
if (user_property && item && *item_num) {
mqtt5_user_property_item_t user_property_item;
uint8_t num = *item_num;
STAILQ_FOREACH(user_property_item, user_property, next) {
if (i < num) {
size_t item_key_len = strlen(user_property_item->key);
size_t item_value_len = strlen(user_property_item->value);
char *key = calloc(1, item_key_len + 1);
ESP_MEM_CHECK(TAG, key, goto err);
memcpy(key, user_property_item->key, item_key_len);
key[item_key_len] = '\0';
char *value = calloc(1, item_value_len + 1);
ESP_MEM_CHECK(TAG, value, {
free(key);
goto err;
});
memcpy(value, user_property_item->value, item_value_len);
value[item_value_len] = '\0';
item[i].key = key;
item[i].value = value;
i ++;
} else {
break;
}
}
*item_num = i;
return ESP_OK;
} else {
ESP_LOGE(TAG, "Input value is NULL or item_num is 0");
return ESP_FAIL;
}
err:
for (j = 0; j < i; j ++) {
if (item[j].key) {
free((char *)item[j].key);
}
if (item[j].value) {
free((char *)item[j].value);
}
}
return ESP_ERR_NO_MEM;
}
uint8_t esp_mqtt5_client_get_user_property_count(mqtt5_user_property_handle_t user_property)
{
uint8_t count = 0;
if (user_property) {
mqtt5_user_property_item_t item;
STAILQ_FOREACH(item, user_property, next) {
count ++;
}
}
return count;
}
void esp_mqtt5_client_delete_user_property(mqtt5_user_property_handle_t user_property)
{
if (user_property) {
mqtt5_user_property_item_t item, tmp;
STAILQ_FOREACH_SAFE(item, user_property, next, tmp) {
STAILQ_REMOVE(user_property, item, mqtt5_user_property, next);
free(item->key);
free(item->value);
free(item);
}
}
free(user_property);
}

915
mqtt_client.c
View File

@ -1,36 +1,4 @@
#include <stdio.h>
#include <stdlib.h>
#include "esp_err.h"
#include "platform.h"
#include "esp_event.h"
#include "mqtt_client.h"
#include "mqtt_msg.h"
#ifdef MQTT_PROTOCOL_5
#include "mqtt5_msg.h"
#endif
#include "esp_transport.h"
#include "esp_transport_tcp.h"
#include "esp_transport_ssl.h"
#include "esp_transport_ws.h"
#include "esp_log.h"
#include "mqtt_outbox.h"
#include "freertos/event_groups.h"
#include <errno.h>
#include <string.h>
#include "mqtt_supported_features.h"
/* using uri parser */
#include "http_parser.h"
#ifdef MQTT_DISABLE_API_LOCKS
# define MQTT_API_LOCK(c)
# define MQTT_API_UNLOCK(c)
#else
# define MQTT_API_LOCK(c) xSemaphoreTakeRecursive(c->api_lock, portMAX_DELAY)
# define MQTT_API_UNLOCK(c) xSemaphoreGiveRecursive(c->api_lock)
#endif /* MQTT_USE_API_LOCKS */
#include "mqtt_client_priv.h"
_Static_assert(sizeof(uint64_t) == sizeof(outbox_tick_t), "mqtt-client tick type size different from outbox tick type");
#ifdef ESP_EVENT_ANY_ID
@ -52,134 +20,12 @@ ESP_EVENT_DEFINE_BASE(MQTT_EVENTS);
#define MQTT_OVER_WS_SCHEME "ws"
#define MQTT_OVER_WSS_SCHEME "wss"
typedef struct mqtt_state {
uint8_t *in_buffer;
uint8_t *out_buffer;
int in_buffer_length;
int out_buffer_length;
size_t message_length;
size_t in_buffer_read_len;
mqtt_message_t *outbound_message;
mqtt_connection_t mqtt_connection;
uint16_t pending_msg_id;
int pending_msg_type;
int pending_publish_qos;
int pending_msg_count;
} mqtt_state_t;
typedef struct {
mqtt_event_callback_t event_handle;
esp_event_loop_handle_t event_loop_handle;
int task_stack;
int task_prio;
char *uri;
char *host;
char *path;
char *scheme;
int port;
bool auto_reconnect;
void *user_context;
int network_timeout_ms;
int refresh_connection_after_ms;
int reconnect_timeout_ms;
char **alpn_protos;
int num_alpn_protos;
char *clientkey_password;
int clientkey_password_len;
bool use_global_ca_store;
esp_err_t ((*crt_bundle_attach)(void *conf));
const char *cacert_buf;
size_t cacert_bytes;
const char *clientcert_buf;
size_t clientcert_bytes;
const char *clientkey_buf;
size_t clientkey_bytes;
const struct psk_key_hint *psk_hint_key;
bool skip_cert_common_name_check;
bool use_secure_element;
void *ds_data;
int message_retransmit_timeout;
} mqtt_config_storage_t;
typedef enum {
MQTT_STATE_INIT = 0,
MQTT_STATE_DISCONNECTED,
MQTT_STATE_CONNECTED,
MQTT_STATE_WAIT_RECONNECT,
} mqtt_client_state_t;
#ifdef MQTT_PROTOCOL_5
typedef struct mqtt5_topic_alias {
char *topic;
uint16_t topic_len;
uint16_t topic_alias;
STAILQ_ENTRY(mqtt5_topic_alias) next;
} mqtt5_topic_alias_t;
STAILQ_HEAD(mqtt5_topic_alias_list_t, mqtt5_topic_alias);
typedef struct mqtt5_topic_alias_list_t *mqtt5_topic_alias_handle_t;
typedef struct mqtt5_topic_alias *mqtt5_topic_alias_item_t;
typedef struct {
esp_mqtt5_connection_property_storage_t connect_property_info;
esp_mqtt5_connection_will_property_storage_t will_property_info;
esp_mqtt5_connection_server_resp_property_t server_resp_property_info;
esp_mqtt5_disconnect_property_config_t disconnect_property_info;
const esp_mqtt5_publish_property_config_t *publish_property_info;
const esp_mqtt5_subscribe_property_config_t *subscribe_property_info;
const esp_mqtt5_unsubscribe_property_config_t *unsubscribe_property_info;
mqtt5_topic_alias_handle_t peer_topic_alias;
} mqtt5_config_storage_t;
static void esp_mqtt5_flow_control(esp_mqtt_client_handle_t client);
static void esp_mqtt5_parse_pubcomp(esp_mqtt_client_handle_t client);
static void esp_mqtt5_parse_puback(esp_mqtt_client_handle_t client);
static void esp_mqtt5_parse_unsuback(esp_mqtt_client_handle_t client);
static void esp_mqtt5_parse_suback(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt5_parse_connack(esp_mqtt_client_handle_t client, int *connect_rsp_code);
static void esp_mqtt5_client_destory(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt5_client_publish_check(esp_mqtt_client_handle_t client, int qos, int retain);
static esp_err_t esp_mqtt5_client_subscribe_check(esp_mqtt_client_handle_t client, int qos);
static void esp_mqtt5_print_error_code(esp_mqtt_client_handle_t client, int code);
static esp_err_t esp_mqtt5_create_default_config(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt5_get_publish_data(esp_mqtt_client_handle_t client, uint8_t *msg_buf, size_t msg_read_len, char **msg_topic, size_t *msg_topic_len, char **msg_data, size_t *msg_data_len);
static esp_err_t esp_mqtt5_client_update_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle, uint16_t topic_alias, char *topic, size_t topic_len);
static char *esp_mqtt5_client_get_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle, uint16_t topic_alias, size_t *topic_length);
static void esp_mqtt5_client_delete_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle);
static esp_err_t esp_mqtt5_user_property_copy(mqtt5_user_property_handle_t user_property_new, const mqtt5_user_property_handle_t user_property_old);
#endif
struct esp_mqtt_client {
esp_transport_list_handle_t transport_list;
esp_transport_handle_t transport;
mqtt_config_storage_t *config;
mqtt_state_t mqtt_state;
mqtt_connect_info_t connect_info;
mqtt_client_state_t state;
uint64_t refresh_connection_tick;
int64_t keepalive_tick;
uint64_t reconnect_tick;
#ifdef MQTT_PROTOCOL_5
mqtt5_config_storage_t *mqtt5_config;
uint16_t send_publish_packet_count; // This is for MQTT v5.0 flow control
#endif
int wait_timeout_ms;
int auto_reconnect;
esp_mqtt_event_t event;
bool run;
bool wait_for_ping_resp;
outbox_handle_t outbox;
EventGroupHandle_t status_bits;
SemaphoreHandle_t api_lock;
TaskHandle_t task_handle;
};
const static int STOPPED_BIT = (1 << 0);
const static int RECONNECT_BIT = (1 << 1);
const static int DISCONNECT_BIT = (1 << 2);
static esp_err_t esp_mqtt_dispatch_event(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt_dispatch_event_with_msgid(esp_mqtt_client_handle_t client);
static void esp_mqtt_destroy_config(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_ms);
static void esp_mqtt_abort_connection(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt_client_ping(esp_mqtt_client_handle_t client);
@ -409,7 +255,7 @@ static esp_err_t esp_mqtt_check_cfg_conflict(const mqtt_config_storage_t *cfg, c
return ret;
}
static bool set_if_config(char const *const new_config, char **old_config)
bool esp_mqtt_set_if_config(char const *const new_config, char **old_config)
{
if (new_config) {
free(*old_config);
@ -529,14 +375,14 @@ esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_cl
}
err = ESP_ERR_NO_MEM;
ESP_MEM_CHECK(TAG, set_if_config(config->host, &client->config->host), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(config->path, &client->config->path), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(config->username, &client->connect_info.username), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(config->password, &client->connect_info.password), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(config->host, &client->config->host), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(config->path, &client->config->path), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(config->username, &client->connect_info.username), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(config->password, &client->connect_info.password), goto _mqtt_set_config_failed);
if (!config->set_null_client_id) {
if (config->client_id) {
ESP_MEM_CHECK(TAG, set_if_config(config->client_id, &client->connect_info.client_id), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(config->client_id, &client->connect_info.client_id), goto _mqtt_set_config_failed);
} else if (client->connect_info.client_id == NULL) {
client->connect_info.client_id = platform_create_id_string();
}
@ -544,8 +390,8 @@ esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_cl
ESP_LOGD(TAG, "MQTT client_id=%s", client->connect_info.client_id);
}
ESP_MEM_CHECK(TAG, set_if_config(config->uri, &client->config->uri), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(config->lwt_topic, &client->connect_info.will_topic), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(config->uri, &client->config->uri), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, esp_mqtt_set_if_config(config->lwt_topic, &client->connect_info.will_topic), goto _mqtt_set_config_failed);
if (config->lwt_msg_len && config->lwt_msg) {
free(client->connect_info.will_message);
@ -716,7 +562,7 @@ _mqtt_set_config_failed:
return err;
}
static void esp_mqtt_destroy_config(esp_mqtt_client_handle_t client)
void esp_mqtt_destroy_config(esp_mqtt_client_handle_t client)
{
if (client->config == NULL) {
return;
@ -2249,744 +2095,3 @@ int esp_mqtt_client_get_outbox_size(esp_mqtt_client_handle_t client)
return outbox_size;
}
#ifdef MQTT_PROTOCOL_5
static void esp_mqtt5_flow_control(esp_mqtt_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
int msg_type = mqtt5_get_type(client->mqtt_state.outbound_message->data);
if (msg_type == MQTT_MSG_TYPE_PUBLISH) {
int msg_qos = mqtt5_get_qos(client->mqtt_state.outbound_message->data);
if (msg_qos > 0) {
client->send_publish_packet_count ++;
ESP_LOGD(TAG, "Sent (%d) qos > 0 publish packet without ack", client->send_publish_packet_count);
}
}
}
}
static void esp_mqtt5_parse_pubcomp(esp_mqtt_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBCOMP return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.data = mqtt5_get_pubcomp_data(client->mqtt_state.in_buffer, &msg_data_len, &client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
client->event.current_data_offset = 0;
}
}
static void esp_mqtt5_parse_puback(esp_mqtt_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.data = mqtt5_get_puback_data(client->mqtt_state.in_buffer, &msg_data_len, &client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
client->event.current_data_offset = 0;
client->send_publish_packet_count --;
}
}
static void esp_mqtt5_parse_unsuback(esp_mqtt_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGI(TAG, "MQTT_MSG_TYPE_UNSUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.data = mqtt5_get_unsuback_data(client->mqtt_state.in_buffer, &msg_data_len, &client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
client->event.current_data_offset = 0;
}
}
static void esp_mqtt5_parse_suback(esp_mqtt_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGI(TAG, "MQTT_MSG_TYPE_SUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
}
}
static esp_err_t esp_mqtt5_parse_connack(esp_mqtt_client_handle_t client, int *connect_rsp_code)
{
size_t len = client->mqtt_state.in_buffer_read_len;
client->mqtt_state.in_buffer_read_len = 0;
uint8_t ack_flag = 0;
if (mqtt5_msg_parse_connack_property(client->mqtt_state.in_buffer, len, &client->connect_info, &client->mqtt5_config->connect_property_info, &client->mqtt5_config->server_resp_property_info, connect_rsp_code, &ack_flag, &client->event.property->user_property) != ESP_OK) {
ESP_LOGE(TAG, "Failed to parse CONNACK packet");
return ESP_FAIL;
}
if (*connect_rsp_code == MQTT_CONNECTION_ACCEPTED) {
ESP_LOGD(TAG, "Connected");
client->event.session_present = ack_flag & 0x01;
return ESP_OK;
}
esp_mqtt5_print_error_code(client, *connect_rsp_code);
return ESP_FAIL;
}
static esp_err_t esp_mqtt5_get_publish_data(esp_mqtt_client_handle_t client, uint8_t *msg_buf, size_t msg_read_len, char **msg_topic, size_t *msg_topic_len, char **msg_data, size_t *msg_data_len)
{
// get property
uint16_t property_len = 0;
esp_mqtt5_publish_resp_property_t property = {0};
*msg_data = mqtt5_get_publish_property_payload(msg_buf, msg_read_len, msg_topic, msg_topic_len, &property, &property_len, msg_data_len, &client->event.property->user_property);
if (*msg_data_len == 0 || *msg_data == NULL) {
ESP_LOGE(TAG, "%s: mqtt5_get_publish_property_payload() failed", __func__);
return ESP_FAIL;
}
if (property.topic_alias > client->mqtt5_config->connect_property_info.topic_alias_maximum) {
ESP_LOGE(TAG, "%s: Broker response topic alias %d is over the max topic alias %d", __func__, property.topic_alias, client->mqtt5_config->connect_property_info.topic_alias_maximum);
return ESP_FAIL;
}
if (property.topic_alias) {
if (*msg_topic_len == 0) {
ESP_LOGI(TAG, "Publish topic is empty, use topic alias");
*msg_topic = esp_mqtt5_client_get_topic_alias(client->mqtt5_config->peer_topic_alias, property.topic_alias, msg_topic_len);
if (!*msg_topic) {
ESP_LOGE(TAG, "%s: esp_mqtt5_client_get_topic_alias() failed", __func__);
return ESP_FAIL;
}
} else {
if (esp_mqtt5_client_update_topic_alias(client->mqtt5_config->peer_topic_alias, property.topic_alias, *msg_topic, *msg_topic_len) != ESP_OK) {
ESP_LOGE(TAG, "%s: esp_mqtt5_client_update_topic_alias() failed", __func__);
return ESP_FAIL;
}
}
}
client->event.property->payload_format_indicator = property.payload_format_indicator;
client->event.property->response_topic = property.response_topic;
client->event.property->response_topic_len = property.response_topic_len;
client->event.property->correlation_data = property.correlation_data;
client->event.property->correlation_data_len = property.correlation_data_len;
client->event.property->content_type = property.content_type;
client->event.property->content_type_len = property.content_type_len;
return ESP_OK;
}
static esp_err_t esp_mqtt5_create_default_config(esp_mqtt_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
client->event.property = calloc(1, sizeof(esp_mqtt5_event_property_t));
ESP_MEM_CHECK(TAG, client->event.property, return ESP_FAIL)
client->mqtt5_config = calloc(1, sizeof(mqtt5_config_storage_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config, return ESP_FAIL)
client->mqtt5_config->server_resp_property_info.max_qos = 2;
client->mqtt5_config->server_resp_property_info.retain_available = true;
client->mqtt5_config->server_resp_property_info.wildcard_subscribe_available = true;
client->mqtt5_config->server_resp_property_info.subscribe_identifiers_available = true;
client->mqtt5_config->server_resp_property_info.shared_subscribe_available = true;
client->mqtt5_config->server_resp_property_info.receive_maximum = 65535;
}
return ESP_OK;
}
static void esp_mqtt5_print_error_code(esp_mqtt_client_handle_t client, int code)
{
switch (code) {
case MQTT5_UNSPECIFIED_ERROR:
ESP_LOGW(TAG, "Unspecified error");
break;
case MQTT5_MALFORMED_PACKET:
ESP_LOGW(TAG, "Malformed Packet");
break;
case MQTT5_PROTOCOL_ERROR:
ESP_LOGW(TAG, "Protocol Error");
break;
case MQTT5_IMPLEMENT_SPECIFIC_ERROR:
ESP_LOGW(TAG, "Implementation specific error");
break;
case MQTT5_UNSUPPORTED_PROTOCOL_VER:
ESP_LOGW(TAG, "Unsupported Protocol Version");
break;
case MQTT5_INVAILD_CLIENT_ID:
ESP_LOGW(TAG, "Client Identifier not valid");
break;
case MQTT5_BAD_USERNAME_OR_PWD:
ESP_LOGW(TAG, "Bad User Name or Password");
break;
case MQTT5_NOT_AUTHORIZED:
ESP_LOGW(TAG, "Not authorized");
break;
case MQTT5_SERVER_UNAVAILABLE:
ESP_LOGW(TAG, "Server unavailable");
break;
case MQTT5_SERVER_BUSY:
ESP_LOGW(TAG, "Server busy");
break;
case MQTT5_BANNED:
ESP_LOGW(TAG, "Banned");
break;
case MQTT5_SERVER_SHUTTING_DOWN:
ESP_LOGW(TAG, "Server shutting down");
break;
case MQTT5_BAD_AUTH_METHOD:
ESP_LOGW(TAG, "Bad authentication method");
break;
case MQTT5_KEEP_ALIVE_TIMEOUT:
ESP_LOGW(TAG, "Keep Alive timeout");
break;
case MQTT5_SESSION_TAKEN_OVER:
ESP_LOGW(TAG, "Session taken over");
break;
case MQTT5_TOPIC_FILTER_INVAILD:
ESP_LOGW(TAG, "Topic Filter invalid");
break;
case MQTT5_TOPIC_NAME_INVAILD:
ESP_LOGW(TAG, "Topic Name invalid");
break;
case MQTT5_PACKET_IDENTIFIER_IN_USE:
ESP_LOGW(TAG, "Packet Identifier in use");
break;
case MQTT5_PACKET_IDENTIFIER_NOT_FOUND:
ESP_LOGW(TAG, "Packet Identifier not found");
break;
case MQTT5_RECEIVE_MAXIMUM_EXCEEDED:
ESP_LOGW(TAG, "Receive Maximum exceeded");
break;
case MQTT5_TOPIC_ALIAS_INVAILD:
ESP_LOGW(TAG, "Topic Alias invalid");
break;
case MQTT5_PACKET_TOO_LARGE:
ESP_LOGW(TAG, "Packet too large");
break;
case MQTT5_MESSAGE_RATE_TOO_HIGH:
ESP_LOGW(TAG, "Message rate too high");
break;
case MQTT5_QUOTA_EXCEEDED:
ESP_LOGW(TAG, "Quota exceeded");
break;
case MQTT5_ADMINISTRATIVE_ACTION:
ESP_LOGW(TAG, "Administrative action");
break;
case MQTT5_PAYLOAD_FORMAT_INVAILD:
ESP_LOGW(TAG, "Payload format invalid");
break;
case MQTT5_RETAIN_NOT_SUPPORT:
ESP_LOGW(TAG, "Retain not supported");
break;
case MQTT5_QOS_NOT_SUPPORT:
ESP_LOGW(TAG, "QoS not supported");
break;
case MQTT5_USE_ANOTHER_SERVER:
ESP_LOGW(TAG, "Use another server");
break;
case MQTT5_SERVER_MOVED:
ESP_LOGW(TAG, "Server moved");
break;
case MQTT5_SHARED_SUBSCR_NOT_SUPPORTED:
ESP_LOGW(TAG, "Shared Subscriptions not supported");
break;
case MQTT5_CONNECTION_RATE_EXCEEDED:
ESP_LOGW(TAG, "Connection rate exceeded");
break;
case MQTT5_MAXIMUM_CONNECT_TIME:
ESP_LOGW(TAG, "Maximum connect time");
break;
case MQTT5_SUBSCRIBE_IDENTIFIER_NOT_SUPPORT:
ESP_LOGW(TAG, "Subscription Identifiers not supported");
break;
case MQTT5_WILDCARD_SUBSCRIBE_NOT_SUPPORT:
ESP_LOGW(TAG, "Wildcard Subscriptions not supported");
break;
default:
ESP_LOGW(TAG, "Connection refused, Unknow reason");
break;
}
}
static esp_err_t esp_mqtt5_client_subscribe_check(esp_mqtt_client_handle_t client, int qos)
{
/* Check Server support QoS level */
if (client->mqtt5_config->server_resp_property_info.max_qos < qos) {
ESP_LOGE(TAG, "Server only support max QoS level %d", client->mqtt5_config->server_resp_property_info.max_qos);
return ESP_FAIL;
}
return ESP_OK;
}
static esp_err_t esp_mqtt5_client_publish_check(esp_mqtt_client_handle_t client, int qos, int retain)
{
/* Check Server support QoS level */
if (client->mqtt5_config->server_resp_property_info.max_qos < qos) {
ESP_LOGE(TAG, "Server only support max QoS level %d", client->mqtt5_config->server_resp_property_info.max_qos);
return ESP_FAIL;
}
/* Check Server support RETAIN */
if (!client->mqtt5_config->server_resp_property_info.retain_available && retain) {
ESP_LOGE(TAG, "Server not support retain");
return ESP_FAIL;
}
/* Flow control to check PUBLISH(No PUBACK or PUBCOMP received) packet sent count(Only record QoS1 and QoS2)*/
if (client->send_publish_packet_count >= client->mqtt5_config->server_resp_property_info.receive_maximum) {
ESP_LOGE(TAG, "Client send more than %d QoS1 and QoS2 PUBLISH packet without no ack", client->mqtt5_config->server_resp_property_info.receive_maximum);
return ESP_FAIL;
}
return ESP_OK;
}
static void esp_mqtt5_client_destory(esp_mqtt_client_handle_t client)
{
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
if (client->mqtt5_config) {
free(client->mqtt5_config->will_property_info.content_type);
free(client->mqtt5_config->will_property_info.response_topic);
free(client->mqtt5_config->will_property_info.correlation_data);
free(client->mqtt5_config->server_resp_property_info.response_info);
esp_mqtt5_client_delete_topic_alias(client->mqtt5_config->peer_topic_alias);
esp_mqtt5_client_delete_user_property(client->mqtt5_config->connect_property_info.user_property);
esp_mqtt5_client_delete_user_property(client->mqtt5_config->will_property_info.user_property);
esp_mqtt5_client_delete_user_property(client->mqtt5_config->disconnect_property_info.user_property);
free(client->mqtt5_config);
}
free(client->event.property);
}
}
static void esp_mqtt5_client_delete_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle)
{
if (topic_alias_handle) {
mqtt5_topic_alias_item_t item, tmp;
STAILQ_FOREACH_SAFE(item, topic_alias_handle, next, tmp) {
STAILQ_REMOVE(topic_alias_handle, item, mqtt5_topic_alias, next);
free(item->topic);
free(item);
}
free(topic_alias_handle);
}
}
static esp_err_t esp_mqtt5_client_update_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle, uint16_t topic_alias, char *topic, size_t topic_len)
{
mqtt5_topic_alias_item_t item;
bool found = false;
STAILQ_FOREACH(item, topic_alias_handle, next) {
if (item->topic_alias == topic_alias) {
found = true;
break;
}
}
if (found) {
if ((item->topic_len != topic_len) || strncmp(topic, item->topic, topic_len)) {
free(item->topic);
item->topic = calloc(1, topic_len);
ESP_MEM_CHECK(TAG, item->topic, return ESP_FAIL);
memcpy(item->topic, topic, topic_len);
item->topic_len = topic_len;
}
} else {
item = calloc(1, sizeof(mqtt5_topic_alias_t));
ESP_MEM_CHECK(TAG, item, return ESP_FAIL);
item->topic_alias = topic_alias;
item->topic_len = topic_len;
item->topic = calloc(1, topic_len);
ESP_MEM_CHECK(TAG, item->topic, {
free(item);
return ESP_FAIL;
});
memcpy(item->topic, topic, topic_len);
STAILQ_INSERT_TAIL(topic_alias_handle, item, next);
}
return ESP_OK;
}
static char *esp_mqtt5_client_get_topic_alias(mqtt5_topic_alias_handle_t topic_alias_handle, uint16_t topic_alias, size_t *topic_length)
{
mqtt5_topic_alias_item_t item;
STAILQ_FOREACH(item, topic_alias_handle, next) {
if (item->topic_alias == topic_alias) {
*topic_length = item->topic_len;
return item->topic;
}
}
*topic_length = 0;
return NULL;
}
static esp_err_t esp_mqtt5_user_property_copy(mqtt5_user_property_handle_t user_property_new, const mqtt5_user_property_handle_t user_property_old)
{
if (!user_property_new || !user_property_old) {
ESP_LOGE(TAG, "Input is NULL");
return ESP_FAIL;
}
mqtt5_user_property_item_t old_item, new_item;
STAILQ_FOREACH(old_item, user_property_old, next) {
new_item = calloc(1, sizeof(mqtt5_user_property_t));
ESP_MEM_CHECK(TAG, new_item, return ESP_FAIL);
new_item->key = strdup(old_item->key);
ESP_MEM_CHECK(TAG, new_item->key, {
free(new_item);
return ESP_FAIL;
});
new_item->value = strdup(old_item->value);
ESP_MEM_CHECK(TAG, new_item->value, {
free(new_item->key);
free(new_item);
return ESP_FAIL;
});
STAILQ_INSERT_TAIL(user_property_new, new_item, next);
}
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_publish_property(esp_mqtt_client_handle_t client, const esp_mqtt5_publish_property_config_t *property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if(client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
/* Check topic alias less than server maximum topic alias */
if (property->topic_alias > client->mqtt5_config->server_resp_property_info.topic_alias_maximum) {
ESP_LOGE(TAG, "Topic alias %d is bigger than server support %d", property->topic_alias, client->mqtt5_config->server_resp_property_info.topic_alias_maximum);
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
client->mqtt5_config->publish_property_info = property;
MQTT_API_UNLOCK(client);
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt_client_handle_t client, const esp_mqtt5_subscribe_property_config_t *property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
if (property->retain_handle > 2) {
ESP_LOGE(TAG, "retain_handle only support 0, 1, 2");
return -1;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (property->is_share_subscribe) {
if (property->no_local_flag) {
// MQTT-3.8.3-4 not allow that No Local bit to 1 on a Shared Subscription
ESP_LOGE(TAG, "Protocol error that no local flag set on shared subscription");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (!client->mqtt5_config->server_resp_property_info.shared_subscribe_available) {
ESP_LOGE(TAG, "MQTT broker not support shared subscribe");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (!property->share_name || !strlen(property->share_name)) {
ESP_LOGE(TAG, "Share name can't be empty for shared subscribe");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
}
client->mqtt5_config->subscribe_property_info = property;
MQTT_API_UNLOCK(client);
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_unsubscribe_property(esp_mqtt_client_handle_t client, const esp_mqtt5_unsubscribe_property_config_t *property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (property->is_share_subscribe) {
if (!client->mqtt5_config->server_resp_property_info.shared_subscribe_available) {
ESP_LOGE(TAG, "MQTT broker not support shared subscribe");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (!property->share_name || !strlen(property->share_name)) {
ESP_LOGE(TAG, "Share name can't be empty for shared subscribe");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
}
client->mqtt5_config->unsubscribe_property_info = property;
MQTT_API_UNLOCK(client);
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_disconnect_property(esp_mqtt_client_handle_t client, const esp_mqtt5_disconnect_property_config_t *property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (property) {
if (property->session_expiry_interval) {
client->mqtt5_config->disconnect_property_info.session_expiry_interval = property->session_expiry_interval;
}
if (property->disconnect_reason) {
client->mqtt5_config->disconnect_property_info.disconnect_reason = property->disconnect_reason;
}
if (property->user_property) {
esp_mqtt5_client_delete_user_property(client->mqtt5_config->disconnect_property_info.user_property);
client->mqtt5_config->disconnect_property_info.user_property = calloc(1, sizeof(struct mqtt5_user_property_list_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config->disconnect_property_info.user_property, {
MQTT_API_UNLOCK(client);
return ESP_ERR_NO_MEM;
});
STAILQ_INIT(client->mqtt5_config->disconnect_property_info.user_property);
if (esp_mqtt5_user_property_copy(client->mqtt5_config->disconnect_property_info.user_property, property->user_property) != ESP_OK) {
ESP_LOGE(TAG, "esp_mqtt5_user_property_copy fail");
free(client->mqtt5_config->disconnect_property_info.user_property);
client->mqtt5_config->disconnect_property_info.user_property = NULL;
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
}
}
MQTT_API_UNLOCK(client);
return ESP_OK;
}
esp_err_t esp_mqtt5_client_set_connect_property(esp_mqtt_client_handle_t client, const esp_mqtt5_connection_property_config_t *connect_property)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
MQTT_API_LOCK(client);
/* Check protocol version */
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
ESP_LOGE(TAG, "MQTT protocol version is not v5");
MQTT_API_UNLOCK(client);
return ESP_FAIL;
}
if (connect_property) {
if (connect_property->session_expiry_interval) {
client->mqtt5_config->connect_property_info.session_expiry_interval = connect_property->session_expiry_interval;
}
if (connect_property->maximum_packet_size) {
if (connect_property->maximum_packet_size > client->mqtt_state.in_buffer_length) {
ESP_LOGW(TAG, "Connect maximum_packet_size property is over buffer_size(%d), Please first change it", client->mqtt_state.in_buffer_length);
MQTT_API_UNLOCK(client);
return ESP_FAIL;
} else {
client->mqtt5_config->connect_property_info.maximum_packet_size = connect_property->maximum_packet_size;
}
} else {
client->mqtt5_config->connect_property_info.maximum_packet_size = client->mqtt_state.in_buffer_length;
}
if (connect_property->receive_maximum) {
client->mqtt5_config->connect_property_info.receive_maximum = connect_property->receive_maximum;
}
if (connect_property->topic_alias_maximum) {
client->mqtt5_config->connect_property_info.topic_alias_maximum = connect_property->topic_alias_maximum;
if (!client->mqtt5_config->peer_topic_alias) {
client->mqtt5_config->peer_topic_alias = calloc(1, sizeof(struct mqtt5_topic_alias_list_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config->peer_topic_alias, goto _mqtt_set_config_failed);
STAILQ_INIT(client->mqtt5_config->peer_topic_alias);
}
}
if (connect_property->request_resp_info) {
client->mqtt5_config->connect_property_info.request_resp_info = connect_property->request_resp_info;
}
if (connect_property->request_problem_info) {
client->mqtt5_config->connect_property_info.request_problem_info = connect_property->request_problem_info;
}
if (connect_property->user_property) {
esp_mqtt5_client_delete_user_property(client->mqtt5_config->connect_property_info.user_property);
client->mqtt5_config->connect_property_info.user_property = calloc(1, sizeof(struct mqtt5_user_property_list_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config->connect_property_info.user_property, goto _mqtt_set_config_failed);
STAILQ_INIT(client->mqtt5_config->connect_property_info.user_property);
if (esp_mqtt5_user_property_copy(client->mqtt5_config->connect_property_info.user_property, connect_property->user_property) != ESP_OK) {
ESP_LOGE(TAG, "esp_mqtt5_user_property_copy fail");
goto _mqtt_set_config_failed;
}
}
if (connect_property->payload_format_indicator) {
client->mqtt5_config->will_property_info.payload_format_indicator = connect_property->payload_format_indicator;
}
if (connect_property->will_delay_interval) {
client->mqtt5_config->will_property_info.will_delay_interval = connect_property->will_delay_interval;
}
if (connect_property->message_expiry_interval) {
client->mqtt5_config->will_property_info.message_expiry_interval = connect_property->message_expiry_interval;
}
ESP_MEM_CHECK(TAG, set_if_config(connect_property->content_type, &client->mqtt5_config->will_property_info.content_type), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(connect_property->response_topic, &client->mqtt5_config->will_property_info.response_topic), goto _mqtt_set_config_failed);
if (connect_property->correlation_data && connect_property->correlation_data_len) {
free(client->mqtt5_config->will_property_info.correlation_data);
client->mqtt5_config->will_property_info.correlation_data = malloc(connect_property->correlation_data_len);
ESP_MEM_CHECK(TAG, client->mqtt5_config->will_property_info.correlation_data, goto _mqtt_set_config_failed);
memcpy(client->mqtt5_config->will_property_info.correlation_data, connect_property->correlation_data, connect_property->correlation_data_len);
client->mqtt5_config->will_property_info.correlation_data_len = connect_property->correlation_data_len;
}
if (connect_property->will_user_property) {
esp_mqtt5_client_delete_user_property(client->mqtt5_config->will_property_info.user_property);
client->mqtt5_config->will_property_info.user_property = calloc(1, sizeof(struct mqtt5_user_property_list_t));
ESP_MEM_CHECK(TAG, client->mqtt5_config->will_property_info.user_property, goto _mqtt_set_config_failed);
STAILQ_INIT(client->mqtt5_config->will_property_info.user_property);
if (esp_mqtt5_user_property_copy(client->mqtt5_config->will_property_info.user_property, connect_property->will_user_property) != ESP_OK) {
ESP_LOGE(TAG, "esp_mqtt5_user_property_copy fail");
goto _mqtt_set_config_failed;
}
}
}
MQTT_API_UNLOCK(client);
return ESP_OK;
_mqtt_set_config_failed:
esp_mqtt_destroy_config(client);
MQTT_API_UNLOCK(client);
return ESP_ERR_NO_MEM;
}
esp_err_t esp_mqtt5_client_set_user_property(mqtt5_user_property_handle_t *user_property, esp_mqtt5_user_property_item_t item[], uint8_t item_num)
{
if (!item_num || !item) {
ESP_LOGE(TAG, "Input value is NULL");
return ESP_FAIL;
}
if (!*user_property) {
*user_property = calloc(1, sizeof(struct mqtt5_user_property_list_t));
ESP_MEM_CHECK(TAG, *user_property, return ESP_ERR_NO_MEM);
STAILQ_INIT(*user_property);
}
for (int i = 0; i < item_num; i ++) {
if (item[i].key && item[i].value) {
mqtt5_user_property_item_t user_property_item = calloc(1, sizeof(mqtt5_user_property_t));
ESP_MEM_CHECK(TAG, user_property_item, goto err);
size_t key_len = strlen(item[i].key);
size_t value_len = strlen(item[i].value);
user_property_item->key = calloc(1, key_len + 1);
ESP_MEM_CHECK(TAG, user_property_item->key, {
free(user_property_item);
goto err;
});
memcpy(user_property_item->key, item[i].key, key_len);
user_property_item->key[key_len] = '\0';
user_property_item->value = calloc(1, value_len + 1);
ESP_MEM_CHECK(TAG, user_property_item->value, {
free(user_property_item->key);
free(user_property_item);
goto err;
});
memcpy(user_property_item->value, item[i].value, value_len);
user_property_item->value[value_len] = '\0';
STAILQ_INSERT_TAIL(*user_property, user_property_item, next);
}
}
return ESP_OK;
err:
esp_mqtt5_client_delete_user_property(*user_property);
*user_property = NULL;
return ESP_ERR_NO_MEM;
}
esp_err_t esp_mqtt5_client_get_user_property(mqtt5_user_property_handle_t user_property, esp_mqtt5_user_property_item_t *item, uint8_t *item_num)
{
int i = 0, j = 0;
if (user_property && item && *item_num) {
mqtt5_user_property_item_t user_property_item;
uint8_t num = *item_num;
STAILQ_FOREACH(user_property_item, user_property, next) {
if (i < num) {
size_t item_key_len = strlen(user_property_item->key);
size_t item_value_len = strlen(user_property_item->value);
char *key = calloc(1, item_key_len + 1);
ESP_MEM_CHECK(TAG, key, goto err);
memcpy(key, user_property_item->key, item_key_len);
key[item_key_len] = '\0';
char *value = calloc(1, item_value_len + 1);
ESP_MEM_CHECK(TAG, value, {
free(key);
goto err;
});
memcpy(value, user_property_item->value, item_value_len);
value[item_value_len] = '\0';
item[i].key = key;
item[i].value = value;
i ++;
} else {
break;
}
}
*item_num = i;
return ESP_OK;
} else {
ESP_LOGE(TAG, "Input value is NULL or item_num is 0");
return ESP_FAIL;
}
err:
for (j = 0; j < i; j ++) {
if (item[j].key) {
free((char *)item[j].key);
}
if (item[j].value) {
free((char *)item[j].value);
}
}
return ESP_ERR_NO_MEM;
}
uint8_t esp_mqtt5_client_get_user_property_count(mqtt5_user_property_handle_t user_property)
{
uint8_t count = 0;
if (user_property) {
mqtt5_user_property_item_t item;
STAILQ_FOREACH(item, user_property, next) {
count ++;
}
}
return count;
}
void esp_mqtt5_client_delete_user_property(mqtt5_user_property_handle_t user_property)
{
if (user_property) {
mqtt5_user_property_item_t item, tmp;
STAILQ_FOREACH_SAFE(item, user_property, next, tmp) {
STAILQ_REMOVE(user_property, item, mqtt5_user_property, next);
free(item->key);
free(item->value);
free(item);
}
}
free(user_property);
}
#endif