From 0580642439fc477701df52598588d2af3dd99a53 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 28 Sep 2015 14:38:33 -0400 Subject: [PATCH] INTEXT-8: Add SNS Outbound Channel Adapter JIRA: https://jira.spring.io/browse/INTEXT-8 Address PR comments --- README.md | 177 ++++++++++----- build.gradle | 2 +- .../aws/config/xml/AWSNamespaceHandler.java | 4 +- .../xml/SnsOutboundChannelAdapterParser.java | 41 ++++ .../config/xml/SnsOutboundGatewayParser.java | 70 ++++++ .../aws/outbound/SnsMessageHandler.java | 206 ++++++++++++++++++ .../integration/aws/support/AwsHeaders.java | 4 + .../aws/support/SnsBodyBuilder.java | 67 ++++++ src/main/resources/META-INF/spring.schemas | 4 +- src/main/resources/META-INF/spring.tooling | 2 +- .../{xml => }/spring-integration-aws-1.0.xsd | 170 ++++++++++++++- .../{xml => }/spring-integration-aws.gif | Bin .../SnsInboundChannelAdapterParserTests.java | 2 +- ...boundChannelAdapterParserTests-context.xml | 33 +++ .../SnsOutboundChannelAdapterParserTests.java | 126 +++++++++++ .../aws/outbound/SnsMessageBuilderTests.java | 76 +++++++ .../aws/outbound/SnsMessageHandlerTests.java | 134 ++++++++++++ 17 files changed, 1056 insertions(+), 62 deletions(-) create mode 100644 src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java create mode 100644 src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java create mode 100644 src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java create mode 100644 src/main/java/org/springframework/integration/aws/support/SnsBodyBuilder.java rename src/main/resources/org/springframework/integration/aws/config/{xml => }/spring-integration-aws-1.0.xsd (79%) rename src/main/resources/org/springframework/integration/aws/config/{xml => }/spring-integration-aws.gif (100%) create mode 100644 src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml create mode 100644 src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java create mode 100644 src/test/java/org/springframework/integration/aws/outbound/SnsMessageBuilderTests.java create mode 100644 src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java diff --git a/README.md b/README.md index 340ab47..d714edd 100644 --- a/README.md +++ b/README.md @@ -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 ``. +can be injected to the ``. ##Amazon Simple Queue Service (SQS) -The `SQS` adapters are fully based on the [Spring Cloud AWS][] foundation, so for more information about the +The `SQS` adapters are fully based on the [Spring Cloud AWS][] foundation, so for more information about the background components and core configuration, please, refer to the documentation of that project. - + ###Outbound Channel Adapter -The SQS Outbound Channel Adapter is presented by the `SqsMessageHandler` implementation -(``) and allows to send message to the SQS `queue` with provided `AmazonSQS` -client. An SQS queue can be configured explicitly on the adapter (using -`org.springframework.integration.expression.ValueExpression`) or as a SpEL `Expression`, which is evaluated against -request message as a root object of evaluation context. In addition the `queue` can be extracted from the message -headers under `AwsHeaders.QUEUE`. - +The SQS Outbound Channel Adapter is presented by the `SqsMessageHandler` implementation +(``) and allows to send message to the SQS `queue` with provided `AmazonSQS` +client. An SQS queue can be configured explicitly on the adapter (using +`org.springframework.integration.expression.ValueExpression`) or as a SpEL `Expression`, which is evaluated against +request message as a root object of evaluation context. In addition the `queue` can be extracted from the message +headers under `AwsHeaders.QUEUE`. + The Java Configuration is pretty simple: ````java @@ -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` (``) is an extension of +The `SnsInboundChannelAdapter` (``) is an extension of `HttpRequestHandlingMessagingGateway` and must be as a part of Spring MVC application. Its URL must be used from the -AWS Management Console to add this endpoint as a subscriber to the SNS Topic. However before receiving any +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` (``) 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 + +```` + +###Outbound Gateway + +The `` 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 `` 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/ diff --git a/build.gradle b/build.gradle index 7f0ce3d..e9dd11f 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java b/src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java index 7723bef..3083919 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java @@ -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()); } } diff --git a/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java b/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java new file mode 100644 index 0000000..8e4886b --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java @@ -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 } + * @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; + } + +} diff --git a/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java b/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java new file mode 100644 index 0000000..52b59aa --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java @@ -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; + } + +} diff --git a/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java new file mode 100644 index 0000000..94395d7 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java @@ -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}). + *

+ * The SNS Message subject can be evaluated as a result of {@link #subjectExpression}. + *

+ * The algorithm to populate SNS Message body is like: + *

    + *
  • + * If the {@code payload instanceof PublishRequest} it is used as is for publishing. + *
    • + *
  • + * If the {@link #bodyExpression} is specified, it is used to be evaluated + * against {@code requestMessage}. + *
    • + *
  • + * 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: + * 
    + * {@code
+ * String bodyExpression =
+ * "SnsBodyBuilder.withDefault(payload).forProtocols(payload.substring(0, 140), 'sms')";
+ * snsMessageHandler.setBodyExpression(spelExpressionParser.parseExpression(bodyExpression));
+ * }
+ *
    + *
    • + *
  • + * Otherwise the {@code payload} (or the {@link #bodyExpression} evaluation result) is converted + * to the {@link String} using {@link #getConversionService()}. + *
    • + *
+ *

+ * 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; + } + } + +} diff --git a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java index 556664c..f0fa480 100644 --- a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java +++ b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java @@ -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"; + } diff --git a/src/main/java/org/springframework/integration/aws/support/SnsBodyBuilder.java b/src/main/java/org/springframework/integration/aws/support/SnsBodyBuilder.java new file mode 100644 index 0000000..cabedca --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/support/SnsBodyBuilder.java @@ -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 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 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); + } + +} diff --git a/src/main/resources/META-INF/spring.schemas b/src/main/resources/META-INF/spring.schemas index 4e9c49d..9e5cd3e 100644 --- a/src/main/resources/META-INF/spring.schemas +++ b/src/main/resources/META-INF/spring.schemas @@ -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 diff --git a/src/main/resources/META-INF/spring.tooling b/src/main/resources/META-INF/spring.tooling index 878f9cd..f00530f 100644 --- a/src/main/resources/META-INF/spring.tooling +++ b/src/main/resources/META-INF/spring.tooling @@ -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 \ No newline at end of file +http\://www.springframework.org/schema/integration/aws@icon=org/springframework/integration/aws/config/spring-integration-aws.gif diff --git a/src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd similarity index 79% rename from src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd rename to src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd index 25eae0a..8d31d12 100644 --- a/src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws-1.0.xsd +++ b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd @@ -233,7 +233,7 @@ - + @@ -591,6 +591,85 @@ + + + + + Defines an outbound SNS Channel Adapter for publishing messages to the topic. + + + + + + + + + + + + + + + Defines an outbound SNS Channel Adapter for publishing messages to the topic. + + + + + + + + + + + + + Identifies the request channel attached to this gateway. + + + + + + + + + + + + Identifies the reply channel attached to this + gateway. + + + + + + + + + + + + + + + @@ -626,4 +705,93 @@ + + + + + + + + + The 'com.amazonaws.services.sns.AmazonSNS' bean reference. + + + + + + + + + + + + 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. + + + + + + + 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. + + + + + + + The Notification Subject. + Mutually exclusive with 'subject-expression'. + This attribute isn't mandatory and the subject can be fully omitted. + + + + + + + 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. + + + + + + + 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. + + + + + + + The 'org.springframework.cloud.aws.core.env.ResourceIdResolver' bean reference. + + + + + + + + + + + + diff --git a/src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws.gif b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws.gif similarity index 100% rename from src/main/resources/org/springframework/integration/aws/config/xml/spring-integration-aws.gif rename to src/main/resources/org/springframework/integration/aws/config/spring-integration-aws.gif diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java index 1a8c9c9..c3892c4 100644 --- a/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java +++ b/src/test/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParserTests.java @@ -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)); diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml new file mode 100644 index 0000000..322cab3 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests-context.xml @@ -0,0 +1,33 @@ + + + + + + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java new file mode 100644 index 0000000..8c3384a --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java @@ -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()); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/outbound/SnsMessageBuilderTests.java b/src/test/java/org/springframework/integration/aws/outbound/SnsMessageBuilderTests.java new file mode 100644 index 0000000..1a106cb --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/outbound/SnsMessageBuilderTests.java @@ -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); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java new file mode 100644 index 0000000..a929250 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java @@ -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 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() { + + @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; + } + + } + +}