Merge branch 'manual-puback' into manual-puback-tests

Author: sbSteveK

include/aws/mqtt/private/v5/mqtt5_options_storage.h

@@ -281,11 +281,6 @@ AWS_MQTT_API void aws_mqtt5_packet_puback_view_log(
     const struct aws_mqtt5_packet_puback_view *puback_view,
     enum aws_log_level level);
 
-AWS_MQTT_API struct aws_mqtt5_manual_puback_entry *s_aws_mqtt_manual_puback_entry_new(
-    struct aws_allocator *allocator,
-    uint16_t packet_id,
-    uint64_t puback_control_id);
-
 /* Subscribe */
 
 AWS_MQTT_API struct aws_mqtt5_operation_subscribe *aws_mqtt5_operation_subscribe_new(

include/aws/mqtt/v5/mqtt5_client.h

@@ -300,8 +300,9 @@ typedef void(aws_mqtt5_publish_completion_fn)(
  * @param puback_result result of the PUBACK operation
  * @param complete_ctx user data passed in with the completion options
  */
-typedef void(
-    aws_mqtt5_manual_puback_completion_fn)(enum aws_mqtt5_manual_puback_result puback_result, void *complete_ctx);
+typedef void(aws_mqtt5_manual_puback_completion_fn)(
+    enum aws_mqtt5_manual_puback_result puback_result,
+    void *completion_user_data);
 
 /**
  * Signature of callback to invoke on Subscribe success/failure.
@@ -773,7 +774,7 @@ AWS_MQTT_API uint64_t aws_mqtt5_client_acquire_puback(
  *
  * @param client mqtt5 client to queue a puback for
  * @param puback_control_id Control ID of aws_mqtt5_manual_puback_entry to send to broker/server
- * @return success/failure of the manual PUBACK operation.
+ * @return success/failure of starting the manual PUBACK operation.
  */
 AWS_MQTT_API
 int aws_mqtt5_client_invoke_puback(

source/v5/mqtt5_client.c

@@ -608,7 +608,7 @@ static int s_manual_puback_transfer(void *context, struct aws_hash_element *elem
 }
 
 /* This is called when the manual puback entry is removed from a hashset to properly decref on removal */
-static void aws_mqtt5_manual_puback_entry_decref(void *value) {
+static void s_aws_mqtt5_manual_puback_entry_decref(void *value) {
     struct aws_mqtt5_manual_puback_entry *manual_puback_entry = value;
     if (manual_puback_entry != NULL) {
         aws_ref_count_release(&manual_puback_entry->ref_count);
@@ -622,9 +622,10 @@ static void s_aws_mqtt5_reset_manual_puback_tables(
     struct aws_mqtt5_client_operational_state *client_operational_state) {
     size_t count = aws_hash_table_get_entry_count(&client_operational_state->manual_puback_control_id_table);
     if (count > 0) {
-        AWS_LOGF_INFO(
+        AWS_LOGF_DEBUG(
             AWS_LS_MQTT5_CLIENT,
-            "id=%p: Clearing %zu manual PUBACK control ids.",
+            "id=%p: Clearing %zu PUBACKs under user control. Previously controlled PUBACKs are no longer valid and "
+            "have been cancelled.",
             (void *)client_operational_state->client,
             count);
         aws_hash_table_clear(&client_operational_state->manual_puback_packet_id_table);
@@ -2606,6 +2607,30 @@ int aws_mqtt5_client_invoke_puback(
     return AWS_OP_SUCCESS;
 }
 
+static void s_aws_mqtt5_manual_puback_entry_destroy(void *object) {
+    if (object == NULL) {
+        return;
+    }
+    struct aws_mqtt5_manual_puback_entry *manual_puback_entry = object;
+    aws_mem_release(manual_puback_entry->allocator, manual_puback_entry);
+}
+
+static struct aws_mqtt5_manual_puback_entry *s_aws_mqtt_manual_puback_entry_new(
+    struct aws_allocator *allocator,
+    uint16_t packet_id,
+    uint64_t puback_control_id) {
+
+    struct aws_mqtt5_manual_puback_entry *manual_puback_entry =
+        aws_mem_calloc(allocator, 1, sizeof(struct aws_mqtt5_manual_puback_entry));
+
+    manual_puback_entry->allocator = allocator;
+    aws_ref_count_init(&manual_puback_entry->ref_count, manual_puback_entry, s_aws_mqtt5_manual_puback_entry_destroy);
+    manual_puback_entry->packet_id = packet_id;
+    manual_puback_entry->puback_control_id = puback_control_id;
+
+    return manual_puback_entry;
+}
+
 uint64_t aws_mqtt5_client_acquire_puback(
     struct aws_mqtt5_client *client,
     const struct aws_mqtt5_packet_publish_view *publish_view) {
@@ -2629,6 +2654,10 @@ uint64_t aws_mqtt5_client_acquire_puback(
         /* In this case we simply provide the same control_id that was already sent before. We do not want to create a
          * second control_id with the same packet_id. It is the user's responsibility to know that they have two PUBACKs
          * for the same PUBLISH. */
+        AWS_LOGF_WARN(
+            AWS_LS_MQTT5_CLIENT,
+            "id=%p: PUBACK acquire called on a PUBLISH that is already under user control.",
+            (void *)client);
         struct aws_mqtt5_manual_puback_entry *entry = elem->value;
         return entry->puback_control_id;
     }
@@ -2674,6 +2703,25 @@ uint64_t aws_mqtt5_client_acquire_puback(
 
     /* Increment next_mqtt5_puback_control_id for next use */
     client->operational_state.next_mqtt5_puback_control_id = current_control_packet_id + 1;
+
+    size_t in_flight_unacked_publishes =
+        aws_hash_table_get_entry_count(&client->operational_state.manual_puback_control_id_table);
+    if (in_flight_unacked_publishes >= 100) {
+        AWS_LOGF_WARN(
+            AWS_LS_MQTT5_CLIENT,
+            "id=%p: 100 or more PUBACKs under user control: %zu. AWS IoT Core limits of 100 in-flight has been met or "
+            "exceeded.",
+            (void *)client,
+            in_flight_unacked_publishes);
+    } else {
+        AWS_LOGF_DEBUG(
+            AWS_LS_MQTT5_CLIENT,
+            "id=%p: Manual PUBACK control taken for a PUBLISH packet. Current in-flight PUBACKs under user control: "
+            "%zu",
+            (void *)client,
+            in_flight_unacked_publishes);
+    }
+
     return manual_puback->puback_control_id;
 
 cleanup:
@@ -2781,7 +2829,7 @@ int aws_mqtt5_client_operational_state_init(
             aws_mqtt_hash_uint16_t,
             aws_mqtt_compare_uint16_t_eq,
             NULL,
-            aws_mqtt5_manual_puback_entry_decref)) {
+            s_aws_mqtt5_manual_puback_entry_decref)) {
         return AWS_OP_ERR;
     }
 
@@ -2792,7 +2840,7 @@ int aws_mqtt5_client_operational_state_init(
             aws_mqtt_hash_uint64_t,
             aws_mqtt_compare_uint64_t_eq,
             NULL,
-            aws_mqtt5_manual_puback_entry_decref)) {
+            s_aws_mqtt5_manual_puback_entry_decref)) {
         return AWS_OP_ERR;
     }
 
@@ -2803,7 +2851,7 @@ int aws_mqtt5_client_operational_state_init(
             aws_mqtt_hash_uint64_t,
             aws_mqtt_compare_uint64_t_eq,
             NULL,
-            aws_mqtt5_manual_puback_entry_decref)) {
+            s_aws_mqtt5_manual_puback_entry_decref)) {
         return AWS_OP_ERR;
     }
 

source/v5/mqtt5_options_storage.c

@@ -2370,30 +2370,6 @@ struct aws_mqtt5_operation_puback *aws_mqtt5_operation_puback_new(
     return NULL;
 }
 
-static void s_aws_mqtt5_manual_puback_entry_destroy(void *object) {
-    if (object == NULL) {
-        return;
-    }
-    struct aws_mqtt5_manual_puback_entry *manual_puback_entry = object;
-    aws_mem_release(manual_puback_entry->allocator, manual_puback_entry);
-}
-
-struct aws_mqtt5_manual_puback_entry *s_aws_mqtt_manual_puback_entry_new(
-    struct aws_allocator *allocator,
-    uint16_t packet_id,
-    uint64_t puback_control_id) {
-
-    struct aws_mqtt5_manual_puback_entry *manual_puback_entry =
-        aws_mem_calloc(allocator, 1, sizeof(struct aws_mqtt5_manual_puback_entry));
-
-    manual_puback_entry->allocator = allocator;
-    aws_ref_count_init(&manual_puback_entry->ref_count, manual_puback_entry, s_aws_mqtt5_manual_puback_entry_destroy);
-    manual_puback_entry->packet_id = packet_id;
-    manual_puback_entry->puback_control_id = puback_control_id;
-
-    return manual_puback_entry;
-}
-
 /*********************************************************************************************************************
  * Unsubscribe
  ********************************************************************************************************************/