mirror of
https://github.com/espressif/esp-mqtt.git
synced 2026-09-26 02:47:35 +00:00
Merge branch 'fix/mqtt5-property-owner' into 'master'
MR: Prevent concurrency issue in MQTT5 properties See merge request espressif/esp-mqtt!345
This commit is contained in:
+18
-6
@@ -220,8 +220,11 @@ esp_err_t esp_mqtt5_client_set_connect_property(esp_mqtt5_client_handle_t client
|
||||
/**
|
||||
* @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
|
||||
* This API stages a borrowed property pointer for one successful publish or enqueue serialization by the
|
||||
* calling task. The property and all referenced data must remain valid until then. A call from the same task
|
||||
* replaces its staged pointer. Publish or enqueue calls from another task proceed without this property and do
|
||||
* not consume it. The owner task must not terminate before consuming the property; otherwise the slot remains
|
||||
* reserved until client destruction. Reuse of a terminated task's handle cannot currently be detected.
|
||||
*
|
||||
* @param client mqtt client handle, must not be NULL
|
||||
* @param property publish property, must not be NULL
|
||||
@@ -229,6 +232,7 @@ esp_err_t esp_mqtt5_client_set_connect_property(esp_mqtt5_client_handle_t client
|
||||
* @return
|
||||
* - ESP_OK on success
|
||||
* - ESP_ERR_INVALID_ARG if client or property is NULL
|
||||
* - ESP_ERR_INVALID_STATE if another task owns the staged publish property
|
||||
* - ESP_FAIL on fail
|
||||
*/
|
||||
esp_err_t esp_mqtt5_client_set_publish_property(esp_mqtt5_client_handle_t client,
|
||||
@@ -237,8 +241,11 @@ esp_err_t esp_mqtt5_client_set_publish_property(esp_mqtt5_client_handle_t client
|
||||
/**
|
||||
* @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
|
||||
* This API stages a borrowed property pointer for one successful subscribe serialization by the calling task.
|
||||
* The property and all referenced data must remain valid until then. A call from the same task replaces its
|
||||
* staged pointer. Subscribe calls from another task proceed without this property and do not consume it. The
|
||||
* owner task must not terminate before consuming the property; otherwise the slot remains reserved until client
|
||||
* destruction. Reuse of a terminated task's handle cannot currently be detected.
|
||||
*
|
||||
* @param client mqtt client handle, must not be NULL
|
||||
* @param property subscribe property, must not be NULL
|
||||
@@ -246,6 +253,7 @@ esp_err_t esp_mqtt5_client_set_publish_property(esp_mqtt5_client_handle_t client
|
||||
* @return
|
||||
* - ESP_OK on success
|
||||
* - ESP_ERR_INVALID_ARG if client or property is NULL
|
||||
* - ESP_ERR_INVALID_STATE if another task owns the staged subscribe property
|
||||
* - ESP_FAIL on fail
|
||||
*/
|
||||
esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt5_client_handle_t client,
|
||||
@@ -254,8 +262,11 @@ esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt5_client_handle_t clie
|
||||
/**
|
||||
* @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
|
||||
* This API stages a borrowed property pointer for one successful unsubscribe serialization by the calling task.
|
||||
* The property and all referenced data must remain valid until then. A call from the same task replaces its
|
||||
* staged pointer. Unsubscribe calls from another task proceed without this property and do not consume it. The
|
||||
* owner task must not terminate before consuming the property; otherwise the slot remains reserved until client
|
||||
* destruction. Reuse of a terminated task's handle cannot currently be detected.
|
||||
*
|
||||
* @param client mqtt client handle, must not be NULL
|
||||
* @param property unsubscribe property, must not be NULL
|
||||
@@ -263,6 +274,7 @@ esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt5_client_handle_t clie
|
||||
* @return
|
||||
* - ESP_OK on success
|
||||
* - ESP_ERR_INVALID_ARG if client or property is NULL
|
||||
* - ESP_ERR_INVALID_STATE if another task owns the staged unsubscribe property
|
||||
* - ESP_FAIL on fail
|
||||
*/
|
||||
esp_err_t esp_mqtt5_client_set_unsubscribe_property(esp_mqtt5_client_handle_t client,
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2022-2025 Espressif Systems (Shanghai) CO LTD
|
||||
* SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD
|
||||
*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
*/
|
||||
@@ -9,6 +9,8 @@
|
||||
|
||||
#include "mqtt5_client.h"
|
||||
#include "mqtt5_msg.h"
|
||||
#include "freertos/FreeRTOS.h"
|
||||
#include "freertos/task.h"
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
@@ -24,14 +26,19 @@ 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 {
|
||||
const void *property;
|
||||
TaskHandle_t owner;
|
||||
} mqtt5_staged_property_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_staged_property_t publish_property;
|
||||
mqtt5_staged_property_t subscribe_property;
|
||||
mqtt5_staged_property_t unsubscribe_property;
|
||||
mqtt5_topic_alias_handle_t peer_topic_alias;
|
||||
} mqtt5_config_storage_t;
|
||||
|
||||
@@ -47,6 +54,9 @@ void esp_mqtt5_client_destory(esp_mqtt5_client_handle_t client);
|
||||
esp_err_t esp_mqtt5_client_check_inflight_maximum(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_staged_property_set(mqtt5_staged_property_t *slot, const void *property);
|
||||
const void *esp_mqtt5_staged_property_get(const mqtt5_staged_property_t *slot);
|
||||
void esp_mqtt5_staged_property_clear(mqtt5_staged_property_t *slot);
|
||||
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);
|
||||
|
||||
+32
-6
@@ -22,6 +22,32 @@ static void esp_mqtt5_client_delete_topic_alias(mqtt5_topic_alias_handle_t topic
|
||||
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);
|
||||
|
||||
esp_err_t esp_mqtt5_staged_property_set(mqtt5_staged_property_t *slot, const void *property)
|
||||
{
|
||||
TaskHandle_t current_task = xTaskGetCurrentTaskHandle();
|
||||
|
||||
if (slot->property && slot->owner != current_task) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
slot->property = property;
|
||||
slot->owner = current_task;
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
const void *esp_mqtt5_staged_property_get(const mqtt5_staged_property_t *slot)
|
||||
{
|
||||
return slot->owner == xTaskGetCurrentTaskHandle() ? slot->property : NULL;
|
||||
}
|
||||
|
||||
void esp_mqtt5_staged_property_clear(mqtt5_staged_property_t *slot)
|
||||
{
|
||||
if (slot->owner == xTaskGetCurrentTaskHandle()) {
|
||||
slot->property = NULL;
|
||||
slot->owner = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
void esp_mqtt5_increment_packet_counter(esp_mqtt5_client_handle_t client)
|
||||
{
|
||||
client->send_publish_packet_count ++;
|
||||
@@ -530,9 +556,9 @@ esp_err_t esp_mqtt5_client_set_publish_property(esp_mqtt5_client_handle_t client
|
||||
return ESP_FAIL;
|
||||
}
|
||||
|
||||
client->mqtt5_config->publish_property_info = property;
|
||||
esp_err_t ret = esp_mqtt5_staged_property_set(&client->mqtt5_config->publish_property, property);
|
||||
MQTT_API_UNLOCK(client);
|
||||
return ESP_OK;
|
||||
return ret;
|
||||
}
|
||||
|
||||
esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt5_client_handle_t client,
|
||||
@@ -583,9 +609,9 @@ esp_err_t esp_mqtt5_client_set_subscribe_property(esp_mqtt5_client_handle_t clie
|
||||
}
|
||||
}
|
||||
|
||||
client->mqtt5_config->subscribe_property_info = property;
|
||||
esp_err_t ret = esp_mqtt5_staged_property_set(&client->mqtt5_config->subscribe_property, property);
|
||||
MQTT_API_UNLOCK(client);
|
||||
return ESP_OK;
|
||||
return ret;
|
||||
}
|
||||
|
||||
esp_err_t esp_mqtt5_client_set_unsubscribe_property(esp_mqtt5_client_handle_t client,
|
||||
@@ -624,9 +650,9 @@ esp_err_t esp_mqtt5_client_set_unsubscribe_property(esp_mqtt5_client_handle_t cl
|
||||
}
|
||||
}
|
||||
|
||||
client->mqtt5_config->unsubscribe_property_info = property;
|
||||
esp_err_t ret = esp_mqtt5_staged_property_set(&client->mqtt5_config->unsubscribe_property, property);
|
||||
MQTT_API_UNLOCK(client);
|
||||
return ESP_OK;
|
||||
return ret;
|
||||
}
|
||||
|
||||
esp_err_t esp_mqtt5_client_set_disconnect_property(esp_mqtt5_client_handle_t client,
|
||||
|
||||
+15
-9
@@ -2417,12 +2417,14 @@ int esp_mqtt_client_subscribe_multiple(esp_mqtt_client_handle_t client,
|
||||
return -1;
|
||||
}
|
||||
|
||||
const esp_mqtt5_subscribe_property_config_t *property =
|
||||
esp_mqtt5_staged_property_get(&client->mqtt5_config->subscribe_property);
|
||||
mqtt5_msg_subscribe(&client->mqtt_state.connection,
|
||||
topic_list, size,
|
||||
&client->mqtt_state.pending_msg_id, client->mqtt5_config->subscribe_property_info);
|
||||
&client->mqtt_state.pending_msg_id, property);
|
||||
|
||||
if (client->mqtt_state.connection.outbound_message.length) {
|
||||
client->mqtt5_config->subscribe_property_info = NULL;
|
||||
if (property && client->mqtt_state.connection.outbound_message.length) {
|
||||
esp_mqtt5_staged_property_clear(&client->mqtt5_config->subscribe_property);
|
||||
}
|
||||
|
||||
#endif
|
||||
@@ -2483,12 +2485,14 @@ int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *top
|
||||
|
||||
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
|
||||
#ifdef MQTT_PROTOCOL_5
|
||||
const esp_mqtt5_unsubscribe_property_config_t *property =
|
||||
esp_mqtt5_staged_property_get(&client->mqtt5_config->unsubscribe_property);
|
||||
mqtt5_msg_unsubscribe(&client->mqtt_state.connection,
|
||||
topic,
|
||||
&client->mqtt_state.pending_msg_id, client->mqtt5_config->unsubscribe_property_info);
|
||||
&client->mqtt_state.pending_msg_id, property);
|
||||
|
||||
if (client->mqtt_state.connection.outbound_message.length) {
|
||||
client->mqtt5_config->unsubscribe_property_info = NULL;
|
||||
if (property && client->mqtt_state.connection.outbound_message.length) {
|
||||
esp_mqtt5_staged_property_clear(&client->mqtt5_config->unsubscribe_property);
|
||||
}
|
||||
|
||||
#endif
|
||||
@@ -2533,14 +2537,16 @@ static int make_publish(esp_mqtt_client_handle_t client, const char *topic, cons
|
||||
|
||||
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
|
||||
#ifdef MQTT_PROTOCOL_5
|
||||
const esp_mqtt5_publish_property_config_t *property =
|
||||
esp_mqtt5_staged_property_get(&client->mqtt5_config->publish_property);
|
||||
mqtt5_msg_publish(&client->mqtt_state.connection,
|
||||
topic, data, len,
|
||||
qos, retain,
|
||||
&pending_msg_id, client->mqtt5_config->publish_property_info,
|
||||
&pending_msg_id, property,
|
||||
client->mqtt5_config->server_resp_property_info.response_info);
|
||||
|
||||
if (client->mqtt_state.connection.outbound_message.length) {
|
||||
client->mqtt5_config->publish_property_info = NULL;
|
||||
if (property && client->mqtt_state.connection.outbound_message.length) {
|
||||
esp_mqtt5_staged_property_clear(&client->mqtt5_config->publish_property);
|
||||
}
|
||||
|
||||
#endif
|
||||
|
||||
@@ -4,9 +4,37 @@
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
*/
|
||||
#include <stdint.h>
|
||||
#include <string.h>
|
||||
|
||||
#include "mqtt_client_priv.h"
|
||||
|
||||
static struct esp_mqtt_client s_client;
|
||||
static mqtt5_config_storage_t s_mqtt5_config;
|
||||
|
||||
esp_mqtt_client_handle_t test_mqtt5_property_client_reset(void)
|
||||
{
|
||||
memset(&s_client, 0, sizeof(s_client));
|
||||
memset(&s_mqtt5_config, 0, sizeof(s_mqtt5_config));
|
||||
s_client.mqtt5_config = &s_mqtt5_config;
|
||||
s_client.mqtt_state.connection.information.protocol_ver = MQTT_PROTOCOL_V_5;
|
||||
return &s_client;
|
||||
}
|
||||
|
||||
mqtt5_staged_property_t *test_mqtt5_publish_property_slot(void)
|
||||
{
|
||||
return &s_mqtt5_config.publish_property;
|
||||
}
|
||||
|
||||
mqtt5_staged_property_t *test_mqtt5_subscribe_property_slot(void)
|
||||
{
|
||||
return &s_mqtt5_config.subscribe_property;
|
||||
}
|
||||
|
||||
mqtt5_staged_property_t *test_mqtt5_unsubscribe_property_slot(void)
|
||||
{
|
||||
return &s_mqtt5_config.unsubscribe_property;
|
||||
}
|
||||
|
||||
esp_err_t test_mqtt5_check_inflight_maximum(uint16_t send_count, uint16_t receive_maximum)
|
||||
{
|
||||
struct esp_mqtt_client client = {0};
|
||||
|
||||
@@ -11,6 +11,10 @@
|
||||
#include "mqtt5_client.h"
|
||||
|
||||
extern "C" {
|
||||
#include "Mockqueue.h"
|
||||
#include "Mocktask.h"
|
||||
#include "mqtt5_client_priv.h"
|
||||
|
||||
esp_err_t test_mqtt5_check_inflight_maximum(uint16_t send_count, uint16_t receive_maximum);
|
||||
int test_mqtt5_increment_packet_counter_with_dup(void);
|
||||
esp_err_t test_mqtt5_set_connect_null_property(void);
|
||||
@@ -18,6 +22,22 @@ extern "C" {
|
||||
esp_err_t test_mqtt5_set_subscribe_null_property(void);
|
||||
esp_err_t test_mqtt5_set_unsubscribe_null_property(void);
|
||||
esp_err_t test_mqtt5_set_disconnect_null_property(void);
|
||||
esp_mqtt_client_handle_t test_mqtt5_property_client_reset(void);
|
||||
mqtt5_staged_property_t *test_mqtt5_publish_property_slot(void);
|
||||
mqtt5_staged_property_t *test_mqtt5_subscribe_property_slot(void);
|
||||
mqtt5_staged_property_t *test_mqtt5_unsubscribe_property_slot(void);
|
||||
}
|
||||
|
||||
static void use_task(TaskHandle_t task)
|
||||
{
|
||||
xTaskGetCurrentTaskHandle_IgnoreAndReturn(task);
|
||||
}
|
||||
|
||||
static esp_mqtt_client_handle_t prepare_property_client()
|
||||
{
|
||||
xQueueTakeMutexRecursive_IgnoreAndReturn(pdTRUE);
|
||||
xQueueGiveMutexRecursive_IgnoreAndReturn(pdTRUE);
|
||||
return test_mqtt5_property_client_reset();
|
||||
}
|
||||
|
||||
TEST_CASE("MQTT5 inflight quota uses an exact upper bound")
|
||||
@@ -53,3 +73,67 @@ TEST_CASE("MQTT5 property setters reject a null client")
|
||||
REQUIRE(esp_mqtt5_client_set_unsubscribe_property(nullptr, &unsubscribe) == ESP_ERR_INVALID_ARG);
|
||||
REQUIRE(esp_mqtt5_client_set_disconnect_property(nullptr, &disconnect) == ESP_ERR_INVALID_ARG);
|
||||
}
|
||||
|
||||
TEST_CASE("MQTT5 staged property enforces task ownership")
|
||||
{
|
||||
int task_a_storage;
|
||||
int task_b_storage;
|
||||
int property_a = 1;
|
||||
int property_b = 2;
|
||||
int replacement = 3;
|
||||
TaskHandle_t task_a = reinterpret_cast<TaskHandle_t>(&task_a_storage);
|
||||
TaskHandle_t task_b = reinterpret_cast<TaskHandle_t>(&task_b_storage);
|
||||
mqtt5_staged_property_t slot = {};
|
||||
use_task(task_a);
|
||||
REQUIRE(esp_mqtt5_staged_property_set(&slot, &property_a) == ESP_OK);
|
||||
REQUIRE(esp_mqtt5_staged_property_get(&slot) == &property_a);
|
||||
use_task(task_b);
|
||||
REQUIRE(esp_mqtt5_staged_property_get(&slot) == nullptr);
|
||||
esp_mqtt5_staged_property_clear(&slot);
|
||||
REQUIRE(slot.property == &property_a);
|
||||
REQUIRE(slot.owner == task_a);
|
||||
REQUIRE(esp_mqtt5_staged_property_set(&slot, &property_b) == ESP_ERR_INVALID_STATE);
|
||||
REQUIRE(slot.property == &property_a);
|
||||
REQUIRE(slot.owner == task_a);
|
||||
use_task(task_a);
|
||||
REQUIRE(esp_mqtt5_staged_property_set(&slot, &replacement) == ESP_OK);
|
||||
REQUIRE(esp_mqtt5_staged_property_get(&slot) == &replacement);
|
||||
esp_mqtt5_staged_property_clear(&slot);
|
||||
REQUIRE(slot.property == nullptr);
|
||||
REQUIRE(slot.owner == nullptr);
|
||||
use_task(task_b);
|
||||
REQUIRE(esp_mqtt5_staged_property_set(&slot, &property_b) == ESP_OK);
|
||||
REQUIRE(esp_mqtt5_staged_property_get(&slot) == &property_b);
|
||||
}
|
||||
|
||||
TEST_CASE("MQTT5 property setters use task-owned slots")
|
||||
{
|
||||
int task_a_storage;
|
||||
int task_b_storage;
|
||||
TaskHandle_t task_a = reinterpret_cast<TaskHandle_t>(&task_a_storage);
|
||||
TaskHandle_t task_b = reinterpret_cast<TaskHandle_t>(&task_b_storage);
|
||||
esp_mqtt5_publish_property_config_t publish_a = {};
|
||||
esp_mqtt5_publish_property_config_t publish_b = {};
|
||||
esp_mqtt5_subscribe_property_config_t subscribe_a = {};
|
||||
esp_mqtt5_subscribe_property_config_t subscribe_b = {};
|
||||
esp_mqtt5_unsubscribe_property_config_t unsubscribe_a = {};
|
||||
esp_mqtt5_unsubscribe_property_config_t unsubscribe_b = {};
|
||||
esp_mqtt_client_handle_t client = prepare_property_client();
|
||||
use_task(task_a);
|
||||
REQUIRE(esp_mqtt5_client_set_publish_property(client, &publish_a) == ESP_OK);
|
||||
REQUIRE(esp_mqtt5_client_set_subscribe_property(client, &subscribe_a) == ESP_OK);
|
||||
REQUIRE(esp_mqtt5_client_set_unsubscribe_property(client, &unsubscribe_a) == ESP_OK);
|
||||
REQUIRE(test_mqtt5_publish_property_slot()->property == &publish_a);
|
||||
REQUIRE(test_mqtt5_subscribe_property_slot()->property == &subscribe_a);
|
||||
REQUIRE(test_mqtt5_unsubscribe_property_slot()->property == &unsubscribe_a);
|
||||
REQUIRE(test_mqtt5_publish_property_slot()->owner == task_a);
|
||||
REQUIRE(test_mqtt5_subscribe_property_slot()->owner == task_a);
|
||||
REQUIRE(test_mqtt5_unsubscribe_property_slot()->owner == task_a);
|
||||
use_task(task_b);
|
||||
REQUIRE(esp_mqtt5_client_set_publish_property(client, &publish_b) == ESP_ERR_INVALID_STATE);
|
||||
REQUIRE(esp_mqtt5_client_set_subscribe_property(client, &subscribe_b) == ESP_ERR_INVALID_STATE);
|
||||
REQUIRE(esp_mqtt5_client_set_unsubscribe_property(client, &unsubscribe_b) == ESP_ERR_INVALID_STATE);
|
||||
REQUIRE(test_mqtt5_publish_property_slot()->property == &publish_a);
|
||||
REQUIRE(test_mqtt5_subscribe_property_slot()->property == &subscribe_a);
|
||||
REQUIRE(test_mqtt5_unsubscribe_property_slot()->property == &unsubscribe_a);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user