Resolve SQS queue by ARN (#619)

Fixes #561

Author: techadri

docs/src/main/asciidoc/sqs.adoc

@@ -134,7 +134,11 @@ public void queueListener(Person person) {
 ----
 
 In this example a queue listener container is started that polls the SQS `queueName` passed to the `MessageMapping`
-annotation. The incoming messages are converted to the target type and then the annotated method `queueListener` is invoked.
+annotation. The incoming messages are converted to the target type and then the annotated method `queueListener` is invoked. There are multiple options to specify the queue that should be listened to:
+
+ * Using the queue name. Works only for queues within the same account and region your application/sqs client is using.
+ * Using the queue url. This allows access to a queue that is in a different region than your application/sqs client is using.
+ * Using the queue arn. Can be used to listen to queues that are owned by another account, but live within the same region your application/sqs client uses.
 
 In addition to the payload, headers can be injected in the listener methods with the `@Header` or `@Headers`
 annotations. `@Header` is used to inject a specific header value while `@Headers` injects a `Map<String, String>`

spring-cloud-aws-messaging/src/main/java/io/awspring/cloud/messaging/support/destination/DynamicQueueUrlDestinationResolver.java

@@ -26,6 +26,7 @@
 import com.amazonaws.services.sqs.model.GetQueueUrlResult;
 import com.amazonaws.services.sqs.model.QueueDoesNotExistException;
 import io.awspring.cloud.core.env.ResourceIdResolver;
+import io.awspring.cloud.core.naming.AmazonResourceName;
 
 import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.messaging.core.DestinationResolver;
@@ -87,7 +88,15 @@ public String resolveDestination(String name) throws DestinationResolutionExcept
 		}
 		else {
 			try {
-				GetQueueUrlResult getQueueUrlResult = this.amazonSqs.getQueueUrl(new GetQueueUrlRequest(queueName));
+				GetQueueUrlRequest request = new GetQueueUrlRequest(queueName);
+
+				if (AmazonResourceName.isValidAmazonResourceName(queueName)) {
+					AmazonResourceName resourceName = AmazonResourceName.fromString(name);
+					request.setQueueName(resourceName.getResourceType());
+					request.setQueueOwnerAWSAccountId(resourceName.getAccount());
+				}
+
+				GetQueueUrlResult getQueueUrlResult = this.amazonSqs.getQueueUrl(request);
 				return getQueueUrlResult.getQueueUrl();
 			}
 			catch (QueueDoesNotExistException e) {

spring-cloud-aws-messaging/src/test/java/io/awspring/cloud/messaging/support/destination/DynamicQueueUrlDestinationResolverTest.java

@@ -37,7 +37,7 @@
 class DynamicQueueUrlDestinationResolverTest {
 
 	@Test
-	void testAutoCreate() throws Exception {
+	void testAutoCreate() {
 		AmazonSQS amazonSqs = mock(AmazonSQS.class);
 		String queueUrl = "https://foo/bar";
 		when(amazonSqs.createQueue(new CreateQueueRequest("foo")))
@@ -50,7 +50,7 @@ void testAutoCreate() throws Exception {
 	}
 
 	@Test
-	void testAbsoluteUrl() throws Exception {
+	void testAbsoluteUrl() {
 		AmazonSQS amazonSqs = mock(AmazonSQS.class);
 		DynamicQueueUrlDestinationResolver dynamicQueueDestinationResolver = new DynamicQueueUrlDestinationResolver(
 				amazonSqs);
@@ -59,7 +59,37 @@ void testAbsoluteUrl() throws Exception {
 	}
 
 	@Test
-	void testNoAutoCreate() throws Exception {
+	void resolveDestination_shouldResolveArnToUrl() {
+		AmazonSQS amazonSqs = mock(AmazonSQS.class);
+		String expectedQueueUrl = "https://sqs-amazon.aws.com/123123123/myQueue";
+		DynamicQueueUrlDestinationResolver dynamicQueueDestinationResolver = new DynamicQueueUrlDestinationResolver(
+				amazonSqs);
+		when(amazonSqs.getQueueUrl(new GetQueueUrlRequest("myQueue").withQueueOwnerAWSAccountId("123123123")))
+				.thenReturn(new GetQueueUrlResult().withQueueUrl(expectedQueueUrl));
+		String arn = "arn:aws:sqs:eu-central-1:123123123:myQueue";
+		assertThat(dynamicQueueDestinationResolver.resolveDestination(arn)).isEqualTo(expectedQueueUrl);
+	}
+
+	@Test
+	void resolveDestination_shouldThrowError_IfQueueDoesNotExist() {
+		AmazonSQS amazonSqs = mock(AmazonSQS.class);
+		AmazonServiceException exception = new QueueDoesNotExistException("AWS.SimpleQueueService.NonExistentQueue");
+		exception.setErrorCode("AWS.SimpleQueueService.NonExistentQueue");
+		String arn = "arn:aws:sqs:eu-central-1:123123123:myQueue";
+		when(amazonSqs.getQueueUrl(new GetQueueUrlRequest("myQueue").withQueueOwnerAWSAccountId("123123123")))
+				.thenThrow(exception);
+		DynamicQueueUrlDestinationResolver dynamicQueueDestinationResolver = new DynamicQueueUrlDestinationResolver(
+				amazonSqs);
+		try {
+			dynamicQueueDestinationResolver.resolveDestination(arn);
+		}
+		catch (DestinationResolutionException e) {
+			assertThat(e.getMessage()).startsWith("The queue does not exist.");
+		}
+	}
+
+	@Test
+	void testNoAutoCreate() {
 		AmazonSQS amazonSqs = mock(AmazonSQS.class);
 		String queueUrl = "https://foo/bar";
 		when(amazonSqs.getQueueUrl(new GetQueueUrlRequest("foo")))
@@ -71,7 +101,7 @@ void testNoAutoCreate() throws Exception {
 	}
 
 	@Test
-	void testInvalidDestinationName() throws Exception {
+	void testInvalidDestinationName() {
 		AmazonSQS amazonSqs = mock(AmazonSQS.class);
 		AmazonServiceException exception = new QueueDoesNotExistException("AWS.SimpleQueueService.NonExistentQueue");
 		exception.setErrorCode("AWS.SimpleQueueService.NonExistentQueue");
@@ -88,7 +118,7 @@ void testInvalidDestinationName() throws Exception {
 	}
 
 	@Test
-	void testPotentiallyNoAccessToPerformGetQueueUrl() throws Exception {
+	void testPotentiallyNoAccessToPerformGetQueueUrl() {
 		AmazonSQS amazonSqs = mock(AmazonSQS.class);
 		AmazonServiceException exception = new QueueDoesNotExistException("AWS.SimpleQueueService.NonExistentQueue");
 		exception.setErrorCode("AWS.SimpleQueueService.NonExistentQueue");
@@ -107,7 +137,7 @@ void testPotentiallyNoAccessToPerformGetQueueUrl() throws Exception {
 	}
 
 	@Test
-	void resolveDestination_withResourceIdResolver_shouldUseIt() throws Exception {
+	void resolveDestination_withResourceIdResolver_shouldUseIt() {
 		AmazonSQS amazonSqs = mock(AmazonSQS.class);
 		ResourceIdResolver resourceIdResolver = mock(ResourceIdResolver.class);
 		when(resourceIdResolver.resolveToPhysicalResourceId(anyString())).thenReturn("http://queue.com");
@@ -121,7 +151,7 @@ void resolveDestination_withResourceIdResolver_shouldUseIt() throws Exception {
 	}
 
 	@Test
-	void resolveDestination_withResourceIdResolver_nonUrlId_shouldGetUrlByResolvedName() throws Exception {
+	void resolveDestination_withResourceIdResolver_nonUrlId_shouldGetUrlByResolvedName() {
 		String queueUrl = "http://queue.com";
 		String resolvedQueueName = "some-queue-name";
 		AmazonSQS amazonSqs = mock(AmazonSQS.class);
@@ -138,7 +168,7 @@ void resolveDestination_withResourceIdResolver_nonUrlId_shouldGetUrlByResolvedNa
 	}
 
 	@Test
-	void instantiation_withNullAmazonClient_shouldThrowAnError() throws Exception {
+	void instantiation_withNullAmazonClient_shouldThrowAnError() {
 		assertThatThrownBy(() -> new DynamicQueueUrlDestinationResolver(null, null))
 				.isInstanceOf(IllegalArgumentException.class);
 	}