diff --git a/build.gradle b/build.gradle index e9dd11f..6604ce8 100644 --- a/build.gradle +++ b/build.gradle @@ -4,6 +4,9 @@ buildscript { repositories { maven { url 'http://repo.spring.io/plugins-release' } } + dependencies { + classpath 'io.spring.gradle:spring-io-plugin:0.0.4.RELEASE' + } } apply plugin: 'java' @@ -15,24 +18,32 @@ apply plugin: 'jacoco' group = 'org.springframework.integration' repositories { + if (version.endsWith('BUILD-SNAPSHOT') || project.hasProperty('platformVersion')) { + maven { url 'https://repo.spring.io/libs-snapshot' } + } maven { url 'http://repo.spring.io/libs-milestone' } - maven { url 'http://repo.spring.io/libs-snapshot' } } sourceCompatibility = targetCompatibility = 1.7 -// See http://www.gradle.org/docs/current/userguide/dependency_management.html#sub:configurations -// and http://www.gradle.org/docs/current/dsl/org.gradle.api.artifacts.ConfigurationContainer.html -configurations { - jacoco //Configuration Group used by Sonar to provide Code Coverage using JaCoCo +if (project.hasProperty('platformVersion')) { + apply plugin: 'spring-io' + + dependencyManagement { + springIoTestRuntime { + imports { + mavenBom "io.spring.platform:platform-bom:${platformVersion}" + } + } + } } ext { commonsIoVersion='2.4' servletApiVersion = '3.1.0' slf4jVersion = '1.7.12' - springCloudAwsVersion = '1.0.3.RELEASE' - springIntegrationVersion = '4.2.4.RELEASE' + springCloudAwsVersion = '1.1.0.M2' + springIntegrationVersion = '4.2.5.RELEASE' idPrefix = 'aws' @@ -103,8 +114,6 @@ ext.xLintArg = '-Xlint:all,-options' [compileJava, compileTestJava]*.options*.compilerArgs = [xLintArg] test { - // suppress all console output during testing unless running `gradle -i` - logging.captureStandardOutput(LogLevel.INFO) maxHeapSize = "1024m" jacoco { append = false @@ -112,6 +121,11 @@ test { } } +tasks.withType(Test).all { + // suppress all console output during testing unless running `gradle -i` + logging.captureStandardOutput(LogLevel.INFO) +} + jacocoTestReport { reports { xml.enabled false @@ -121,6 +135,7 @@ jacocoTestReport { } build.dependsOn jacocoTestReport + task sourcesJar(type: Jar) { classifier = 'sources' from sourceSets.main.allJava diff --git a/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java b/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java index 12ae20a..44c294f 100644 --- a/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java +++ b/src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -77,7 +77,7 @@ public class SqsMessageDrivenChannelAdapterParser extends AbstractSingleBeanDefi IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "payload-type"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delete-message-on-exception"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "message-deletion-policy"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-number-of-messages"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "visibility-timeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "wait-time-out"); diff --git a/src/main/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapter.java index 76b28ea..990585f 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -41,6 +41,7 @@ import org.springframework.util.Assert; import org.springframework.web.multipart.MultipartResolver; import com.amazonaws.services.sns.AmazonSNS; +import com.fasterxml.jackson.databind.JsonNode; /** * The {@link HttpRequestHandlingMessagingGateway} extension for the Amazon WS SNS HTTP(S) endpoints. @@ -63,13 +64,15 @@ import com.amazonaws.services.sns.AmazonSNS; *

* For the convenience on the underlying message flow routing a {@link AwsHeaders#SNS_MESSAGE_TYPE} * header is present. - * * @author Artem Bilan */ public class SnsInboundChannelAdapter extends HttpRequestHandlingMessagingGateway { private final NotificationStatusResolver notificationStatusResolver; + private final MappingJackson2HttpMessageConverter jackson2HttpMessageConverter = + new MappingJackson2HttpMessageConverter(); + private volatile boolean handleNotificationStatus; private volatile Expression payloadExpression; @@ -89,7 +92,7 @@ public class SnsInboundChannelAdapter extends HttpRequestHandlingMessagingGatewa super.setRequestMapping(requestMapping); super.setStatusCodeExpression(new ValueExpression<>(HttpStatus.NO_CONTENT)); super.setMessageConverters( - Collections.>singletonList(new MappingJackson2HttpMessageConverter())); + Collections.>singletonList(this.jackson2HttpMessageConverter)); super.setRequestPayloadType(HashMap.class); } @@ -109,7 +112,7 @@ public class SnsInboundChannelAdapter extends HttpRequestHandlingMessagingGatewa @SuppressWarnings("unchecked") protected void send(Object object) { Message message = (Message) object; - HashMap payload = (HashMap) message.getPayload(); + Map payload = (HashMap) message.getPayload(); AbstractIntegrationMessageBuilder messageToSendBuilder; if (this.payloadExpression != null) { messageToSendBuilder = getMessageBuilderFactory() @@ -122,7 +125,8 @@ public class SnsInboundChannelAdapter extends HttpRequestHandlingMessagingGatewa String type = payload.get("Type"); if ("SubscriptionConfirmation".equals(type) || "UnsubscribeConfirmation".equals(type)) { - NotificationStatus notificationStatus = this.notificationStatusResolver.resolveNotificationStatus(payload); + JsonNode content = this.jackson2HttpMessageConverter.getObjectMapper().valueToTree(payload); + NotificationStatus notificationStatus = this.notificationStatusResolver.resolveNotificationStatus(content); if (this.handleNotificationStatus) { messageToSendBuilder.setHeader(AwsHeaders.NOTIFICATION_STATUS, notificationStatus); } @@ -194,8 +198,8 @@ public class SnsInboundChannelAdapter extends HttpRequestHandlingMessagingGatewa super(amazonSns); } - protected NotificationStatus resolveNotificationStatus(HashMap content) { - return (NotificationStatus) super.doResolverArgumentFromNotificationMessage(content); + protected NotificationStatus resolveNotificationStatus(JsonNode content) { + return (NotificationStatus) doResolveArgumentFromNotificationMessage(content, null, null); } } 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 6859990..b55bbba 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/SqsMessageDrivenChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -28,7 +28,8 @@ import org.springframework.cloud.aws.core.env.ResourceIdResolver; import org.springframework.cloud.aws.messaging.config.SimpleMessageListenerContainerFactory; import org.springframework.cloud.aws.messaging.listener.QueueMessageHandler; import org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer; -import org.springframework.core.task.TaskExecutor; +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.messaging.Message; @@ -59,13 +60,15 @@ public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport private SimpleMessageListenerContainer listenerContainer; + private SqsMessageDeletionPolicy messageDeletionPolicy = SqsMessageDeletionPolicy.NO_REDRIVE; + public SqsMessageDrivenChannelAdapter(AmazonSQSAsync amazonSqs, String... queues) { Assert.noNullElements(queues, "'queues' must not be empty"); this.simpleMessageListenerContainerFactory.setAmazonSqs(amazonSqs); this.queues = Arrays.copyOf(queues, queues.length); } - public void setTaskExecutor(TaskExecutor taskExecutor) { + public void setTaskExecutor(AsyncTaskExecutor taskExecutor) { this.simpleMessageListenerContainerFactory.setTaskExecutor(taskExecutor); } @@ -89,8 +92,9 @@ public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport this.simpleMessageListenerContainerFactory.setDestinationResolver(destinationResolver); } - public void setDeleteMessageOnException(Boolean deleteMessageOnException) { - this.simpleMessageListenerContainerFactory.setDeleteMessageOnException(deleteMessageOnException); + public void setMessageDeletionPolicy(SqsMessageDeletionPolicy messageDeletionPolicy) { + Assert.notNull(messageDeletionPolicy, "'messageDeletionPolicy' must not be null."); + this.messageDeletionPolicy = messageDeletionPolicy; } @Override @@ -126,7 +130,7 @@ public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport @Override public Map getHandlerMethods() { Set queues = new HashSet<>(Arrays.asList(SqsMessageDrivenChannelAdapter.this.queues)); - return Collections.singletonMap(new MappingInformation(queues), null); + return Collections.singletonMap(new MappingInformation(queues, messageDeletionPolicy), null); } @Override @@ -134,13 +138,13 @@ public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport MessageHeaders headers = message.getHeaders(); Message messageToSend = getMessageBuilderFactory() .fromMessage(message) - .removeHeaders(QueueMessageHandler.Headers.LOGICAL_RESOURCE_ID_MESSAGE_HEADER_KEY, + .removeHeaders("LogicalResourceId", "MessageId", "ReceiptHandle") .setHeader(AwsHeaders.MESSAGE_ID, headers.get("MessageId")) .setHeader(AwsHeaders.RECEIPT_HANDLE, headers.get("ReceiptHandle")) .setHeader(AwsHeaders.QUEUE, - headers.get(QueueMessageHandler.Headers.LOGICAL_RESOURCE_ID_MESSAGE_HEADER_KEY)) + headers.get("LogicalResourceId")) .build(); sendMessage(messageToSend); } diff --git a/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd index 8d31d12..5086734 100644 --- a/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd +++ b/src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd @@ -443,13 +443,13 @@ - The 'org.springframework.core.task.TaskExecutor' to run the underlying listener task + The 'org.springframework.core.task.AsyncTaskExecutor' to run the underlying listener task from the 'org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer'. - + @@ -500,23 +500,34 @@ - + - Defines if a message must be deleted or not if the message handling throws an exception - By default this value is set to 'true' which means that the message is deleted to - avoid poison messages. If the 'error-channel' is specified and the direct error flow - doesn't rethrow a 'MessagingException' this attribute doesn't have an effect. If this is - to 'false' it is a responsibility of the error flow to remove the message from - AmazonSQS. + Defines the policy that must be used for the deletion of SQS messages once + they were processed. The default policy is NO_REDRIVE because it is the safest + way to avoid poison messages and have + a safe way to avoid the loss of messages (i.e. using a dead letter queue). + + + + + + + + + + + + + diff --git a/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml index 31a839a..d11dfa8 100644 --- a/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml +++ b/src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml @@ -14,15 +14,10 @@ - + - - - - - + destinationResolver() { + DestinationResolver destinationResolver = Mockito.mock(DestinationResolver.class); + doThrow(DestinationResolutionException.class) + .when(destinationResolver) + .resolveDestination(anyString()); + return destinationResolver; + } @Test public void testSqsMessageDrivenChannelAdapterParser() { @@ -82,12 +97,11 @@ public class SqsMessageDrivenChannelAdapterParserTests { assertEquals(5, TestUtils.getPropertyValue(listenerContainer, "maxNumberOfMessages")); assertEquals(200, TestUtils.getPropertyValue(listenerContainer, "visibilityTimeout")); assertEquals(40, TestUtils.getPropertyValue(listenerContainer, "waitTimeOut")); - assertFalse(TestUtils.getPropertyValue(listenerContainer, "deleteMessageOnException", Boolean.class)); @SuppressWarnings("rawtypes") - Set queues = TestUtils.getPropertyValue(listenerContainer, "queues", Set.class); - assertTrue(queues.contains("foo")); - assertTrue(queues.contains("bar")); + Map queues = TestUtils.getPropertyValue(listenerContainer, "registeredQueues", Map.class); + assertTrue(queues.keySet().contains("foo")); + assertTrue(queues.keySet().contains("bar")); assertEquals(100, this.sqsMessageDrivenChannelAdapter.getPhase()); assertFalse(this.sqsMessageDrivenChannelAdapter.isAutoStartup()); @@ -96,6 +110,9 @@ public class SqsMessageDrivenChannelAdapterParserTests { assertSame(this.nullChannel, TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, "errorChannel")); assertEquals(2000L, TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, "messagingTemplate.sendTimeout")); + assertEquals(SqsMessageDeletionPolicy.NEVER, + TestUtils.getPropertyValue(this.sqsMessageDrivenChannelAdapter, "messageDeletionPolicy", + SqsMessageDeletionPolicy.class)); } }