CAMEL-22786 - Camel-AWS: Extract common logic for clients instantiation in a separated module

components/camel-aws/camel-aws-common/pom.xml

@@ -67,6 +67,11 @@
             <artifactId>aws-core</artifactId>
             <version>${aws-java-sdk2-version}</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>netty-nio-client</artifactId>
+            <version>${aws-java-sdk2-version}</version>
+        </dependency>
     </dependencies>
 
 </project>

components/camel-aws/camel-aws-common/src/main/java/org/apache/camel/component/aws/common/AwsClientBuilderUtil.java

@@ -31,11 +31,14 @@
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.awscore.client.builder.AwsClientBuilder;
 import software.amazon.awssdk.core.SdkClient;
+import software.amazon.awssdk.core.client.builder.SdkAsyncClientBuilder;
 import software.amazon.awssdk.core.client.builder.SdkSyncClientBuilder;
 import software.amazon.awssdk.http.SdkHttpClient;
 import software.amazon.awssdk.http.SdkHttpConfigurationOption;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
 import software.amazon.awssdk.http.apache.ProxyConfiguration;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.utils.AttributeMap;
 
@@ -143,6 +146,98 @@ public static <B extends AwsClientBuilder<B, C> & SdkSyncClientBuilder<B, C>, C
         return buildClient(config, builderSupplier, null);
     }
 
+    /**
+     * Build an AWS async client with the given configuration.
+     *
+     * @param  config                The common AWS configuration
+     * @param  builderSupplier       Supplier for the service-specific async client builder (e.g.,
+     *                               KinesisAsyncClient::builder)
+     * @param  serviceSpecificConfig Optional consumer for service-specific configuration
+     * @param  <B>                   The builder type (must extend both AwsClientBuilder and SdkAsyncClientBuilder)
+     * @param  <C>                   The client type
+     * @return                       The configured AWS async client
+     */
+    @SuppressWarnings("unchecked")
+    public static <B extends AwsClientBuilder<B, C> & SdkAsyncClientBuilder<B, C>, C extends SdkClient> C buildAsyncClient(
+            AwsCommonConfiguration config,
+            Supplier<B> builderSupplier,
+            Consumer<B> serviceSpecificConfig) {
+
+        B clientBuilder = builderSupplier.get();
+        NettyNioAsyncHttpClient.Builder httpClientBuilder = null;
+        boolean httpClientConfigured = false;
+
+        // 1. Configure proxy
+        if (ObjectHelper.isNotEmpty(config.getProxyHost())
+                && ObjectHelper.isNotEmpty(config.getProxyPort())) {
+            LOG.trace("Configuring async proxy: {}:{}", config.getProxyHost(), config.getProxyPort());
+            software.amazon.awssdk.http.nio.netty.ProxyConfiguration proxyConfig
+                    = software.amazon.awssdk.http.nio.netty.ProxyConfiguration.builder()
+                            .scheme(config.getProxyProtocol().toString())
+                            .host(config.getProxyHost())
+                            .port(config.getProxyPort())
+                            .build();
+            httpClientBuilder = NettyNioAsyncHttpClient.builder().proxyConfiguration(proxyConfig);
+            httpClientConfigured = true;
+        }
+
+        // 2. Configure credentials
+        AwsCredentialsProvider credentialsProvider = resolveCredentialsProvider(config);
+        if (credentialsProvider != null) {
+            clientBuilder.credentialsProvider(credentialsProvider);
+        }
+
+        // 3. Apply HTTP client builder if configured (before trust all certs check)
+        if (httpClientConfigured) {
+            clientBuilder.httpClientBuilder(httpClientBuilder);
+        }
+
+        // 4. Configure region
+        if (ObjectHelper.isNotEmpty(config.getRegion())) {
+            clientBuilder.region(Region.of(config.getRegion()));
+        }
+
+        // 5. Configure endpoint override
+        if (config.isOverrideEndpoint() && ObjectHelper.isNotEmpty(config.getUriEndpointOverride())) {
+            clientBuilder.endpointOverride(URI.create(config.getUriEndpointOverride()));
+        }
+
+        // 6. Configure trust all certificates
+        if (config.isTrustAllCertificates()) {
+            if (httpClientBuilder == null) {
+                httpClientBuilder = NettyNioAsyncHttpClient.builder();
+            }
+            SdkAsyncHttpClient asyncHttpClient = httpClientBuilder.buildWithDefaults(
+                    AttributeMap.builder()
+                            .put(SdkHttpConfigurationOption.TRUST_ALL_CERTIFICATES, Boolean.TRUE)
+                            .build());
+            clientBuilder.httpClient(asyncHttpClient);
+            clientBuilder.httpClientBuilder(null);
+        }
+
+        // 7. Apply service-specific configuration
+        if (serviceSpecificConfig != null) {
+            serviceSpecificConfig.accept(clientBuilder);
+        }
+
+        return clientBuilder.build();
+    }
+
+    /**
+     * Build an AWS async client with the given configuration, without service-specific configuration.
+     *
+     * @param  config          The common AWS configuration
+     * @param  builderSupplier Supplier for the service-specific async client builder
+     * @param  <B>             The builder type
+     * @param  <C>             The client type
+     * @return                 The configured AWS async client
+     */
+    public static <B extends AwsClientBuilder<B, C> & SdkAsyncClientBuilder<B, C>, C extends SdkClient> C buildAsyncClient(
+            AwsCommonConfiguration config,
+            Supplier<B> builderSupplier) {
+        return buildAsyncClient(config, builderSupplier, null);
+    }
+
     /**
      * Resolve the appropriate credentials provider based on configuration.
      * <p>

components/camel-aws/camel-aws2-ddb/pom.xml

@@ -36,6 +36,10 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-aws-common</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.camel</groupId>
             <artifactId>camel-support</artifactId>

components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddb/Ddb2Configuration.java

@@ -17,6 +17,7 @@
 package org.apache.camel.component.aws2.ddb;
 
 import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.aws.common.AwsCommonConfiguration;
 import org.apache.camel.spi.Metadata;
 import org.apache.camel.spi.UriParam;
 import org.apache.camel.spi.UriParams;
@@ -25,7 +26,7 @@
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 
 @UriParams
-public class Ddb2Configuration implements Cloneable {
+public class Ddb2Configuration implements Cloneable, AwsCommonConfiguration {
 
     @UriPath
     @Metadata(required = true)
@@ -297,7 +298,7 @@ public void setUseDefaultCredentialsProvider(Boolean useDefaultCredentialsProvid
         this.useDefaultCredentialsProvider = useDefaultCredentialsProvider;
     }
 
-    public Boolean isUseDefaultCredentialsProvider() {
+    public boolean isUseDefaultCredentialsProvider() {
         return useDefaultCredentialsProvider;
     }
 

components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddb/Ddb2Endpoint.java

@@ -88,7 +88,7 @@ public void doStart() throws Exception {
         super.doStart();
 
         ddbClient = configuration.getAmazonDDBClient() != null
-                ? configuration.getAmazonDDBClient() : Ddb2ClientFactory.getDynamoDBClient(configuration).getDynamoDBClient();
+                ? configuration.getAmazonDDBClient() : Ddb2ClientFactory.getDynamoDBClient(configuration);
 
         String tableName = getConfiguration().getTableName();
         LOG.trace("Querying whether table [{}] already exists...", tableName);

components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddb/client/Ddb2ClientFactory.java

@@ -16,35 +16,27 @@
  */
 package org.apache.camel.component.aws2.ddb.client;
 
+import org.apache.camel.component.aws.common.AwsClientBuilderUtil;
 import org.apache.camel.component.aws2.ddb.Ddb2Configuration;
-import org.apache.camel.component.aws2.ddb.client.impl.Ddb2ClientIAMOptimizedImpl;
-import org.apache.camel.component.aws2.ddb.client.impl.Ddb2ClientIAMProfileOptimizedImpl;
-import org.apache.camel.component.aws2.ddb.client.impl.Ddb2ClientSessionTokenImpl;
-import org.apache.camel.component.aws2.ddb.client.impl.Ddb2ClientStandardImpl;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 
 /**
- * Factory class to return the correct type of AWS DynamoDB client.
+ * Factory class to create AWS DynamoDB clients using common configuration.
  */
 public final class Ddb2ClientFactory {
 
     private Ddb2ClientFactory() {
     }
 
     /**
-     * Return the correct AWS DynamoDB client (based on remote vs local).
+     * Create a DynamoDB client based on configuration.
      *
-     * @param  configuration configuration
-     * @return               DynamoDBClient
+     * @param  configuration The DynamoDB configuration
+     * @return               Configured DynamoDbClient
      */
-    public static Ddb2InternalClient getDynamoDBClient(Ddb2Configuration configuration) {
-        if (Boolean.TRUE.equals(configuration.isUseDefaultCredentialsProvider())) {
-            return new Ddb2ClientIAMOptimizedImpl(configuration);
-        } else if (Boolean.TRUE.equals(configuration.isUseProfileCredentialsProvider())) {
-            return new Ddb2ClientIAMProfileOptimizedImpl(configuration);
-        } else if (Boolean.TRUE.equals(configuration.isUseSessionCredentials())) {
-            return new Ddb2ClientSessionTokenImpl(configuration);
-        } else {
-            return new Ddb2ClientStandardImpl(configuration);
-        }
+    public static DynamoDbClient getDynamoDBClient(Ddb2Configuration configuration) {
+        return AwsClientBuilderUtil.buildClient(
+                configuration,
+                DynamoDbClient::builder);
     }
 }