Added support for instance profile credentials

nevado-jms/pom.xml

@@ -5,12 +5,12 @@
     <parent>
         <groupId>org.skyscreamer</groupId>
         <artifactId>nevado</artifactId>
-        <version>1.3.3-SNAPSHOT</version>
+        <version>1.3.2</version>
     </parent>
 
     <groupId>org.skyscreamer</groupId>
     <artifactId>nevado-jms</artifactId>
-    <version>1.3.3-SNAPSHOT</version>
+    <version>1.3.2-OA7</version>
     <packaging>jar</packaging>
 
     <name>Nevado JMS</name>
@@ -51,29 +51,59 @@
         <dependency>
             <groupId>org.json</groupId>
             <artifactId>json</artifactId>
-            <version>20090211</version>
-        </dependency>
-        <dependency>
-            <groupId>org.codehaus.jackson</groupId>
-            <artifactId>jackson-core-asl</artifactId>
-            <version>1.9.12</version>
+            <version>20231013</version>
         </dependency>
+<!--        <dependency>-->
+<!--            <groupId>org.codehaus.jackson</groupId>-->
+<!--            <artifactId>jackson-core-asl</artifactId>-->
+<!--            <version>1.9.12</version>-->
+<!--        </dependency>-->
         <dependency>
             <groupId>joda-time</groupId>
             <artifactId>joda-time</artifactId>
             <version>2.2</version>
         </dependency>
 
         <dependency>
-            <groupId>log4j</groupId>
-            <artifactId>log4j</artifactId>
-            <version>1.2.17</version>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-1.2-api</artifactId>
+            <version>2.23.1</version> <!-- or latest stable -->
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-core</artifactId>
+            <version>2.23.1</version>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-sqs</artifactId>
+            <version>1.12.377</version>
+            <exclusions>
+                <!-- Aws SDK pulls in Jackson version 2.12.7, which conflicts a bit with the 2.20. -->
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.dataformat</groupId>
+                    <artifactId>jackson-dataformat-cbor</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>com.amazonaws</groupId>
-            <artifactId>aws-java-sdk</artifactId>
-            <version>1.6.12</version>
+            <artifactId>aws-java-sdk-sns</artifactId>
+            <version>1.12.377</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <version>2.20.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <version>2.20</version>
         </dependency>
 
         <dependency>
@@ -90,16 +120,23 @@
         </dependency>
         <dependency>
             <groupId>org.springframework</groupId>
-            <artifactId>spring-beans</artifactId>
+            <artifactId>spring-context</artifactId>
             <version>${spring.version}</version>
             <scope>test</scope>
         </dependency>
+
         <dependency>
-            <groupId>org.springframework.integration</groupId>
-            <artifactId>spring-integration-jms</artifactId>
-            <version>2.0.0.RELEASE</version>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-beans</artifactId>
+            <version>${spring.version}</version>
             <scope>test</scope>
         </dependency>
+<!--        <dependency>-->
+<!--            <groupId>org.springframework.integration</groupId>-->
+<!--            <artifactId>spring-integration-jms</artifactId>-->
+<!--            <version>2.0.0.RELEASE</version>-->
+<!--            <scope>test</scope>-->
+<!--        </dependency>-->
         <dependency>
             <groupId>activemq</groupId>
             <artifactId>activemq</artifactId>
@@ -116,13 +153,13 @@
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
-            <version>4.10</version>
+            <version>4.13.2</version>
             <scope>test</scope>
         </dependency>
     </dependencies>
 
     <properties>
-        <spring.version>3.0.5.RELEASE</spring.version>
+        <spring.version>4.3.21.RELEASE</spring.version>
     </properties>
 
     <build>

nevado-jms/src/main/java/org/skyscreamer/nevado/jms/NevadoSession.java

@@ -355,8 +355,10 @@ protected String getDurableEndpointQueueName(String durableSubscriptionName) {
 
         if (_connection.getClientID() != null)
         {
-            queueName += "_client-" + _connection.getClientID() + "";
+            queueName += "_" + _connection.getClientID() + "";
         }
+
+
         return queueName;
     }
 

nevado-jms/src/main/java/org/skyscreamer/nevado/jms/connector/amazonaws/AmazonAwsSQSConnector.java

@@ -4,20 +4,17 @@
 import com.amazonaws.AmazonServiceException;
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.Protocol;
-import com.amazonaws.auth.AWSCredentials;
-import com.amazonaws.auth.BasicAWSCredentials;
-import com.amazonaws.services.sns.AmazonSNS;
-import com.amazonaws.services.sns.AmazonSNSAsync;
-import com.amazonaws.services.sns.AmazonSNSAsyncClient;
-import com.amazonaws.services.sns.AmazonSNSClient;
+import com.amazonaws.auth.*;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.client.builder.ExecutorFactory;
+import com.amazonaws.services.sns.*;
 import com.amazonaws.services.sns.model.*;
-import com.amazonaws.services.sqs.AmazonSQS;
-import com.amazonaws.services.sqs.AmazonSQSAsyncClient;
-import com.amazonaws.services.sqs.AmazonSQSClient;
+import com.amazonaws.services.sqs.*;
 import com.amazonaws.services.sqs.model.CreateQueueRequest;
 import com.amazonaws.services.sqs.model.CreateQueueResult;
 import com.amazonaws.services.sqs.model.ListQueuesRequest;
 import com.amazonaws.services.sqs.model.ListQueuesResult;
+import org.apache.commons.lang.StringUtils;
 import org.skyscreamer.nevado.jms.connector.AbstractSQSConnector;
 import org.skyscreamer.nevado.jms.connector.SQSMessage;
 import org.skyscreamer.nevado.jms.connector.SQSQueue;
@@ -53,13 +50,12 @@ public class AmazonAwsSQSConnector extends AbstractSQSConnector {
     private boolean _testAlwaysPasses = false;
 
 
-    public AmazonAwsSQSConnector(String awsAccessKey, String awsSecretKey, boolean isSecure, long receiveCheckIntervalMs) {
-        this(awsAccessKey, awsSecretKey, isSecure, receiveCheckIntervalMs, false);
+    public AmazonAwsSQSConnector(String awsAccessKey, String awsSecretKey, String awsSQSEndpoint, String awsSNSEndpoint, boolean isSecure, long receiveCheckIntervalMs) {
+        this(awsAccessKey, awsSecretKey, awsSQSEndpoint, awsSNSEndpoint, isSecure, receiveCheckIntervalMs, false);
     }
 
-    public AmazonAwsSQSConnector(String awsAccessKey, String awsSecretKey, boolean isSecure, long receiveCheckIntervalMs, boolean isAsync) {
+    public AmazonAwsSQSConnector(String awsAccessKey, String awsSecretKey, String awsSQSEndpoint, String awsSNSEndpoint, boolean isSecure, long receiveCheckIntervalMs, boolean isAsync) {
         super(receiveCheckIntervalMs, isAsync);
-        AWSCredentials awsCredentials = new BasicAWSCredentials(awsAccessKey, awsSecretKey);
         ClientConfiguration clientConfiguration = new ClientConfiguration();
         String proxyHost = System.getProperty("http.proxyHost");
         String proxyPort = System.getProperty("http.proxyPort");
@@ -70,13 +66,77 @@ public AmazonAwsSQSConnector(String awsAccessKey, String awsSecretKey, boolean i
             }
         }
         clientConfiguration.setProtocol(isSecure ? Protocol.HTTPS : Protocol.HTTP);
+        AwsClientBuilder.EndpointConfiguration awsSQSEndpointConfiguration = new AwsClientBuilder.EndpointConfiguration(awsSQSEndpoint, null);
+        AwsClientBuilder.EndpointConfiguration awsSNSEndpointConfiguration = new AwsClientBuilder.EndpointConfiguration(awsSNSEndpoint, null);
+
         if (isAsync) {
-            ExecutorService executorService = Executors.newSingleThreadExecutor();
-            _amazonSQS = new AmazonSQSAsyncClient(awsCredentials, clientConfiguration, executorService);
-            _amazonSNS = new AmazonSNSAsyncClient(awsCredentials, clientConfiguration, executorService);
-        } else {
-            _amazonSQS = new AmazonSQSClient(awsCredentials, clientConfiguration);
-            _amazonSNS = new AmazonSNSClient(awsCredentials, clientConfiguration);
+            ExecutorFactory executorFactory = new ExecutorFactory() {
+                @Override
+                public ExecutorService newExecutor() {
+                    return Executors.newSingleThreadExecutor();
+                }
+            };
+
+            if(StringUtils.isNotEmpty(awsAccessKey) && StringUtils.isNotEmpty(awsSecretKey)) {
+                BasicAWSCredentials basicAWSCredentials = new BasicAWSCredentials(awsAccessKey, awsSecretKey);
+                _amazonSQS = AmazonSQSAsyncClientBuilder.standard()
+                        .withCredentials(new AWSStaticCredentialsProvider(basicAWSCredentials))
+                        .withClientConfiguration(clientConfiguration)
+                        .withExecutorFactory(executorFactory)
+                        .withEndpointConfiguration(awsSQSEndpointConfiguration)
+                        .build();
+                _amazonSNS = AmazonSNSAsyncClientBuilder.standard()
+                        .withCredentials(new AWSStaticCredentialsProvider(basicAWSCredentials))
+                        .withClientConfiguration(clientConfiguration)
+                        .withExecutorFactory(executorFactory)
+                        .withEndpointConfiguration(awsSNSEndpointConfiguration)
+                        .build();
+            }
+            else {
+                _amazonSQS = AmazonSQSAsyncClientBuilder
+                        .standard()
+                        .withClientConfiguration(clientConfiguration)
+                        .withCredentials(InstanceProfileCredentialsProvider.getInstance())
+                        .withExecutorFactory(executorFactory)
+                        .withEndpointConfiguration(awsSQSEndpointConfiguration)
+                        .build();
+                _amazonSNS = AmazonSNSAsyncClientBuilder
+                        .standard()
+                        .withClientConfiguration(clientConfiguration)
+                        .withCredentials(InstanceProfileCredentialsProvider.getInstance())
+                        .withExecutorFactory(executorFactory)
+                        .withEndpointConfiguration(awsSNSEndpointConfiguration)
+                        .build();
+            }
+        }
+        else {
+            if(StringUtils.isNotEmpty(awsAccessKey) && StringUtils.isNotEmpty(awsSecretKey)) {
+                BasicAWSCredentials basicAWSCredentials = new BasicAWSCredentials(awsAccessKey, awsSecretKey);
+                _amazonSQS = AmazonSQSClientBuilder.standard()
+                    .withCredentials(new AWSStaticCredentialsProvider(basicAWSCredentials))
+                    .withClientConfiguration(clientConfiguration)
+                    .withEndpointConfiguration(awsSQSEndpointConfiguration)
+                    .build();
+                _amazonSNS = AmazonSNSClientBuilder.standard()
+                    .withCredentials(new AWSStaticCredentialsProvider(basicAWSCredentials))
+                    .withClientConfiguration(clientConfiguration)
+                    .withEndpointConfiguration(awsSNSEndpointConfiguration)
+                    .build();
+            }
+            else {
+                _amazonSQS = AmazonSQSClientBuilder
+                        .standard()
+                        .withClientConfiguration(clientConfiguration)
+                        .withCredentials(InstanceProfileCredentialsProvider.getInstance())
+                        .withEndpointConfiguration(awsSQSEndpointConfiguration)
+                        .build();
+                _amazonSNS = AmazonSNSClientBuilder
+                        .standard()
+                        .withClientConfiguration(clientConfiguration)
+                        .withCredentials(InstanceProfileCredentialsProvider.getInstance())
+                        .withEndpointConfiguration(awsSNSEndpointConfiguration)
+                        .build();
+            }
         }
     }
 

nevado-jms/src/main/java/org/skyscreamer/nevado/jms/connector/amazonaws/AmazonAwsSQSConnectorFactory.java

@@ -15,19 +15,13 @@ public class AmazonAwsSQSConnectorFactory extends AbstractSQSConnectorFactory {
 
     @Override
     public AmazonAwsSQSConnector getInstance(String awsAccessKey, String awsSecretKey, String awsSQSEndpoint, String awsSNSEndpoint) {
-        AmazonAwsSQSConnector amazonAwsSQSConnector = createConnector(awsAccessKey, awsSecretKey);
+        AmazonAwsSQSConnector amazonAwsSQSConnector = createConnector(awsAccessKey, awsSecretKey, awsSQSEndpoint, awsSNSEndpoint);
         amazonAwsSQSConnector.setTestAlwaysPasses(_testAlwaysPasses);
-        if (StringUtils.isNotEmpty(awsSQSEndpoint)) {
-            amazonAwsSQSConnector.getAmazonSQS().setEndpoint(awsSQSEndpoint);
-        }
-        if (StringUtils.isNotEmpty(awsSNSEndpoint)) {
-            amazonAwsSQSConnector.getAmazonSNS().setEndpoint(awsSNSEndpoint);
-        }
         return amazonAwsSQSConnector;
     }
 
-    protected AmazonAwsSQSConnector createConnector(String awsAccessKey, String awsSecretKey) {
-        return new AmazonAwsSQSConnector(awsAccessKey, awsSecretKey, _isSecure, _receiveCheckIntervalMs, _useAsyncSend);
+    protected AmazonAwsSQSConnector createConnector(String awsAccessKey, String awsSecretKey, String awsSQSEndpoint, String awsSNSEndpoint) {
+        return new AmazonAwsSQSConnector(awsAccessKey, awsSecretKey, awsSQSEndpoint, awsSNSEndpoint, _isSecure, _receiveCheckIntervalMs, _useAsyncSend);
     }
 
     public void setUseAsyncSend(boolean useAsyncSend) {

nevado-jms/src/main/java/org/skyscreamer/nevado/jms/connector/amazonaws/PlainTextAmazonSQSConnector.java

@@ -1,43 +1,43 @@
-package org.skyscreamer.nevado.jms.connector.amazonaws;
-
-import javax.jms.JMSException;
-import javax.jms.TextMessage;
-
-import org.skyscreamer.nevado.jms.message.NevadoMessage;
-import org.skyscreamer.nevado.jms.message.NevadoTextMessage;
-
-/**
- * Overrides the serialisation handling from AbstractSQSConnector so that raw
- * text messages can be received/sent without wrapping.
- * 
- * @author qi.chen
- */
-public class PlainTextAmazonSQSConnector extends AmazonAwsSQSConnector {
-
-	public PlainTextAmazonSQSConnector(String awsAccessKey, String awsSecretKey, boolean isSecure,
-			long receiveCheckIntervalMs) {
-		super(awsAccessKey, awsSecretKey, isSecure, receiveCheckIntervalMs);
-	}
-
-	public PlainTextAmazonSQSConnector(String awsAccessKey, String awsSecretKey, boolean isSecure,
-			long receiveCheckIntervalMs, boolean isAsync) {
-		super(awsAccessKey, awsSecretKey, isSecure, receiveCheckIntervalMs, isAsync);
-	}
-
-	@Override
-	protected String serializeMessage(NevadoMessage message) throws JMSException {
-		if (message instanceof TextMessage) {
-			return ((TextMessage) message).getText();
-		} else {
-			throw new UnsupportedOperationException(getClass().getSimpleName() + " does not support " + message);
-		}
-	}
-
-	@Override
-	protected NevadoMessage deserializeMessage(String serializedMessage) throws JMSException {
-		NevadoTextMessage out = new NevadoTextMessage();
-		out.setText(serializedMessage);
-		return out;
-	}
-
-}
+package org.skyscreamer.nevado.jms.connector.amazonaws;
+
+import javax.jms.JMSException;
+import javax.jms.TextMessage;
+
+import org.skyscreamer.nevado.jms.message.NevadoMessage;
+import org.skyscreamer.nevado.jms.message.NevadoTextMessage;
+
+/**
+ * Overrides the serialisation handling from AbstractSQSConnector so that raw
+ * text messages can be received/sent without wrapping.
+ * 
+ * @author qi.chen
+ */
+public class PlainTextAmazonSQSConnector extends AmazonAwsSQSConnector {
+
+	public PlainTextAmazonSQSConnector(String awsAccessKey, String awsSecretKey, String awsSQSEndpoint, String awsSNSEndpoint, boolean isSecure,
+			long receiveCheckIntervalMs) {
+		super(awsAccessKey, awsSecretKey, awsSQSEndpoint, awsSNSEndpoint, isSecure, receiveCheckIntervalMs);
+	}
+
+	public PlainTextAmazonSQSConnector(String awsAccessKey, String awsSecretKey, String awsSQSEndpoint, String awsSNSEndpoint, boolean isSecure,
+			long receiveCheckIntervalMs, boolean isAsync) {
+		super(awsAccessKey, awsSecretKey, awsSQSEndpoint, awsSNSEndpoint, isSecure, receiveCheckIntervalMs, isAsync);
+	}
+
+	@Override
+	protected String serializeMessage(NevadoMessage message) throws JMSException {
+		if (message instanceof TextMessage) {
+			return ((TextMessage) message).getText();
+		} else {
+			throw new UnsupportedOperationException(getClass().getSimpleName() + " does not support " + message);
+		}
+	}
+
+	@Override
+	protected NevadoMessage deserializeMessage(String serializedMessage) throws JMSException {
+		NevadoTextMessage out = new NevadoTextMessage();
+		out.setText(serializedMessage);
+		return out;
+	}
+
+}