diff --git a/examples/custom_outbox/main/custom_outbox.cpp b/examples/custom_outbox/main/custom_outbox.cpp index 07688e2..0f943cd 100644 --- a/examples/custom_outbox/main/custom_outbox.cpp +++ b/examples/custom_outbox/main/custom_outbox.cpp @@ -222,20 +222,25 @@ struct outbox_t { queue.clear(); } - outbox_item_handle_t enqueue(outbox_message_handle_t message, outbox_tick_t tick) noexcept + outbox_item_handle_t enqueue(mqtt_message_t *message, outbox_tick_t tick) noexcept { + if (message == nullptr || message->data == nullptr) { + return nullptr; + } + try { auto &item = - queue.emplace_back(std::pmr::vector {message->data, message->data + message->len}, - outbox_item::id_t{message->msg_id}, - outbox_item::type_t{message->msg_type}, - outbox_item::qos_t{message->msg_qos}, + queue.emplace_back(std::pmr::vector {message->data, message->data + message->length}, + outbox_item::id_t{message->id}, + outbox_item::type_t{message->type}, + outbox_item::qos_t{message->qos}, tick, QUEUED ); total_size += item.get_size(); - ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%d, size=%" PRIu64, message->msg_id, message->msg_type, - message->len + message->remaining_len, outbox_get_size(this)); + ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%zu, size=%" PRIu64, message->id, message->type, + message->length, outbox_get_size(this)); + mqtt_msg_destroy(message); return &item; } catch (const std::exception &e) { return nullptr; @@ -305,7 +310,7 @@ extern "C" { } } - outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, outbox_message_handle_t message, outbox_tick_t tick) + outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, mqtt_message_t *message, outbox_tick_t tick) { return outbox->enqueue(message, tick); } diff --git a/lib/include/mqtt_msg.h b/lib/include/mqtt_msg.h index ae0c03a..7446168 100644 --- a/lib/include/mqtt_msg.h +++ b/lib/include/mqtt_msg.h @@ -102,8 +102,13 @@ char *mqtt_get_suback_data(uint8_t *buffer, size_t *length); uint16_t mqtt_get_id(uint8_t *buffer, size_t length); int mqtt_has_valid_msg_hdr(uint8_t *buffer, size_t length); -esp_err_t mqtt_msg_buffer_init(mqtt_message_t *message, int buffer_size); +mqtt_message_t *mqtt_msg_create(size_t buffer_length); +void mqtt_msg_destroy(mqtt_message_t *message); +esp_err_t mqtt_msg_buffer_init(mqtt_message_t *message, size_t buffer_size); void mqtt_msg_buffer_destroy(mqtt_message_t *message); +mqtt_message_t *mqtt_msg_dup(const mqtt_message_t *message); +esp_err_t mqtt_msg_copy(mqtt_message_t *message, const mqtt_message_t *src); +esp_err_t mqtt_msg_append(mqtt_message_t *message, const uint8_t *data, size_t length); mqtt_message_t *mqtt_msg_connect(mqtt_message_t *message, mqtt_connect_info_t *info); mqtt_message_t *mqtt_msg_publish(mqtt_message_t *message, const char *topic, const char *data, int data_length, int qos, int retain, uint16_t *message_id); diff --git a/lib/include/mqtt_outbox.h b/lib/include/mqtt_outbox.h index 5ae78c6..8108eff 100644 --- a/lib/include/mqtt_outbox.h +++ b/lib/include/mqtt_outbox.h @@ -10,6 +10,7 @@ */ #ifndef _MQTT_OUTOBX_H_ #define _MQTT_OUTOBX_H_ +#include "mqtt_msg.h" #include "platform.h" #include "esp_err.h" @@ -21,18 +22,7 @@ struct outbox_item; typedef struct outbox_t *outbox_handle_t; typedef struct outbox_item *outbox_item_handle_t; -typedef struct outbox_message *outbox_message_handle_t; -typedef long long outbox_tick_t; - -typedef struct outbox_message { - const uint8_t *data; - int len; - int msg_id; - int msg_qos; - int msg_type; - const uint8_t *remaining_data; - int remaining_len; -} outbox_message_t; +typedef uint64_t outbox_tick_t; typedef enum pending_state { QUEUED, @@ -42,7 +32,15 @@ typedef enum pending_state { } pending_state_t; outbox_handle_t outbox_init(void); -outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, outbox_message_handle_t message, outbox_tick_t tick); +/** + * @brief Stores a message in the outbox + * + * On success the outbox takes ownership of @c message and releases it when the item is deleted. + * On failure the ownership stays with the caller. + * + * @return handle of the enqueued item, NULL on failure + */ +outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, mqtt_message_t *message, outbox_tick_t tick); outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox, pending_state_t pending, outbox_tick_t *tick); outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id); uint8_t *outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos); diff --git a/lib/mqtt_msg.c b/lib/mqtt_msg.c index 0411f1c..3943700 100644 --- a/lib/mqtt_msg.c +++ b/lib/mqtt_msg.c @@ -3,9 +3,11 @@ * * SPDX-License-Identifier: BSD-3-Clause */ +#include #include #include "esp_heap_caps.h" #include "mqtt_msg.h" +#include "esp_log.h" #include "mqtt_config.h" #include "platform.h" @@ -465,7 +467,7 @@ mqtt_message_t *mqtt_msg_publish(mqtt_message_t *message, const char *topic, con message->buffer_length - message->length); message->length = message->buffer_length; message->fragmented_msg_total_length = data_length + - message->fragmented_msg_data_offset; + message->fragmented_msg_data_offset; } else { memcpy(message->buffer + message->length, data, data_length); message->length += data_length; @@ -630,10 +632,10 @@ int mqtt_has_valid_msg_hdr(uint8_t *buffer, size_t length) } } -esp_err_t mqtt_msg_buffer_init(mqtt_message_t *message, int buffer_size) +static esp_err_t msg_buffer_init(mqtt_message_t *message, size_t buffer_size, uint32_t caps) { memset(message, 0, sizeof(mqtt_message_t)); - message->buffer = (uint8_t *)heap_caps_calloc(buffer_size, sizeof(uint8_t), MQTT_BUFFER_MEMORY); + message->buffer = (uint8_t *)heap_caps_calloc(buffer_size, sizeof(uint8_t), caps); if (!message->buffer) { return ESP_ERR_NO_MEM; @@ -644,9 +646,107 @@ esp_err_t mqtt_msg_buffer_init(mqtt_message_t *message, int buffer_size) return ESP_OK; } +esp_err_t mqtt_msg_buffer_init(mqtt_message_t *message, size_t buffer_size) +{ + return msg_buffer_init(message, buffer_size, MQTT_BUFFER_MEMORY); +} + void mqtt_msg_buffer_destroy(mqtt_message_t *message) { if (message) { free(message->buffer); } } + +/* Messages created here are owned by the outbox, hence the outbox memory capabilities. */ +mqtt_message_t *mqtt_msg_create(size_t buffer_size) +{ + mqtt_message_t *message = heap_caps_calloc(1, sizeof(mqtt_message_t), MQTT_OUTBOX_MEMORY); + + if (!message) { + return NULL; + } + + if (msg_buffer_init(message, buffer_size, MQTT_OUTBOX_MEMORY) != ESP_OK) { + free(message); + return NULL; + } + + return message; +} + +void mqtt_msg_destroy(mqtt_message_t *message) +{ + if (message) { + mqtt_msg_buffer_destroy(message); + free(message); + } +} + +/* `data` may be offset from `buffer` start to leave room for the fixed header, so the copy is + placed at `data_offset` and the payload is taken from `src->data`, not from `src->buffer`. */ +static esp_err_t msg_copy_at_offset(mqtt_message_t *message, const mqtt_message_t *src, size_t data_offset) +{ + if (message->buffer_length < data_offset + src->length) { + return ESP_ERR_INVALID_SIZE; + } + + message->length = src->length; + message->fragmented_msg_data_offset = src->fragmented_msg_data_offset; + message->fragmented_msg_total_length = src->fragmented_msg_total_length; + message->read_len = src->read_len; + message->id = src->id; + message->type = src->type; + message->qos = src->qos; + message->data = message->buffer + data_offset; + memcpy(message->data, src->data, src->length); + return ESP_OK; +} + +mqtt_message_t *mqtt_msg_dup(const mqtt_message_t *message) +{ + if (!message || !message->buffer || !message->data) { + return NULL; + } + + mqtt_message_t *new_message = mqtt_msg_create(message->buffer_length); + + if (!new_message) { + return NULL; + } + + if (msg_copy_at_offset(new_message, message, message->data - message->buffer) != ESP_OK) { + mqtt_msg_destroy(new_message); + return NULL; + } + + return new_message; +} + +/* The copy is compacted: `data` starts at the beginning of the destination buffer, so only + `src->length` bytes are required in it. */ +esp_err_t mqtt_msg_copy(mqtt_message_t *message, const mqtt_message_t *src) +{ + if (!message || !src || !message->buffer || !src->data) { + return ESP_ERR_INVALID_ARG; + } + + return msg_copy_at_offset(message, src, 0); +} + +esp_err_t mqtt_msg_append(mqtt_message_t *message, const uint8_t *data, size_t length) +{ + if (!message || !message->buffer || !message->data || !data) { + return ESP_ERR_INVALID_ARG; + } + + size_t data_offset = message->data - message->buffer; + + if (data_offset + message->length + length > message->buffer_length) { + return ESP_ERR_INVALID_SIZE; + } + + memcpy(message->data + message->length, data, length); + message->length += length; + return ESP_OK; +} diff --git a/lib/mqtt_outbox.c b/lib/mqtt_outbox.c index abf2f17..1eb4a7e 100644 --- a/lib/mqtt_outbox.c +++ b/lib/mqtt_outbox.c @@ -10,17 +10,14 @@ #include "mqtt_config.h" #include "sys/queue.h" #include "esp_heap_caps.h" +#include "mqtt_msg.h" #include "esp_log.h" #ifndef CONFIG_MQTT_CUSTOM_OUTBOX static const char *TAG = "outbox"; typedef struct outbox_item { - char *buffer; - int len; - int msg_id; - int msg_type; - int msg_qos; + mqtt_message_t *message; outbox_tick_t tick; pending_state_t pending; STAILQ_ENTRY(outbox_item) next; @@ -44,31 +41,23 @@ outbox_handle_t outbox_init(void) return outbox; } -outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, outbox_message_handle_t message, outbox_tick_t tick) +static void outbox_item_free(outbox_item_handle_t item) +{ + mqtt_msg_destroy(item->message); + free(item); +} + +outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, mqtt_message_t *message, outbox_tick_t tick) { outbox_item_handle_t item = calloc(1, sizeof(outbox_item_t)); ESP_MEM_CHECK(TAG, item, return NULL); - item->msg_id = message->msg_id; - item->msg_type = message->msg_type; - item->msg_qos = message->msg_qos; + item->message = message; item->tick = tick; - item->len = message->len + message->remaining_len; item->pending = QUEUED; - item->buffer = heap_caps_malloc(message->len + message->remaining_len, MQTT_OUTBOX_MEMORY); - ESP_MEM_CHECK(TAG, item->buffer, { - free(item); - return NULL; - }); - memcpy(item->buffer, message->data, message->len); - - if (message->remaining_data) { - memcpy(item->buffer + message->len, message->remaining_data, message->remaining_len); - } - STAILQ_INSERT_TAIL(outbox->list, item, next); - outbox->size += item->len; - ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%d, size=%"PRIu64, message->msg_id, message->msg_type, - message->len + message->remaining_len, outbox_get_size(outbox)); + outbox->size += item->message->length; + ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%zu, size=%"PRIu64, message->id, message->type, + message->length, outbox_get_size(outbox)); return item; } @@ -76,7 +65,7 @@ outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id) { outbox_item_handle_t item; STAILQ_FOREACH(item, outbox->list, next) { - if (item->msg_id == msg_id) { + if (item->message->id == msg_id) { return item; } } @@ -104,11 +93,10 @@ esp_err_t outbox_delete_item(outbox_handle_t outbox, outbox_item_handle_t item_t STAILQ_FOREACH(item, outbox->list, next) { if (item == item_to_delete) { STAILQ_REMOVE(outbox->list, item, outbox_item, next); - outbox->size -= item->len; - ESP_LOGD(TAG, "DELETE_ITEM msgid=%d, msg_type=%d, remain size=%"PRIu64, item_to_delete->msg_id, - item_to_delete->msg_type, outbox_get_size(outbox)); - free(item->buffer); - free(item); + outbox->size -= item->message->length; + ESP_LOGD(TAG, "DELETE_ITEM msgid=%d, msg_type=%d, remain size=%"PRIu64, item->message->id, + item->message->type, outbox_get_size(outbox)); + outbox_item_free(item); return ESP_OK; } } @@ -118,11 +106,11 @@ esp_err_t outbox_delete_item(outbox_handle_t outbox, outbox_item_handle_t item_t uint8_t *outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos) { if (item) { - *len = item->len; - *msg_id = item->msg_id; - *msg_type = item->msg_type; - *qos = item->msg_qos; - return (uint8_t *)item->buffer; + *len = item->message->length; + *msg_id = item->message->id; + *msg_type = item->message->type; + *qos = item->message->qos; + return (uint8_t *)item->message->data; } return NULL; @@ -132,12 +120,11 @@ esp_err_t outbox_delete(outbox_handle_t outbox, int msg_id, int msg_type) { outbox_item_handle_t item, tmp; STAILQ_FOREACH_SAFE(item, outbox->list, next, tmp) { - if (item->msg_id == msg_id && (0xFF & (item->msg_type)) == msg_type) { + if (item->message->id == msg_id && (0xFF & (item->message->type)) == msg_type) { STAILQ_REMOVE(outbox->list, item, outbox_item, next); - outbox->size -= item->len; + outbox->size -= item->message->length; + outbox_item_free(item); ESP_LOGD(TAG, "DELETE msgid=%d, msg_type=%d, remain size=%"PRIu64, msg_id, msg_type, outbox_get_size(outbox)); - free(item->buffer); - free(item); return ESP_OK; } } @@ -184,10 +171,9 @@ int outbox_delete_single_expired(outbox_handle_t outbox, outbox_tick_t current_t STAILQ_FOREACH(item, outbox->list, next) { if (current_tick - item->tick > timeout) { STAILQ_REMOVE(outbox->list, item, outbox_item, next); - free(item->buffer); - outbox->size -= item->len; - msg_id = item->msg_id; - free(item); + outbox->size -= item->message->length; + msg_id = item->message->id; + outbox_item_free(item); ESP_LOGD(TAG, "DELETE_SINGLE_EXPIRED msgid=%d, remain size=%"PRIu64, msg_id, outbox_get_size(outbox)); return msg_id; } @@ -202,10 +188,9 @@ int outbox_delete_expired(outbox_handle_t outbox, outbox_tick_t current_tick, ou STAILQ_FOREACH_SAFE(item, outbox->list, next, tmp) { if (current_tick - item->tick > timeout) { STAILQ_REMOVE(outbox->list, item, outbox_item, next); - free(item->buffer); - outbox->size -= item->len; - ESP_LOGD(TAG, "DELETE_EXPIRED msgid=%d, remain size=%"PRIu64, item->msg_id, outbox_get_size(outbox)); - free(item); + outbox->size -= item->message->length; + ESP_LOGD(TAG, "DELETE_EXPIRED msgid=%d, remain size=%"PRIu64, item->message->id, outbox_get_size(outbox)); + outbox_item_free(item); deleted_items ++; } } @@ -219,18 +204,26 @@ uint64_t outbox_get_size(outbox_handle_t outbox) void outbox_delete_all_items(outbox_handle_t outbox) { + if (!outbox) { + return; + } + outbox_item_handle_t item, tmp; STAILQ_FOREACH_SAFE(item, outbox->list, next, tmp) { STAILQ_REMOVE(outbox->list, item, outbox_item, next); - outbox->size -= item->len; - ESP_LOGD(TAG, "DELETE_ALL_ITEMS msgid=%d, msg_type=%d, remain size=%"PRIu64, item->msg_id, item->msg_type, - outbox_get_size(outbox)); - free(item->buffer); - free(item); + outbox->size -= item->message->length; + ESP_LOGD(TAG, "DELETE_ALL_ITEMS msgid=%d, msg_type=%d, remain size=%"PRIu64, item->message->id, + item->message->type, outbox_get_size(outbox)); + outbox_item_free(item); } } + void outbox_destroy(outbox_handle_t outbox) { + if (!outbox) { + return; + } + outbox_delete_all_items(outbox); free(outbox->list); free(outbox); diff --git a/lib/mqtt_outbox/CMakeLists.txt b/lib/mqtt_outbox/CMakeLists.txt index 126fd99..2ef8859 100644 --- a/lib/mqtt_outbox/CMakeLists.txt +++ b/lib/mqtt_outbox/CMakeLists.txt @@ -2,7 +2,8 @@ set(srcs "${CMAKE_CURRENT_LIST_DIR}/../mqtt_outbox.c") add_library(mqtt_outbox_lib ${srcs}) target_include_directories(mqtt_outbox_lib PUBLIC - ${CMAKE_CURRENT_LIST_DIR}/../include) + ${CMAKE_CURRENT_LIST_DIR}/../include + ${CMAKE_CURRENT_LIST_DIR}/../../include) idf_component_get_property(heap heap COMPONENT_LIB) idf_component_get_property(log log COMPONENT_LIB) target_link_libraries(mqtt_outbox_lib PRIVATE ${heap} ${log}) diff --git a/mqtt5_client.c b/mqtt5_client.c index c837fd3..96149fc 100644 --- a/mqtt5_client.c +++ b/mqtt5_client.c @@ -519,6 +519,7 @@ static esp_err_t esp_mqtt5_user_property_copy(mqtt5_user_property_handle_t user_ ESP_LOGE(TAG, "Input is NULL"); return ESP_FAIL; } + return mqtt5_msg_copy_user_property(user_property_new, user_property_old); } diff --git a/mqtt_client.c b/mqtt_client.c index 05a9852..8673e4b 100644 --- a/mqtt_client.c +++ b/mqtt_client.c @@ -1403,20 +1403,36 @@ static bool remove_initiator_message(esp_mqtt_client_handle_t client, int msg_ty return false; } -static outbox_item_handle_t mqtt_enqueue(esp_mqtt_client_handle_t client, const uint8_t *remaining_data, int remaining_len) +static outbox_item_handle_t mqtt_enqueue(outbox_handle_t outbox, mqtt_message_t *message, const uint8_t *remaining_data, + int remaining_len) { - ESP_LOGD(TAG, "mqtt_enqueue id: %d, type=%d successful", - client->mqtt_state.connection.outbound_message.id, client->mqtt_state.connection.outbound_message.type); - outbox_message_t msg = { 0 }; - msg.data = client->mqtt_state.connection.outbound_message.data; - msg.len = client->mqtt_state.connection.outbound_message.length; - msg.msg_id = client->mqtt_state.connection.outbound_message.id; - msg.msg_type = client->mqtt_state.connection.outbound_message.type; - msg.msg_qos = client->mqtt_state.connection.outbound_message.qos; - msg.remaining_data = remaining_data; - msg.remaining_len = remaining_len; - //Copy to queue buffer - return outbox_enqueue(client->outbox, &msg, platform_tick_get_ms()); + ESP_LOGD(TAG, "mqtt_enqueue id: %d, type=%d", message->id, message->type); + size_t append_len = (remaining_data && remaining_len > 0) ? (size_t)remaining_len : 0; + mqtt_message_t *new_message = mqtt_msg_create(message->length + append_len); + + if (!new_message) { + ESP_LOGE(TAG, "Failed to allocate message to be stored in the outbox"); + return NULL; + } + + if (mqtt_msg_copy(new_message, message) != ESP_OK) { + mqtt_msg_destroy(new_message); + return NULL; + } + + if (append_len > 0 && mqtt_msg_append(new_message, remaining_data, append_len) != ESP_OK) { + mqtt_msg_destroy(new_message); + return NULL; + } + + outbox_item_handle_t item = outbox_enqueue(outbox, new_message, platform_tick_get_ms()); + + if (!item) { + /* The outbox takes ownership of the message only when the item is created. */ + mqtt_msg_destroy(new_message); + } + + return item; } /* @@ -2377,7 +2393,7 @@ int esp_mqtt_client_subscribe_multiple(esp_mqtt_client_handle_t client, client->mqtt_state.connection.outbound_message.type = mqtt_get_type(client->mqtt_state.connection.outbound_message.data); //move pending msg to outbox (if have) - if (!mqtt_enqueue(client, NULL, 0)) { + if (!mqtt_enqueue(client->outbox, &client->mqtt_state.connection.outbound_message, NULL, 0)) { MQTT_API_UNLOCK(client); return -1; } @@ -2445,7 +2461,7 @@ int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *top ESP_LOGD(TAG, "unsubscribe, topic\"%s\", id: %d", topic, client->mqtt_state.connection.outbound_message.id); client->mqtt_state.connection.outbound_message.type = mqtt_get_type(client->mqtt_state.connection.outbound_message.data); - if (!mqtt_enqueue(client, NULL, 0)) { + if (!mqtt_enqueue(client->outbox, &client->mqtt_state.connection.outbound_message, NULL, 0)) { MQTT_API_UNLOCK(client); return -1; } @@ -2516,14 +2532,15 @@ static inline int mqtt_client_enqueue_publish(esp_mqtt_client_handle_t client, c // by default store as QUEUED (not transmitted yet) only for messages which would fit outbound buffer if (client->mqtt_state.connection.outbound_message.fragmented_msg_total_length == 0) { - if (!mqtt_enqueue(client, NULL, 0)) { + if (!mqtt_enqueue(client->outbox, &client->mqtt_state.connection.outbound_message, NULL, 0)) { return -1; } } else { int first_fragment = client->mqtt_state.connection.outbound_message.length - client->mqtt_state.connection.outbound_message.fragmented_msg_data_offset; - if (!mqtt_enqueue(client, (const uint8_t *)data + first_fragment, len - first_fragment)) { + if (!mqtt_enqueue(client->outbox, &client->mqtt_state.connection.outbound_message, + ((const uint8_t *)data) + first_fragment, len - first_fragment)) { return -1; } diff --git a/test/mqtt_outbox_host_test/main/CMakeLists.txt b/test/mqtt_outbox_host_test/main/CMakeLists.txt index bd3e3ec..7958a2e 100644 --- a/test/mqtt_outbox_host_test/main/CMakeLists.txt +++ b/test/mqtt_outbox_host_test/main/CMakeLists.txt @@ -10,8 +10,19 @@ target_link_options(${COMPONENT_LIB} INTERFACE ${SANITIZER_FLAGS}) add_subdirectory(${CMAKE_CURRENT_LIST_DIR}/../../../lib/mqtt_outbox ${CMAKE_CURRENT_BINARY_DIR}/mqtt_outbox) +# mqtt_outbox.c now owns mqtt_message_t and calls mqtt_msg_destroy(). +add_library(mqtt_msg_lib "${CMAKE_CURRENT_LIST_DIR}/../../../lib/mqtt_msg.c") +target_include_directories(mqtt_msg_lib PUBLIC + ${CMAKE_CURRENT_LIST_DIR}/../../../lib/include + ${CMAKE_CURRENT_LIST_DIR}/../../../include) +idf_component_get_property(heap heap COMPONENT_LIB) +idf_component_get_property(log log COMPONENT_LIB) +target_link_libraries(mqtt_msg_lib PRIVATE ${heap} ${log}) +target_compile_options(mqtt_msg_lib PRIVATE ${SANITIZER_FLAGS}) + target_link_libraries(${COMPONENT_LIB} PUBLIC idf::mqtt::outbox + mqtt_msg_lib Catch2::Catch2WithMain rapidcheck rapidcheck_catch) @@ -19,4 +30,6 @@ target_link_libraries(${COMPONENT_LIB} PUBLIC if(CONFIG_GCOV_ENABLED) target_compile_options(mqtt_outbox_lib PUBLIC --coverage -fprofile-arcs -ftest-coverage) target_link_options(mqtt_outbox_lib PUBLIC --coverage -fprofile-arcs -ftest-coverage) + target_compile_options(mqtt_msg_lib PUBLIC --coverage -fprofile-arcs -ftest-coverage) + target_link_options(mqtt_msg_lib PUBLIC --coverage -fprofile-arcs -ftest-coverage) endif() diff --git a/test/mqtt_outbox_host_test/main/test_outbox.cpp b/test/mqtt_outbox_host_test/main/test_outbox.cpp index 579f935..29149ce 100644 --- a/test/mqtt_outbox_host_test/main/test_outbox.cpp +++ b/test/mqtt_outbox_host_test/main/test_outbox.cpp @@ -21,6 +21,7 @@ extern "C" { #include "mqtt_outbox.h" +#include "mqtt_msg.h" } struct OutboxGuard { @@ -38,20 +39,36 @@ struct OutboxGuard { outbox_handle_t handle; }; -static outbox_message_t make_msg(int msg_id, int qos, int msg_type, - const char *payload, int len) +static mqtt_message_t *make_msg(int msg_id, int qos, int msg_type, + const char *payload, int len) { - outbox_message_t message{}; - message.msg_id = msg_id; - message.msg_qos = qos; - message.msg_type = msg_type; - message.data = reinterpret_cast(const_cast(payload)); - message.len = len; - message.remaining_data = nullptr; - message.remaining_len = 0; + REQUIRE(len >= 0); + mqtt_message_t *message = mqtt_msg_create(static_cast(len)); + REQUIRE(message != nullptr); + + if (len > 0) { + REQUIRE(mqtt_msg_append(message, reinterpret_cast(payload), + static_cast(len)) == ESP_OK); + } + + message->id = static_cast(msg_id); + message->qos = qos; + message->type = msg_type; return message; } +/* Transfers ownership on success; destroys the message if enqueue fails. */ +static outbox_item_handle_t enqueue(outbox_handle_t outbox, mqtt_message_t *message, outbox_tick_t tick) +{ + outbox_item_handle_t item = outbox_enqueue(outbox, message, tick); + + if (item == nullptr) { + mqtt_msg_destroy(message); + } + + return item; +} + TEST_CASE("Outbox lifecycle") { SECTION("init returns a non-null handle") { @@ -61,8 +78,7 @@ TEST_CASE("Outbox lifecycle") } SECTION("destroy on a non-empty outbox reclaims all items") { OutboxGuard outbox; - auto message = make_msg(1, 1, 3, "hello", 5); - REQUIRE(outbox_enqueue(outbox.handle, &message, 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, "hello", 5), 0) != nullptr); // destructor calls outbox_destroy — LSan catches leaks if it is omitted } } @@ -71,23 +87,19 @@ TEST_CASE("Outbox enqueue") { OutboxGuard outbox; SECTION("enqueued item starts in QUEUED state") { - auto message = make_msg(42, 1, 3, "data", 4); - outbox_item_handle_t item = outbox_enqueue(outbox.handle, &message, 0); + outbox_item_handle_t item = enqueue(outbox.handle, make_msg(42, 1, 3, "data", 4), 0); REQUIRE(item != nullptr); REQUIRE(outbox_item_get_pending(item) == QUEUED); } SECTION("size increases by item length after enqueue") { REQUIRE(outbox_get_size(outbox.handle) == 0); const char payload[] = "hello"; - auto message = make_msg(1, 1, 3, payload, static_cast(sizeof(payload) - 1)); - outbox_enqueue(outbox.handle, &message, 0); + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, payload, static_cast(sizeof(payload) - 1)), 0) != nullptr); REQUIRE(outbox_get_size(outbox.handle) == sizeof(payload) - 1); } SECTION("multiple enqueues accumulate size") { - auto message1 = make_msg(1, 1, 3, "abc", 3); - auto message2 = make_msg(2, 1, 3, "de", 2); - outbox_enqueue(outbox.handle, &message1, 0); - outbox_enqueue(outbox.handle, &message2, 0); + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, "abc", 3), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(2, 1, 3, "de", 2), 0) != nullptr); REQUIRE(outbox_get_size(outbox.handle) == 5); } } @@ -96,12 +108,9 @@ TEST_CASE("Outbox FIFO dequeue") { OutboxGuard outbox; SECTION("dequeue returns items in enqueue order") { - auto message1 = make_msg(10, 1, 3, "first", 5); - auto message2 = make_msg(20, 1, 3, "second", 6); - auto message3 = make_msg(30, 1, 3, "third", 5); - outbox_enqueue(outbox.handle, &message1, 0); - outbox_enqueue(outbox.handle, &message2, 0); - outbox_enqueue(outbox.handle, &message3, 0); + REQUIRE(enqueue(outbox.handle, make_msg(10, 1, 3, "first", 5), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(20, 1, 3, "second", 6), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(30, 1, 3, "third", 5), 0) != nullptr); uint16_t id; int type, qos; size_t len; @@ -117,8 +126,7 @@ TEST_CASE("Outbox FIFO dequeue") REQUIRE(id == 20); } SECTION("dequeue returns nullptr when no item matches state") { - auto message = make_msg(1, 1, 3, "x", 1); - outbox_enqueue(outbox.handle, &message, 0); + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, "x", 1), 0) != nullptr); REQUIRE(outbox_dequeue(outbox.handle, TRANSMITTED, nullptr) == nullptr); REQUIRE(outbox_dequeue(outbox.handle, ACKNOWLEDGED, nullptr) == nullptr); } @@ -131,8 +139,7 @@ TEST_CASE("Outbox state transitions") { OutboxGuard outbox; SECTION("set_pending changes the state of an item") { - auto message = make_msg(5, 1, 3, "msg", 3); - outbox_enqueue(outbox.handle, &message, 0); + REQUIRE(enqueue(outbox.handle, make_msg(5, 1, 3, "msg", 3), 0) != nullptr); REQUIRE(outbox_set_pending(outbox.handle, 5, TRANSMITTED) == ESP_OK); outbox_item_handle_t item = outbox_dequeue(outbox.handle, TRANSMITTED, nullptr); REQUIRE(item != nullptr); @@ -143,8 +150,7 @@ TEST_CASE("Outbox state transitions") REQUIRE(id == 5); } SECTION("full state cycle: QUEUED -> TRANSMITTED -> ACKNOWLEDGED -> CONFIRMED") { - auto message = make_msg(7, 2, 3, "qos2", 4); - outbox_enqueue(outbox.handle, &message, 0); + REQUIRE(enqueue(outbox.handle, make_msg(7, 2, 3, "qos2", 4), 0) != nullptr); REQUIRE(outbox_set_pending(outbox.handle, 7, TRANSMITTED) == ESP_OK); REQUIRE(outbox_set_pending(outbox.handle, 7, ACKNOWLEDGED) == ESP_OK); REQUIRE(outbox_set_pending(outbox.handle, 7, CONFIRMED) == ESP_OK); @@ -159,10 +165,8 @@ TEST_CASE("Outbox concurrent states") { OutboxGuard outbox; SECTION("QUEUED head does not shadow a later TRANSMITTED item") { - auto message1 = make_msg(1, 1, 3, "first", 5); - auto message2 = make_msg(2, 1, 3, "second", 6); - outbox_enqueue(outbox.handle, &message1, 0); - outbox_enqueue(outbox.handle, &message2, 0); + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, "first", 5), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(2, 1, 3, "second", 6), 0) != nullptr); // Promote the second item only outbox_set_pending(outbox.handle, 2, TRANSMITTED); // dequeue(QUEUED) must still return msg 1 @@ -178,8 +182,7 @@ TEST_CASE("Outbox concurrent states") REQUIRE(id == 2); } SECTION("TRANSMITTED item is not visible to dequeue(QUEUED)") { - auto message = make_msg(3, 1, 3, "only", 4); - outbox_enqueue(outbox.handle, &message, 0); + REQUIRE(enqueue(outbox.handle, make_msg(3, 1, 3, "only", 4), 0) != nullptr); outbox_set_pending(outbox.handle, 3, TRANSMITTED); REQUIRE(outbox_dequeue(outbox.handle, QUEUED, nullptr) == nullptr); } @@ -189,10 +192,8 @@ TEST_CASE("Outbox lookup by msg_id") { OutboxGuard outbox; SECTION("outbox_get returns the correct item") { - auto message1 = make_msg(100, 1, 3, "a", 1); - auto message2 = make_msg(200, 1, 3, "b", 1); - outbox_enqueue(outbox.handle, &message1, 0); - outbox_enqueue(outbox.handle, &message2, 0); + REQUIRE(enqueue(outbox.handle, make_msg(100, 1, 3, "a", 1), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(200, 1, 3, "b", 1), 0) != nullptr); outbox_item_handle_t item = outbox_get(outbox.handle, 200); REQUIRE(item != nullptr); uint16_t id; int type, qos; size_t len; @@ -200,17 +201,13 @@ TEST_CASE("Outbox lookup by msg_id") REQUIRE(id == 200); } SECTION("outbox_get returns nullptr for unknown msg_id") { - auto message = make_msg(1, 1, 3, "x", 1); - outbox_enqueue(outbox.handle, &message, 0); + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, "x", 1), 0) != nullptr); REQUIRE(outbox_get(outbox.handle, 999) == nullptr); } SECTION("msg_id zero finds queued QoS 0 behind a QoS 1 head") { - auto qos1 = make_msg(1, 1, 3, "qos1", 4); - auto qos0_first = make_msg(0, 0, 3, "first", 5); - auto qos0_second = make_msg(0, 0, 3, "second", 6); - outbox_enqueue(outbox.handle, &qos1, 0); - outbox_enqueue(outbox.handle, &qos0_first, 0); - outbox_enqueue(outbox.handle, &qos0_second, 0); + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, "qos1", 4), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(0, 0, 3, "first", 5), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(0, 0, 3, "second", 6), 0) != nullptr); REQUIRE(outbox_dequeue(outbox.handle, QUEUED, nullptr) == outbox_get(outbox.handle, 1)); outbox_item_handle_t item = outbox_get(outbox.handle, 0); REQUIRE(item != nullptr); @@ -235,16 +232,14 @@ TEST_CASE("Outbox delete by msg_id and type") { OutboxGuard outbox; SECTION("deleted item is no longer found") { - auto message = make_msg(55, 1, 3, "del", 3); - outbox_enqueue(outbox.handle, &message, 0); + REQUIRE(enqueue(outbox.handle, make_msg(55, 1, 3, "del", 3), 0) != nullptr); REQUIRE(outbox_get_size(outbox.handle) == 3); REQUIRE(outbox_delete(outbox.handle, 55, 3) == ESP_OK); REQUIRE(outbox_get(outbox.handle, 55) == nullptr); REQUIRE(outbox_get_size(outbox.handle) == 0); } SECTION("delete with wrong type returns ESP_FAIL") { - auto message = make_msg(56, 1, 3, "x", 1); - outbox_enqueue(outbox.handle, &message, 0); + REQUIRE(enqueue(outbox.handle, make_msg(56, 1, 3, "x", 1), 0) != nullptr); REQUIRE(outbox_delete(outbox.handle, 56, 99) == ESP_FAIL); REQUIRE(outbox_get(outbox.handle, 56) != nullptr); } @@ -257,8 +252,7 @@ TEST_CASE("Outbox delete by item handle") { OutboxGuard outbox; SECTION("item is removed and size decreases") { - auto message = make_msg(77, 1, 3, "item", 4); - outbox_item_handle_t item = outbox_enqueue(outbox.handle, &message, 0); + outbox_item_handle_t item = enqueue(outbox.handle, make_msg(77, 1, 3, "item", 4), 0); REQUIRE(item != nullptr); REQUIRE(outbox_get_size(outbox.handle) == 4); REQUIRE(outbox_delete_item(outbox.handle, item) == ESP_OK); @@ -272,20 +266,16 @@ TEST_CASE("Outbox expiry") OutboxGuard outbox; SECTION("delete_expired removes items older than timeout") { // enqueue with tick=0; current_tick=200, timeout=100 → age=200 > 100 - auto message1 = make_msg(1, 1, 3, "old", 3); - auto message2 = make_msg(2, 1, 3, "fresh", 5); - outbox_enqueue(outbox.handle, &message1, 0); - outbox_enqueue(outbox.handle, &message2, 150); // tick=150, age=50 at current_tick=200 + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, "old", 3), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(2, 1, 3, "fresh", 5), 150) != nullptr); // tick=150, age=50 at current_tick=200 int deleted = outbox_delete_expired(outbox.handle, 200, 100); REQUIRE(deleted == 1); REQUIRE(outbox_get(outbox.handle, 1) == nullptr); REQUIRE(outbox_get(outbox.handle, 2) != nullptr); } SECTION("delete_single_expired removes exactly one item") { - auto message1 = make_msg(10, 1, 3, "old1", 4); - auto message2 = make_msg(20, 1, 3, "old2", 4); - outbox_enqueue(outbox.handle, &message1, 0); - outbox_enqueue(outbox.handle, &message2, 0); + REQUIRE(enqueue(outbox.handle, make_msg(10, 1, 3, "old1", 4), 0) != nullptr); + REQUIRE(enqueue(outbox.handle, make_msg(20, 1, 3, "old2", 4), 0) != nullptr); // Both are expired at current_tick=500, timeout=100; only one is removed int id = outbox_delete_single_expired(outbox.handle, 500, 100); REQUIRE(id >= 0); @@ -295,8 +285,7 @@ TEST_CASE("Outbox expiry") REQUIRE(remaining == 1); } SECTION("delete_expired returns 0 when no items are expired") { - auto message = make_msg(1, 1, 3, "fresh", 5); - outbox_enqueue(outbox.handle, &message, 1000); + REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, "fresh", 5), 1000) != nullptr); int deleted = outbox_delete_expired(outbox.handle, 1050, 100); REQUIRE(deleted == 0); } @@ -319,8 +308,8 @@ TEST_CASE("Outbox size invariant (RapidCheck)") for (int i = 0; i < count; ++i) { int len = *rc::gen::inRange(1, 32); std::string payload(len, 'x'); - auto message = make_msg(i + 1, 1, 3, payload.c_str(), len); - const bool enqueued = outbox_enqueue(outbox.handle, &message, 0) != nullptr; + mqtt_message_t *message = make_msg(i + 1, 1, 3, payload.c_str(), len); + const bool enqueued = enqueue(outbox.handle, message, 0) != nullptr; RC_ASSERT(enqueued); items.push_back({i + 1, len}); expected_size += len; @@ -355,9 +344,9 @@ TEST_CASE("Outbox FIFO ordering property (RapidCheck)") int id = i + 1; msg_ids.push_back(id); std::string payload = "p" + std::to_string(id); - auto message = make_msg(id, 1, 3, payload.c_str(), - static_cast(payload.size())); - const bool enqueued = outbox_enqueue(outbox.handle, &message, 0) != nullptr; + mqtt_message_t *message = make_msg(id, 1, 3, payload.c_str(), + static_cast(payload.size())); + const bool enqueued = enqueue(outbox.handle, message, 0) != nullptr; RC_ASSERT(enqueued); } @@ -379,3 +368,101 @@ TEST_CASE("Outbox FIFO ordering property (RapidCheck)") RC_ASSERT(no_queued_items); }); } + +TEST_CASE("mqtt_msg_dup points data into its own buffer") +{ + mqtt_message_t *src = mqtt_msg_create(16); + REQUIRE(src != nullptr); + const uint8_t payload[] = {0x10, 0x20, 0x30, 0x40}; + const size_t offset = 3; + src->data = src->buffer + offset; + memcpy(src->data, payload, sizeof(payload)); + src->length = sizeof(payload); + src->id = 9; + src->type = 3; + src->qos = 1; + mqtt_message_t *dup = mqtt_msg_dup(src); + REQUIRE(dup != nullptr); + REQUIRE(dup->buffer != src->buffer); + REQUIRE(dup->data == dup->buffer + offset); + REQUIRE(dup->data != src->data); + REQUIRE(dup->length == sizeof(payload)); + REQUIRE(dup->id == 9); + REQUIRE(memcmp(dup->data, payload, sizeof(payload)) == 0); + src->data[0] = 0xFF; + REQUIRE(dup->data[0] == 0x10); + mqtt_msg_destroy(src); + mqtt_msg_destroy(dup); +} + +TEST_CASE("mqtt_msg_append copies at the previous length") +{ + mqtt_message_t *message = mqtt_msg_create(16); + REQUIRE(message != nullptr); + const uint8_t first[] = {'A', 'B'}; + const uint8_t second[] = {'C', 'D', 'E'}; + REQUIRE(mqtt_msg_append(message, first, sizeof(first)) == ESP_OK); + REQUIRE(message->length == sizeof(first)); + REQUIRE(mqtt_msg_append(message, second, sizeof(second)) == ESP_OK); + REQUIRE(message->length == 5); + REQUIRE(std::string(reinterpret_cast(message->data), message->length) == "ABCDE"); + mqtt_msg_destroy(message); +} + +TEST_CASE("mqtt_msg_copy compacting a header-offset message") +{ + mqtt_message_t *src = mqtt_msg_create(16); + REQUIRE(src != nullptr); + const uint8_t payload[] = {'P', 'Q', 'R'}; + src->data = src->buffer + 2; + memcpy(src->data, payload, sizeof(payload)); + src->length = sizeof(payload); + src->id = 11; + src->type = 3; + src->qos = 1; + mqtt_message_t *dst = mqtt_msg_create(sizeof(payload)); + REQUIRE(dst != nullptr); + REQUIRE(mqtt_msg_copy(dst, src) == ESP_OK); + REQUIRE(dst->data == dst->buffer); + REQUIRE(dst->length == sizeof(payload)); + REQUIRE(memcmp(dst->data, payload, sizeof(payload)) == 0); + mqtt_msg_destroy(src); + mqtt_msg_destroy(dst); +} + +TEST_CASE("Outbox stores compacted copy plus remaining fragment") +{ + OutboxGuard outbox; + mqtt_message_t *src = mqtt_msg_create(16); + REQUIRE(src != nullptr); + const uint8_t first_fragment[] = {'H', 'e'}; + const uint8_t remaining[] = {'l', 'l', 'o'}; + const size_t offset = 2; + src->data = src->buffer + offset; + memcpy(src->data, first_fragment, sizeof(first_fragment)); + src->length = sizeof(first_fragment); + src->id = 8; + src->type = 3; + src->qos = 1; + src->fragmented_msg_data_offset = sizeof(first_fragment); + src->fragmented_msg_total_length = sizeof(first_fragment) + sizeof(remaining); + mqtt_message_t *stored = mqtt_msg_create(src->length + sizeof(remaining)); + REQUIRE(stored != nullptr); + REQUIRE(mqtt_msg_copy(stored, src) == ESP_OK); + REQUIRE(mqtt_msg_append(stored, remaining, sizeof(remaining)) == ESP_OK); + REQUIRE(stored->data == stored->buffer); + REQUIRE(stored->length == 5); + outbox_item_handle_t item = enqueue(outbox.handle, stored, 25); + REQUIRE(item != nullptr); + uint16_t id; + int type, qos; + size_t len; + uint8_t *data = outbox_item_get_data(item, &len, &id, &type, &qos); + REQUIRE(id == 8); + REQUIRE(len == 5); + REQUIRE(std::string(reinterpret_cast(data), len) == "Hello"); + outbox_tick_t tick = 0; + REQUIRE(outbox_dequeue(outbox.handle, QUEUED, &tick) == item); + REQUIRE(tick == 25); + mqtt_msg_destroy(src); +}