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

Fix FIFO logic in the SnsMessageHandler

Browse Source 
The `messageGroupId` and `messageDeduplicationId` properties
must be set into a `PublishRequest` when topic is in FIFO mode.
Otherwise, they must not be set and fully ignoring them is enough
This commit is contained in:
abilan
2023-03-31 14:27:48 -04:00
parent 10b94e386d
commit 853c410f70
2 changed files with 29 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,7 +29,6 @@ import software.amazon.awssdk.services.sns.model.MessageAttributeValue;
import software.amazon.awssdk.services.sns.model.PublishRequest;
import software.amazon.awssdk.services.sns.model.PublishResponse;
import org.springframework.core.log.LogMessage;
import org.springframework.expression.Expression;
import org.springframework.expression.TypeLocator;
import org.springframework.expression.common.LiteralExpression;
@@ -209,32 +208,43 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler<Map<String, Mes
else {
Assert.state(this.topicArnExpression != null, "'topicArn' or 'topicArnExpression' must be specified.");
PublishRequest.Builder publishRequest = PublishRequest.builder();
String topicArn = this.topicArnExpression.getValue(getEvaluationContext(), message, String.class);
publishRequest.topicArn(this.topicArnResolver.resolveTopicArn(topicArn).toString());
String topic = this.topicArnExpression.getValue(getEvaluationContext(), message, String.class);
String topicArn = this.topicArnResolver.resolveTopicArn(topic).toString();
publishRequest.topicArn(topicArn);
if (this.subjectExpression != null) {
String subject = this.subjectExpression.getValue(getEvaluationContext(), message, String.class);
publishRequest.subject(subject);
}
if (this.messageGroupIdExpression != null) {
if (!topicArn.endsWith(".fifo")) {
logger.warn(LogMessage.format("a messageGroupId will be set for non-FIFO topic '%s'", topicArn));
if (topicArn.endsWith(".fifo")) {
String messageGroupId = null;
if (this.messageGroupIdExpression != null) {
messageGroupId =
this.messageGroupIdExpression.getValue(getEvaluationContext(), message, String.class);
}
String messageGroupId =
this.messageGroupIdExpression.getValue(getEvaluationContext(), message, String.class);
publishRequest.messageGroupId(messageGroupId);
}
Assert.notNull(messageGroupId, () ->
"The 'messageGroupIdExpression' [" + this.messageGroupIdExpression + "] " +
"must not evaluate to null. The failed request message is " + message);
if (this.messageDeduplicationIdExpression != null) {
if (!topicArn.endsWith(".fifo")) {
logger.warn(
LogMessage.format("a messageDeduplicationId will be set for non-FIFO topic '%s'", topicArn));
publishRequest.messageGroupId(messageGroupId);
String messageDeduplicationId = null;
if (this.messageDeduplicationIdExpression != null) {
messageDeduplicationId =
this.messageDeduplicationIdExpression.getValue(getEvaluationContext(), message,
String.class);
}
String messageDeduplicationId =
this.messageDeduplicationIdExpression.getValue(getEvaluationContext(), message, String.class);
Assert.notNull(messageDeduplicationId, () ->
"The 'messageDeduplicationIdExpression' [" + this.messageDeduplicationIdExpression + "] " +
"must not evaluate to null. The failed request message is " + message);
publishRequest.messageDeduplicationId(messageDeduplicationId);
}
else if (this.messageGroupIdExpression != null || this.messageDeduplicationIdExpression != null) {
logger.info("The 'messageGroupIdExpression' and 'messageDeduplicationIdExpression' properties " +
"are ignored for non-FIFO topics.");
}
Object snsMessage = message.getPayload();

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

@@ -90,7 +90,7 @@ public class SnsMessageHandlerTests {
PublishRequest publishRequest = captor.getValue();
assertThat(publishRequest.messageStructure()).isEqualTo("json");
assertThat(publishRequest.topicArn()).isEqualTo("arn:aws:sns:eu-west-1:111111111111:topic");
assertThat(publishRequest.topicArn()).isEqualTo("arn:aws:sns:eu-west-1:111111111111:topic.fifo");
assertThat(publishRequest.subject()).isEqualTo("subject");
assertThat(publishRequest.messageGroupId()).isEqualTo("SUBJECT");
assertThat(publishRequest.messageDeduplicationId()).isEqualTo("BAR");
@@ -105,7 +105,7 @@ public class SnsMessageHandlerTests {
assertThat(messageAttributes.get("foo").stringValue()).isEqualTo("bar");
assertThat(reply.getHeaders().get(AwsHeaders.MESSAGE_ID)).isEqualTo("111");
assertThat(reply.getHeaders().get(AwsHeaders.TOPIC)).isEqualTo("arn:aws:sns:eu-west-1:111111111111:topic");
assertThat(reply.getHeaders().get(AwsHeaders.TOPIC)).isEqualTo("arn:aws:sns:eu-west-1:111111111111:topic.fifo");
assertThat(reply.getPayload()).isSameAs(payload);
}
@@ -121,7 +121,7 @@ public class SnsMessageHandlerTests {
willAnswer(invocation ->
CompletableFuture.completedFuture(
CreateTopicResponse.builder()
.topicArn("arn:aws:sns:eu-west-1:111111111111:topic")
.topicArn("arn:aws:sns:eu-west-1:111111111111:topic.fifo")
.build()))
.given(mock)
.createTopic(any(Consumer.class));