diff --git a/build.gradle b/build.gradle index a9d33a2..9150bda 100644 --- a/build.gradle +++ b/build.gradle @@ -30,7 +30,7 @@ ext { commonsIoVersion='2.4' jacocoVersion = '0.7.2.201409121644' slf4jVersion = '1.7.8' - springCloudAwsVersion = '1.0.0.BUILD-SNAPSHOT' + springCloudAwsVersion = '1.0.0.RELEASE' springIntegrationVersion = '4.1.2.RELEASE' idPrefix = 'aws' diff --git a/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java b/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java index f918b31..12ae20a 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java @@ -57,7 +57,11 @@ public class SqsMessageDrivenChannelAdapterParser extends AbstractSingleBeanDefi @Override protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - builder.addConstructorArgReference(element.getAttribute(AmazonWSParserUtils.SQS_REF)) + String sqs = element.getAttribute(AmazonWSParserUtils.SQS_REF); + if (!StringUtils.hasText(sqs)) { + parserContext.getReaderContext().error("'sqs' attribute is required.", element); + } + builder.addConstructorArgReference(sqs) .addConstructorArgValue(element.getAttribute("queues")); String channelName = element.getAttribute("channel"); if (!StringUtils.hasText(channelName)) { diff --git a/src/main/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParser.java b/src/main/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParser.java index 010da41..8c0928d 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParser.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParser.java @@ -31,15 +31,41 @@ import org.springframework.util.StringUtils; * The parser for the {@code } * * @author Artem Bilan + * @author Rahul Pilani */ public class SqsOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { + public static final String QUEUE_MESSAGING_TEMPLATE_REF = "queue-messaging-template"; + @Override protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SqsMessageHandler.class); - builder.addConstructorArgReference(element.getAttribute(AmazonWSParserUtils.SQS_REF)); - if (StringUtils.hasText(element.getAttribute(AmazonWSParserUtils.RESOURCE_ID_RESOLVER_REF))) { - builder.addConstructorArgReference(element.getAttribute(AmazonWSParserUtils.RESOURCE_ID_RESOLVER_REF)); + + String template = element.getAttribute(QUEUE_MESSAGING_TEMPLATE_REF); + boolean hasTemplate = StringUtils.hasText(template); + String sqs = element.getAttribute(AmazonWSParserUtils.SQS_REF); + boolean hasSqs = StringUtils.hasText(sqs); + String resourceIdResolver = element.getAttribute(AmazonWSParserUtils.RESOURCE_ID_RESOLVER_REF); + boolean hasResourceIdResolver = StringUtils.hasText(resourceIdResolver); + if (hasTemplate && (hasSqs || hasResourceIdResolver)) { + parserContext.getReaderContext().error(QUEUE_MESSAGING_TEMPLATE_REF + + " should not be defined in conjunction with " + AmazonWSParserUtils.SQS_REF + + " or " + AmazonWSParserUtils.RESOURCE_ID_RESOLVER_REF, element); + } + + if (!hasTemplate && !hasSqs) { + parserContext.getReaderContext().error("One of " + QUEUE_MESSAGING_TEMPLATE_REF + " or " + + AmazonWSParserUtils.SQS_REF + " must be defined.", element); + } + + if (hasSqs) { + builder.addConstructorArgReference(sqs); + } + if (hasResourceIdResolver) { + builder.addConstructorArgReference(resourceIdResolver); + } + if (hasTemplate) { + builder.addConstructorArgReference(template); } BeanDefinition queue = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("queue", "queue-expression", parserContext, element, false); 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 5712bf5..23cf639 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java @@ -34,6 +34,7 @@ import com.amazonaws.services.sqs.AmazonSQS; * The {@link AbstractMessageHandler} implementation for the Amazon SQS {@code sendMessage}. * * @author Artem Bilan + * @author Rahul Pilani * * @see QueueMessagingTemplate * @see org.springframework.cloud.aws.messaging.core.QueueMessageChannel @@ -51,7 +52,12 @@ public class SqsMessageHandler extends AbstractMessageHandler implements Integra } public SqsMessageHandler(AmazonSQS amazonSqs, ResourceIdResolver resourceIdResolver) { - this.template = new QueueMessagingTemplate(amazonSqs, resourceIdResolver); + this(new QueueMessagingTemplate(amazonSqs, resourceIdResolver)); + } + + public SqsMessageHandler(QueueMessagingTemplate template) { + Assert.notNull(template, "template must not be null."); + this.template = template; } public void setQueue(String queue) { diff --git a/src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd b/src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd index b3bee51..5b92d2a 100644 --- a/src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd +++ b/src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd @@ -1,7 +1,6 @@ + + + + A bean reference to the QueueMessagingTemplate that this class is supposed to use. + This attribute is mutually exclusive with the 'sqs' and 'resource-id-resolver' + attributes. + + + + + + + + @@ -514,7 +528,7 @@ Base type for the 'sqs-message-driven-channel-adapter' and 'sqs-outbound-channel-adapter' elements. - + The 'com.amazonaws.services.sqs.AmazonSQS' bean reference. diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad.xml b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad.xml new file mode 100644 index 0000000..d2f9296 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad.xml @@ -0,0 +1,27 @@ + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad2.xml b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad2.xml new file mode 100644 index 0000000..f5cff53 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad2.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad3.xml b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad3.xml new file mode 100644 index 0000000..540735d --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad3.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad4.xml b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad4.xml new file mode 100644 index 0000000..cef4f0e --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad4.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-good.xml b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-good.xml new file mode 100644 index 0000000..7f8d726 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-good.xml @@ -0,0 +1,33 @@ + + + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests.java new file mode 100644 index 0000000..393eb2f --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests.java @@ -0,0 +1,69 @@ +/* + * Copyright 2015 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.config.xml; + +import org.junit.Assert; +import org.junit.Test; + +import org.springframework.beans.factory.BeanDefinitionStoreException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.aws.outbound.SqsMessageHandler; +import org.springframework.integration.test.util.TestUtils; + +/** + * @author Rahul Pilani + * @author Artem Bilan + */ +public class SqsOutboundChannelAdapterParserTests { + + @Test(expected = BeanDefinitionStoreException.class) + public void test_sqs_resource_resolver_defined_with_queue_messaging_template() { + new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad.xml", this.getClass()); + } + + @Test(expected = BeanDefinitionStoreException.class) + public void test_sqs_defined_with_queue_messaging_template() { + new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad2.xml", this.getClass()); + } + + @Test(expected = BeanDefinitionStoreException.class) + public void test_resource_resolver_defined_with_queue_messaging_template() { + new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad3.xml", this.getClass()); + } + + @Test(expected = BeanDefinitionStoreException.class) + public void test_neither_sqs_nor_queue_messaging_template_defined() { + new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-bad4.xml", this.getClass()); + } + + @Test + public void test_happy_path_with_queue_messaging_template() { + ApplicationContext applicationContext = + new ClassPathXmlApplicationContext("SqsOutboundChannelAdapterParserTests-context-good.xml", this.getClass()); + + SqsMessageHandler handlerWithTemplate = + applicationContext.getBean("sqsOutboundChannelAdapterWithQueueMessagingTemplate.handler", + SqsMessageHandler.class); + Assert.assertNotNull(TestUtils.getPropertyValue(handlerWithTemplate, "template")); + + SqsMessageHandler handlerWithSqs = applicationContext.getBean("sqsOutboundChannelAdapterWithSqs.handler", + SqsMessageHandler.class); + Assert.assertNotNull(TestUtils.getPropertyValue(handlerWithSqs, "template")); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/outbound/AbstractSqsMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/AbstractSqsMessageHandlerTests.java new file mode 100644 index 0000000..4e61b49 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/outbound/AbstractSqsMessageHandlerTests.java @@ -0,0 +1,91 @@ +/* + * Copyright 2015 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.outbound; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.support.MessageBuilder; + +import com.amazonaws.services.sqs.AmazonSQS; +import com.amazonaws.services.sqs.model.SendMessageRequest; + +/** + * Parent class to contain tests that exercise the SqsMessageHandler class + * + * Subclasses can instantiate SqsMessageHandler in their own way. + * + * @author Rahul Pilani + * @author Artem Bilan + */ +public abstract class AbstractSqsMessageHandlerTests { + + @Autowired + protected AmazonSQS amazonSqs; + + @Autowired + protected MessageChannel sqsSendChannel; + + @Autowired + protected SqsMessageHandler sqsMessageHandler; + + @Test + public void testSqsMessageHandler() { + Message message = MessageBuilder.withPayload("message").build(); + try { + this.sqsSendChannel.send(message); + } + catch (Exception e) { + assertThat(e, instanceOf(MessageHandlingException.class)); + assertThat(e.getCause(), instanceOf(IllegalStateException.class)); + } + + this.sqsMessageHandler.setQueue("foo"); + this.sqsSendChannel.send(message); + ArgumentCaptor sendMessageRequestArgumentCaptor = + ArgumentCaptor.forClass(SendMessageRequest.class); + verify(this.amazonSqs).sendMessage(sendMessageRequestArgumentCaptor.capture()); + assertEquals("http://queue-url.com/foo", sendMessageRequestArgumentCaptor.getValue().getQueueUrl()); + + message = MessageBuilder.withPayload("message").setHeader(AwsHeaders.QUEUE, "bar").build(); + this.sqsSendChannel.send(message); + verify(this.amazonSqs, times(2)).sendMessage(sendMessageRequestArgumentCaptor.capture()); + assertEquals("http://queue-url.com/bar", sendMessageRequestArgumentCaptor.getValue().getQueueUrl()); + + SpelExpressionParser spelExpressionParser = new SpelExpressionParser(); + Expression expression = spelExpressionParser.parseExpression("headers.foo"); + this.sqsMessageHandler.setQueueExpression(expression); + message = MessageBuilder.withPayload("message").setHeader("foo", "baz").build(); + this.sqsSendChannel.send(message); + verify(this.amazonSqs, times(3)).sendMessage(sendMessageRequestArgumentCaptor.capture()); + assertEquals("http://queue-url.com/baz", sendMessageRequestArgumentCaptor.getValue().getQueueUrl()); + } + +} 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 db3044d..8a4508f 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java +++ b/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java @@ -16,89 +16,36 @@ package org.springframework.integration.aws.outbound; -import static org.hamcrest.Matchers.instanceOf; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertThat; import static org.mockito.Matchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import org.junit.Test; import org.junit.runner.RunWith; -import org.mockito.ArgumentCaptor; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.expression.Expression; -import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.aws.support.AwsHeaders; import org.springframework.integration.config.EnableIntegration; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessageHandlingException; -import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.model.GetQueueUrlRequest; import com.amazonaws.services.sqs.model.GetQueueUrlResult; -import com.amazonaws.services.sqs.model.SendMessageRequest; /** + * Instantiating SqsMessageHandler using amazonSqs. + * * @author Artem Bilan + * @author Rahul Pilani */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration -public class SqsMessageHandlerTests { +public class SqsMessageHandlerTests extends AbstractSqsMessageHandlerTests { - @Autowired - private AmazonSQS amazonSqs; - - @Autowired - private MessageChannel sqsSendChannel; - - @Autowired - private SqsMessageHandler sqsMessageHandler; - - @Test - public void testSqsMessageHandler() { - Message message = MessageBuilder.withPayload("message").build(); - try { - this.sqsSendChannel.send(message); - } - catch (Exception e) { - assertThat(e, instanceOf(MessageHandlingException.class)); - assertThat(e.getCause(), instanceOf(IllegalStateException.class)); - } - - this.sqsMessageHandler.setQueue("foo"); - this.sqsSendChannel.send(message); - ArgumentCaptor sendMessageRequestArgumentCaptor = - ArgumentCaptor.forClass(SendMessageRequest.class); - verify(this.amazonSqs).sendMessage(sendMessageRequestArgumentCaptor.capture()); - assertEquals("http://queue-url.com/foo", sendMessageRequestArgumentCaptor.getValue().getQueueUrl()); - - message = MessageBuilder.withPayload("message").setHeader(AwsHeaders.QUEUE, "bar").build(); - this.sqsSendChannel.send(message); - verify(this.amazonSqs, times(2)).sendMessage(sendMessageRequestArgumentCaptor.capture()); - assertEquals("http://queue-url.com/bar", sendMessageRequestArgumentCaptor.getValue().getQueueUrl()); - - SpelExpressionParser spelExpressionParser = new SpelExpressionParser(); - Expression expression = spelExpressionParser.parseExpression("headers.foo"); - this.sqsMessageHandler.setQueueExpression(expression); - message = MessageBuilder.withPayload("message").setHeader("foo", "baz").build(); - this.sqsSendChannel.send(message); - verify(this.amazonSqs, times(3)).sendMessage(sendMessageRequestArgumentCaptor.capture()); - assertEquals("http://queue-url.com/baz", sendMessageRequestArgumentCaptor.getValue().getQueueUrl()); - } @Configuration @EnableIntegration diff --git a/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerWithQueueMessagingTemplateTests.java b/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerWithQueueMessagingTemplateTests.java new file mode 100644 index 0000000..be5a1e2 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerWithQueueMessagingTemplateTests.java @@ -0,0 +1,86 @@ +/* + * Copyright 2015 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.outbound; + +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; + +import org.junit.runner.RunWith; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.cloud.aws.messaging.core.QueueMessagingTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.MessageHandler; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.amazonaws.services.sqs.AmazonSQS; +import com.amazonaws.services.sqs.model.GetQueueUrlRequest; +import com.amazonaws.services.sqs.model.GetQueueUrlResult; + +/** + * Instantiating SqsMessageHandler using QueueMessagingTemplate. + * + * @author Rahul Pilani + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class SqsMessageHandlerWithQueueMessagingTemplateTests extends AbstractSqsMessageHandlerTests { + + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + public AmazonSQS amazonSqs() { + AmazonSQS amazonSqs = mock(AmazonSQS.class); + + doAnswer(new Answer() { + + @Override + public GetQueueUrlResult answer(InvocationOnMock invocation) throws Throwable { + GetQueueUrlRequest getQueueUrlRequest = (GetQueueUrlRequest) invocation.getArguments()[0]; + GetQueueUrlResult queueUrl = new GetQueueUrlResult(); + queueUrl.setQueueUrl("http://queue-url.com/" + getQueueUrlRequest.getQueueName()); + return queueUrl; + } + + }).when(amazonSqs).getQueueUrl(any(GetQueueUrlRequest.class)); + + return amazonSqs; + } + + @Bean + public QueueMessagingTemplate queueMessagingTemplate() { + return new QueueMessagingTemplate(amazonSqs()); + } + + @Bean + @ServiceActivator(inputChannel = "sqsSendChannel") + public MessageHandler sqsMessageHandler() { + return new SqsMessageHandler(queueMessagingTemplate()); + } + + } + +}