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

INTEXT-7: Add SQS Adapters

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

Upgrade to Gradle 2.3

Minor Doc Polishing
This commit is contained in:
Artem Bilan
2015-02-12 16:29:00 +02:00
committed by Gary Russell
parent d4dedef383
commit ec4a93142d
22 changed files with 1203 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
build.gradle
View File

@@ -2,7 +2,7 @@ description = 'Spring Integration AWS Support'
buildscript {
repositories {
maven { url 'http://repo.spring.io/plugins-snapshot' }
maven { url 'http://repo.spring.io/plugins-release' }
}
}
@@ -15,11 +15,10 @@ group = 'org.springframework.integration'
repositories {
maven { url 'http://repo.spring.io/libs-milestone' }
maven { url 'http://repo.spring.io/plugins-release' }
maven { url 'http://repo.spring.io/libs-snapshot' }
}
sourceCompatibility=1.6
targetCompatibility=1.6
sourceCompatibility = targetCompatibility = 1.7
// See http://www.gradle.org/docs/current/userguide/dependency_management.html#sub:configurations
// and http://www.gradle.org/docs/current/dsl/org.gradle.api.artifacts.ConfigurationContainer.html
@@ -29,8 +28,9 @@ configurations {
ext {
commonsIoVersion='2.4'
jacocoVersion = '0.7.1.201405082137'
springCloudAwsVersion = '1.0.0.RC2'
jacocoVersion = '0.7.2.201409121644'
slf4jVersion = '1.7.8'
springCloudAwsVersion = '1.0.0.BUILD-SNAPSHOT'
springIntegrationVersion = '4.1.2.RELEASE'
idPrefix = 'aws'
@@ -56,15 +56,13 @@ dependencies {
testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion"
testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion"
jacoco "org.jacoco:org.jacoco.agent:$jacocoVersion:runtime"
}
eclipse {
project {
natures += 'org.springframework.ide.eclipse.core.springnature'
}
}
eclipse.project.natures += 'org.springframework.ide.eclipse.core.springnature'
javadoc {
group = 'Documentation'
@@ -249,6 +247,6 @@ task dist(dependsOn: assemble) {
task wrapper(type: Wrapper) {
description = 'Generates gradlew[.bat] scripts'
gradleVersion = '1.12'
gradleVersion = '2.3'
distributionUrl = "http://services.gradle.org/distributions/gradle-${gradleVersion}-all.zip"
}

BIN
gradle/wrapper/gradle-wrapper.jar vendored
View File

Binary file not shown.

4
gradle/wrapper/gradle-wrapper.properties vendored
View File

@@ -1,6 +1,6 @@
#Thu Aug 14 15:52:49 EEST 2014
#Mon Feb 16 15:46:30 EET 2015
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists
distributionUrl=http\://services.gradle.org/distributions/gradle-1.12-all.zip
distributionUrl=http\://services.gradle.org/distributions/gradle-2.3-all.zip

5
src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.config.xml;
import org.springframework.integration.aws.s3.config.xml.AmazonS3InboundChannelAdapterParser;
@@ -23,6 +24,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
* The namespace handler for "int-aws" namespace
*
* @author Amol Nayak
* @author Artem Bilan
* @since 0.5
*/
public class AWSNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@@ -31,6 +33,9 @@ public class AWSNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public void init() {
this.registerBeanDefinitionParser("s3-outbound-channel-adapter", new AmazonS3OutboundChannelAdapterParser());
this.registerBeanDefinitionParser("s3-inbound-channel-adapter", new AmazonS3InboundChannelAdapterParser());
this.registerBeanDefinitionParser("sqs-outbound-channel-adapter", new SqsOutboundChannelAdapterParser());
this.registerBeanDefinitionParser("sqs-message-driven-channel-adapter",
new SqsMessageDrivenChannelAdapterParser());
}
}

36
src/main/java/org/springframework/integration/aws/config/xml/AmazonWSParserUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-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.
@@ -13,8 +13,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
@@ -22,23 +25,29 @@ import org.springframework.integration.aws.core.AWSCredentials;
import org.springframework.integration.aws.core.BasicAWSCredentials;
import org.springframework.integration.aws.core.PropertiesAWSCredentials;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* The utility class for the namespace parsers
*
* @author Amol Nayak
*
* @author Artem Bilan
* @since 0.5
*
*/
public final class AmazonWSParserUtils {
public static final String ACCESS_KEY = "accessKey";
public static final String SECRET_KEY = "secretKey";
public static final String PROPERTIES_FILE = "propertiesFile";
public static final String CREDENTIALS_REF = "credentials-ref";
public static final String SQS_REF = "sqs";
public static final String RESOURCE_ID_RESOLVER_REF = "resource-id-resolver";
private AmazonWSParserUtils() {
throw new AssertionError("Cannot instantiate the utility class");
}
@@ -48,11 +57,10 @@ public final class AmazonWSParserUtils {
* Registers the {@link AWSCredentials} bean with the current ApplicationContext if
* accessKey and secretKey is given, if the credentials-ref is given, the given value
* is returned.
*
* @param element
* @param parserContext
*/
public static String getAmazonWSCredentials(Element element,ParserContext parserContext) {
public static String getAmazonWSCredentials(Element element, ParserContext parserContext) {
//TODO: Some mechanism to use the same instance with same ACCESS_KEY to be implemented
String accessKey = element.getAttribute(ACCESS_KEY);
String secretKey = element.getAttribute(SECRET_KEY);
@@ -60,31 +68,32 @@ public final class AmazonWSParserUtils {
String credentialsRef = element.getAttribute(CREDENTIALS_REF);
String awsCredentialsGeneratedName;
if(StringUtils.hasText(credentialsRef)) {
if(StringUtils.hasText(propertiesFile)
if (StringUtils.hasText(credentialsRef)) {
if (StringUtils.hasText(propertiesFile)
|| StringUtils.hasText(accessKey)
|| StringUtils.hasText(secretKey)) {
parserContext.getReaderContext().error("When " + CREDENTIALS_REF + " is specified, " +
"do not specify the " + PROPERTIES_FILE + " attribute or the "
+ SECRET_KEY + " and " + ACCESS_KEY + " attributes", element);
+ SECRET_KEY + " and " + ACCESS_KEY + " attributes", element);
}
awsCredentialsGeneratedName = credentialsRef;
}
else {
if(StringUtils.hasText(propertiesFile)) {
if(StringUtils.hasText(accessKey) && StringUtils.hasText(secretKey)) {
if (StringUtils.hasText(propertiesFile)) {
if (StringUtils.hasText(accessKey) && StringUtils.hasText(secretKey)) {
parserContext.getReaderContext().error("When " + ACCESS_KEY + " and " + SECRET_KEY +
" are specified, do not specify the " + PROPERTIES_FILE + " attribute", element);
}
BeanDefinitionBuilder builder =
BeanDefinitionBuilder.genericBeanDefinition(PropertiesAWSCredentials.class);
BeanDefinitionBuilder.genericBeanDefinition(PropertiesAWSCredentials.class);
builder.addConstructorArgValue(propertiesFile);
awsCredentialsGeneratedName = BeanDefinitionReaderUtils.registerWithGeneratedName(
builder.getBeanDefinition(), parserContext.getRegistry());
} else {
}
else {
BeanDefinitionBuilder builder
= BeanDefinitionBuilder.genericBeanDefinition(BasicAWSCredentials.class);
= BeanDefinitionBuilder.genericBeanDefinition(BasicAWSCredentials.class);
builder.addConstructorArgValue(accessKey);
builder.addConstructorArgValue(secretKey);
awsCredentialsGeneratedName = BeanDefinitionReaderUtils.registerWithGeneratedName(
@@ -95,4 +104,5 @@ public final class AmazonWSParserUtils {
return awsCredentialsGeneratedName;
}
}

82
src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,82 @@
/*
* 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.w3c.dom.Element;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.aws.inbound.SqsMessageDrivenChannelAdapter;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
* The parser for the {@code <int-aws:sqs-message-driven-channel-adapter>}
*
* @author Artem Bilan
*/
public class SqsMessageDrivenChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
@Override
protected Class<?> getBeanClass(Element element) {
return SqsMessageDrivenChannelAdapter.class;
}
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
String id = super.resolveId(element, definition, parserContext);
if (!element.hasAttribute("channel")) {
// the created channel will get the 'id', so the adapter's bean name includes a suffix
id = id + ".adapter";
}
if (!StringUtils.hasText(id)) {
id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry());
}
return id;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
builder.addConstructorArgReference(element.getAttribute(AmazonWSParserUtils.SQS_REF))
.addConstructorArgValue(element.getAttribute("queues"));
String channelName = element.getAttribute("channel");
if (!StringUtils.hasText(channelName)) {
channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
}
builder.addPropertyReference("outputChannel", channelName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
AmazonWSParserUtils.RESOURCE_ID_RESOLVER_REF);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "payload-type");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delete-message-on-exception");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-number-of-messages");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "visibility-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "wait-time-out");
}
}

52
src/main/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* 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.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.aws.outbound.SqsMessageHandler;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
* The parser for the {@code <int-aws:sqs-outbound-channel-adapter>}
*
* @author Artem Bilan
*/
public class SqsOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@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));
}
BeanDefinition queue = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("queue",
"queue-expression", parserContext, element, false);
if (queue != null) {
builder.addPropertyValue("queueExpression", queue);
}
return builder.getBeanDefinition();
}
}

4
src/main/java/org/springframework/integration/aws/config/xml/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides the parser classes for Integration AWS Namespace.
*/
package org.springframework.integration.aws.config.xml;

150
src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java Normal file
View File

@@ -0,0 +1,150 @@
/*
* 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.inbound;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.cloud.aws.core.env.ResourceIdResolver;
import org.springframework.cloud.aws.messaging.config.SimpleMessageListenerContainerFactory;
import org.springframework.cloud.aws.messaging.listener.QueueMessageHandler;
import org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.aws.support.AwsHeaders;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.messaging.handler.HandlerMethod;
import org.springframework.util.Assert;
import com.amazonaws.services.sqs.AmazonSQSAsync;
/**
* The {@link MessageProducerSupport} implementation for the Amazon SQS {@code receiveMessage}.
* Works in 'listener' manner and delegates hard to the {@link SimpleMessageListenerContainer}.
*
* @author Artem Bilan
* *
* @see SimpleMessageListenerContainerFactory
* @see SimpleMessageListenerContainer
* @see QueueMessageHandler
*/
public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport
implements DisposableBean {
private final SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory =
new SimpleMessageListenerContainerFactory();
private final String[] queues;
private SimpleMessageListenerContainer listenerContainer;
public SqsMessageDrivenChannelAdapter(AmazonSQSAsync amazonSqs, String... queues) {
Assert.noNullElements(queues, "'queues' must not be empty");
this.simpleMessageListenerContainerFactory.setAmazonSqs(amazonSqs);
this.queues = Arrays.copyOf(queues, queues.length);
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.simpleMessageListenerContainerFactory.setTaskExecutor(taskExecutor);
}
public void setMaxNumberOfMessages(Integer maxNumberOfMessages) {
this.simpleMessageListenerContainerFactory.setMaxNumberOfMessages(maxNumberOfMessages);
}
public void setVisibilityTimeout(Integer visibilityTimeout) {
this.simpleMessageListenerContainerFactory.setVisibilityTimeout(visibilityTimeout);
}
public void setWaitTimeOut(Integer waitTimeOut) {
this.simpleMessageListenerContainerFactory.setWaitTimeOut(waitTimeOut);
}
public void setResourceIdResolver(ResourceIdResolver resourceIdResolver) {
this.simpleMessageListenerContainerFactory.setResourceIdResolver(resourceIdResolver);
}
public void setDestinationResolver(DestinationResolver<String> destinationResolver) {
this.simpleMessageListenerContainerFactory.setDestinationResolver(destinationResolver);
}
public void setDeleteMessageOnException(Boolean deleteMessageOnException) {
this.simpleMessageListenerContainerFactory.setDeleteMessageOnException(deleteMessageOnException);
}
@Override
protected void onInit() {
super.onInit();
this.listenerContainer = this.simpleMessageListenerContainerFactory.createSimpleMessageListenerContainer();
this.listenerContainer.setMessageHandler(new IntegrationQueueMessageHandler());
try {
this.listenerContainer.afterPropertiesSet();
}
catch (Exception e) {
throw new BeanCreationException("Cannot instantiate 'SimpleMessageListenerContainer'", e);
}
}
@Override
protected void doStart() {
this.listenerContainer.start();
}
@Override
protected void doStop() {
this.listenerContainer.stop();
}
@Override
public void destroy() throws Exception {
this.listenerContainer.destroy();
}
private class IntegrationQueueMessageHandler extends QueueMessageHandler {
@Override
public Map<MappingInformation, HandlerMethod> getHandlerMethods() {
Set<String> queues = new HashSet<>(Arrays.asList(SqsMessageDrivenChannelAdapter.this.queues));
return Collections.singletonMap(new MappingInformation(queues), null);
}
@Override
protected void handleMessageInternal(Message<?> message, String lookupDestination) {
MessageHeaders headers = message.getHeaders();
Message<?> messageToSend = getMessageBuilderFactory()
.fromMessage(message)
.removeHeaders(QueueMessageHandler.Headers.LOGICAL_RESOURCE_ID_MESSAGE_HEADER_KEY,
"MessageId",
"ReceiptHandle")
.setHeader(AwsHeaders.MESSAGE_ID, headers.get("MessageId"))
.setHeader(AwsHeaders.RECEIPT_HANDLE, headers.get("ReceiptHandle"))
.setHeader(AwsHeaders.QUEUE,
headers.get(QueueMessageHandler.Headers.LOGICAL_RESOURCE_ID_MESSAGE_HEADER_KEY))
.build();
sendMessage(messageToSend);
}
}
}

4
src/main/java/org/springframework/integration/aws/inbound/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes which represent inbound AWS components.
*/
package org.springframework.integration.aws.inbound;

89
src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java Normal file
View File

@@ -0,0 +1,89 @@
/*
* 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 org.springframework.cloud.aws.core.env.ResourceIdResolver;
import org.springframework.cloud.aws.messaging.core.QueueMessagingTemplate;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.aws.support.AwsHeaders;
import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.amazonaws.services.sqs.AmazonSQS;
/**
* The {@link AbstractMessageHandler} implementation for the Amazon SQS {@code sendMessage}.
*
* @author Artem Bilan
*
* @see QueueMessagingTemplate
* @see org.springframework.cloud.aws.messaging.core.QueueMessageChannel
*/
public class SqsMessageHandler extends AbstractMessageHandler implements IntegrationEvaluationContextAware {
private final QueueMessagingTemplate template;
private Expression queueExpression;
private EvaluationContext evaluationContext;
public SqsMessageHandler(AmazonSQS amazonSqs) {
this(amazonSqs, null);
}
public SqsMessageHandler(AmazonSQS amazonSqs, ResourceIdResolver resourceIdResolver) {
this.template = new QueueMessagingTemplate(amazonSqs, resourceIdResolver);
}
public void setQueue(String queue) {
Assert.hasText(queue, "'queue' must not be empty");
this.queueExpression = new LiteralExpression(queue);
}
public void setQueueExpression(Expression queueExpression) {
Assert.notNull(queueExpression, "'queueExpression' must not be null");
this.queueExpression = queueExpression;
}
@Override
public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
this.evaluationContext = evaluationContext;
}
@Override
protected void onInit() throws Exception {
Assert.notNull(this.evaluationContext);
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
String queue = message.getHeaders().get(AwsHeaders.QUEUE, String.class);
if (!StringUtils.hasText(queue) && this.queueExpression != null) {
queue = this.queueExpression.getValue(this.evaluationContext, message, String.class);
}
Assert.state(queue != null, "'queue' must not be null for sending an SQS message. " +
"Consider configuring this handler with a 'queue'( or 'queueExpression') or supply an " +
"'aws_queue' message header");
this.template.send(queue, message);
}
}

4
src/main/java/org/springframework/integration/aws/outbound/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides classes which represent outbound AWS components.
*/
package org.springframework.integration.aws.outbound;

32
src/main/java/org/springframework/integration/aws/support/AwsHeaders.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* 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.support;
/**
* @author Artem Bilan
*/
public abstract class AwsHeaders {
private static final String PREFIX = "aws_";
public static final String QUEUE = PREFIX + "queue";
public static final String MESSAGE_ID = PREFIX + "messageId";
public static final String RECEIPT_HANDLE = PREFIX + "receiptHandle";
}

4
src/main/java/org/springframework/integration/aws/support/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides support classes for AWS components.
*/
package org.springframework.integration.aws.support;

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

@@ -1,17 +1,17 @@
<?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"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
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"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd"/>
schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd"/>
<xsd:annotation>
@@ -20,46 +20,46 @@
]]></xsd:documentation>
</xsd:annotation>
<xsd:attributeGroup name="awsAdaptersCommonAttributes">
<xsd:attribute name="accessKey" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the access key to be used to connect to the AWS service.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="secretKey" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the secret key corresponding to the access key to be used to
authenticate the user and connect to the AWS service.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="propertiesFile" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specifies the properties file used that will be used to hold the access key
and the secret key of the AWS. This file expects two properties accessKey and
secretKey for Access key and Secret Key respectively. This attribute is mutually
exclusive to the accessKey, secretKey and credentials-ref attribute.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="credentials-ref" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.aws.core.AWSCredentials"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
If none of the default mechanisms work, you may provide a custom implementation
of org.springframework.integration.aws.core.AWSCredentials. Various use cases are
when you might want to read the credentials from a database or any other secure store
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup name="awsAdaptersCommonAttributes">
<xsd:attribute name="accessKey" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the access key to be used to connect to the AWS service.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="secretKey" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify the secret key corresponding to the access key to be used to
authenticate the user and connect to the AWS service.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="propertiesFile" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specifies the properties file used that will be used to hold the access key
and the secret key of the AWS. This file expects two properties accessKey and
secretKey for Access key and Secret Key respectively. This attribute is mutually
exclusive to the accessKey, secretKey and credentials-ref attribute.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="credentials-ref" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.aws.core.AWSCredentials"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
If none of the default mechanisms work, you may provide a custom implementation
of org.springframework.integration.aws.core.AWSCredentials. Various use cases are
when you might want to read the credentials from a database or any other secure store
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:attributeGroup>
<xsd:element name="s3-outbound-channel-adapter">
@@ -166,7 +166,8 @@
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.aws.s3.FileNameGenerationStrategy"/>
<tool:expected-type
type="org.springframework.integration.aws.s3.FileNameGenerationStrategy"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
@@ -201,7 +202,8 @@
<tool:expected-type type="org.springframework.integration.aws.s3.core.AmazonS3Operations"/>
</xsd:appinfo>
<xsd:documentation>
Reference to the bean with an implementation of org.springframework.integration.aws.s3.core.AmazonS3Operations
Reference to the bean with an implementation of
org.springframework.integration.aws.s3.core.AmazonS3Operations
that would be used to perform the operations on the S3 bucket. If not provided, the
default implementation used is
org.springframework.integration.aws.s3.core.DefaultAmazonS3Operations which is the
@@ -266,7 +268,8 @@
<tool:expected-type type="org.springframework.integration.aws.s3.core.AmazonS3Operations"/>
</xsd:appinfo>
<xsd:documentation>
Reference to the bean with an implementation of org.springframework.integration.aws.s3.core.AmazonS3Operations
Reference to the bean with an implementation of
org.springframework.integration.aws.s3.core.AmazonS3Operations
that would be used to perform the operations on the S3 bucket. If not provided, the
default implementation used is
org.springframework.integration.aws.s3.core.DefaultAmazonS3Operations which uses AWS SDK.
@@ -314,7 +317,8 @@
<xsd:documentation>
The maximum number of objects returned in the listOperation performed on the S3 bucket
The default value used internally is 100. That is not more than 100 objects would be returned
in one call to list the objects, subsequent calls will be made to the Web service to retrieve the next
in one call to list the objects, subsequent calls will be made to the Web service to retrieve
the next
batch of objects.
</xsd:documentation>
</xsd:annotation>
@@ -348,4 +352,197 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="sqs-outbound-channel-adapter">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an outbound SQS Channel Adapter for sending messages to queues.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="baseSqsAdapterType">
<xsd:choice minOccurs="0" maxOccurs="2">
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
minOccurs="0" maxOccurs="1"/>
</xsd:choice>
<xsd:attribute name="queue">
<xsd:annotation>
<xsd:documentation>
The Amazon queue name or URL.
Mutually exclusive with 'queue-expression'.
This attribute isn't mandatory and the queue can be specified in message headers
with the 'AwsHeaders.QUEUE' header name.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="queue-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression that resolves to an Amazon queue or its URL.
The 'requestMessage' is the root object for evaluation context.
Mutually exclusive with 'queue'.
This attribute isn't mandatory and the queue can be specified in message headers with
the 'AwsHeaders.QUEUE' header name.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="sqs-message-driven-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures an endpoint ('SqsMessageDrivenChannelAdapter') that will receive
Amazon SQS message from the provided 'queues'.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="baseSqsAdapterType">
<xsd:attribute name="error-channel" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which error Messages should be sent.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Maximum amount of time in milliseconds to wait when sending a message
to the channel if such channel may block.
For example, a Queue Channel can block until space is available
if its maximum capacity has been reached.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="queues" use="required">
<xsd:annotation>
<xsd:documentation>
Comma-separated SQS queue names or their URLs.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The 'org.springframework.core.task.TaskExecutor' to run the underlying listener task
from the
'org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.core.task.TaskExecutor"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="max-number-of-messages">
<xsd:annotation>
<xsd:documentation>
Configure the maximum number of messages that should be retrieved during one
poll to the Amazon SQS system. This number must be a positive, non-zero number that
has a maximum number of 10. Values higher then 10 are currently
not supported by the queueing system. Defaults to 1.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="visibility-timeout">
<xsd:annotation>
<xsd:documentation>
Configures the duration (in seconds) that the received messages are hidden from
subsequent poll requests after being retrieved from the system.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="wait-time-out">
<xsd:annotation>
<xsd:documentation>
<![CDATA[
Configures the wait timeout that the poll request will wait for new message
to arrive if the are currently no messages on the queue.
Higher values will reduce poll request to the system significantly. The value should
be between 1 and 20. For more information read the
http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-long-polling.html
]]>
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="destination-resolver">
<xsd:annotation>
<xsd:documentation>
A reference to a bean that implements the
'org.springframework.messaging.core.DestinationResolver' interface.
E.g.
'org.springframework.cloud.aws.messaging.support.destination.DynamicQueueUrlDestinationResolver'
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.core.DestinationResolver"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="delete-message-on-exception">
<xsd:annotation>
<xsd:documentation>
Defines if a message must be deleted or not if the message handling throws an exception
By default this value is set to 'true' which means that the message is deleted to
avoid poison messages. If the 'error-channel' is specified and the direct error flow
doesn't rethrow a 'MessagingException' this attribute doesn't have an effect. If this is
to 'false' it is a responsibility of the error flow to remove the message from
AmazonSQS.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="baseSqsAdapterType">
<xsd:annotation>
<xsd:documentation>
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:annotation>
<xsd:documentation>
The 'com.amazonaws.services.sqs.AmazonSQS' bean reference.
Must be 'AmazonSQSAsync' for the 'sqs-message-driven-channel-adapter'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="com.amazonaws.services.sqs.AmazonSQS"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="resource-id-resolver">
<xsd:annotation>
<xsd:documentation>
The 'org.springframework.cloud.aws.core.env.ResourceIdResolver' bean reference.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
</xsd:complexType>
</xsd:schema>

43
src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml Normal file
View File

@@ -0,0 +1,43 @@
<?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-1.0.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>
<bean id="taskExecutor" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.core.task.TaskExecutor"/>
</bean>
<bean id="destinationResolver"
class="org.springframework.cloud.aws.messaging.support.destination.DynamicQueueUrlDestinationResolver">
<constructor-arg ref="sqs"/>
<constructor-arg ref="resourceIdResolver"/>
</bean>
<int-aws:sqs-message-driven-channel-adapter sqs="sqs"
auto-startup="false"
channel="errorChannel"
error-channel="nullChannel"
task-executor="taskExecutor"
phase="100"
id="sqsMessageDrivenChannelAdapter"
queues="foo, bar"
delete-message-on-exception="false"
max-number-of-messages="5"
visibility-timeout="200"
wait-time-out="40"
send-timeout="2000"
destination-resolver="destinationResolver"
resource-id-resolver="resourceIdResolver"/>
</beans>

101
src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.util.Set;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.aws.core.env.ResourceIdResolver;
import org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.aws.inbound.SqsMessageDrivenChannelAdapter;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.amazonaws.services.sqs.AmazonSQS;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class SqsMessageDrivenChannelAdapterParserTests {
@Autowired
private AmazonSQS amazonSqs;
@Autowired
private ResourceIdResolver resourceIdResolver;
@Autowired
private DestinationResolver<?> destinationResolver;
@Autowired
private TaskExecutor taskExecutor;
@Autowired
private MessageChannel errorChannel;
@Autowired
private NullChannel nullChannel;
@Autowired
private SqsMessageDrivenChannelAdapter sqsMessageDrivenChannelAdapter;
@Test
public void testSqsMessageDrivenChannelAdapterParser() {
SimpleMessageListenerContainer listenerContainer =
TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, "listenerContainer",
SimpleMessageListenerContainer.class);
assertSame(this.amazonSqs, TestUtils.getPropertyValue(listenerContainer, "amazonSqs"));
assertSame(this.resourceIdResolver, TestUtils.getPropertyValue(listenerContainer, "resourceIdResolver"));
assertSame(this.taskExecutor, TestUtils.getPropertyValue(listenerContainer, "taskExecutor"));
assertSame(this.destinationResolver, TestUtils.getPropertyValue(listenerContainer, "destinationResolver"));
assertFalse(listenerContainer.isRunning());
assertEquals(5, TestUtils.getPropertyValue(listenerContainer, "maxNumberOfMessages"));
assertEquals(200, TestUtils.getPropertyValue(listenerContainer, "visibilityTimeout"));
assertEquals(40, TestUtils.getPropertyValue(listenerContainer, "waitTimeOut"));
assertFalse(TestUtils.getPropertyValue(listenerContainer, "deleteMessageOnException", Boolean.class));
@SuppressWarnings("rawtypes")
Set queues = TestUtils.getPropertyValue(listenerContainer, "queues", Set.class);
assertTrue(queues.contains("foo"));
assertTrue(queues.contains("bar"));
assertEquals(100, this.sqsMessageDrivenChannelAdapter.getPhase());
assertFalse(this.sqsMessageDrivenChannelAdapter.isAutoStartup());
assertFalse(this.sqsMessageDrivenChannelAdapter.isRunning());
assertSame(this.errorChannel, TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, "outputChannel"));
assertSame(this.nullChannel, TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, "errorChannel"));
assertEquals(2000L, TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter,
"messagingTemplate.sendTimeout"));
}
}

24
src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests-context.xml Normal file
View File

@@ -0,0 +1,24 @@
<?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-1.0.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>
<int-aws:sqs-outbound-channel-adapter sqs="sqs"
auto-startup="false"
channel="errorChannel"
phase="100"
id="sqsOutboundChannelAdapter"
queue="foo"
resource-id-resolver="resourceIdResolver"/>
</beans>

79
src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests.java Normal file
View File

@@ -0,0 +1,79 @@
/*
* 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.aws.core.env.ResourceIdResolver;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.amazonaws.services.sqs.AmazonSQS;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class SqsMessageHandlerParserTests {
@Autowired
private AmazonSQS amazonSqs;
@Autowired
private ResourceIdResolver resourceIdResolver;
@Autowired
private MessageChannel errorChannel;
@Autowired
private EventDrivenConsumer sqsOutboundChannelAdapter;
@Autowired
@Qualifier("sqsOutboundChannelAdapter.handler")
private MessageHandler sqsOutboundChannelAdapterHandler;
@Test
public void testSqsMessageHandlerParser() {
assertSame(this.amazonSqs,
TestUtils.getPropertyValue(this.sqsOutboundChannelAdapterHandler, "template.amazonSqs"));
assertSame(this.resourceIdResolver, TestUtils.getPropertyValue(this.sqsOutboundChannelAdapterHandler,
"template.destinationResolver.targetDestinationResolver.resourceIdResolver"));
assertEquals("foo", TestUtils.getPropertyValue(this.sqsOutboundChannelAdapterHandler,
"queueExpression.literalValue"));
assertEquals(100, this.sqsOutboundChannelAdapter.getPhase());
assertFalse(this.sqsOutboundChannelAdapter.isAutoStartup());
assertFalse(this.sqsOutboundChannelAdapter.isRunning());
assertSame(this.errorChannel, TestUtils.getPropertyValue(this.sqsOutboundChannelAdapter, "inputChannel"));
assertSame(this.sqsOutboundChannelAdapterHandler,
TestUtils.getPropertyValue(this.sqsOutboundChannelAdapter, "handler"));
}
}

110
src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java Normal file
View File

@@ -0,0 +1,110 @@
/*
* 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.inbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.aws.support.AwsHeaders;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessageProducer;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.amazonaws.services.sqs.AmazonSQSAsync;
import com.amazonaws.services.sqs.model.GetQueueAttributesRequest;
import com.amazonaws.services.sqs.model.GetQueueAttributesResult;
import com.amazonaws.services.sqs.model.GetQueueUrlRequest;
import com.amazonaws.services.sqs.model.GetQueueUrlResult;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.amazonaws.services.sqs.model.ReceiveMessageResult;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class SqsMessageDrivenChannelAdapterTests {
@Autowired
private PollableChannel inputChannel;
@Test
public void testSqsMessageDrivenChannelAdapter() {
org.springframework.messaging.Message<?> receive = this.inputChannel.receive(1000);
assertNotNull(receive);
assertThat((String) receive.getPayload(), Matchers.isOneOf("messageContent", "messageContent2"));
assertEquals("testQueue", receive.getHeaders().get(AwsHeaders.QUEUE));
receive = this.inputChannel.receive(1000);
assertNotNull(receive);
assertThat((String) receive.getPayload(), Matchers.isOneOf("messageContent", "messageContent2"));
assertEquals("testQueue", receive.getHeaders().get(AwsHeaders.QUEUE));
}
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@Bean
public AmazonSQSAsync amazonSqs() {
AmazonSQSAsync sqs = mock(AmazonSQSAsync.class);
when(sqs.getQueueUrl(new GetQueueUrlRequest("testQueue"))).thenReturn(new GetQueueUrlResult().
withQueueUrl("http://testQueue.amazonaws.com"));
when(sqs.receiveMessage(new ReceiveMessageRequest("http://testQueue.amazonaws.com")
.withAttributeNames("All")
.withMessageAttributeNames("All")
.withMaxNumberOfMessages(10)))
.thenReturn(new ReceiveMessageResult()
.withMessages(new Message().withBody("messageContent"),
new Message().withBody("messageContent2")))
.thenReturn(new ReceiveMessageResult());
when(sqs.getQueueAttributes(any(GetQueueAttributesRequest.class)))
.thenReturn(new GetQueueAttributesResult());
return sqs;
}
@Bean
public PollableChannel inputChannel() {
return new QueueChannel();
}
@Bean
public MessageProducer sqsMessageDrivenChannelAdapter() {
SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(amazonSqs(), "testQueue");
adapter.setOutputChannel(inputChannel());
return adapter;
}
}
}

134
src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java Normal file
View File

@@ -0,0 +1,134 @@
/*
* 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.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;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class SqsMessageHandlerTests {
@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
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
@ServiceActivator(inputChannel = "sqsSendChannel")
public MessageHandler sqsMessageHandler() {
return new SqsMessageHandler(amazonSqs());
}
}
}

4
src/test/resources/log4j.properties
View File

@@ -1,8 +1,8 @@
log4j.rootCategory=INFO, stdout
log4j.rootCategory=WARN, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n
log4j.category.org.springframework=WARN
log4j.category.org.springframework.integration.aws=DEBUG
log4j.category.org.springframework.integration.aws=WARN