diff --git a/README.md b/README.md index 451b8e9..340ab47 100644 --- a/README.md +++ b/README.md @@ -21,10 +21,10 @@ This guide intends to explain briefly the various adapters available for [Amazon * **Amazon Simple Email Service (SES)** * **Amazon Simple Storage Service (S3)** -* **Amazon Simple Queue Service (SQS)** (Development complete, coming soon) +* **Amazon Simple Queue Service (SQS)** +* **Amazon Simple Notification Service (SNS)** * **Amazon DynamoDB** (Analysis ongoing) * **Amazon SimpleDB** (Not initiated) -* **Amazon SNS** (Not initiated) Sample XML Namespace configurations for each adapter as well as sample code snippets are provided wherever necessary. Of the above libraries, *SES* and *SNS* provide outbound adapters only. All other services have inbound @@ -71,6 +71,183 @@ There is no adapter for SES, since [Spring Cloud AWS][] provides implementations `org.springframework.mail.MailSender` - `SimpleEmailServiceMailSender` and `SimpleEmailServiceJavaMailSender`, which can be injected to the ``. +##Amazon Simple Queue Service (SQS) + +The `SQS` adapters are fully based on the [Spring Cloud AWS][] foundation, so for more information about the +background components and core configuration, please, refer to the documentation of that project. + +###Outbound Channel Adapter + +The SQS Outbound Channel Adapter is presented by the `SqsMessageHandler` implementation +(``) and allows to send message to the SQS `queue` with provided `AmazonSQS` +client. An SQS queue can be configured explicitly on the adapter (using +`org.springframework.integration.expression.ValueExpression`) or as a SpEL `Expression`, which is evaluated against +request message as a root object of evaluation context. In addition the `queue` can be extracted from the message +headers under `AwsHeaders.QUEUE`. + +The Java Configuration is pretty simple: + +````java +@SpringBootApplication +public static class MyConfiguration { + + @Autowired + private AmazonSQS amazonSqs; + + @Bean + public QueueMessagingTemplate queueMessagingTemplate() { + return new QueueMessagingTemplate(this.amazonSqs); + } + + @Bean + @ServiceActivator(inputChannel = "sqsSendChannel") + public MessageHandler sqsMessageHandler() { + return new SqsMessageHandler(queueMessagingTemplate()); + } + +} +```` + +An XML variant may look like: + +````xml + + + +```` + +###Inbound Channel Adapter + +The SQS Inbound Channel Adapter is a `message-driven` implementation for the `MessageProducer` and is represented with +`SqsMessageDrivenChannelAdapter`. This channel adapter is based on the +`org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer` to receive messages from the +provided `queues` in async manner and send an enhanced Spring Integration Message to the provided `MessageChannel`. +The enhancements includes `AwsHeaders.MESSAGE_ID`, `AwsHeaders.RECEIPT_HANDLE` and `AwsHeaders.QUEUE` message headers. + +The Java Configuration is pretty simple: + +````java +@SpringBootApplication +public static class MyConfiguration { + + @Autowired + private AmazonSQSAsync amazonSqs; + + @Bean + public PollableChannel inputChannel() { + return new QueueChannel(); + } + + @Bean + public MessageProducer sqsMessageDrivenChannelAdapter() { + SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(this.amazonSqs, "myQueue"); + adapter.setOutputChannel(inputChannel()); + return adapter; + } +} +```` + +An XML variant may look like: + +````xml + + + +```` + +The `SqsMessageDrivenChannelAdapter` exposes all `SimpleMessageListenerContainer` attributes to configure and one an +important of them is `deleteMessageOnException`, which is `true` by default. Having that to `false`, it is a +responsibility of end-application to delete message or not on exceptions. E.g. in the error flow on the +`error-channel` of this channel adapter. For this purpose a `AwsHeaders.RECEIPT_HANDLE` message header must be used +for the message deletion: + +````java +MessageHeaders headers = message.getHeaders(); +this.amazonSqs.deleteMessageAsync( + new DeleteMessageRequest(headers.get(AwsHeaders.QUEUE), headers.get(AwsHeaders.RECEIPT_HANDLE))); +```` + +##Amazon Simple Notification Service (SNS) + +Amazon SNS is a publish-subscribe messaging system that allows clients to publish notification to a particular topic. +Other interested clients may subscribe using different protocols like HTTP/HTTPS, e-mail or an Amazon SQS queue to +receive the messages. Plus mobile devices can be registered as subscribers from the AWS Management Console. + +Unfortunately [Spring Cloud AWS][] doesn't provide flexible components which can be used from the channel adapter +implementations, but Amazon SNS API is pretty simple, from other side, hence Spring Integration AWS SNS Support is +straightforward and just allows to provide channel adapter foundation for Spring Integration applications. + +Since e-mail, SMS and mobile devices subscription/unsubscription confirmation is out of the Spring Integration +application scope and can be done only from the AWS Management Console, we provide only HTTP/HTTPS SNS endpoint in +face of `SnsInboundChannelAdapter`. The SQS-to-SNS subscription can be done with the simple usage of +`com.amazonaws.services.sns.util.Topics#subscribeQueue()`, which confirms subscription automatically. + +###Inbound Channel Adapter + +The `SnsInboundChannelAdapter` (``) is an extension of +`HttpRequestHandlingMessagingGateway` and must be as a part of Spring MVC application. Its URL must be used from the +AWS Management Console to add this endpoint as a subscriber to the SNS Topic. However before receiving any +notification itself this HTTP endpoint must confirm the subscription. + +See `SnsInboundChannelAdapter` JavaDocs for more information. + +An important option of this adapter to consider is `handleNotificationStatus`. This `boolean` flag indicates if the +adapter should send `SubscriptionConfirmation/UnsubscribeConfirmation` message to the `output-channel` or not. If +that the `AwsHeaders.NOTIFICATION_STATUS` message header is present in the message with the `NotificationStatus` +object, which can be used in the downstream flow to confirm subscription or not. Or "re-confirm" it in case of +`UnsubscribeConfirmation` message. + +In addition the `AwsHeaders#SNS_MESSAGE_TYPE` message header is represent to simplify a routing in the downstream flow. + +The Java Configuration is pretty simple: + +````java +@SpringBootApplication +public static class MyConfiguration { + + @Autowired + private AmazonSNS amazonSns; + + @Bean + public PollableChannel inputChannel() { + return new QueueChannel(); + } + + @Bean + public HttpRequestHandler sqsMessageDrivenChannelAdapter() { + SnsInboundChannelAdapter adapter = new SnsInboundChannelAdapter(amazonSns(), "/mySampleTopic"); + adapter.setRequestChannel(inputChannel()); + adapter.setHandleNotificationStatus(true); + return adapter; + } +} +```` + +An XML variant may look like: + +````xml + +``` + +Note: by default the message `payload` is a `Map` converted from the received Topic JSON message. For the convenient +the `payload-expression` is provided with the `Message` as a root object of the evaluation context. Hence even some +HTTP headers, populated by the `DefaultHttpHeaderMapper`, are available for the evaluation context. + [Spring Cloud AWS]: https://github.com/spring-cloud/spring-cloud-aws [AWS SDK for Java]: http://aws.amazon.com/sdkforjava/ [Amazon Web Services]: http://aws.amazon.com/ diff --git a/build.gradle b/build.gradle index 284cdd3..7f0ce3d 100644 --- a/build.gradle +++ b/build.gradle @@ -10,6 +10,7 @@ apply plugin: 'java' apply from: "${rootProject.projectDir}/publish-maven.gradle" apply plugin: 'eclipse' apply plugin: 'idea' +apply plugin: 'jacoco' group = 'org.springframework.integration' @@ -28,15 +29,15 @@ configurations { ext { commonsIoVersion='2.4' - jacocoVersion = '0.7.2.201409121644' - slf4jVersion = '1.7.8' + servletApiVersion = '3.1.0' + slf4jVersion = '1.7.12' springCloudAwsVersion = '1.0.3.RELEASE' - springIntegrationVersion = '4.1.6.RELEASE' + springIntegrationVersion = '4.2.0.RELEASE' idPrefix = 'aws' linkHomepage = 'https://github.com/spring-projects/spring-integration-aws' - linkCi = 'https://build.spring.io/browse/INTEXT-AWS' + linkCi = 'https://build.spring.io/browse/INTEXT' linkIssue = 'https://jira.spring.io/browse/INTEXT' linkScmUrl = 'https://github.com/spring-projects/spring-integration-aws' linkScmConnection = 'https://github.com/spring-projects/spring-integration-aws.git' @@ -47,19 +48,21 @@ ext.javadocLinks = [ "http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/" ] as String[] +jacoco { + toolVersion = "0.7.2.201409121644" +} dependencies { compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion" + compile ("org.springframework.integration:spring-integration-http:$springIntegrationVersion", optional) + compile("javax.servlet:javax.servlet-api:$servletApiVersion", provided) compile "org.springframework.cloud:spring-cloud-aws-messaging:$springCloudAwsVersion" compile "commons-io:commons-io:$commonsIoVersion" 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' @@ -102,9 +105,22 @@ ext.xLintArg = '-Xlint:all,-options' test { // suppress all console output during testing unless running `gradle -i` logging.captureStandardOutput(LogLevel.INFO) - jvmArgs "-javaagent:${configurations.jacoco.asPath}=destfile=${buildDir}/jacoco.exec,includes=*" + maxHeapSize = "1024m" + jacoco { + append = false + destinationFile = file("$buildDir/jacoco.exec") + } } +jacocoTestReport { + reports { + xml.enabled false + csv.enabled false + html.destination "${buildDir}/reports/jacoco/html" + } +} + +build.dependsOn jacocoTestReport task sourcesJar(type: Jar) { classifier = 'sources' from sourceSets.main.allJava @@ -247,6 +263,6 @@ task dist(dependsOn: assemble) { task wrapper(type: Wrapper) { description = 'Generates gradlew[.bat] scripts' - gradleVersion = '2.3' + gradleVersion = '2.5' 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 085a1cd..30d399d 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 81c2b60..741e43d 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,6 +1,6 @@ -#Mon Feb 16 15:46:30 EET 2015 +#Wed Sep 23 15:22:04 EDT 2015 distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists -distributionUrl=http\://services.gradle.org/distributions/gradle-2.3-all.zip +distributionUrl=http\://services.gradle.org/distributions/gradle-2.5-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 b7974dc..7723bef 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 @@ -31,11 +31,11 @@ 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()); + registerBeanDefinitionParser("s3-outbound-channel-adapter", new AmazonS3OutboundChannelAdapterParser()); + registerBeanDefinitionParser("s3-inbound-channel-adapter", new AmazonS3InboundChannelAdapterParser()); + registerBeanDefinitionParser("sqs-outbound-channel-adapter", new SqsOutboundChannelAdapterParser()); + registerBeanDefinitionParser("sqs-message-driven-channel-adapter", new SqsMessageDrivenChannelAdapterParser()); + registerBeanDefinitionParser("sns-inbound-channel-adapter", new SnsInboundChannelAdapterParser()); } } 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 4370c8f..b748fd0 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 @@ -46,6 +46,8 @@ public final class AmazonWSParserUtils { public static final String SQS_REF = "sqs"; + public static final String SNS_REF = "sns"; + public static final String RESOURCE_ID_RESOLVER_REF = "resource-id-resolver"; private AmazonWSParserUtils() { diff --git a/src/main/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParser.java b/src/main/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParser.java new file mode 100644 index 0000000..b25374b --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParser.java @@ -0,0 +1,81 @@ +/* + * 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.config.BeanDefinition; +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.SnsInboundChannelAdapter; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; + +/** + * The parser for the {@code } + * + * @author Artem Bilan + */ +public class SnsInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser { + + @Override + protected Class getBeanClass(Element element) { + return SnsInboundChannelAdapter.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.SNS_REF)) + .addConstructorArgValue(element.getAttribute("path")); + String channelName = element.getAttribute("channel"); + if (!StringUtils.hasText(channelName)) { + channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext); + } + builder.addPropertyReference("requestChannel", channelName); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "handle-notification-status"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout", "requestTimeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE); + BeanDefinition payloadExpressionDef = + IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("payload-expression", element); + if (payloadExpressionDef != null) { + builder.addPropertyValue("payloadExpression", payloadExpressionDef); + } + } + +} diff --git a/src/main/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapter.java new file mode 100644 index 0000000..76b28ea --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapter.java @@ -0,0 +1,203 @@ +/* + * 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.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.springframework.cloud.aws.messaging.endpoint.NotificationStatus; +import org.springframework.cloud.aws.messaging.endpoint.NotificationStatusHandlerMethodArgumentResolver; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.http.inbound.HttpRequestHandlingMessagingGateway; +import org.springframework.integration.http.inbound.RequestMapping; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.web.multipart.MultipartResolver; + +import com.amazonaws.services.sns.AmazonSNS; + +/** + * The {@link HttpRequestHandlingMessagingGateway} extension for the Amazon WS SNS HTTP(S) endpoints. + * Accepts all {@code x-amz-sns-message-type}s, converts the received Topic JSON message to the + * {@link Map} using {@link MappingJackson2HttpMessageConverter} and send it to the provided + * {@link #requestChannel} as {@link Message} {@code payload}. + *

+ * The mapped url must be configured inside the Amazon Web Service platform as a subscription. + * Before receiving any notification itself this HTTP endpoint must confirm the subscription. + *

+ * The {@link #handleNotificationStatus} flag (defaults to {@code false}) indicates that + * this endpoint should send the {@code SubscriptionConfirmation/UnsubscribeConfirmation} + * messages to the the provided {@link #requestChannel}. If that, the {@link AwsHeaders#NOTIFICATION_STATUS} + * header is populated with the {@link NotificationStatus} value. In that case it is a responsibility of + * the application to {@link NotificationStatus#confirmSubscription()} or not. + *

+ * By default this endpoint just does {@link NotificationStatus#confirmSubscription()} + * for the {@code SubscriptionConfirmation} message type. + * And does nothing for the {@code UnsubscribeConfirmation}. + *

+ * For the convenience on the underlying message flow routing a {@link AwsHeaders#SNS_MESSAGE_TYPE} + * header is present. + * + * @author Artem Bilan + */ +public class SnsInboundChannelAdapter extends HttpRequestHandlingMessagingGateway { + + private final NotificationStatusResolver notificationStatusResolver; + + private volatile boolean handleNotificationStatus; + + private volatile Expression payloadExpression; + + private EvaluationContext evaluationContext; + + public SnsInboundChannelAdapter(AmazonSNS amazonSns, String... path) { + super(false); + Assert.notNull(amazonSns, "'amazonSns' must not be null."); + Assert.notNull(path, "'path' must not be null."); + Assert.noNullElements(path, "'path' must not contain null elements."); + this.notificationStatusResolver = new NotificationStatusResolver(amazonSns); + RequestMapping requestMapping = new RequestMapping(); + requestMapping.setMethods(HttpMethod.POST); + requestMapping.setHeaders("x-amz-sns-message-type"); + requestMapping.setPathPatterns(path); + super.setRequestMapping(requestMapping); + super.setStatusCodeExpression(new ValueExpression<>(HttpStatus.NO_CONTENT)); + super.setMessageConverters( + Collections.>singletonList(new MappingJackson2HttpMessageConverter())); + super.setRequestPayloadType(HashMap.class); + } + + public void setHandleNotificationStatus(boolean handleNotificationStatus) { + this.handleNotificationStatus = handleNotificationStatus; + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + if (this.payloadExpression != null) { + this.evaluationContext = createEvaluationContext(); + } + } + + @Override + @SuppressWarnings("unchecked") + protected void send(Object object) { + Message message = (Message) object; + HashMap payload = (HashMap) message.getPayload(); + AbstractIntegrationMessageBuilder messageToSendBuilder; + if (this.payloadExpression != null) { + messageToSendBuilder = getMessageBuilderFactory() + .withPayload(this.payloadExpression.getValue(this.evaluationContext, message)) + .copyHeaders(message.getHeaders()); + } + else { + messageToSendBuilder = getMessageBuilderFactory().fromMessage(message); + } + + String type = payload.get("Type"); + if ("SubscriptionConfirmation".equals(type) || "UnsubscribeConfirmation".equals(type)) { + NotificationStatus notificationStatus = this.notificationStatusResolver.resolveNotificationStatus(payload); + if (this.handleNotificationStatus) { + messageToSendBuilder.setHeader(AwsHeaders.NOTIFICATION_STATUS, notificationStatus); + } + else { + if ("SubscriptionConfirmation".equals(type)) { + notificationStatus.confirmSubscription(); + } + return; + } + } + messageToSendBuilder.setHeader(AwsHeaders.SNS_MESSAGE_TYPE, type) + .setHeader(AwsHeaders.MESSAGE_ID, payload.get("MessageId")); + super.send(messageToSendBuilder.build()); + } + + @Override + public void setPayloadExpression(Expression payloadExpression) { + this.payloadExpression = payloadExpression; + } + + @Override + public void setHeaderExpressions(Map headerExpressions) { + throw new UnsupportedOperationException(); + } + + @Override + public void setMessageConverters(List> messageConverters) { + throw new UnsupportedOperationException(); + } + + @Override + public void setMergeWithDefaultConverters(boolean mergeWithDefaultConverters) { + throw new UnsupportedOperationException(); + } + + @Override + public void setHeaderMapper(HeaderMapper headerMapper) { + throw new UnsupportedOperationException(); + } + + @Override + public void setRequestMapping(RequestMapping requestMapping) { + throw new UnsupportedOperationException(); + } + + @Override + public void setRequestPayloadType(Class requestPayloadType) { + throw new UnsupportedOperationException(); + } + + @Override + public void setExtractReplyPayload(boolean extractReplyPayload) { + throw new UnsupportedOperationException(); + } + + @Override + public void setMultipartResolver(MultipartResolver multipartResolver) { + throw new UnsupportedOperationException(); + } + + @Override + public void setStatusCodeExpression(Expression statusCodeExpression) { + throw new UnsupportedOperationException(); + } + + private static class NotificationStatusResolver extends NotificationStatusHandlerMethodArgumentResolver { + + public NotificationStatusResolver(AmazonSNS amazonSns) { + super(amazonSns); + } + + protected NotificationStatus resolveNotificationStatus(HashMap content) { + return (NotificationStatus) super.doResolverArgumentFromNotificationMessage(content); + } + + } + +} diff --git a/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java index 6126de0..c6b6236 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java @@ -22,7 +22,7 @@ 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.context.IntegrationContextUtils; +import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -74,7 +74,7 @@ public class SqsMessageHandler extends AbstractMessageHandler { @Override protected void onInit() throws Exception { super.onInit(); - this.evaluationContext = IntegrationContextUtils.getEvaluationContext(getBeanFactory()); + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); } @Override @@ -90,4 +90,3 @@ public class SqsMessageHandler extends AbstractMessageHandler { } } - diff --git a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java index 7ee6b0e..556664c 100644 --- a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java +++ b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java @@ -29,4 +29,8 @@ public abstract class AwsHeaders { public static final String RECEIPT_HANDLE = PREFIX + "receiptHandle"; + public static final String NOTIFICATION_STATUS = PREFIX + "notificationStatus"; + + public static final String SNS_MESSAGE_TYPE = PREFIX + "snsMessageType"; + } 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 5b92d2a..1b3f0f0 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 @@ -522,6 +522,80 @@ + + + + Defines an SNS inbound HTTP-based Channel Adapter - SnsInboundChannelAdapter. + + + + + + + + The 'com.amazonaws.services.sns.AmazonSNS' bean reference. + + + + + + + + + + + + Comma-separated URI paths (e.g., /orderId/{order}). + Ant-style path patterns are also supported (e.g. /myPath/*.do). + + + + + + + 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. + + + + + + + Allows you to specify SpEL expression to construct a Message payload. + The root evaluation object is a raw Message as a result of the 'HttpServletRequest' + conversion in the 'HttpRequestHandlingEndpointSupport' super class. + + + + + + + + + + + + The MessagingGateway's 'error-channel' where to send an ErrorMessage in case + of Exception is caused from original message flow. + + + + + + + Flag to indicate if the 'SubscriptionConfirmation' and 'UnsubscribeConfirmation' + SNS messages should sent to the 'channel' or not. If 'true' the + 'AwsHeaders.NOTIFICATION_STATUS' message header is populated with the 'NotificationStatus' + value. In this case it is an application responsibility to 'confirm' subscription or not using + that 'NotificationStatus' object. Defaults to 'false'. + + + + + + @@ -536,8 +610,7 @@ - + @@ -549,8 +622,7 @@ - + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests-context.xml b/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000..3f711fc --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests-context.xml @@ -0,0 +1,23 @@ + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java new file mode 100644 index 0000000..1a8c9c9 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java @@ -0,0 +1,80 @@ +/* + * 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.assertArrayEquals; +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.beans.factory.annotation.Qualifier; +import org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer; +import org.springframework.integration.aws.inbound.SnsInboundChannelAdapter; +import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.amazonaws.services.sns.AmazonSNS; + +/** + * @author Artem Bilan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class SnsInboundChannelAdapterParserTests { + + @Autowired + private AmazonSNS amazonSns; + + @Autowired + private MessageChannel errorChannel; + + @Autowired + private NullChannel nullChannel; + + @Autowired + @Qualifier("snsInboundChannelAdapter") + private SnsInboundChannelAdapter snsInboundChannelAdapter; + + + @Test + public void testSqsMessageDrivenChannelAdapterParser() { + assertSame(this.amazonSns, TestUtils.getPropertyValue(this.snsInboundChannelAdapter, + "notificationStatusResolver.amazonSns")); + assertTrue(TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "handleNotificationStatus", Boolean.class)); + assertArrayEquals(new String[] {"/foo"}, + TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "requestMapping.pathPatterns", String[].class)); + assertEquals("payload.Message", + TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "payloadExpression.expression")); + assertFalse(this.snsInboundChannelAdapter.isRunning()); + assertEquals(100, this.snsInboundChannelAdapter.getPhase()); + assertFalse(this.snsInboundChannelAdapter.isAutoStartup()); + assertSame(this.errorChannel, TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "requestChannel")); + assertSame(this.nullChannel, TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "errorChannel")); + assertEquals(2000L, TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "messagingTemplate.sendTimeout")); + } + +} 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 index 6761df8..31a839a 100644 --- 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 @@ -5,7 +5,7 @@ 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"> + http://www.springframework.org/schema/cloud/aws/messaging http://www.springframework.org/schema/cloud/aws/messaging/spring-cloud-aws-messaging.xsd"> 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 index cc69062..4c84db0 100644 --- 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 @@ -5,7 +5,7 @@ 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"> + http://www.springframework.org/schema/cloud/aws/messaging http://www.springframework.org/schema/cloud/aws/messaging/spring-cloud-aws-messaging.xsd"> diff --git a/src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java new file mode 100644 index 0000000..40bc528 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapterTests.java @@ -0,0 +1,181 @@ +/* + * 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.assertTrue; +import static org.mockito.Mockito.verify; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mockito; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.cloud.aws.messaging.endpoint.NotificationStatus; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.Resource; +import org.springframework.http.MediaType; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; +import org.springframework.util.StreamUtils; +import org.springframework.web.HttpRequestHandler; +import org.springframework.web.context.WebApplicationContext; + +import com.amazonaws.services.sns.AmazonSNS; + +/** + * @author Artem Bilan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +@WebAppConfiguration +@DirtiesContext +public class SnsInboundChannelAdapterTests { + + @Autowired + private WebApplicationContext context; + + @Autowired + private AmazonSNS amazonSns; + + @Autowired + private PollableChannel inputChannel; + + @Value("classpath:org/springframework/integration/aws/inbound/subscriptionConfirmation.json") + private Resource subscriptionConfirmation; + + @Value("classpath:org/springframework/integration/aws/inbound/notificationMessage.json") + private Resource notificationMessage; + + @Value("classpath:org/springframework/integration/aws/inbound/unsubscribeConfirmation.json") + private Resource unsubscribeConfirmation; + + private MockMvc mockMvc; + + @Before + public void setUp() throws Exception { + this.mockMvc = MockMvcBuilders.webAppContextSetup(this.context).build(); + } + + @Test + public void testSubscriptionConfirmation() throws Exception { + this.mockMvc.perform( + post("/mySampleTopic") + .header("x-amz-sns-message-type", "SubscriptionConfirmation") + .contentType(MediaType.APPLICATION_JSON) + .content(StreamUtils.copyToByteArray(this.subscriptionConfirmation.getInputStream()))) + .andExpect(status().isNoContent()); + + Message receive = inputChannel.receive(10000); + assertNotNull(receive); + assertTrue(receive.getHeaders().containsKey(AwsHeaders.SNS_MESSAGE_TYPE)); + assertEquals("SubscriptionConfirmation", receive.getHeaders().get(AwsHeaders.SNS_MESSAGE_TYPE)); + + assertTrue(receive.getHeaders().containsKey(AwsHeaders.NOTIFICATION_STATUS)); + NotificationStatus notificationStatus = (NotificationStatus) receive.getHeaders() + .get(AwsHeaders.NOTIFICATION_STATUS); + + notificationStatus.confirmSubscription(); + + verify(this.amazonSns) + .confirmSubscription("arn:aws:sns:eu-west-1:111111111111:mySampleTopic", "111"); + } + + @Test + @SuppressWarnings("unchecked") + public void testNotification() throws Exception { + this.mockMvc.perform( + post("/mySampleTopic") + .header("x-amz-sns-message-type", "Notification") + .contentType(MediaType.APPLICATION_JSON) + .content(StreamUtils.copyToByteArray(this.notificationMessage.getInputStream()))) + .andExpect(status().isNoContent()); + + Message receive = inputChannel.receive(10000); + assertNotNull(receive); + Map payload = (Map) receive.getPayload(); + + assertEquals("foo", payload.get("Subject")); + assertEquals("bar", payload.get("Message")); + } + + @Test + public void testUnsubscribe() throws Exception { + this.mockMvc.perform( + post("/mySampleTopic") + .header("x-amz-sns-message-type", "UnsubscribeConfirmation") + .contentType(MediaType.APPLICATION_JSON) + .content(StreamUtils.copyToByteArray(this.unsubscribeConfirmation.getInputStream()))) + .andExpect(status().isNoContent()); + + Message receive = inputChannel.receive(10000); + assertNotNull(receive); + assertTrue(receive.getHeaders().containsKey(AwsHeaders.SNS_MESSAGE_TYPE)); + assertEquals("UnsubscribeConfirmation", receive.getHeaders().get(AwsHeaders.SNS_MESSAGE_TYPE)); + + assertTrue(receive.getHeaders().containsKey(AwsHeaders.NOTIFICATION_STATUS)); + NotificationStatus notificationStatus = (NotificationStatus) receive.getHeaders() + .get(AwsHeaders.NOTIFICATION_STATUS); + + notificationStatus.confirmSubscription(); + + verify(this.amazonSns) + .confirmSubscription("arn:aws:sns:eu-west-1:111111111111:mySampleTopic", "233"); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + public AmazonSNS amazonSns() { + return Mockito.mock(AmazonSNS.class); + } + + @Bean + public PollableChannel inputChannel() { + return new QueueChannel(); + } + + @Bean + public HttpRequestHandler sqsMessageDrivenChannelAdapter() { + SnsInboundChannelAdapter adapter = new SnsInboundChannelAdapter(amazonSns(), "/mySampleTopic"); + adapter.setRequestChannel(inputChannel()); + adapter.setHandleNotificationStatus(true); + return adapter; + } + + } + +} diff --git a/src/test/java/org/springframework/integration/aws/inbound/notificationMessage.json b/src/test/java/org/springframework/integration/aws/inbound/notificationMessage.json new file mode 100644 index 0000000..dd5c48c --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/inbound/notificationMessage.json @@ -0,0 +1,26 @@ +{ + "Type": "Notification", + "MessageId": "f2c15fec-c617-5b08-b54d-13c4099fec60", + "TopicArn": "arn:aws:sns:eu-west-1:111111111111:mySampleTopic", + "Subject": "foo", + "Message": "bar", + "Timestamp": "2014-06-28T14:12:24.418Z", + "SignatureVersion": "1", + "Signature": "XDvKSAnhxECrAmyIrs0Dsfbp/tnKD1IvoOOYTU28FtbUoxr/CgziuW87yZwTuSNNbHJbdD3BEjHS0vKewm0xBeQ0PToDkgtoORXo5RWnmShDQ2nhkthFhZnNulKtmFtRogjBtCwbz8sPnbOCSk21ruyXNdV2RUbdDalndAW002CWEQmYMxFSN6OXUtMueuT610aX+tqeYP4Z6+8WTWLWjAuVyy7rOI6KHYBcVDhKtskvTOPZ4tiVohtQdQbO2Gjuh1vblRzzwMkfaoFTSWImd4pFXxEsv/fq9aGIlqq9xEryJ0w2huFwI5gxyhvGt0RnTd9YvmAEC+WzdJDOqaDNxg==", + "SigningCertURL": "https://sns.eu-west-1.amazonaws.com/SimpleNotificationService-e372f8ca30337fdb084e8ac449342c77.pem", + "UnsubscribeURL": "https://sns.eu-west-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:eu-west-1:721324560415:mySampleTopic:9859a6c9-6083-4690-ab02-d1aead3442df", + "MessageAttributes": { + "AWS.SNS.MOBILE.MPNS.Type": { + "Type": "String", + "Value": "token" + }, + "AWS.SNS.MOBILE.WNS.Type": { + "Type": "String", + "Value": "wns/badge" + }, + "AWS.SNS.MOBILE.MPNS.NotificationClass": { + "Type": "String", + "Value": "realtime" + } + } +} diff --git a/src/test/java/org/springframework/integration/aws/inbound/subscriptionConfirmation.json b/src/test/java/org/springframework/integration/aws/inbound/subscriptionConfirmation.json new file mode 100644 index 0000000..8869637 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/inbound/subscriptionConfirmation.json @@ -0,0 +1,12 @@ +{ + "Type": "SubscriptionConfirmation", + "MessageId": "e267b24c-5532-472f-889d-c2cdd2143bbc", + "Token": "111", + "TopicArn": "arn:aws:sns:eu-west-1:111111111111:mySampleTopic", + "Message": "You have chosen to subscribe to the topic arn:aws:sns:eu-west-1:721324560415:mySampleTopic.To confirm the subscription, visit the SubscribeURL included in this message.", + "SubscribeURL": "https://sns.eu-west-1.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:eu-west-1:111111111111:mySampleTopic&Token=111", + "Timestamp": "2014-06-28T10:22:18.086Z", + "SignatureVersion": "1", + "Signature": "JLdRUR+uhP4cyVW6bRuUSAkUosFMJyO7g7WCAwEUJoB4y8vQE1uDUWGpbQSEbruVTjPEM8hFsf4/95NftfM0W5IgND1uSnv4P/4AYyL+q0bLOJlquzXrw4w2NX3QShS3y+r/gXzo7p/UP4NOr35MGCEGPqHAEe1Coc5S0eaP3JvKU6xY1tcop6ze2RNHTwzhM43dda2bnjPYogAJzA5uHfmSjs3cMVvPCckj3zdLyvxISp+RgrogdvlNyu9ycND1SxagmbzjkBaqvF/4aiSYFxsEXX4e9zuNuHGmXGWgm1ppYUGLSPPJruCsPUa7Ii1mYvpX7SezuFZlAAXXBk0mHg==", + "SigningCertURL": "https://sns.eu-west-1.amazonaws.com/SimpleNotificationService-e372f8ca30337fdb084e8ac449342c77.pem" +} diff --git a/src/test/java/org/springframework/integration/aws/inbound/unsubscribeConfirmation.json b/src/test/java/org/springframework/integration/aws/inbound/unsubscribeConfirmation.json new file mode 100644 index 0000000..36317b5 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/inbound/unsubscribeConfirmation.json @@ -0,0 +1,12 @@ +{ + "Type": "UnsubscribeConfirmation", + "MessageId": "7b9a6321-45d5-461a-bc35-9c2d18ed9dbe", + "Token": "233", + "TopicArn": "arn:aws:sns:eu-west-1:111111111111:mySampleTopic", + "Message": "You have chosen to deactivate subscription arn:aws:sns:eu-west-1:111111111111:mySampleTopic:f64111de-e681-4820-a8be-474f64c1bbf8.\nTo cancel this operation and restore the subscription, visit the SubscribeURL included in this message.", + "SubscribeURL": "https://sns.eu-west-1.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:eu-west-1:721324560415:mySampleTopic&Token=233", + "Timestamp": "2014-06-28T19:33:00.497Z", + "SignatureVersion": "1", + "Signature": "EAb0k1XoD2h+u4j2GB42wEFCVUFKEaS/2W+p+3wK1GfZDZn4LeDLbkxJ1LYF/A1CYL+sCJ4ZmLFI1axm0V/p+fESkNbKQoQotMgma+PtA6KnmRrKEU8O6nUELqeVPWFAoQ9ZsW9FCAXVDXoPxiqHNTH+tC7mzAsvajyp4aTm/POqkRKBl+A/7dHUqfHGup/FJhLNgTAciBZSloa5EuBKxInJQfoZjy3DU8qKXXhmKKRdyVwOGEuReo/njy4c3Phtn0+logu2PZUKqkGTuJZVbapHmcTq+0MqIh05sevLDmTEfBlsmNThhWIyCza/t68RRlqm9cRLjINSWfrv1Xkrpw==", + "SigningCertURL": "https://sns.eu-west-1.amazonaws.com/SimpleNotificationService-e372f8ca30337fdb084e8ac449342c77.pem" +}