SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade to AWS SDK v2

Browse Source 
* Upgrade to Spring Integration AWS `3.0.1`
* Upgrade to Spring Cloud AWS `3.0.1`
* Use `spring-cloud-aws-dependencies` BOM to manage Spring Cloud AWS deps, as well as AWS SDK
* Remove `AmazonS3Properties` in favor of similar properties in the Spring Cloud AWS auto-configuration
* Remove `CompatibleStorageAmazonS3Configuration` in favor of its code migration to the single `AmazonS3Configuration`
* Fix `supplier`, `consumer`, `source` and `sink` for S3 to use API and programing model from the mentioned upgrades
* Add extra `dataflow-configuration-metadata` entries for the mentioned properties from the Spring Cloud AWS auto-configuration
* Regenerate README for `s3-source` and `s3-sink`
* Fix `metadata-store-common` to use the latest Spring Integration AWS for DynamoDB store impl

* * Fix `AmazonS3ConfigurationTests` for missed `AwsAutoConfiguration`
in an `ApplicationContextRunner` setup

* * Remove unused import in the `MetadataStoreAutoConfigurationTests`

* * Remove unused import in the `AbstractAwsS3ConsumerMockTests`
This commit is contained in:
Artem Bilan
2023-07-19 09:36:09 -04:00
committed by GitHub
parent 511900fed7
commit 1df9fd9ca5
21 changed files with 342 additions and 471 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
common/aws-s3-common/pom.xml
View File

@@ -11,12 +11,7 @@
<artifactId>aws-s3-common</artifactId>
<name>aws-s3-common</name>
<description>aws-s3 consumer</description>
<properties>
<spring-integration-aws.version>2.5.2</spring-integration-aws.version>
<spring-cloud-aws.version>2.4.2</spring-cloud-aws.version>
</properties>
<description>aws-s3 common</description>
<dependencies>
<dependency>
@@ -26,8 +21,12 @@
</dependency>
<dependency>
<groupId>io.awspring.cloud</groupId>
<artifactId>spring-cloud-starter-aws</artifactId>
<version>${spring-cloud-aws.version}</version>
<artifactId>spring-cloud-aws-starter-s3</artifactId>
</dependency>
<dependency>
<groupId>software.amazon.awssdk.crt</groupId>
<artifactId>aws-crt</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>

38
common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Configuration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -16,39 +16,35 @@
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 io.awspring.cloud.autoconfigure.context.ContextCredentialsAutoConfiguration;
import io.awspring.cloud.autoconfigure.context.ContextRegionProviderAutoConfiguration;
import io.awspring.cloud.context.annotation.ConditionalOnMissingAmazonClient;
import io.awspring.cloud.context.config.annotation.ContextDefaultConfigurationRegistrar;
import io.awspring.cloud.core.region.RegionProvider;
import java.net.URI;
import io.awspring.cloud.autoconfigure.s3.S3AutoConfiguration;
import io.awspring.cloud.autoconfigure.s3.S3CrtAsyncClientAutoConfiguration;
import io.awspring.cloud.autoconfigure.s3.properties.S3Properties;
import software.amazon.awssdk.services.s3.S3Client;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.integration.aws.support.S3SessionFactory;
/**
* @author Artem Bilan
*/
@AutoConfiguration
@ConditionalOnMissingAmazonClient(AmazonS3.class)
@Import({
ContextCredentialsAutoConfiguration.class,
ContextDefaultConfigurationRegistrar.class,
ContextRegionProviderAutoConfiguration.class
})
@AutoConfigureAfter({S3AutoConfiguration.class, S3CrtAsyncClientAutoConfiguration.class})
public class AmazonS3Configuration {
@Bean
@ConditionalOnMissingBean
public AmazonS3 amazonS3(AWSCredentialsProvider awsCredentialsProvider, RegionProvider regionProvider) {
return AmazonS3ClientBuilder.standard()
.withCredentials(awsCredentialsProvider)
.withRegion(regionProvider.getRegion().getName())
.build();
public S3SessionFactory s3SessionFactory(S3Client amazonS3, S3Properties s3Properties) {
S3SessionFactory s3SessionFactory = new S3SessionFactory(amazonS3);
URI endpoint = s3Properties.getEndpoint();
if (endpoint != null) {
s3SessionFactory.setEndpoint(String.join(":", endpoint.getHost(), String.valueOf(endpoint.getPort())));
}
return s3SessionFactory;
}
}

53
common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Properties.java
View File

@@ -1,53 +0,0 @@
/*
* Copyright 2020-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 org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author Timo Salm
* @author David Turanski
*/
@ConfigurationProperties("s3.common")
public class AmazonS3Properties {
/**
* Optional endpoint url to connect to s3 compatible storage.
*/
private String endpointUrl;
/**
* Use path style access.
*/
private boolean pathStyleAccess;
public boolean isPathStyleAccess() {
return pathStyleAccess;
}
public void setPathStyleAccess(boolean pathStyleAccess) {
this.pathStyleAccess = pathStyleAccess;
}
public String getEndpointUrl() {
return this.endpointUrl;
}
public void setEndpointUrl(String endpointUrl) {
this.endpointUrl = endpointUrl;
}
}

83
common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/CompatibleStorageAmazonS3Configuration.java
View File

@@ -1,83 +0,0 @@
/*
* Copyright 2020-2022 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 java.net.URI;
import java.net.URISyntaxException;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import io.awspring.cloud.core.env.ResourceIdResolver;
import io.awspring.cloud.core.region.RegionProvider;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.aws.support.S3SessionFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;
/**
* @author Timo Salm
* @author David Turanski
* @author Artem Bilan
*/
@AutoConfiguration
@EnableConfigurationProperties(AmazonS3Properties.class)
@AutoConfigureBefore(AmazonS3Configuration.class)
public class CompatibleStorageAmazonS3Configuration {
@Bean
@ConditionalOnProperty("s3.common.endpoint-url")
public AmazonS3 compatibleStorageAmazonS3(AWSCredentialsProvider awsCredentialsProvider,
RegionProvider regionProvider,
AmazonS3Properties amazonS3Properties) {
final AmazonS3ClientBuilder builder = AmazonS3ClientBuilder.standard();
final EndpointConfiguration endpointConfiguration = new EndpointConfiguration(
amazonS3Properties.getEndpointUrl(), regionProvider.getRegion().getName());
builder.setEndpointConfiguration(endpointConfiguration);
return builder
.withCredentials(awsCredentialsProvider)
.withPathStyleAccessEnabled(amazonS3Properties.isPathStyleAccess())
.build();
}
@Bean
@ConditionalOnMissingBean
public S3SessionFactory s3SessionFactory(AmazonS3 amazonS3, @Nullable ResourceIdResolver resourceIdResolver,
AmazonS3Properties amazonS3Properties) {
S3SessionFactory s3SessionFactory = new S3SessionFactory(amazonS3, resourceIdResolver);
if (StringUtils.hasText(amazonS3Properties.getEndpointUrl())) {
URI uri;
try {
uri = new URI(amazonS3Properties.getEndpointUrl());
}
catch (URISyntaxException e) {
throw new IllegalArgumentException(amazonS3Properties.getEndpointUrl() + " is not a valid URI");
}
s3SessionFactory.setEndpoint(String.join(":", uri.getHost(), String.valueOf(uri.getPort())));
}
return s3SessionFactory;
}
}

1
common/aws-s3-common/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
View File

@@ -1,2 +1 @@
org.springframework.cloud.fn.common.aws.s3.CompatibleStorageAmazonS3Configuration
org.springframework.cloud.fn.common.aws.s3.AmazonS3Configuration

48
common/aws-s3-common/src/test/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3ConfigurationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2023 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.
@@ -16,19 +16,25 @@
package org.springframework.cloud.fn.common.aws.s3;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import io.awspring.cloud.core.region.RegionProvider;
import io.awspring.cloud.autoconfigure.core.AwsAutoConfiguration;
import io.awspring.cloud.autoconfigure.s3.S3AutoConfiguration;
import io.awspring.cloud.autoconfigure.s3.S3CrtAsyncClientAutoConfiguration;
import io.awspring.cloud.core.region.StaticRegionProvider;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.regions.providers.AwsRegionProvider;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.S3Utilities;
import software.amazon.awssdk.services.s3.model.GetUrlRequest;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.test.util.TestUtils;
/**
* @author Timo Salm
@@ -39,7 +45,9 @@ public class AmazonS3ConfigurationTests {
private final ApplicationContextRunner runner = new ApplicationContextRunner()
.withConfiguration(
AutoConfigurations.of(
CompatibleStorageAmazonS3Configuration.class,
AwsAutoConfiguration.class,
S3AutoConfiguration.class,
S3CrtAsyncClientAutoConfiguration.class,
AmazonS3Configuration.class))
.withUserConfiguration(TestConfiguration.class);
@@ -48,22 +56,26 @@ public class AmazonS3ConfigurationTests {
@Test
public void testAmazonS3Configuration() {
runner.withPropertyValues().run(context -> {
final AmazonS3Client amazonS3 = (AmazonS3Client) context.getBean(AmazonS3.class);
S3Client amazonS3 = context.getBean(S3Client.class);
Assertions.assertNotNull(amazonS3);
Assertions.assertEquals(TEST_REGION_NAME, amazonS3.getRegionName());
Assertions.assertTrue(amazonS3.getResourceUrl("b", "k")
.startsWith("https://s3.eu-central-1.amazonaws.com"));
S3Utilities utilities = amazonS3.utilities();
Assertions.assertEquals(TEST_REGION_NAME,
TestUtils.getPropertyValue(utilities, "region", Region.class).id());
Assertions.assertTrue(
utilities.getUrl(GetUrlRequest.builder().bucket("b").key("k").build()).toString()
.startsWith("https://s3.eu-central-1.amazonaws.com"));
});
}
@Test
public void testAmazonS3ConfigurationForS3CompatibleStorage() {
runner.withPropertyValues(
"s3.common.endpoint-url=http://localhost:8080"
"spring.cloud.aws.s3.endpoint=http://localhost:8080"
).run(context -> {
final AmazonS3Client amazonS3 = (AmazonS3Client) context.getBean(AmazonS3.class);
S3Client amazonS3 = context.getBean(S3Client.class);
Assertions.assertNotNull(amazonS3);
Assertions.assertTrue(amazonS3.getResourceUrl("b", "k")
S3Utilities utilities = amazonS3.utilities();
Assertions.assertTrue(utilities.getUrl(GetUrlRequest.builder().bucket("b").key("k").build()).toString()
.startsWith("http://localhost:8080"));
});
}
@@ -71,13 +83,13 @@ public class AmazonS3ConfigurationTests {
private static class TestConfiguration {
@Bean
RegionProvider regionProvider() {
AwsRegionProvider regionProvider() {
return new StaticRegionProvider(TEST_REGION_NAME);
}
@Bean
AWSCredentialsProvider awsCredentialsProvider() {
return new AWSStaticCredentialsProvider(new BasicAWSCredentials("accessKey", "secretKey"));
AwsCredentialsProvider awsCredentialsProvider() {
return StaticCredentialsProvider.create(AwsBasicCredentials.create("accessKey", "secretKey"));
}
}

18
common/metadata-store-common/pom.xml
View File

@@ -13,13 +13,6 @@
<name>metadata-store-common</name>
<description>metadata-store common</description>
<properties>
<aws-java-sdk.version>1.12.322</aws-java-sdk.version>
<spring-integration-aws.version>2.5.2</spring-integration-aws.version>
<spring-integration-hazelcast.version>6.1.1</spring-integration-hazelcast.version>
<curator.version>5.3.0</curator.version>
</properties>
<dependencies>
<!--Redis-->
@@ -46,6 +39,12 @@
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.awspring.cloud</groupId>
<artifactId>spring-cloud-aws-starter</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
@@ -103,9 +102,8 @@
<optional>true</optional>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-dynamodb</artifactId>
<version>${aws-java-sdk.version}</version>
<groupId>software.amazon.awssdk</groupId>
<artifactId>dynamodb</artifactId>
<optional>true</optional>
</dependency>

22
common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-2023 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.
@@ -16,14 +16,12 @@
package org.springframework.cloud.fn.common.metadata.store;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder;
import com.hazelcast.core.HazelcastInstance;
import io.awspring.cloud.core.region.RegionProvider;
import io.awspring.cloud.autoconfigure.core.AwsClientBuilderConfigurer;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryForever;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.AutoConfiguration;
@@ -49,6 +47,7 @@ import org.springframework.jdbc.core.JdbcTemplate;
* @author Artem Bilan
* @author David Turanski
* @author Corneil du Plessis
*
* @since 2.0.2
*/
@AutoConfiguration
@@ -142,20 +141,13 @@ public class MetadataStoreAutoConfiguration {
@Bean
@ConditionalOnMissingBean
public AmazonDynamoDBAsync dynamoDB(AWSCredentialsProvider awsCredentialsProvider,
RegionProvider regionProvider) {
return AmazonDynamoDBAsyncClientBuilder.standard()
.withCredentials(awsCredentialsProvider)
.withRegion(
regionProvider.getRegion()
.getName())
.build();
public DynamoDbAsyncClient dynamoDB(AwsClientBuilderConfigurer awsClientBuilderConfigurer) {
return awsClientBuilderConfigurer.configure(DynamoDbAsyncClient.builder()).build();
}
@Bean
@ConditionalOnMissingBean
public ConcurrentMetadataStore dynamoDbMetadataStore(AmazonDynamoDBAsync dynamoDB,
public ConcurrentMetadataStore dynamoDbMetadataStore(DynamoDbAsyncClient dynamoDB,
MetadataStoreProperties metadataStoreProperties) {
MetadataStoreProperties.DynamoDb dynamoDbProperties = metadataStoreProperties.getDynamoDb();

32
common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-2023 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.
@@ -19,16 +19,19 @@ package org.springframework.cloud.fn.common.metadata.store;
import java.beans.Introspector;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Predicate;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
import com.amazonaws.services.dynamodbv2.model.DescribeTableResult;
import io.awspring.cloud.core.region.RegionProvider;
import org.apache.curator.test.TestingServer;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.ArgumentMatchers;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.regions.providers.AwsRegionProvider;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest;
import software.amazon.awssdk.services.dynamodb.model.DescribeTableResponse;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -47,7 +50,6 @@ import org.springframework.integration.redis.metadata.RedisMetadataStore;
import org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStore;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.Mockito.mock;
@@ -122,23 +124,23 @@ public class MetadataStoreAutoConfigurationTests {
protected static class DynamoDbMockConfig {
@Bean
public static AmazonDynamoDBAsync dynamoDB() {
AmazonDynamoDBAsync dynamoDb = mock(AmazonDynamoDBAsync.class);
willReturn(new DescribeTableResult())
public static DynamoDbAsyncClient dynamoDB() {
DynamoDbAsyncClient dynamoDb = mock(DynamoDbAsyncClient.class);
willReturn(CompletableFuture.completedFuture(DescribeTableResponse.builder().build()))
.given(dynamoDb)
.describeTable(any(DescribeTableRequest.class));
.describeTable(ArgumentMatchers.<Consumer<DescribeTableRequest.Builder>>any());
return dynamoDb;
}
@Bean
public static AWSCredentialsProvider awsCredentialsProvider() {
return mock(AWSCredentialsProvider.class);
public static AwsCredentialsProvider awsCredentialsProvider() {
return mock(AwsCredentialsProvider.class);
}
@Bean
public static RegionProvider regionProvider() {
return mock(RegionProvider.class);
public static AwsRegionProvider regionProvider() {
return mock(AwsRegionProvider.class);
}
}

15
consumer/s3-consumer/README.adoc
View File

@@ -1,9 +1,9 @@
# AWS S3 Consumer
= 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
== Beans for injection
You can import `AwsS3ConsumerConfiguration` in the application and then inject the following bean.
@@ -11,18 +11,17 @@ You can import `AwsS3ConsumerConfiguration` in the application and then inject t
You can use `s3Consumer` as a qualifier when injecting.
## Configuration Options
== Configuration Options
All configuration properties are prefixed with `s3.consumer`.
There are also properties that need to be used with the prefix `s3.common`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerProperties.java[AwsS3ConsumerProperties] and
link:../../common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Properties.java[AmazonS3Properties].
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerProperties.java[AwsS3ConsumerProperties] and `io.awspring.cloud.autoconfigure.s3.properties.S3Properties` & `io.awspring.cloud.autoconfigure.s3.properties.S3CrtClientProperties` from Spring Cloud AWS auto-configuration.
## Examples
== 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
== 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.
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.

4
consumer/s3-consumer/pom.xml
View File

@@ -24,6 +24,10 @@
<artifactId>file-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>s3-transfer-manager</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>

90
consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2023 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.
@@ -16,56 +16,94 @@
package org.springframework.cloud.fn.consumer.s3;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.transfer.internal.S3ProgressListener;
import io.awspring.cloud.core.env.ResourceIdResolver;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.transfer.s3.S3TransferManager;
import software.amazon.awssdk.transfer.s3.progress.TransferListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.aws.outbound.S3MessageHandler;
import org.springframework.integration.aws.support.AwsHeaders;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
@Configuration
@Configuration(proxyBeanMethods = false)
@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;
public Consumer<Message<?>> s3Consumer(IntegrationFlow s3ConsumerFlow) {
return s3ConsumerFlow.getInputChannel()::send;
}
@Bean
public MessageHandler amazonS3MessageHandler(AmazonS3 amazonS3, ResourceIdResolver resourceIdResolver,
AwsS3ConsumerProperties s3ConsumerProperties) {
S3MessageHandler s3MessageHandler;
public IntegrationFlow s3ConsumerFlow(@Nullable TransferListener transferListener,
MessageHandler amazonS3MessageHandler) {
return flow -> flow
.enrichHeaders(headers -> headers.header(AwsHeaders.TRANSFER_LISTENER, transferListener))
.handle(amazonS3MessageHandler);
}
@Bean
public MessageHandler amazonS3MessageHandler(S3TransferManager s3TransferManager,
AwsS3ConsumerProperties s3ConsumerProperties,
BeanFactory beanFactory,
@Nullable BiConsumer<PutObjectRequest.Builder, Message<?>> uploadMetadataProvider) {
Expression bucketExpression = s3ConsumerProperties.getBucketExpression();
if (s3ConsumerProperties.getBucket() != null) {
s3MessageHandler = new S3MessageHandler(amazonS3, s3ConsumerProperties.getBucket());
bucketExpression = new ValueExpression<>(s3ConsumerProperties.getBucket());
}
else {
s3MessageHandler = new S3MessageHandler(amazonS3, s3ConsumerProperties.getBucketExpression());
}
s3MessageHandler.setResourceIdResolver(resourceIdResolver);
S3MessageHandler s3MessageHandler = new S3MessageHandler(s3TransferManager, bucketExpression);
s3MessageHandler.setKeyExpression(s3ConsumerProperties.getKeyExpression());
Expression aclExpression;
if (s3ConsumerProperties.getAcl() != null) {
s3MessageHandler.setObjectAclExpression(new ValueExpression<>(s3ConsumerProperties.getAcl()));
aclExpression = new ValueExpression<>(s3ConsumerProperties.getAcl());
}
else {
s3MessageHandler.setObjectAclExpression(s3ConsumerProperties.getAclExpression());
aclExpression = s3ConsumerProperties.getAclExpression();
}
BiConsumer<PutObjectRequest.Builder, Message<?>> metadataProviderToUse = uploadMetadataProvider;
if (aclExpression != null) {
EvaluationContext evaluationContext = IntegrationContextUtils.getEvaluationContext(beanFactory);
metadataProviderToUse =
(builder, message) -> {
Object aclValue = aclExpression.getValue(evaluationContext, message);
Assert.notNull(aclValue,
() -> String.format("The expression '%s' for message '%s' returned null",
aclExpression, message));
builder.acl(aclValue.toString());
if (uploadMetadataProvider != null) {
uploadMetadataProvider.accept(builder, message);
}
};
}
if (metadataProviderToUse != null) {
s3MessageHandler.setUploadMetadataProvider(metadataProviderToUse);
}
s3MessageHandler.setUploadMetadataProvider(this.uploadMetadataProvider);
s3MessageHandler.setProgressListener(this.s3ProgressListener);
return s3MessageHandler;
}
}

10
consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2023 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.
@@ -16,9 +16,9 @@
package org.springframework.cloud.fn.consumer.s3;
import com.amazonaws.services.s3.model.CannedAccessControlList;
import jakarta.validation.constraints.AssertTrue;
import org.hibernate.validator.constraints.Length;
import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.expression.Expression;
@@ -49,7 +49,7 @@ public class AwsS3ConsumerProperties {
/**
* S3 Object access control list.
*/
private CannedAccessControlList acl;
private ObjectCannedACL acl;
/**
* Expression to evaluate S3 Object access control list.
@@ -81,11 +81,11 @@ public class AwsS3ConsumerProperties {
this.keyExpression = keyExpression;
}
public CannedAccessControlList getAcl() {
public ObjectCannedACL getAcl() {
return this.acl;
}
public void setAcl(CannedAccessControlList acl) {
public void setAcl(ObjectCannedACL acl) {
this.acl = acl;
}

88
consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AbstractAwsS3ConsumerMockTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -18,20 +18,19 @@ package org.springframework.cloud.fn.consumer.s3;
import java.io.InputStream;
import java.nio.file.Path;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.function.BiConsumer;
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 software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
import software.amazon.awssdk.transfer.s3.S3TransferManager;
import software.amazon.awssdk.transfer.s3.progress.TransferListener;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -39,24 +38,21 @@ 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.BDDMockito.willReturn;
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.common.endpointUrl=foo",
"s3.consumer.bucket=" + AbstractAwsS3ConsumerMockTests.S3_BUCKET })
"spring.cloud.aws.credentials.accessKey=" + AbstractAwsS3ConsumerMockTests.AWS_ACCESS_KEY,
"spring.cloud.aws.credentials.secretKey=" + AbstractAwsS3ConsumerMockTests.AWS_SECRET_KEY,
"spring.cloud.aws.region.static=" + AbstractAwsS3ConsumerMockTests.AWS_REGION,
"spring.cloud.aws.s3.endpoint=s3://foo",
"s3.consumer.bucket=" + AbstractAwsS3ConsumerMockTests.S3_BUCKET})
public abstract class AbstractAwsS3ConsumerMockTests {
protected static final String AWS_ACCESS_KEY = "test.accessKey";
@@ -71,13 +67,10 @@ public abstract class AbstractAwsS3ConsumerMockTests {
protected static Path temporaryRemoteFolder;
@Autowired
private AmazonS3Client amazonS3;
private S3AsyncClient amazonS3;
@Autowired
protected S3MessageHandler s3MessageHandler;
@Autowired
protected CountDownLatch aclLatch;
protected S3TransferManager s3TransferManager;
@Autowired
protected CountDownLatch transferCompletedLatch;
@@ -87,60 +80,43 @@ public abstract class AbstractAwsS3ConsumerMockTests {
@BeforeEach
public void setupTest() {
Object transferManager = TestUtils.getPropertyValue(this.s3MessageHandler, "transferManager");
S3AsyncClient amazonS3 = spy(this.amazonS3);
AmazonS3 amazonS3 = spy(this.amazonS3);
willReturn(CompletableFuture.completedFuture(PutObjectResponse.builder().build()))
.given(amazonS3)
.putObject(any(PutObjectRequest.class), any(AsyncRequestBody.class));
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);
new DirectFieldAccessor(this.s3TransferManager).setPropertyValue("s3AsyncClient", amazonS3);
}
@SpringBootApplication
public static class S3ConsumerTestApplication {
@Bean
public CountDownLatch aclLatch() {
return new CountDownLatch(1);
}
@Bean
public CountDownLatch transferCompletedLatch() {
return new CountDownLatch(1);
}
@Bean
public S3ProgressListener s3ProgressListener() {
return new S3ProgressListener() {
public TransferListener transferListener() {
return new TransferListener() {
@Override
public void onPersistableTransfer(PersistableTransfer persistableTransfer) {
public void transferComplete(Context.TransferComplete context) {
transferCompletedLatch().countDown();
}
@Override
public void progressChanged(ProgressEvent progressEvent) {
if (ProgressEventType.TRANSFER_COMPLETED_EVENT.equals(progressEvent.getEventType())) {
transferCompletedLatch().countDown();
}
}
};
}
@Bean
public S3MessageHandler.UploadMetadataProvider uploadMetadataProvider() {
return (metadata, message) -> {
public BiConsumer<PutObjectRequest.Builder, Message<?>> uploadMetadataProvider() {
return (builder, message) -> {
if (message.getPayload() instanceof InputStream) {
metadata.setContentLength(1);
metadata.setContentType(MediaType.APPLICATION_JSON_VALUE);
metadata.setContentDisposition("test.json");
builder.contentLength(1L)
.contentType(MediaType.APPLICATION_JSON_VALUE)
.contentDisposition("test.json");
}
};
}

65
consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadFileTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -19,17 +19,14 @@ 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 reactor.test.StepVerifier;
import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.utils.Md5Utils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
@@ -40,15 +37,15 @@ 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")
@TestPropertySource(properties = "s3.consumer.acl=PUBLIC_READ_WRITE")
public class AmazonS3UploadFileTests extends AbstractAwsS3ConsumerMockTests {
@Test
public void test() throws Exception {
AmazonS3 amazonS3Client = TestUtils.getPropertyValue(this.s3MessageHandler, "transferManager.s3",
AmazonS3.class);
S3AsyncClient amazonS3Client =
TestUtils.getPropertyValue(this.s3TransferManager, "s3AsyncClient", S3AsyncClient.class);
File file = new File(this.temporaryRemoteFolder.toFile(), "foo.mp3");
File file = new File(temporaryRemoteFolder.toFile(), "foo.mp3");
file.createNewFile();
Message<?> message = MessageBuilder.withPayload(file)
.build();
@@ -57,34 +54,26 @@ public class AmazonS3UploadFileTests extends AbstractAwsS3ConsumerMockTests {
ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor =
ArgumentCaptor.forClass(PutObjectRequest.class);
verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture());
ArgumentCaptor<AsyncRequestBody> asyncRequestBodyArgumentCaptor =
ArgumentCaptor.forClass(AsyncRequestBody.class);
verify(amazonS3Client, atLeastOnce())
.putObject(putObjectRequestArgumentCaptor.capture(), asyncRequestBodyArgumentCaptor.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();
assertThat(putObjectRequest.bucket()).isEqualTo(S3_BUCKET);
assertThat(putObjectRequest.key()).isEqualTo("foo.mp3");
assertThat(putObjectRequest.contentMD5()).isEqualTo(Md5Utils.md5AsBase64(file));
assertThat(putObjectRequest.contentLength()).isEqualTo(0L);
assertThat(putObjectRequest.contentType()).isEqualTo("audio/mpeg");
assertThat(putObjectRequest.acl()).isEqualTo(ObjectCannedACL.PUBLIC_READ_WRITE);
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);
AsyncRequestBody asyncRequestBody = asyncRequestBodyArgumentCaptor.getValue();
StepVerifier.create(asyncRequestBody)
.assertNext(buffer -> assertThat(buffer.array()).isEmpty())
.expectComplete()
.verify();
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);
}
}

44
consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadInputStreamTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -18,13 +18,14 @@ 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 reactor.test.StepVerifier;
import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.utils.Md5Utils;
import software.amazon.awssdk.utils.StringInputStream;
import org.springframework.http.MediaType;
import org.springframework.integration.test.util.TestUtils;
@@ -41,8 +42,8 @@ public class AmazonS3UploadInputStreamTests extends AbstractAwsS3ConsumerMockTes
@Test
public void test() throws Exception {
AmazonS3 amazonS3Client = TestUtils.getPropertyValue(this.s3MessageHandler, "transferManager.s3",
AmazonS3.class);
S3AsyncClient amazonS3Client =
TestUtils.getPropertyValue(this.s3TransferManager, "s3AsyncClient", S3AsyncClient.class);
InputStream payload = new StringInputStream("a");
Message<?> message = MessageBuilder.withPayload(payload)
@@ -53,18 +54,25 @@ public class AmazonS3UploadInputStreamTests extends AbstractAwsS3ConsumerMockTes
ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor =
ArgumentCaptor.forClass(PutObjectRequest.class);
verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture());
ArgumentCaptor<AsyncRequestBody> asyncRequestBodyArgumentCaptor =
ArgumentCaptor.forClass(AsyncRequestBody.class);
verify(amazonS3Client, atLeastOnce())
.putObject(putObjectRequestArgumentCaptor.capture(), asyncRequestBodyArgumentCaptor.capture());
PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue();
assertThat(putObjectRequest.getBucketName()).isEqualTo(S3_BUCKET);
assertThat(putObjectRequest.getKey()).isEqualTo("myInputStream");
assertThat(putObjectRequest.getFile()).isNull();
assertThat(putObjectRequest.getInputStream()).isNotNull();
assertThat(putObjectRequest.bucket()).isEqualTo(S3_BUCKET);
assertThat(putObjectRequest.key()).isEqualTo("myInputStream");
assertThat(putObjectRequest.contentMD5()).isEqualTo(Md5Utils.md5AsBase64(payload));
assertThat(putObjectRequest.contentLength()).isEqualTo(1L);
assertThat(putObjectRequest.contentType()).isEqualTo(MediaType.APPLICATION_JSON_VALUE);
assertThat(putObjectRequest.contentDisposition()).isEqualTo("test.json");
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");
AsyncRequestBody asyncRequestBody = asyncRequestBodyArgumentCaptor.getValue();
StepVerifier.create(asyncRequestBody.map(buffer -> new String(buffer.array())))
.expectNext("a")
.expectComplete()
.verify();
}
}

12
supplier/s3-supplier/README.adoc
View File

@@ -1,4 +1,4 @@
# AWS S3 Supplier
= 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
@@ -6,7 +6,7 @@ The `Supplier` uses the AWS S3 support provided by Spring Integration and spring
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
== Beans for injection
You can import the `AwsS3SupplierConfiguration` in the application and then inject the following bean.
@@ -18,21 +18,21 @@ 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
== Configuration Options
All configuration properties are prefixed with `s3.supplier`.
There are also properties that need to be used with the prefix `s3.common` and `file.consumer`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java[AwsS3SupplierProperties],
link:../../common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileConsumerProperties.java[FileConsumerProperties], and
link:../../common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Properties.java[AmazonS3Properties].
`io.awspring.cloud.autoconfigure.s3.properties.S3Properties` from Spring Cloud AWS auto-configuration..
A `ComponentCustomizer<S3InboundFileSynchronizingMessageSource>` bean can be added in the target project to provide any custom options for the `S3InboundFileSynchronizingMessageSource` configuration used by the `s3Supplier`.
## Tests
== 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
== 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.

65
supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java
View File

@@ -22,11 +22,11 @@ import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.ListObjectsRequest;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
import software.amazon.awssdk.services.s3.model.S3Object;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -57,7 +57,7 @@ import org.springframework.util.StringUtils;
* @author David Turanski
*/
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties({ AwsS3SupplierProperties.class, FileConsumerProperties.class })
@EnableConfigurationProperties({AwsS3SupplierProperties.class, FileConsumerProperties.class})
public class AwsS3SupplierConfiguration {
protected static final String METADATA_STORE_PREFIX = "s3-metadata-";
@@ -66,20 +66,17 @@ public class AwsS3SupplierConfiguration {
protected final FileConsumerProperties fileConsumerProperties;
protected final AmazonS3 amazonS3;
protected final S3SessionFactory s3SessionFactory;
protected final ConcurrentMetadataStore metadataStore;
public AwsS3SupplierConfiguration(AwsS3SupplierProperties awsS3SupplierProperties,
FileConsumerProperties fileConsumerProperties,
AmazonS3 amazonS3,
S3SessionFactory s3SessionFactory, ConcurrentMetadataStore metadataStore) {
S3SessionFactory s3SessionFactory,
ConcurrentMetadataStore metadataStore) {
this.awsS3SupplierProperties = awsS3SupplierProperties;
this.fileConsumerProperties = fileConsumerProperties;
this.amazonS3 = amazonS3;
this.s3SessionFactory = s3SessionFactory;
this.metadataStore = metadataStore;
}
@@ -94,8 +91,8 @@ public class AwsS3SupplierConfiguration {
}
@Bean
public ChainFileListFilter<S3ObjectSummary> filter(ConcurrentMetadataStore metadataStore) {
ChainFileListFilter<S3ObjectSummary> chainFilter = new ChainFileListFilter<>();
public ChainFileListFilter<S3Object> filter(ConcurrentMetadataStore metadataStore) {
ChainFileListFilter<S3Object> chainFilter = new ChainFileListFilter<>();
if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) {
chainFilter.addFilter(
new S3SimplePatternFileListFilter(this.awsS3SupplierProperties.getFilenamePattern()));
@@ -111,12 +108,10 @@ public class AwsS3SupplierConfiguration {
SynchronizingConfiguration(AwsS3SupplierProperties awsS3SupplierProperties,
FileConsumerProperties fileConsumerProperties,
AmazonS3 amazonS3,
S3SessionFactory s3SessionFactory,
ConcurrentMetadataStore concurrentMetadataStore) {
super(awsS3SupplierProperties, fileConsumerProperties, amazonS3, s3SessionFactory,
concurrentMetadataStore);
super(awsS3SupplierProperties, fileConsumerProperties, s3SessionFactory, concurrentMetadataStore);
}
@Bean
@@ -130,7 +125,7 @@ public class AwsS3SupplierConfiguration {
}
@Bean
public S3InboundFileSynchronizer s3InboundFileSynchronizer(ChainFileListFilter<S3ObjectSummary> filter) {
public S3InboundFileSynchronizer s3InboundFileSynchronizer(ChainFileListFilter<S3Object> filter) {
S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(s3SessionFactory);
synchronizer.setDeleteRemoteFiles(this.awsS3SupplierProperties.isDeleteRemoteFiles());
@@ -169,10 +164,10 @@ public class AwsS3SupplierConfiguration {
ListOnlyConfiguration(AwsS3SupplierProperties awsS3SupplierProperties,
FileConsumerProperties fileConsumerProperties,
AmazonS3 amazonS3, S3SessionFactory s3SessionFactory, ConcurrentMetadataStore metadataStore) {
S3SessionFactory s3SessionFactory,
ConcurrentMetadataStore metadataStore) {
super(awsS3SupplierProperties, fileConsumerProperties, amazonS3, s3SessionFactory,
metadataStore);
super(awsS3SupplierProperties, fileConsumerProperties, s3SessionFactory, metadataStore);
}
@Bean
@@ -186,19 +181,19 @@ public class AwsS3SupplierConfiguration {
}
@Bean
Predicate<S3ObjectSummary> listOnlyFilter() {
Predicate<S3ObjectSummary> predicate = s -> true;
Predicate<S3Object> listOnlyFilter(AwsS3SupplierProperties awsS3SupplierProperties) {
Predicate<S3Object> predicate = s -> true;
if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) {
Pattern pattern = Pattern.compile(this.awsS3SupplierProperties.getFilenamePattern());
predicate = (S3ObjectSummary summary) -> pattern.matcher(summary.getKey()).matches();
predicate = (S3Object summary) -> pattern.matcher(summary.key()).matches();
}
else if (this.awsS3SupplierProperties.getFilenameRegex() != null) {
predicate = (S3ObjectSummary summary) -> this.awsS3SupplierProperties.getFilenameRegex()
.matcher(summary.getKey()).matches();
predicate = (S3Object summary) -> this.awsS3SupplierProperties.getFilenameRegex()
.matcher(summary.key()).matches();
}
predicate = predicate.and((S3ObjectSummary summary) -> {
final String key = METADATA_STORE_PREFIX + summary.getBucketName() + "-" + summary.getKey();
final String lastModified = String.valueOf(summary.getLastModified().getTime());
predicate = predicate.and((S3Object summary) -> {
final String key = METADATA_STORE_PREFIX + awsS3SupplierProperties.getRemoteDir() + "-" + summary.key();
final String lastModified = String.valueOf(summary.lastModified().toEpochMilli());
final String storedLastModified = this.metadataStore.get(key);
boolean result = !lastModified.equals(storedLastModified);
if (result) {
@@ -211,16 +206,18 @@ public class AwsS3SupplierConfiguration {
}
@Bean
ReactiveMessageSourceProducer s3ListingMessageProducer(AmazonS3 amazonS3,
AwsS3SupplierProperties awsS3SupplierProperties, Predicate<S3ObjectSummary> filter) {
ReactiveMessageSourceProducer s3ListingMessageProducer(S3Client amazonS3,
AwsS3SupplierProperties awsS3SupplierProperties, Predicate<S3Object> filter) {
ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
listObjectsRequest.setBucketName(awsS3SupplierProperties.getRemoteDir());
return new ReactiveMessageSourceProducer(
(MessageSource<List<S3ObjectSummary>>) () -> {
List<S3ObjectSummary> summaryList = amazonS3.listObjects(listObjectsRequest)
.getObjectSummaries().stream()
.filter(filter).collect(Collectors.toList());
(MessageSource<List<S3Object>>) () -> {
List<S3Object> summaryList =
amazonS3.listObjects(ListObjectsRequest.builder()
.bucket(awsS3SupplierProperties.getRemoteDir())
.build())
.contents()
.stream()
.filter(filter).collect(Collectors.toList());
return summaryList.isEmpty() ? null : new GenericMessage<>(summaryList);
});
}

85
supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java
View File

@@ -19,22 +19,25 @@ package org.springframework.cloud.fn.supplier.s3;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.time.Instant;
import java.time.Period;
import java.util.HashMap;
import java.util.Map;
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 software.amazon.awssdk.core.ResponseInputStream;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
import software.amazon.awssdk.services.s3.model.ListObjectsResponse;
import software.amazon.awssdk.services.s3.model.S3Object;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -49,16 +52,14 @@ 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.common.endpointUrl=foo",
"spring.cloud.aws.credentials.accessKey=" + AbstractAwsS3SupplierMockTests.AWS_ACCESS_KEY,
"spring.cloud.aws.credentials.secretKey=" + AbstractAwsS3SupplierMockTests.AWS_SECRET_KEY,
"spring.cloud.aws.region.static=" + AbstractAwsS3SupplierMockTests.AWS_REGION,
"spring.cloud.aws.s3.endpoint=s3://foo",
"s3.supplier.remoteDir=" + AbstractAwsS3SupplierMockTests.S3_BUCKET
})
@DirtiesContext
@@ -76,7 +77,7 @@ public abstract class AbstractAwsS3SupplierMockTests {
protected static final String S3_BUCKET = "S3_BUCKET";
protected static List<S3Object> S3_OBJECTS;
protected static Map<S3Object, InputStream> S3_OBJECTS;
@Autowired
Supplier<Flux<Message<?>>> s3Supplier;
@@ -99,14 +100,18 @@ public abstract class AbstractAwsS3SupplierMockTests {
File otherFile = new File(f, "otherFile");
FileCopyUtils.copy("Other\nOther2".getBytes(), otherFile);
S3_OBJECTS = new ArrayList<>();
S3_OBJECTS = new HashMap<>();
Instant instant = Instant.now().plus(Period.ofDays(1));
for (File file : f.listFiles()) {
S3Object s3Object = new S3Object();
s3Object.setBucketName(S3_BUCKET);
s3Object.setKey("subdir/" + file.getName());
s3Object.setObjectContent(new FileInputStream(file));
S3_OBJECTS.add(s3Object);
S3Object s3Object =
S3Object.builder()
.key("subdir/" + file.getName())
.lastModified(instant)
.build();
S3_OBJECTS.put(s3Object, new FileInputStream(file));
}
final String local = temporaryRemoteFolder.toAbsolutePath() + "/local";
@@ -118,8 +123,7 @@ public abstract class AbstractAwsS3SupplierMockTests {
@AfterAll
public static void tearDown() {
System.clearProperty("s3.supplier.localDir");
S3_OBJECTS.stream()
.map(S3Object::getObjectContent)
S3_OBJECTS.values()
.forEach(stream -> {
try {
stream.close();
@@ -135,27 +139,24 @@ public abstract class AbstractAwsS3SupplierMockTests {
@Bean
@Primary
public AmazonS3 amazonS3Mock() {
AmazonS3 amazonS3 = mock(AmazonS3.class);
willReturn(Region.US_West).given(amazonS3).getRegion();
public S3Client amazonS3Mock() {
S3Client amazonS3 = mock(S3Client.class);
Calendar calendar = Calendar.getInstance();
calendar.add(Calendar.DATE, 1);
ListObjectsResponse listObjectsResponse =
ListObjectsResponse.builder().contents(S3_OBJECTS.keySet()).isTruncated(false).build();
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);
}
willAnswer(invocation -> listObjectsResponse)
.given(amazonS3)
.listObjects(any(ListObjectsRequest.class));
willAnswer(invocation -> objectListing).given(amazonS3).listObjects(any(ListObjectsRequest.class));
for (final S3Object s3Object : S3_OBJECTS) {
willAnswer(invocation -> s3Object).given(amazonS3).getObject(S3_BUCKET, s3Object.getKey());
for (Map.Entry<S3Object, InputStream> s3Object : S3_OBJECTS.entrySet()) {
willAnswer(invocation ->
new ResponseInputStream<>(GetObjectResponse.builder().build(), s3Object.getValue()))
.given(amazonS3)
.getObject(GetObjectRequest.builder()
.bucket(S3_BUCKET)
.key(s3Object.getKey().key())
.build());
}
return amazonS3;
}

2
supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java
View File

@@ -32,7 +32,7 @@ import static org.assertj.core.api.Assertions.assertThat;
@TestPropertySource(properties = {
"file.consumer.mode=lines",
"s3.supplier.filenamePattern=*/otherFile",
"file.consumer.with-markers=false" })
"file.consumer.with-markers=false"})
public class AmazonS3LinesTransferredTests extends AbstractAwsS3SupplierMockTests {
@Test

23
supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java
View File

@@ -19,10 +19,10 @@ package org.springframework.cloud.fn.supplier.s3;
import java.time.Duration;
import java.util.HashSet;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import software.amazon.awssdk.services.s3.model.S3Object;
import org.springframework.messaging.Message;
import org.springframework.test.context.TestPropertySource;
@@ -43,22 +43,19 @@ public class AmazonS3ListOnlyTests extends AbstractAwsS3SupplierMockTests {
keys.add("subdir/otherFile");
StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext(message -> {
S3ObjectSummary summary = (S3ObjectSummary) message.getPayload();
assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET);
assertThat(keys).contains(summary.getKey());
keys.remove(summary.getKey());
S3Object s3Object = (S3Object) message.getPayload();
assertThat(keys).contains(s3Object.key());
keys.remove(s3Object.key());
})
.assertNext(message -> {
S3ObjectSummary summary = (S3ObjectSummary) message.getPayload();
assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET);
assertThat(keys).contains(summary.getKey());
keys.remove(summary.getKey());
S3Object s3Object = (S3Object) message.getPayload();
assertThat(keys).contains(s3Object.key());
keys.remove(s3Object.key());
})
.assertNext(message -> {
S3ObjectSummary summary = (S3ObjectSummary) message.getPayload();
assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET);
assertThat(keys).contains(summary.getKey());
keys.remove(summary.getKey());
S3Object s3Object = (S3Object) message.getPayload();
assertThat(keys).contains(s3Object.key());
keys.remove(s3Object.key());
})
.thenCancel()
.verifyLater();