From aee5bdc1af8743afb53feaa482d95f79839b3669 Mon Sep 17 00:00:00 2001 From: abilan Date: Wed, 15 Mar 2023 13:41:05 -0400 Subject: [PATCH] Localstack for SqsMessageDrivenChannelAdapterTest * Remove `AwsHeaders` which are now covered by the `SqsHeaders` --- .../SqsMessageDrivenChannelAdapter.java | 5 +- .../integration/aws/support/AwsHeaders.java | 15 -- .../SqsMessageDrivenChannelAdapterTests.java | 150 +++++------------- 3 files changed, 47 insertions(+), 123 deletions(-) diff --git a/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java index fceec2f..a37e5bf 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java @@ -22,6 +22,7 @@ import java.util.Collection; import io.awspring.cloud.sqs.config.SqsMessageListenerContainerFactory; import io.awspring.cloud.sqs.listener.MessageListener; import io.awspring.cloud.sqs.listener.SqsContainerOptions; +import io.awspring.cloud.sqs.listener.SqsHeaders; import io.awspring.cloud.sqs.listener.SqsMessageListenerContainer; import software.amazon.awssdk.services.sqs.SqsAsyncClient; @@ -44,6 +45,7 @@ import org.springframework.util.Assert; * @see SqsMessageListenerContainerFactory * @see SqsMessageListenerContainerFactory * @see MessageListener + * @see SqsHeaders */ @ManagedResource @IntegrationManagedResource @@ -76,8 +78,7 @@ public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport { sqsContainerOptionsBuilder.fromBuilder(this.sqsContainerOptions.toBuilder())); } this.sqsMessageListenerContainerFactory.messageListener(new IntegrationMessageListener()); - SqsMessageListenerContainerFactory containerFactory = this.sqsMessageListenerContainerFactory.build(); - this.listenerContainer = containerFactory.createContainer(this.queues); + this.listenerContainer = this.sqsMessageListenerContainerFactory.build().createContainer(this.queues); } @Override diff --git a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java index c4b6c5f..97480aa 100644 --- a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java +++ b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java @@ -30,11 +30,6 @@ public abstract class AwsHeaders { */ public static final String QUEUE = PREFIX + "queue"; - /** - * The {@value RECEIVED_QUEUE} header for receiving data from SQS. - */ - public static final String RECEIVED_QUEUE = PREFIX + "receivedQueue"; - /** * The {@value TOPIC} header for sending/receiving data over SNS. */ @@ -45,16 +40,6 @@ public abstract class AwsHeaders { */ public static final String MESSAGE_ID = PREFIX + "messageId"; - /** - * The {@value RECEIPT_HANDLE} header for received SQS message. - */ - public static final String RECEIPT_HANDLE = PREFIX + "receiptHandle"; - - /** - * The {@value ACKNOWLEDGMENT} header for received SQS message. - */ - public static final String ACKNOWLEDGMENT = PREFIX + "acknowledgment"; - /** * The {@value NOTIFICATION_STATUS} header for SNS notification status. */ diff --git a/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java index 4d3a07a..cf1bfeb 100644 --- a/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java @@ -16,136 +16,87 @@ package org.springframework.integration.aws.inbound; -import java.util.concurrent.CompletableFuture; +import java.util.Map; -import org.junit.jupiter.api.Disabled; +import io.awspring.cloud.sqs.listener.SqsHeaders; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import software.amazon.awssdk.services.sqs.SqsAsyncClient; -import software.amazon.awssdk.services.sqs.model.GetQueueAttributesRequest; -import software.amazon.awssdk.services.sqs.model.GetQueueAttributesResponse; -import software.amazon.awssdk.services.sqs.model.GetQueueUrlRequest; -import software.amazon.awssdk.services.sqs.model.GetQueueUrlResponse; -import software.amazon.awssdk.services.sqs.model.Message; -import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest; -import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse; +import software.amazon.awssdk.services.sqs.model.MessageAttributeValue; +import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.aws.LocalstackContainerTest; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; -import org.springframework.integration.config.ExpressionControlBusFactoryBean; import org.springframework.integration.core.MessageProducer; -import org.springframework.integration.test.util.TestUtils; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; -import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.BDDMockito.given; -import static org.mockito.BDDMockito.mock; /** * @author Artem Bilan */ -@Disabled("Revise in favor of Local Stack") @SpringJUnitConfig @DirtiesContext -public class SqsMessageDrivenChannelAdapterTests { +public class SqsMessageDrivenChannelAdapterTests implements LocalstackContainerTest { + + private static SqsAsyncClient AMAZON_SQS; + + private static String testQueueUrl; @Autowired private PollableChannel inputChannel; - @Autowired - private SqsMessageDrivenChannelAdapter sqsMessageDrivenChannelAdapter; - - @Autowired - private MessageChannel controlBusInput; - - @Autowired - private PollableChannel controlBusOutput; + @BeforeAll + static void setup() { + AMAZON_SQS = LocalstackContainerTest.sqsClient(); + testQueueUrl = AMAZON_SQS.createQueue(request -> request.queueName("testQueue")).join().queueUrl(); + } @Test void testSqsMessageDrivenChannelAdapter() { - assertThat( - TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, "listenerContainer.queueStopTimeout")) - .isEqualTo(20000L); - org.springframework.messaging.Message receive = this.inputChannel.receive(1000); + Map attributes = + Map.of("someAttribute", + MessageAttributeValue.builder() + .stringValue("someValue") + .dataType("String") + .build()); + + AMAZON_SQS.sendMessageBatch(request -> + request.queueUrl(testQueueUrl) + .entries(SendMessageBatchRequestEntry.builder() + .messageBody("messageContent") + .id("messageContent_id") + .messageAttributes(attributes) + .build(), + SendMessageBatchRequestEntry.builder() + .messageBody("messageContent2") + .id("messageContent2_id") + .messageAttributes(attributes) + .build())); + + org.springframework.messaging.Message receive = this.inputChannel.receive(10000); assertThat(receive).isNotNull(); assertThat((String) receive.getPayload()).isIn("messageContent", "messageContent2"); - assertThat(receive.getHeaders().get(AwsHeaders.RECEIVED_QUEUE)).isEqualTo("testQueue"); - receive = this.inputChannel.receive(1000); + assertThat(receive.getHeaders().get(SqsHeaders.SQS_QUEUE_NAME_HEADER)).isEqualTo("testQueue"); + assertThat(receive.getHeaders().get("someAttribute")).isEqualTo("someValue"); + + receive = this.inputChannel.receive(10000); assertThat(receive).isNotNull(); assertThat((String) receive.getPayload()).isIn("messageContent", "messageContent2"); - assertThat(receive.getHeaders().get(AwsHeaders.RECEIVED_QUEUE)).isEqualTo("testQueue"); - - try { - this.controlBusInput.send(new GenericMessage<>("@sqsMessageDrivenChannelAdapter.stop('testQueue')")); - } - catch (Exception e) { - // May fail with NPE. See - // https://github.com/spring-cloud/spring-cloud-aws/issues/232 - } - this.controlBusInput.send(new GenericMessage<>("@sqsMessageDrivenChannelAdapter.isRunning('testQueue')")); - - receive = this.controlBusOutput.receive(1000); - assertThat(receive).isNotNull(); - assertThat((Boolean) receive.getPayload()).isFalse(); - - this.controlBusInput.send(new GenericMessage<>("@sqsMessageDrivenChannelAdapter.start('testQueue')")); - this.controlBusInput.send(new GenericMessage<>("@sqsMessageDrivenChannelAdapter.isRunning('testQueue')")); - - receive = this.controlBusOutput.receive(1000); - assertThat(receive).isNotNull(); - assertThat((Boolean) receive.getPayload()).isTrue(); - - assertThatThrownBy( - () -> this.controlBusInput.send(new GenericMessage<>("@sqsMessageDrivenChannelAdapter.start('foo')"))) - .hasCauseExactlyInstanceOf(IllegalArgumentException.class) - .hasStackTraceContaining("Queue with name 'foo' does not exist"); - - assertThat(this.sqsMessageDrivenChannelAdapter.getQueues()).isEqualTo(new String[] {"testQueue"}); + assertThat(receive.getHeaders().get(SqsHeaders.SQS_QUEUE_NAME_HEADER)).isEqualTo("testQueue"); + assertThat(receive.getHeaders().get("someAttribute")).isEqualTo("someValue"); } @Configuration @EnableIntegration public static class ContextConfiguration { - @Bean - public SqsAsyncClient amazonSqs() { - SqsAsyncClient sqs = mock(SqsAsyncClient.class); - given(sqs.getQueueUrl(GetQueueUrlRequest.builder().queueName("testQueue").build())) - .willReturn(CompletableFuture.completedFuture( - GetQueueUrlResponse.builder().queueUrl("http://testQueue.amazonaws.com").build())); - - given(sqs.receiveMessage( - ReceiveMessageRequest.builder() - .queueUrl("http://testQueue.amazonaws.com") - .maxNumberOfMessages(10) - .attributeNamesWithStrings("All") - .messageAttributeNames("All") - .waitTimeSeconds(20) - .build())) - .willReturn( - CompletableFuture.completedFuture( - ReceiveMessageResponse.builder() - .messages(Message.builder().body("messageContent").build(), - Message.builder().body("messageContent2").build()) - .build())) - .willReturn(CompletableFuture.completedFuture(ReceiveMessageResponse.builder().build())); - - given(sqs.getQueueAttributes(any(GetQueueAttributesRequest.class))) - .willReturn(CompletableFuture.completedFuture(GetQueueAttributesResponse.builder().build())); - - return sqs; - } - @Bean public PollableChannel inputChannel() { return new QueueChannel(); @@ -153,24 +104,11 @@ public class SqsMessageDrivenChannelAdapterTests { @Bean public MessageProducer sqsMessageDrivenChannelAdapter() { - SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(amazonSqs(), "testQueue"); + SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(AMAZON_SQS, "testQueue"); adapter.setOutputChannel(inputChannel()); return adapter; } - @Bean - @ServiceActivator(inputChannel = "controlBusInput") - public ExpressionControlBusFactoryBean controlBus() { - ExpressionControlBusFactoryBean controlBusFactoryBean = new ExpressionControlBusFactoryBean(); - controlBusFactoryBean.setOutputChannel(controlBusOutput()); - return controlBusFactoryBean; - } - - @Bean - public PollableChannel controlBusOutput() { - return new QueueChannel(); - } - } }