From 6e5e9f191be1c0a225b220c5b2051cf06ff7523e Mon Sep 17 00:00:00 2001 From: skelly201 Date: Mon, 17 Sep 2018 14:03:05 -0400 Subject: [PATCH] Add SqsMessageHandler DestinationResolver ctor --- .../aws/outbound/SqsMessageHandler.java | 14 ++++-- .../aws/outbound/SqsMessageHandlerTests.java | 49 +++++++++++++++++++ 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java index 1a62399..9c494cb 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java +++ b/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 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()); } diff --git a/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java index f5432af..0d2000c 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java +++ b/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 message = MessageBuilder.withPayload("message").build(); + + this.sqsMessageHandlerWithAutoQueueCreate.setQueue("foo"); + this.sqsSendChannelWithAutoCreate.send(message); + ArgumentCaptor createQueueRequestArgumentCaptor = + ArgumentCaptor.forClass(CreateQueueRequest.class); + verify(this.amazonSqs).createQueue(createQueueRequestArgumentCaptor.capture()); + assertThat(createQueueRequestArgumentCaptor.getValue().getQueueName()).isEqualTo("foo"); + + ArgumentCaptor 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); + } + } }