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

INTEXT-8: Add SNS Outbound Channel Adapter

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

Address PR comments
This commit is contained in:
Artem Bilan
2015-09-28 14:38:33 -04:00
parent dc66af65eb
commit 0580642439
17 changed files with 1056 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
README.md
View File

@@ -5,13 +5,13 @@ Spring Integration Extension for Amazon Web Services (AWS)
## Amazon Web Services (AWS)
Launched in 2006, [Amazon Web Services][] (AWS) provides key infrastructure services for business through
its cloud computing platform. Using cloud computing businesses can adopt a new business model whereby they
do not have to plan and invest in procuring their own IT infrastructure. They can use the infrastructure and services
provided by the cloud service provider and pay as they use the services. Visit [http://aws.amazon.com/products/]
Launched in 2006, [Amazon Web Services][] (AWS) provides key infrastructure services for business through
its cloud computing platform. Using cloud computing businesses can adopt a new business model whereby they
do not have to plan and invest in procuring their own IT infrastructure. They can use the infrastructure and services
provided by the cloud service provider and pay as they use the services. Visit [http://aws.amazon.com/products/]
for more details about various products offered by Amazon as a part their cloud computing services.
*Spring Integration Extension for Amazon Web Services* provides Spring Integration adapters for the various services
*Spring Integration Extension for Amazon Web Services* provides Spring Integration adapters for the various services
provided by the [AWS SDK for Java][].
Note the Spring Integration AWS Extension is based on the [Spring Cloud AWS][] project.
@@ -26,34 +26,34 @@ This guide intends to explain briefly the various adapters available for [Amazon
* **Amazon DynamoDB** (Analysis ongoing)
* **Amazon SimpleDB** (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
and outbound adapters. The *SQS* inbound adapter is capable of receiving notifications sent out from *SNS*
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
and outbound adapters. The *SQS* inbound adapter is capable of receiving notifications sent out from *SNS*
where the topic is an *SQS* Queue.
For *DymamoDB* and *SimpleDB*, besides providing *Inbound*- and *Outbound Adapters*,
For *DymamoDB* and *SimpleDB*, besides providing *Inbound*- and *Outbound Adapters*,
a *MessageStore* implementation is provided, too.
# Executing the test cases.
All test cases for the adapters are present in the *src/test/java* folder. On executing the build, maven surefire
All test cases for the adapters are present in the *src/test/java* folder. On executing the build, maven surefire
plugin will execute all the tests.
> Please note that all the tests ending with **AWSTests.java* connect to the actual [Amazon Web Services][]
and are excluded by default in the maven build. All other tests rely on mocking to test the functionality.
> Please note that all the tests ending with **AWSTests.java* connect to the actual [Amazon Web Services][]
and are excluded by default in the maven build. All other tests rely on mocking to test the functionality.
You need to execute the **AWSTests.java* manually to test the connectivity to AWS using your credentials.
All these **AWSTests.java* tests look for the file *awscredentials.properties* in the classpath.
All these **AWSTests.java* tests look for the file *awscredentials.properties* in the classpath.
To be on the safe side, create the following file at *src/test/resources*:
*spring-integration-aws/src/test/resources/awscredentials.properties*. It is added to the *.gitignore* file by default.
*spring-integration-aws/src/test/resources/awscredentials.properties*. It is added to the *.gitignore* file by default.
This will prevent this file to be checked in accidentally and revealing your credentials.
This file needs to have two properties *accessKey* and *secretKey*, holding the values of your access key
This file needs to have two properties *accessKey* and *secretKey*, holding the values of your access key
and secret key respectively.
> **Note: AWS Services are chargeable and we recommend not to execute the **AWSTests.java* as part
of your regular builds. AWS does provide a free tier which is sufficient to perform your tests without being charged
(not true for DynamoDB though), however keep a check on your account usage regularly.
> **Note: AWS Services are chargeable and we recommend not to execute the **AWSTests.java* as part
of your regular builds. AWS does provide a free tier which is sufficient to perform your tests without being charged
(not true for DynamoDB though), however keep a check on your account usage regularly.
Get more information about AWS free tier at [http://aws.amazon.com/free/][]**
#Adapters
@@ -69,22 +69,22 @@ Get more information about AWS free tier at [http://aws.amazon.com/free/][]**
There is no adapter for SES, since [Spring Cloud AWS][] provides implementations for
`org.springframework.mail.MailSender` - `SimpleEmailServiceMailSender` and `SimpleEmailServiceJavaMailSender`, which
can be injected to the `<int-mail:outbound-channel-adapter>`.
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
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 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
@@ -120,9 +120,9 @@ An XML variant may look like:
###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
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.
@@ -165,12 +165,12 @@ An XML variant may look like:
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
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(
@@ -179,34 +179,34 @@ this.amazonSqs.deleteMessageAsync(
##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.
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
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.
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
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
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
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:
@@ -222,7 +222,7 @@ public static class MyConfiguration {
public PollableChannel inputChannel() {
return new QueueChannel();
}
@Bean
public HttpRequestHandler sqsMessageDrivenChannelAdapter() {
SnsInboundChannelAdapter adapter = new SnsInboundChannelAdapter(amazonSns(), "/mySampleTopic");
@@ -245,8 +245,75 @@ An XML variant may look like:
```
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.
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.
###Outbound Channel Adapter
The `SnsMessageHandler` (`<int-aws:sns-outbound-channel-adapter>`) is a simple one-way Outbound Channel Adapter
to send Topic Notification using `AmasonSNS` service.
This Channel Adapter (`MessageHandler`) accepts these options:
- `topic-arn` (`topic-arn-expression`) - the SNS Topic to send notification for. The `ResourceIdResolver` can be used
from the SpEL definition to determine the target Topic Arn from the local logical name;
- `subject` (`subject-expression`) - the SNS Notification Subject;
- `body-expression` - the SpEL expression to evaluate the `message` property for the
`com.amazonaws.services.sns.model.PublishRequest`.
See `SnsMessageHandler` JavaDocs for more information.
The Java Config looks like:
````java
@Bean
public MessageHandler snsMessageHandler() {
SnsMessageHandler handler = new SnsMessageHandler(amazonSns());
adapter.setTopicArn("arn:aws:sns:eu-west:123456789012:test);
String bodyExpression = "SnsBodyBuilder.withDefault(payload).forProtocols(payload.substring(0, 140), 'sms')";
handler.setBodyExpression(spelExpressionParser.parseExpression(bodyExpression));
return handler;
}
````
NOTE: the `bodyExpression` can be evaluated to a `org.springframework.integration.aws.support.SnsBodyBuilder`
allowing the configuration of a `json` `messageStructure` for the `PublishRequest` and provide separate messages
for different protocols.
The same `SnsBodyBuilder` rule is applied for the raw `payload` if the `bodyExpression` hasn't been configured.
NOTE: if the `payload` of `requestMessage` is a `com.amazonaws.services.sns.model.PublishRequest` already,
the `SnsMessageHandler` doesn't do anything with it and it is sent as-is.
The XML variant may look like:
````xml
<int-aws:sns-outbound-channel-adapter
id="snsAdapter"
sns="amazonSns"
channel="notificationChannel"
topic-arn="foo"
subject="bar"
body-expression="payload.toUpperCase()"/>
````
###Outbound Gateway
The `<int-aws:sns-outbound-gateway>` is fully similar to the one-way channel adapter.
The only difference that in gateway mode the `SnsMessageHandler` produces the reply `Message` as:
````java
return getMessageBuilderFactory()
.withPayload(publishRequest)
.setHeader(AwsHeaders.TOPIC, publishRequest.getTopicArn())
.setHeader(AwsHeaders.SNS_PUBLISHED_MESSAGE_ID, publishResult.getMessageId());
````
The reply from the `<int-aws:sns-outbound-gateway>` may be useful to track and correlate the SNS message
in the downstream flow.
With Java configuration there is just enough to use constructor of `SnsMessageHandler` with `produceReply` boolean
flag to `true` ot switch it to the gateway mode.
By default the `SnsMessageHandler` is one-way `MessageHandler`.
[Spring Cloud AWS]: https://github.com/spring-cloud/spring-cloud-aws
[AWS SDK for Java]: http://aws.amazon.com/sdkforjava/

2
build.gradle
View File

@@ -32,7 +32,7 @@ ext {
servletApiVersion = '3.1.0'
slf4jVersion = '1.7.12'
springCloudAwsVersion = '1.0.3.RELEASE'
springIntegrationVersion = '4.2.0.RELEASE'
springIntegrationVersion = '4.2.4.RELEASE'
idPrefix = 'aws'

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

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2015 the original author or authors.
* Copyright 2013-2016 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.
@@ -36,6 +36,8 @@ public class AWSNamespaceHandler extends AbstractIntegrationNamespaceHandler {
registerBeanDefinitionParser("sqs-outbound-channel-adapter", new SqsOutboundChannelAdapterParser());
registerBeanDefinitionParser("sqs-message-driven-channel-adapter", new SqsMessageDrivenChannelAdapterParser());
registerBeanDefinitionParser("sns-inbound-channel-adapter", new SnsInboundChannelAdapterParser());
registerBeanDefinitionParser("sns-outbound-channel-adapter", new SnsOutboundChannelAdapterParser());
registerBeanDefinitionParser("sns-outbound-gateway", new SnsOutboundGatewayParser());
}
}

41
src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2016 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.support.AbstractBeanDefinition;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
/**
* The parser for the {@code <int-aws:sns-outbound-channel-adapter>}
* @author Artem Bilan
*/
public class SnsOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
AbstractBeanDefinition beanDefinition =
new SnsOutboundGatewayParser()
.parseHandler(element, parserContext)
.getBeanDefinition();
beanDefinition.getConstructorArgumentValues().addIndexedArgumentValue(1, false);
return beanDefinition;
}
}

70
src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java Normal file
View File

@@ -0,0 +1,70 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.aws.outbound.SnsMessageHandler;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
/**
* @author Artem Bilan
*/
public class SnsOutboundGatewayParser extends AbstractConsumerEndpointParser {
@Override
protected String getInputChannelAttributeName() {
return "request-channel";
}
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
String sns = element.getAttribute(AmazonWSParserUtils.SNS_REF);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SnsMessageHandler.class)
.addConstructorArgReference(sns)
.addConstructorArgValue(true);
BeanDefinition topic = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("topic-arn",
"topic-arn-expression", parserContext, element, false);
if (topic != null) {
builder.addPropertyValue("topicArnExpression", topic);
}
BeanDefinition subject = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("subject",
"subject-expression", parserContext, element, false);
if (subject != null) {
builder.addPropertyValue("subjectExpression", subject);
}
BeanDefinition message =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("body-expression", element);
if (message != null) {
builder.addPropertyValue("bodyExpression", message);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
return builder;
}
}

206
src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java Normal file
View File

@@ -0,0 +1,206 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.outbound;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.TypeLocator;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.support.StandardTypeLocator;
import org.springframework.integration.aws.support.AwsHeaders;
import org.springframework.integration.aws.support.SnsBodyBuilder;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import com.amazonaws.services.sns.AmazonSNS;
import com.amazonaws.services.sns.model.PublishRequest;
import com.amazonaws.services.sns.model.PublishResult;
/**
* The {@link AbstractReplyProducingMessageHandler} implementation to send SNS Notifications
* ({@link AmazonSNS#publish(PublishRequest)}) to the provided {@code topicArn}
* (or evaluated at runtime against {@link Message}).
* <p>
* The SNS Message subject can be evaluated as a result of {@link #subjectExpression}.
* <p>
* The algorithm to populate SNS Message body is like:
* <ul>
* <li>
* If the {@code payload instanceof PublishRequest} it is used as is for publishing.
* </li>
* <li>
* If the {@link #bodyExpression} is specified, it is used to be evaluated
* against {@code requestMessage}.
* </li>
* <li>
* If the evaluation result (or {@code payload}) is instance of {@link SnsBodyBuilder},
* the SNS Message is built from there and the {@code messageStructure}
* of the {@link PublishRequest} is set to {@code json}.
* For the convenience the package {@code org.springframework.integration.aws.support} is imported
* to the {@link #evaluationContext} to allow bypass it for the {@link SnsBodyBuilder}
* from the {@link #bodyExpression} definition. For example:
* <pre class="code">
* {@code
* String bodyExpression =
* "SnsBodyBuilder.withDefault(payload).forProtocols(payload.substring(0, 140), 'sms')";
* snsMessageHandler.setBodyExpression(spelExpressionParser.parseExpression(bodyExpression));
* }
* </pre>
* </li>
* <li>
* Otherwise the {@code payload} (or the {@link #bodyExpression} evaluation result) is converted
* to the {@link String} using {@link #getConversionService()}.
* </li>
* </ul>
* <p>
* If this {@link AbstractReplyProducingMessageHandler} is configured with {@link #produceReply} as
* {@code true}, the reply message is composed to be sent to the {@code outputChannel} or
* {@code replyChannel}. The reply message's {@code payload} is exactly the {@link PublishRequest}
* object, which has been just published to SNS. Also this message has {@link AwsHeaders#TOPIC}
* and {@link AwsHeaders#SNS_PUBLISHED_MESSAGE_ID} headers to track published SNS message in the
* downstream.
*
* @author Artem Bilan
*
* @see AmazonSNS
* @see PublishRequest
* @see SnsBodyBuilder
*/
public class SnsMessageHandler extends AbstractReplyProducingMessageHandler {
private final AmazonSNS amazonSns;
private final boolean produceReply;
private EvaluationContext evaluationContext;
private Expression topicArnExpression;
private Expression subjectExpression;
private Expression bodyExpression;
public SnsMessageHandler(AmazonSNS amazonSns) {
this(amazonSns, false);
}
public SnsMessageHandler(AmazonSNS amazonSns, boolean produceReply) {
Assert.notNull(amazonSns, "amazonSns must not be null.");
this.amazonSns = amazonSns;
this.produceReply = produceReply;
}
public void setTopicArn(String topicArn) {
Assert.hasText(topicArn, "topicArn must not be empty.");
this.topicArnExpression = new LiteralExpression(topicArn);
}
public void setTopicArnExpression(Expression topicArnExpression) {
Assert.notNull(topicArnExpression, "topicArnExpression must not be null.");
this.topicArnExpression = topicArnExpression;
}
public void setSubject(String subject) {
Assert.hasText(subject, "subject must not be empty.");
this.subjectExpression = new LiteralExpression(subject);
}
public void setSubjectExpression(Expression subjectExpression) {
Assert.notNull(subjectExpression, "subjectExpression must not be null.");
this.subjectExpression = subjectExpression;
}
/**
* The {@link Expression} to produce the SNS notification message.
* If it evaluates to the {@link SnsBodyBuilder} the {@code messageStructure}
* of the {@link PublishRequest} is set to {@code json}.
* Otherwise the {@link #getConversionService()} is used to convert the evaluation result
* to the {@link String} without setting the {@code messageStructure}.
* @param bodyExpression the {@link Expression} to produce the SNS notification message.
*/
public void setBodyExpression(Expression bodyExpression) {
Assert.notNull(bodyExpression, "bodyExpression must not be null.");
this.bodyExpression = bodyExpression;
}
@Override
protected void doInit() {
super.doInit();
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
TypeLocator typeLocator = this.evaluationContext.getTypeLocator();
if (typeLocator instanceof StandardTypeLocator) {
/*
* Register the 'org.springframework.integration.aws.support' package
* you don't need a FQCN for the 'SnsMessageBuilder'.
*/
((StandardTypeLocator) typeLocator).registerImport("org.springframework.integration.aws.support");
}
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
Object payload = requestMessage.getPayload();
PublishRequest publishRequest = null;
if (payload instanceof PublishRequest) {
publishRequest = (PublishRequest) payload;
}
else {
publishRequest = new PublishRequest();
if (this.topicArnExpression != null) {
String topicArn = this.topicArnExpression.getValue(this.evaluationContext, requestMessage, String.class);
publishRequest.setTopicArn(topicArn);
}
if (this.subjectExpression != null) {
String subject = this.subjectExpression.getValue(this.evaluationContext, requestMessage, String.class);
publishRequest.setSubject(subject);
}
Object snsMessage = requestMessage.getPayload();
if (this.bodyExpression != null) {
snsMessage = this.bodyExpression.getValue(this.evaluationContext, requestMessage);
}
if (snsMessage instanceof SnsBodyBuilder) {
publishRequest.withMessageStructure("json")
.setMessage(((SnsBodyBuilder) snsMessage).build());
}
else {
publishRequest.setMessage(getConversionService().convert(snsMessage, String.class));
}
}
PublishResult publishResult = this.amazonSns.publish(publishRequest);
if (this.produceReply) {
return getMessageBuilderFactory()
.withPayload(publishRequest)
.setHeader(AwsHeaders.TOPIC, publishRequest.getTopicArn())
.setHeader(AwsHeaders.SNS_PUBLISHED_MESSAGE_ID, publishResult.getMessageId());
}
else {
return null;
}
}
}

4
src/main/java/org/springframework/integration/aws/support/AwsHeaders.java
View File

@@ -25,6 +25,8 @@ public abstract class AwsHeaders {
public static final String QUEUE = PREFIX + "queue";
public static final String TOPIC = PREFIX + "topic";
public static final String MESSAGE_ID = PREFIX + "messageId";
public static final String RECEIPT_HANDLE = PREFIX + "receiptHandle";
@@ -33,4 +35,6 @@ public abstract class AwsHeaders {
public static final String SNS_MESSAGE_TYPE = PREFIX + "snsMessageType";
public static final String SNS_PUBLISHED_MESSAGE_ID = PREFIX + "snsPublishedMessageId";
}

67
src/main/java/org/springframework/integration/aws/support/SnsBodyBuilder.java Normal file
View File

@@ -0,0 +1,67 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.support;
import java.util.HashMap;
import java.util.Map;
import org.springframework.util.Assert;
/**
* An utility class to simplify an SNS Message body building.
* Can be used from the {@code SnsMessageHandler#bodyExpression} definition or
* directly in case of manual {@link com.amazonaws.services.sns.model.PublishRequest}
* building.
*
* @author Artem Bilan
*/
public class SnsBodyBuilder {
private final Map<String, String> snsMessage = new HashMap<>();
private SnsBodyBuilder(String defaultMessage) {
Assert.hasText(defaultMessage, "defaultMessage must not be empty.");
this.snsMessage.put("default", defaultMessage);
}
public SnsBodyBuilder forProtocols(String message, String... protocols) {
Assert.hasText(message, "message must not be empty.");
Assert.notEmpty(protocols, "protocols must not be empty.");
for (String protocol : protocols) {
Assert.hasText(protocol, "protocols must not contain empty elements.");
this.snsMessage.put(protocol, message);
}
return this;
}
public String build() {
StringBuilder stringBuilder = new StringBuilder("{");
for (Map.Entry<String, String> entry : snsMessage.entrySet()) {
stringBuilder.append("\"")
.append(entry.getKey())
.append("\":\"")
.append(entry.getValue().replaceAll("\"", "\\\\\""))
.append("\",");
}
return stringBuilder.substring(0, stringBuilder.length() - 1) + "}";
}
public static SnsBodyBuilder withDefault(String defaultMessage) {
return new SnsBodyBuilder(defaultMessage);
}
}

4
src/main/resources/META-INF/spring.schemas
View File

@@ -1,2 +1,2 @@
http\://www.springframework.org/schema/integration/aws/spring-integration-aws-1.0.xsd=org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd
http\://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd=org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd
http\://www.springframework.org/schema/integration/aws/spring-integration-aws-1.0.xsd=org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd
http\://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd=org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd

2
src/main/resources/META-INF/spring.tooling
View File

@@ -1,3 +1,3 @@
http\://www.springframework.org/schema/integration/aws@name=Integration AWS Namespace
http\://www.springframework.org/schema/integration/aws@prefix=int-aws
http\://www.springframework.org/schema/integration/aws@icon=org/springframework/integration/aws/config/xml/spring-integration-aws.gif
http\://www.springframework.org/schema/integration/aws@icon=org/springframework/integration/aws/config/spring-integration-aws.gif

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

@@ -233,7 +233,7 @@
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -591,6 +591,85 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="sns-outbound-channel-adapter">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an outbound SNS Channel Adapter for publishing messages to the topic.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="baseSnsAdapterType">
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="sns-outbound-gateway">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an outbound SNS Channel Adapter for publishing messages to the topic.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="baseSnsAdapterType">
<xsd:attribute name="request-channel" use="required" 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>
Identifies the request channel attached to this gateway.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-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>
Identifies the reply channel attached to this
gateway.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Allows you to specify how long this gateway will wait for
the reply message to be sent successfully to the reply channel
before throwing an exception. This attribute only applies when the
channel might block, for example when using a bounded queue channel that
is currently full.
Also, keep in mind that when sending to a DirectChannel, the
invocation will occur in the sender's thread. Therefore,
the failing of the send operation may be caused by other
components further downstream.
The "reply-timeout" attribute maps to the "sendTimeout" property of the
underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
The attribute will default, if not specified, to '-1', meaning that
by default, the Gateway will wait indefinitely. The value is
specified in milliseconds.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="baseSqsAdapterType">
<xsd:annotation>
<xsd:documentation>
@@ -626,4 +705,93 @@
</xsd:complexType>
<xsd:complexType name="baseSnsAdapterType">
<xsd:choice minOccurs="0" maxOccurs="2">
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
minOccurs="0"/>
</xsd:choice>
<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="topic-arn">
<xsd:annotation>
<xsd:documentation>
The Amazon SNS Topic ARN.
Mutually exclusive with 'topic-arn-expression'.
This attribute isn't mandatory and the topic can be specified on the 'PublishRequest'
payload of the request Message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="topic-arn-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression that resolves to an Amazon SNS Topic ARN.
The 'requestMessage' is the root object for evaluation context.
Mutually exclusive with 'topic-arn'.
This attribute isn't mandatory and the the topic can be specified on the
'com.amazonaws.services.sns.model.PublishRequest'
payload of the request Message.
The 'org.springframework.cloud.aws.core.env.ResourceIdResolver' bean can be used from
the expression to resolve logical topic name to the real ARN.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="subject">
<xsd:annotation>
<xsd:documentation>
The Notification Subject.
Mutually exclusive with 'subject-expression'.
This attribute isn't mandatory and the subject can be fully omitted.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="subject-expression">
<xsd:annotation>
<xsd:documentation>
The SpEL expression for Notification Subject.
The 'requestMessage' is the root object for evaluation context.
Mutually exclusive with 'subject-expression'.
This attribute isn't mandatory and the subject can be fully omitted.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="body-expression">
<xsd:annotation>
<xsd:documentation>
The SpEL expression evaluating the 'message' object for
the 'com.amazonaws.services.sns.model.PublishRequest'.
The 'requestMessage' is the root object for evaluation context.
The 'org.springframework.integration.aws.support' package is registered with the
`EvaluationContext` to simplify usage of the `SnsBodyBuilder` from expression definition.
This attribute isn't mandatory and the 'payload' of request Message can be used directly.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="resource-id-resolver">
<xsd:annotation>
<xsd:documentation>
The 'org.springframework.cloud.aws.core.env.ResourceIdResolver' bean reference.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.cloud.aws.core.env.ResourceIdResolver"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
</xsd:complexType>
</xsd:schema>

0
src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws.gif → src/main/resources/org/springframework/integration/aws/config/spring-integration-aws.gif
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 539 B

After

Width:  |  Height:  |  Size: 539 B

2
src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java
View File

@@ -61,7 +61,7 @@ public class SnsInboundChannelAdapterParserTests {
@Test
public void testSqsMessageDrivenChannelAdapterParser() {
public void testSnsInboundChannelAdapterParser() {
assertSame(this.amazonSns, TestUtils.getPropertyValue(this.snsInboundChannelAdapter,
"notificationStatusResolver.amazonSns"));
assertTrue(TestUtils.getPropertyValue(this.snsInboundChannelAdapter, "handleNotificationStatus", Boolean.class));

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

@@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
xmlns:int="http://www.springframework.org/schema/integration"
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/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<bean id="amazonSns" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="com.amazonaws.services.sns.AmazonSNS"/>
</bean>
<int-aws:sns-outbound-channel-adapter id="defaultAdapter" sns="amazonSns">
<int-aws:request-handler-advice-chain>
<bean class="org.springframework.integration.handler.advice.RequestHandlerRetryAdvice"/>
</int-aws:request-handler-advice-chain>
</int-aws:sns-outbound-channel-adapter>
<int:channel id="notificationChannel"/>
<int-aws:sns-outbound-gateway
id="snsGateway"
sns="amazonSns"
request-channel="notificationChannel"
reply-channel="errorChannel"
topic-arn="foo"
subject="bar"
body-expression="payload.toUpperCase()"
auto-startup="false"
phase="201"/>
</beans>

126
src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java Normal file
View File

@@ -0,0 +1,126 @@
/*
* Copyright 2016 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.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.expression.Expression;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.amazonaws.services.sns.AmazonSNS;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class SnsOutboundChannelAdapterParserTests {
@Autowired
private AmazonSNS amazonSns;
@Autowired
@Qualifier("defaultAdapter")
private MessageChannel defaultAdapterChannel;
@Autowired
@Qualifier("errorChannel")
private MessageChannel errorChannel;
@Autowired
@Qualifier("defaultAdapter.adapter")
private AbstractEndpoint defaultAdapter;
@Autowired
@Qualifier("defaultAdapter.handler")
private MessageHandler defaultAdapterHandler;
@Autowired
@Qualifier("notificationChannel")
private MessageChannel notificationChannel;
@Autowired
@Qualifier("snsGateway")
private AbstractEndpoint snsGateway;
@Autowired
@Qualifier("snsGateway.handler")
private MessageHandler snsGatewayHandler;
@Test
public void testSnsOutboundChannelAdapterDefaultParser() throws Exception {
Object handler = TestUtils.getPropertyValue(this.defaultAdapter, "handler");
assertFalse(AopUtils.isAopProxy(handler));
assertSame(this.defaultAdapterHandler, handler);
assertThat(TestUtils.getPropertyValue(handler, "adviceChain", List.class).get(0),
instanceOf(RequestHandlerRetryAdvice.class));
assertSame(this.defaultAdapterChannel, TestUtils.getPropertyValue(this.defaultAdapter, "inputChannel"));
assertSame(this.amazonSns, TestUtils.getPropertyValue(this.defaultAdapterHandler, "amazonSns"));
assertNotNull(TestUtils.getPropertyValue(this.defaultAdapterHandler, "evaluationContext"));
assertNull(TestUtils.getPropertyValue(this.defaultAdapterHandler, "topicArnExpression"));
assertNull(TestUtils.getPropertyValue(this.defaultAdapterHandler, "subjectExpression"));
assertNull(TestUtils.getPropertyValue(this.defaultAdapterHandler, "bodyExpression"));
}
@Test
public void testSnsOutboundChannelAdapterParser() {
assertSame(this.notificationChannel, TestUtils.getPropertyValue(this.snsGateway, "inputChannel"));
assertSame(this.snsGatewayHandler, TestUtils.getPropertyValue(this.snsGateway, "handler"));
assertFalse(TestUtils.getPropertyValue(this.snsGateway, "autoStartup", Boolean.class));
assertEquals(new Integer(201), TestUtils.getPropertyValue(this.snsGateway, "phase", Integer.class));
assertTrue(TestUtils.getPropertyValue(this.snsGatewayHandler, "produceReply", Boolean.class));
assertSame(this.errorChannel, TestUtils.getPropertyValue(this.snsGatewayHandler, "outputChannel"));
assertSame(this.amazonSns, TestUtils.getPropertyValue(this.snsGatewayHandler, "amazonSns"));
assertNotNull(TestUtils.getPropertyValue(this.snsGatewayHandler, "evaluationContext"));
assertEquals("foo",
TestUtils.getPropertyValue(this.snsGatewayHandler, "topicArnExpression", Expression.class)
.getExpressionString());
assertEquals("bar",
TestUtils.getPropertyValue(this.snsGatewayHandler, "subjectExpression", Expression.class)
.getExpressionString());
assertEquals("payload.toUpperCase()",
TestUtils.getPropertyValue(this.snsGatewayHandler, "bodyExpression", Expression.class)
.getExpressionString());
}
}

76
src/test/java/org/springframework/integration/aws/outbound/SnsMessageBuilderTests.java Normal file
View File

@@ -0,0 +1,76 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.outbound;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import org.junit.Test;
import org.springframework.integration.aws.support.SnsBodyBuilder;
/**
* @author Artem Bilan
*/
public class SnsMessageBuilderTests {
@Test
public void testSnsMessageBuilder() {
try {
SnsBodyBuilder.withDefault("");
fail("IllegalArgumentException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(IllegalArgumentException.class));
assertThat(e.getMessage(), containsString("defaultMessage must not be empty."));
}
String message = SnsBodyBuilder.withDefault("foo").build();
assertEquals("{\"default\":\"foo\"}", message);
try {
SnsBodyBuilder.withDefault("foo")
.forProtocols("{\"foo\" : \"bar\"}")
.build();
fail("IllegalArgumentException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(IllegalArgumentException.class));
assertThat(e.getMessage(), containsString("protocols must not be empty."));
}
try {
SnsBodyBuilder.withDefault("foo")
.forProtocols("{\"foo\" : \"bar\"}", "")
.build();
fail("IllegalArgumentException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(IllegalArgumentException.class));
assertThat(e.getMessage(), containsString("protocols must not contain empty elements."));
}
message = SnsBodyBuilder.withDefault("foo")
.forProtocols("{\"foo\" : \"bar\"}", "sms")
.build();
assertEquals("{\"default\":\"foo\",\"sms\":\"{\\\"foo\\\" : \\\"bar\\\"}\"}", message);
}
}

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

@@ -0,0 +1,134 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.outbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.aws.support.AwsHeaders;
import org.springframework.integration.aws.support.SnsBodyBuilder;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.amazonaws.services.sns.AmazonSNS;
import com.amazonaws.services.sns.model.PublishRequest;
import com.amazonaws.services.sns.model.PublishResult;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@DirtiesContext
public class SnsMessageHandlerTests {
private static SpelExpressionParser PARSER = new SpelExpressionParser();
@Autowired
private MessageChannel sendToSnsChannel;
@Autowired
private AmazonSNS amazonSNS;
@Test
public void testSnsMessageHandler() {
SnsBodyBuilder payload = SnsBodyBuilder.withDefault("foo")
.forProtocols("{\"foo\" : \"bar\"}", "sms");
QueueChannel replyChannel = new QueueChannel();
Message<?> message = MessageBuilder.withPayload(payload)
.setHeader("topic", "topic")
.setHeader("subject", "subject")
.setReplyChannel(replyChannel)
.build();
this.sendToSnsChannel.send(message);
Message<?> reply = replyChannel.receive(1000);
assertNotNull(reply);
ArgumentCaptor<PublishRequest> captor = ArgumentCaptor.forClass(PublishRequest.class);
verify(this.amazonSNS).publish(captor.capture());
PublishRequest publishRequest = captor.getValue();
assertEquals("json", publishRequest.getMessageStructure());
assertEquals("topic", publishRequest.getTopicArn());
assertEquals("subject", publishRequest.getSubject());
assertEquals("{\"default\":\"foo\",\"sms\":\"{\\\"foo\\\" : \\\"bar\\\"}\"}", publishRequest.getMessage());
assertEquals("111", reply.getHeaders().get(AwsHeaders.SNS_PUBLISHED_MESSAGE_ID));
assertEquals("topic", reply.getHeaders().get(AwsHeaders.TOPIC));
assertSame(publishRequest, reply.getPayload());
}
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@Bean
public AmazonSNS amazonSNS() {
AmazonSNS mock = mock(AmazonSNS.class);
doAnswer(new Answer<PublishResult>() {
@Override
public PublishResult answer(InvocationOnMock invocation) throws Throwable {
return new PublishResult().withMessageId("111");
}
}).when(mock).publish(any(PublishRequest.class));
return mock;
}
@Bean
@ServiceActivator(inputChannel = "sendToSnsChannel")
public MessageHandler snsMessageHandler() {
SnsMessageHandler snsMessageHandler = new SnsMessageHandler(amazonSNS(), true);
snsMessageHandler.setTopicArnExpression(PARSER.parseExpression("headers.topic"));
snsMessageHandler.setSubjectExpression(PARSER.parseExpression("headers.subject"));
snsMessageHandler.setBodyExpression(PARSER.parseExpression("payload"));
return snsMessageHandler;
}
}
}