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

GH-196: Add SNS FIFO support

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/196

* Add support for SNS FIFO message group and deduplication IDs
* Add Javadoc, clean up formatting
* Fix Javadoc formatting
* Add SNS FIFO information to README
* Change XML snippet indentation to tabs
* Code samples throughout the README mix tabs and spaces, but this
snippet was using tabs, so this updates the new lines to match.
* Minor code style clean up

**Cherry-pick to `main`**
This commit is contained in:
Christopher Smith
2022-12-09 13:14:45 -05:00
committed by abilan
parent 2eb0ba2c3f
commit 2b77c3810f
6 changed files with 128 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
README.md
View File

@@ -437,9 +437,14 @@ The Java Config looks like:
@Bean
public MessageHandler snsMessageHandler() {
SnsMessageHandler handler = new SnsMessageHandler(amazonSns());
adapter.setTopicArn("arn:aws:sns:eu-west:123456789012:test);
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));
// message-group ID and deduplication ID are used for FIFO topics
handler.setMessageGroupId("foo-messages");
String deduplicationExpression = "headers.id";
handler.setMessageDeduplicationIdExpression(spelExpressionParser.parseExpression(deduplicationExpression))''
return handler;
}
````
@@ -457,12 +462,17 @@ The XML variant may look like:
channel="notificationChannel"
topic-arn="foo"
subject="bar"
message-group-id="foo-messages"
message-deduplication-id-expression="headers.id"
body-expression="payload.toUpperCase()"/>
````
Starting with _version 2.0_, the `SnsMessageHandler` can be configured with the `HeaderMapper` to map message headers to the SNS message attributes.
See `SnsHeaderMapper` implementation for more information and also consult with [Amazon SNS Message Attributes][] about value types and restrictions.
Starting with _version 2.5.3_, the `SnsMessageHandler` supports sending to SNS FIFO topics using the `messageGroupId`/`messageGroupIdExpression`
and `messageDeduplicationIdExpression` properties.
## Metadata Store for Amazon DynamoDB
The `DynamoDbMetadataStore`, a `ConcurrentMetadataStore` implementation, is provided to keep the metadata for Spring Integration components in the distributed Amazon DynamoDB store.

5
src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2022 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.
@@ -30,6 +30,7 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
* The parser for the {@code <int-aws:sns-outbound-channel-adapter>}.
*
* @author Artem Bilan
* @author Christopher Smith
*/
public class SnsOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@@ -42,6 +43,8 @@ public class SnsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap
AwsParserUtils.populateExpressionAttribute("topic-arn", builder, element, parserContext);
AwsParserUtils.populateExpressionAttribute("subject", builder, element, parserContext);
AwsParserUtils.populateExpressionAttribute("message-group-id", builder, element, parserContext);
AwsParserUtils.populateExpressionAttribute("message-deduplication-id", builder, element, parserContext);
BeanDefinition message = IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("body-expression",
element);

71
src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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.
@@ -20,6 +20,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;
import org.springframework.core.log.LogMessage;
import org.springframework.expression.Expression;
import org.springframework.expression.TypeLocator;
import org.springframework.expression.common.LiteralExpression;
@@ -71,6 +72,8 @@ import io.awspring.cloud.core.env.ResourceIdResolver;
* </ul>
*
* @author Artem Bilan
* @author Christopher Smith
*
* @see AmazonSNSAsync
* @see PublishRequest
* @see SnsBodyBuilder
@@ -83,6 +86,10 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler<Map<String, Mes
private Expression subjectExpression;
private Expression messageGroupIdExpression;
private Expression messageDeduplicationIdExpression;
private Expression bodyExpression;
private ResourceIdResolver resourceIdResolver;
@@ -113,10 +120,51 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler<Map<String, Mes
this.subjectExpression = subjectExpression;
}
/**
* A fixed message-group ID to be set for messages sent to an SNS FIFO topic
* from this handler.
* Equivalent to calling {{@link #setMessageGroupIdExpression(Expression)} with
* a literal string expression.
* @param messageGroupId the group ID to be used for all messages sent from this handler
* @since 2.5.3
*/
public void setMessageGroupId(String messageGroupId) {
Assert.hasText(messageGroupId, "messageGroupId must not be empty.");
this.messageGroupIdExpression = new LiteralExpression(messageGroupId);
}
/**
* The {@link Expression} to determine the
* <a href="https://docs.aws.amazon.com/sns/latest/dg/fifo-message-grouping.html">message group</a>
* for messages sent to an SNS FIFO topic from this handler.
* @param messageGroupIdExpression the {@link Expression} to produce the message-group ID
* @since 2.5.3
*/
public void setMessageGroupIdExpression(Expression messageGroupIdExpression) {
Assert.notNull(messageGroupIdExpression, "messageGroupIdExpression must not be null.");
this.messageGroupIdExpression = messageGroupIdExpression;
}
/**
* The {@link Expression} to determine the deduplication ID for this message.
* SNS FIFO topics
* <a href="https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html">require a message deduplication ID to be specified</a>,
* either in the adapter configuration or on a {@link PublishRequest} payload
* of the request {@link Message}, unless content-based deduplication is enabled
* on the topic.
* @param messageDeduplicationIdExpression the {@link Expression} to produce the message deduplication ID
* @since 2.5.3
*/
public void setMessageDeduplicationIdExpression(Expression messageDeduplicationIdExpression) {
Assert.notNull(messageDeduplicationIdExpression, "messageDeduplicationIdExpression must not be null.");
this.messageDeduplicationIdExpression = messageDeduplicationIdExpression;
}
/**
* 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 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
@@ -172,6 +220,25 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler<Map<String, Mes
publishRequest.setSubject(subject);
}
if (this.messageGroupIdExpression != null) {
if (!topicArn.endsWith(".fifo")) {
logger.warn(LogMessage.format("a messageGroupId will be set for non-FIFO topic '%s'", topicArn));
}
String messageGroupId =
this.messageGroupIdExpression.getValue(getEvaluationContext(), message, String.class);
publishRequest.setMessageGroupId(messageGroupId);
}
if (this.messageDeduplicationIdExpression != null) {
if (!topicArn.endsWith(".fifo")) {
logger.warn(
LogMessage.format("a messageDeduplicationId will be set for non-FIFO topic '%s'", topicArn));
}
String messageDeduplicationId =
this.messageDeduplicationIdExpression.getValue(getEvaluationContext(), message, String.class);
publishRequest.setMessageDeduplicationId(messageDeduplicationId);
}
Object snsMessage = message.getPayload();
if (this.bodyExpression != null) {

37
src/main/resources/org/springframework/integration/aws/config/spring-integration-aws.xsd
View File

@@ -531,7 +531,7 @@
Asynchronous callback handler for events in the lifecycle of the request. Users can provide an
implementation of the callback methods in this interface to receive notification of successful or
unsuccessful completion of the operation.
By default successful reply is sent to the 'success-channel' and error message to the
By default, successful reply is sent to the 'success-channel' and error message to the
'failure-channel' if they are provided.
</xsd:documentation>
</xsd:annotation>
@@ -960,7 +960,7 @@
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
This attribute isn't mandatory and the topic can be specified on the
'com.amazonaws.services.sns.model.PublishRequest'
payload of the request Message.
</xsd:documentation>
@@ -985,6 +985,39 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-group-id">
<xsd:annotation>
<xsd:documentation>
The message group ID.
Mutually exclusive with 'message-group-id-expression'.
SNS FIFO topics require a message group to be specified, either in
the adapter configuration or on a 'PublishRequest' payload
of the request Message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-group-id-expression">
<xsd:annotation>
<xsd:documentation>
The SpEL expression for the message group ID.
Mutually exclusive with 'message-group-id'.
SNS FIFO topics require a message group to be specified, either in
the adapter configuration or on a 'PublishRequest' payload
of the request Message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-deduplication-id-expression">
<xsd:annotation>
<xsd:documentation>
The SpEL expression for the message deduplication ID.
SNS FIFO topics require a message deduplication ID to be specified, either in
the adapter configuration or on a 'PublishRequest' payload
of the request Message, unless content-based deduplication is enabled
on the topic.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="body-expression">
<xsd:annotation>
<xsd:documentation>

5
src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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,7 @@ import io.awspring.cloud.core.env.ResourceIdResolver;
/**
* @author Artem Bilan
* @author Christopher Smith
*/
@SpringJUnitConfig
@DirtiesContext
@@ -79,6 +80,8 @@ class SnsOutboundChannelAdapterParserTests {
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "amazonSns")).isSameAs(this.amazonSns);
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "evaluationContext")).isNotNull();
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "topicArnExpression")).isNull();
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "messageGroupIdExpression")).isNull();
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "messageDeduplicationIdExpression")).isNull();
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "subjectExpression")).isNull();
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "bodyExpression")).isNull();
assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "resourceIdResolver"))

5
src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java
View File

@@ -55,6 +55,7 @@ import com.amazonaws.services.sns.model.PublishResult;
/**
* @author Artem Bilan
* @author Christopher Smith
*/
@SpringJUnitConfig
@DirtiesContext
@@ -92,6 +93,8 @@ public class SnsMessageHandlerTests {
assertThat(publishRequest.getMessageStructure()).isEqualTo("json");
assertThat(publishRequest.getTopicArn()).isEqualTo("topic");
assertThat(publishRequest.getSubject()).isEqualTo("subject");
assertThat(publishRequest.getMessageGroupId()).isEqualTo("SUBJECT");
assertThat(publishRequest.getMessageDeduplicationId()).isEqualTo("BAR");
assertThat(publishRequest.getMessage())
.isEqualTo("{\"default\":\"foo\",\"sms\":\"{\\\"foo\\\" : \\\"bar\\\"}\"}");
@@ -136,6 +139,8 @@ public class SnsMessageHandlerTests {
public MessageHandler snsMessageHandler() {
SnsMessageHandler snsMessageHandler = new SnsMessageHandler(amazonSNS());
snsMessageHandler.setTopicArnExpression(PARSER.parseExpression("headers.topic"));
snsMessageHandler.setMessageGroupIdExpression(PARSER.parseExpression("headers.subject.toUpperCase()"));
snsMessageHandler.setMessageDeduplicationIdExpression(PARSER.parseExpression("headers.foo.toUpperCase()"));
snsMessageHandler.setSubjectExpression(PARSER.parseExpression("headers.subject"));
snsMessageHandler.setBodyExpression(PARSER.parseExpression("payload"));
snsMessageHandler.setOutputChannel(resultChannel());