diff --git a/include/mqtt5_client.h b/include/mqtt5_client.h index f46a379..25b27c8 100644 --- a/include/mqtt5_client.h +++ b/include/mqtt5_client.h @@ -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, diff --git a/lib/include/mqtt5_client_priv.h b/lib/include/mqtt5_client_priv.h index 8f54478..c8bcc2a 100644 --- a/lib/include/mqtt5_client_priv.h +++ b/lib/include/mqtt5_client_priv.h @@ -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); diff --git a/mqtt5_client.c b/mqtt5_client.c index db777b0..59a0d39 100644 --- a/mqtt5_client.c +++ b/mqtt5_client.c @@ -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, diff --git a/mqtt_client.c b/mqtt_client.c index b9a4496..505400a 100644 --- a/mqtt_client.c +++ b/mqtt_client.c @@ -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 diff --git a/test/host/main/mqtt5_client_test_adapter.c b/test/host/main/mqtt5_client_test_adapter.c index 3bf1452..1d46349 100644 --- a/test/host/main/mqtt5_client_test_adapter.c +++ b/test/host/main/mqtt5_client_test_adapter.c @@ -4,9 +4,37 @@ * SPDX-License-Identifier: Apache-2.0 */ #include +#include #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}; diff --git a/test/host/main/test_mqtt5_client.cpp b/test/host/main/test_mqtt5_client.cpp index 99d0a60..52ed1fc 100644 --- a/test/host/main/test_mqtt5_client.cpp +++ b/test/host/main/test_mqtt5_client.cpp @@ -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(&task_a_storage); + TaskHandle_t task_b = reinterpret_cast(&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(&task_a_storage); + TaskHandle_t task_b = reinterpret_cast(&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); +}