Revert "Merge branch 'main' of github.com:awslabs/aws-c-mqtt into get_eventloop_group"

This reverts commit e28e58b, reversing
changes made to 9fc2f57.

Author: xiazhvera

include/aws/mqtt/private/request-response/protocol_adapter.h

@@ -15,7 +15,7 @@ struct aws_allocator;
 struct aws_event_loop;
 struct aws_mqtt_client_connection;
 struct aws_mqtt5_client;
-struct aws_mqtt_request_response_publish_event;
+struct aws_mqtt_rr_incoming_publish_event;
 
 /*
  * The request-response protocol adapter is a translation layer that sits between the request-response native client
@@ -99,7 +99,7 @@ typedef void(aws_protocol_adapter_subscription_event_fn)(
     void *user_data);
 
 typedef void(aws_protocol_adapter_incoming_publish_fn)(
-    const struct aws_mqtt_request_response_publish_event *publish,
+    const struct aws_mqtt_rr_incoming_publish_event *publish,
     void *user_data);
 
 typedef void(aws_protocol_adapter_terminate_callback_fn)(void *user_data);

include/aws/mqtt/request-response/request_response_client.h

@@ -31,22 +31,6 @@ struct aws_mqtt_request_operation_response_path {
     struct aws_byte_cursor correlation_token_json_path;
 };
 
-/*
- * An event emitted by a streaming operation's subscription.
- */
-struct aws_mqtt_request_response_publish_event {
-    struct aws_byte_cursor payload;
-    struct aws_byte_cursor topic;
-    /* Below are MQTT optional fields. For MQTT3, they will always be empty, as MQTT3 does not support them. For MQTT5,
-     * they will be set if they are present in a packet. */
-    const struct aws_byte_cursor *content_type;
-    size_t user_property_count;
-    const struct aws_mqtt5_user_property *user_properties;
-    /* Even though this field is supposed to be used by MQTT broker to determine if a message-to-be-sent is expired,
-     * certain services use this field to specify client-side timeouts. */
-    const uint32_t *message_expiry_interval_seconds;
-};
-
 /*
  * Callback signature for request-response completion.
  *

source/request-response/protocol_adapter.c

@@ -347,7 +347,7 @@ static void s_protocol_adapter_mqtt311_listener_publish_received(
 
     struct aws_mqtt_protocol_adapter_311_impl *adapter = userdata;
 
-    struct aws_mqtt_request_response_publish_event publish_event = {
+    struct aws_mqtt_rr_incoming_publish_event publish_event = {
         .topic = *topic,
         .payload = *payload,
     };
@@ -803,7 +803,7 @@ static bool s_protocol_adapter_mqtt5_listener_publish_received(
     void *user_data) {
     struct aws_mqtt_protocol_adapter_5_impl *adapter = user_data;
 
-    struct aws_mqtt_request_response_publish_event publish_event = {
+    struct aws_mqtt_rr_incoming_publish_event publish_event = {
         .topic = publish->topic,
         .payload = publish->payload,
         .content_type = publish->content_type,

source/request-response/request_response_client.c

@@ -873,9 +873,22 @@ static void s_aws_rr_client_protocol_adapter_subscription_event_callback(
 }
 
 static void s_apply_publish_to_streaming_operation_list(
-    struct aws_rr_operation_list_topic_filter_entry *entry,
-    const struct aws_mqtt_request_response_publish_event *publish_event) {
-    AWS_FATAL_ASSERT(entry != NULL);
+    const struct aws_linked_list *operations,
+    const struct aws_byte_cursor *topic_filter,
+    const struct aws_mqtt_rr_incoming_publish_event *publish_event,
+    void *user_data) {
+
+    AWS_FATAL_ASSERT(operations != NULL);
+
+    struct aws_mqtt_request_response_client *rr_client = user_data;
+
+    AWS_LOGF_DEBUG(
+        AWS_LS_MQTT_REQUEST_RESPONSE,
+        "id=%p: request-response client incoming publish on topic '" PRInSTR
+        "' matches streaming subscription on topic filter '" PRInSTR "'",
+        (void *)rr_client,
+        AWS_BYTE_CURSOR_PRI(publish_event->topic),
+        AWS_BYTE_CURSOR_PRI(*topic_filter));
 
     struct aws_linked_list_node *node = aws_linked_list_begin(&entry->operations);
     while (node != aws_linked_list_end(&entry->operations)) {
@@ -897,8 +910,8 @@ static void s_apply_publish_to_streaming_operation_list(
             continue;
         }
 
-        void *user_data = operation->storage.streaming_storage.options.user_data;
-        (*incoming_publish_callback)(publish_event, user_data);
+        void *operation_user_data = operation->storage.streaming_storage.options.user_data;
+        (*incoming_publish_callback)(publish_event, operation_user_data);
 
         AWS_LOGF_DEBUG(
             AWS_LS_MQTT_REQUEST_RESPONSE,
@@ -913,7 +926,7 @@ static void s_apply_publish_to_streaming_operation_list(
 static void s_complete_operation_with_correlation_token(
     struct aws_mqtt_request_response_client *rr_client,
     struct aws_byte_cursor correlation_token,
-    const struct aws_mqtt_request_response_publish_event *publish_event) {
+    const struct aws_mqtt_rr_incoming_publish_event *publish_event) {
     struct aws_hash_element *hash_element = NULL;
 
     if (aws_hash_table_find(&rr_client->operations_by_correlation_tokens, &correlation_token, &hash_element)) {
@@ -965,7 +978,16 @@ static void s_complete_operation_with_correlation_token(
 static void s_apply_publish_to_response_path_entry(
     struct aws_mqtt_request_response_client *rr_client,
     struct aws_rr_response_path_entry *entry,
-    const struct aws_mqtt_request_response_publish_event *publish_event) {
+    const struct aws_mqtt_rr_incoming_publish_event *publish_event,
+    void *user_data) {
+
+    struct aws_mqtt_request_response_client *rr_client = user_data;
+
+    AWS_LOGF_DEBUG(
+        AWS_LS_MQTT_REQUEST_RESPONSE,
+        "id=%p: request-response client incoming publish on topic '" PRInSTR "' matches response path",
+        (void *)rr_client,
+        AWS_BYTE_CURSOR_PRI(publish_event->topic));
 
     struct aws_json_value *json_payload = NULL;
 
@@ -1044,7 +1066,7 @@ static void s_apply_publish_to_response_path_entry(
 }
 
 static void s_aws_rr_client_protocol_adapter_incoming_publish_callback(
-    const struct aws_mqtt_request_response_publish_event *publish_event,
+    const struct aws_mqtt_rr_incoming_publish_event *publish_event,
     void *user_data) {
 
     struct aws_mqtt_request_response_client *rr_client = user_data;

tests/request-response/protocol_adapter_tests.c

@@ -111,7 +111,7 @@ static void s_rr_mqtt_protocol_adapter_test_on_subscription_event(
 }
 
 static void s_rr_mqtt_protocol_adapter_test_on_incoming_publish(
-    const struct aws_mqtt_request_response_publish_event *publish,
+    const struct aws_mqtt_rr_incoming_publish_event *publish,
     void *user_data) {
     struct aws_request_response_protocol_adapter_test_fixture *fixture = user_data;
 

tests/request-response/request_response_client_tests.c

@@ -310,7 +310,7 @@ static void s_rrc_fixture_streaming_operation_subscription_status_callback(
 }
 
 static void s_rrc_fixture_streaming_operation_incoming_publish_callback(
-    const struct aws_mqtt_request_response_publish_event *publish_event,
+    const struct aws_mqtt_rr_incoming_publish_event *publish_event,
     void *user_data) {
     struct aws_rr_client_fixture_streaming_record *record = user_data;
     struct aws_rr_client_test_fixture *fixture = record->fixture;