Capture fields from PUBLISH for req-resp stream operations (#888)

sfod · web-flow · commit 83d451ebfe7b · 2025-05-13T11:42:00.000-07:00
diff --git a/src/main/java/software/amazon/awssdk/crt/iot/IncomingPublishEvent.java b/src/main/java/software/amazon/awssdk/crt/iot/IncomingPublishEvent.java
@@ -5,12 +5,25 @@
 
 package software.amazon.awssdk.crt.iot;
 
+import software.amazon.awssdk.crt.mqtt5.packets.UserProperty;
+
+import java.util.List;
+
+/**
+ * An event that describes an incoming publish message received on a streaming operation.
+ */
 public class IncomingPublishEvent {
 
     private final byte[] payload;
 
     private final String topic;
 
+    private String contentType;
+
+    private List<UserProperty> userProperties;
+
+    private Long messageExpiryIntervalSeconds;
+
     private IncomingPublishEvent(byte[] payload, String topic) {
         this.payload = payload;
         this.topic = topic;
@@ -33,4 +46,31 @@ public byte[] getPayload() {
     public String getTopic() {
         return topic;
     }
+
+    /**
+     * Gets the content type of the IncomingPublishEvent.
+     *
+     * @return Content type of the IncomingPublishEvent.
+     */
+    public String getContentType() {
+        return contentType;
+    }
+
+    /**
+     * Gets the user properties of the IncomingPublishEvent.
+     *
+     * @return User properties of the IncomingPublishEvent.
+     */
+    public List<UserProperty> getUserProperties() {
+        return userProperties;
+    }
+
+    /**
+     * Gets the message expiry interval seconds of the IncomingPublishEvent.
+     *
+     * @return Message expiry interval seconds of the IncomingPublishEvent.
+     */
+    public Long getMessageExpiryIntervalSeconds() {
+        return messageExpiryIntervalSeconds;
+    }
 }
diff --git a/src/main/resources/META-INF/native-image/software.amazon.awssdk/crt/aws-crt/jni-config.json b/src/main/resources/META-INF/native-image/software.amazon.awssdk/crt/aws-crt/jni-config.json
@@ -1015,6 +1015,17 @@
           "java.lang.String"
         ]
       }
+    ],
+    "fields": [
+      {
+        "name": "contentType"
+      },
+      {
+        "name": "userProperties"
+      },
+      {
+        "name": "messageExpiryIntervalSeconds"
+      }
     ]
   },
   {
diff --git a/src/native/java_class_ids.c b/src/native/java_class_ids.c
@@ -2407,6 +2407,19 @@ static void s_cache_incoming_publish_event_properties(JNIEnv *env) {
 
     incoming_publish_event_properties.constructor_method_id = (*env)->GetMethodID(
         env, incoming_publish_event_properties.incoming_publish_event_class, "<init>", "([BLjava/lang/String;)V");
+
+    incoming_publish_event_properties.content_type_id = (*env)->GetFieldID(
+        env, incoming_publish_event_properties.incoming_publish_event_class, "contentType", "Ljava/lang/String;");
+
+    incoming_publish_event_properties.user_properties_field_id = (*env)->GetFieldID(
+        env, incoming_publish_event_properties.incoming_publish_event_class, "userProperties", "Ljava/util/List;");
+
+    incoming_publish_event_properties.message_expiry_interval_seconds_id = (*env)->GetFieldID(
+        env,
+        incoming_publish_event_properties.incoming_publish_event_class,
+        "messageExpiryIntervalSeconds",
+        "Ljava/lang/Long;");
+
     AWS_FATAL_ASSERT(incoming_publish_event_properties.constructor_method_id);
 }
 
diff --git a/src/native/java_class_ids.h b/src/native/java_class_ids.h
@@ -1006,6 +1006,9 @@ extern struct java_mqtt_request_response_properties mqtt_request_response_proper
 struct java_incoming_publish_event_properties {
     jclass incoming_publish_event_class;
     jmethodID constructor_method_id;
+    jfieldID content_type_id;
+    jfieldID user_properties_field_id;
+    jfieldID message_expiry_interval_seconds_id;
 };
 extern struct java_incoming_publish_event_properties incoming_publish_event_properties;
 
diff --git a/src/native/mqtt_request_response.c b/src/native/mqtt_request_response.c
@@ -10,6 +10,7 @@
 
 #include "java_class_ids.h"
 #include "mqtt5_client_jni.h"
+#include "mqtt5_utils.h"
 #include "mqtt_connection.h"
 
 /* on 32-bit platforms, casting pointers to longs throws a warning we don't need */
@@ -726,6 +727,37 @@ static void s_aws_mqtt_streaming_operation_incoming_publish_callback(
         goto done;
     }
 
+    if (s_set_jni_string_field_in_packet(
+            env,
+            publish_event->content_type,
+            java_incoming_publish_event,
+            incoming_publish_event_properties.content_type_id,
+            "content type",
+            true)) {
+        goto done;
+    }
+
+    if (publish_event->user_properties != NULL && publish_event->user_property_count > 0) {
+        if (s_set_user_properties_field(
+                env,
+                publish_event->user_property_count,
+                publish_event->user_properties,
+                java_incoming_publish_event,
+                incoming_publish_event_properties.user_properties_field_id)) {
+            goto done;
+        }
+    }
+
+    if (s_set_jni_uint32_t_field_in_packet(
+            env,
+            publish_event->message_expiry_interval_seconds,
+            java_incoming_publish_event,
+            incoming_publish_event_properties.message_expiry_interval_seconds_id,
+            "message expiry interval seconds",
+            true)) {
+        goto done;
+    }
+
     (*env)->CallVoidMethod(
         env,
         binding->java_incoming_publish_event_callback,
diff --git a/src/test/java/software/amazon/awssdk/crt/test/MqttRequestResponseClientTests.java b/src/test/java/software/amazon/awssdk/crt/test/MqttRequestResponseClientTests.java
@@ -21,8 +21,10 @@
 import software.amazon.awssdk.crt.mqtt5.OnConnectionSuccessReturn;
 import software.amazon.awssdk.crt.mqtt5.packets.ConnectPacket;
 import software.amazon.awssdk.crt.mqtt5.packets.PublishPacket;
+import software.amazon.awssdk.crt.mqtt5.packets.UserProperty;
 
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -699,14 +701,21 @@ public void ShadowUpdatedStreamingOperationOpenCloseMqtt311() {
         doStreamingOperationOpenClosedTest(MqttVersion.Mqtt311);
     }
 
-    void doStreamingOperationIncomingPublishTest(MqttVersion version) {
+    public interface StreamOperationPublishHelper {
+        String getTopic();
+        PublishPacket createMqtt5PublishPacket();
+        MqttMessage createMqtt3Message();
+        void assertPublishEvent(IncomingPublishEvent event);
+    }
+
+    void doStreamingOperationIncomingPublishTest(MqttVersion version, StreamOperationPublishHelper helper) {
         this.context = new TestContext(version, null);
         CompletableFuture<Boolean> subscribed = new CompletableFuture<>();
         CompletableFuture<IncomingPublishEvent> publishEvent = new CompletableFuture<>();
 
-        String fakeShadowTopic = "not/a/shadow/topic/" + (UUID.randomUUID()).toString();
+        String topic = helper.getTopic();
         StreamingOperationOptions streamingOptions = StreamingOperationOptions.builder()
-                .withTopic(fakeShadowTopic)
+                .withTopic(topic)
                 .withSubscriptionStatusEventCallback((event) -> {
                     if (event.getType() == SubscriptionStatusEventType.SUBSCRIPTION_ESTABLISHED) {
                         subscribed.complete(true);
@@ -723,48 +732,135 @@ void doStreamingOperationIncomingPublishTest(MqttVersion version) {
         try {
             subscribed.get();
         } catch (Exception ex) {
-            Assert.assertTrue(false);
+            Assert.fail(ex.getMessage());
         }
 
-        String originalPayloadAsString = "IncomingPublishTest";
-        byte[] originalPayload = originalPayloadAsString.getBytes(StandardCharsets.UTF_8);
-
         if (version == MqttVersion.Mqtt5) {
-            PublishPacket publishPacket = new PublishPacket.PublishPacketBuilder()
-                    .withPayload(originalPayload)
-                    .withTopic(fakeShadowTopic)
-                    .withQOS(QOS.AT_LEAST_ONCE)
-                    .build();
+            PublishPacket publishPacket = helper.createMqtt5PublishPacket();
             this.context.mqtt5Client.publish(publishPacket);
         } else {
-            this.context.mqtt311Client.publish(new MqttMessage(fakeShadowTopic, originalPayload, QualityOfService.AT_LEAST_ONCE));
+            this.context.mqtt311Client.publish(helper.createMqtt3Message());
         }
 
         try {
             IncomingPublishEvent event = publishEvent.get();
+            helper.assertPublishEvent(event);
+        } catch (Exception | Error ex) {
+            Assert.fail(ex.getMessage());
+        } finally {
+            stream.close();
+        }
+    }
+
+    class StreamOperationPublishHelper_RequiredFieldsOnly implements StreamOperationPublishHelper {
+        private final String payload = "IncomingPublishTest";
+        private final String topic = "not/a/shadow/topic/" + (UUID.randomUUID()).toString();
+
+        @Override
+        public String getTopic() {
+            return topic;
+        }
+
+        @Override
+        public PublishPacket createMqtt5PublishPacket() {
+            byte[] payloadBytes = payload.getBytes(StandardCharsets.UTF_8);
+            return new PublishPacket.PublishPacketBuilder()
+                .withPayload(payloadBytes)
+                .withTopic(topic)
+                .withQOS(QOS.AT_LEAST_ONCE)
+                .build();
+        }
+
+        @Override
+        public MqttMessage createMqtt3Message() {
+            byte[] payloadBytes = payload.getBytes(StandardCharsets.UTF_8);
+            return new MqttMessage(topic, payloadBytes, QualityOfService.AT_LEAST_ONCE);
+        }
+
+        @Override
+        public void assertPublishEvent(IncomingPublishEvent event) {
             Assert.assertNotNull(event);
-            String payloadAsString = new String(event.getPayload(), StandardCharsets.UTF_8);
-            Assert.assertEquals(originalPayloadAsString, payloadAsString);
-            Assert.assertEquals(fakeShadowTopic, event.getTopic());
-        } catch (Exception ex) {
-            Assert.assertTrue(false);
+            String eventPayload = new String(event.getPayload(), StandardCharsets.UTF_8);
+            Assert.assertEquals("payload", payload, eventPayload);
+            Assert.assertEquals("topic", topic, event.getTopic());
+
+            Assert.assertNull("content-type", event.getContentType());
+            Assert.assertNull("user-properties", event.getUserProperties());
+            Assert.assertNull("message-expiry-interval-seconds", event.getMessageExpiryIntervalSeconds());
         }
+    }
 
-        stream.close();
+    class StreamOperationPublishHelper_AllFields implements StreamOperationPublishHelper {
+        private final String payload = "IncomingPublishTest";
+        private final String topic = "not/a/shadow/topic/" + (UUID.randomUUID()).toString();
+        private final String contentType = "application/json";
+        private ArrayList<UserProperty> userProperties;
+
+        @Override
+        public String getTopic() {
+            return topic;
+        }
+
+        @Override
+        public PublishPacket createMqtt5PublishPacket() {
+            byte[] payloadBytes = payload.getBytes(StandardCharsets.UTF_8);
+
+            userProperties = new ArrayList<UserProperty>();
+            userProperties.add(new UserProperty("Hello", "World"));
+
+            return new PublishPacket.PublishPacketBuilder()
+                .withPayload(payloadBytes)
+                .withTopic(topic)
+                .withQOS(QOS.AT_LEAST_ONCE)
+                .withContentType(contentType)
+                .withUserProperties(userProperties)
+                .withMessageExpiryIntervalSeconds(100L)
+                .build();
+        }
+
+        @Override
+        public MqttMessage createMqtt3Message() {
+            byte[] payloadBytes = payload.getBytes(StandardCharsets.UTF_8);
+            return new MqttMessage(topic, payloadBytes, QualityOfService.AT_LEAST_ONCE);
+        }
+
+        @Override
+        public void assertPublishEvent(IncomingPublishEvent event) {
+            Assert.assertNotNull(event);
+            String eventPayload = new String(event.getPayload(), StandardCharsets.UTF_8);
+            Assert.assertEquals("payload", payload, eventPayload);
+            Assert.assertEquals("topic", topic, event.getTopic());
+            Assert.assertEquals("content-type", contentType, event.getContentType());
+            Assert.assertEquals("user-properties size", userProperties.size(), event.getUserProperties().size());
+            for (int i = 0; i < userProperties.size(); i++) {
+                Assert.assertEquals("user-property " + i + " key", userProperties.get(i).key, event.getUserProperties().get(i).key);
+                Assert.assertEquals("user-property " + i + " value", userProperties.get(i).value, event.getUserProperties().get(i).value);
+            }
+            Assert.assertNotNull("message-expiry-interval-seconds should not be null", event.getMessageExpiryIntervalSeconds());
+            // We can't check for the exact value here as it'll be decremented by the server part.
+            Assert.assertTrue("message-expiry-interval-seconds should be a positive number", event.getMessageExpiryIntervalSeconds() > 0);
+        }
+    }
+
+    @Test
+    public void StreamingOperationIncomingPublishMqtt5_RequiredFieldsOnly() {
+        skipIfNetworkUnavailable();
+
+        doStreamingOperationIncomingPublishTest(MqttVersion.Mqtt5, new StreamOperationPublishHelper_RequiredFieldsOnly());
     }
 
     @Test
-    public void StreamingOperationIncomingPublishMqtt5() {
+    public void StreamingOperationIncomingPublishMqtt5_AllFields() {
         skipIfNetworkUnavailable();
 
-        doStreamingOperationIncomingPublishTest(MqttVersion.Mqtt5);
+        doStreamingOperationIncomingPublishTest(MqttVersion.Mqtt5, new StreamOperationPublishHelper_AllFields());
     }
 
     @Test
     public void StreamingOperationIncomingPublishMqtt311() {
         skipIfNetworkUnavailable();
 
-        doStreamingOperationIncomingPublishTest(MqttVersion.Mqtt311);
+        doStreamingOperationIncomingPublishTest(MqttVersion.Mqtt311, new StreamOperationPublishHelper_RequiredFieldsOnly());
     }
 
     void doStreamingOperationReopenFailureTest(MqttVersion version) {

Original file line number	Diff line number	Diff line change
`@@ -1015,6 +1015,17 @@`
`1015`	`1015`	`"java.lang.String"`
`1016`	`1016`	`]`
`1017`	`1017`	`}`
	`1018`	`+ ],`
	`1019`	`+ "fields": [`
	`1020`	`+ {`
	`1021`	`+ "name": "contentType"`
	`1022`	`+ },`
	`1023`	`+ {`
	`1024`	`+ "name": "userProperties"`
	`1025`	`+ },`
	`1026`	`+ {`
	`1027`	`+ "name": "messageExpiryIntervalSeconds"`
	`1028`	`+ }`
`1018`	`1029`	`]`
`1019`	`1030`	`},`
`1020`	`1031`	`{`