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

Upgrade to Spring Cloud AWS 1.1.0.M2

Browse Source
This commit is contained in:
Artem Bilan
2016-02-23 13:59:26 -05:00
parent 7e3271dfad
commit b6c57b29be
7 changed files with 94 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
build.gradle
View File

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

4
src/main/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParser.java
View File

@@ -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");

18
src/main/java/org/springframework/integration/aws/inbound/SnsInboundChannelAdapter.java
View File

@@ -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;
* <p>
* 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.<HttpMessageConverter<?>>singletonList(new MappingJackson2HttpMessageConverter()));
Collections.<HttpMessageConverter<?>>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<String, String> payload = (HashMap<String, String>) message.getPayload();
Map<String, String> payload = (HashMap<String, String>) 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<String, String> content) {
return (NotificationStatus) super.doResolverArgumentFromNotificationMessage(content);
protected NotificationStatus resolveNotificationStatus(JsonNode content) {
return (NotificationStatus) doResolveArgumentFromNotificationMessage(content, null, null);
}
}

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

@@ -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<MappingInformation, HandlerMethod> getHandlerMethods() {
Set<String> 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);
}

29
src/main/resources/org/springframework/integration/aws/config/spring-integration-aws-1.0.xsd
View File

@@ -443,13 +443,13 @@
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.core.task.TaskExecutor"/>
<tool:expected-type type="org.springframework.core.task.AsyncTaskExecutor"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -500,23 +500,34 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="delete-message-on-exception">
<xsd:attribute name="message-deletion-policy" default="NO_REDRIVE">
<xsd:annotation>
<xsd:documentation>
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).
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="messageDeletionPolicy xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:simpleType name="messageDeletionPolicy">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="NO_REDRIVE" />
<xsd:enumeration value="ALWAYS" />
<xsd:enumeration value="NEVER" />
<xsd:enumeration value="ON_SUCCESS" />
</xsd:restriction>
</xsd:simpleType>
<xsd:element name="sns-inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>

11
src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests-context.xml
View File

@@ -14,15 +14,10 @@
</bean>
<bean id="taskExecutor" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.core.task.TaskExecutor"/>
<constructor-arg value="org.springframework.core.task.AsyncTaskExecutor"/>
</bean>
<bean id="destinationResolver"
class="org.springframework.cloud.aws.messaging.support.destination.DynamicQueueUrlDestinationResolver">
<constructor-arg ref="sqs"/>
<constructor-arg ref="resourceIdResolver"/>
</bean>
<bean class="org.springframework.integration.aws.config.xml.SqsMessageDrivenChannelAdapterParserTests"/>
<int-aws:sqs-message-driven-channel-adapter sqs="sqs"
auto-startup="false"
@@ -32,7 +27,7 @@
phase="100"
id="sqsMessageDrivenChannelAdapter"
queues="foo, bar"
delete-message-on-exception="false"
message-deletion-policy="NEVER"
max-number-of-messages="5"
visibility-timeout="200"
wait-time-out="40"

27
src/test/java/org/springframework/integration/aws/config/xml/SqsMessageDrivenChannelAdapterParserTests.java
View File

@@ -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.
@@ -20,20 +20,27 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doThrow;
import java.util.Map;
import java.util.Set;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.aws.core.env.ResourceIdResolver;
import org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer;
import org.springframework.cloud.aws.messaging.listener.SqsMessageDeletionPolicy;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.aws.inbound.SqsMessageDrivenChannelAdapter;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -68,6 +75,14 @@ public class SqsMessageDrivenChannelAdapterParserTests {
@Autowired
private SqsMessageDrivenChannelAdapter sqsMessageDrivenChannelAdapter;
@Bean
public DestinationResolver<?> 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));
}
}