FTP supplier/source
AWS S3 apps migration s3-supplier/s3-source s3-consumer/s3-sink Fix merge conflicts Addressing PR review comments Rebasing Add ConditionalOnMissingClass to Spel converter config Test cleanup * Fix `AwsS3SourceTests` to properly mock before starting endpoint
This commit is contained in:
Soby Chacko
38
common/aws-s3-common/pom.xml
Normal file
38
common/aws-s3-common/pom.xml
Normal file
@@ -0,0 +1,38 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<artifactId>aws-s3-common</artifactId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
<name>aws-s3-common</name>
|
||||
<description>aws-s3 consumer</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>spring-functions-parent</artifactId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
<relativePath>../../spring-functions-parent</relativePath>
|
||||
</parent>
|
||||
|
||||
<properties>
|
||||
<spring-integration-aws.version>2.3.0.RELEASE</spring-integration-aws.version>
|
||||
<spring-cloud-aws.version>2.2.2.RELEASE</spring-cloud-aws.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-aws</artifactId>
|
||||
<version>${spring-integration-aws.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-aws</artifactId>
|
||||
<version>${spring-cloud-aws.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-file</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,45 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.common.aws.s3;
|
||||
|
||||
import com.amazonaws.auth.AWSCredentialsProvider;
|
||||
import com.amazonaws.services.s3.AmazonS3;
|
||||
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
|
||||
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.cloud.aws.context.annotation.ConditionalOnMissingAmazonClient;
|
||||
import org.springframework.cloud.aws.core.region.RegionProvider;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
@Configuration
|
||||
@ConditionalOnMissingAmazonClient(AmazonS3.class)
|
||||
public class AmazonS3Configuration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public AmazonS3 amazonS3(AWSCredentialsProvider awsCredentialsProvider, RegionProvider regionProvider) {
|
||||
return AmazonS3ClientBuilder.standard()
|
||||
.withCredentials(awsCredentialsProvider)
|
||||
.withRegion(regionProvider.getRegion().getName())
|
||||
.build();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
||||
org.springframework.cloud.fn.common.aws.s3.AmazonS3Configuration
|
||||
@@ -21,7 +21,7 @@ import java.beans.Introspector;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
|
||||
import org.springframework.boot.context.properties.ConfigurationPropertiesBinding;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
@@ -38,7 +38,7 @@ import org.springframework.integration.json.JsonPropertyAccessor;
|
||||
|
||||
@Configuration
|
||||
@AutoConfigureAfter(name = "org.springframework.cloud.stream.config.SpelExpressionConverterConfiguration")
|
||||
@ConditionalOnMissingBean(type = "org.springframework.cloud.stream.config.SpelExpressionConverterConfiguration")
|
||||
@ConditionalOnMissingClass("org.springframework.cloud.stream.config.SpelExpressionConverterConfiguration")
|
||||
public class SpelExpressionConverterConfiguration {
|
||||
@Bean
|
||||
public static SpelPropertyAccessorRegistrar spelPropertyAccessorRegistrar() {
|
||||
|
||||
26
consumer/s3-consumer/README.adoc
Normal file
26
consumer/s3-consumer/README.adoc
Normal file
@@ -0,0 +1,26 @@
|
||||
# AWS S3 Consumer
|
||||
|
||||
A consumer that allows you to upload files to AWS S3.
|
||||
The consumer uses the AWS S3 support from Spring Integration and Spring Cloud AWS.
|
||||
|
||||
## Beans for injection
|
||||
|
||||
You can import `AwsS3ConsumerConfiguration` in the application and then inject the following bean.
|
||||
|
||||
`Consumer<Message<?>> s3Consumer`
|
||||
|
||||
You can use `s3Consumer` as a qualifier when injecting.
|
||||
|
||||
## Configuration Options
|
||||
|
||||
All configuration properties are prefixed with `s3.consumer`.
|
||||
|
||||
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerProperties.java[AwsS3ConsumerProperties].
|
||||
|
||||
## Examples
|
||||
|
||||
See this link:src/test/java/org/springframework/cloud/fn/consumer/s3[test suite] for the various ways, this consumer is used.
|
||||
|
||||
## Other usage
|
||||
|
||||
See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/s3-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream based S3 Sink application.
|
||||
52
consumer/s3-consumer/pom.xml
Normal file
52
consumer/s3-consumer/pom.xml
Normal file
@@ -0,0 +1,52 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<artifactId>s3-consumer</artifactId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
<name>s3-consumer</name>
|
||||
<description>s3 consumer</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>spring-functions-parent</artifactId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
<relativePath>../../spring-functions-parent</relativePath>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>aws-s3-common</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>file-common</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>config-common</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-validation</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1,71 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.consumer.s3;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import com.amazonaws.services.s3.AmazonS3;
|
||||
import com.amazonaws.services.s3.transfer.internal.S3ProgressListener;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cloud.aws.core.env.ResourceIdResolver;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.aws.outbound.S3MessageHandler;
|
||||
import org.springframework.integration.expression.ValueExpression;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
|
||||
@Configuration
|
||||
@EnableConfigurationProperties(AwsS3ConsumerProperties.class)
|
||||
public class AwsS3ConsumerConfiguration {
|
||||
|
||||
@Autowired(required = false)
|
||||
private S3MessageHandler.UploadMetadataProvider uploadMetadataProvider;
|
||||
|
||||
@Autowired(required = false)
|
||||
private S3ProgressListener s3ProgressListener;
|
||||
|
||||
@Bean
|
||||
public Consumer<Message<?>> s3Consumer() {
|
||||
return amazonS3MessageHandler(null, null, null)::handleMessage;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public MessageHandler amazonS3MessageHandler(AmazonS3 amazonS3, ResourceIdResolver resourceIdResolver,
|
||||
AwsS3ConsumerProperties s3ConsumerProperties) {
|
||||
S3MessageHandler s3MessageHandler;
|
||||
if (s3ConsumerProperties.getBucket() != null) {
|
||||
s3MessageHandler = new S3MessageHandler(amazonS3, s3ConsumerProperties.getBucket());
|
||||
}
|
||||
else {
|
||||
s3MessageHandler = new S3MessageHandler(amazonS3, s3ConsumerProperties.getBucketExpression());
|
||||
}
|
||||
s3MessageHandler.setResourceIdResolver(resourceIdResolver);
|
||||
s3MessageHandler.setKeyExpression(s3ConsumerProperties.getKeyExpression());
|
||||
if (s3ConsumerProperties.getAcl() != null) {
|
||||
s3MessageHandler.setObjectAclExpression(new ValueExpression<>(s3ConsumerProperties.getAcl()));
|
||||
}
|
||||
else {
|
||||
s3MessageHandler.setObjectAclExpression(s3ConsumerProperties.getAclExpression());
|
||||
}
|
||||
s3MessageHandler.setUploadMetadataProvider(this.uploadMetadataProvider);
|
||||
s3MessageHandler.setProgressListener(this.s3ProgressListener);
|
||||
return s3MessageHandler;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.consumer.s3;
|
||||
|
||||
import javax.validation.constraints.AssertTrue;
|
||||
|
||||
import com.amazonaws.services.s3.model.CannedAccessControlList;
|
||||
import org.hibernate.validator.constraints.Length;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.expression.Expression;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
@ConfigurationProperties("s3.consumer")
|
||||
@Validated
|
||||
public class AwsS3ConsumerProperties {
|
||||
|
||||
/**
|
||||
* AWS bucket for target file(s) to store.
|
||||
*/
|
||||
private String bucket;
|
||||
|
||||
/**
|
||||
* Expression to evaluate AWS bucket name.
|
||||
*/
|
||||
private Expression bucketExpression;
|
||||
|
||||
/**
|
||||
* Expression to evaluate S3 Object key.
|
||||
*/
|
||||
private Expression keyExpression;
|
||||
|
||||
/**
|
||||
* S3 Object access control list.
|
||||
*/
|
||||
private CannedAccessControlList acl;
|
||||
|
||||
/**
|
||||
* Expression to evaluate S3 Object access control list.
|
||||
*/
|
||||
private Expression aclExpression;
|
||||
|
||||
@Length(min = 3)
|
||||
public String getBucket() {
|
||||
return this.bucket;
|
||||
}
|
||||
|
||||
public void setBucket(String bucket) {
|
||||
this.bucket = bucket;
|
||||
}
|
||||
|
||||
public Expression getBucketExpression() {
|
||||
return this.bucketExpression;
|
||||
}
|
||||
|
||||
public void setBucketExpression(Expression bucketExpression) {
|
||||
this.bucketExpression = bucketExpression;
|
||||
}
|
||||
|
||||
public Expression getKeyExpression() {
|
||||
return this.keyExpression;
|
||||
}
|
||||
|
||||
public void setKeyExpression(Expression keyExpression) {
|
||||
this.keyExpression = keyExpression;
|
||||
}
|
||||
|
||||
public CannedAccessControlList getAcl() {
|
||||
return this.acl;
|
||||
}
|
||||
|
||||
public void setAcl(CannedAccessControlList acl) {
|
||||
this.acl = acl;
|
||||
}
|
||||
|
||||
public Expression getAclExpression() {
|
||||
return this.aclExpression;
|
||||
}
|
||||
|
||||
public void setAclExpression(Expression aclExpression) {
|
||||
this.aclExpression = aclExpression;
|
||||
}
|
||||
|
||||
@AssertTrue(message = "Exactly one of 'bucket' or 'bucketExpression' must be set")
|
||||
public boolean isMutuallyExclusiveBucketAndBucketExpression() {
|
||||
return (this.bucket != null && this.bucketExpression == null) ||
|
||||
(this.bucket == null && this.bucketExpression != null);
|
||||
}
|
||||
|
||||
@AssertTrue(message = "Only one of 'acl' or 'aclExpression' must be set")
|
||||
public boolean isMutuallyExclusiveAclAndAclExpression() {
|
||||
return this.acl == null || this.aclExpression == null;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.consumer.s3;
|
||||
|
||||
import java.io.InputStream;
|
||||
import java.nio.file.Path;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import com.amazonaws.event.ProgressEvent;
|
||||
import com.amazonaws.event.ProgressEventType;
|
||||
import com.amazonaws.services.s3.AmazonS3;
|
||||
import com.amazonaws.services.s3.AmazonS3Client;
|
||||
import com.amazonaws.services.s3.model.PutObjectRequest;
|
||||
import com.amazonaws.services.s3.model.PutObjectResult;
|
||||
import com.amazonaws.services.s3.model.SetObjectAclRequest;
|
||||
import com.amazonaws.services.s3.transfer.PersistableTransfer;
|
||||
import com.amazonaws.services.s3.transfer.internal.S3ProgressListener;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.integration.aws.outbound.S3MessageHandler;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.BDDMockito.willAnswer;
|
||||
import static org.mockito.Mockito.spy;
|
||||
|
||||
@DirtiesContext
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
|
||||
properties = {
|
||||
"cloud.aws.stack.auto=false",
|
||||
"cloud.aws.credentials.accessKey=" + AbstractAwsS3ConsumerMockTests.AWS_ACCESS_KEY,
|
||||
"cloud.aws.credentials.secretKey=" + AbstractAwsS3ConsumerMockTests.AWS_SECRET_KEY,
|
||||
"cloud.aws.region.static=" + AbstractAwsS3ConsumerMockTests.AWS_REGION,
|
||||
"s3.consumer.bucket=" + AbstractAwsS3ConsumerMockTests.S3_BUCKET })
|
||||
public abstract class AbstractAwsS3ConsumerMockTests {
|
||||
|
||||
protected static final String AWS_ACCESS_KEY = "test.accessKey";
|
||||
|
||||
protected static final String AWS_SECRET_KEY = "test.secretKey";
|
||||
|
||||
protected static final String AWS_REGION = "us-gov-west-1";
|
||||
|
||||
protected static final String S3_BUCKET = "S3_BUCKET";
|
||||
|
||||
@TempDir
|
||||
protected static Path temporaryRemoteFolder;
|
||||
|
||||
@Autowired
|
||||
private AmazonS3Client amazonS3;
|
||||
|
||||
@Autowired
|
||||
protected S3MessageHandler s3MessageHandler;
|
||||
|
||||
@Autowired
|
||||
protected CountDownLatch aclLatch;
|
||||
|
||||
@Autowired
|
||||
protected CountDownLatch transferCompletedLatch;
|
||||
|
||||
@Autowired
|
||||
protected Consumer<Message<?>> s3Consumer;
|
||||
|
||||
@BeforeEach
|
||||
public void setupTest() {
|
||||
Object transferManager = TestUtils.getPropertyValue(this.s3MessageHandler, "transferManager");
|
||||
|
||||
AmazonS3 amazonS3 = spy(this.amazonS3);
|
||||
|
||||
willAnswer(invocation -> new PutObjectResult()).given(amazonS3)
|
||||
.putObject(any(PutObjectRequest.class));
|
||||
|
||||
willAnswer(invocation -> {
|
||||
aclLatch.countDown();
|
||||
return null;
|
||||
}).given(amazonS3)
|
||||
.setObjectAcl(any(SetObjectAclRequest.class));
|
||||
|
||||
new DirectFieldAccessor(transferManager).setPropertyValue("s3", amazonS3);
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
public static class S3ConsumerApplication {
|
||||
|
||||
@Bean
|
||||
public CountDownLatch aclLatch() {
|
||||
return new CountDownLatch(1);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public CountDownLatch transferCompletedLatch() {
|
||||
return new CountDownLatch(1);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public S3ProgressListener s3ProgressListener() {
|
||||
return new S3ProgressListener() {
|
||||
|
||||
@Override
|
||||
public void onPersistableTransfer(PersistableTransfer persistableTransfer) {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void progressChanged(ProgressEvent progressEvent) {
|
||||
if (ProgressEventType.TRANSFER_COMPLETED_EVENT.equals(progressEvent.getEventType())) {
|
||||
transferCompletedLatch().countDown();
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public S3MessageHandler.UploadMetadataProvider uploadMetadataProvider() {
|
||||
return (metadata, message) -> {
|
||||
if (message.getPayload() instanceof InputStream) {
|
||||
metadata.setContentLength(1);
|
||||
metadata.setContentType(MediaType.APPLICATION_JSON_VALUE);
|
||||
metadata.setContentDisposition("test.json");
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.consumer.s3;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import com.amazonaws.event.ProgressEventType;
|
||||
import com.amazonaws.event.ProgressListener;
|
||||
import com.amazonaws.services.s3.AmazonS3;
|
||||
import com.amazonaws.services.s3.model.CannedAccessControlList;
|
||||
import com.amazonaws.services.s3.model.ObjectMetadata;
|
||||
import com.amazonaws.services.s3.model.PutObjectRequest;
|
||||
import com.amazonaws.services.s3.model.SetObjectAclRequest;
|
||||
import com.amazonaws.services.s3.transfer.internal.S3ProgressPublisher;
|
||||
import com.amazonaws.util.Md5Utils;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.test.context.TestPropertySource;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.Mockito.atLeastOnce;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
@TestPropertySource(properties = "s3.consumer.acl=PublicReadWrite")
|
||||
public class AmazonS3UploadFileTests extends AbstractAwsS3ConsumerMockTests {
|
||||
|
||||
@Test
|
||||
public void test() throws Exception {
|
||||
AmazonS3 amazonS3Client = TestUtils.getPropertyValue(this.s3MessageHandler, "transferManager.s3",
|
||||
AmazonS3.class);
|
||||
|
||||
File file = new File(this.temporaryRemoteFolder.toFile(), "foo.mp3");
|
||||
file.createNewFile();
|
||||
Message<?> message = MessageBuilder.withPayload(file)
|
||||
.build();
|
||||
|
||||
this.s3Consumer.accept(message);
|
||||
|
||||
ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor =
|
||||
ArgumentCaptor.forClass(PutObjectRequest.class);
|
||||
verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture());
|
||||
|
||||
PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue();
|
||||
assertThat(putObjectRequest.getBucketName()).isEqualTo(S3_BUCKET);
|
||||
assertThat(putObjectRequest.getKey()).isEqualTo("foo.mp3");
|
||||
assertThat(putObjectRequest.getFile()).isNotNull();
|
||||
assertThat(putObjectRequest.getInputStream()).isNull();
|
||||
|
||||
ObjectMetadata metadata = putObjectRequest.getMetadata();
|
||||
assertThat(metadata.getContentMD5()).isEqualTo(Md5Utils.md5AsBase64(file));
|
||||
assertThat(metadata.getContentLength()).isEqualTo(0L);
|
||||
assertThat(metadata.getContentType()).isEqualTo("audio/mpeg");
|
||||
|
||||
ProgressListener listener = putObjectRequest.getGeneralProgressListener();
|
||||
S3ProgressPublisher.publishProgress(listener, ProgressEventType.TRANSFER_COMPLETED_EVENT);
|
||||
|
||||
assertThat(this.transferCompletedLatch.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(this.aclLatch.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
|
||||
ArgumentCaptor<SetObjectAclRequest> setObjectAclRequestArgumentCaptor =
|
||||
ArgumentCaptor.forClass(SetObjectAclRequest.class);
|
||||
verify(amazonS3Client).setObjectAcl(setObjectAclRequestArgumentCaptor.capture());
|
||||
|
||||
SetObjectAclRequest setObjectAclRequest = setObjectAclRequestArgumentCaptor.getValue();
|
||||
|
||||
assertThat(setObjectAclRequest.getBucketName()).isEqualTo(S3_BUCKET);
|
||||
assertThat(setObjectAclRequest.getKey()).isEqualTo("foo.mp3");
|
||||
assertThat(setObjectAclRequest.getAcl()).isNull();
|
||||
assertThat(setObjectAclRequest.getCannedAcl()).isEqualTo(CannedAccessControlList.PublicReadWrite);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.consumer.s3;
|
||||
|
||||
import java.io.InputStream;
|
||||
|
||||
import com.amazonaws.services.s3.AmazonS3;
|
||||
import com.amazonaws.services.s3.model.ObjectMetadata;
|
||||
import com.amazonaws.services.s3.model.PutObjectRequest;
|
||||
import com.amazonaws.util.Md5Utils;
|
||||
import com.amazonaws.util.StringInputStream;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.test.context.TestPropertySource;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.Mockito.atLeastOnce;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
@TestPropertySource(properties = "s3.consumer.key-expression=headers.key")
|
||||
public class AmazonS3UploadInputStreamTests extends AbstractAwsS3ConsumerMockTests {
|
||||
|
||||
@Test
|
||||
public void test() throws Exception {
|
||||
AmazonS3 amazonS3Client = TestUtils.getPropertyValue(this.s3MessageHandler, "transferManager.s3",
|
||||
AmazonS3.class);
|
||||
|
||||
InputStream payload = new StringInputStream("a");
|
||||
Message<?> message = MessageBuilder.withPayload(payload)
|
||||
.setHeader("key", "myInputStream")
|
||||
.build();
|
||||
|
||||
this.s3Consumer.accept(message);
|
||||
|
||||
ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor =
|
||||
ArgumentCaptor.forClass(PutObjectRequest.class);
|
||||
verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture());
|
||||
|
||||
PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue();
|
||||
assertThat(putObjectRequest.getBucketName()).isEqualTo(S3_BUCKET);
|
||||
assertThat(putObjectRequest.getKey()).isEqualTo("myInputStream");
|
||||
assertThat(putObjectRequest.getFile()).isNull();
|
||||
assertThat(putObjectRequest.getInputStream()).isNotNull();
|
||||
|
||||
ObjectMetadata metadata = putObjectRequest.getMetadata();
|
||||
assertThat(metadata.getContentMD5()).isEqualTo(Md5Utils.md5AsBase64(payload));
|
||||
assertThat(metadata.getContentLength()).isEqualTo(1L);
|
||||
assertThat(metadata.getContentType()).isEqualTo(MediaType.APPLICATION_JSON_VALUE);
|
||||
assertThat(metadata.getContentDisposition()).isEqualTo("test.json");
|
||||
}
|
||||
}
|
||||
3
pom.xml
3
pom.xml
@@ -41,6 +41,7 @@
|
||||
</properties>
|
||||
|
||||
<modules>
|
||||
<module>common/aws-s3-common</module>
|
||||
<module>common/config-common</module>
|
||||
<module>common/file-common</module>
|
||||
<module>common/ftp-common</module>
|
||||
@@ -65,6 +66,7 @@
|
||||
<module>consumer/sftp-consumer</module>
|
||||
<module>consumer/tcp-consumer</module>
|
||||
<module>consumer/websocket-consumer</module>
|
||||
<module>consumer/s3-consumer</module>
|
||||
|
||||
<module>function/filter-function</module>
|
||||
<module>function/header-enricher-function</module>
|
||||
@@ -86,6 +88,7 @@
|
||||
<module>supplier/time-supplier</module>
|
||||
<module>supplier/rabbit-supplier</module>
|
||||
<module>supplier/websocket-supplier</module>
|
||||
<module>supplier/s3-supplier</module>
|
||||
|
||||
<module>spring-functions-parent</module>
|
||||
</modules>
|
||||
|
||||
@@ -140,5 +140,4 @@ public class FtpSupplierConfiguration {
|
||||
return () -> Flux.from(ftpReadingFlow());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
35
supplier/s3-supplier/README.adoc
Normal file
35
supplier/s3-supplier/README.adoc
Normal file
@@ -0,0 +1,35 @@
|
||||
# AWS S3 Supplier
|
||||
|
||||
This module provides an S3 supplier that can be reused and composed in other applications.
|
||||
The `Supplier` uses the AWS S3 support provided by Spring Integration and spring cloud aws project
|
||||
`S3Supplier` is implemented as a `java.util.function.Supplier`.
|
||||
This supplier gives you a reactive stream of files from the provided directory as the supplier has a signature of `Supplier<Flux<Message<?>>>`.
|
||||
Users have to subscribe to this `Flux` and receive the data.
|
||||
|
||||
## Beans for injection
|
||||
|
||||
You can import the `AwsS3SupplierConfiguration` in the application and then inject the following bean.
|
||||
|
||||
`s3Supplier`
|
||||
|
||||
You need to inject this as `Supplier<Flux<Message<?>>>`.
|
||||
|
||||
You can use `s3Supplier` as a qualifier when injecting.
|
||||
|
||||
Once injected, you can use the `get` method of the `Supplier` to invoke it and then subscribe to the returned `Flux`.
|
||||
|
||||
## Configuration Options
|
||||
|
||||
All configuration properties are prefixed with `s3.supplier`.
|
||||
There are also properties that need to be used with the prefix `file.consumer`.
|
||||
|
||||
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java[AwsS3upplierProperties].
|
||||
See link:../../common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileConsumerProperties.java[this] also.
|
||||
|
||||
## Tests
|
||||
|
||||
See this link:src/test/java/org/springframework/cloud/fn/supplier/s3[test suite] for the various ways, this supplier is used.
|
||||
|
||||
## Other usage
|
||||
|
||||
See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/s3-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes an AWS S3 Source.
|
||||
53
supplier/s3-supplier/pom.xml
Normal file
53
supplier/s3-supplier/pom.xml
Normal file
@@ -0,0 +1,53 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<artifactId>s3-supplier</artifactId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
<name>s3-supplier</name>
|
||||
<description>s3 supplier</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>spring-functions-parent</artifactId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
<relativePath>../../spring-functions-parent</relativePath>
|
||||
</parent>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>aws-s3-common</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>file-common</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-validation</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.projectreactor</groupId>
|
||||
<artifactId>reactor-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,116 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.supplier.s3;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.Arrays;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import com.amazonaws.services.s3.AmazonS3;
|
||||
import com.amazonaws.services.s3.model.S3ObjectSummary;
|
||||
import org.reactivestreams.Publisher;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cloud.aws.core.env.ResourceIdResolver;
|
||||
import org.springframework.cloud.fn.common.file.FileConsumerProperties;
|
||||
import org.springframework.cloud.fn.common.file.FileUtils;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.aws.inbound.S3InboundFileSynchronizer;
|
||||
import org.springframework.integration.aws.inbound.S3InboundFileSynchronizingMessageSource;
|
||||
import org.springframework.integration.aws.support.S3SessionFactory;
|
||||
import org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter;
|
||||
import org.springframework.integration.aws.support.filters.S3RegexPatternFileListFilter;
|
||||
import org.springframework.integration.aws.support.filters.S3SimplePatternFileListFilter;
|
||||
import org.springframework.integration.core.MessageSource;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.integration.file.filters.ChainFileListFilter;
|
||||
import org.springframework.integration.file.filters.FileListFilter;
|
||||
import org.springframework.integration.metadata.SimpleMetadataStore;
|
||||
import org.springframework.integration.util.IntegrationReactiveUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({AwsS3SupplierProperties.class, FileConsumerProperties.class})
|
||||
public class AwsS3SupplierConfiguration {
|
||||
|
||||
private final AwsS3SupplierProperties awsS3SupplierProperties;
|
||||
private final FileConsumerProperties fileConsumerProperties;
|
||||
private final AmazonS3 amazonS3;
|
||||
private final ResourceIdResolver resourceIdResolver;
|
||||
|
||||
public AwsS3SupplierConfiguration(AwsS3SupplierProperties awsS3SupplierProperties,
|
||||
FileConsumerProperties fileConsumerProperties,
|
||||
AmazonS3 amazonS3,
|
||||
ResourceIdResolver resourceIdResolver) {
|
||||
this.awsS3SupplierProperties = awsS3SupplierProperties;
|
||||
this.fileConsumerProperties = fileConsumerProperties;
|
||||
this.amazonS3 = amazonS3;
|
||||
this.resourceIdResolver = resourceIdResolver;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public S3InboundFileSynchronizer s3InboundFileSynchronizer() {
|
||||
S3SessionFactory s3SessionFactory = new S3SessionFactory(this.amazonS3, this.resourceIdResolver);
|
||||
S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(s3SessionFactory);
|
||||
synchronizer.setDeleteRemoteFiles(this.awsS3SupplierProperties.isDeleteRemoteFiles());
|
||||
synchronizer.setPreserveTimestamp(this.awsS3SupplierProperties.isPreserveTimestamp());
|
||||
String remoteDir = this.awsS3SupplierProperties.getRemoteDir();
|
||||
synchronizer.setRemoteDirectory(remoteDir);
|
||||
synchronizer.setRemoteFileSeparator(this.awsS3SupplierProperties.getRemoteFileSeparator());
|
||||
synchronizer.setTemporaryFileSuffix(this.awsS3SupplierProperties.getTmpFileSuffix());
|
||||
|
||||
FileListFilter<S3ObjectSummary> filter = null;
|
||||
if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) {
|
||||
filter = new S3SimplePatternFileListFilter(this.awsS3SupplierProperties.getFilenamePattern());
|
||||
}
|
||||
else if (this.awsS3SupplierProperties.getFilenameRegex() != null) {
|
||||
filter = new S3RegexPatternFileListFilter(this.awsS3SupplierProperties.getFilenameRegex());
|
||||
}
|
||||
if (filter != null) {
|
||||
synchronizer.setFilter(new ChainFileListFilter<>(Arrays.asList(filter,
|
||||
new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3-metadata-"))));
|
||||
}
|
||||
return synchronizer;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public MessageSource<File> s3MessageSource() {
|
||||
S3InboundFileSynchronizingMessageSource s3MessageSource =
|
||||
new S3InboundFileSynchronizingMessageSource(s3InboundFileSynchronizer());
|
||||
s3MessageSource.setLocalDirectory(this.awsS3SupplierProperties.getLocalDir());
|
||||
s3MessageSource.setAutoCreateLocalDirectory(this.awsS3SupplierProperties.isAutoCreateLocalDir());
|
||||
return s3MessageSource;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Publisher<Message<Object>> s3SupplierFlow() {
|
||||
return FileUtils.enhanceFlowForReadingMode(IntegrationFlows
|
||||
.from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource())), fileConsumerProperties)
|
||||
.toReactivePublisher();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Supplier<Flux<Message<?>>> s3Supplier() {
|
||||
return () -> Flux.from(s3SupplierFlow());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.supplier.s3;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import javax.validation.constraints.AssertTrue;
|
||||
import javax.validation.constraints.NotBlank;
|
||||
import javax.validation.constraints.NotNull;
|
||||
|
||||
import org.hibernate.validator.constraints.Length;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.validation.annotation.Validated;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
*/
|
||||
@ConfigurationProperties("s3.supplier")
|
||||
@Validated
|
||||
public class AwsS3SupplierProperties {
|
||||
|
||||
/**
|
||||
* AWS S3 bucket resource.
|
||||
*/
|
||||
private String remoteDir = "bucket";
|
||||
|
||||
/**
|
||||
* Temporary file suffix.
|
||||
*/
|
||||
private String tmpFileSuffix = ".tmp";
|
||||
|
||||
/**
|
||||
* Remote File separator.
|
||||
*/
|
||||
private String remoteFileSeparator = "/";
|
||||
|
||||
/**
|
||||
* Delete or not remote files after processing.
|
||||
*/
|
||||
private boolean deleteRemoteFiles = false;
|
||||
|
||||
/**
|
||||
* The local directory to store files.
|
||||
*/
|
||||
private File localDir = new File(System.getProperty("java.io.tmpdir"), "s3-supplier");
|
||||
|
||||
/**
|
||||
* Create or not the local directory.
|
||||
*/
|
||||
private boolean autoCreateLocalDir = true;
|
||||
|
||||
/**
|
||||
* The pattern to filter remote files.
|
||||
*/
|
||||
private String filenamePattern;
|
||||
|
||||
/**
|
||||
* The regexp to filter remote files.
|
||||
*/
|
||||
private Pattern filenameRegex;
|
||||
|
||||
/**
|
||||
* To transfer or not the timestamp of the remote file to the local one.
|
||||
*/
|
||||
private boolean preserveTimestamp = true;
|
||||
|
||||
@Length(min = 3)
|
||||
public String getRemoteDir() {
|
||||
return this.remoteDir;
|
||||
}
|
||||
|
||||
public final void setRemoteDir(String remoteDir) {
|
||||
this.remoteDir = remoteDir;
|
||||
}
|
||||
|
||||
@NotBlank
|
||||
public String getTmpFileSuffix() {
|
||||
return tmpFileSuffix;
|
||||
}
|
||||
|
||||
public void setTmpFileSuffix(String tmpFileSuffix) {
|
||||
this.tmpFileSuffix = tmpFileSuffix;
|
||||
}
|
||||
|
||||
@NotBlank
|
||||
public String getRemoteFileSeparator() {
|
||||
return remoteFileSeparator;
|
||||
}
|
||||
|
||||
public void setRemoteFileSeparator(String remoteFileSeparator) {
|
||||
this.remoteFileSeparator = remoteFileSeparator;
|
||||
}
|
||||
|
||||
public boolean isAutoCreateLocalDir() {
|
||||
return autoCreateLocalDir;
|
||||
}
|
||||
|
||||
public void setAutoCreateLocalDir(boolean autoCreateLocalDir) {
|
||||
this.autoCreateLocalDir = autoCreateLocalDir;
|
||||
}
|
||||
|
||||
public boolean isDeleteRemoteFiles() {
|
||||
return deleteRemoteFiles;
|
||||
}
|
||||
|
||||
public void setDeleteRemoteFiles(boolean deleteRemoteFiles) {
|
||||
this.deleteRemoteFiles = deleteRemoteFiles;
|
||||
}
|
||||
|
||||
@NotNull
|
||||
public File getLocalDir() {
|
||||
return localDir;
|
||||
}
|
||||
|
||||
public final void setLocalDir(File localDir) {
|
||||
this.localDir = localDir;
|
||||
}
|
||||
|
||||
public String getFilenamePattern() {
|
||||
return filenamePattern;
|
||||
}
|
||||
|
||||
public void setFilenamePattern(String filenamePattern) {
|
||||
this.filenamePattern = filenamePattern;
|
||||
}
|
||||
|
||||
public Pattern getFilenameRegex() {
|
||||
return filenameRegex;
|
||||
}
|
||||
|
||||
public void setFilenameRegex(Pattern filenameRegex) {
|
||||
this.filenameRegex = filenameRegex;
|
||||
}
|
||||
|
||||
public boolean isPreserveTimestamp() {
|
||||
return preserveTimestamp;
|
||||
}
|
||||
|
||||
public void setPreserveTimestamp(boolean preserveTimestamp) {
|
||||
this.preserveTimestamp = preserveTimestamp;
|
||||
}
|
||||
|
||||
@AssertTrue(message = "filenamePattern and filenameRegex are mutually exclusive")
|
||||
public boolean isExclusivePatterns() {
|
||||
return !(this.filenamePattern != null && this.filenameRegex != null);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.supplier.s3;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.FileInputStream;
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Path;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Calendar;
|
||||
import java.util.List;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import com.amazonaws.services.s3.AmazonS3;
|
||||
import com.amazonaws.services.s3.model.ListObjectsRequest;
|
||||
import com.amazonaws.services.s3.model.ObjectListing;
|
||||
import com.amazonaws.services.s3.model.Region;
|
||||
import com.amazonaws.services.s3.model.S3Object;
|
||||
import com.amazonaws.services.s3.model.S3ObjectSummary;
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.integration.dsl.StandardIntegrationFlow;
|
||||
import org.springframework.integration.test.context.SpringIntegrationTest;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.util.FileCopyUtils;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.BDDMockito.willAnswer;
|
||||
import static org.mockito.BDDMockito.willReturn;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
|
||||
properties = {
|
||||
"cloud.aws.stack.auto=false",
|
||||
"cloud.aws.credentials.accessKey=" + AbstractAwsS3SupplierMockTests.AWS_ACCESS_KEY,
|
||||
"cloud.aws.credentials.secretKey=" + AbstractAwsS3SupplierMockTests.AWS_SECRET_KEY,
|
||||
"cloud.aws.region.static=" + AbstractAwsS3SupplierMockTests.AWS_REGION,
|
||||
"s3.supplier.remoteDir=" + AbstractAwsS3SupplierMockTests.S3_BUCKET})
|
||||
@DirtiesContext
|
||||
@SpringIntegrationTest(noAutoStartup = "*")
|
||||
public abstract class AbstractAwsS3SupplierMockTests {
|
||||
|
||||
@TempDir
|
||||
protected static Path temporaryRemoteFolder;
|
||||
|
||||
protected static final String AWS_ACCESS_KEY = "test.accessKey";
|
||||
|
||||
protected static final String AWS_SECRET_KEY = "test.secretKey";
|
||||
|
||||
protected static final String AWS_REGION = "us-gov-west-1";
|
||||
|
||||
protected static final String S3_BUCKET = "S3_BUCKET";
|
||||
|
||||
protected static List<S3Object> S3_OBJECTS;
|
||||
|
||||
@Autowired
|
||||
Supplier<Flux<Message<?>>> s3Supplier;
|
||||
|
||||
@Autowired
|
||||
protected AwsS3SupplierProperties awsS3SupplierProperties;
|
||||
|
||||
@Autowired
|
||||
StandardIntegrationFlow standardIntegrationFlow;
|
||||
|
||||
@BeforeAll
|
||||
public static void setup() throws IOException {
|
||||
final String remote = temporaryRemoteFolder.toAbsolutePath() + "/remote";
|
||||
File f = new File(remote);
|
||||
f.mkdirs();
|
||||
File aFile = new File(f, "1.test");
|
||||
FileCopyUtils.copy("Hello".getBytes(), aFile);
|
||||
File bFile = new File(f, "2.test");
|
||||
FileCopyUtils.copy("Bye".getBytes(), bFile);
|
||||
File otherFile = new File(f, "otherFile");
|
||||
FileCopyUtils.copy("Other\nOther2".getBytes(), otherFile);
|
||||
|
||||
S3_OBJECTS = new ArrayList<>();
|
||||
|
||||
for (File file : f.listFiles()) {
|
||||
S3Object s3Object = new S3Object();
|
||||
s3Object.setBucketName(S3_BUCKET);
|
||||
s3Object.setKey(file.getName());
|
||||
s3Object.setObjectContent(new FileInputStream(file));
|
||||
S3_OBJECTS.add(s3Object);
|
||||
}
|
||||
|
||||
final String local = temporaryRemoteFolder.toAbsolutePath() + "/local";
|
||||
File f1 = new File(local);
|
||||
f1.mkdirs();
|
||||
System.setProperty("s3.supplier.localDir", f1.getAbsolutePath());
|
||||
}
|
||||
|
||||
@AfterAll
|
||||
public static void tearDown() {
|
||||
System.clearProperty("s3.supplier.localDir");
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
public static class S3SupplierApplication {
|
||||
|
||||
@Bean
|
||||
@Primary
|
||||
public AmazonS3 amazonS3Mock() {
|
||||
AmazonS3 amazonS3 = mock(AmazonS3.class);
|
||||
willReturn(Region.US_West).given(amazonS3).getRegion();
|
||||
|
||||
Calendar calendar = Calendar.getInstance();
|
||||
calendar.add(Calendar.DATE, 1);
|
||||
|
||||
willAnswer(invocation -> {
|
||||
ObjectListing objectListing = new ObjectListing();
|
||||
List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
|
||||
for (S3Object s3Object : S3_OBJECTS) {
|
||||
S3ObjectSummary s3ObjectSummary = new S3ObjectSummary();
|
||||
s3ObjectSummary.setBucketName(S3_BUCKET);
|
||||
s3ObjectSummary.setKey(s3Object.getKey());
|
||||
s3ObjectSummary.setLastModified(calendar.getTime());
|
||||
objectSummaries.add(s3ObjectSummary);
|
||||
}
|
||||
return objectListing;
|
||||
}).given(amazonS3).listObjects(any(ListObjectsRequest.class));
|
||||
|
||||
for (final S3Object s3Object : S3_OBJECTS) {
|
||||
willAnswer(invocation -> s3Object).given(amazonS3).getObject(S3_BUCKET, s3Object.getKey());
|
||||
}
|
||||
return amazonS3;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.supplier.s3;
|
||||
|
||||
import java.io.File;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.test.context.TestPropertySource;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@TestPropertySource(properties = {"file.consumer.mode=ref",
|
||||
"s3.supplier.filenameRegex=.*\\\\.test$"})
|
||||
public class AmazonS3FilesTransferredTests extends AbstractAwsS3SupplierMockTests {
|
||||
|
||||
@Test
|
||||
public void test() {
|
||||
final Flux<Message<?>> messageFlux = s3Supplier.get();
|
||||
StepVerifier stepVerifier =
|
||||
StepVerifier.create(messageFlux)
|
||||
.assertNext((message) -> {
|
||||
assertThat(new File(message.getPayload().toString().replaceAll("\"", "")))
|
||||
.isEqualTo(new File(this.awsS3SupplierProperties.getLocalDir() + File.separator + "1.test"));
|
||||
}
|
||||
)
|
||||
.assertNext((message) -> {
|
||||
assertThat(new File(message.getPayload().toString().replaceAll("\"", "")))
|
||||
.isEqualTo(new File(this.awsS3SupplierProperties.getLocalDir() + File.separator + "2.test"));
|
||||
})
|
||||
.thenCancel()
|
||||
.verifyLater();
|
||||
standardIntegrationFlow.start();
|
||||
stepVerifier.verify();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
/*
|
||||
* Copyright 2016-2020 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
|
||||
*
|
||||
* https://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.cloud.fn.supplier.s3;
|
||||
|
||||
import java.io.File;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import org.springframework.integration.file.FileHeaders;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.test.context.TestPropertySource;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@TestPropertySource(properties = {
|
||||
"file.consumer.mode=lines",
|
||||
"s3.supplier.filenamePattern=otherFile",
|
||||
"file.consumer.with-markers=false" })
|
||||
public class AmazonS3LinesTransferredTests extends AbstractAwsS3SupplierMockTests {
|
||||
|
||||
@Test
|
||||
public void test() throws Exception {
|
||||
final Flux<Message<?>> messageFlux = s3Supplier.get();
|
||||
StepVerifier stepVerifier =
|
||||
StepVerifier.create(messageFlux)
|
||||
.assertNext((message) -> {
|
||||
assertThat(message.getPayload().toString()).isEqualTo("Other");
|
||||
assertThat(message.getHeaders().containsKey(FileHeaders.ORIGINAL_FILE)).isTrue();
|
||||
assertThat(message.getHeaders().containsValue(
|
||||
new File(this.awsS3SupplierProperties.getLocalDir(), "otherFile"))).isTrue();
|
||||
}
|
||||
)
|
||||
.assertNext((message) -> {
|
||||
assertThat(message.getPayload().toString()).isEqualTo("Other2");
|
||||
})
|
||||
.thenCancel()
|
||||
.verifyLater();
|
||||
standardIntegrationFlow.start();
|
||||
stepVerifier.verify();
|
||||
|
||||
assertThat(this.awsS3SupplierProperties.getLocalDir().list().length).isEqualTo(1);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user