Add SqsMessageHandler DestinationResolver ctor

src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -52,6 +52,7 @@ import com.amazonaws.services.sqs.model.SendMessageResult;
  * @author Artem Bilan
  * @author Rahul Pilani
  * @author Taylor Wicksell
+ * @author Seth Kelly
  *
  * @see AmazonSQSAsync#sendMessageAsync(SendMessageRequest, AsyncHandler)
  * @see com.amazonaws.handlers.AsyncHandler
@@ -75,12 +76,19 @@ public class SqsMessageHandler extends AbstractAwsMessageHandler<Map<String, Mes
 
 
 	public SqsMessageHandler(AmazonSQSAsync amazonSqs) {
-		this(amazonSqs, null);
+		this(amazonSqs, (ResourceIdResolver) null);
 	}
 
 	public SqsMessageHandler(AmazonSQSAsync amazonSqs, ResourceIdResolver resourceIdResolver) {
+		this(amazonSqs, new DynamicQueueUrlDestinationResolver(amazonSqs, resourceIdResolver));
+	}
+
+	public SqsMessageHandler(AmazonSQSAsync amazonSqs, DestinationResolver<?> destinationResolver) {
+		Assert.notNull(amazonSqs, "'amazonSqs' must not be null");
+		Assert.notNull(destinationResolver, "'destinationResolver' must not be null");
+
 		this.amazonSqs = amazonSqs;
-		this.destinationResolver = new DynamicQueueUrlDestinationResolver(amazonSqs, resourceIdResolver);
+		this.destinationResolver = destinationResolver;
 		doSetHeaderMapper(new SqsHeaderMapper());
 	}
 

src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java

@@ -30,6 +30,7 @@ import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.aws.messaging.support.destination.DynamicQueueUrlDestinationResolver;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.expression.Expression;
@@ -43,11 +44,14 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import com.amazonaws.handlers.AsyncHandler;
 import com.amazonaws.services.sqs.AmazonSQSAsync;
+import com.amazonaws.services.sqs.model.CreateQueueRequest;
+import com.amazonaws.services.sqs.model.CreateQueueResult;
 import com.amazonaws.services.sqs.model.GetQueueUrlRequest;
 import com.amazonaws.services.sqs.model.GetQueueUrlResult;
 import com.amazonaws.services.sqs.model.MessageAttributeValue;
@@ -58,9 +62,11 @@ import com.amazonaws.services.sqs.model.SendMessageRequest;
  *
  * @author Artem Bilan
  * @author Rahul Pilani
+ * @author Seth Kelly
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
 public class SqsMessageHandlerTests {
 
 	@Autowired
@@ -69,9 +75,15 @@ public class SqsMessageHandlerTests {
 	@Autowired
 	protected MessageChannel sqsSendChannel;
 
+	@Autowired
+	protected MessageChannel sqsSendChannelWithAutoCreate;
+
 	@Autowired
 	protected SqsMessageHandler sqsMessageHandler;
 
+	@Autowired
+	protected SqsMessageHandler sqsMessageHandlerWithAutoQueueCreate;
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testSqsMessageHandler() {
@@ -123,6 +135,26 @@ public class SqsMessageHandlerTests {
 
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testSqsMessageHandlerWithAutoQueueCreate() {
+		Message<String> message = MessageBuilder.withPayload("message").build();
+
+		this.sqsMessageHandlerWithAutoQueueCreate.setQueue("foo");
+		this.sqsSendChannelWithAutoCreate.send(message);
+		ArgumentCaptor<CreateQueueRequest> createQueueRequestArgumentCaptor =
+				ArgumentCaptor.forClass(CreateQueueRequest.class);
+		verify(this.amazonSqs).createQueue(createQueueRequestArgumentCaptor.capture());
+		assertThat(createQueueRequestArgumentCaptor.getValue().getQueueName()).isEqualTo("foo");
+
+		ArgumentCaptor<SendMessageRequest> sendMessageRequestArgumentCaptor =
+				ArgumentCaptor.forClass(SendMessageRequest.class);
+		verify(this.amazonSqs)
+				.sendMessageAsync(sendMessageRequestArgumentCaptor.capture(), any(AsyncHandler.class));
+		assertThat(sendMessageRequestArgumentCaptor.getValue().getQueueUrl())
+				.isEqualTo("http://queue-url.com/foo");
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {
@@ -140,6 +172,15 @@ public class SqsMessageHandlerTests {
 					.given(amazonSqs)
 					.getQueueUrl(any(GetQueueUrlRequest.class));
 
+			willAnswer(invocation -> {
+				CreateQueueRequest createQueueRequest = (CreateQueueRequest) invocation.getArguments()[0];
+				CreateQueueResult queueUrl = new CreateQueueResult();
+				queueUrl.setQueueUrl("http://queue-url.com/" + createQueueRequest.getQueueName());
+				return queueUrl;
+			})
+					.given(amazonSqs)
+					.createQueue(any(CreateQueueRequest.class));
+
 			return amazonSqs;
 		}
 
@@ -149,6 +190,14 @@ public class SqsMessageHandlerTests {
 			return new SqsMessageHandler(amazonSqs());
 		}
 
+		@Bean
+		@ServiceActivator(inputChannel = "sqsSendChannelWithAutoCreate")
+		public MessageHandler sqsMessageHandlerWithAutoQueueCreate() {
+			DynamicQueueUrlDestinationResolver destinationResolver = new DynamicQueueUrlDestinationResolver(amazonSqs(), null);
+			destinationResolver.setAutoCreate(true);
+			return new SqsMessageHandler(amazonSqs(), destinationResolver);
+		}
+
 	}
 
 }