refactor: Makes the outbox to store msg type

- This remove the difference between types and simplify the message
enqueuing from the outbox perspective.

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

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

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

lib/mqtt_msg.c

@@ -3,9 +3,11 @@
  *
  * SPDX-License-Identifier: BSD-3-Clause
  */
+#include <stdlib.h>
 #include <string.h>
 #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;
+}

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

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

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

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

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()

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<uint8_t *>(const_cast<char *>(payload));
-    message.len      = len;
-    message.remaining_data = nullptr;
-    message.remaining_len  = 0;
+    REQUIRE(len >= 0);
+    mqtt_message_t *message = mqtt_msg_create(static_cast<size_t>(len));
+    REQUIRE(message != nullptr);
+
+    if (len > 0) {
+        REQUIRE(mqtt_msg_append(message, reinterpret_cast<const uint8_t *>(payload),
+                                static_cast<size_t>(len)) == ESP_OK);
+    }
+
+    message->id = static_cast<uint16_t>(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<int>(sizeof(payload) - 1));
-        outbox_enqueue(outbox.handle, &message, 0);
+        REQUIRE(enqueue(outbox.handle, make_msg(1, 1, 3, payload, static_cast<int>(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<int>(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<int>(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<char *>(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<char *>(data), len) == "Hello");
+    outbox_tick_t tick = 0;
+    REQUIRE(outbox_dequeue(outbox.handle, QUEUED, &tick) == item);
+    REQUIRE(tick == 25);
+    mqtt_msg_destroy(src);
+}