espressif/esp-mqtt
1
0
Fork 1
mirror of https://github.com/espressif/esp-mqtt.git synced 2025-07-29 18:28:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'feature/support_mqtt_5' into 'master'

Browse Source 
mqtt: Support MQTT5 protocol

See merge request espressif/esp-mqtt!125
This commit is contained in:
David Čermák
2022-06-14 21:40:57 +08:00
parent 0aa0a342a1 fdf2aeb36f
commit 493822dc9d
9 changed files with 2680 additions and 182 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

11
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" {
@ -114,7 +117,8 @@ typedef enum esp_mqtt_transport_t {
typedef enum esp_mqtt_protocol_ver_t {
MQTT_PROTOCOL_UNDEFINED = 0,
MQTT_PROTOCOL_V_3_1,
MQTT_PROTOCOL_V_3_1_1
MQTT_PROTOCOL_V_3_1_1,
MQTT_PROTOCOL_V_5,
} esp_mqtt_protocol_ver_t;
/**
@ -161,6 +165,10 @@ typedef struct esp_mqtt_event_t {
bool retain; /*!< Retained flag of the message associated with this event */
int qos; /*!< qos of the messages associated with this event */
bool dup; /*!< dup flag of the message associated with this event */
esp_mqtt_protocol_ver_t protocol_ver; /*!< MQTT protocol version used for connection, defaults to value from menuconfig*/
#ifdef CONFIG_MQTT_PROTOCOL_5
esp_mqtt5_event_property_t *property; /*!< MQTT 5 property associated with this event */
#endif
} esp_mqtt_event_t;
typedef esp_mqtt_event_t *esp_mqtt_event_handle_t;
@ -420,7 +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 __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

142
lib/include/mqtt5_msg.h Normal file
View File

@ -0,0 +1,142 @@
#ifndef MQTT5_MSG_H
#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"
#ifdef __cplusplus
extern "C" {
#endif
enum mqtt_properties_type {
MQTT5_PROPERTY_PAYLOAD_FORMAT_INDICATOR = 0x01,
MQTT5_PROPERTY_MESSAGE_EXPIRY_INTERVAL = 0x02,
MQTT5_PROPERTY_CONTENT_TYPE = 0x03,
MQTT5_PROPERTY_RESPONSE_TOPIC = 0x08,
MQTT5_PROPERTY_CORRELATION_DATA = 0x09,
MQTT5_PROPERTY_SUBSCRIBE_IDENTIFIER = 0x0B,
MQTT5_PROPERTY_SESSION_EXPIRY_INTERVAL = 0x11,
MQTT5_PROPERTY_ASSIGNED_CLIENT_IDENTIFIER = 0x12,
MQTT5_PROPERTY_SERVER_KEEP_ALIVE = 0x13,
MQTT5_PROPERTY_AUTHENTICATION_METHOD = 0x15,
MQTT5_PROPERTY_AUTHENTICATION_DATA = 0x16,
MQTT5_PROPERTY_REQUEST_PROBLEM_INFO = 0x17,
MQTT5_PROPERTY_WILL_DELAY_INTERVAL = 0x18,
MQTT5_PROPERTY_REQUEST_RESP_INFO = 0x19,
MQTT5_PROPERTY_RESP_INFO = 0x1A,
MQTT5_PROPERTY_SERVER_REFERENCE = 0x1C,
MQTT5_PROPERTY_REASON_STRING = 0x1F,
MQTT5_PROPERTY_RECEIVE_MAXIMUM = 0x21,
MQTT5_PROPERTY_TOPIC_ALIAS_MAXIMIM = 0x22,
MQTT5_PROPERTY_TOPIC_ALIAS = 0x23,
MQTT5_PROPERTY_MAXIMUM_QOS = 0x24,
MQTT5_PROPERTY_RETAIN_AVAILABLE = 0x25,
MQTT5_PROPERTY_USER_PROPERTY = 0x26,
MQTT5_PROPERTY_MAXIMUM_PACKET_SIZE = 0x27,
MQTT5_PROPERTY_WILDCARD_SUBSCR_AVAILABLE = 0x28,
MQTT5_PROPERTY_SUBSCR_IDENTIFIER_AVAILABLE = 0x29,
MQTT5_PROPERTY_SHARED_SUBSCR_AVAILABLE = 0x2A,
};
typedef struct mqtt5_user_property {
char *key;
char *value;
STAILQ_ENTRY(mqtt5_user_property) next;
} mqtt5_user_property_t;
STAILQ_HEAD(mqtt5_user_property_list_t, mqtt5_user_property);
typedef struct mqtt5_user_property *mqtt5_user_property_item_t;
typedef struct {
uint32_t maximum_packet_size;
uint16_t receive_maximum;
uint16_t topic_alias_maximum;
uint8_t max_qos;
bool retain_available;
bool wildcard_subscribe_available;
bool subscribe_identifiers_available;
bool shared_subscribe_available;
char *response_info;
} esp_mqtt5_connection_server_resp_property_t;
typedef struct {
bool payload_format_indicator;
uint32_t message_expiry_interval;
uint16_t topic_alias;
char *response_topic;
int response_topic_len;
char *correlation_data;
uint16_t correlation_data_len;
char *content_type;
int content_type_len;
uint16_t subscribe_id;
} esp_mqtt5_publish_resp_property_t;
typedef struct {
uint32_t session_expiry_interval;
uint32_t maximum_packet_size;
uint16_t receive_maximum;
uint16_t topic_alias_maximum;
bool request_resp_info;
bool request_problem_info;
mqtt5_user_property_handle_t user_property;
} esp_mqtt5_connection_property_storage_t;
typedef struct {
uint32_t will_delay_interval;
uint32_t message_expiry_interval;
bool payload_format_indicator;
char *content_type;
char *response_topic;
char *correlation_data;
uint16_t correlation_data_len;
mqtt5_user_property_handle_t user_property;
} esp_mqtt5_connection_will_property_storage_t;
#define mqtt5_get_type mqtt_get_type
#define mqtt5_get_dup mqtt_get_dup
#define mqtt5_set_dup mqtt_set_dup
#define mqtt5_get_qos mqtt_get_qos
#define mqtt5_get_retain mqtt_get_retain
#define mqtt5_msg_init mqtt_msg_init
#define mqtt5_get_total_length mqtt_get_total_length
#define mqtt5_has_valid_msg_hdr mqtt_has_valid_msg_hdr
#define mqtt5_msg_pingreq mqtt_msg_pingreq
#define mqtt5_msg_pingresp mqtt_msg_pingresp
#define mqtt5_get_unsuback_data mqtt5_get_suback_data
#define mqtt5_get_pubcomp_data mqtt5_get_puback_data
uint16_t mqtt5_get_id(uint8_t *buffer, size_t length);
char *mqtt5_get_publish_property_payload(uint8_t *buffer, size_t buffer_length, char **msg_topic, size_t *msg_topic_len, esp_mqtt5_publish_resp_property_t *resp_property, uint16_t *property_len, size_t *payload_len, mqtt5_user_property_handle_t *user_property);
char *mqtt5_get_suback_data(uint8_t *buffer, size_t *length, mqtt5_user_property_handle_t *user_property);
char *mqtt5_get_puback_data(uint8_t *buffer, size_t *length, mqtt5_user_property_handle_t *user_property);
mqtt_message_t *mqtt5_msg_connect(mqtt_connection_t *connection, mqtt_connect_info_t *info, esp_mqtt5_connection_property_storage_t *property, esp_mqtt5_connection_will_property_storage_t *will_property);
mqtt_message_t *mqtt5_msg_publish(mqtt_connection_t *connection, const char *topic, const char *data, int data_length, int qos, int retain, uint16_t *message_id, const esp_mqtt5_publish_property_config_t *property, const char *resp_info);
esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, mqtt_connect_info_t *connection_info, esp_mqtt5_connection_property_storage_t *connection_property, esp_mqtt5_connection_server_resp_property_t *resp_property, int *reason_code, uint8_t *ack_flag, mqtt5_user_property_handle_t *user_property);
int mqtt5_msg_get_reason_code(uint8_t *buffer, size_t length);
mqtt_message_t *mqtt5_msg_subscribe(mqtt_connection_t *connection, const char *topic, int qos, uint16_t *message_id, const esp_mqtt5_subscribe_property_config_t *property);
mqtt_message_t *mqtt5_msg_unsubscribe(mqtt_connection_t *connection, const char *topic, uint16_t *message_id, const esp_mqtt5_unsubscribe_property_config_t *property);
mqtt_message_t *mqtt5_msg_disconnect(mqtt_connection_t *connection, esp_mqtt5_disconnect_property_config_t *disconnect_property_info);
mqtt_message_t *mqtt5_msg_pubcomp(mqtt_connection_t *connection, uint16_t message_id);
mqtt_message_t *mqtt5_msg_pubrel(mqtt_connection_t *connection, uint16_t message_id);
mqtt_message_t *mqtt5_msg_pubrec(mqtt_connection_t *connection, uint16_t message_id);
mqtt_message_t *mqtt5_msg_puback(mqtt_connection_t *connection, uint16_t message_id);
#ifdef __cplusplus
}
#endif
#endif /* MQTT5_MSG_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

4
lib/include/mqtt_config.h
View File

@ -12,6 +12,10 @@
#define MQTT_PROTOCOL_311
#endif
#ifdef CONFIG_MQTT_PROTOCOL_5
#define MQTT_PROTOCOL_5
#endif
#define MQTT_RECON_DEFAULT_MS (10*1000)
#define MQTT_POLL_READ_TIMEOUT_MS (1000)

1040
lib/mqtt5_msg.c Normal file
View File

File diff suppressed because it is too large Load Diff

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);
}

438
mqtt_client.c
View File

@ -1,33 +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"
#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
@ -49,90 +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;
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;
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);
@ -362,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);
@ -482,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();
}
@ -497,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);
@ -546,6 +439,11 @@ esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_cl
client->connect_info.protocol_ver = MQTT_PROTOCOL_V_3_1_1;
#else
client->connect_info.protocol_ver = MQTT_PROTOCOL_V_3_1;
#endif
} else if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifndef MQTT_PROTOCOL_5
ESP_LOGE(TAG, "Please first enable MQTT_PROTOCOL_5 feature in menuconfig");
goto _mqtt_set_config_failed;
#endif
}
@ -664,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;
@ -683,6 +581,9 @@ static void esp_mqtt_destroy_config(esp_mqtt_client_handle_t client)
free(client->connect_info.client_id);
free(client->connect_info.username);
free(client->connect_info.password);
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_client_destory(client);
#endif
memset(&client->connect_info, 0, sizeof(mqtt_connect_info_t));
#ifdef MQTT_SUPPORTED_FEATURE_EVENT_LOOP
if (client->config->event_loop_handle) {
@ -746,18 +647,32 @@ static inline esp_err_t esp_mqtt_write(esp_mqtt_client_handle_t client)
static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_ms)
{
int read_len, connect_rsp_code;
int read_len, connect_rsp_code = 0;
client->wait_for_ping_resp = false;
client->mqtt_state.outbound_message = mqtt_msg_connect(&client->mqtt_state.mqtt_connection,
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_connect(&client->mqtt_state.mqtt_connection,
&client->connect_info, &client->mqtt5_config->connect_property_info, &client->mqtt5_config->will_property_info);
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_connect(&client->mqtt_state.mqtt_connection,
&client->connect_info);
}
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Connect message cannot be created");
return ESP_FAIL;
}
client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data,
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.pending_msg_id = mqtt5_get_id(client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
#endif
} else {
client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
}
ESP_LOGD(TAG, "Sending MQTT CONNECT message, type: %d, id: %04X",
client->mqtt_state.pending_msg_type,
client->mqtt_state.pending_msg_id);
@ -784,28 +699,36 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
ESP_LOGE(TAG, "Invalid MSG_TYPE response: %d, read_len: %d", mqtt_get_type(client->mqtt_state.in_buffer), read_len);
return ESP_FAIL;
}
client->mqtt_state.in_buffer_read_len = 0;
connect_rsp_code = mqtt_get_connect_return_code(client->mqtt_state.in_buffer);
if (connect_rsp_code == MQTT_CONNECTION_ACCEPTED) {
ESP_LOGD(TAG, "Connected");
return ESP_OK;
}
switch (connect_rsp_code) {
case MQTT_CONNECTION_REFUSE_PROTOCOL:
ESP_LOGW(TAG, "Connection refused, bad protocol");
break;
case MQTT_CONNECTION_REFUSE_SERVER_UNAVAILABLE:
ESP_LOGW(TAG, "Connection refused, server unavailable");
break;
case MQTT_CONNECTION_REFUSE_BAD_USERNAME:
ESP_LOGW(TAG, "Connection refused, bad username or password");
break;
case MQTT_CONNECTION_REFUSE_NOT_AUTHORIZED:
ESP_LOGW(TAG, "Connection refused, not authorized");
break;
default:
ESP_LOGW(TAG, "Connection refused, Unknow reason");
break;
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
if (esp_mqtt5_parse_connack(client, &connect_rsp_code) == ESP_OK) {
return ESP_OK;
}
#endif
} else {
client->mqtt_state.in_buffer_read_len = 0;
connect_rsp_code = mqtt_get_connect_return_code(client->mqtt_state.in_buffer);
if (connect_rsp_code == MQTT_CONNECTION_ACCEPTED) {
ESP_LOGD(TAG, "Connected");
return ESP_OK;
}
switch (connect_rsp_code) {
case MQTT_CONNECTION_REFUSE_PROTOCOL:
ESP_LOGW(TAG, "Connection refused, bad protocol");
break;
case MQTT_CONNECTION_REFUSE_SERVER_UNAVAILABLE:
ESP_LOGW(TAG, "Connection refused, server unavailable");
break;
case MQTT_CONNECTION_REFUSE_BAD_USERNAME:
ESP_LOGW(TAG, "Connection refused, bad username or password");
break;
case MQTT_CONNECTION_REFUSE_NOT_AUTHORIZED:
ESP_LOGW(TAG, "Connection refused, not authorized");
break;
default:
ESP_LOGW(TAG, "Connection refused, Unknow reason");
break;
}
}
/* propagate event with connection refused error */
client->event.event_id = MQTT_EVENT_ERROR;
@ -888,7 +811,11 @@ esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *co
mqtt_msg_init(&client->mqtt_state.mqtt_connection, client->mqtt_state.out_buffer,
client->mqtt_state.out_buffer_length);
#ifdef MQTT_PROTOCOL_5
if (esp_mqtt5_create_default_config(client) != ESP_OK) {
goto _mqtt_init_failed;
}
#endif
return client;
_mqtt_init_failed:
esp_mqtt_client_destroy(client);
@ -998,12 +925,21 @@ static esp_err_t mqtt_write_data(esp_mqtt_client_handle_t client)
esp_mqtt_client_dispatch_transport_error(client);
return ESP_FAIL;
}
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_flow_control(client);
#endif
return ESP_OK;
}
static esp_err_t esp_mqtt_dispatch_event_with_msgid(esp_mqtt_client_handle_t client)
{
client->event.msg_id = mqtt_get_id(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->event.msg_id = mqtt5_get_id(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length);
#endif
} else {
client->event.msg_id = mqtt_get_id(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length);
}
return esp_mqtt_dispatch_event(client);
}
@ -1011,17 +947,26 @@ static esp_err_t esp_mqtt_dispatch_event(esp_mqtt_client_handle_t client)
{
client->event.user_context = client->config->user_context;
client->event.client = client;
client->event.protocol_ver = client->connect_info.protocol_ver;
esp_err_t ret = ESP_FAIL;
if (client->config->event_handle) {
return client->config->event_handle(&client->event);
ret = client->config->event_handle(&client->event);
} else {
#ifdef MQTT_SUPPORTED_FEATURE_EVENT_LOOP
esp_event_post_to(client->config->event_loop_handle, MQTT_EVENTS, client->event.event_id, &client->event, sizeof(client->event), portMAX_DELAY);
return esp_event_loop_run(client->config->event_loop_handle, 0);
ret = esp_event_loop_run(client->config->event_loop_handle, 0);
#else
return ESP_FAIL;
#endif
}
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_client_delete_user_property(client->event.property->user_property);
client->event.property->user_property = NULL;
#endif
}
return ret;
}
static esp_err_t deliver_publish(esp_mqtt_client_handle_t client)
@ -1033,28 +978,41 @@ static esp_err_t deliver_publish(esp_mqtt_client_handle_t client)
size_t msg_data_offset = 0;
char *msg_topic = NULL, *msg_data = NULL;
// get topic
msg_topic = mqtt_get_publish_topic(msg_buf, &msg_topic_len);
if (msg_topic == NULL) {
ESP_LOGE(TAG, "%s: mqtt_get_publish_topic() failed", __func__);
return ESP_FAIL;
}
ESP_LOGD(TAG, "%s: msg_topic_len=%zu", __func__, msg_topic_len);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
if (esp_mqtt5_get_publish_data(client, msg_buf, msg_read_len, &msg_topic, &msg_topic_len, &msg_data, &msg_data_len) != ESP_OK) {
ESP_LOGE(TAG, "%s: esp_mqtt5_get_publish_data() failed", __func__);
return ESP_FAIL;
}
#endif
} else {
// get topic
msg_topic = mqtt_get_publish_topic(msg_buf, &msg_topic_len);
if (msg_topic == NULL) {
ESP_LOGE(TAG, "%s: mqtt_get_publish_topic() failed", __func__);
return ESP_FAIL;
}
ESP_LOGD(TAG, "%s: msg_topic_len=%zu", __func__, msg_topic_len);
// get payload
msg_data = mqtt_get_publish_data(msg_buf, &msg_data_len);
if (msg_data_len > 0 && msg_data == NULL) {
ESP_LOGE(TAG, "%s: mqtt_get_publish_data() failed", __func__);
return ESP_FAIL;
// get payload
msg_data = mqtt_get_publish_data(msg_buf, &msg_data_len);
if (msg_data_len > 0 && msg_data == NULL) {
ESP_LOGE(TAG, "%s: mqtt_get_publish_data() failed", __func__);
return ESP_FAIL;
}
}
// post data event
client->event.retain = mqtt_get_retain(msg_buf);
client->event.msg_id = mqtt_get_id(msg_buf, msg_data_len);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->event.msg_id = mqtt5_get_id(msg_buf, msg_read_len);
#endif
} else {
client->event.msg_id = mqtt_get_id(msg_buf, msg_read_len);
}
client->event.qos = mqtt_get_qos(msg_buf);
client->event.dup = mqtt_get_dup(msg_buf);
client->event.total_data_len = msg_data_len + msg_total_len - msg_read_len;
post_data_event:
ESP_LOGD(TAG, "Get data len= %zu, topic len=%zu, total_data: %d offset: %zu", msg_data_len, msg_topic_len,
client->event.total_data_len, msg_data_offset);
@ -1093,7 +1051,13 @@ static esp_err_t deliver_suback(esp_mqtt_client_handle_t client)
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
char *msg_data = NULL;
msg_data = mqtt_get_suback_data(msg_buf, &msg_data_len);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
msg_data = mqtt5_get_suback_data(msg_buf, &msg_data_len, &client->event.property->user_property);
#endif
} else {
msg_data = mqtt_get_suback_data(msg_buf, &msg_data_len);
}
if (msg_data_len <= 0) {
ESP_LOGE(TAG, "Failed to acquire suback data");
return ESP_FAIL;
@ -1277,9 +1241,8 @@ err:
static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
{
uint8_t msg_type;
uint8_t msg_qos;
uint16_t msg_id;
uint8_t msg_type = 0, msg_qos = 0;
uint16_t msg_id = 0;
/* non-blocking receive in order not to block other tasks */
int recv = mqtt_message_receive(client, 0);
@ -1295,13 +1258,22 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
// If the message was valid, get the type, quality of service and id of the message
msg_type = mqtt_get_type(client->mqtt_state.in_buffer);
msg_qos = mqtt_get_qos(client->mqtt_state.in_buffer);
msg_id = mqtt_get_id(client->mqtt_state.in_buffer, read_len);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
msg_id = mqtt5_get_id(client->mqtt_state.in_buffer, read_len);
#endif
} else {
msg_id = mqtt_get_id(client->mqtt_state.in_buffer, read_len);
}
ESP_LOGD(TAG, "msg_type=%d, msg_id=%d", msg_type, msg_id);
switch (msg_type) {
case MQTT_MSG_TYPE_SUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_SUBSCRIBE, msg_id)) {
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_parse_suback(client);
#endif
ESP_LOGD(TAG, "deliver_suback, message_length_read=%zu, message_length=%zu", client->mqtt_state.in_buffer_read_len, client->mqtt_state.message_length);
if (deliver_suback(client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to deliver suback message id=%d", msg_id);
@ -1311,6 +1283,9 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
break;
case MQTT_MSG_TYPE_UNSUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_UNSUBSCRIBE, msg_id)) {
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_parse_unsuback(client);
#endif
ESP_LOGD(TAG, "UnSubscribe successful");
client->event.event_id = MQTT_EVENT_UNSUBSCRIBED;
esp_mqtt_dispatch_event_with_msgid(client);
@ -1323,9 +1298,21 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
return ESP_FAIL;
}
if (msg_qos == 1) {
client->mqtt_state.outbound_message = mqtt_msg_puback(&client->mqtt_state.mqtt_connection, msg_id);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_puback(&client->mqtt_state.mqtt_connection, msg_id);
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_puback(&client->mqtt_state.mqtt_connection, msg_id);
}
} else if (msg_qos == 2) {
client->mqtt_state.outbound_message = mqtt_msg_pubrec(&client->mqtt_state.mqtt_connection, msg_id);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_pubrec(&client->mqtt_state.mqtt_connection, msg_id);
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_pubrec(&client->mqtt_state.mqtt_connection, msg_id);
}
}
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Publish response message PUBACK or PUBREC cannot be created");
@ -1344,13 +1331,23 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
case MQTT_MSG_TYPE_PUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish");
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_parse_puback(client);
#endif
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_PUBREC:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBREC");
client->mqtt_state.outbound_message = mqtt_msg_pubrel(&client->mqtt_state.mqtt_connection, msg_id);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBREC return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
client->mqtt_state.outbound_message = mqtt5_msg_pubrel(&client->mqtt_state.mqtt_connection, msg_id);
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_pubrel(&client->mqtt_state.mqtt_connection, msg_id);
}
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Publish response message PUBREL cannot be created");
return ESP_FAIL;
@ -1361,7 +1358,14 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
break;
case MQTT_MSG_TYPE_PUBREL:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBREL");
client->mqtt_state.outbound_message = mqtt_msg_pubcomp(&client->mqtt_state.mqtt_connection, msg_id);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBREL return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len));
client->mqtt_state.outbound_message = mqtt5_msg_pubcomp(&client->mqtt_state.mqtt_connection, msg_id);
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_pubcomp(&client->mqtt_state.mqtt_connection, msg_id);
}
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Publish response message PUBCOMP cannot be created");
return ESP_FAIL;
@ -1373,6 +1377,9 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBCOMP");
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "Receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish");
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_parse_pubcomp(client);
#endif
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event_with_msgid(client);
}
@ -1500,7 +1507,9 @@ static void esp_mqtt_task(void *pv)
break;
}
client->event.event_id = MQTT_EVENT_CONNECTED;
client->event.session_present = mqtt_get_connect_session_present(client->mqtt_state.in_buffer);
if (client->connect_info.protocol_ver != MQTT_PROTOCOL_V_5) {
client->event.session_present = mqtt_get_connect_session_present(client->mqtt_state.in_buffer);
}
client->state = MQTT_STATE_CONNECTED;
esp_mqtt_dispatch_event_with_msgid(client);
client->refresh_connection_tick = platform_tick_get_ms();
@ -1654,7 +1663,18 @@ esp_err_t esp_mqtt_client_reconnect(esp_mqtt_client_handle_t client)
static esp_err_t send_disconnect_msg(esp_mqtt_client_handle_t client)
{
// Notify the broker we are disconnecting
client->mqtt_state.outbound_message = mqtt_msg_disconnect(&client->mqtt_state.mqtt_connection);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_disconnect(&client->mqtt_state.mqtt_connection, &client->mqtt5_config->disconnect_property_info);
if (client->mqtt_state.outbound_message->length) {
esp_mqtt5_client_delete_user_property(client->mqtt5_config->disconnect_property_info.user_property);
client->mqtt5_config->disconnect_property_info.user_property = NULL;
memset(&client->mqtt5_config->disconnect_property_info, 0, sizeof(esp_mqtt5_disconnect_property_config_t));
}
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_disconnect(&client->mqtt_state.mqtt_connection);
}
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Disconnect message cannot be created");
return ESP_FAIL;
@ -1729,9 +1749,25 @@ int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic
MQTT_API_UNLOCK(client);
return -1;
}
client->mqtt_state.outbound_message = mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection,
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
if (esp_mqtt5_client_subscribe_check(client, qos) != ESP_OK) {
ESP_LOGI(TAG, "MQTT5 subscribe check fail");
MQTT_API_UNLOCK(client);
return -1;
}
client->mqtt_state.outbound_message = mqtt5_msg_subscribe(&client->mqtt_state.mqtt_connection,
topic, qos,
&client->mqtt_state.pending_msg_id, client->mqtt5_config->subscribe_property_info);
if (client->mqtt_state.outbound_message->length) {
client->mqtt5_config->subscribe_property_info = NULL;
}
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection,
topic, qos,
&client->mqtt_state.pending_msg_id);
}
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Subscribe message cannot be created");
MQTT_API_UNLOCK(client);
@ -1770,9 +1806,20 @@ int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *top
ESP_LOGE(TAG, "Client has not connected");
return -1;
}
client->mqtt_state.outbound_message = mqtt_msg_unsubscribe(&client->mqtt_state.mqtt_connection,
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_unsubscribe(&client->mqtt_state.mqtt_connection,
topic,
&client->mqtt_state.pending_msg_id, client->mqtt5_config->unsubscribe_property_info);
if (client->mqtt_state.outbound_message->length) {
client->mqtt5_config->unsubscribe_property_info = NULL;
}
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_unsubscribe(&client->mqtt_state.mqtt_connection,
topic,
&client->mqtt_state.pending_msg_id);
}
if (client->mqtt_state.outbound_message->length == 0) {
MQTT_API_UNLOCK(client);
ESP_LOGE(TAG, "Unubscribe message cannot be created");
@ -1803,17 +1850,28 @@ static inline int mqtt_client_enqueue_priv(esp_mqtt_client_handle_t client, cons
int len, int qos, int retain, bool store)
{
uint16_t pending_msg_id = 0;
mqtt_message_t *publish_msg = mqtt_msg_publish(&client->mqtt_state.mqtt_connection,
topic, data, len,
qos, retain,
&pending_msg_id);
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_publish(&client->mqtt_state.mqtt_connection,
topic, data, len,
qos, retain,
&pending_msg_id, client->mqtt5_config->publish_property_info, client->mqtt5_config->server_resp_property_info.response_info);
if (client->mqtt_state.outbound_message->length) {
client->mqtt5_config->publish_property_info = NULL;
}
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_publish(&client->mqtt_state.mqtt_connection,
topic, data, len,
qos, retain,
&pending_msg_id);
}
if (publish_msg->length == 0) {
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Publish message cannot be created");
return -1;
}
/* We have to set as pending all the qos>0 messages */
client->mqtt_state.outbound_message = publish_msg;
if (qos > 0 || store) {
client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
client->mqtt_state.pending_msg_id = pending_msg_id;
@ -1850,6 +1908,16 @@ int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic,
}
#endif
#ifdef MQTT_PROTOCOL_5
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
if (esp_mqtt5_client_publish_check(client, qos, retain) != ESP_OK) {
ESP_LOGI(TAG, "MQTT5 publish check fail");
MQTT_API_UNLOCK(client);
return -1;
}
}
#endif
/* Acceptable publish messages:
data == NULL, len == 0: publish null message
data valid, len == 0: publish all data, payload len is determined from string length
@ -1955,6 +2023,15 @@ int esp_mqtt_client_enqueue(esp_mqtt_client_handle_t client, const char *topic,
}
MQTT_API_LOCK(client);
#ifdef MQTT_PROTOCOL_5
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
if (esp_mqtt5_client_publish_check(client, qos, retain) != ESP_OK) {
ESP_LOGI(TAG, "esp_mqtt_client_enqueue check fail");
MQTT_API_UNLOCK(client);
return -1;
}
}
#endif
int ret = mqtt_client_enqueue_priv(client, topic, data, len, qos, retain, store);
MQTT_API_UNLOCK(client);
if (ret == 0 && store == false) {
@ -2017,3 +2094,4 @@ int esp_mqtt_client_get_outbox_size(esp_mqtt_client_handle_t client)
return outbox_size;
}