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

Localstack for SqsMessageDrivenChannelAdapterTest

Browse Source 
* Remove `AwsHeaders` which are now covered by the `SqsHeaders`
This commit is contained in:
abilan
2023-03-15 13:41:05 -04:00
parent d003e1b850
commit aee5bdc1af
3 changed files with 47 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java
View File

@@ -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

15
src/main/java/org/springframework/integration/aws/support/AwsHeaders.java
View File

@@ -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.
*/

150
src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java
View File

@@ -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<String, MessageAttributeValue> 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();
}
}
}