bring in mqtt3 changes

Author: xiazhvera

include/aws/mqtt/client.h

@@ -693,6 +693,21 @@ int aws_mqtt_client_connection_get_stats(
     struct aws_mqtt_client_connection *connection,
     struct aws_mqtt_connection_operation_statistics *stats);
 
+/**
+ * Sets IoT SDK metrics configuration for the connection.
+ * These metrics will be appended to the username field during connection.
+ *
+ * NOTE: DO NOT USE METADATA. Metadata will not be set.
+ *
+ * \param connection The connection object
+ * \param metrics The IoT SDK metrics configuration (pass NULL to disable metrics)
+ * \returns AWS_OP_SUCCESS if successful, AWS_OP_ERR otherwise
+ */
+AWS_MQTT_API
+int aws_mqtt_client_connection_set_metrics(
+    struct aws_mqtt_client_connection *connection,
+    const struct aws_mqtt_iot_sdk_metrics *metrics);
+
 AWS_EXTERN_C_END
 AWS_POP_SANE_WARNING_LEVEL
 

include/aws/mqtt/mqtt.h

@@ -111,6 +111,44 @@ enum aws_mqtt_log_subject {
 /** Function called on cleanup of a userdata. */
 typedef void(aws_mqtt_userdata_cleanup_fn)(void *userdata);
 
+/**
+ * Metadata entry for IoT SDK metrics
+ */
+struct aws_mqtt_metadata_entry {
+    struct aws_byte_cursor key;
+    struct aws_byte_cursor value;
+};
+
+/**
+ * IoT SDK metrics configuration structure
+ */
+struct aws_mqtt_iot_sdk_metrics {
+    /**
+     * TODO: This is a place holder for metadata entries.
+     * The metadata_entries is not added to username yet.
+     *
+     * DO NOT USE.
+     *
+     * Array of metadata entries
+     */
+    // struct aws_mqtt_metadata_entry *metadata_entries;
+
+    /**
+     * TODO: This is a place holder for metadata entries.
+     * The metadata_entries is not added to username yet.
+     *
+     * DO NOT USE.
+     *
+     * Number of metadata entries
+     */
+    // size_t metadata_count;
+
+    /**
+     * Library name string (SDK attribute)
+     */
+    struct aws_byte_cursor library_name;
+};
+
 AWS_EXTERN_C_BEGIN
 
 AWS_MQTT_API

include/aws/mqtt/private/client_impl.h

@@ -365,6 +365,12 @@ struct aws_mqtt_client_connection_311_impl {
      * Statistics tracking operational state
      */
     struct aws_mqtt_connection_operation_statistics_impl operation_statistics_impl;
+
+    /**
+     * IoT SDK metrics configuration
+     */
+    struct aws_mqtt_iot_sdk_metrics_storage *metrics;
+    struct aws_byte_buf username_with_metrics_buf;
 };
 
 struct aws_channel_handler_vtable *aws_mqtt_get_client_channel_vtable(void);

include/aws/mqtt/private/client_impl_shared.h

@@ -122,6 +122,8 @@ struct aws_mqtt_client_connection_vtable {
 
     int (*get_stats_fn)(void *impl, struct aws_mqtt_connection_operation_statistics *stats);
 
+    int (*set_metrics_fn)(void *impl, const struct aws_mqtt_iot_sdk_metrics *metrics);
+
     enum aws_mqtt311_impl_type (*get_impl_type)(const void *impl);
 
     struct aws_event_loop *(*get_event_loop)(const void *impl);
@@ -132,6 +134,28 @@ struct aws_mqtt_client_connection {
     void *impl;
 };
 
+struct aws_mqtt_iot_sdk_metrics_storage {
+    struct aws_allocator *allocator;
+
+    struct aws_mqtt_iot_sdk_metrics storage_view;
+
+    struct aws_array_list metadata_entries;
+
+    struct aws_byte_cursor library_name;
+
+    struct aws_byte_buf storage;
+};
+
+/* IoT SDK Metrics */
+AWS_MQTT_API int aws_mqtt_iot_sdk_metrics_storage_init(
+    struct aws_mqtt_iot_sdk_metrics_storage *metrics_storage,
+    struct aws_allocator *allocator,
+    const struct aws_mqtt_iot_sdk_metrics *metrics_options);
+
+AWS_MQTT_API size_t aws_mqtt_iot_sdk_metrics_compute_storage_size(const struct aws_mqtt_iot_sdk_metrics *metrics);
+
+AWS_MQTT_API void aws_mqtt_iot_sdk_metrics_storage_clean_up(struct aws_mqtt_iot_sdk_metrics_storage *metrics_storage);
+
 AWS_MQTT_API enum aws_mqtt311_impl_type aws_mqtt_client_connection_get_impl_type(
     const struct aws_mqtt_client_connection *connection);
 
@@ -144,4 +168,38 @@ AWS_MQTT_API bool aws_mqtt_byte_cursor_hash_equality(const void *a, const void *
 AWS_MQTT_API struct aws_event_loop *aws_mqtt_client_connection_get_event_loop(
     const struct aws_mqtt_client_connection *connection);
 
+/**
+ * Appends SDK metrics to the username field
+ *
+ * @param original_username The original username
+ * @param metrics The metrics configuration
+ * @param output_username Buffer to store the modified username. If the function succeed, caller is responsible to
+ * release the memory for output_username.
+ *
+ * @return AWS_OP_SUCCESS on success, AWS_OP_ERR on failure
+ */
+AWS_MQTT_API
+int aws_mqtt_append_sdk_metrics_to_username(
+    struct aws_allocator *allocator,
+    const struct aws_byte_cursor *original_username,
+    const struct aws_mqtt_iot_sdk_metrics metrics,
+    struct aws_byte_buf *output_username);
+
+/**
+ * Get final username length
+ */
+AWS_MQTT_API
+size_t aws_mqtt_append_sdk_metrics_to_username_size(
+    const struct aws_byte_cursor *original_username,
+    const struct aws_mqtt_iot_sdk_metrics metrics);
+
+/**
+ * Validates that all string fields in aws_mqtt_iot_sdk_metrics are valid UTF-8
+ *
+ * @param metrics The metrics structure to validate
+ * @return AWS_OP_SUCCESS if all strings are valid UTF-8, AWS_OP_ERR otherwise
+ */
+AWS_MQTT_API
+int aws_mqtt_validate_iot_sdk_metrics_utf8(const struct aws_mqtt_iot_sdk_metrics *metrics);
+
 #endif /* AWS_MQTT_PRIVATE_CLIENT_IMPL_SHARED_H */

source/client.c

@@ -623,8 +623,32 @@ static void s_mqtt_client_init(
             &connect, topic_cur, connection->will.qos, connection->will.retain, payload_cur);
     }
 
-    if (connection->username) {
-        struct aws_byte_cursor username_cur = aws_byte_cursor_from_string(connection->username);
+    if (connection->username || connection->metrics) {
+        struct aws_byte_cursor username_cur;
+        if (connection->username) {
+            username_cur = aws_byte_cursor_from_string(connection->username);
+        } else {
+            /* Create empty username cursor when username is null but metrics is set */
+            username_cur = aws_byte_cursor_from_c_str("");
+        }
+
+        aws_byte_buf_clean_up(&connection->username_with_metrics_buf);
+
+        /* Apply metrics to username if configured */
+        if (connection->metrics) {
+            if (aws_mqtt_append_sdk_metrics_to_username(
+                    connection->allocator,
+                    &username_cur,
+                    connection->metrics->storage_view,
+                    &connection->username_with_metrics_buf) == AWS_OP_SUCCESS) {
+                username_cur = aws_byte_cursor_from_buf(&connection->username_with_metrics_buf);
+            } else {
+                AWS_LOGF_WARN(
+                    AWS_LS_MQTT_CLIENT,
+                    "id=%p: Failed to apply metrics to username, using original",
+                    (void *)connection);
+            }
+        }
 
         AWS_LOGF_DEBUG(
             AWS_LS_MQTT_CLIENT,
@@ -844,6 +868,14 @@ static void s_mqtt_client_connection_destroy_final(struct aws_mqtt_client_connec
         connection->http_proxy_config = NULL;
     }
 
+    /* Clean up metrics */
+    if (connection->metrics) {
+        aws_mqtt_iot_sdk_metrics_storage_clean_up(connection->metrics);
+        connection->metrics = NULL;
+    }
+
+    aws_byte_buf_clean_up(&connection->username_with_metrics_buf);
+
     aws_mqtt_client_release(connection->client);
 
     /* Frees all allocated memory */
@@ -3341,6 +3373,46 @@ int aws_mqtt_client_connection_set_on_operation_statistics_handler(
     return AWS_OP_SUCCESS;
 }
 
+static int s_aws_mqtt_client_connection_311_set_metrics(void *impl, const struct aws_mqtt_iot_sdk_metrics *metrics) {
+
+    struct aws_mqtt_client_connection_311_impl *connection = impl;
+
+    AWS_PRECONDITION(connection);
+    if (s_check_connection_state_for_configuration(connection)) {
+        return aws_raise_error(AWS_ERROR_INVALID_STATE);
+    }
+
+    if (metrics != NULL && aws_mqtt_validate_iot_sdk_metrics_utf8(metrics) == AWS_OP_ERR) {
+        AWS_LOGF_DEBUG(
+            AWS_LS_MQTT_CLIENT, "id=%p: Invalid utf8 or forbidden codepoints in metrics.", (void *)connection);
+        return aws_raise_error(AWS_ERROR_INVALID_UTF8);
+    }
+
+    // TODO: Validate metadata value, when we had metadata in place.
+
+    AWS_LOGF_TRACE(AWS_LS_MQTT_CLIENT, "id=%p: Setting IoT SDK metrics", (void *)connection);
+
+    /* Clean up existing metrics if any */
+    if (connection->metrics) {
+        aws_mqtt_iot_sdk_metrics_storage_clean_up(connection->metrics);
+        connection->metrics = NULL;
+    }
+
+    if (metrics) {
+        /* Allocate metrics storage */
+        connection->metrics = aws_mem_calloc(connection->allocator, 1, sizeof(struct aws_mqtt_iot_sdk_metrics_storage));
+
+        /* Initialize metrics storage */
+        if (aws_mqtt_iot_sdk_metrics_storage_init(connection->metrics, connection->allocator, metrics)) {
+            aws_mem_release(connection->allocator, connection->metrics);
+            connection->metrics = NULL;
+            return AWS_OP_ERR;
+        }
+    }
+
+    return AWS_OP_SUCCESS;
+}
+
 static struct aws_mqtt_client_connection *s_aws_mqtt_client_connection_311_acquire(void *impl) {
     struct aws_mqtt_client_connection_311_impl *connection = impl;
 
@@ -3390,6 +3462,7 @@ static struct aws_mqtt_client_connection_vtable s_aws_mqtt_client_connection_311
     .unsubscribe_fn = s_aws_mqtt_client_connection_311_unsubscribe,
     .publish_fn = s_aws_mqtt_client_connection_311_publish,
     .get_stats_fn = s_aws_mqtt_client_connection_311_get_stats,
+    .set_metrics_fn = s_aws_mqtt_client_connection_311_set_metrics,
     .get_impl_type = s_aws_mqtt_client_connection_311_get_impl,
     .get_event_loop = s_aws_mqtt_client_connection_311_get_event_loop,
 };
@@ -3493,6 +3566,8 @@ struct aws_mqtt_client_connection *aws_mqtt_client_connection_new(struct aws_mqt
     connection->handler.vtable = aws_mqtt_get_client_channel_vtable();
     connection->handler.impl = connection;
 
+    aws_byte_buf_init(&connection->username_with_metrics_buf, connection->allocator, 0);
+
     aws_mqtt311_callback_set_manager_init(&connection->callback_manager, connection->allocator, &connection->base);
 
     return &connection->base;
@@ -3511,3 +3586,11 @@ struct aws_mqtt_client_connection *aws_mqtt_client_connection_new(struct aws_mqt
 
     return NULL;
 }
+
+int aws_mqtt_client_connection_set_metrics(
+    struct aws_mqtt_client_connection *connection,
+    const struct aws_mqtt_iot_sdk_metrics *metrics) {
+
+    AWS_PRECONDITION(connection);
+    return connection->vtable->set_metrics_fn(connection->impl, metrics);
+}