Merge branch 'main' into more_algos

Author: DmitriyMusatkin

src/main/java/software/amazon/awssdk/crt/internal/IoTDeviceSDKMetrics.java

@@ -0,0 +1,21 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+package software.amazon.awssdk.crt.internal;
+
+/**
+ * @internal
+ * IoT Device SDK Metrics Structure. Not for external usage. 
+ */
+public class IoTDeviceSDKMetrics {
+    private String libraryName;
+
+    public IoTDeviceSDKMetrics() {
+        this.libraryName = "IoTDeviceSDK/Java";
+    }
+
+    public String getLibraryName() {
+        return libraryName;
+    }
+}

src/main/java/software/amazon/awssdk/crt/mqtt/MqttClientConnection.java

@@ -17,6 +17,7 @@
 import software.amazon.awssdk.crt.mqtt5.Mqtt5Client;
 import software.amazon.awssdk.crt.mqtt5.Mqtt5ClientOptions;
 import software.amazon.awssdk.crt.mqtt5.packets.ConnectPacket;
+import software.amazon.awssdk.crt.internal.IoTDeviceSDKMetrics;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
@@ -77,6 +78,7 @@ private static MqttConnectionConfig s_toMqtt3ConnectionConfig(Mqtt5ClientOptions
         options.setProtocolOperationTimeoutMs(mqtt5options.getAckTimeoutSeconds() != null
                 ? Math.toIntExact(mqtt5options.getAckTimeoutSeconds()) * 1000
                 : 0);
+        options.setMetricsEnabled(mqtt5options.getMetricsEnabled());
         return options;
     }
 
@@ -162,6 +164,10 @@ private void SetupConfig(MqttConnectionConfig config) throws MqttException {
                 mqttClientConnectionSetLogin(getNativeHandle(), config.getUsername(), config.getPassword());
             }
 
+            if (config.getMetricsEnabled()) {
+                mqttClientConnectionSetMetrics(getNativeHandle(), new IoTDeviceSDKMetrics());
+            }
+
             if (config.getMinReconnectTimeoutSecs() != 0L && config.getMaxReconnectTimeoutSecs() != 0L) {
                 mqttClientConnectionSetReconnectTimeout(getNativeHandle(), config.getMinReconnectTimeoutSecs(),
                         config.getMaxReconnectTimeoutSecs());
@@ -502,6 +508,9 @@ private static native boolean mqttClientConnectionSetWill(long connection, Strin
     private static native void mqttClientConnectionSetLogin(long connection, String username, String password)
             throws CrtRuntimeException;
 
+    private static native void mqttClientConnectionSetMetrics(long connection, IoTDeviceSDKMetrics metrics)
+            throws CrtRuntimeException;
+
     private static native void mqttClientConnectionSetReconnectTimeout(long connection, long minTimeout,
             long maxTimeout)
             throws CrtRuntimeException;

src/main/java/software/amazon/awssdk/crt/mqtt/MqttConnectionConfig.java

@@ -46,6 +46,9 @@ public final class MqttConnectionConfig extends CrtResource {
     private HttpProxyOptions proxyOptions;
     private Consumer<WebsocketHandshakeTransformArgs> websocketHandshakeTransform;
 
+    /* metrics */
+    private boolean metricsEnabled = true;
+
     public MqttConnectionConfig() {}
 
 
@@ -538,6 +541,24 @@ public Consumer<WebsocketHandshakeTransformArgs> getWebsocketHandshakeTransform(
         return websocketHandshakeTransform;
     }
 
+    /**
+     * Enables or disables IoT Device SDK metrics collection. The metrics includes SDK name, version, and platform.
+     *
+     * @param enabled true to enable metrics, false to disable
+     */
+    public void setMetricsEnabled(boolean enabled) {
+        this.metricsEnabled = enabled;
+    }
+
+    /**
+     * Queries whether IoT Device SDK metrics collection is enabled
+     *
+     * @return true if metrics are enabled, false if disabled
+     */
+    public boolean getMetricsEnabled() {
+        return metricsEnabled;
+    }
+
     /**
      * Creates a (shallow) clone of this config object
      *
@@ -567,6 +588,7 @@ public MqttConnectionConfig clone() {
             clone.setWebsocketHandshakeTransform(getWebsocketHandshakeTransform());
 
             clone.setReconnectTimeoutSecs(getMinReconnectTimeoutSecs(), getMaxReconnectTimeoutSecs());
+            clone.setMetricsEnabled(getMetricsEnabled());
 
             // success, bump up the ref count so we can escape the try-with-resources block
             clone.addRef();

src/main/java/software/amazon/awssdk/crt/mqtt5/Mqtt5Client.java

@@ -202,6 +202,22 @@ public Mqtt5ClientOperationStatistics getOperationStatistics() {
         return mqtt5ClientInternalGetOperationStatistics(getNativeHandle());
     }
 
+    /**
+     * Sends a publish acknowledgement packet for a QoS 1 PUBLISH that was previously acquired
+     * for manual control.
+     *
+     * <p>To use manual publish acknowledgement control, call
+     * {@link PublishReturn#acquirePublishAcknowledgementControl()} within the
+     * {@link Mqtt5ClientOptions.PublishEvents#onMessageReceived} callback of a QoS 1 PUBLISH to obtain a
+     * {@link Mqtt5PublishAcknowledgementControlHandle}. Then call this method to send the publish acknowledgement.</p>
+     *
+     * @param publishAcknowledgementControlHandle An opaque handle obtained from
+     *        {@link PublishReturn#acquirePublishAcknowledgementControl()}.
+     */
+    public void invokePublishAcknowledgement(Mqtt5PublishAcknowledgementControlHandle publishAcknowledgementControlHandle) {
+        mqtt5ClientInternalInvokePublishAcknowledgement(getNativeHandle(), publishAcknowledgementControlHandle.getControlId());
+    }
+
     /**
      * Returns the connectivity state for the Mqtt5Client.
      * @return True if the client is connected, false otherwise
@@ -277,4 +293,5 @@ private static native long mqtt5ClientNew(
     private static native void mqtt5ClientInternalUnsubscribe(long client, UnsubscribePacket unsubscribe_options, CompletableFuture<UnsubAckPacket> unsubscribe_suback);
     private static native void mqtt5ClientInternalWebsocketHandshakeComplete(long connection, byte[] marshalledRequest, Throwable throwable, long nativeUserData) throws CrtRuntimeException;
     private static native Mqtt5ClientOperationStatistics mqtt5ClientInternalGetOperationStatistics(long client);
+    private static native void mqtt5ClientInternalInvokePublishAcknowledgement(long client, long controlId);
 }

src/main/java/software/amazon/awssdk/crt/mqtt5/Mqtt5ClientOptions.java

@@ -12,6 +12,7 @@
 
 import software.amazon.awssdk.crt.mqtt5.packets.ConnectPacket;
 import software.amazon.awssdk.crt.mqtt.MqttConnectionConfig;
+import software.amazon.awssdk.crt.internal.IoTDeviceSDKMetrics;
 
 import java.util.Map;
 import java.util.function.Function;
@@ -45,6 +46,12 @@ public class Mqtt5ClientOptions {
     private Consumer<Mqtt5WebsocketHandshakeTransformArgs> websocketHandshakeTransform;
     private PublishEvents publishEvents;
     private TopicAliasingOptions topicAliasingOptions;
+    // Indicates whether AWS IoT Metrics are enabled for this client, default to true.
+    // We don't expose iotDeviceSDKMetrics in the builder, and only allow setting
+    // metricsEnabled for now.
+    private boolean metricsEnabled = true;
+    private IoTDeviceSDKMetrics iotDeviceSDKMetrics;
+    
 
     /**
      * Returns the host name of the MQTT server to connect to.
@@ -263,6 +270,24 @@ public TopicAliasingOptions getTopicAliasingOptions() {
         return this.topicAliasingOptions;
     }
 
+    /**
+     * Returns whether AWS IoT Device SDK metrics collection is enabled
+     *
+     * @return true if metrics are enabled, false otherwise
+     */
+    public boolean getMetricsEnabled() {
+        return this.metricsEnabled;
+    }
+
+    /**
+     * Enables or disables IoT Device SDK metrics collection. The metrics includes SDK name, version, and platform.
+     *
+     * @param enabled true to enable metrics, false to disable
+     */
+    public void setMetricsEnabled(boolean enabled) {
+        this.metricsEnabled = enabled;
+    }
+
     /**
      * Creates a Mqtt5ClientOptionsBuilder instance
      * @param builder The builder to get the Mqtt5ClientOptions values from
@@ -289,6 +314,8 @@ public Mqtt5ClientOptions(Mqtt5ClientOptionsBuilder builder) {
         this.websocketHandshakeTransform = builder.websocketHandshakeTransform;
         this.publishEvents = builder.publishEvents;
         this.topicAliasingOptions = builder.topicAliasingOptions;
+        this.metricsEnabled = builder.metricsEnabled;
+        this.iotDeviceSDKMetrics = new IoTDeviceSDKMetrics();
     }
 
     /*******************************************************************************
@@ -345,7 +372,15 @@ public interface LifecycleEvents {
      */
     public interface PublishEvents {
         /**
-         * Called when an MQTT PUBLISH packet is received by the client
+         * Called when an MQTT PUBLISH packet is received by the client.
+         *
+         * <p>To take manual control of the publish acknowledgement for a QoS 1 message, call
+         * {@link PublishReturn#acquirePublishAcknowledgementControl()} within this callback. If you do so,
+         * the client will NOT automatically send the publish acknowledgement. You are responsible for calling
+         * {@link Mqtt5Client#invokePublishAcknowledgement(Mqtt5PublishAcknowledgementControlHandle)} later.</p>
+         *
+         * <p>If you do not call {@code acquirePublishAcknowledgementControl()} within the callback , 
+         * the client will automatically send the publish acknowledgement after this callback returns.</p>
          *
          * @param client The client that has received the message
          * @param publishReturn All of the data that was received from the server
@@ -583,6 +618,7 @@ static final public class Mqtt5ClientOptionsBuilder {
         private Consumer<Mqtt5WebsocketHandshakeTransformArgs> websocketHandshakeTransform;
         private PublishEvents publishEvents;
         private TopicAliasingOptions topicAliasingOptions;
+        private boolean metricsEnabled = true;
 
         /**
          * Sets the host name of the MQTT server to connect to.
@@ -850,6 +886,17 @@ public Mqtt5ClientOptionsBuilder withTopicAliasingOptions(TopicAliasingOptions o
             return this;
         }
 
+        /**
+         * Enables or disables IoT Device SDK metrics collection. The metrics includes SDK name, version, and platform.
+         *
+         * @param enabled true to enable metrics, false to disable.
+         * @return The Mqtt5ClientOptionsBuilder after setting the metrics option
+         */
+        public Mqtt5ClientOptionsBuilder withMetricsEnabled(boolean enabled) {
+            this.metricsEnabled = enabled;
+            return this;
+        }
+
         /**
          * Creates a new Mqtt5ClientOptionsBuilder instance
          *

src/main/java/software/amazon/awssdk/crt/mqtt5/Mqtt5PublishAcknowledgementControlHandle.java

@@ -0,0 +1,37 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+package software.amazon.awssdk.crt.mqtt5;
+
+/**
+ * An opaque handle representing manual control over a publish acknowledgement for a received
+ * PUBLISH packet.
+ *
+ * <p>This class cannot be instantiated directly. Instances are only created by the CRT library.</p>
+ */
+public class Mqtt5PublishAcknowledgementControlHandle {
+
+    private final long controlId;
+
+    /**
+     * Creates a new Mqtt5PublishAcknowledgementControlHandle.
+     *
+     * @param controlId The native publish acknowledgement control ID returned by
+     *                  aws_mqtt5_client_acquire_publish_acknowledgement.
+     */
+    Mqtt5PublishAcknowledgementControlHandle(long controlId) {
+        this.controlId = controlId;
+    }
+
+    /**
+     * Returns the native publish acknowledgement control ID, used by
+     * {@link Mqtt5Client#invokePublishAcknowledgement(Mqtt5PublishAcknowledgementControlHandle)}
+     * to pass the control ID to native code.
+     *
+     * @return The native publish acknowledgement control ID.
+     */
+    long getControlId() {
+        return controlId;
+    }
+}

src/main/java/software/amazon/awssdk/crt/mqtt5/PublishReturn.java

@@ -14,6 +14,22 @@
 public class PublishReturn {
     private PublishPacket publishPacket;
 
+    /**
+     * The already-acquired publish acknowledgement control ID, eagerly acquired by native code
+     * before the {@link Mqtt5ClientOptions.PublishEvents#onMessageReceived} callback is invoked.
+     * For QoS 0 messages this is 0 (no publish acknowledgement required).
+     * After {@code acquirePublishAcknowledgementControl()} is called, this is set to 0 to prevent
+     * double-use.
+     */
+    private long controlId;
+
+    /**
+     * The threadID of the {@link Mqtt5ClientOptions.PublishEvents#onMessageReceived} callback.
+     * This is set at the beginning of the callback and then used to verify 
+     * {@code acquirePublishAcknowledgementControl()} is being called from within.
+     */
+    private long threadID;
+
     /**
      * Returns the PublishPacket returned from the server or Null if none was returned.
      * @return The PublishPacket returned from the server.
@@ -22,12 +38,50 @@ public PublishPacket getPublishPacket() {
         return publishPacket;
     }
 
+    /**
+     * Acquires manual control over the publish acknowledgement for this PUBLISH message,
+     * preventing the client from automatically sending an acknowledgement. The returned handle can be
+     * passed to {@link Mqtt5Client#invokePublishAcknowledgement(Mqtt5PublishAcknowledgementControlHandle)}
+     * at a later time to send the publish acknowledgement to the broker.
+     *
+     * <p><b>Important:</b> This method must be called within the
+     * {@link Mqtt5ClientOptions.PublishEvents#onMessageReceived} callback. Calling it outside the
+     * callback (wrong thread) or after it has already been called will return {@code null}.</p>
+     *
+     * <p>This method may only be called once per received PUBLISH. Subsequent calls will return
+     * {@code null}.</p>
+     *
+     * <p>If this method is not called, the client will automatically send a publish acknowledgment
+     * for QoS 1 messages when the callback returns.</p>
+     *
+     * @return A {@link Mqtt5PublishAcknowledgementControlHandle} that can be used to manually send
+     *         the acknowledgement, or {@code null} if called outside the callback, called more than
+     *         once, or called on a QoS 0 message.
+     */
+    public synchronized Mqtt5PublishAcknowledgementControlHandle acquirePublishAcknowledgementControl() {
+        if (controlId == 0 || threadID != Thread.currentThread().getId()) {
+            return null;
+        }
+        long acquiredControlId = controlId;
+        /* Zero out so it can't be double-called */
+        controlId = 0;
+        return new Mqtt5PublishAcknowledgementControlHandle(acquiredControlId);
+    }
+
     /**
      * This is only called in JNI to make a new PublishReturn with a PUBLISH packet.
-     * @param newPublishPacket The PubAckPacket data for QoS 1 packets. Can be null if result is non QoS 1.
-     * @return A newly created PublishResult
+     * The controlId is eagerly acquired by native code prior to
+     * {@link Mqtt5ClientOptions.PublishEvents#onMessageReceived} being called.
+     * The threadID is set to the calling thread and is used when
+     * {@link #acquirePublishAcknowledgementControl()} is called to guarantee the result
+     * is accurate and enforces the requirement of calling it from within the callback.
+     *
+     * @param newPublishPacket The PublishPacket data received from the server.
+     * @param controlId The pre-acquired publish acknowledgement control ID (0 for QoS 0 messages).
      */
-    private PublishReturn(PublishPacket newPublishPacket) {
+    private PublishReturn(PublishPacket newPublishPacket, long controlId) {
         this.publishPacket = newPublishPacket;
+        this.controlId = controlId;
+        this.threadID = Thread.currentThread().getId();
     }
 }

src/main/resources/META-INF/native-image/software.amazon.awssdk/crt/aws-crt/jni-config.json

@@ -1256,6 +1256,14 @@
       }
     ]
   },
+  {
+    "name": "software.amazon.awssdk.crt.internal.IoTDeviceSDKMetrics",
+    "fields": [
+      {
+        "name": "libraryName"
+      }
+    ]
+  },
   {
     "name": "software.amazon.awssdk.crt.mqtt5.Mqtt5ClientOptions",
     "fields": [
@@ -1306,6 +1314,12 @@
       },
       {
         "name": "topicAliasingOptions"
+      },
+      {
+        "name": "iotDeviceSDKMetrics"
+      },
+      {
+        "name": "metricsEnabled"
       }
     ],
     "methods": [
@@ -1549,8 +1563,13 @@
       {
         "name": "<init>",
         "parameterTypes": [
-          "software.amazon.awssdk.crt.mqtt5.packets.PublishPacket"
+          "software.amazon.awssdk.crt.mqtt5.packets.PublishPacket",
+          "long"
         ]
+      },
+      {
+        "name": "acquirePublishAcknowledgementControl",
+        "parameterTypes": []
       }
     ]
   },