diff --git a/build.gradle b/build.gradle index f95cfaa..a9d33a2 100644 --- a/build.gradle +++ b/build.gradle @@ -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" } diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar index 0087cd3..085a1cd 100644 Binary files a/gradle/wrapper/gradle-wrapper.jar and b/gradle/wrapper/gradle-wrapper.jar differ diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 81aa59b..81c2b60 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -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 diff --git a/src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java b/src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java index 8260558..b7974dc 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java @@ -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()); } } diff --git a/src/main/java/org/springframework/integration/aws/config/xml/AmazonWSParserUtils.java b/src/main/java/org/springframework/integration/aws/config/xml/AmazonWSParserUtils.java index 9cfcfee..4370c8f 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/AmazonWSParserUtils.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/AmazonWSParserUtils.java @@ -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; } + } 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 new file mode 100644 index 0000000..f918b31 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java @@ -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 } + * + * @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"); + } + +} 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 new file mode 100644 index 0000000..010da41 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParser.java @@ -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 } + * + * @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(); + } + +} diff --git a/src/main/java/org/springframework/integration/aws/config/xml/package-info.java b/src/main/java/org/springframework/integration/aws/config/xml/package-info.java new file mode 100644 index 0000000..28e56de --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/config/xml/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides the parser classes for Integration AWS Namespace. + */ +package org.springframework.integration.aws.config.xml; diff --git a/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java new file mode 100644 index 0000000..6859990 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java @@ -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 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 getHandlerMethods() { + Set 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); + } + + } + +} diff --git a/src/main/java/org/springframework/integration/aws/inbound/package-info.java b/src/main/java/org/springframework/integration/aws/inbound/package-info.java new file mode 100644 index 0000000..f48c7d9 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/inbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes which represent inbound AWS components. + */ +package org.springframework.integration.aws.inbound; diff --git a/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java new file mode 100644 index 0000000..5712bf5 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java @@ -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); + } + +} diff --git a/src/main/java/org/springframework/integration/aws/outbound/package-info.java b/src/main/java/org/springframework/integration/aws/outbound/package-info.java new file mode 100644 index 0000000..843e9d1 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/outbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes which represent outbound AWS components. + */ +package org.springframework.integration.aws.outbound; diff --git a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java new file mode 100644 index 0000000..7ee6b0e --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java @@ -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"; + +} diff --git a/src/main/java/org/springframework/integration/aws/support/package-info.java b/src/main/java/org/springframework/integration/aws/support/package-info.java new file mode 100644 index 0000000..d774de6 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/support/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides support classes for AWS components. + */ +package org.springframework.integration.aws.support; 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 e57b115..b3bee51 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,17 +1,17 @@ + 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"> + schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd"/> @@ -20,46 +20,46 @@ ]]> - - - - - Specify the access key to be used to connect to the AWS service. - - - - - - - Specify the secret key corresponding to the access key to be used to - authenticate the user and connect to the AWS service. - - - - - - - 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. - - - - - - - - - - - - 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 - - - + + + + + Specify the access key to be used to connect to the AWS service. + + + + + + + Specify the secret key corresponding to the access key to be used to + authenticate the user and connect to the AWS service. + + + + + + + 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. + + + + + + + + + + + + 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 + + + @@ -166,7 +166,8 @@ - + @@ -201,7 +202,8 @@ - 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 @@ - 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 @@ 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. @@ -348,4 +352,197 @@ + + + + + Defines an outbound SQS Channel Adapter for sending messages to queues. + + + + + + + + + + + + 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. + + + + + + + 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. + + + + + + + + + + + + Configures an endpoint ('SqsMessageDrivenChannelAdapter') that will receive + Amazon SQS message from the provided 'queues'. + + + + + + + + + Message Channel to which error Messages should be sent. + + + + + + + + + + + + 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. + + + + + + + Comma-separated SQS queue names or their URLs. + + + + + + + The 'org.springframework.core.task.TaskExecutor' to run the underlying listener task + from the + 'org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer'. + + + + + + + + + + + + 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. + + + + + + + Configures the duration (in seconds) that the received messages are hidden from + subsequent poll requests after being retrieved from the system. + + + + + + + + + + + + + + A reference to a bean that implements the + 'org.springframework.messaging.core.DestinationResolver' interface. + E.g. + 'org.springframework.cloud.aws.messaging.support.destination.DynamicQueueUrlDestinationResolver' + + + + + + + + + + + + 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. + + + + + + + + + + + + Base type for the 'sqs-message-driven-channel-adapter' and 'sqs-outbound-channel-adapter' elements. + + + + + + The 'com.amazonaws.services.sqs.AmazonSQS' bean reference. + Must be 'AmazonSQSAsync' for the 'sqs-message-driven-channel-adapter'. + + + + + + + + + + + + The 'org.springframework.cloud.aws.core.env.ResourceIdResolver' bean reference. + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml new file mode 100644 index 0000000..6761df8 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml @@ -0,0 +1,43 @@ + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests.java new file mode 100644 index 0000000..9fa7286 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests.java @@ -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")); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests-context.xml b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests-context.xml new file mode 100644 index 0000000..cc69062 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests-context.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests.java new file mode 100644 index 0000000..889a5d3 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageHandlerParserTests.java @@ -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")); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java new file mode 100644 index 0000000..cf77d5d --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java @@ -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; + } + + } + +} diff --git a/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java new file mode 100644 index 0000000..db3044d --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java @@ -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 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 + 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 + @ServiceActivator(inputChannel = "sqsSendChannel") + public MessageHandler sqsMessageHandler() { + return new SqsMessageHandler(amazonSqs()); + } + + } + +} diff --git a/src/test/resources/log4j.properties b/src/test/resources/log4j.properties index 8e9bc68..bf146e4 100644 --- a/src/test/resources/log4j.properties +++ b/src/test/resources/log4j.properties @@ -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