Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INTEXT-141: Add SqsMessageHandler template ctor

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-141

Adding in a constructor to the `SqsMessageHandler` so that `QueueMessagingTemplate` can be configured independently.

Updated with comments

- Updated the `spring-integration-aws-1.0.xsd` to include the `queue-messaging-template` param.
- Updated the `SqsOutboundChannelAdapterParserTest` to look for the `queue-messaging-template` param, and validate that `sqs` and `resource-resolver-id` params are mutually exclusive to `queue-messaging-template`.
- Added tests for the parser.
- Added test for the new constructor.

Updated with feedback
- Validation for `sqs` parameter missing in xml config
- Made `SqsMessageHandlerWithQueueMessagingTemplateTests` use the same tests as `SqsMessageHandler` tests by moving the tests to a common parent class: `AbstractSqsMessageHandlerTest`
- Other minor fixes.

Whitespace and error message fixes.

Polishing
This commit is contained in:
Rahul Pilani
2015-03-10 08:40:49 +02:00
committed by Artem Bilan
parent ec4a93142d
commit a551224f9d
14 changed files with 445 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

6
src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java
View File

@@ -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)) {

32
src/main/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParser.java
View File

@@ -31,15 +31,41 @@ import org.springframework.util.StringUtils;
* The parser for the {@code <int-aws:sqs-outbound-channel-adapter>}
*
* @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);

8
src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java
View File

@@ -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) {

18
src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd
View File

@@ -1,7 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/aws"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/aws"
@@ -387,6 +386,21 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="queue-messaging-template">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.cloud.aws.messaging.core.QueueMessagingTemplate"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -514,7 +528,7 @@
Base type for the 'sqs-message-driven-channel-adapter' and 'sqs-outbound-channel-adapter' elements.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="sqs" use="required">
<xsd:attribute name="sqs">
<xsd:annotation>
<xsd:documentation>
The 'com.amazonaws.services.sqs.AmazonSQS' bean reference.

27
src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad.xml Normal file
View File

@@ -0,0 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
xmlns:aws-messaging="http://www.springframework.org/schema/cloud/aws/messaging"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/aws http://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd
http://www.springframework.org/schema/cloud/aws/messaging http://www.springframework.org/schema/cloud/aws/messaging/spring-cloud-aws-messaging.xsd">
<aws-messaging:sqs-async-client id="sqs"/>
<bean id="resourceIdResolver" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
</bean>
<aws-messaging:queue-messaging-template amazon-sqs="sqs" id="queueMessagingTemplate"/>
<int-aws:sqs-outbound-channel-adapter sqs="sqs"
auto-startup="false"
channel="errorChannel"
phase="100"
id="sqsOutboundChannelAdapter"
queue="foo"
resource-id-resolver="resourceIdResolver"
queue-messaging-template="queueMessagingTemplate"/>
</beans>

26
src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad2.xml Normal file
View File

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
xmlns:aws-messaging="http://www.springframework.org/schema/cloud/aws/messaging"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/aws http://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd
http://www.springframework.org/schema/cloud/aws/messaging http://www.springframework.org/schema/cloud/aws/messaging/spring-cloud-aws-messaging.xsd">
<aws-messaging:sqs-async-client id="sqs"/>
<bean id="resourceIdResolver" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
</bean>
<aws-messaging:queue-messaging-template amazon-sqs="sqs" id="queueMessagingTemplate"/>
<int-aws:sqs-outbound-channel-adapter sqs="sqs"
auto-startup="false"
channel="errorChannel"
phase="100"
id="sqsOutboundChannelAdapter"
queue="foo"
queue-messaging-template="queueMessagingTemplate"/>
</beans>

26
src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad3.xml Normal file
View File

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
xmlns:aws-messaging="http://www.springframework.org/schema/cloud/aws/messaging"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/aws http://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd
http://www.springframework.org/schema/cloud/aws/messaging http://www.springframework.org/schema/cloud/aws/messaging/spring-cloud-aws-messaging.xsd">
<aws-messaging:sqs-async-client id="sqs"/>
<bean id="resourceIdResolver" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
</bean>
<aws-messaging:queue-messaging-template amazon-sqs="sqs" id="queueMessagingTemplate"/>
<int-aws:sqs-outbound-channel-adapter auto-startup="false"
channel="errorChannel"
phase="100"
id="sqsOutboundChannelAdapter"
queue="foo"
resource-id-resolver="resourceIdResolver"
queue-messaging-template="queueMessagingTemplate"/>
</beans>

25
src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-bad4.xml Normal file
View File

@@ -0,0 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
xmlns:aws-messaging="http://www.springframework.org/schema/cloud/aws/messaging"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/aws http://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd
http://www.springframework.org/schema/cloud/aws/messaging http://www.springframework.org/schema/cloud/aws/messaging/spring-cloud-aws-messaging.xsd">
<aws-messaging:sqs-async-client id="sqs"/>
<bean id="resourceIdResolver" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
</bean>
<aws-messaging:queue-messaging-template amazon-sqs="sqs" id="queueMessagingTemplate"/>
<int-aws:sqs-outbound-channel-adapter auto-startup="false"
channel="errorChannel"
phase="100"
id="sqsOutboundChannelAdapter"
queue="foo"
resource-id-resolver="resourceIdResolver"/>
</beans>

33
src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests-context-good.xml Normal file
View File

@@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
xmlns:aws-messaging="http://www.springframework.org/schema/cloud/aws/messaging"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/aws http://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd
http://www.springframework.org/schema/cloud/aws/messaging http://www.springframework.org/schema/cloud/aws/messaging/spring-cloud-aws-messaging.xsd">
<aws-messaging:sqs-async-client id="sqs"/>
<bean id="resourceIdResolver" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
</bean>
<aws-messaging:queue-messaging-template amazon-sqs="sqs" id="queueMessagingTemplate"/>
<int-aws:sqs-outbound-channel-adapter sqs="sqs"
auto-startup="false"
channel="errorChannel"
phase="100"
id="sqsOutboundChannelAdapterWithSqs"
queue="foo"
resource-id-resolver="resourceIdResolver"/>
<int-aws:sqs-outbound-channel-adapter auto-startup="false"
channel="errorChannel"
phase="100"
id="sqsOutboundChannelAdapterWithQueueMessagingTemplate"
queue="foo"
queue-messaging-template="queueMessagingTemplate"/>
</beans>

69
src/test/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParserTests.java Normal file
View File

@@ -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"));
}
}

91
src/test/java/org/springframework/integration/aws/outbound/AbstractSqsMessageHandlerTests.java Normal file
View File

@@ -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<String> 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<SendMessageRequest> 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());
}
}

61
src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java
View File

@@ -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<String> 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<SendMessageRequest> 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

86
src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerWithQueueMessagingTemplateTests.java Normal file
View File

@@ -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<GetQueueUrlResult>() {
@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());
}
}
}