diff --git a/build.gradle b/build.gradle index 0275fb2..938f49d 100644 --- a/build.gradle +++ b/build.gradle @@ -3,7 +3,7 @@ buildscript { maven { url 'http://repo.spring.io/plugins-release' } } dependencies { - classpath 'io.spring.gradle:spring-io-plugin:0.0.4.RELEASE' + classpath 'io.spring.gradle:spring-io-plugin:0.0.5.RELEASE' } } @@ -56,7 +56,7 @@ ext { servletApiVersion = '3.1.0' slf4jVersion = '1.7.21' springCloudAwsVersion = '1.1.0.RELEASE' - springIntegrationVersion = '4.2.8.RELEASE' + springIntegrationVersion = '4.2.9.RELEASE' idPrefix = 'aws' 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 c5ecc6a..4ed5dac 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java @@ -32,6 +32,9 @@ import org.springframework.cloud.aws.messaging.listener.SqsMessageDeletionPolicy import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.integration.aws.support.AwsHeaders; import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.core.DestinationResolver; @@ -51,6 +54,8 @@ import com.amazonaws.services.sqs.AmazonSQSAsync; * @see SimpleMessageListenerContainer * @see QueueMessageHandler */ +@ManagedResource +@IntegrationManagedResource public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport implements DisposableBean { private final SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory = @@ -140,6 +145,21 @@ public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport imple this.listenerContainer.stop(); } + @ManagedOperation + public void stop(String logicalQueueName) { + this.listenerContainer.stop(logicalQueueName); + } + + @ManagedOperation + public void start(String logicalQueueName) { + this.listenerContainer.start(logicalQueueName); + } + + @ManagedOperation + public boolean isRunning(String logicalQueueName) { + return this.listenerContainer.isRunning(logicalQueueName); + } + @Override public void destroy() throws Exception { this.listenerContainer.destroy(); 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 6432fc1..4081529 100644 --- a/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapterTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.aws.inbound; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.mock; import static org.mockito.Matchers.any; @@ -27,12 +28,16 @@ import org.junit.runner.RunWith; 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.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.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -58,6 +63,12 @@ public class SqsMessageDrivenChannelAdapterTests { @Autowired private SqsMessageDrivenChannelAdapter sqsMessageDrivenChannelAdapter; + @Autowired + private MessageChannel controlBusInput; + + @Autowired + private PollableChannel controlBusOutput; + @Test public void testSqsMessageDrivenChannelAdapter() { assertThat(TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, @@ -71,6 +82,25 @@ public class SqsMessageDrivenChannelAdapterTests { assertThat(receive).isNotNull(); assertThat((String) receive.getPayload()).isIn("messageContent", "messageContent2"); assertThat(receive.getHeaders().get(AwsHeaders.QUEUE)).isEqualTo("testQueue"); + + this.controlBusInput.send(new GenericMessage<>("@sqsMessageDrivenChannelAdapter.stop('testQueue')")); + 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) + .hasMessageContaining("Queue with name 'foo' does not exist"); } @Configuration @@ -111,6 +141,19 @@ public class SqsMessageDrivenChannelAdapterTests { 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(); + } + } } diff --git a/src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java index b210100..43014c2 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java +++ b/src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java @@ -34,6 +34,7 @@ import java.io.IOException; import java.io.InputStream; import java.util.Arrays; import java.util.Collections; +import java.util.Date; import java.util.HashMap; import java.util.LinkedList; import java.util.List; @@ -317,7 +318,9 @@ public class S3MessageHandlerTests { AmazonS3 amazonS3 = mock(AmazonS3.class); given(amazonS3.putObject(any(PutObjectRequest.class))).willReturn(new PutObjectResult()); - given(amazonS3.getObjectMetadata(any(GetObjectMetadataRequest.class))).willReturn(new ObjectMetadata()); + ObjectMetadata objectMetadata = new ObjectMetadata(); + objectMetadata.setLastModified(new Date()); + given(amazonS3.getObjectMetadata(any(GetObjectMetadataRequest.class))).willReturn(objectMetadata); given(amazonS3.copyObject(any(CopyObjectRequest.class))).willReturn(new CopyObjectResult()); ObjectListing objectListing = spy(new ObjectListing());