From 2b77c3810fa0ce2125d9d32f1761b3e08c7cf128 Mon Sep 17 00:00:00 2001 From: Christopher Smith Date: Fri, 9 Dec 2022 13:14:45 -0500 Subject: [PATCH] GH-196: Add SNS FIFO support 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`** --- README.md | 12 +++- .../xml/SnsOutboundChannelAdapterParser.java | 5 +- .../aws/outbound/SnsMessageHandler.java | 71 ++++++++++++++++++- .../aws/config/spring-integration-aws.xsd | 37 +++++++++- .../SnsOutboundChannelAdapterParserTests.java | 5 +- .../aws/outbound/SnsMessageHandlerTests.java | 5 ++ 6 files changed, 128 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index f7756ad..0fa318e 100644 --- a/README.md +++ b/README.md @@ -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. 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 index 0bf1576..cdc7053 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java @@ -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 }. * * @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); diff --git a/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java index d845f05..07bc4d5 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java @@ -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; * * * @author Artem Bilan + * @author Christopher Smith + * * @see AmazonSNSAsync * @see PublishRequest * @see SnsBodyBuilder @@ -83,6 +86,10 @@ public class SnsMessageHandler extends AbstractAwsMessageHandlermessage group + * 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 + * require a message deduplication ID to be specified, + * 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 @@ -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. @@ -985,6 +985,39 @@ + + + + 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. + + + + + + + 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. + + + + + + + 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. + + + 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 index 8369801..94bffc3 100644 --- a/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java +++ b/src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java @@ -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")) diff --git a/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java index 96141f5..1ebc70d 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java +++ b/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java @@ -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());