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

Add S3MessageHandler Namespace Support

Browse Source 
* Add `<int-aws:s3-outbound-channel-adapter>` and `<int-aws:s3-outbound-gateway>`
components. Their parsers and tests for them.
* Remove the old S3 Outbound Channel Adapter implementation and its tests

Rename `AwsNamespaceHandler` properly

`git mv -f File file` does the trick

Add `.travis.yml` to enable Travis CI on PRs
This commit is contained in:
Artem Bilan
2016-03-29 17:03:13 -04:00
committed by Gary Russell
parent a0893cb88e
commit 5167b36b4d
24 changed files with 662 additions and 1133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -9,3 +9,4 @@ target
/*.iml
/*.ipr
/*.iws
bin/

2
.travis.yml Normal file
View File

@@ -0,0 +1,2 @@
language: java
jdk: oraclejdk8

66
README.md
View File

@@ -62,7 +62,51 @@ Get more information about AWS free tier at [http://aws.amazon.com/free/][]**
###Introduction
The S3 Channel Adapters are based on the `AmazonS3` template and `TransferManager`.
See their specification and JavaDocs for more information.
###Outbound Channel Adapter
The S3 Outbound Channel Adapter is represented by the `S3MessageHandler` (`<int-aws:s3-outbound-channel-adapter>`
and `<int-aws:s3-outbound-gateway>`) and allows to perform `upload`, `download` and `copy`
(see `S3MessageHandler.Command` enum) operations in the provided S3 bucket.
The Java Configuration is:
````java
@SpringBootApplication
public static class MyConfiguration {
@Autowired
private AmazonS3 amazonS3;
@Bean
@ServiceActivator(inputChannel = "s3UploadChannel")
public MessageHandler s3MessageHandler() {
return new S3MessageHandler(amazonS3(), "myBuck");
}
}
````
With this config you can send message with the `java.io.File` as `payload` and the `transferManager.upload()`
operation will be performed, where the file name is used as a S3 Object key.
An XML variant may look like:
````xml
<bean id="transferManager" class="com.amazonaws.services.s3.transfer.TransferManager"/>
<int-aws:s3-outbound-channel-adapter transfer-manager="transferManager"
channel="s3SendChannel"
bucket="foo"
command="DOWNLOAD"
key="myDirectory"/>
````
See more information in the `S3MessageHandler` JavaDocs and `<int-aws:s3-outbound-channel-adapter>` &
`<int-aws:s3-outbound-gateway>` descriptions.
###Inbound Channel Adapter
##Simple Email Service (SES)
@@ -91,19 +135,19 @@ The Java Configuration is pretty simple:
@SpringBootApplication
public static class MyConfiguration {
@Autowired
private AmazonSQS amazonSqs;
@Autowired
private AmazonSQS amazonSqs;
@Bean
public QueueMessagingTemplate queueMessagingTemplate() {
return new QueueMessagingTemplate(this.amazonSqs);
}
@Bean
public QueueMessagingTemplate queueMessagingTemplate() {
return new QueueMessagingTemplate(this.amazonSqs);
}
@Bean
@ServiceActivator(inputChannel = "sqsSendChannel")
public MessageHandler sqsMessageHandler() {
return new SqsMessageHandler(queueMessagingTemplate());
}
@Bean
@ServiceActivator(inputChannel = "sqsSendChannel")
public MessageHandler sqsMessageHandler() {
return new SqsMessageHandler(queueMessagingTemplate());
}
}
````

2
build.gradle
View File

@@ -41,7 +41,7 @@ if (project.hasProperty('platformVersion')) {
ext {
commonsIoVersion='2.4'
servletApiVersion = '3.1.0'
slf4jVersion = '1.7.12'
slf4jVersion = '1.7.13'
springCloudAwsVersion = '1.1.0.M2'
springIntegrationVersion = '4.2.5.RELEASE'

69
src/main/java/org/springframework/integration/aws/config/xml/AbstractAWSOutboundChannelAdapterParser.java
View File

@@ -1,69 +0,0 @@
/*
* Copyright 2002-2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.config.xml;
import static org.springframework.integration.aws.config.xml.AmazonWSParserUtils.getAmazonWSCredentials;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.messaging.MessageHandler;
import org.w3c.dom.Element;
/**
* The common adapter parser for all AWS Outbound channel adapters
*
* @author Amol Nayak
* @author Rob Harrop
*
* @since 0.5
*
*/
public abstract class AbstractAWSOutboundChannelAdapterParser extends
AbstractOutboundChannelAdapterParser {
/* (non-Javadoc)
* @see org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser#parseConsumer(org.w3c.dom.Element, org.springframework.beans.factory.xml.ParserContext)
*/
@Override
protected final AbstractBeanDefinition parseConsumer(Element element,
ParserContext parserContext) {
String awsCredentialsGeneratedName = getAmazonWSCredentials(element,parserContext);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(getMessageHandlerImplementation());
builder.addConstructorArgReference(awsCredentialsGeneratedName);
processBeanDefinition(builder,awsCredentialsGeneratedName,element,parserContext);
return builder.getBeanDefinition();
}
protected abstract Class<? extends MessageHandler> getMessageHandlerImplementation();
/**
* The subclasses can override this method to set additional attributes and perform some
* additional operations on the {@link BeanDefinitionBuilder}
*
* @param builder
* @param awsCredentialsGeneratedName
* @param element
* @param context
*/
protected void processBeanDefinition(BeanDefinitionBuilder builder,String awsCredentialsGeneratedName,
Element element,ParserContext context) {
//Default implementation does nothing
}
}

6
src/main/java/org/springframework/integration/aws/config/xml/AWSNamespaceHandler.java → src/main/java/org/springframework/integration/aws/config/xml/AwsNamespaceHandler.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.aws.config.xml;
import org.springframework.integration.aws.s3.config.xml.AmazonS3InboundChannelAdapterParser;
import org.springframework.integration.aws.s3.config.xml.AmazonS3OutboundChannelAdapterParser;
import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
/**
@@ -27,11 +26,12 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
* @author Artem Bilan
* @since 0.5
*/
public class AWSNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public class AwsNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public void init() {
registerBeanDefinitionParser("s3-outbound-channel-adapter", new AmazonS3OutboundChannelAdapterParser());
registerBeanDefinitionParser("s3-outbound-channel-adapter", new S3OutboundChannelAdapterParser());
registerBeanDefinitionParser("s3-outbound-gateway", new S3OutboundGatewayParser());
registerBeanDefinitionParser("s3-inbound-channel-adapter", new AmazonS3InboundChannelAdapterParser());
registerBeanDefinitionParser("sqs-outbound-channel-adapter", new SqsOutboundChannelAdapterParser());
registerBeanDefinitionParser("sqs-message-driven-channel-adapter", new SqsMessageDrivenChannelAdapterParser());

7
src/main/java/org/springframework/integration/aws/config/xml/AmazonWSParserUtils.java → src/main/java/org/springframework/integration/aws/config/xml/AwsParserUtils.java
View File

@@ -34,7 +34,7 @@ import org.springframework.util.StringUtils;
* @since 0.5
*
*/
public final class AmazonWSParserUtils {
public final class AwsParserUtils {
public static final String ACCESS_KEY = "accessKey";
@@ -48,10 +48,11 @@ public final class AmazonWSParserUtils {
public static final String SNS_REF = "sns";
public static final String S3_REF = "s3";
public static final String RESOURCE_ID_RESOLVER_REF = "resource-id-resolver";
private AmazonWSParserUtils() {
throw new AssertionError("Cannot instantiate the utility class");
private AwsParserUtils() {
}

40
src/main/java/org/springframework/integration/aws/config/xml/S3OutboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
/**
* The parser for the {@code <int-aws:sns-outbound-channel-adapter>}
* @author Artem Bilan
*/
public class S3OutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
AbstractBeanDefinition beanDefinition = new S3OutboundGatewayParser()
.parseHandler(element, parserContext)
.getBeanDefinition();
beanDefinition.getConstructorArgumentValues().addIndexedArgumentValue(2, false);
return beanDefinition;
}
}

102
src/main/java/org/springframework/integration/aws/config/xml/S3OutboundGatewayParser.java Normal file
View File

@@ -0,0 +1,102 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.aws.outbound.S3MessageHandler;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
* @author Artem Bilan
*/
public class S3OutboundGatewayParser extends AbstractConsumerEndpointParser {
@Override
protected String getInputChannelAttributeName() {
return "request-channel";
}
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
String s3 = element.getAttribute(AwsParserUtils.S3_REF);
boolean hasS3 = StringUtils.hasText(s3);
String transferManager = element.getAttribute("transfer-manager");
boolean hasTransferManager = StringUtils.hasText(transferManager);
if (hasS3 == hasTransferManager) {
parserContext.getReaderContext()
.error("One and only of 's3' and 'transfer-manager' attributes must be provided", element);
}
BeanDefinition bucketExpression =
IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("bucket", "bucket-expression",
parserContext, element, true);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(S3MessageHandler.class)
.addConstructorArgReference(hasS3 ? s3 : transferManager)
.addConstructorArgValue(bucketExpression)
.addConstructorArgValue(true);
BeanDefinition commandExpression =
IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("command",
"command-expression", parserContext, element, false);
if (commandExpression != null) {
builder.addPropertyValue("commandExpression", commandExpression);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "progress-listener");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "upload-metadata-provider");
BeanDefinition keyExpression =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("key-expression", element);
if (keyExpression != null) {
builder.addPropertyValue("keyExpression", keyExpression);
}
BeanDefinition objectAclExpression =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("object-acl-expression", element);
if (objectAclExpression != null) {
builder.addPropertyValue("objectAclExpression", objectAclExpression);
}
BeanDefinition destinationBucketExpression =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("destination-bucket-expression",
element);
if (destinationBucketExpression != null) {
builder.addPropertyValue("destinationBucketExpression", destinationBucketExpression);
}
BeanDefinition destinationKeyExpression =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("destination-key-expression", element);
if (destinationKeyExpression != null) {
builder.addPropertyValue("destinationKeyExpression", destinationKeyExpression);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
return builder;
}
}

2
src/main/java/org/springframework/integration/aws/config/xml/SnsInboundChannelAdapterParser.java
View File

@@ -59,7 +59,7 @@ public class SnsInboundChannelAdapterParser extends AbstractSingleBeanDefinition
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
builder.addConstructorArgReference(element.getAttribute(AmazonWSParserUtils.SNS_REF))
builder.addConstructorArgReference(element.getAttribute(AwsParserUtils.SNS_REF))
.addConstructorArgValue(element.getAttribute("path"));
String channelName = element.getAttribute("channel");
if (!StringUtils.hasText(channelName)) {

2
src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundGatewayParser.java
View File

@@ -37,7 +37,7 @@ public class SnsOutboundGatewayParser extends AbstractConsumerEndpointParser {
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
String sns = element.getAttribute(AmazonWSParserUtils.SNS_REF);
String sns = element.getAttribute(AwsParserUtils.SNS_REF);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SnsMessageHandler.class)
.addConstructorArgReference(sns)

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

@@ -57,7 +57,7 @@ public class SqsMessageDrivenChannelAdapterParser extends AbstractSingleBeanDefi
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String sqs = element.getAttribute(AmazonWSParserUtils.SQS_REF);
String sqs = element.getAttribute(AwsParserUtils.SQS_REF);
if (!StringUtils.hasText(sqs)) {
parserContext.getReaderContext().error("'sqs' attribute is required.", element);
}
@@ -70,7 +70,7 @@ public class SqsMessageDrivenChannelAdapterParser extends AbstractSingleBeanDefi
builder.addPropertyReference("outputChannel", channelName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
AmazonWSParserUtils.RESOURCE_ID_RESOLVER_REF);
AwsParserUtils.RESOURCE_ID_RESOLVER_REF);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");

10
src/main/java/org/springframework/integration/aws/config/xml/SqsOutboundChannelAdapterParser.java
View File

@@ -43,19 +43,19 @@ public class SqsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap
String template = element.getAttribute(QUEUE_MESSAGING_TEMPLATE_REF);
boolean hasTemplate = StringUtils.hasText(template);
String sqs = element.getAttribute(AmazonWSParserUtils.SQS_REF);
String sqs = element.getAttribute(AwsParserUtils.SQS_REF);
boolean hasSqs = StringUtils.hasText(sqs);
String resourceIdResolver = element.getAttribute(AmazonWSParserUtils.RESOURCE_ID_RESOLVER_REF);
String resourceIdResolver = element.getAttribute(AwsParserUtils.RESOURCE_ID_RESOLVER_REF);
boolean hasResourceIdResolver = StringUtils.hasText(resourceIdResolver);
if (hasTemplate && (hasSqs || hasResourceIdResolver)) {
parserContext.getReaderContext().error(QUEUE_MESSAGING_TEMPLATE_REF +
" should not be defined in conjunction with " + AmazonWSParserUtils.SQS_REF
+ " or " + AmazonWSParserUtils.RESOURCE_ID_RESOLVER_REF, element);
" should not be defined in conjunction with " + AwsParserUtils.SQS_REF
+ " or " + AwsParserUtils.RESOURCE_ID_RESOLVER_REF, element);
}
if (!hasTemplate && !hasSqs) {
parserContext.getReaderContext().error("One of " + QUEUE_MESSAGING_TEMPLATE_REF + " or "
+ AmazonWSParserUtils.SQS_REF + " must be defined.", element);
+ AwsParserUtils.SQS_REF + " must be defined.", element);
}
if (hasSqs) {

16
src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java
View File

@@ -219,7 +219,7 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler {
* Specify a {@link S3ProgressListener} for upload and download operations.
* @param s3ProgressListener the {@link S3ProgressListener} to use.
*/
public void setS3ProgressListener(S3ProgressListener s3ProgressListener) {
public void setProgressListener(S3ProgressListener s3ProgressListener) {
this.s3ProgressListener = s3ProgressListener;
}
@@ -241,16 +241,10 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
Command command;
if (this.commandExpression instanceof ValueExpression) {
command = (Command) this.commandExpression.getValue();
}
else {
command = this.commandExpression.getValue(this.evaluationContext, requestMessage, Command.class);
Assert.state(command != null, "'commandExpression' ["
+ this.commandExpression.getExpressionString()
+ "] cannot evaluate to null.");
}
Command command = this.commandExpression.getValue(this.evaluationContext, requestMessage, Command.class);
Assert.state(command != null, "'commandExpression' ["
+ this.commandExpression.getExpressionString()
+ "] cannot evaluate to null.");
Transfer transfer = null;

230
src/main/java/org/springframework/integration/aws/s3/AmazonS3MessageHandler.java
View File

@@ -1,230 +0,0 @@
/*
* Copyright 2002-2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.s3;
import static org.springframework.integration.aws.s3.AmazonS3MessageHeaders.METADATA;
import static org.springframework.integration.aws.s3.AmazonS3MessageHeaders.OBJECT_ACLS;
import static org.springframework.integration.aws.s3.AmazonS3MessageHeaders.USER_METADATA;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.InputStream;
import java.util.Collection;
import java.util.Map;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.expression.Expression;
import org.springframework.messaging.Message;
import org.springframework.integration.aws.core.AWSCredentials;
import org.springframework.integration.aws.s3.core.AmazonS3Object;
import org.springframework.integration.aws.s3.core.AmazonS3OperationException;
import org.springframework.integration.aws.s3.core.AmazonS3Operations;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.util.Assert;
/**
* The Message handler for the S3 outbound channel adapter
*
* @author Amol Nayak
* @author Rob Harrop
* @author Artem Bilan
*
* @since 0.5
*
*/
public class AmazonS3MessageHandler extends AbstractMessageHandler {
private final AWSCredentials credentials;
private final AmazonS3Operations operations;
private volatile String charset = "UTF-8";
private volatile String bucket;
private volatile ExpressionEvaluatingMessageProcessor<String> remoteDirectoryProcessor;
private volatile FileNameGenerationStrategy fileNameGenerator = new DefaultFileNameGenerationStrategy();
private volatile boolean fileNameGeneratorSet;
@Override
protected void onInit() throws Exception {
super.onInit();
Assert.hasText(bucket,"Bucket not set'");
Assert.notNull(remoteDirectoryProcessor,
"Remote Directory processor should be present, set the remote directory expression");
if (!this.fileNameGeneratorSet && this.fileNameGenerator instanceof BeanFactoryAware) {
((BeanFactoryAware) this.fileNameGenerator).setBeanFactory(getBeanFactory());
}
this.remoteDirectoryProcessor.setBeanFactory(getBeanFactory());
}
/**
* The constructor that initializes {@link AmazonS3MessageHandler} with the provided
* implementation of {@link AmazonS3Operations} and using the provided {@link AWSCredentials}
* @param credentials
* @param operations
*/
public AmazonS3MessageHandler(AWSCredentials credentials, AmazonS3Operations operations) {
Assert.notNull(operations,"s3 operations is null");
Assert.notNull(credentials,"AWS Credentials are null");
this.credentials = credentials;
this.operations = operations;
}
/**
* The handler implementation for the Amazon S3 used to put objects in the remote AWS S3 bucket
* the message should contain a valid payload of type {@link File}, {@link InputStream},
* byte[] or {@link String}. Various predetermined headers as defined in {@link AmazonS3MessageHeaders}
* are extracted from the message and an {@link AmazonS3Object} is constructed that is provided to
* the {@link AmazonS3Operations} implementation to be uploaded in S3.
* @param message
*/
@Override
@SuppressWarnings("unchecked")
protected void handleMessageInternal(Message<?> message) throws Exception {
Object payload = message.getPayload();
//The payload can be only of type java.io.File, java.io.InputStream, byte[] or String
File file = null;
InputStream in = null;
//potentially unsafe operation if the types are not as those expected
Map<String, String> userMetaData = getHeaderValue(message,USER_METADATA,Map.class);
Map<String, Object> metaData = getHeaderValue(message,METADATA,Map.class);
Map<String, Collection<String>> objectAcls = getHeaderValue(message,OBJECT_ACLS,Map.class);
AmazonS3ObjectBuilder builder = AmazonS3ObjectBuilder
.getInstance()
.withMetaData(metaData)
.withUserMetaData(userMetaData)
.withObjectACL(objectAcls);
String folder = this.remoteDirectoryProcessor.processMessage(message);
String objectName = this.fileNameGenerator.generateFileName(message);
if(payload instanceof File) {
file = (File)payload;
}
else if (payload instanceof InputStream) {
in = (InputStream)payload;
}
else if(payload instanceof byte[]) {
in = new ByteArrayInputStream((byte[])payload);
}
else if(payload instanceof String) {
in = new ByteArrayInputStream(((String)payload).getBytes(charset));
}
else {
throw new AmazonS3OperationException
(credentials.getAccessKey(),
bucket, objectName, "The Message payload is of unexpected type "
+ payload.getClass().getCanonicalName() + ", only supported types are"
+" java.io.File, java.io.InputStream, byte[] and java.lang.String");
}
if(file != null) {
builder.fromFile(file);
}
else {
builder.fromInputStream(in);
}
AmazonS3Object object = builder.build();
if(logger.isDebugEnabled()) {
logger.debug("Uploading Object to bucket " + bucket + ", to folder " + folder + ", with object name " + objectName);
}
operations.putObject(bucket, folder, objectName, object);
}
/**
* The common helper method that would read the message header and checks if it is of a particular type or not
* @param <T>
* @param message
* @param headerName
* @param expectedType
*/
@SuppressWarnings("unchecked")
private <T> T getHeaderValue(Message<?> message, String headerName, Class<T> expectedType) {
T header = null;
Object genericHeader = message.getHeaders().get(headerName);
if(genericHeader == null) {
return null;
}
if(expectedType.isAssignableFrom(genericHeader.getClass())) {
header = (T) genericHeader;
}
else {
logger.warn("Found header " + USER_METADATA + " in the message but was not of required type");
}
return header;
}
/**
* Sets the charset for the String payload received
* @param charset
*/
public void setCharset(String charset) {
Assert.hasText(charset,"'charset' should be non null, non empty string");
this.charset = charset;
}
/**
* Sets the S3 Bucket to which the files are to be uploaded
* @param bucket
*/
public void setBucket(String bucket) {
Assert.hasText(bucket, "'bucket' should be non null, non empty string");
this.bucket = bucket;
}
/**
* Sets the directory evaluating expression for finding the remote directory in S3
* @param expression
*/
public void setRemoteDirectoryExpression(Expression expression) {
Assert.notNull(expression, "Remote directory expression is null");
remoteDirectoryProcessor = new ExpressionEvaluatingMessageProcessor<String>(expression);
}
/**
* Sets the file name generation strategy
* @param fileNameGenerator
*/
public void setFileNameGenerator(FileNameGenerationStrategy fileNameGenerator) {
Assert.notNull(fileNameGenerator,"File name generator is null");
this.fileNameGenerator = fileNameGenerator;
this.fileNameGeneratorSet = true;
}
}

7
src/main/java/org/springframework/integration/aws/s3/config/xml/AmazonS3InboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-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.
@@ -13,8 +13,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.s3.config.xml;
import static org.springframework.integration.aws.config.xml.AmazonWSParserUtils.getAmazonWSCredentials;
import org.w3c.dom.Element;
@@ -23,6 +23,7 @@ import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.aws.config.xml.AwsParserUtils;
import org.springframework.integration.aws.s3.AmazonS3InboundSynchronizationMessageSource;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
@@ -60,7 +61,7 @@ public class AmazonS3InboundChannelAdapterParser extends
@Override
protected BeanMetadataElement parseSource(Element element,
ParserContext parserContext) {
String awsCredentials = getAmazonWSCredentials(element, parserContext);
String awsCredentials = AwsParserUtils.getAmazonWSCredentials(element, parserContext);
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(AmazonS3InboundSynchronizationMessageSource.class);
builder.addPropertyReference(AWS_CREDENTIAL, awsCredentials);

159
src/main/java/org/springframework/integration/aws/s3/config/xml/AmazonS3OutboundChannelAdapterParser.java
View File

@@ -1,159 +0,0 @@
/*
* Copyright 2002-2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.s3.config.xml;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.aws.config.xml.AbstractAWSOutboundChannelAdapterParser;
import org.springframework.integration.aws.s3.AmazonS3MessageHandler;
import org.springframework.integration.aws.s3.DefaultFileNameGenerationStrategy;
import org.springframework.integration.aws.s3.core.DefaultAmazonS3Operations;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* The namespace parser for outbound-channel-parser for the aws-s3 namespace
*
* @author Amol Nayak
* @author Rob Harrop
*
* @since 0.5
*
*/
public class AmazonS3OutboundChannelAdapterParser extends
AbstractAWSOutboundChannelAdapterParser {
private static final String S3_OPERATIONS = "s3-operations";
private static final String AWS_ENDPOINT = "aws-endpoint";
private static final String S3_BUCKET = "bucket";
private static final String CHARSET = "charset";
private static final String MULTIPART_THRESHOLD = "multipart-upload-threshold";
private static final String TEMPORARY_DIRECTORY = "temporary-directory";
private static final String TEMPORARY_SUFFIX = "temporary-suffix";
private static final String THREADPOOL_EXECUTOR = "thread-pool-executor";
private static final String REMOTE_DIRECTORY = "remote-directory";
private static final String REMOTE_DIRECTORY_EXPRESSION = "remote-directory-expression";
private static final String FILE_NAME_GENERATOR = "file-name-generator";
private static final String FILE_NAME_GENERATION_EXPRESSION = "file-name-generation-expression";
/* (non-Javadoc)
* @see org.springframework.integration.aws.core.config.AbstractAWSOutboundChannelAdapterParser#getMessageHandlerImplementation()
*/
@Override
protected Class<? extends MessageHandler> getMessageHandlerImplementation() {
return AmazonS3MessageHandler.class;
}
/**
* This is where we will be instantiating the AmazonS3Operations instance and
* passing it to the MessageHandler
*/
@Override
protected void processBeanDefinition(BeanDefinitionBuilder builder,
String awsCredentialsGeneratedName,Element element, ParserContext context) {
//TODO: When we will have more than one implementations, also provision with an enum
//for the operation
String s3Operations = element.getAttribute(S3_OPERATIONS);
String operationsService;
if(StringUtils.hasText(s3Operations)) {
//custom implementation provided
if(element.hasAttribute(MULTIPART_THRESHOLD)
|| element.hasAttribute(TEMPORARY_DIRECTORY)
|| element.hasAttribute(TEMPORARY_SUFFIX)
|| element.hasAttribute(THREADPOOL_EXECUTOR)) {
throw new BeanDefinitionStoreException("Attributes '" + MULTIPART_THRESHOLD + "', '"
+ TEMPORARY_DIRECTORY + "', '" + TEMPORARY_SUFFIX + "' and '" + THREADPOOL_EXECUTOR
+ " are mutually exclusive to the '" + S3_OPERATIONS + "' attribute");
}
operationsService = s3Operations;
}
else {
BeanDefinitionBuilder s3OpBuilder = BeanDefinitionBuilder.genericBeanDefinition(DefaultAmazonS3Operations.class);
s3OpBuilder.addConstructorArgReference(awsCredentialsGeneratedName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(s3OpBuilder, element, MULTIPART_THRESHOLD);
IntegrationNamespaceUtils.setValueIfAttributeDefined(s3OpBuilder, element, TEMPORARY_DIRECTORY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(s3OpBuilder, element, TEMPORARY_SUFFIX,"temporaryFileSuffix");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(s3OpBuilder, element, THREADPOOL_EXECUTOR);
IntegrationNamespaceUtils.setValueIfAttributeDefined(s3OpBuilder, element, AWS_ENDPOINT);
operationsService = BeanDefinitionReaderUtils.registerWithGeneratedName(s3OpBuilder.getBeanDefinition(), context.getRegistry());
}
//Set the bucket and charset
builder.addConstructorArgReference(operationsService);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CHARSET);
builder.addPropertyValue(S3_BUCKET, element.getAttribute(S3_BUCKET)); //Mandatory
//Get the remote directory expression or remote directory literal string
String remoteDirectoryLiteral = element.getAttribute(REMOTE_DIRECTORY);
String remoteDirectoryExpression = element.getAttribute(REMOTE_DIRECTORY_EXPRESSION);
boolean hasRemoteDirectoryExpression = StringUtils.hasText(remoteDirectoryExpression);
boolean hasRemoteDirectoryLiteral = StringUtils.hasText(remoteDirectoryLiteral);
if(!(hasRemoteDirectoryExpression ^ hasRemoteDirectoryLiteral)) {
throw new BeanDefinitionStoreException("Exactly one of " + REMOTE_DIRECTORY + " or "
+ REMOTE_DIRECTORY_EXPRESSION + " is required");
}
AbstractBeanDefinition expression;
if(hasRemoteDirectoryLiteral) {
expression = BeanDefinitionBuilder.genericBeanDefinition(LiteralExpression.class)
.addConstructorArgValue(remoteDirectoryLiteral)
.getBeanDefinition();
}
else {
expression = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class)
.addConstructorArgValue(remoteDirectoryExpression)
.getBeanDefinition();
}
builder.addPropertyValue("remoteDirectoryExpression", expression);
//Get the File generation strategy
String fileNameGenerator = element.getAttribute(FILE_NAME_GENERATOR);
String fileNameGenerationExpression = element.getAttribute(FILE_NAME_GENERATION_EXPRESSION);
boolean hasFileGenerator = StringUtils.hasText(fileNameGenerator);
boolean hasFileGenerationExpression = StringUtils.hasText(fileNameGenerationExpression);
if(hasFileGenerationExpression && hasFileGenerator) {
throw new BeanDefinitionStoreException("Attributes '" + FILE_NAME_GENERATION_EXPRESSION + "' and '"
+ FILE_NAME_GENERATOR + "' are mutually exclusive, at most one might be specified");
}
if(hasFileGenerator) {
builder.addPropertyReference("fileNameGenerator", fileNameGenerator);
}
else {
BeanDefinitionBuilder fileNameGeneratorBuilder =
BeanDefinitionBuilder.genericBeanDefinition(DefaultFileNameGenerationStrategy.class);
String tempDirectorySuffix = element.getAttribute(TEMPORARY_SUFFIX);
if(StringUtils.hasText(tempDirectorySuffix)) {
fileNameGeneratorBuilder.addPropertyValue("temporarySuffix",tempDirectorySuffix);
}
if(hasFileGenerationExpression) {
fileNameGeneratorBuilder.addPropertyValue("fileNameExpression", fileNameGenerationExpression);
}
builder.addPropertyValue("fileNameGenerator", fileNameGeneratorBuilder.getBeanDefinition());
}
}
}

2
src/main/resources/META-INF/spring.handlers
View File

@@ -1 +1 @@
http\://www.springframework.org/schema/integration/aws=org.springframework.integration.aws.config.xml.AWSNamespaceHandler
http\://www.springframework.org/schema/integration/aws=org.springframework.integration.aws.config.xml.AwsNamespaceHandler

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

@@ -62,161 +62,228 @@
</xsd:attributeGroup>
<xsd:element name="s3-outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines a Consumer Endpoint for the 'org.springframework.integration.aws.outbound.S3MessageHandler'
with one-way behaviour to perform Amazon S3 operations .
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an outbound S3 Channel Adapter for Uploading files to Amazon S3
</xsd:documentation>
</xsd:annotation>
<xsd:choice minOccurs="0" maxOccurs="2">
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType" minOccurs="0"
maxOccurs="1"/>
</xsd:choice>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attributeGroup ref="awsAdaptersCommonAttributes"/>
<xsd:attribute name="bucket" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
The mandatory attribute that would be used to provide the AWS bucket to which
the objects needs to be uploaded.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="charset" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Relevant only when the payload of the message to the outbound adapter is of
type java.lang.String. The default charset is UTF-8.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="multipart-upload-threshold" type="xsd:integer">
<xsd:annotation>
<xsd:documentation>
Using 'Amazon Multipart Upload' you can upload data as a set of parts using
parallel threads. This non-negative integer value representing bytes which is used
to provide the threshold after which the upload to the S3 bucket will be done
using 'Amazon Multipart Upload'. Amazon recommends the size to be 100 MB.
The minimum threshold for 'Amazon Multipart Upload' is 5120 bytes.
If the attribute is not specified, then the value used is the default value used
by the underlying implementation. The default implementation uses AWS SDK which
uses Multi part upload after 16 MB. The maximum value for this is 2 GB. Any value
greater than 2 GB will not throw an exception but the value will be set to
2 GB internally for the threshold.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="temporary-directory" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
If the payload of the message is an InputStream, byte[] or String
the contents are written to a temporary file in the provided temporary directory
location before being uploaded to the S3. In absence of this attribute, the
value is defaulted to the value of the system property "java.io.tmpdir"
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="temporary-suffix" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The suffix for the files if a temporary file is to be generated. The value
defaults to ".writing"
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="thread-pool-executor" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.concurrent.ThreadPoolExecutor"/>
</tool:annotation>
<xsd:documentation>
The thread pool executor to be used for multi part uploads.
If none is provided, the default one used by the underlying SDK
or library will be used.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="remote-directory" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The String literal that gives the remote folder in the provided bucket where
the files will be uploaded. This attribute is mutually exclusive to
remote-directory-expression
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="remote-directory-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
This attribute is mutually exclusive with the remote-directory attribute
and is used to provide an expression that would be evaluated against the incoming
message to derive the remote directory name in the given bucket.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="file-name-generator" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.aws.s3.FileNameGenerationStrategy"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The instance that would be used to generate the name of the file that would be
stored in S3. If none is specified then
org.springframework.integration.aws.s3.DefaultFileNameGenerationStrategy would be used.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="file-name-generation-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The filename generation expression that is mutually exclusive to the
file-name-generator attribute. The default expression is "headers['file_name']"
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="s3-operations" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:expected-type type="org.springframework.integration.aws.s3.core.AmazonS3Operations"/>
</xsd:appinfo>
<xsd:documentation>
Reference to the bean with an implementation of
org.springframework.integration.aws.s3.core.AmazonS3Operations
that would be used to perform the operations on the S3 bucket. If not provided, the
default implementation used is
org.springframework.integration.aws.s3.core.DefaultAmazonS3Operations which is the
implementation using the AWS SDK.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="aws-endpoint" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The String that gives the endpoint to use for the adapter, if none is
specified the default used is s3.amazonaws.com.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:complexContent>
<xsd:extension base="s3OutboundAttributes">
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="s3-outbound-gateway">
<xsd:annotation>
<xsd:documentation>
Defines a Consumer Endpoint for the 'org.springframework.integration.aws.outbound.S3MessageHandler'
with request-reply behaviour to perform Amazon S3 operations.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="s3OutboundAttributes">
<xsd:attribute name="request-channel" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Identifies the request channel attached to this gateway.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Identifies the reply channel attached to this gateway.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Allows you to specify how long this gateway will wait for
the reply message to be sent successfully to the reply channel
before throwing an exception. This attribute only applies when the
channel might block, for example when using a bounded queue channel that
is currently full.
Also, keep in mind that when sending to a DirectChannel, the
invocation will occur in the sender's thread. Therefore,
the failing of the send operation may be caused by other
components further downstream.
The "reply-timeout" attribute maps to the "sendTimeout" property of the
underlying 'MessagingTemplate' instance
(org.springframework.integration.core.MessagingTemplate).
The attribute will default, if not specified, to '-1', meaning that
by default, the Gateway will wait indefinitely. The value is
specified in milliseconds.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="s3OutboundAttributes">
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType" minOccurs="0"
maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="s3">
<xsd:annotation>
<xsd:documentation>
Reference to an instance of 'com.amazonaws.services.s3.AmazonS3'.
Mutually exclusive with the 'transfer-manager'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="com.amazonaws.services.s3.AmazonS3"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="transfer-manager">
<xsd:annotation>
<xsd:documentation>
Reference to an instance of 'com.amazonaws.services.s3.transfer.TransferManager'.
Mutually exclusive with the 's3'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="com.amazonaws.services.s3.transfer.TransferManager"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="bucket">
<xsd:annotation>
<xsd:documentation>
The S3 bucket to use.
Mutually exclusive with 'bucket-expression'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="bucket-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to evaluate S3 bucket at runtime against request message.
Mutually exclusive with 'bucket'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="command">
<xsd:annotation>
<xsd:documentation>
The S3MessageHandler operation command.
Mutually exclusive with 'command-expression'.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="s3CommandType xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="command-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to evaluate S3MessageHandler operation command at runtime against request message.
Mutually exclusive with 'command'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="progress-listener">
<xsd:annotation>
<xsd:documentation>
Reference to an instance of 'com.amazonaws.services.s3.transfer.internal.S3ProgressListener'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="com.amazonaws.services.s3.transfer.internal.S3ProgressListener"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="upload-metadata-provider">
<xsd:annotation>
<xsd:documentation>
Reference to an instance of
'org.springframework.integration.aws.outbound.S3MessageHandler$UploadMetadataProvider'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.aws.outbound.S3MessageHandler$UploadMetadataProvider"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to evaluate S3Object key at runtime against request message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="object-acl-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to evaluate S3Object ACL at runtime against request message
for the 'upload' operation.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="destination-bucket-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to evaluate destination S3 bucket at runtime against request message
for the 'copy' operation.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="destination-key-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to evaluate destination S3Object key at runtime against request message
for the 'copy' operation.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:simpleType name="s3CommandType">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="UPLOAD" />
<xsd:enumeration value="DOWNLOAD" />
<xsd:enumeration value="COPY" />
</xsd:restriction>
</xsd:simpleType>
<xsd:element name="s3-inbound-channel-adapter">
<xsd:complexType>
<xsd:annotation>
@@ -443,7 +510,8 @@
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The 'org.springframework.core.task.AsyncTaskExecutor' 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>
@@ -520,10 +588,10 @@
<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:enumeration value="NO_REDRIVE"/>
<xsd:enumeration value="ALWAYS"/>
<xsd:enumeration value="NEVER"/>
<xsd:enumeration value="ON_SUCCESS"/>
</xsd:restriction>
</xsd:simpleType>
@@ -630,7 +698,7 @@
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel" />
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
@@ -642,7 +710,7 @@
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel" />
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
@@ -674,7 +742,7 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="id" type="xsd:string" />
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:extension>
</xsd:complexContent>
@@ -802,6 +870,14 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order">
<xsd:annotation>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>

51
src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests-context.xml Normal file
View File

@@ -0,0 +1,51 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/aws http://www.springframework.org/schema/integration/aws/spring-integration-aws.xsd">
<bean id="s3" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="com.amazonaws.services.s3.AmazonS3"/>
</bean>
<bean id="s3ProgressListener" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="com.amazonaws.services.s3.transfer.internal.S3ProgressListener"/>
</bean>
<bean id="uploadMetadataProvider" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.integration.aws.outbound.S3MessageHandler$UploadMetadataProvider"/>
</bean>
<int-aws:s3-outbound-channel-adapter s3="s3"
auto-startup="false"
channel="errorChannel"
phase="100"
id="s3OutboundChannelAdapter"
bucket="foo"
key-expression="payload.name"
command="COPY"
destination-bucket-expression="'bar'"
destination-key-expression="'baz'"
object-acl-expression="'qux'"
progress-listener="s3ProgressListener"
upload-metadata-provider="uploadMetadataProvider"/>
<bean id="transferManager" class="com.amazonaws.services.s3.transfer.TransferManager"/>
<int-aws:s3-outbound-gateway transfer-manager="transferManager"
request-channel="errorChannel"
id="s3OutboundGateway"
bucket-expression="'FOO'"
command-expression="'DOWNLOAD'"
reply-channel="nullChannel"/>
<!--Invalid configs-->
<!--One of 'bucket' or 'bucket-expression' is required-->
<!--<int-aws:s3-outbound-channel-adapter s3="s3" id="bucketRequired"/>-->
<!--One and only of 's3' and 'transfer-manager' attributes must be provided-->
<!--<int-aws:s3-outbound-channel-adapter s3="s3" transfer-manager="transferManager" id="onlyOneS3OrTransferManager"/>-->
</beans>

144
src/test/java/org/springframework/integration/aws/config/xml/S3MessageHandlerParserTests.java Normal file
View File

@@ -0,0 +1,144 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.config.xml;
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 org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.aws.outbound.S3MessageHandler;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.transfer.TransferManager;
import com.amazonaws.services.s3.transfer.internal.S3ProgressListener;
/**
* @author Artem Bilan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class S3MessageHandlerParserTests {
@Autowired
private AmazonS3 amazonS3;
@Autowired
private TransferManager transferManager;
@Autowired
private MessageChannel errorChannel;
@Autowired
private MessageChannel nullChannel;
@Autowired
private EventDrivenConsumer s3OutboundChannelAdapter;
@Autowired
@Qualifier("s3OutboundChannelAdapter.handler")
private MessageHandler s3OutboundChannelAdapterHandler;
@Autowired
private EventDrivenConsumer s3OutboundGateway;
@Autowired
@Qualifier("s3OutboundGateway.handler")
private MessageHandler s3OutboundGatewayHandler;
@Autowired
private S3ProgressListener progressListener;
@Autowired
private S3MessageHandler.UploadMetadataProvider uploadMetadataProvider;
@Autowired
private BeanFactory beanFactory;
@Test
public void testS3OutboundChannelAdapterParser() {
assertSame(this.amazonS3,
TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "transferManager.s3"));
assertEquals("foo", TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler,
"bucketExpression.literalValue"));
assertEquals("'bar'", TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler,
"destinationBucketExpression.expression"));
assertEquals("'baz'", TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler,
"destinationKeyExpression.expression"));
assertEquals("payload.name", TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler,
"keyExpression.expression"));
assertEquals("'qux'", TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler,
"objectAclExpression.expression"));
assertEquals(S3MessageHandler.Command.COPY.name(),
TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "commandExpression.literalValue"));
assertFalse(TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "produceReply", Boolean.class));
assertSame(this.progressListener,
TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "s3ProgressListener"));
assertSame(this.uploadMetadataProvider,
TestUtils.getPropertyValue(this.s3OutboundChannelAdapterHandler, "uploadMetadataProvider"));
assertEquals(100, this.s3OutboundChannelAdapter.getPhase());
assertFalse(this.s3OutboundChannelAdapter.isAutoStartup());
assertFalse(this.s3OutboundChannelAdapter.isRunning());
assertSame(this.errorChannel, TestUtils.getPropertyValue(this.s3OutboundChannelAdapter, "inputChannel"));
assertSame(this.s3OutboundChannelAdapterHandler,
TestUtils.getPropertyValue(this.s3OutboundChannelAdapter, "handler"));
}
@Test
public void testS3OutboundGatewayParser() {
assertSame(this.transferManager,
TestUtils.getPropertyValue(this.s3OutboundGatewayHandler, "transferManager"));
assertEquals("'FOO'", TestUtils.getPropertyValue(this.s3OutboundGatewayHandler,
"bucketExpression.expression"));
Expression commandExpression =
TestUtils.getPropertyValue(this.s3OutboundGatewayHandler, "commandExpression", Expression.class);
assertEquals("'" + S3MessageHandler.Command.DOWNLOAD.name() + "'",
TestUtils.getPropertyValue(commandExpression, "expression"));
StandardEvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
S3MessageHandler.Command command =
commandExpression.getValue(evaluationContext, S3MessageHandler.Command.class);
assertEquals(S3MessageHandler.Command.DOWNLOAD, command);
assertTrue(TestUtils.getPropertyValue(this.s3OutboundGatewayHandler, "produceReply", Boolean.class));
assertSame(this.nullChannel, TestUtils.getPropertyValue(this.s3OutboundGatewayHandler, "outputChannel"));
assertTrue(this.s3OutboundGateway.isRunning());
assertSame(this.errorChannel, TestUtils.getPropertyValue(this.s3OutboundGateway, "inputChannel"));
assertSame(this.s3OutboundGatewayHandler, TestUtils.getPropertyValue(this.s3OutboundGateway, "handler"));
}
}

7
src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java
View File

@@ -57,6 +57,7 @@ import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.http.MediaType;
import org.springframework.integration.annotation.ServiceActivator;
@@ -383,7 +384,9 @@ public class S3MessageHandlerTests {
public MessageHandler s3MessageHandler() {
S3MessageHandler s3MessageHandler = new S3MessageHandler(amazonS3(), "myBucket");
s3MessageHandler.setCommandExpression(PARSER.parseExpression("headers.s3Command"));
s3MessageHandler.setKeyExpression(PARSER.parseExpression("payload instanceof T(java.io.File) ? payload.name : headers.key"));
Expression keyExpression =
PARSER.parseExpression("payload instanceof T(java.io.File) ? payload.name : headers.key");
s3MessageHandler.setKeyExpression(keyExpression);
s3MessageHandler.setObjectAclExpression(new ValueExpression<>(CannedAccessControlList.PublicReadWrite));
s3MessageHandler.setUploadMetadataProvider(new S3MessageHandler.UploadMetadataProvider() {
@@ -397,7 +400,7 @@ public class S3MessageHandlerTests {
}
});
s3MessageHandler.setS3ProgressListener(s3ProgressListener());
s3MessageHandler.setProgressListener(s3ProgressListener());
return s3MessageHandler;
}

271
src/test/java/org/springframework/integration/aws/s3/AmazonS3MessageHandlerTests.java
View File

@@ -1,271 +0,0 @@
/*
* Copyright 2002-2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.s3;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.springframework.integration.aws.s3.AmazonS3MessageHeaders.FILE_NAME;
import static org.springframework.integration.aws.s3.AmazonS3MessageHeaders.METADATA;
import static org.springframework.integration.aws.s3.AmazonS3MessageHeaders.OBJECT_ACLS;
import static org.springframework.integration.aws.s3.AmazonS3MessageHeaders.USER_METADATA;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.InputStream;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.aws.core.BasicAWSCredentials;
import org.springframework.integration.aws.s3.core.AmazonS3Object;
import org.springframework.integration.aws.s3.core.AmazonS3Operations;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
/**
* The test class for {@link AmazonS3MessageHandler}, we rely on mock of {@link AmazonS3Operations}
* to test the behavior.
*
* @author Amol Nayak
* @author Rob Harrop
* @author Artem Bilan
*
* @since 0.5
*
*/
public class AmazonS3MessageHandlerTests {
private static AmazonS3Operations operations;
private static PutObjectParameterHolder holder = new PutObjectParameterHolder();
@Rule
public TemporaryFolder tempFolder = new TemporaryFolder();
@BeforeClass
public static void setup() {
operations = Mockito.mock(AmazonS3Operations.class);
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock inv) {
Object[] args = inv.getArguments();
holder.setBucket((String)args[0]);
holder.setFolder((String)args[1]);
holder.setObjectName((String)args[2]);
holder.setS3Object((AmazonS3Object)args[3]);
return null;
}
}).
when(operations)
.putObject(anyString(), anyString(), anyString(), any(AmazonS3Object.class));
}
private AmazonS3MessageHandler getHandler() {
AmazonS3MessageHandler handler = new AmazonS3MessageHandler(new BasicAWSCredentials(), operations);
//set the remote directory to root by default
handler.setRemoteDirectoryExpression(new LiteralExpression("/"));
handler.setBucket("TestBucket");
handler.setBeanFactory(Mockito.mock(BeanFactory.class));
handler.afterPropertiesSet();
return handler;
}
private static class PutObjectParameterHolder {
private String bucket;
private String folder;
private String objectName;
private AmazonS3Object s3Object;
public String getBucket() {
return bucket;
}
public void setBucket(String bucket) {
this.bucket = bucket;
}
public String getFolder() {
return folder;
}
public void setFolder(String folder) {
this.folder = folder;
}
public String getObjectName() {
return objectName;
}
public void setObjectName(String objectName) {
this.objectName = objectName;
}
public AmazonS3Object getS3Object() {
return s3Object;
}
public void setS3Object(AmazonS3Object s3Object) {
this.s3Object = s3Object;
}
}
/**
* Tests with a message payload of type {@link String}
*/
@Test
public void withStringPayload() {
Message<String> message = MessageBuilder.withPayload("Test String").build();
AmazonS3MessageHandler handler = getHandler();
handler.handleMessage(message);
AmazonS3Object object = holder.getS3Object();
Assert.assertNotNull(object.getInputStream());
Assert.assertNull(object.getFileSource());
assertCommonValues(message,object);
}
/**
* Tests with a message with payload of type {@link InputStream}
*/
@Test
public void withInputStreamPayload() {
InputStream bin = new ByteArrayInputStream("SomeString".getBytes());
Message<InputStream> message = MessageBuilder.withPayload(bin).build();
AmazonS3MessageHandler handler = getHandler();
handler.handleMessage(message);
AmazonS3Object object = holder.getS3Object();
Assert.assertNotNull(object.getInputStream());
Assert.assertNull(object.getFileSource());
assertCommonValues(message,object);
}
/**
* Tests with a message with payload of type byte[]
*/
@Test
public void withByteArrayPayload() {
Message<byte[]> message = MessageBuilder.withPayload("String".getBytes()).build();
AmazonS3MessageHandler handler = getHandler();
handler.handleMessage(message);
AmazonS3Object object = holder.getS3Object();
Assert.assertNotNull(object.getInputStream());
Assert.assertNull(object.getFileSource());
assertCommonValues(message,object);
}
/**
* Tests with a message with payload of type {@link File} which is a file with temporary suffix
*/
@Test
public void withTempFileTypePayload() throws Exception {
final File file = tempFolder.newFile("TempFile.txt.writing");
messageWithFileTypePayload(file);
}
/**
* Tests with a message with payload of type {@link File} which is a file without temporary suffix
*/
@Test
public void withFileTypePayload() throws Exception {
final File file = tempFolder.newFile("TempFile.txt");
messageWithFileTypePayload(file);
}
/**
*Test case to with message of an incompatible type, {@link Integer} in this case.
*
*/
@Test(expected=MessageHandlingException.class)
public void withIncompatiblePayload() {
Message<Integer> message = MessageBuilder.withPayload(1).build();
AmazonS3MessageHandler handler = getHandler();
handler.handleMessage(message);
}
/**
* Tests with all the header provided in the message
*/
@Test
public void withAllHeaders() {
Map<String, Collection<String>> acls = new HashMap<String, Collection<String>>();
acls.put("test@test.com", Arrays.asList("Read", "Write acp"));
Message<String> message = MessageBuilder.withPayload("Test Content")
.setHeader(FILE_NAME, "TestFileName.txt")
.setHeader(USER_METADATA, Collections.singletonMap("UserMD", "UserMD"))
.setHeader(METADATA, Collections.singletonMap("Metadata", "Metadata"))
.setHeader(OBJECT_ACLS, acls)
.setHeader("remoteDirectory", "/remote")
.build();
AmazonS3MessageHandler handler = getHandler();
SpelExpressionParser parser = new SpelExpressionParser();
handler.setRemoteDirectoryExpression(parser.parseExpression("headers['remoteDirectory']"));
handler.afterPropertiesSet();
handler.handleMessage(message);
Assert.assertEquals("TestBucket", holder.getBucket());
Assert.assertEquals("TestFileName.txt", holder.getObjectName());
Assert.assertEquals("/remote", holder.getFolder());
AmazonS3Object object = holder.getS3Object();
Assert.assertNotNull(object);
Assert.assertNotNull(object.getInputStream());
Assert.assertNotNull(object.getMetaData());
Assert.assertNotNull(object.getUserMetaData());
Assert.assertNotNull(object.getObjectACL());
Assert.assertEquals(2,object.getObjectACL().getGrants().size());
}
/**
* The common method to test messages with payload of type {@link File}
* @param file
*/
private void messageWithFileTypePayload(File file) throws Exception {
file.createNewFile();
Message<File> message = MessageBuilder.withPayload(file).build();
AmazonS3MessageHandler handler = getHandler();
handler.handleMessage(message);
AmazonS3Object object = holder.getS3Object();
Assert.assertEquals("TempFile.txt", holder.getObjectName());
Assert.assertNotNull(object.getFileSource());
Assert.assertNull(object.getInputStream());
Assert.assertNull(object.getMetaData());
Assert.assertNull(object.getObjectACL());
Assert.assertNull(object.getUserMetaData());
file.delete();
}
/**
* The method used to assert the values for tests with String, InputStream and byte[] parameters
* @param message
*/
private void assertCommonValues(Message<?> message,AmazonS3Object object) {
Assert.assertEquals(message.getHeaders().getId().toString() + ".ext", holder.getObjectName());
Assert.assertEquals("/", holder.getFolder());
Assert.assertEquals("TestBucket", holder.getBucket());
Assert.assertNotNull(object);
Assert.assertNull(object.getMetaData());
Assert.assertNull(object.getObjectACL());
Assert.assertNull(object.getUserMetaData());
}
}

201
src/test/java/org/springframework/integration/aws/s3/config/xml/AmazonS3OutboundChannelAdapterParserTests.java
View File

@@ -1,201 +0,0 @@
/*
* Copyright 2002-2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aws.s3.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.springframework.integration.test.util.TestUtils.getPropertyValue;
import java.net.URI;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.integration.aws.s3.AmazonS3MessageHandler;
import org.springframework.integration.aws.s3.FileNameGenerationStrategy;
import org.springframework.integration.aws.s3.core.AmazonS3Operations;
import org.springframework.integration.aws.s3.core.DefaultAmazonS3Operations;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
/**
* The test case for the aws-s3 namespace's {@link AmazonS3OutboundChannelAdapterParser} class
* @author Amol Nayak
* @author Rob Harrop
* @author Karthikeyan Palanivelu
* @since 0.5
*/
public class AmazonS3OutboundChannelAdapterParserTests {
private volatile static int adviceCalled;
/**
* Test case for the xml definition with a custom implementation of {@link AmazonS3Operations}
*/
@Test
public void withCustomOperations() {
ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:s3-valid-outbound-cases.xml");
EventDrivenConsumer consumer = ctx.getBean("withCustomService", EventDrivenConsumer.class);
AmazonS3MessageHandler handler = getPropertyValue(consumer, "handler", AmazonS3MessageHandler.class);
assertEquals(AmazonS3DummyOperations.class, getPropertyValue(handler, "operations").getClass());
Expression expression =
getPropertyValue(handler, "remoteDirectoryProcessor.expression", Expression.class);
assertNotNull(expression);
assertEquals(LiteralExpression.class, expression.getClass());
assertEquals("/", getPropertyValue(expression, "literalValue", String.class));
ctx.close();
}
/**
* Test case for the xml definition with the default implementation of {@link AmazonS3Operations}
*/
@Test
public void withDefaultOperationsImplementation() {
ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:s3-valid-outbound-cases.xml");
EventDrivenConsumer consumer = ctx.getBean("withDefaultServices", EventDrivenConsumer.class);
AmazonS3MessageHandler handler = getPropertyValue(consumer, "handler", AmazonS3MessageHandler.class);
assertEquals(DefaultAmazonS3Operations.class, getPropertyValue(handler, "operations").getClass());
Expression expression =
getPropertyValue(handler, "remoteDirectoryProcessor.expression", Expression.class);
assertNotNull(expression);
assertEquals(SpelExpression.class, expression.getClass());
assertEquals("headers['remoteDirectory']", getPropertyValue(expression, "expression", String.class));
assertEquals("TestBucket", getPropertyValue(handler, "bucket", String.class));
assertEquals("US-ASCII", getPropertyValue(handler, "charset", String.class));
assertEquals("dummy", getPropertyValue(handler, "credentials.accessKey", String.class));
assertEquals("dummy", getPropertyValue(handler, "credentials.secretKey", String.class));
assertEquals("dummy", getPropertyValue(handler, "operations.credentials.accessKey", String.class));
assertEquals("dummy", getPropertyValue(handler, "operations.credentials.secretKey", String.class));
assertEquals(5120, getPropertyValue(handler, "operations.multipartUploadThreshold", Long.class).longValue());
assertEquals(".write", getPropertyValue(handler, "operations.temporaryFileSuffix", String.class));
assertEquals(".write", getPropertyValue(handler, "fileNameGenerator.temporarySuffix", String.class));
assertEquals("headers['name']", getPropertyValue(handler, "fileNameGenerator.fileNameExpression", String.class));
assertEquals(ctx.getBean("executor"), getPropertyValue(handler, "operations.threadPoolExecutor"));
ctx.close();
}
/**
* Test case for the xml definition with a custom implementation of {@link FileNameGenerationStrategy}
*/
@Test
public void withCustomNameGenerator() {
ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("s3-valid-outbound-cases.xml");
EventDrivenConsumer consumer = ctx.getBean("withCustomNameGenerator", EventDrivenConsumer.class);
AmazonS3MessageHandler handler = getPropertyValue(consumer, "handler", AmazonS3MessageHandler.class);
assertEquals(DummyFileNameGenerator.class, getPropertyValue(handler, "fileNameGenerator").getClass());
ctx.close();
}
/**
* Test case for the xml definition with a custom AWS endpoint
*/
@Test
public void withCustomEndpoint() {
ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("s3-valid-outbound-cases.xml");
EventDrivenConsumer consumer = ctx.getBean("withCustomEndpoint", EventDrivenConsumer.class);
AmazonS3MessageHandler handler = getPropertyValue(consumer, "handler", AmazonS3MessageHandler.class);
assertEquals("http://s3-eu-west-1.amazonaws.com",
getPropertyValue(handler, "operations.client.endpoint", URI.class).toString());
ctx.close();
}
/**
* Multi part upload should have a size of 5120 and above, any value less than 5120 will
* thrown an exception
*/
@Test(expected = BeanCreationException.class)
public void withMultiUploadLessThan5120() {
new ClassPathXmlApplicationContext("s3-multiupload-lessthan-5120.xml").close();
}
/**
* Test with both the custom file generator and expression attribute set.
*/
@Test(expected = BeanDefinitionStoreException.class)
public void withBothFileGeneratorAndExpression() {
new ClassPathXmlApplicationContext("s3-both-customfilegenerator-and-expression.xml").close();
}
/**
* When custom implementation of {@link AmazonS3Operations} is provided, the attributes
* multipart-upload-threshold, temporary-directory, temporary-suffix and thread-pool-executor
* are not allowed
*/
@Test(expected = BeanDefinitionStoreException.class)
public void withCustomOperationsAndDisallowedAttributes() {
new ClassPathXmlApplicationContext("s3-custom-operations-with-disallowed-attributes.xml").close();
}
/**
* Tests the outbound channel adapter definition with a valid combination of attributes along with
* request handler chain.
*/
@Test
public void withHandlerAdviceChain() {
ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("s3-valid-outbound-cases.xml");
EventDrivenConsumer consumer = ctx.getBean("withHandlerChain", EventDrivenConsumer.class);
MessageHandler handler = getPropertyValue(consumer, "handler", MessageHandler.class);
handler.handleMessage(new GenericMessage<String>("String content: Test AWS advice chain"));
assertEquals(1, adviceCalled);
ctx.close();
}
/**
* Test case for the xml definition with the Channel Attributes.
*/
@Test
public void withChannelAttributeImplementation() {
ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:s3-valid-outbound-cases.xml");
EventDrivenConsumer consumer = ctx.getBean("withChannelAdapterAttributes", EventDrivenConsumer.class);
AmazonS3MessageHandler handler = getPropertyValue(consumer, "handler", AmazonS3MessageHandler.class);
assertEquals(DefaultAmazonS3Operations.class, getPropertyValue(handler, "operations").getClass());
assertEquals("input", getPropertyValue(consumer, "inputChannel.beanName", String.class));
assertEquals("US-ASCII", getPropertyValue(handler, "charset", String.class));
assertEquals(100, getPropertyValue(consumer, "phase"));
assertFalse(getPropertyValue(consumer, "autoStartup", Boolean.class));
ctx.close();
}
public static class DummyFileNameGenerator implements FileNameGenerationStrategy {
@Override
public String generateFileName(Message<?> message) {
return null;
}
}
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
adviceCalled++;
return callback.execute();
}
}
}