diff --git a/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/java/io/awspring/cloud/samples/ses/MailSendingApplication.java b/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/java/io/awspring/cloud/samples/ses/MailSendingApplication.java
index 298803fc4..9f7b6e691 100644
--- a/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/java/io/awspring/cloud/samples/ses/MailSendingApplication.java
+++ b/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/java/io/awspring/cloud/samples/ses/MailSendingApplication.java
@@ -15,15 +15,40 @@
*/
package io.awspring.cloud.samples.ses;
+import java.io.File;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
+import org.springframework.core.io.ClassPathResource;
import org.springframework.mail.MailSender;
import org.springframework.mail.SimpleMailMessage;
+import org.springframework.mail.javamail.JavaMailSender;
+import org.springframework.mail.javamail.MimeMessageHelper;
import software.amazon.awssdk.services.ses.SesClient;
import software.amazon.awssdk.services.ses.model.VerifyEmailAddressRequest;
+/**
+ * A sample application to demonstrate sending simple emails and emails with attachments.
+ *
+ * To run this sample application, you need to do either of the following.
+ *
+ *
+ *
+ */
@SpringBootApplication
public class MailSendingApplication {
private static final String SENDER = "something@foo.bar";
@@ -37,7 +62,9 @@ public static void main(String[] args) {
ApplicationRunner applicationRunner(MailSender mailSender, SesClient sesClient) {
return args -> {
sendAnEmail(mailSender, sesClient);
- // check localstack logs for sent email
+ sendAnEmailWithAttachment(mailSender, sesClient);
+ sendHtmlEmail(mailSender, sesClient);
+ // check localstack logs for sent email, if you use localstack for running this sample
};
}
@@ -56,4 +83,69 @@ public static void sendAnEmail(MailSender mailSender, SesClient sesClient) {
mailSender.send(simpleMailMessage);
}
+ /**
+ * To send emails with attachments, you must provide the Java Mail API and an implementation of the API in the
+ * classpath. See the dependencies provided in this sample app. If you don't provider an implementation of the Java
+ * Mail API, you would get the following exception at runtime.
+ *
+ *
+ * java.lang.IllegalStateException: Not provider of jakarta.mail.util.StreamProvider was found
+ *
+ *
+ * @param mailSender A {@link JavaMailSender}.
+ * @param sesClient An {@link SesClient}.
+ */
+ public static void sendAnEmailWithAttachment(MailSender mailSender, SesClient sesClient) {
+ // e-mail address has to verified before we email it. If it is not verified SES will return error.
+ sesClient.verifyEmailAddress(VerifyEmailAddressRequest.builder().emailAddress(RECIPIENT).build());
+ sesClient.verifyEmailAddress(VerifyEmailAddressRequest.builder().emailAddress(SENDER).build());
+
+ // A JavaMailSender is needed. Spring Cloud AWS SES automatically configures a JavaMailSender when it finds
+ // the Java Mail API in the classpath. At runtime, an implementation of teh Java Mail API must also be
+ // available.
+ JavaMailSender javaMailSender = (JavaMailSender) mailSender;
+ javaMailSender.send(mimeMessage -> {
+ MimeMessageHelper helper = new MimeMessageHelper(mimeMessage, true, "UTF-8");
+ helper.addTo(RECIPIENT);
+ helper.setFrom(SENDER);
+ File resource = new ClassPathResource("answer.txt").getFile();
+ helper.addAttachment("answer.txt", resource.getAbsoluteFile());
+ helper.setSubject("What is the meaning of life, the universe, and everything?");
+ helper.setText("Open the attached file for the answer you are seeking", false);
+ });
+ }
+
+ /**
+ * To send HTML emails, you must provide the Java Mail API and an implementation of the API in the classpath. See
+ * the dependencies provided in this sample app. If you don't provider an implementation of the Java Mail API, you
+ * would get the following exception at runtime.
+ *
+ *
+ * java.lang.IllegalStateException: Not provider of jakarta.mail.util.StreamProvider was found
+ *
+ *
+ * @param mailSender A {@link JavaMailSender}.
+ * @param sesClient An {@link SesClient}.
+ */
+ public static void sendHtmlEmail(MailSender mailSender, SesClient sesClient) {
+ // e-mail address has to verified before we email it. If it is not verified SES will return error.
+ sesClient.verifyEmailAddress(VerifyEmailAddressRequest.builder().emailAddress(RECIPIENT).build());
+ sesClient.verifyEmailAddress(VerifyEmailAddressRequest.builder().emailAddress(SENDER).build());
+
+ // A JavaMailSender is needed. Spring Cloud AWS SES automatically configures a JavaMailSender when it finds
+ // the Java Mail API in the classpath. At runtime, an implementation of the Java Mail API must also be
+ // available.
+ JavaMailSender javaMailSender = (JavaMailSender) mailSender;
+ javaMailSender.send(mimeMessage -> {
+ MimeMessageHelper helper = new MimeMessageHelper(mimeMessage, true, "UTF-8");
+ helper.addTo(RECIPIENT);
+ helper.setFrom(SENDER);
+ helper.setSubject("What is the meaning of life, the universe, and everything?");
+ String htmlMessage = """
+ What is the meaning of life, the universe, and everything?
+ 42
+ """;
+ helper.setText(htmlMessage, true);
+ });
+ }
}
diff --git a/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/resources/answer.txt b/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/resources/answer.txt
new file mode 100644
index 000000000..2980103fe
--- /dev/null
+++ b/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/resources/answer.txt
@@ -0,0 +1,2 @@
+Question to Spring Cloud AWS: What is the meaning of life, the universe, and everything?
+Answer from Spring Cloud AWS: 42
diff --git a/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/resources/application.properties b/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/resources/application.properties
new file mode 100644
index 000000000..ff20e53b4
--- /dev/null
+++ b/spring-cloud-aws-samples/spring-cloud-aws-ses-sample/src/main/resources/application.properties
@@ -0,0 +1,5 @@
+# Localstack configuration
+spring.cloud.aws.endpoint=http://localhost:4566
+spring.cloud.aws.region.static=us-east-1
+spring.cloud.aws.credentials.access-key=noop
+spring.cloud.aws.credentials.secret-key=noop
diff --git a/spring-cloud-aws-samples/spring-cloud-aws-sns-sample/pom.xml b/spring-cloud-aws-samples/spring-cloud-aws-sns-sample/pom.xml
index b48b2ad36..640e01fa3 100644
--- a/spring-cloud-aws-samples/spring-cloud-aws-sns-sample/pom.xml
+++ b/spring-cloud-aws-samples/spring-cloud-aws-sns-sample/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws-samples
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
4.0.0
spring-cloud-aws-sns-sample
diff --git a/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/pom.xml b/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/pom.xml
index 3044593b2..6d2c4a6c2 100644
--- a/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/pom.xml
+++ b/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws-samples
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
4.0.0
spring-cloud-aws-sqs-sample
diff --git a/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/src/main/java/io/awspring/cloud/sqs/sample/SpringSqsListenMultipleQueues.java b/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/src/main/java/io/awspring/cloud/sqs/sample/SpringSqsListenMultipleQueues.java
new file mode 100644
index 000000000..e08dd8708
--- /dev/null
+++ b/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/src/main/java/io/awspring/cloud/sqs/sample/SpringSqsListenMultipleQueues.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2013-2023 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
+ *
+ * https://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 io.awspring.cloud.sqs.sample;
+
+import io.awspring.cloud.sqs.annotation.SqsListener;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
+import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import software.amazon.awssdk.services.sqs.model.Message;
+
+@Configuration
+public class SpringSqsListenMultipleQueues {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SpringSqsListenMultipleQueues.class);
+
+ private static final String ORDER_QUEUE = "order-queue";
+ private static final String WITHDRAWAL_QUEUE = "withdrawal-queue";
+
+ @SqsListener(queueNames = { ORDER_QUEUE, WITHDRAWAL_QUEUE })
+ void listen(Message message) {
+ LOGGER.info("Received message {}", message);
+ }
+
+ @Bean
+ public ApplicationRunner sendMessageToQueues(SqsTemplate sqsTemplate) {
+ return args -> {
+ sqsTemplate.sendAsync(ORDER_QUEUE, new OrderMessage(UUID.randomUUID(), "john@awsspringcloud.com"));
+
+ sqsTemplate.sendAsync(WITHDRAWAL_QUEUE,
+ new WithdrawalMessage(UUID.randomUUID(), "Mary", "mary@awsspringcloud.com"));
+ };
+ }
+
+ private record WithdrawalMessage(UUID transaction, String customerName, String customerEmail) {
+ }
+
+ private record OrderMessage(UUID orderId, String customerEmail) {
+ }
+
+}
diff --git a/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/src/main/java/io/awspring/cloud/sqs/sample/SqsManualAckSample.java b/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/src/main/java/io/awspring/cloud/sqs/sample/SqsManualAckSample.java
new file mode 100644
index 000000000..dbfe5b4e8
--- /dev/null
+++ b/spring-cloud-aws-samples/spring-cloud-aws-sqs-sample/src/main/java/io/awspring/cloud/sqs/sample/SqsManualAckSample.java
@@ -0,0 +1,77 @@
+package io.awspring.cloud.sqs.sample;
+
+import io.awspring.cloud.sqs.annotation.SqsListener;
+import io.awspring.cloud.sqs.config.SqsMessageListenerContainerFactory;
+import io.awspring.cloud.sqs.listener.acknowledgement.Acknowledgement;
+import io.awspring.cloud.sqs.listener.acknowledgement.AcknowledgementResultCallback;
+import io.awspring.cloud.sqs.listener.acknowledgement.handler.AcknowledgementMode;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
+
+import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.util.Collection;
+import java.util.UUID;
+
+@Configuration
+public class SqsManualAckSample {
+
+ public static final String NEW_USER_QUEUE = "new-user-queue";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SqsManualAckSample.class);
+
+ @Bean
+ public ApplicationRunner sendMessageToQueue(SqsTemplate sqsTemplate) {
+ LOGGER.info("Sending message");
+ return args -> sqsTemplate.send(to -> to.queue(NEW_USER_QUEUE)
+ .payload(new User(UUID.randomUUID(), "John"))
+ );
+ }
+
+ @Bean
+ public SqsTemplate sqsTemplate(SqsAsyncClient sqsAsyncClient) {
+ return SqsTemplate.builder()
+ .sqsAsyncClient(sqsAsyncClient)
+ .build();
+ }
+
+ @SqsListener(NEW_USER_QUEUE)
+ public void listen(Message message) {
+ LOGGER.info("Message received on listen method at {}", OffsetDateTime.now());
+ Acknowledgement.acknowledge(message);
+ }
+
+ @Bean
+ SqsMessageListenerContainerFactory defaultSqsListenerContainerFactory(SqsAsyncClient sqsAsyncClient) {
+ return SqsMessageListenerContainerFactory
+ .builder()
+ .configure(options -> options
+ .acknowledgementMode(AcknowledgementMode.MANUAL)
+ .acknowledgementInterval(Duration.ofSeconds(3)) // NOTE: With acknowledgementInterval 3 seconds, we can batch and ack async.
+ .acknowledgementThreshold(0)
+ )
+ .acknowledgementResultCallback(new AckResultCallback())
+ .sqsAsyncClient(sqsAsyncClient)
+ .build();
+ }
+
+ public record User(UUID id, String name) {
+ }
+
+ static class AckResultCallback implements AcknowledgementResultCallback {
+ @Override
+ public void onSuccess(Collection> messages) {
+ LOGGER.info("Ack with success at {}", OffsetDateTime.now()); }
+
+ @Override
+ public void onFailure(Collection> messages, Throwable t) {
+ LOGGER.error("Ack with fail", t);
+ }
+ }
+}
diff --git a/spring-cloud-aws-secrets-manager/pom.xml b/spring-cloud-aws-secrets-manager/pom.xml
index 3e0a90673..7b7b97462 100644
--- a/spring-cloud-aws-secrets-manager/pom.xml
+++ b/spring-cloud-aws-secrets-manager/pom.xml
@@ -5,7 +5,7 @@
io.awspring.cloud
spring-cloud-aws
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
4.0.0
diff --git a/spring-cloud-aws-ses/pom.xml b/spring-cloud-aws-ses/pom.xml
index c4ae39f43..ad515e902 100644
--- a/spring-cloud-aws-ses/pom.xml
+++ b/spring-cloud-aws-ses/pom.xml
@@ -5,7 +5,7 @@
io.awspring.cloud
spring-cloud-aws
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
4.0.0
diff --git a/spring-cloud-aws-sns/pom.xml b/spring-cloud-aws-sns/pom.xml
index 3bad5ae28..ee4722d1a 100644
--- a/spring-cloud-aws-sns/pom.xml
+++ b/spring-cloud-aws-sns/pom.xml
@@ -6,7 +6,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
4.0.0
diff --git a/spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/TopicsListingTopicArnResolver.java b/spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/TopicsListingTopicArnResolver.java
index eb2a6e740..5c67c0530 100644
--- a/spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/TopicsListingTopicArnResolver.java
+++ b/spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/TopicsListingTopicArnResolver.java
@@ -71,7 +71,7 @@ private Arn doRecursiveCall(@Nullable String token, String topicName) {
private Arn checkIfArnIsInList(String topicName, ListTopicsResponse listTopicsResponse) {
Optional arn = listTopicsResponse.topics().stream().map(Topic::topicArn)
- .filter(ta -> ta.contains(topicName)).findFirst();
+ .filter(ta -> ta.endsWith(":" + topicName)).findFirst();
if (arn.isPresent()) {
return Arn.fromString(arn.get());
}
diff --git a/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/core/TopicsListingTopicArnResolverTest.java b/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/core/TopicsListingTopicArnResolverTest.java
new file mode 100644
index 000000000..7215bf538
--- /dev/null
+++ b/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/core/TopicsListingTopicArnResolverTest.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright 2013-2023 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
+ *
+ * https://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 io.awspring.cloud.sns.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+import java.util.Arrays;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.arns.Arn;
+import software.amazon.awssdk.services.sns.SnsClient;
+import software.amazon.awssdk.services.sns.model.ListTopicsResponse;
+import software.amazon.awssdk.services.sns.model.Topic;
+
+class TopicsListingTopicArnResolverTest {
+ private final SnsClient snsClient = mock(SnsClient.class);
+
+ private final TopicsListingTopicArnResolver resolver = new TopicsListingTopicArnResolver(snsClient);
+
+ /**
+ * SNS topic ARN should be resolved by full topic name rather by just a substring. i.e. "topic1" should not be
+ * resolved to arn:aws:sns:eu-west-1:123456789012:topic11 but rather to arn:aws:sns:eu-west-1:123456789012:topic1
+ */
+ @Test
+ void shouldResolveArnBasedOnTopicName() {
+ given(snsClient.listTopics()).willReturn(aResponseContainingArnsForTopicNames("topic11", "topic1"));
+
+ Arn arn = resolver.resolveTopicArn("topic1");
+
+ assertThat(arn.toString()).isEqualTo("arn:aws:sns:eu-west-1:123456789012:topic1");
+ }
+
+ private ListTopicsResponse aResponseContainingArnsForTopicNames(String... topicNames) {
+ return ListTopicsResponse.builder().topics(stubTopics(topicNames)).build();
+ }
+
+ private List stubTopics(String... topicNames) {
+ return Arrays.stream(topicNames).map(this::createTopic).toList();
+ }
+
+ private Topic createTopic(String name) {
+ return Topic.builder().topicArn("arn:aws:sns:eu-west-1:123456789012:" + name).build();
+ }
+}
diff --git a/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/integration/SnsTemplateIntegrationTest.java b/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/integration/SnsTemplateIntegrationTest.java
index 935df0e88..3ff39bd43 100644
--- a/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/integration/SnsTemplateIntegrationTest.java
+++ b/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/integration/SnsTemplateIntegrationTest.java
@@ -64,7 +64,7 @@ class SnsTemplateIntegrationTest {
@Container
static LocalStackContainer localstack = new LocalStackContainer(
- DockerImageName.parse("localstack/localstack:2.0.0")).withServices(SNS).withServices(SQS).withReuse(true);
+ DockerImageName.parse("localstack/localstack:1.4.0")).withServices(SNS).withServices(SQS).withReuse(true);
@BeforeAll
public static void createSnsTemplate() {
diff --git a/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/sms/SnsSmsTemplateIntegrationTest.java b/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/sms/SnsSmsTemplateIntegrationTest.java
index d7e7bf90c..ba526e3d4 100644
--- a/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/sms/SnsSmsTemplateIntegrationTest.java
+++ b/spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/sms/SnsSmsTemplateIntegrationTest.java
@@ -43,7 +43,7 @@ class SnsSmsTemplateIntegrationTest {
@Container
static LocalStackContainer localstack = new LocalStackContainer(
- DockerImageName.parse("localstack/localstack:2.0.0")).withServices(SNS).withEnv("DEBUG", "1");
+ DockerImageName.parse("localstack/localstack:1.4.0")).withServices(SNS).withEnv("DEBUG", "1");
@BeforeAll
public static void createSnsTemplate() {
diff --git a/spring-cloud-aws-sqs/pom.xml b/spring-cloud-aws-sqs/pom.xml
index bdbdee8c9..550db18c5 100644
--- a/spring-cloud-aws-sqs/pom.xml
+++ b/spring-cloud-aws-sqs/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
4.0.0
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/MessageExecutionThread.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/MessageExecutionThread.java
index 0a78eeb5e..56061fa75 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/MessageExecutionThread.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/MessageExecutionThread.java
@@ -15,6 +15,8 @@
*/
package io.awspring.cloud.sqs;
+import org.springframework.lang.Nullable;
+
/**
* A {@link Thread} implementation for processing messages.
* @author Tomaz Fernandes
@@ -30,7 +32,7 @@ public class MessageExecutionThread extends Thread {
* @param runnable see {@link Thread} javadoc.
* @param nextThreadName see {@link Thread} javadoc.
*/
- public MessageExecutionThread(ThreadGroup threadGroup, Runnable runnable, String nextThreadName) {
+ public MessageExecutionThread(@Nullable ThreadGroup threadGroup, Runnable runnable, String nextThreadName) {
super(threadGroup, runnable, nextThreadName);
}
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/MessageExecutionThreadFactory.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/MessageExecutionThreadFactory.java
index ab9aee1a4..65ae1af0e 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/MessageExecutionThreadFactory.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/MessageExecutionThreadFactory.java
@@ -29,6 +29,21 @@
*/
public class MessageExecutionThreadFactory extends CustomizableThreadFactory {
+ /**
+ * Create a new MessageExecutionThreadFactory with default thread name prefix.
+ */
+ public MessageExecutionThreadFactory() {
+ super();
+ }
+
+ /**
+ * Create a new MessageExecutionThreadFactory with the given thread name prefix.
+ * @param threadNamePrefix the prefix to use for the names of newly created threads
+ */
+ public MessageExecutionThreadFactory(String threadNamePrefix) {
+ super(threadNamePrefix);
+ }
+
@Override
public Thread createThread(Runnable runnable) {
MessageExecutionThread thread = new MessageExecutionThread(getThreadGroup(), runnable, nextThreadName());
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractContainerOptions.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractContainerOptions.java
index 5a6df4050..3945b3270 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractContainerOptions.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractContainerOptions.java
@@ -37,6 +37,8 @@ public abstract class AbstractContainerOptions,
private final int maxMessagesPerPoll;
+ private final boolean autoStartup;
+
private final Duration pollTimeout;
private final Duration maxDelayBetweenPolls;
@@ -71,6 +73,7 @@ public abstract class AbstractContainerOptions,
protected AbstractContainerOptions(Builder builder) {
this.maxConcurrentMessages = builder.maxConcurrentMessages;
this.maxMessagesPerPoll = builder.maxMessagesPerPoll;
+ this.autoStartup = builder.autoStartup;
this.pollTimeout = builder.pollTimeout;
this.maxDelayBetweenPolls = builder.maxDelayBetweenPolls;
this.listenerShutdownTimeout = builder.listenerShutdownTimeout;
@@ -99,6 +102,11 @@ public int getMaxMessagesPerPoll() {
return this.maxMessagesPerPoll;
}
+ @Override
+ public boolean isAutoStartup() {
+ return this.autoStartup;
+ }
+
@Override
public Duration getPollTimeout() {
return this.pollTimeout;
@@ -176,6 +184,8 @@ protected abstract static class Builder,
private static final int DEFAULT_MAX_MESSAGES_PER_POLL = 10;
+ private static final boolean DEFAULT_AUTO_STARTUP = true;
+
private static final Duration DEFAULT_POLL_TIMEOUT = Duration.ofSeconds(10);
private static final Duration DEFAULT_SEMAPHORE_TIMEOUT = Duration.ofSeconds(10);
@@ -196,6 +206,8 @@ protected abstract static class Builder,
private int maxMessagesPerPoll = DEFAULT_MAX_MESSAGES_PER_POLL;
+ private boolean autoStartup = DEFAULT_AUTO_STARTUP;
+
private Duration pollTimeout = DEFAULT_POLL_TIMEOUT;
private Duration maxDelayBetweenPolls = DEFAULT_SEMAPHORE_TIMEOUT;
@@ -233,6 +245,7 @@ protected Builder() {
protected Builder(AbstractContainerOptions options) {
this.maxConcurrentMessages = options.maxConcurrentMessages;
this.maxMessagesPerPoll = options.maxMessagesPerPoll;
+ this.autoStartup = options.autoStartup;
this.pollTimeout = options.pollTimeout;
this.maxDelayBetweenPolls = options.maxDelayBetweenPolls;
this.listenerShutdownTimeout = options.listenerShutdownTimeout;
@@ -261,6 +274,12 @@ public B maxMessagesPerPoll(int maxMessagesPerPoll) {
return self();
}
+ @Override
+ public B autoStartup(boolean autoStartup) {
+ this.autoStartup = autoStartup;
+ return self();
+ }
+
@Override
public B pollTimeout(Duration pollTimeout) {
Assert.notNull(pollTimeout, "pollTimeout cannot be null");
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractMessageListenerContainer.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractMessageListenerContainer.java
index 5375407d0..9566fbb7a 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractMessageListenerContainer.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractMessageListenerContainer.java
@@ -69,6 +69,8 @@ public abstract class AbstractMessageListenerContainer acknowledgementResultCallback = new AsyncAcknowledgementResultCallback() {
};
+ private int phase = DEFAULT_PHASE;
+
/**
* Create an instance with the provided {@link ContainerOptions}
* @param containerOptions the options instance.
@@ -162,6 +164,14 @@ public void setComponentFactories(Collection> co
this.containerComponentFactories = containerComponentFactories;
}
+ /**
+ * Set the phase for the SmartLifecycle for this container instance.
+ * @param phase the phase.
+ */
+ public void setPhase(int phase) {
+ this.phase = phase;
+ }
+
/**
* Returns the {@link ContainerOptions} instance for this container. Changed options will take effect on container
* restart.
@@ -252,6 +262,15 @@ public boolean isRunning() {
return this.isRunning;
}
+ public int getPhase() {
+ return this.phase;
+ }
+
+ @Override
+ public boolean isAutoStartup() {
+ return containerOptions.isAutoStartup();
+ }
+
@Override
public void start() {
if (this.isRunning) {
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractPipelineMessageListenerContainer.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractPipelineMessageListenerContainer.java
index 3201c700e..6808f647a 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractPipelineMessageListenerContainer.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/AbstractPipelineMessageListenerContainer.java
@@ -241,7 +241,7 @@ protected TaskExecutor createTaskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
int poolSize = getContainerOptions().getMaxConcurrentMessages() * this.messageSources.size();
executor.setMaxPoolSize(poolSize);
- executor.setCorePoolSize(getContainerOptions().getMaxMessagesPerPoll());
+ executor.setCorePoolSize(poolSize);
// Necessary due to a small racing condition between releasing the permit and releasing the thread.
executor.setQueueCapacity(poolSize);
executor.setAllowCoreThreadTimeOut(true);
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/ContainerOptions.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/ContainerOptions.java
index 968099937..a85509eaf 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/ContainerOptions.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/ContainerOptions.java
@@ -50,6 +50,13 @@ public interface ContainerOptions, B extends Co
*/
int getMaxMessagesPerPoll();
+ /**
+ * Checks whether the container should be started automatically or manually. Default is true.
+ *
+ * @return true if the container starts automatically, false if it should be started manually
+ */
+ boolean isAutoStartup();
+
/**
* Set the maximum time the polling thread should wait for a full batch of permits to be available before trying to
* acquire a partial batch if so configured. A poll is only actually executed if at least one permit is available.
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/ContainerOptionsBuilder.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/ContainerOptionsBuilder.java
index 8eb436473..6d706a744 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/ContainerOptionsBuilder.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/ContainerOptionsBuilder.java
@@ -46,6 +46,15 @@ public interface ContainerOptionsBuilder
*/
B maxMessagesPerPoll(int maxMessagesPerPoll);
+ /**
+ * Set whether the container should be started automatically or manually. By default, the container is set to start
+ * automatically.
+ *
+ * @param autoStartup true if the container is set to start automatically, false if it should be started manually
+ * @return this instance.
+ */
+ B autoStartup(boolean autoStartup);
+
/**
* Set the maximum time the polling thread should wait for a full batch of permits to be available before trying to
* acquire a partial batch if so configured. A poll is only actually executed if at least one permit is available.
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/DefaultListenerContainerRegistry.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/DefaultListenerContainerRegistry.java
index 4b3051406..04a2b288f 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/DefaultListenerContainerRegistry.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/DefaultListenerContainerRegistry.java
@@ -18,10 +18,13 @@
import io.awspring.cloud.sqs.LifecycleHandler;
import java.util.Collection;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.context.SmartLifecycle;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -50,6 +53,8 @@ public class DefaultListenerContainerRegistry implements MessageListenerContaine
private volatile boolean running = false;
+ private int phase = MessageListenerContainer.DEFAULT_PHASE;
+
@Override
public void registerListenerContainer(MessageListenerContainer listenerContainer) {
Assert.notNull(listenerContainer, "listenerContainer cannot be null");
@@ -75,7 +80,9 @@ public MessageListenerContainer getContainerById(String id) {
public void start() {
synchronized (this.lifecycleMonitor) {
logger.debug("Starting {}", getClass().getSimpleName());
- LifecycleHandler.get().start(this.listenerContainers.values());
+ List> containersToStart = this.listenerContainers.values().stream()
+ .filter(SmartLifecycle::isAutoStartup).collect(Collectors.toList());
+ LifecycleHandler.get().start(containersToStart);
this.running = true;
logger.debug("{} started", getClass().getSimpleName());
}
@@ -95,4 +102,12 @@ public boolean isRunning() {
return this.running;
}
+ @Override
+ public int getPhase() {
+ return phase;
+ }
+
+ public void setPhase(int phase) {
+ this.phase = phase;
+ }
}
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/MessageListenerContainer.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/MessageListenerContainer.java
index 02bc862bc..aa0effcac 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/MessageListenerContainer.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/MessageListenerContainer.java
@@ -28,6 +28,8 @@
*/
public interface MessageListenerContainer extends SmartLifecycle {
+ int DEFAULT_PHASE = Integer.MAX_VALUE;
+
/**
* Get the container id.
* @return the id.
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/SqsMessageListenerContainer.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/SqsMessageListenerContainer.java
index 2f79772c0..9bf78e84b 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/SqsMessageListenerContainer.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/SqsMessageListenerContainer.java
@@ -31,6 +31,7 @@
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.context.SmartLifecycle;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
@@ -177,6 +178,8 @@ public static class Builder {
private AcknowledgementResultCallback acknowledgementResultCallback;
+ private Integer phase;
+
public Builder id(String id) {
this.id = id;
return this;
@@ -250,6 +253,11 @@ public Builder configure(Consumer options) {
return this;
}
+ public Builder phase(Integer phase) {
+ this.phase = phase;
+ return this;
+ }
+
// @formatter:off
public SqsMessageListenerContainer build() {
SqsMessageListenerContainer container = new SqsMessageListenerContainer<>(this.sqsAsyncClient);
@@ -262,9 +270,11 @@ public SqsMessageListenerContainer build() {
.acceptIfNotNull(this.acknowledgementResultCallback, container::setAcknowledgementResultCallback)
.acceptIfNotNull(this.asyncAcknowledgementResultCallback, container::setAcknowledgementResultCallback)
.acceptIfNotNull(this.containerComponentFactories, container::setComponentFactories)
- .acceptIfNotEmpty(this.queueNames, container::setQueueNames);
+ .acceptIfNotEmpty(this.queueNames, container::setQueueNames)
+ .acceptIfNotNullOrElse(container::setPhase, this.phase, DEFAULT_PHASE);
this.messageInterceptors.forEach(container::addMessageInterceptor);
this.asyncMessageInterceptors.forEach(container::addMessageInterceptor);
+
container.configure(this.optionsConsumer);
return container;
}
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/AbstractMessagingTemplate.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/AbstractMessagingTemplate.java
index 692b38782..ba5886796 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/AbstractMessagingTemplate.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/operations/AbstractMessagingTemplate.java
@@ -356,7 +356,7 @@ protected MessageConversionContext getSendMessageConversionContext(Message void logSendMessageResult(String endpointToUse, Message message, @Nullable Throwable t) {
if (t == null) {
logger.trace("Message {} successfully sent to endpoint {} with id {}", message,
- MessageHeaderUtils.getId(message), endpointToUse);
+ endpointToUse, MessageHeaderUtils.getId(message));
}
else {
logger.error("Error sending message {} to endpoint {}", MessageHeaderUtils.getId(message), endpointToUse,
@@ -372,7 +372,7 @@ private void logSendMessageBatchResult(String endpointToUse, Collection addAdditionalReceiveHeaders(SqsReceiveOptionsImpl op
@Override
protected org.springframework.messaging.Message preProcessMessageForSend(String endpointToUse,
org.springframework.messaging.Message message) {
- return FifoUtils.isFifo(endpointToUse) ? addMissingFifoSendHeaders(message) : message;
+ return FifoUtils.isFifo(endpointToUse) ? addMissingFifoSendHeaders(endpointToUse, message) : message;
}
@Override
protected Collection> preProcessMessagesForSend(String endpointToUse,
Collection> messages) {
return FifoUtils.isFifo(endpointToUse)
- ? messages.stream().map(this::addMissingFifoSendHeaders).collect(Collectors.toList())
+ ? messages.stream().map(message -> addMissingFifoSendHeaders(endpointToUse, message)).toList()
: messages;
}
- private org.springframework.messaging.Message addMissingFifoSendHeaders(
+ private org.springframework.messaging.Message addMissingFifoSendHeaders(String endpointName,
org.springframework.messaging.Message message) {
- return MessageHeaderUtils.addHeadersIfAbsent(message,
- Map.of(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, UUID.randomUUID().toString(),
- SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER,
- UUID.randomUUID().toString()));
+
+ Set additionalAttributes = Set.of(QueueAttributeName.CONTENT_BASED_DEDUPLICATION);
+ String contentBasedDedupQueueAttribute = getQueueAttributes(endpointName, additionalAttributes).join()
+ .getQueueAttribute(QueueAttributeName.CONTENT_BASED_DEDUPLICATION);
+
+ boolean isContentBasedDedup = Boolean.parseBoolean(contentBasedDedupQueueAttribute);
+ Map defaultHeaders;
+ if (isContentBasedDedup) {
+ defaultHeaders = Map.of(MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, UUID.randomUUID().toString());
+ }
+ else {
+ defaultHeaders = Map.of(MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, UUID.randomUUID().toString(),
+ MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER, UUID.randomUUID().toString());
+ }
+
+ return MessageHeaderUtils.addHeadersIfAbsent(message, defaultHeaders);
}
@Override
@@ -408,10 +423,21 @@ private boolean isSkipAttribute(MessageSystemAttributeName name) {
}
private CompletableFuture getQueueAttributes(String endpointName) {
- return this.queueAttributesCache.computeIfAbsent(endpointName,
- newName -> QueueAttributesResolver.builder().sqsAsyncClient(this.sqsAsyncClient).queueName(newName)
- .queueNotFoundStrategy(this.queueNotFoundStrategy).queueAttributeNames(this.queueAttributeNames)
- .build().resolveQueueAttributes());
+ return getQueueAttributes(endpointName, Collections.emptySet());
+ }
+
+ private CompletableFuture getQueueAttributes(String endpointName,
+ Set additionalAttributes) {
+ return this.queueAttributesCache.computeIfAbsent(endpointName, newName -> {
+ // ensure we have the content based dedupe config to determine default fifo send headers
+ Set namesToRequest = new HashSet<>(queueAttributeNames);
+ if (additionalAttributes != null && !additionalAttributes.isEmpty()) {
+ namesToRequest.addAll(additionalAttributes);
+ }
+ return QueueAttributesResolver.builder().sqsAsyncClient(this.sqsAsyncClient).queueName(newName)
+ .queueNotFoundStrategy(this.queueNotFoundStrategy).queueAttributeNames(namesToRequest).build()
+ .resolveQueueAttributes();
+ });
}
@Override
diff --git a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/AbstractMessagingMessageConverter.java b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/AbstractMessagingMessageConverter.java
index 8c2ca7929..d3b8607c6 100644
--- a/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/AbstractMessagingMessageConverter.java
+++ b/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/support/converter/AbstractMessagingMessageConverter.java
@@ -205,13 +205,14 @@ public MessageConversionContext createMessageConversionContext() {
@Override
public S fromMessagingMessage(Message message, @Nullable MessageConversionContext context) {
+ // We must make sure the message id stays consistent throughout this process
MessageHeaders headers = getMessageHeaders(message);
- S messageWithHeaders = this.headerMapper.fromHeaders(headers);
- Object payload = Objects
- .requireNonNull(this.payloadMessageConverter.toMessage(message.getPayload(), message.getHeaders()),
- () -> "payloadMessageConverter returned null message for message " + message)
- .getPayload();
- return doConvertMessage(messageWithHeaders, payload);
+ Message convertedMessage = Objects.requireNonNull(
+ this.payloadMessageConverter.toMessage(message.getPayload(), message.getHeaders()),
+ () -> "payloadMessageConverter returned null message for message " + message);
+ MessageHeaders completeHeaders = MessageHeaderUtils.addHeadersIfAbsent(headers, convertedMessage.getHeaders());
+ S messageWithHeaders = this.headerMapper.fromHeaders(completeHeaders);
+ return doConvertMessage(messageWithHeaders, convertedMessage.getPayload());
}
private MessageHeaders getMessageHeaders(Message message) {
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/BaseSqsIntegrationTest.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/BaseSqsIntegrationTest.java
index 79a97753a..f6e88c179 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/BaseSqsIntegrationTest.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/BaseSqsIntegrationTest.java
@@ -48,7 +48,7 @@ abstract class BaseSqsIntegrationTest {
protected static final boolean purgeQueues = true;
- private static final String LOCAL_STACK_VERSION = "localstack/localstack:2.0.0";
+ private static final String LOCAL_STACK_VERSION = "localstack/localstack:1.4.0";
static LocalStackContainer localstack = new LocalStackContainer(DockerImageName.parse(LOCAL_STACK_VERSION));
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsFifoIntegrationTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsFifoIntegrationTests.java
index 410e59a02..a2aef3538 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsFifoIntegrationTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsFifoIntegrationTests.java
@@ -37,6 +37,7 @@
import io.awspring.cloud.sqs.listener.acknowledgement.AcknowledgementResultCallback;
import io.awspring.cloud.sqs.listener.acknowledgement.handler.AcknowledgementHandler;
import io.awspring.cloud.sqs.listener.acknowledgement.handler.OnSuccessAcknowledgementHandler;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
@@ -48,7 +49,6 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -59,24 +59,23 @@
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.StopWatch;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
-import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry;
-import software.amazon.awssdk.services.sqs.model.SendMessageBatchResponse;
/**
* Integration tests for handling SQS FIFO queues.
*
* @author Tomaz Fernandes
+ * @author Mikhail Strokov
*/
@SpringBootTest
class SqsFifoIntegrationTests extends BaseSqsIntegrationTest {
@@ -101,16 +100,13 @@ class SqsFifoIntegrationTests extends BaseSqsIntegrationTest {
static final String FIFO_MANUALLY_CREATE_BATCH_FACTORY_QUEUE_NAME = "fifo_manually_create_batch_factory_test_queue.fifo";
- private static final String TEST_SQS_ASYNC_CLIENT_BEAN_NAME = "testSqsAsyncClient";
-
private static final String ERROR_ON_ACK_FACTORY = "errorOnAckFactory";
@Autowired
LatchContainer latchContainer;
@Autowired
- @Qualifier(TEST_SQS_ASYNC_CLIENT_BEAN_NAME)
- SqsAsyncClient sqsAsyncClient;
+ SqsTemplate sqsTemplate;
@Autowired
ObjectMapper objectMapper;
@@ -176,6 +172,7 @@ public void afterSingletonsInstantiated() {
loadSimulator.setBound(1000);
loadSimulator.setRandom(true);
}
+
}
@Test
@@ -184,15 +181,14 @@ void receivesMessagesInOrder() throws Exception {
String messageGroupId = UUID.randomUUID().toString();
List values = IntStream.range(0, this.settings.messagesPerTest).mapToObj(String::valueOf)
.collect(toList());
- String queueUrl = fetchQueueUrl(FIFO_RECEIVES_MESSAGES_IN_ORDER_QUEUE_NAME);
- sendMessageTo(queueUrl, values, messageGroupId);
+ sqsTemplate.sendMany(FIFO_RECEIVES_MESSAGES_IN_ORDER_QUEUE_NAME,
+ createMessagesFromValues(messageGroupId, values));
assertThat(latchContainer.receivesMessageLatch.await(settings.latchTimeoutSeconds, TimeUnit.SECONDS)).isTrue();
assertThat(receivesMessageInOrderListener.receivedMessages).containsExactlyElementsOf(values);
}
@Test
void receivesMessagesInOrderFromManyMessageGroups() throws Exception {
- String queueUrl = fetchQueueUrl(FIFO_RECEIVES_MESSAGE_IN_ORDER_MANY_GROUPS_QUEUE_NAME);
int messagesPerTest = Math.max(this.settings.messagesPerTest, 30);
int numberOfMessageGroups = messagesPerTest / Math.max(this.settings.messagesPerMessageGroup, 10);
int messagesPerMessageGroup = Math.max(messagesPerTest / numberOfMessageGroups, 1);
@@ -206,7 +202,21 @@ void receivesMessagesInOrderFromManyMessageGroups() throws Exception {
LoadSimulator loadSimulator = new LoadSimulator().setLoadEnabled(true).setRandom(true).setBound(20);
IntStream.range(0, messageGroups.size()).forEach(index -> {
if (this.settings.sendMessages) {
- sendMessageTo(queueUrl, values, messageGroups.get(index));
+ try {
+ if (useLocalStackClient) {
+ sqsTemplate.sendMany(FIFO_RECEIVES_MESSAGE_IN_ORDER_MANY_GROUPS_QUEUE_NAME,
+ createMessagesFromValues(messageGroups.get(index), values));
+ }
+ else {
+ sqsTemplate.sendManyAsync(FIFO_RECEIVES_MESSAGE_IN_ORDER_MANY_GROUPS_QUEUE_NAME,
+ createMessagesFromValues(messageGroups.get(index), values));
+ }
+ }
+ catch (Exception e) {
+ logger.error("Error sending messages to queue {}",
+ FIFO_RECEIVES_MESSAGE_IN_ORDER_MANY_GROUPS_QUEUE_NAME, e);
+ throw (RuntimeException) e;
+ }
}
if (index % 10 == 0) {
loadSimulator.runLoad();
@@ -233,8 +243,8 @@ void stopsProcessingAfterException() throws Exception {
List values = IntStream.range(0, this.settings.messagesPerTest).mapToObj(String::valueOf)
.collect(toList());
String messageGroupId = UUID.randomUUID().toString();
- String queueUrl = fetchQueueUrl(FIFO_STOPS_PROCESSING_ON_ERROR_QUEUE_NAME);
- sendMessageTo(queueUrl, values, messageGroupId);
+ sqsTemplate.sendMany(FIFO_STOPS_PROCESSING_ON_ERROR_QUEUE_NAME,
+ createMessagesFromValues(messageGroupId, values));
assertThat(latchContainer.stopsProcessingOnErrorLatch1.await(settings.latchTimeoutSeconds, TimeUnit.SECONDS))
.isTrue();
logger.debug("receivedMessagesBeforeException: {}", stopsOnErrorListener.receivedMessagesBeforeException);
@@ -263,8 +273,8 @@ void stopsProcessingAfterAckException() throws Exception {
List values = IntStream.range(0, this.settings.messagesPerTest).mapToObj(String::valueOf)
.collect(toList());
String messageGroupId = UUID.randomUUID().toString();
- String queueUrl = fetchQueueUrl(FIFO_STOPS_PROCESSING_ON_ACK_ERROR_ERROR_QUEUE_NAME);
- sendMessageTo(queueUrl, values, messageGroupId);
+ sqsTemplate.sendMany(FIFO_STOPS_PROCESSING_ON_ACK_ERROR_ERROR_QUEUE_NAME,
+ createMessagesFromValues(messageGroupId, values));
assertThat(latchContainer.stopsProcessingOnAckErrorLatch1.await(settings.latchTimeoutSeconds, TimeUnit.SECONDS))
.isTrue();
logger.debug("Messages consumed before error: {}", messagesContainer.stopsProcessingOnAckErrorBeforeThrown);
@@ -289,10 +299,12 @@ void receivesBatchesManyGroups() throws Exception {
String messageGroupId1 = UUID.randomUUID().toString();
String messageGroupId2 = UUID.randomUUID().toString();
String messageGroupId3 = UUID.randomUUID().toString();
- String queueUrl = fetchQueueUrl(FIFO_RECEIVES_BATCHES_MANY_GROUPS_QUEUE_NAME);
- sendMessageTo(queueUrl, values, messageGroupId1);
- sendMessageTo(queueUrl, values, messageGroupId2);
- sendMessageTo(queueUrl, values, messageGroupId3);
+ sqsTemplate.sendMany(FIFO_RECEIVES_BATCHES_MANY_GROUPS_QUEUE_NAME,
+ createMessagesFromValues(messageGroupId1, values));
+ sqsTemplate.sendMany(FIFO_RECEIVES_BATCHES_MANY_GROUPS_QUEUE_NAME,
+ createMessagesFromValues(messageGroupId2, values));
+ sqsTemplate.sendMany(FIFO_RECEIVES_BATCHES_MANY_GROUPS_QUEUE_NAME,
+ createMessagesFromValues(messageGroupId3, values));
assertThat(latchContainer.receivesBatchManyGroupsLatch.await(settings.latchTimeoutSeconds, TimeUnit.SECONDS))
.isTrue();
assertThat(receivesBatchesFromManyGroupsListener.receivedMessages.get(messageGroupId1))
@@ -305,10 +317,10 @@ void receivesBatchesManyGroups() throws Exception {
@Test
void manuallyCreatesContainer() throws Exception {
- String queueUrl = fetchQueueUrl(FIFO_MANUALLY_CREATE_CONTAINER_QUEUE_NAME);
List values = IntStream.range(0, this.settings.messagesPerTest).mapToObj(String::valueOf)
.collect(toList());
- sendMessageTo(queueUrl, values, UUID.randomUUID().toString());
+ sqsTemplate.sendMany(FIFO_MANUALLY_CREATE_CONTAINER_QUEUE_NAME,
+ createMessagesFromValues(UUID.randomUUID().toString(), values));
assertThat(latchContainer.manuallyCreatedContainerLatch.await(settings.latchTimeoutSeconds, TimeUnit.SECONDS))
.isTrue();
assertThat(messagesContainer.manuallyCreatedContainerMessages).containsExactlyElementsOf(values);
@@ -316,10 +328,10 @@ void manuallyCreatesContainer() throws Exception {
@Test
void manuallyCreatesBatchContainer() throws Exception {
- String queueUrl = fetchQueueUrl(FIFO_MANUALLY_CREATE_BATCH_CONTAINER_QUEUE_NAME);
List values = IntStream.range(0, this.settings.messagesPerTest).mapToObj(String::valueOf)
.collect(toList());
- sendMessageTo(queueUrl, values, UUID.randomUUID().toString());
+ sqsTemplate.sendMany(FIFO_MANUALLY_CREATE_BATCH_CONTAINER_QUEUE_NAME,
+ createMessagesFromValues(UUID.randomUUID().toString(), values));
assertThat(
latchContainer.manuallyCreatedBatchContainerLatch.await(settings.latchTimeoutSeconds, TimeUnit.SECONDS))
.isTrue();
@@ -328,10 +340,10 @@ void manuallyCreatesBatchContainer() throws Exception {
@Test
void manuallyCreatesFactory() throws Exception {
- String queueUrl = fetchQueueUrl(FIFO_MANUALLY_CREATE_FACTORY_QUEUE_NAME);
List values = IntStream.range(0, this.settings.messagesPerTest).mapToObj(String::valueOf)
.collect(toList());
- sendMessageTo(queueUrl, values, UUID.randomUUID().toString());
+ sqsTemplate.sendMany(FIFO_MANUALLY_CREATE_FACTORY_QUEUE_NAME,
+ createMessagesFromValues(UUID.randomUUID().toString(), values));
assertThat(latchContainer.manuallyCreatedFactoryLatch.await(settings.latchTimeoutSeconds, TimeUnit.SECONDS))
.isTrue();
assertThat(messagesContainer.manuallyCreatedFactoryMessages).containsExactlyElementsOf(values);
@@ -339,16 +351,28 @@ void manuallyCreatesFactory() throws Exception {
@Test
void manuallyCreatesBatchFactory() throws Exception {
- String queueUrl = fetchQueueUrl(FIFO_MANUALLY_CREATE_BATCH_FACTORY_QUEUE_NAME);
List values = IntStream.range(0, this.settings.messagesPerTest).mapToObj(String::valueOf)
.collect(toList());
- sendMessageTo(queueUrl, values, UUID.randomUUID().toString());
+ sqsTemplate.sendMany(FIFO_MANUALLY_CREATE_BATCH_FACTORY_QUEUE_NAME,
+ createMessagesFromValues(UUID.randomUUID().toString(), values));
assertThat(
latchContainer.manuallyCreatedBatchFactoryLatch.await(settings.latchTimeoutSeconds, TimeUnit.SECONDS))
.isTrue();
assertThat(messagesContainer.manuallyCreatedBatchFactoryMessages).containsExactlyElementsOf(values);
}
+ private Message createMessage(String body, String messageGroupId) {
+ return MessageBuilder.withPayload(body)
+ .setHeader(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER, messageGroupId)
+ .setHeader(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER,
+ UUID.randomUUID().toString())
+ .build();
+ }
+
+ private List> createMessagesFromValues(String messageGroupId, List values) {
+ return values.stream().map(value -> createMessage(value, messageGroupId)).toList();
+ }
+
static class ReceivesMessageInOrderListener {
List receivedMessages = Collections.synchronizedList(new ArrayList<>());
@@ -466,57 +490,7 @@ void listen(List> messages) {
messages.forEach(msg -> latchContainer.receivesBatchManyGroupsLatch.countDown());
logger.trace("Finished processing messages {} for group id {}", values, messageGroupId);
}
- }
- private void sendMessageTo(String queueUrl, List messageBodies, String messageGroupId) {
- try {
- if (useLocalStackClient) {
- sendManyTo(queueUrl, messageBodies, messageGroupId).join();
- }
- else {
- sendManyTo(queueUrl, messageBodies, messageGroupId);
- }
- }
- catch (Exception e) {
- logger.error("Error sending messages to queue {}", queueUrl, e);
- throw (RuntimeException) e;
- }
- }
-
- private CompletableFuture sendManyTo(String queueUrl, List messageBodies, String messageGroupId) {
- return IntStream.range(0, (int) Math.ceil(messageBodies.size() / 10.))
- .mapToObj(index -> messageBodies.subList(index * 10, Math.min((index + 1) * 10, messageBodies.size())))
- .reduce(CompletableFuture.completedFuture(null), (previousFuture, messages) -> previousFuture
- .thenCompose(theVoid -> doSendMessageTo(queueUrl, messages, messageGroupId).thenRun(() -> {
- })), (a, b) -> a);
- }
-
- AtomicInteger messagesSent = new AtomicInteger();
-
- private CompletableFuture doSendMessageTo(String queueUrl, List messageBodies,
- String messageGroupId) {
- return sqsAsyncClient.sendMessageBatch(req -> req
- .entries(messageBodies.stream().map(body -> createEntry(body, messageGroupId)).collect(toList()))
- .queueUrl(queueUrl).build()).whenComplete((v, t) -> {
- if (t != null) {
- logger.error("Error sending messages", t);
- }
- else {
- int sent = messagesSent.addAndGet(messageBodies.size());
- if (sent % 1000 == 0) {
- logger.debug("Sent {} messages to queue {}", sent, queueUrl);
- }
- }
- });
- }
-
- private SendMessageBatchRequestEntry createEntry(String body, String messageGroupId) {
- return SendMessageBatchRequestEntry.builder().messageBody(body).id(UUID.randomUUID().toString())
- .messageGroupId(messageGroupId).messageDeduplicationId(UUID.randomUUID().toString()).build();
- }
-
- private String fetchQueueUrl(String receivesMessageQueueName) throws InterruptedException, ExecutionException {
- return this.sqsAsyncClient.getQueueUrl(req -> req.queueName(receivesMessageQueueName)).get().queueUrl();
}
static class LatchContainer {
@@ -799,9 +773,10 @@ ObjectMapper objectMapper() {
return new ObjectMapper();
}
- @Bean(name = TEST_SQS_ASYNC_CLIENT_BEAN_NAME)
- SqsAsyncClient sqsAsyncClientProducer() {
- return BaseSqsIntegrationTest.createHighThroughputAsyncClient();
+ @Bean
+ SqsTemplate sqsTemplate() {
+ return SqsTemplate.builder().sqsAsyncClient(BaseSqsIntegrationTest.createHighThroughputAsyncClient())
+ .build();
}
}
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java
index 943e3443b..4b4b1018d 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsIntegrationTests.java
@@ -17,6 +17,7 @@
import static java.util.Collections.singletonMap;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.awspring.cloud.sqs.CompletableFutures;
@@ -45,6 +46,7 @@
import io.awspring.cloud.sqs.listener.sink.MessageSink;
import io.awspring.cloud.sqs.listener.source.AbstractSqsMessageSource;
import io.awspring.cloud.sqs.listener.source.MessageSource;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
import java.lang.reflect.Method;
import java.time.Duration;
import java.util.ArrayList;
@@ -52,8 +54,10 @@
import java.util.Collections;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
@@ -73,18 +77,18 @@
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
+import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.context.TestPropertySource;
import org.springframework.util.Assert;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
-import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequest;
-import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry;
/**
* Integration tests for SQS integration.
*
* @author Tomaz Fernandes
+ * @author Mikhail Strokov
*/
@SpringBootTest
@TestPropertySource(properties = { "property.one=1", "property.five.seconds=5s",
@@ -94,8 +98,6 @@ class SqsIntegrationTests extends BaseSqsIntegrationTest {
private static final Logger logger = LoggerFactory.getLogger(SqsIntegrationTests.class);
- private static final String TEST_SQS_ASYNC_CLIENT_BEAN_NAME = "testSqsAsyncClient";
-
static final String RECEIVES_MESSAGE_QUEUE_NAME = "receives_message_test_queue";
static final String RECEIVES_MESSAGE_BATCH_QUEUE_NAME = "receives_message_batch_test_queue";
@@ -116,8 +118,12 @@ class SqsIntegrationTests extends BaseSqsIntegrationTest {
static final String MANUALLY_CREATE_CONTAINER_QUEUE_NAME = "manually_create_container_test_queue";
+ static final String MANUALLY_CREATE_INACTIVE_CONTAINER_QUEUE_NAME = "manually_create_inactive_container_test_queue";
+
static final String MANUALLY_CREATE_FACTORY_QUEUE_NAME = "manually_create_factory_test_queue";
+ static final String MAX_CONCURRENT_MESSAGES_QUEUE_NAME = "max_concurrent_messages_test_queue";
+
static final String LOW_RESOURCE_FACTORY = "lowResourceFactory";
static final String MANUAL_ACK_FACTORY = "manualAcknowledgementFactory";
@@ -141,22 +147,29 @@ static void beforeTests() {
createQueue(client, RESOLVES_PARAMETER_TYPES_QUEUE_NAME,
singletonMap(QueueAttributeName.VISIBILITY_TIMEOUT, "20")),
createQueue(client, MANUALLY_CREATE_CONTAINER_QUEUE_NAME),
- createQueue(client, MANUALLY_CREATE_FACTORY_QUEUE_NAME)).join();
+ createQueue(client, MANUALLY_CREATE_INACTIVE_CONTAINER_QUEUE_NAME),
+ createQueue(client, MANUALLY_CREATE_FACTORY_QUEUE_NAME),
+ createQueue(client, MAX_CONCURRENT_MESSAGES_QUEUE_NAME)).join();
}
@Autowired
LatchContainer latchContainer;
@Autowired
- @Qualifier(TEST_SQS_ASYNC_CLIENT_BEAN_NAME)
- SqsAsyncClient sqsAsyncClient;
+ SqsTemplate sqsTemplate;
@Autowired
ObjectMapper objectMapper;
+ @Autowired
+ @Qualifier("inactiveContainer")
+ MessageListenerContainer inactiveMessageListenerContainer;
+
@Test
void receivesMessage() throws Exception {
- sendMessageTo(RECEIVES_MESSAGE_QUEUE_NAME, "receivesMessage-payload");
+ String messageBody = "receivesMessage-payload";
+ sqsTemplate.send(RECEIVES_MESSAGE_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", RECEIVES_MESSAGE_QUEUE_NAME, messageBody);
assertThat(latchContainer.receivesMessageLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchContainer.invocableHandlerMethodLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchContainer.acknowledgementCallbackSuccessLatch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -164,59 +177,88 @@ void receivesMessage() throws Exception {
@Test
void receivesMessageBatch() throws Exception {
- sendMessageTo(RECEIVES_MESSAGE_BATCH_QUEUE_NAME, "receivesMessageBatch-payload");
+ String messageBody = "receivesMessageBatch-payload";
+ sqsTemplate.send(RECEIVES_MESSAGE_BATCH_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", RECEIVES_MESSAGE_BATCH_QUEUE_NAME, messageBody);
assertThat(latchContainer.receivesMessageBatchLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchContainer.acknowledgementCallbackBatchLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void receivesMessageAsync() throws Exception {
- sendMessageTo(RECEIVES_MESSAGE_ASYNC_QUEUE_NAME, "receivesMessageAsync-payload");
+ String messageBody = "receivesMessageAsync-payload";
+ sqsTemplate.send(RECEIVES_MESSAGE_ASYNC_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", RECEIVES_MESSAGE_ASYNC_QUEUE_NAME, messageBody);
assertThat(latchContainer.receivesMessageAsyncLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void doesNotAckOnError() throws Exception {
- sendMessageTo(DOES_NOT_ACK_ON_ERROR_QUEUE_NAME, "doesNotAckOnError-payload");
+ String messageBody = "doesNotAckOnError-payload";
+ sqsTemplate.send(DOES_NOT_ACK_ON_ERROR_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", DOES_NOT_ACK_ON_ERROR_QUEUE_NAME, messageBody);
assertThat(latchContainer.doesNotAckLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchContainer.acknowledgementCallbackErrorLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void doesNotAckOnErrorAsync() throws Exception {
- sendMessageTo(DOES_NOT_ACK_ON_ERROR_ASYNC_QUEUE_NAME, "doesNotAckOnErrorAsync-payload");
+ String messageBody = "doesNotAckOnErrorAsync-payload";
+ sqsTemplate.send(DOES_NOT_ACK_ON_ERROR_ASYNC_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", DOES_NOT_ACK_ON_ERROR_ASYNC_QUEUE_NAME,
+ messageBody);
assertThat(latchContainer.doesNotAckAsyncLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void doesNotAckOnErrorBatch() throws Exception {
- List values = IntStream.range(0, 10).mapToObj(index -> "doesNotAckOnErrorBatch-payload-" + index)
- .collect(Collectors.toList());
- sendMessageBatch(DOES_NOT_ACK_ON_ERROR_BATCH_QUEUE_NAME, values);
+ List> messages = IntStream.range(0, 10)
+ .mapToObj(index -> "doesNotAckOnErrorBatch-payload-" + index)
+ .map(payload -> MessageBuilder.withPayload(payload).build()).collect(Collectors.toList());
+ sqsTemplate.sendManyAsync(DOES_NOT_ACK_ON_ERROR_BATCH_QUEUE_NAME, messages);
+ logger.debug("Sent messages to queue {} with messages {}", DOES_NOT_ACK_ON_ERROR_BATCH_QUEUE_NAME, messages);
assertThat(latchContainer.doesNotAckBatchLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void doesNotAckOnErrorBatchAsync() throws Exception {
- List values = IntStream.range(0, 10).mapToObj(index -> "doesNotAckOnErrorBatchAsync-payload-" + index)
- .collect(Collectors.toList());
- sendMessageBatch(DOES_NOT_ACK_ON_ERROR_BATCH_ASYNC_QUEUE_NAME, values);
+ List> messages = IntStream.range(0, 10)
+ .mapToObj(index -> "doesNotAckOnErrorBatchAsync-payload-" + index)
+ .map(payload -> MessageBuilder.withPayload(payload).build()).collect(Collectors.toList());
+ sqsTemplate.sendManyAsync(DOES_NOT_ACK_ON_ERROR_BATCH_ASYNC_QUEUE_NAME, messages);
+ logger.debug("Sent messages to queue {} with messages {}", DOES_NOT_ACK_ON_ERROR_BATCH_ASYNC_QUEUE_NAME,
+ messages);
assertThat(latchContainer.doesNotAckBatchAsyncLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void resolvesManyParameterTypes() throws Exception {
- sendMessageTo(RESOLVES_PARAMETER_TYPES_QUEUE_NAME, "many-parameter-types-payload");
+ String messageBody = "many-parameter-types-payload";
+ sqsTemplate.send(RESOLVES_PARAMETER_TYPES_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", RESOLVES_PARAMETER_TYPES_QUEUE_NAME, messageBody);
assertThat(latchContainer.manyParameterTypesLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchContainer.manyParameterTypesSecondLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void manuallyCreatesContainer() throws Exception {
- sendMessageTo(MANUALLY_CREATE_CONTAINER_QUEUE_NAME, "Testing manually creates container");
+ String messageBody = "Testing manually creates container";
+ sqsTemplate.send(MANUALLY_CREATE_CONTAINER_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", MANUALLY_CREATE_CONTAINER_QUEUE_NAME, messageBody);
assertThat(latchContainer.manuallyCreatedContainerLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
+ @Test
+ void manuallyCreatesInactiveContainer() throws Exception {
+ String messageBody = "Testing manually creates inactive container";
+ assertThat(inactiveMessageListenerContainer.isRunning()).isFalse();
+ sqsTemplate.send(MANUALLY_CREATE_INACTIVE_CONTAINER_QUEUE_NAME, messageBody);
+ inactiveMessageListenerContainer.start();
+ logger.debug("Sent message to queue {} with messageBody {}", MANUALLY_CREATE_INACTIVE_CONTAINER_QUEUE_NAME,
+ messageBody);
+ assertThat(latchContainer.manuallyInactiveCreatedContainerLatch.await(10, TimeUnit.SECONDS)).isTrue();
+ }
+
// @formatter:off
@Test
void manuallyStartsContainerAndChangesComponent() throws Exception {
@@ -230,7 +272,9 @@ void manuallyStartsContainerAndChangesComponent() throws Exception {
.pollTimeout(Duration.ofSeconds(3)))
.build();
container.start();
- sendMessageTo(MANUALLY_START_CONTAINER, "MyTest");
+ String messageBody1 = "MyTest";
+ sqsTemplate.send(MANUALLY_START_CONTAINER, messageBody1);
+ logger.debug("Sent message to queue {} with messageBody {}", MANUALLY_START_CONTAINER, messageBody1);
assertThat(latchContainer.manuallyStartedContainerLatch.await(10, TimeUnit.SECONDS)).isTrue();
container.stop();
container.setMessageListener(msg -> latchContainer.manuallyStartedContainerLatch2.countDown());
@@ -238,7 +282,9 @@ void manuallyStartsContainerAndChangesComponent() throws Exception {
builder.acknowledgementMode(AcknowledgementMode.ALWAYS);
container.configure(options -> options.fromBuilder(builder));
container.start();
- sendMessageTo(MANUALLY_START_CONTAINER, "MyTest2");
+ String messageBody2 = "MyTest2";
+ sqsTemplate.send(MANUALLY_START_CONTAINER, messageBody2);
+ logger.debug("Sent message to queue {} with messageBody {}", MANUALLY_START_CONTAINER, messageBody2);
assertThat(latchContainer.manuallyStartedContainerLatch2.await(10, TimeUnit.SECONDS)).isTrue();
container.stop();
}
@@ -246,31 +292,26 @@ void manuallyStartsContainerAndChangesComponent() throws Exception {
@Test
void manuallyCreatesFactory() throws Exception {
- sendMessageTo(MANUALLY_CREATE_FACTORY_QUEUE_NAME, "Testing manually creates factory");
+ String messageBody = "Testing manually creates factory";
+ sqsTemplate.send(MANUALLY_CREATE_FACTORY_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", MANUALLY_CREATE_FACTORY_QUEUE_NAME, messageBody);
assertThat(latchContainer.manuallyCreatedFactoryLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchContainer.manuallyCreatedFactorySourceFactoryLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchContainer.manuallyCreatedFactorySinkLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
- private void sendMessageTo(String queueName, String messageBody) {
- String queueUrl = fetchQueueUrl(queueName);
- sqsAsyncClient.sendMessage(req -> req.messageBody(messageBody).queueUrl(queueUrl).build()).join();
- logger.debug("Sent message to queue {} with messageBody {}", queueName, messageBody);
- }
-
- private void sendMessageBatch(String queueName, Collection messageBodies) {
- String queueUrl = fetchQueueUrl(queueName);
- sqsAsyncClient.sendMessageBatch(SendMessageBatchRequest.builder().queueUrl(queueUrl)
- .entries(messageBodies.stream()
- .map(payload -> SendMessageBatchRequestEntry.builder().messageBody(payload)
- .id(UUID.randomUUID().toString()).build())
- .collect(Collectors.toList()))
- .build()).join();
- logger.debug("Sent messages to queue {} with messageBodies {}", queueName, messageBodies);
- }
-
- private String fetchQueueUrl(String receivesMessageQueueName) {
- return sqsAsyncClient.getQueueUrl(req -> req.queueName(receivesMessageQueueName)).join().queueUrl();
+ @Test
+ void maxConcurrentMessages() {
+ List> messages1 = IntStream.range(0, 10)
+ .mapToObj(index -> "maxConcurrentMessages-payload-" + index)
+ .map(payload -> MessageBuilder.withPayload(payload).build()).collect(Collectors.toList());
+ List> messages2 = IntStream.range(10, 20)
+ .mapToObj(index -> "maxConcurrentMessages-payload-" + index)
+ .map(payload -> MessageBuilder.withPayload(payload).build()).collect(Collectors.toList());
+ sqsTemplate.sendManyAsync(MAX_CONCURRENT_MESSAGES_QUEUE_NAME, messages1);
+ sqsTemplate.sendManyAsync(MAX_CONCURRENT_MESSAGES_QUEUE_NAME, messages2);
+ logger.debug("Sent messages to queue {} with messages {} and {}", MAX_CONCURRENT_MESSAGES_QUEUE_NAME, messages1, messages2);
+ assertDoesNotThrow(() -> latchContainer.maxConcurrentMessagesBarrier.await(10, TimeUnit.SECONDS));
}
static class ReceivesMessageListener {
@@ -397,6 +438,18 @@ void listen(Message message, MessageHeaders headers, Acknowledgement ack
}
}
+ static class MaxConcurrentMessagesListener {
+
+ @Autowired
+ LatchContainer latchContainer;
+
+ @SqsListener(queueNames = MAX_CONCURRENT_MESSAGES_QUEUE_NAME, maxMessagesPerPoll = "10", maxConcurrentMessages = "20", id = "max-concurrent-messages")
+ void listen(String message) throws BrokenBarrierException, InterruptedException {
+ logger.debug("Received message in Listener Method: " + message);
+ latchContainer.maxConcurrentMessagesBarrier.await();
+ }
+ }
+
static class LatchContainer {
final CountDownLatch receivesMessageLatch = new CountDownLatch(1);
@@ -419,6 +472,8 @@ static class LatchContainer {
final CountDownLatch acknowledgementCallbackSuccessLatch = new CountDownLatch(1);
final CountDownLatch acknowledgementCallbackBatchLatch = new CountDownLatch(1);
final CountDownLatch acknowledgementCallbackErrorLatch = new CountDownLatch(1);
+ final CountDownLatch manuallyInactiveCreatedContainerLatch = new CountDownLatch(1);
+ final CyclicBarrier maxConcurrentMessagesBarrier = new CyclicBarrier(21);
}
@@ -526,7 +581,8 @@ public void onSuccess(Collection> messages) {
}
@Bean
- public MessageListenerContainer manuallyCreatedContainer(SqsAsyncClient client) throws Exception {
+ public MessageListenerContainer manuallyCreatedContainer() throws Exception {
+ SqsAsyncClient client = BaseSqsIntegrationTest.createAsyncClient();
String queueUrl = client.getQueueUrl(req -> req.queueName(MANUALLY_CREATE_CONTAINER_QUEUE_NAME)).get()
.queueUrl();
return SqsMessageListenerContainer
@@ -540,6 +596,23 @@ public MessageListenerContainer manuallyCreatedContainer(SqsAsyncClient
.build();
}
+ @Bean("inactiveContainer")
+ public MessageListenerContainer manuallyCreatedInactiveContainer() throws Exception {
+ SqsAsyncClient client = BaseSqsIntegrationTest.createAsyncClient();
+ String queueUrl = client.getQueueUrl(req -> req.queueName(MANUALLY_CREATE_INACTIVE_CONTAINER_QUEUE_NAME)).get()
+ .queueUrl();
+ return SqsMessageListenerContainer
+ .builder()
+ .queueNames(queueUrl)
+ .sqsAsyncClient(client)
+ .configure(options -> options
+ .autoStartup(false)
+ .maxDelayBetweenPolls(Duration.ofSeconds(1))
+ .pollTimeout(Duration.ofSeconds(3)))
+ .messageListener(msg -> {latchContainer.manuallyInactiveCreatedContainerLatch.countDown();})
+ .build();
+ }
+
@Bean
public SqsMessageListenerContainer manuallyCreatedFactory() {
SqsMessageListenerContainerFactory factory = new SqsMessageListenerContainerFactory<>();
@@ -609,6 +682,11 @@ ResolvesParameterTypesListener resolvesParameterTypesListener() {
return new ResolvesParameterTypesListener();
}
+ @Bean
+ MaxConcurrentMessagesListener maxConcurrentMessagesListener() {
+ return new MaxConcurrentMessagesListener();
+ }
+
@Bean
SqsListenerConfigurer customizer() {
return registrar -> {
@@ -632,9 +710,9 @@ ObjectMapper objectMapper() {
return new ObjectMapper();
}
- @Bean(name = TEST_SQS_ASYNC_CLIENT_BEAN_NAME)
- SqsAsyncClient sqsAsyncClientProducer() {
- return BaseSqsIntegrationTest.createHighThroughputAsyncClient();
+ @Bean
+ SqsTemplate sqsTemplate() {
+ return SqsTemplate.builder().sqsAsyncClient(BaseSqsIntegrationTest.createAsyncClient()).build();
}
private AsyncMessageInterceptor testInterceptor() {
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsInterceptorIntegrationTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsInterceptorIntegrationTests.java
index c84ef9f63..9ce4631fe 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsInterceptorIntegrationTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsInterceptorIntegrationTests.java
@@ -28,6 +28,7 @@
import io.awspring.cloud.sqs.listener.acknowledgement.handler.AcknowledgementMode;
import io.awspring.cloud.sqs.listener.errorhandler.AsyncErrorHandler;
import io.awspring.cloud.sqs.listener.interceptor.AsyncMessageInterceptor;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
import io.awspring.cloud.sqs.support.converter.MessagingMessageHeaders;
import java.time.Duration;
import java.util.ArrayList;
@@ -36,14 +37,12 @@
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -58,14 +57,13 @@
* Integration tests for SQS interceptors.
*
* @author Tomaz Fernandes
+ * @author Mikhail Strokov
*/
@SpringBootTest
class SqsInterceptorIntegrationTests extends BaseSqsIntegrationTest {
private static final Logger logger = LoggerFactory.getLogger(SqsInterceptorIntegrationTests.class);
- private static final String TEST_SQS_ASYNC_CLIENT_BEAN_NAME = "testSqsAsyncClient";
-
static final String RECEIVES_CHANGED_MESSAGE_ON_COMPONENTS_QUEUE_NAME = "receives_changed_message_on_components_test_queue";
static final String RECEIVES_CHANGED_MESSAGE_ON_ERROR_QUEUE_NAME = "receives_changed_message_on_error_test_queue";
@@ -87,35 +85,28 @@ static void beforeTests() {
LatchContainer latchContainer;
@Autowired
- @Qualifier(TEST_SQS_ASYNC_CLIENT_BEAN_NAME)
- SqsAsyncClient sqsAsyncClient;
+ SqsTemplate sqsTemplate;
@Autowired(required = false)
ReceivesChangedPayloadListener receivesChangedPayloadListener;
@Test
void shouldReceiveChangedMessageOnComponents() throws Exception {
- sendMessageTo(RECEIVES_CHANGED_MESSAGE_ON_COMPONENTS_QUEUE_NAME, SHOULD_CHANGE_PAYLOAD);
+ sqsTemplate.send(RECEIVES_CHANGED_MESSAGE_ON_COMPONENTS_QUEUE_NAME, SHOULD_CHANGE_PAYLOAD);
+ logger.debug("Sent message to queue {} with messageBody {}", RECEIVES_CHANGED_MESSAGE_ON_COMPONENTS_QUEUE_NAME,
+ SHOULD_CHANGE_PAYLOAD);
assertThat(latchContainer.receivesChangedMessageLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(receivesChangedPayloadListener.receivedMessages).containsExactly(CHANGED_PAYLOAD);
}
@Test
void shouldReceiveChangedMessageOnComponentsWhenError() throws Exception {
- sendMessageTo(RECEIVES_CHANGED_MESSAGE_ON_ERROR_QUEUE_NAME, SHOULD_CHANGE_PAYLOAD);
+ sqsTemplate.send(RECEIVES_CHANGED_MESSAGE_ON_ERROR_QUEUE_NAME, SHOULD_CHANGE_PAYLOAD);
+ logger.debug("Sent message to queue {} with messageBody {}", RECEIVES_CHANGED_MESSAGE_ON_ERROR_QUEUE_NAME,
+ SHOULD_CHANGE_PAYLOAD);
assertThat(latchContainer.receivesChangedMessageOnErrorLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
- private void sendMessageTo(String queueName, String messageBody) throws InterruptedException, ExecutionException {
- String queueUrl = fetchQueueUrl(queueName);
- sqsAsyncClient.sendMessage(req -> req.messageBody(messageBody).queueUrl(queueUrl).build()).get();
- logger.debug("Sent message to queue {} with messageBody {}", queueName, messageBody);
- }
-
- private String fetchQueueUrl(String receivesMessageQueueName) throws InterruptedException, ExecutionException {
- return sqsAsyncClient.getQueueUrl(req -> req.queueName(receivesMessageQueueName)).get().queueUrl();
- }
-
static class ReceivesChangedPayloadListener {
@Autowired
@@ -199,9 +190,9 @@ LatchContainer latchContainer() {
return this.latchContainer;
}
- @Bean(name = TEST_SQS_ASYNC_CLIENT_BEAN_NAME)
- SqsAsyncClient sqsAsyncClientProducer() {
- return BaseSqsIntegrationTest.createAsyncClient();
+ @Bean
+ SqsTemplate sqsTemplate() {
+ return SqsTemplate.builder().sqsAsyncClient(BaseSqsIntegrationTest.createAsyncClient()).build();
}
private AsyncMessageInterceptor getMessageInterceptor() {
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsLoadIntegrationTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsLoadIntegrationTests.java
index d766a2907..8c3364893 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsLoadIntegrationTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsLoadIntegrationTests.java
@@ -18,7 +18,6 @@
import static java.util.Collections.singletonMap;
import static org.assertj.core.api.Assertions.assertThat;
-import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.awspring.cloud.sqs.MessageHeaderUtils;
import io.awspring.cloud.sqs.annotation.SqsListener;
@@ -32,12 +31,12 @@
import io.awspring.cloud.sqs.listener.acknowledgement.SqsAcknowledgementExecutor;
import io.awspring.cloud.sqs.listener.source.AbstractSqsMessageSource;
import io.awspring.cloud.sqs.listener.source.MessageSource;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
-import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
@@ -51,22 +50,22 @@
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.StopWatch;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
-import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry;
/**
* Load test for SQS integration.
*
* @author Tomaz Fernandes
+ * @author Mikhail Strokov
*/
@SpringBootTest
class SqsLoadIntegrationTests extends BaseSqsIntegrationTest {
@@ -81,19 +80,13 @@ class SqsLoadIntegrationTests extends BaseSqsIntegrationTest {
private static final String RECEIVE_BATCH_2_QUEUE_NAME = "receive_batch_test_queue_2";
- private static final String TEST_SQS_ASYNC_CLIENT_BEAN_NAME = "testSqsAsyncClient";
-
private static final String HIGH_THROUGHPUT_FACTORY_NAME = "highThroughputFactory";
@Autowired
LatchContainer latchContainer;
@Autowired
- @Qualifier(TEST_SQS_ASYNC_CLIENT_BEAN_NAME)
- SqsAsyncClient sqsAsyncClient;
-
- @Autowired
- ObjectMapper objectMapper;
+ SqsTemplate sqsTemplate;
@Autowired
Settings settings;
@@ -164,16 +157,14 @@ private void testWithLoad(String queue1, String queue2, Collection recei
CountDownLatch listenerLatch, CountDownLatch acknowledgementLatch)
throws InterruptedException, ExecutionException {
Assert.isTrue(settings.totalMessages >= 20, "Minimum of 20 messages");
- String queueUrl1 = fetchQueueUrl(queue1);
- String queueUrl2 = fetchQueueUrl(queue2);
LoadSimulator sendLoadSimulator = new LoadSimulator();
sendLoadSimulator.setLoadEnabled(settings.totalMessages > 1000);
logger.debug("Starting watch");
StopWatch watch = new StopWatch();
watch.start();
IntStream.range(0, Math.max(settings.totalMessages / 20, 1)).forEach(index -> {
- sendMessageBatchAsync(queueUrl1);
- sendMessageBatchAsync(queueUrl2);
+ sendMessageBatchAsync(queue1);
+ sendMessageBatchAsync(queue2);
if (index % 20 == 0) {
sendLoadSimulator.runLoad(50);
}
@@ -203,21 +194,20 @@ private void testWithLoad(String queue1, String queue2, Collection recei
AtomicInteger bodyInteger = new AtomicInteger();
- private void sendMessageBatchAsync(String queueUrl) {
+ private void sendMessageBatchAsync(String queueName) {
if (!settings.sendMessages) {
return;
}
- Collection batchEntries = getBatchEntries();
- doSendMessageBatch(queueUrl, batchEntries);
+ Collection> messages = getMessages();
+ doSendMessageBatch(queueName, messages);
}
- private void doSendMessageBatch(String queueUrl, Collection batchEntries) {
- sqsAsyncClient.sendMessageBatch(req -> req.entries(batchEntries).queueUrl(queueUrl).build())
- .thenRun(this::logSend).exceptionally(t -> {
- logger.error("Error sending messages - retrying", t);
- doSendMessageBatch(queueUrl, batchEntries);
- return null;
- });
+ private void doSendMessageBatch(String queueName, Collection> messages) {
+ sqsTemplate.sendManyAsync(queueName, messages).thenRun(this::logSend).exceptionally(t -> {
+ logger.error("Error sending messages - retrying", t);
+ doSendMessageBatch(queueName, messages);
+ return null;
+ });
}
private void logSend() {
@@ -227,26 +217,16 @@ private void logSend() {
}
}
- private Collection getBatchEntries() {
+ private Collection> getMessages() {
return IntStream.range(0, Math.min(settings.totalMessages / 2, 10)).mapToObj(index -> {
- String id = UUID.randomUUID().toString();
- logger.trace("Sending message with id {}", id);
- return SendMessageBatchRequestEntry.builder().id(id).messageBody(getBody()).build();
+ Message message = MessageBuilder.withPayload(getBody()).build();
+ logger.trace("Sending message with id {}", message.getHeaders().get("id"));
+ return message;
}).collect(Collectors.toList());
}
- private String getBody() {
- try {
- return this.objectMapper.writeValueAsString(
- new MyPojo("MyPojo - " + bodyInteger.incrementAndGet(), "MyPojo - secondValue"));
- }
- catch (JsonProcessingException e) {
- throw new RuntimeException(e);
- }
- }
-
- private String fetchQueueUrl(String receivesMessageQueueName) throws InterruptedException, ExecutionException {
- return sqsAsyncClient.getQueueUrl(req -> req.queueName(receivesMessageQueueName)).get().queueUrl();
+ private Object getBody() {
+ return new MyPojo("MyPojo - " + bodyInteger.incrementAndGet(), "MyPojo - secondValue");
}
static class MessageContainer {
@@ -276,7 +256,7 @@ static class ReceiveManyFromTwoQueuesListener {
@SqsListener(queueNames = { RECEIVE_FROM_MANY_1_QUEUE_NAME,
RECEIVE_FROM_MANY_2_QUEUE_NAME }, factory = HIGH_THROUGHPUT_FACTORY_NAME, id = "many-from-two-queues")
- void listen(Message message) throws Exception {
+ void listen(Message message) throws Exception {
logger.trace("Started processing {}", MessageHeaderUtils.getId(message));
if (this.messageContainer.receivedByListener.contains(MessageHeaderUtils.getId(message))) {
logger.warn("Received duplicated message: {}", message);
@@ -404,9 +384,10 @@ ObjectMapper objectMapper() {
return new ObjectMapper();
}
- @Bean(name = TEST_SQS_ASYNC_CLIENT_BEAN_NAME)
- SqsAsyncClient sqsAsyncClientProducer() {
- return BaseSqsIntegrationTest.createHighThroughputAsyncClient();
+ @Bean
+ SqsTemplate sqsTemplate() {
+ return SqsTemplate.builder().sqsAsyncClient(BaseSqsIntegrationTest.createHighThroughputAsyncClient())
+ .build();
}
private final AtomicInteger acks = new AtomicInteger();
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsMessageConversionIntegrationTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsMessageConversionIntegrationTests.java
index 769717925..12a28b805 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsMessageConversionIntegrationTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/integration/SqsMessageConversionIntegrationTests.java
@@ -17,20 +17,20 @@
import static org.assertj.core.api.Assertions.assertThat;
-import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.awspring.cloud.sqs.annotation.SqsListener;
import io.awspring.cloud.sqs.config.SqsBootstrapConfiguration;
import io.awspring.cloud.sqs.config.SqsMessageListenerContainerFactory;
import io.awspring.cloud.sqs.listener.SqsHeaders;
import io.awspring.cloud.sqs.listener.interceptor.AsyncMessageInterceptor;
+import io.awspring.cloud.sqs.operations.SqsTemplate;
+import io.awspring.cloud.sqs.support.converter.MessagingMessageHeaders;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -43,15 +43,16 @@
import org.springframework.context.annotation.Import;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
-import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
/**
* Integration tests for SQS message conversion.
*
* @author Tomaz Fernandes
+ * @author Mikhail Strokov
*/
@SpringBootTest
class SqsMessageConversionIntegrationTests extends BaseSqsIntegrationTest {
@@ -69,10 +70,7 @@ class SqsMessageConversionIntegrationTests extends BaseSqsIntegrationTest {
LatchContainer latchContainer;
@Autowired
- SqsAsyncClient sqsAsyncClient;
-
- @Autowired
- ObjectMapper objectMapper;
+ SqsTemplate sqsTemplate;
@BeforeAll
static void beforeTests() {
@@ -87,45 +85,58 @@ static void beforeTests() {
@Test
void resolvesPojoParameterTypes() throws Exception {
- sendMessageTo(RESOLVES_POJO_TYPES_QUEUE_NAME, new MyPojo("pojoParameterType", "secondValue"));
+ MyPojo messageBody = new MyPojo("pojoParameterType", "secondValue");
+ sqsTemplate.send(RESOLVES_POJO_TYPES_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", RESOLVES_POJO_TYPES_QUEUE_NAME, messageBody);
assertThat(latchContainer.resolvesPojoLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void resolvesPojoMessage() throws Exception {
- sendMessageTo(RESOLVES_POJO_MESSAGE_QUEUE_NAME, new MyPojo("resolvesPojoMessage", "secondValue"));
+ MyPojo messageBody = new MyPojo("resolvesPojoMessage", "secondValue");
+ sqsTemplate.send(RESOLVES_POJO_MESSAGE_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", RESOLVES_POJO_MESSAGE_QUEUE_NAME, messageBody);
assertThat(latchContainer.resolvesPojoMessageLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void resolvesPojoList() throws Exception {
- sendMessageTo(RESOLVES_POJO_LIST_QUEUE_NAME, new MyPojo("resolvesPojoList", "secondValue"));
+ MyPojo payload = new MyPojo("resolvesPojoList", "secondValue");
+ sqsTemplate.send(RESOLVES_POJO_LIST_QUEUE_NAME, payload);
+ logger.debug("Sent message to queue {} with messageBody {}", RESOLVES_POJO_LIST_QUEUE_NAME, payload);
assertThat(latchContainer.resolvesPojoListLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void resolvesPojoMessageList() throws Exception {
- sendMessageTo(RESOLVES_POJO_MESSAGE_LIST_QUEUE_NAME, new MyPojo("resolvesPojoMessageList", "secondValue"));
+ MyPojo messageBody = new MyPojo("resolvesPojoMessageList", "secondValue");
+ sqsTemplate.send(RESOLVES_POJO_MESSAGE_LIST_QUEUE_NAME, messageBody);
+ logger.debug("Sent message to queue {} with messageBody {}", RESOLVES_POJO_MESSAGE_LIST_QUEUE_NAME,
+ messageBody);
assertThat(latchContainer.resolvesPojoMessageListLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void resolvesPojoFromHeader() throws Exception {
- sendMessageTo(RESOLVES_POJO_FROM_HEADER_QUEUE_NAME, new MyPojo("pojoParameterType", "secondValue"),
- getHeaderMapping(MyPojo.class));
+ MyPojo payload = new MyPojo("pojoParameterType", "secondValue");
+ sqsTemplate.send(RESOLVES_POJO_FROM_HEADER_QUEUE_NAME,
+ MessageBuilder.createMessage(payload, new MessagingMessageHeaders(getHeaderMapping(MyPojo.class))));
+ logger.debug("Sent message to queue {} with messageBody {}", RESOLVES_POJO_FROM_HEADER_QUEUE_NAME, payload);
assertThat(latchContainer.resolvesPojoFromMappingLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void resolvesMyOtherPojoFromHeader() throws Exception {
- sendMessageTo(RESOLVES_MY_OTHER_POJO_FROM_HEADER_QUEUE_NAME,
- new MyOtherPojo("pojoParameterType", "secondValue"), getHeaderMapping(MyOtherPojo.class));
+ MyOtherPojo payload = new MyOtherPojo("pojoParameterType", "secondValue");
+ sqsTemplate.send(RESOLVES_MY_OTHER_POJO_FROM_HEADER_QUEUE_NAME, MessageBuilder.createMessage(payload,
+ new MessagingMessageHeaders(getHeaderMapping(MyOtherPojo.class))));
+ logger.debug("Sent message to queue {} with messageBody {}", RESOLVES_MY_OTHER_POJO_FROM_HEADER_QUEUE_NAME,
+ payload);
assertThat(latchContainer.resolvesMyOtherPojoFromMappingLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
- private Map getHeaderMapping(Class clazz) {
- return Collections.singletonMap(SqsHeaders.SQS_DEFAULT_TYPE_HEADER,
- MessageAttributeValue.builder().stringValue(clazz.getName()).dataType("String").build());
+ private Map getHeaderMapping(Class clazz) {
+ return Collections.singletonMap(SqsHeaders.SQS_DEFAULT_TYPE_HEADER, clazz.getName());
}
static class ResolvesPojoListener {
@@ -305,29 +316,10 @@ ObjectMapper objectMapper() {
}
@Bean
- SqsAsyncClient sqsAsyncClientProducer() {
- return BaseSqsIntegrationTest.createHighThroughputAsyncClient();
+ SqsTemplate sqsTemplate() {
+ return SqsTemplate.builder().sqsAsyncClient(BaseSqsIntegrationTest.createAsyncClient()).build();
}
- }
-
- private void sendMessageTo(String queueName, Object messageBody)
- throws InterruptedException, ExecutionException, JsonProcessingException {
- String queueUrl = sqsAsyncClient.getQueueUrl(req -> req.queueName(queueName)).get().queueUrl();
- String payload = objectMapper.writeValueAsString(messageBody);
- sqsAsyncClient.sendMessage(req -> req.messageBody(payload).queueUrl(queueUrl).build()).get();
- logger.debug("Sent message to queue {} with messageBody {}", queueName, messageBody);
- }
- private void sendMessageTo(String queueName, Object messageBody,
- Map messageAttributes)
- throws InterruptedException, ExecutionException, JsonProcessingException {
- String queueUrl = sqsAsyncClient.getQueueUrl(req -> req.queueName(queueName)).get().queueUrl();
- String payload = objectMapper.writeValueAsString(messageBody);
- sqsAsyncClient
- .sendMessage(
- req -> req.messageBody(payload).queueUrl(queueUrl).messageAttributes(messageAttributes).build())
- .get();
- logger.debug("Sent message to queue {} with messageBody {}", queueName, messageBody);
}
static class MyPojo implements MyInterface {
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/AbstractMessageListenerContainerTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/AbstractMessageListenerContainerTests.java
index 999b72a5e..f7e450286 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/AbstractMessageListenerContainerTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/AbstractMessageListenerContainerTests.java
@@ -73,6 +73,7 @@ void shouldAdaptBlockingComponents() {
.isInstanceOf(AsyncComponentAdapters.AbstractThreadingComponentAdapter.class)
.extracting("blockingMessageInterceptor").isEqualTo(interceptor);
+ assertThat(container.getPhase()).isEqualTo(MessageListenerContainer.DEFAULT_PHASE);
}
@Test
@@ -101,6 +102,7 @@ void shouldSetAsyncComponents() {
assertThat(container.getAcknowledgementResultCallback()).isEqualTo(callback);
assertThat(container.getContainerComponentFactories()).containsExactlyElementsOf(componentFactories);
assertThat(container.getMessageInterceptors()).containsExactly(interceptor);
+ assertThat(container.getPhase()).isEqualTo(MessageListenerContainer.DEFAULT_PHASE);
}
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/DefaultListenerContainerRegistryTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/DefaultListenerContainerRegistryTests.java
index e4e7c34e3..b05426ca5 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/DefaultListenerContainerRegistryTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/DefaultListenerContainerRegistryTests.java
@@ -20,6 +20,7 @@
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
import org.junit.jupiter.api.Test;
@@ -38,6 +39,7 @@ void shouldRegisterListenerContainer() {
given(container.getId()).willReturn(id);
DefaultListenerContainerRegistry registry = new DefaultListenerContainerRegistry();
registry.registerListenerContainer(container);
+ assertThat(registry.getPhase()).isEqualTo(MessageListenerContainer.DEFAULT_PHASE);
}
@Test
@@ -46,9 +48,11 @@ void shouldGetListenerContainer() {
String id = "test-container-id";
given(container.getId()).willReturn(id);
DefaultListenerContainerRegistry registry = new DefaultListenerContainerRegistry();
+ registry.setPhase(2);
registry.registerListenerContainer(container);
MessageListenerContainer containerFromRegistry = registry.getContainerById(id);
assertThat(containerFromRegistry).isEqualTo(container);
+ assertThat(registry.getPhase()).isEqualTo(2);
}
@Test
@@ -78,8 +82,11 @@ void shouldStartAndStopAllListenerContainers() {
String id2 = "test-container-id-2";
String id3 = "test-container-id-3";
given(container1.getId()).willReturn(id1);
+ given(container1.isAutoStartup()).willReturn(true);
given(container2.getId()).willReturn(id2);
+ given(container2.isAutoStartup()).willReturn(true);
given(container3.getId()).willReturn(id3);
+ given(container3.isAutoStartup()).willReturn(true);
DefaultListenerContainerRegistry registry = new DefaultListenerContainerRegistry();
registry.registerListenerContainer(container1);
registry.registerListenerContainer(container2);
@@ -96,6 +103,21 @@ void shouldStartAndStopAllListenerContainers() {
then(container3).should().stop();
}
+ @Test
+ void shouldNotStartContainerWithAutoStartupFalse() {
+ MessageListenerContainer container1 = mock(MessageListenerContainer.class);
+ String id1 = "test-container-id-1";
+ given(container1.getId()).willReturn(id1);
+ DefaultListenerContainerRegistry registry = new DefaultListenerContainerRegistry();
+ registry.registerListenerContainer(container1);
+ registry.start();
+ assertThat(registry.isRunning()).isTrue();
+ registry.stop();
+ assertThat(registry.isRunning()).isFalse();
+ then(container1).should(times(0)).start();
+ then(container1).should().stop();
+ }
+
@Test
void shouldThrowIfIdAlreadyPresent() {
MessageListenerContainer container = mock(MessageListenerContainer.class);
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/SqsMessageListenerContainerTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/SqsMessageListenerContainerTests.java
index 0dbc77fcf..b4bfaa204 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/SqsMessageListenerContainerTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/SqsMessageListenerContainerTests.java
@@ -61,11 +61,12 @@ void shouldCreateFromBuilderWithBlockingComponents() {
List> componentFactories = Collections
.singletonList(componentFactory);
List queueNames = Arrays.asList("test-queue-name-1", "test-queue-name-2");
+ Integer phase = 2;
SqsMessageListenerContainer container = SqsMessageListenerContainer.builder().messageListener(listener)
.sqsAsyncClient(client).errorHandler(errorHandler).componentFactories(componentFactories)
.acknowledgementResultCallback(callback).messageInterceptor(interceptor1)
- .messageInterceptor(interceptor2).queueNames(queueNames).build();
+ .messageInterceptor(interceptor2).queueNames(queueNames).phase(phase).build();
assertThat(container.getMessageListener())
.isInstanceOf(AsyncComponentAdapters.AbstractThreadingComponentAdapter.class)
@@ -90,6 +91,8 @@ void shouldCreateFromBuilderWithBlockingComponents() {
assertThat(container).extracting("sqsAsyncClient").isEqualTo(client);
assertThat(container.getQueueNames()).containsExactlyElementsOf(queueNames);
+
+ assertThat(container.getPhase()).isEqualTo(phase);
}
@Test
@@ -114,6 +117,7 @@ void shouldCreateFromBuilderWithAsyncComponents() {
assertThat(container.getErrorHandler()).isEqualTo(errorHandler);
assertThat(container.getAcknowledgementResultCallback()).isEqualTo(callback);
assertThat(container.getMessageInterceptors()).containsExactly(interceptor1, interceptor2);
+ assertThat(container.getPhase()).isEqualTo(MessageListenerContainer.DEFAULT_PHASE);
}
@Test
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/source/AbstractPollingMessageSourceTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/source/AbstractPollingMessageSourceTests.java
index a07e9dcad..40cb111c2 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/source/AbstractPollingMessageSourceTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/listener/source/AbstractPollingMessageSourceTests.java
@@ -37,6 +37,7 @@
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
@@ -53,9 +54,8 @@ class AbstractPollingMessageSourceTests {
private static final Logger logger = LoggerFactory.getLogger(AbstractPollingMessageSourceTests.class);
- // @RepeatedTest(400)
@Test
- void shouldAcquireAndReleaseFullPermits() throws Exception {
+ void shouldAcquireAndReleaseFullPermits() {
String testName = "shouldAcquireAndReleaseFullPermits";
SemaphoreBackPressureHandler backPressureHandler = SemaphoreBackPressureHandler.builder()
@@ -73,36 +73,47 @@ void shouldAcquireAndReleaseFullPermits() throws Exception {
@Override
protected CompletableFuture> doPollForMessages(int messagesToRequest) {
- doSleep(100);
- // Since BackPressureMode.ALWAYS_POLL_MAX_MESSAGES, should always be 10.
- assertThat(messagesToRequest).isEqualTo(10);
- assertAvailablePermits(backPressureHandler, 0);
- boolean firstPoll = hasReceived.compareAndSet(false, true);
- if (firstPoll) {
- // No permits released yet, should be TM low
- assertThroughputMode(backPressureHandler, "low");
- }
- else if (hasMadeSecondPoll.compareAndSet(false, true)) {
- // Permits returned, should be high
- assertThroughputMode(backPressureHandler, "high");
- }
- else {
- // Already returned full permits, should be low
- assertThroughputMode(backPressureHandler, "low");
- }
- return CompletableFuture
- .supplyAsync(() -> firstPoll
+ return CompletableFuture.supplyAsync(() -> {
+ try {
+ // Since BackPressureMode.ALWAYS_POLL_MAX_MESSAGES, should always be 10.
+ assertThat(messagesToRequest).isEqualTo(10);
+ assertAvailablePermits(backPressureHandler, 0);
+ boolean firstPoll = hasReceived.compareAndSet(false, true);
+ if (firstPoll) {
+ logger.debug("First poll");
+ // No permits released yet, should be TM low
+ assertThroughputMode(backPressureHandler, "low");
+ }
+ else if (hasMadeSecondPoll.compareAndSet(false, true)) {
+ logger.debug("Second poll");
+ // Permits returned, should be high
+ assertThroughputMode(backPressureHandler, "high");
+ }
+ else {
+ logger.debug("Third poll");
+ // Already returned full permits, should be low
+ assertThroughputMode(backPressureHandler, "low");
+ }
+ return firstPoll
? (Collection) List.of(Message.builder()
.messageId(UUID.randomUUID().toString()).body("message").build())
- : Collections. emptyList(), threadPool)
- .whenComplete((v, t) -> pollingCounter.countDown());
+ : Collections. emptyList();
+ }
+ catch (Throwable t) {
+ logger.error("Error", t);
+ throw new RuntimeException(t);
+ }
+ }, threadPool).whenComplete((v, t) -> {
+ if (t == null) {
+ pollingCounter.countDown();
+ }
+ });
}
};
source.setBackPressureHandler(backPressureHandler);
source.setMessageSink((msgs, context) -> {
assertAvailablePermits(backPressureHandler, 9);
- doSleep(500); // Longer than acquire timout + polling sleep
msgs.forEach(msg -> context.runBackPressureReleaseCallback());
return CompletableFuture.runAsync(processingCounter::countDown);
});
@@ -112,20 +123,23 @@ else if (hasMadeSecondPoll.compareAndSet(false, true)) {
source.setTaskExecutor(createTaskExecutor(testName));
source.setAcknowledgementProcessor(getAcknowledgementProcessor());
source.start();
- assertThat(pollingCounter.await(2, TimeUnit.SECONDS)).isTrue();
- assertThat(processingCounter.await(2, TimeUnit.SECONDS)).isTrue();
+ assertThat(doAwait(pollingCounter)).isTrue();
+ assertThat(doAwait(processingCounter)).isTrue();
}
- // @RepeatedTest(400)
+ private static final AtomicInteger testCounter = new AtomicInteger();
+
@Test
- void shouldAcquireAndReleasePartialPermits() throws Exception {
+ void shouldAcquireAndReleasePartialPermits() {
String testName = "shouldAcquireAndReleasePartialPermits";
SemaphoreBackPressureHandler backPressureHandler = SemaphoreBackPressureHandler.builder()
- .acquireTimeout(Duration.ofMillis(200)).batchSize(10).totalPermits(10)
+ .acquireTimeout(Duration.ofMillis(150)).batchSize(10).totalPermits(10)
.throughputConfiguration(BackPressureMode.AUTO).build();
- ExecutorService threadPool = Executors.newCachedThreadPool();
+ ExecutorService threadPool = Executors
+ .newCachedThreadPool(new MessageExecutionThreadFactory("test " + testCounter.incrementAndGet()));
CountDownLatch pollingCounter = new CountDownLatch(4);
CountDownLatch processingCounter = new CountDownLatch(1);
+ CountDownLatch processingLatch = new CountDownLatch(1);
AtomicBoolean hasThrownError = new AtomicBoolean(false);
AbstractPollingMessageSource source = new AbstractPollingMessageSource<>() {
@@ -138,44 +152,45 @@ void shouldAcquireAndReleasePartialPermits() throws Exception {
@Override
protected CompletableFuture> doPollForMessages(int messagesToRequest) {
- try {
- // Give it some time between returning empty and polling again
- doSleep(100);
-
- // Will only be true the first time it sets hasReceived to true
- boolean shouldReturnMessage = hasReceived.compareAndSet(false, true);
- if (shouldReturnMessage) {
- // First poll, should have 10
- logger.debug("First poll - should request 10 messages");
- assertThat(messagesToRequest).isEqualTo(10);
- assertAvailablePermits(backPressureHandler, 0);
- // No permits have been released yet
- assertThroughputMode(backPressureHandler, "low");
- }
- else if (hasAcquired9.compareAndSet(false, true)) {
- // Second poll, should have 9
- logger.debug("Second poll - should request 9 messages");
- assertThat(messagesToRequest).isEqualTo(9);
- assertAvailablePermitsLessThanOrEqualTo(backPressureHandler, 1);
- // Has released 9 permits, should be TM HIGH
- assertThroughputMode(backPressureHandler, "high");
- }
- else {
- boolean thirdPoll = hasMadeThirdPoll.compareAndSet(false, true);
- // Third poll or later, should have 10 again
- logger.debug("Third poll - should request 10 messages");
- assertThat(messagesToRequest).isEqualTo(10);
- assertAvailablePermits(backPressureHandler, 0);
- if (thirdPoll) {
- // Hasn't yet returned a full batch, should be TM High
+ return CompletableFuture.supplyAsync(() -> {
+ try {
+ // Give it some time between returning empty and polling again
+ // doSleep(100);
+
+ // Will only be true the first time it sets hasReceived to true
+ boolean shouldReturnMessage = hasReceived.compareAndSet(false, true);
+ if (shouldReturnMessage) {
+ // First poll, should have 10
+ logger.debug("First poll - should request 10 messages");
+ assertThat(messagesToRequest).isEqualTo(10);
+ assertAvailablePermits(backPressureHandler, 0);
+ // No permits have been released yet
+ assertThroughputMode(backPressureHandler, "low");
+ }
+ else if (hasAcquired9.compareAndSet(false, true)) {
+ // Second poll, should have 9
+ logger.debug("Second poll - should request 9 messages");
+ assertThat(messagesToRequest).isEqualTo(9);
+ assertAvailablePermitsLessThanOrEqualTo(backPressureHandler, 1);
+ // Has released 9 permits, should be TM HIGH
assertThroughputMode(backPressureHandler, "high");
+ processingLatch.countDown(); // Release processing now
}
else {
- // Has returned all permits in third poll
- assertThroughputMode(backPressureHandler, "low");
+ boolean thirdPoll = hasMadeThirdPoll.compareAndSet(false, true);
+ // Third poll or later, should have 10 again
+ logger.debug("Third poll - should request 10 messages");
+ assertThat(messagesToRequest).isEqualTo(10);
+ assertAvailablePermits(backPressureHandler, 0);
+ if (thirdPoll) {
+ // Hasn't yet returned a full batch, should be TM High
+ assertThroughputMode(backPressureHandler, "high");
+ }
+ else {
+ // Has returned all permits in third poll
+ assertThroughputMode(backPressureHandler, "low");
+ }
}
- }
- return CompletableFuture.supplyAsync(() -> {
if (shouldReturnMessage) {
logger.debug("shouldReturnMessage, returning one message");
return (Collection) List.of(
@@ -183,19 +198,21 @@ else if (hasAcquired9.compareAndSet(false, true)) {
}
logger.debug("should not return message, returning empty list");
return Collections. emptyList();
- }, threadPool).whenComplete((v, t) -> pollingCounter.countDown());
- }
- catch (Error e) {
- hasThrownError.set(true);
- return CompletableFuture.failedFuture(new RuntimeException(e));
- }
+ }
+ catch (Error e) {
+ hasThrownError.set(true);
+ throw new RuntimeException("Error polling for messages", e);
+ }
+ }, threadPool).whenComplete((v, t) -> pollingCounter.countDown());
}
};
source.setBackPressureHandler(backPressureHandler);
source.setMessageSink((msgs, context) -> {
+ logger.debug("Processing {} messages", msgs.size());
assertAvailablePermits(backPressureHandler, 9);
- doSleep(500); // Longer than acquire timout + polling sleep
+ assertThat(doAwait(processingLatch)).isTrue();
+ logger.debug("Finished processing {} messages", msgs.size());
msgs.forEach(msg -> context.runBackPressureReleaseCallback());
return CompletableFuture.completedFuture(null).thenRun(processingCounter::countDown);
});
@@ -204,12 +221,22 @@ else if (hasAcquired9.compareAndSet(false, true)) {
source.setTaskExecutor(createTaskExecutor(testName));
source.setAcknowledgementProcessor(getAcknowledgementProcessor());
source.start();
- assertThat(processingCounter.await(2, TimeUnit.SECONDS)).isTrue();
- assertThat(pollingCounter.await(2, TimeUnit.SECONDS)).isTrue();
+ assertThat(doAwait(processingCounter)).isTrue();
+ assertThat(doAwait(pollingCounter)).isTrue();
source.stop();
assertThat(hasThrownError.get()).isFalse();
}
+ private static boolean doAwait(CountDownLatch processingLatch) {
+ try {
+ return processingLatch.await(4, TimeUnit.SECONDS);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for latch", e);
+ }
+ }
+
private void assertThroughputMode(SemaphoreBackPressureHandler backPressureHandler, String expectedThroughputMode) {
assertThat(ReflectionTestUtils.getField(backPressureHandler, "currentThroughputMode"))
.extracting(Object::toString).extracting(String::toLowerCase)
@@ -243,7 +270,6 @@ protected TaskExecutor createTaskExecutor(String testName) {
int poolSize = 10;
executor.setMaxPoolSize(poolSize);
executor.setCorePoolSize(10);
- // Necessary due to a small racing condition between releasing the permit and releasing the thread.
executor.setQueueCapacity(poolSize);
executor.setAllowCoreThreadTimeOut(true);
executor.setThreadFactory(createThreadFactory(testName));
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/operations/SqsTemplateTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/operations/SqsTemplateTests.java
index 56b124184..0a21fa51b 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/operations/SqsTemplateTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/operations/SqsTemplateTests.java
@@ -40,6 +40,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.function.Consumer;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.messaging.Message;
@@ -70,13 +71,20 @@
@SuppressWarnings("unchecked")
class SqsTemplateTests {
+ SqsAsyncClient mockClient;
+
+ @BeforeEach
+ void beforeEach() {
+ mockClient = mock(SqsAsyncClient.class);
+ }
+
@Test
void shouldSendWithOptions() {
String queue = "test-queue";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
UUID uuid = UUID.randomUUID();
String sequenceNumber = "1234";
SendMessageResponse response = SendMessageResponse.builder().messageId(uuid.toString())
@@ -108,10 +116,10 @@ void shouldSendWithOptions() {
@Test
void shouldSendFifoWithOptions() {
String queue = "test-queue";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
UUID uuid = UUID.randomUUID();
String sequenceNumber = "1234";
SendMessageResponse response = SendMessageResponse.builder().messageId(uuid.toString())
@@ -141,7 +149,6 @@ void shouldSendFifoWithOptions() {
@Test
void shouldAddFifoHeadersToSend() {
String queue = "test-queue.fifo";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
@@ -151,6 +158,7 @@ void shouldAddFifoHeadersToSend() {
.sequenceNumber(sequenceNumber).build();
given(mockClient.sendMessage(any(SendMessageRequest.class)))
.willReturn(CompletableFuture.completedFuture(response));
+ mockQueueAttributes(mockClient, Map.of(QueueAttributeName.CONTENT_BASED_DEDUPLICATION, "false"));
SqsOperations template = SqsTemplate.newTemplate(mockClient);
String payload = "test-payload";
SendResult result = template.send(queue, payload);
@@ -172,14 +180,58 @@ void shouldAddFifoHeadersToSend() {
assertThat(capturedRequest.messageDeduplicationId()).isEqualTo(messageDeduplicationId);
}
+ private static void mockQueueAttributes(SqsAsyncClient mockClient, Map attributes) {
+ GetQueueAttributesResponse queueAttributesResponse = GetQueueAttributesResponse.builder().attributes(attributes)
+ .build();
+ given(mockClient.getQueueAttributes(any(Consumer.class)))
+ .willReturn(CompletableFuture.completedFuture(queueAttributesResponse));
+ }
+
+ @Test
+ void shouldAddFifoHeadersToSendWithContentBasedDeduplicationQueueConfig() {
+ String queue = "test-queue.fifo";
+
+ GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
+ given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
+ .willReturn(CompletableFuture.completedFuture(urlResponse));
+ UUID uuid = UUID.randomUUID();
+ String sequenceNumber = "1234";
+ SendMessageResponse response = SendMessageResponse.builder().messageId(uuid.toString())
+ .sequenceNumber(sequenceNumber).build();
+ given(mockClient.sendMessage(any(SendMessageRequest.class)))
+ .willReturn(CompletableFuture.completedFuture(response));
+ mockQueueAttributes(mockClient, Map.of(QueueAttributeName.CONTENT_BASED_DEDUPLICATION, "true"));
+ SqsOperations template = SqsTemplate.newTemplate(mockClient);
+ String payload = "test-payload";
+ SendResult result = template.send(queue, payload);
+ assertThat(result.endpoint()).isEqualTo(queue);
+ MessageHeaders resultHeaders = result.message().getHeaders();
+ assertThat(resultHeaders).containsKey(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER)
+ .doesNotContainKey(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER);
+ String messageDeduplicationId = resultHeaders
+ .get(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_DEDUPLICATION_ID_HEADER, String.class);
+ String messageGroupId = resultHeaders.get(SqsHeaders.MessageSystemAttributes.SQS_MESSAGE_GROUP_ID_HEADER,
+ String.class);
+ assertThat(result.message().getPayload()).isEqualTo(payload);
+ ArgumentCaptor captor = ArgumentCaptor.forClass(SendMessageRequest.class);
+ then(mockClient).should().sendMessage(captor.capture());
+ SendMessageRequest capturedRequest = captor.getValue();
+ assertThat(capturedRequest.queueUrl()).isEqualTo(queue);
+ assertThat(capturedRequest.messageBody()).isEqualTo(payload);
+ assertThat(capturedRequest.messageGroupId()).isEqualTo(messageGroupId);
+ assertThat(capturedRequest.messageDeduplicationId()).isEqualTo(messageDeduplicationId);
+ }
+
@Test
void shouldSendWithDefaultEndpoint() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
+
UUID uuid = UUID.randomUUID();
String sequenceNumber = "1234";
SendMessageResponse response = SendMessageResponse.builder().messageId(uuid.toString())
@@ -200,10 +252,11 @@ void shouldSendWithDefaultEndpoint() {
void shouldWrapSendError() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
given(mockClient.sendMessage(any(SendMessageRequest.class)))
.willReturn(CompletableFuture.failedFuture(new RuntimeException("Expected send error")));
SqsOperations template = SqsTemplate.builder().sqsAsyncClient(mockClient)
@@ -221,16 +274,17 @@ void shouldWrapSendError() {
void shouldSendWithQueueAndPayload() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
UUID uuid = UUID.randomUUID();
String sequenceNumber = "1234";
SendMessageResponse response = SendMessageResponse.builder().messageId(uuid.toString())
.sequenceNumber(sequenceNumber).build();
given(mockClient.sendMessage(any(SendMessageRequest.class)))
.willReturn(CompletableFuture.completedFuture(response));
+
SqsOperations template = SqsTemplate.newSyncTemplate(mockClient);
SendResult result = template.send(queue, payload);
assertThat(result.endpoint()).isEqualTo(queue);
@@ -248,10 +302,11 @@ void shouldSendWithQueueAndMessageAndHeaders() {
String headerName = "headerName";
String headerValue = "headerValue";
Message message = MessageBuilder.withPayload(payload).setHeader(headerName, headerValue).build();
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
UUID uuid = UUID.randomUUID();
String sequenceNumber = "1234";
SendMessageResponse response = SendMessageResponse.builder().messageId(uuid.toString())
@@ -281,10 +336,12 @@ void shouldSendBatch() {
Message message1 = MessageBuilder.withPayload(payload1).setHeader(headerName1, headerValue1).build();
Message message2 = MessageBuilder.withPayload(payload2).setHeader(headerName2, headerValue2).build();
List> messages = List.of(message1, message2);
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
+
SendMessageBatchResponse response = SendMessageBatchResponse.builder().successful(
builder -> builder.id(message1.getHeaders().getId().toString()).messageId(UUID.randomUUID().toString()),
builder -> builder.id(message2.getHeaders().getId().toString()).messageId(UUID.randomUUID().toString()))
@@ -322,10 +379,11 @@ void shouldAddFailedToTheBatchResult() {
Message message1 = MessageBuilder.withPayload(payload1).build();
Message message2 = MessageBuilder.withPayload(payload2).build();
List> messages = List.of(message1, message2);
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
String testErrorMessage = "test error message";
String code = "BC01";
boolean senderFault = true;
@@ -362,10 +420,11 @@ void shouldThrowIfHasFailedMessagesInBatchByDefault() {
Message message1 = MessageBuilder.withPayload(payload1).build();
Message message2 = MessageBuilder.withPayload(payload2).build();
List> messages = List.of(message1, message2);
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
String testErrorMessage = "test error message";
String code = "BC01";
boolean senderFault = true;
@@ -402,7 +461,8 @@ void shouldThrowIfHasFailedMessagesInBatchByDefault() {
void shouldCreateByDefaultIfQueueNotFound() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
+ mockQueueAttributes(mockClient, Map.of());
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class))).willReturn(CompletableFuture
.failedFuture(QueueDoesNotExistException.builder().message("test queue not found").build()));
given(mockClient.createQueue(any(Consumer.class)))
@@ -426,15 +486,10 @@ void shouldCreateByDefaultIfQueueNotFound() {
void shouldThrowIfQueueNotFound() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class))).willReturn(CompletableFuture
.failedFuture(QueueDoesNotExistException.builder().message("test queue not found").build()));
- UUID uuid = UUID.randomUUID();
- String sequenceNumber = "1234";
- SendMessageResponse response = SendMessageResponse.builder().messageId(uuid.toString())
- .sequenceNumber(sequenceNumber).build();
- given(mockClient.sendMessage(any(SendMessageRequest.class)))
- .willReturn(CompletableFuture.completedFuture(response));
+
SqsOperations template = SqsTemplate.builder().sqsAsyncClient(mockClient)
.configure(options -> options.queueNotFoundStrategy(QueueNotFoundStrategy.FAIL)).buildSyncTemplate();
assertThatThrownBy(() -> template.send(to -> to.queue(queue).payload(payload)))
@@ -446,10 +501,11 @@ void shouldThrowIfQueueNotFound() {
@Test
void shouldReceiveEmpty() {
String queue = "test-queue";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder().build();
given(mockClient.receiveMessage(any(ReceiveMessageRequest.class)))
.willReturn(CompletableFuture.completedFuture(receiveMessageResponse));
@@ -463,10 +519,11 @@ void shouldReceiveEmpty() {
void shouldReceiveFromDefaultEndpoint() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder().messages(builder -> builder
.messageId(UUID.randomUUID().toString()).receiptHandle("test-receipt-handle").body(payload).build())
.build();
@@ -488,10 +545,12 @@ void shouldConvertToPayloadClass() throws Exception {
String queue = "test-queue";
SampleRecord payload = new SampleRecord("first-prop", "second-prop");
String payloadString = new ObjectMapper().writeValueAsString(payload);
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
+
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder()
.messages(builder -> builder.messageId(UUID.randomUUID().toString())
.receiptHandle("test-receipt-handle").body(payloadString).build())
@@ -514,10 +573,11 @@ void shouldConvertToDefaultClass() throws Exception {
String queue = "test-queue";
SampleRecord payload = new SampleRecord("first-prop", "second-prop");
String payloadString = new ObjectMapper().writeValueAsString(payload);
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder()
.messages(builder -> builder.messageId(UUID.randomUUID().toString())
.receiptHandle("test-receipt-handle").body(payloadString).build())
@@ -547,10 +607,11 @@ record SampleRecord(String firstProperty, String secondProperty) {
void shouldUseCustomConverter() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ContextAwareMessagingMessageConverter converter = mock(
ContextAwareMessagingMessageConverter.class);
String receiptHandle = "test-receipt-handle";
@@ -578,10 +639,11 @@ void shouldReceiveAndNotAcknowledge() {
String queue = "test-queue";
String payload = "test-payload";
String receiptHandle = "test-receipt-handle";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder().messages(builder -> builder
.messageId(UUID.randomUUID().toString()).receiptHandle(receiptHandle).body(payload).build()).build();
given(mockClient.receiveMessage(any(ReceiveMessageRequest.class)))
@@ -600,10 +662,11 @@ void shouldReceiveAndNotAcknowledge() {
void shouldWrapFullAcknowledgementError() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder().messages(builder -> builder
.messageId(UUID.randomUUID().toString()).receiptHandle("test-receipt-handle").body(payload).build())
.build();
@@ -627,10 +690,11 @@ void shouldWrapPartialAcknowledgementError() {
String queue = "test-queue";
String payload1 = "test-payload-1";
String payload2 = "test-payload-2";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
String messageId1 = UUID.randomUUID().toString();
String messageId2 = UUID.randomUUID().toString();
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder().messages(
@@ -663,7 +727,7 @@ void shouldReceiveFromDefaultSettings() {
String headerValue1 = "headerValue";
String headerName2 = "headerName2";
String headerValue2 = "headerValue2";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
@@ -672,10 +736,7 @@ void shouldReceiveFromDefaultSettings() {
.build();
given(mockClient.receiveMessage(any(ReceiveMessageRequest.class)))
.willReturn(CompletableFuture.completedFuture(receiveMessageResponse));
- GetQueueAttributesResponse attributesResponse = GetQueueAttributesResponse.builder()
- .attributes(Map.of(QueueAttributeName.QUEUE_ARN, "queue-arn")).build();
- given(mockClient.getQueueAttributes(any(Consumer.class)))
- .willReturn(CompletableFuture.completedFuture(attributesResponse));
+ mockQueueAttributes(mockClient, Map.of(QueueAttributeName.QUEUE_ARN, "queue-arn"));
DeleteMessageBatchResponse deleteResponse = DeleteMessageBatchResponse.builder()
.successful(builder -> builder.id(UUID.randomUUID().toString())).build();
given(mockClient.deleteMessageBatch(any(DeleteMessageBatchRequest.class)))
@@ -727,10 +788,12 @@ void shouldReceiveFromOptions() {
String headerValue1 = "headerValue";
String headerName2 = "headerName2";
String headerValue2 = "headerValue2";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
+
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder().messages(builder -> builder
.messageId(UUID.randomUUID().toString()).receiptHandle("test-receipt-handle").body(payload).build())
.build();
@@ -762,10 +825,11 @@ void shouldReceiveFromOptions() {
void shouldReceiveFifoWithGivenAttemptId() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
String messageGroupId = UUID.randomUUID().toString();
String deduplicationId = UUID.randomUUID().toString();
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder()
@@ -802,10 +866,11 @@ void shouldReceiveFifoWithGivenAttemptId() {
void shouldReceiveFifoWithRandomAttemptId() {
String queue = "test-queue.fifo";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of(QueueAttributeName.CONTENT_BASED_DEDUPLICATION, "false"));
String messageGroupId = UUID.randomUUID().toString();
String deduplicationId = UUID.randomUUID().toString();
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder()
@@ -840,10 +905,11 @@ void shouldReceiveFifoWithRandomAttemptId() {
void shouldReceiveBatchWithDefaultValues() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder()
.messages(
builder -> builder.messageId(UUID.randomUUID().toString())
@@ -881,10 +947,12 @@ void shouldReceiveBatchWithDefaultValues() {
void shouldReceiveBatchWithQueueAndPayload() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
+
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder()
.messages(
builder -> builder.messageId(UUID.randomUUID().toString())
@@ -929,10 +997,11 @@ void shouldReceiveBatchWithQueueAndPayload() {
void shouldReceiveBatchWithOptions() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder()
.messages(
builder -> builder.messageId(UUID.randomUUID().toString()).receiptHandle("test-receipt-handle")
@@ -970,10 +1039,11 @@ void shouldReceiveBatchWithOptions() {
void shouldReceiveBatchFifo() {
String queue = "test-queue";
String payload = "test-payload";
- SqsAsyncClient mockClient = mock(SqsAsyncClient.class);
+
GetQueueUrlResponse urlResponse = GetQueueUrlResponse.builder().queueUrl(queue).build();
given(mockClient.getQueueUrl(any(GetQueueUrlRequest.class)))
.willReturn(CompletableFuture.completedFuture(urlResponse));
+ mockQueueAttributes(mockClient, Map.of());
ReceiveMessageResponse receiveMessageResponse = ReceiveMessageResponse.builder()
.messages(
builder -> builder.messageId(UUID.randomUUID().toString()).receiptHandle("test-receipt-handle")
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapperTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapperTests.java
index 11a887710..e9a770aa1 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapperTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsHeaderMapperTests.java
@@ -58,11 +58,11 @@ void shouldAddStringMessageAttributes() {
String headerName = "stringAttribute";
String headerValue = "myString";
Message message = Message.builder().body("payload")
- .messageAttributes(
- Map.of(headerName,
- MessageAttributeValue.builder().dataType(MessageAttributeDataTypes.STRING)
- .stringValue(headerValue).build()))
- .messageId(UUID.randomUUID().toString()).build();
+ .messageAttributes(
+ Map.of(headerName,
+ MessageAttributeValue.builder().dataType(MessageAttributeDataTypes.STRING)
+ .stringValue(headerValue).build()))
+ .messageId(UUID.randomUUID().toString()).build();
MessageHeaders headers = mapper.toHeaders(message);
assertThat(headers.get(headerName)).isEqualTo(headerValue);
}
@@ -73,11 +73,11 @@ void shouldAddStringCustomMessageAttributes() {
String headerName = "stringAttribute";
String headerValue = "myString";
Message message = Message.builder().body("payload")
- .messageAttributes(
- Map.of(headerName,
- MessageAttributeValue.builder().dataType(MessageAttributeDataTypes.STRING + ".Array")
- .stringValue(headerValue).build()))
- .messageId(UUID.randomUUID().toString()).build();
+ .messageAttributes(
+ Map.of(headerName,
+ MessageAttributeValue.builder().dataType(MessageAttributeDataTypes.STRING + ".Array")
+ .stringValue(headerValue).build()))
+ .messageId(UUID.randomUUID().toString()).build();
MessageHeaders headers = mapper.toHeaders(message);
assertThat(headers.get(headerName)).isEqualTo(headerValue);
}
@@ -88,10 +88,8 @@ void shouldDefaultToStringIfDataTypeUnknownMessageAttributes() {
String headerName = "stringAttribute";
String headerValue = "myString";
Message message = Message.builder().body("payload")
- .messageAttributes(
- Map.of(headerName,
- MessageAttributeValue.builder().dataType("invalid data type")
- .stringValue(headerValue).build()))
+ .messageAttributes(Map.of(headerName,
+ MessageAttributeValue.builder().dataType("invalid data type").stringValue(headerValue).build()))
.messageId(UUID.randomUUID().toString()).build();
MessageHeaders headers = mapper.toHeaders(message);
assertThat(headers.get(headerName)).isEqualTo(headerValue);
@@ -103,11 +101,11 @@ void shouldAddBinaryMessageAttributes() {
String headerName = "stringAttribute";
SdkBytes headerValue = SdkBytes.fromUtf8String("myString");
Message message = Message.builder().body("payload")
- .messageAttributes(
- Map.of(headerName,
- MessageAttributeValue.builder().dataType(MessageAttributeDataTypes.BINARY)
- .binaryValue(headerValue).build()))
- .messageId(UUID.randomUUID().toString()).build();
+ .messageAttributes(
+ Map.of(headerName,
+ MessageAttributeValue.builder().dataType(MessageAttributeDataTypes.BINARY)
+ .binaryValue(headerValue).build()))
+ .messageId(UUID.randomUUID().toString()).build();
MessageHeaders headers = mapper.toHeaders(message);
assertThat(headers.get(headerName)).isEqualTo(headerValue);
}
diff --git a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsMessagingMessageConverterTests.java b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsMessagingMessageConverterTests.java
index 3daa58433..2043a3b42 100644
--- a/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsMessagingMessageConverterTests.java
+++ b/spring-cloud-aws-sqs/src/test/java/io/awspring/cloud/sqs/support/converter/SqsMessagingMessageConverterTests.java
@@ -29,6 +29,7 @@
import org.junit.jupiter.api.Test;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.MessageBuilder;
import software.amazon.awssdk.services.sqs.model.Message;
import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
@@ -114,6 +115,26 @@ void shouldUseProvidedPayloadConverter() throws Exception {
assertThat(resultMessage.getPayload()).isEqualTo(myPojo);
}
+ @Test
+ void shouldUseHeadersFromPayloadConverter() {
+ MessageConverter payloadConverter = mock(MessageConverter.class);
+ org.springframework.messaging.Message convertedMessageWithContentType = MessageBuilder.withPayload("example")
+ .setHeader("contentType", "application/json").build();
+ when(payloadConverter.toMessage(any(MyPojo.class), any())).thenReturn(convertedMessageWithContentType);
+
+ SqsMessagingMessageConverter converter = new SqsMessagingMessageConverter();
+ converter.setPayloadMessageConverter(payloadConverter);
+ converter.setPayloadTypeMapper(msg -> MyPojo.class);
+
+ org.springframework.messaging.Message message = MessageBuilder.createMessage(new MyPojo(),
+ new MessageHeaders(null));
+ Message resultMessage = converter.fromMessagingMessage(message);
+
+ assertThat(resultMessage.messageId()).isEqualTo(message.getHeaders().getId().toString());
+ assertThat(resultMessage.messageAttributes()).containsEntry("contentType",
+ MessageAttributeValue.builder().stringValue("application/json").dataType("String").build());
+ }
+
static class MyPojo {
private String myProperty = "myValue";
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter-dynamodb/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter-dynamodb/pom.xml
index a0fc97864..cf3f1ad18 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter-dynamodb/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter-dynamodb/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
4.0.0
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter-metrics/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter-metrics/pom.xml
index 755d46446..ea5623d10 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter-metrics/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter-metrics/pom.xml
@@ -22,7 +22,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
spring-cloud-aws-starter-metrics
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter-parameter-store/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter-parameter-store/pom.xml
index b6ecc90ab..26144771e 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter-parameter-store/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter-parameter-store/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
4.0.0
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter-s3/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter-s3/pom.xml
index 29084c9ec..af431ecdd 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter-s3/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter-s3/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
4.0.0
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter-secrets-manager/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter-secrets-manager/pom.xml
index 2640b9068..e27b5f949 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter-secrets-manager/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter-secrets-manager/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
4.0.0
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter-ses/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter-ses/pom.xml
index 8b4ab78fe..d1e77733f 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter-ses/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter-ses/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
4.0.0
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter-sns/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter-sns/pom.xml
index 4690fdaee..0b3474e29 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter-sns/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter-sns/pom.xml
@@ -6,7 +6,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter-sqs/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter-sqs/pom.xml
index a8b5a6cc7..af296d78b 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter-sqs/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter-sqs/pom.xml
@@ -6,7 +6,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
diff --git a/spring-cloud-aws-starters/spring-cloud-aws-starter/pom.xml b/spring-cloud-aws-starters/spring-cloud-aws-starter/pom.xml
index af0c7b755..421b47f56 100644
--- a/spring-cloud-aws-starters/spring-cloud-aws-starter/pom.xml
+++ b/spring-cloud-aws-starters/spring-cloud-aws-starter/pom.xml
@@ -5,7 +5,7 @@
spring-cloud-aws
io.awspring.cloud
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
../../pom.xml
4.0.0
diff --git a/spring-cloud-aws-test/pom.xml b/spring-cloud-aws-test/pom.xml
index 9b2fa4683..aab3522fa 100644
--- a/spring-cloud-aws-test/pom.xml
+++ b/spring-cloud-aws-test/pom.xml
@@ -22,7 +22,7 @@
io.awspring.cloud
spring-cloud-aws
- 3.0.0-SNAPSHOT
+ 3.0.2-SNAPSHOT
spring-cloud-aws-test
Spring Cloud AWS Test
diff --git a/spring-cloud-aws-test/src/test/java/io/awspring/cloud/test/sqs/BaseSqsIntegrationTest.java b/spring-cloud-aws-test/src/test/java/io/awspring/cloud/test/sqs/BaseSqsIntegrationTest.java
index 71e57268e..53603cf3d 100644
--- a/spring-cloud-aws-test/src/test/java/io/awspring/cloud/test/sqs/BaseSqsIntegrationTest.java
+++ b/spring-cloud-aws-test/src/test/java/io/awspring/cloud/test/sqs/BaseSqsIntegrationTest.java
@@ -32,7 +32,7 @@ abstract class BaseSqsIntegrationTest {
@Container
static LocalStackContainer localstack = new LocalStackContainer(
- DockerImageName.parse("localstack/localstack:2.0.0")).withReuse(true);
+ DockerImageName.parse("localstack/localstack:1.4.0")).withReuse(true);
@BeforeAll
static void beforeAll() throws IOException, InterruptedException {