INTEXT-147: Add SnsInboundChannelAdapter
JIRA: https://jira.spring.io/browse/INTEXT-147 INTEXT-147: Documentation Upgrade dependencies
This commit is contained in:
Artem Bilan
181
README.md
181
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 `<int-mail:outbound-channel-adapter>`.
|
||||
|
||||
##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
|
||||
(`<int-aws:sqs-outbound-channel-adapter>`) 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
|
||||
<aws-messaging:sqs-async-client id="sqs"/>
|
||||
|
||||
<int-aws:sqs-outbound-channel-adapter sqs="sqs"
|
||||
channel="sqsSendChannel"
|
||||
queue="foo"/>
|
||||
````
|
||||
|
||||
###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
|
||||
<aws-messaging:sqs-async-client id="sqs"/>
|
||||
|
||||
<int-aws:sqs-message-driven-channel-adapter id="sqsInboundChannel"
|
||||
sqs="sqs"
|
||||
error-channel="myErrorChannel"
|
||||
queues="foo, bar"
|
||||
delete-message-on-exception="false"
|
||||
max-number-of-messages="5"
|
||||
visibility-timeout="200"
|
||||
wait-time-out="40"
|
||||
send-timeout="2000"/>
|
||||
````
|
||||
|
||||
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` (`<int-aws:sns-inbound-channel-adapter>`) 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
|
||||
<int-aws:sns-inbound-channel-adapter sns="amazonSns"
|
||||
path="/foo"
|
||||
channel="snsChannel"
|
||||
error-channel="errorChannel"
|
||||
handle-notification-status="true"
|
||||
payload-expression="payload.Message"/>
|
||||
```
|
||||
|
||||
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/
|
||||
|
||||
34
build.gradle
34
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"
|
||||
}
|
||||
|
||||
BIN
gradle/wrapper/gradle-wrapper.jar
vendored
BIN
gradle/wrapper/gradle-wrapper.jar
vendored
Binary file not shown.
4
gradle/wrapper/gradle-wrapper.properties
vendored
4
gradle/wrapper/gradle-wrapper.properties
vendored
@@ -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
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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 <int-aws:sns-inbound-channel-adapter>}
|
||||
*
|
||||
* @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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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}.
|
||||
* <p>
|
||||
* 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.
|
||||
* <p>
|
||||
* 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.
|
||||
* <p>
|
||||
* By default this endpoint just does {@link NotificationStatus#confirmSubscription()}
|
||||
* for the {@code SubscriptionConfirmation} message type.
|
||||
* And does nothing for the {@code UnsubscribeConfirmation}.
|
||||
* <p>
|
||||
* 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.<HttpMessageConverter<?>>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<String, String> payload = (HashMap<String, String>) 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<String, Expression> headerExpressions) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setMessageConverters(List<HttpMessageConverter<?>> messageConverters) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setMergeWithDefaultConverters(boolean mergeWithDefaultConverters) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setHeaderMapper(HeaderMapper<HttpHeaders> 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<String, String> content) {
|
||||
return (NotificationStatus) super.doResolverArgumentFromNotificationMessage(content);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -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";
|
||||
|
||||
}
|
||||
|
||||
@@ -522,6 +522,80 @@
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="sns-inbound-channel-adapter">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines an SNS inbound HTTP-based Channel Adapter - SnsInboundChannelAdapter.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:complexType>
|
||||
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
|
||||
<xsd:attribute name="sns" use="required">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
The 'com.amazonaws.services.sns.AmazonSNS' bean reference.
|
||||
</xsd:documentation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type type="com.amazonaws.services.sns.AmazonSNS"/>
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="path" type="xsd:string" use="required">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Comma-separated URI paths (e.g., /orderId/{order}).
|
||||
Ant-style path patterns are also supported (e.g. /myPath/*.do).
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="send-timeout" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Maximum amount of time in milliseconds to wait when sending
|
||||
a message to the channel if such channel may block.
|
||||
For example, a Queue Channel can block until space
|
||||
is available if its maximum capacity has been reached.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="payload-expression" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
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.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="error-channel" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type type="org.springframework.messaging.MessageChannel" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
<xsd:documentation>
|
||||
The MessagingGateway's 'error-channel' where to send an ErrorMessage in case
|
||||
of Exception is caused from original message flow.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="handle-notification-status" default="false">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
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'.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:complexType name="baseSqsAdapterType">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
@@ -536,8 +610,7 @@
|
||||
</xsd:documentation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type
|
||||
type="com.amazonaws.services.sqs.AmazonSQS"/>
|
||||
<tool:expected-type type="com.amazonaws.services.sqs.AmazonSQS"/>
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
@@ -549,8 +622,7 @@
|
||||
</xsd:documentation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type
|
||||
type="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
|
||||
<tool:expected-type type="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
|
||||
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">
|
||||
|
||||
<bean id="amazonSns" class="org.mockito.Mockito" factory-method="mock">
|
||||
<constructor-arg value="com.amazonaws.services.sns.AmazonSNS"/>
|
||||
</bean>
|
||||
|
||||
<int-aws:sns-inbound-channel-adapter sns="amazonSns"
|
||||
path="/foo"
|
||||
auto-startup="false"
|
||||
channel="errorChannel"
|
||||
error-channel="nullChannel"
|
||||
phase="100"
|
||||
id="snsInboundChannelAdapter"
|
||||
send-timeout="2000"
|
||||
handle-notification-status="true"
|
||||
payload-expression="payload.Message"/>
|
||||
|
||||
</beans>
|
||||
@@ -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"));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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">
|
||||
|
||||
<aws-messaging:sqs-async-client id="sqs"/>
|
||||
|
||||
|
||||
@@ -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">
|
||||
|
||||
<aws-messaging:sqs-async-client id="sqs"/>
|
||||
|
||||
|
||||
@@ -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<String, String> payload = (Map<String, String>) 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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
Reference in New Issue
Block a user