diff --git a/common/aws-s3-common/pom.xml b/common/aws-s3-common/pom.xml index 410d54de..172c0ee4 100644 --- a/common/aws-s3-common/pom.xml +++ b/common/aws-s3-common/pom.xml @@ -11,12 +11,7 @@ aws-s3-common aws-s3-common - aws-s3 consumer - - - 2.5.2 - 2.4.2 - + aws-s3 common @@ -26,8 +21,12 @@ io.awspring.cloud - spring-cloud-starter-aws - ${spring-cloud-aws.version} + spring-cloud-aws-starter-s3 + + + + software.amazon.awssdk.crt + aws-crt org.springframework.integration diff --git a/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Configuration.java b/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Configuration.java index 14a8b77c..6f9f07e5 100644 --- a/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Configuration.java +++ b/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Configuration.java @@ -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; } } diff --git a/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Properties.java b/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Properties.java deleted file mode 100644 index b954fcb7..00000000 --- a/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Properties.java +++ /dev/null @@ -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; - } -} diff --git a/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/CompatibleStorageAmazonS3Configuration.java b/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/CompatibleStorageAmazonS3Configuration.java deleted file mode 100644 index bc0fd5e4..00000000 --- a/common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/CompatibleStorageAmazonS3Configuration.java +++ /dev/null @@ -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; - } - -} diff --git a/common/aws-s3-common/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/common/aws-s3-common/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index 9966a9e5..cf831442 100644 --- a/common/aws-s3-common/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/common/aws-s3-common/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1,2 +1 @@ -org.springframework.cloud.fn.common.aws.s3.CompatibleStorageAmazonS3Configuration org.springframework.cloud.fn.common.aws.s3.AmazonS3Configuration diff --git a/common/aws-s3-common/src/test/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3ConfigurationTests.java b/common/aws-s3-common/src/test/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3ConfigurationTests.java index 37a780a0..ab1cf09c 100644 --- a/common/aws-s3-common/src/test/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3ConfigurationTests.java +++ b/common/aws-s3-common/src/test/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3ConfigurationTests.java @@ -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")); } } diff --git a/common/metadata-store-common/pom.xml b/common/metadata-store-common/pom.xml index 46608d00..d8086754 100644 --- a/common/metadata-store-common/pom.xml +++ b/common/metadata-store-common/pom.xml @@ -13,13 +13,6 @@ metadata-store-common metadata-store common - - 1.12.322 - 2.5.2 - 6.1.1 - 5.3.0 - - @@ -46,6 +39,12 @@ true + + io.awspring.cloud + spring-cloud-aws-starter + true + + org.springframework.boot spring-boot-starter-logging @@ -103,9 +102,8 @@ true - com.amazonaws - aws-java-sdk-dynamodb - ${aws-java-sdk.version} + software.amazon.awssdk + dynamodb true diff --git a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java index e92489f3..fa2b9fa6 100644 --- a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java +++ b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java @@ -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(); diff --git a/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java b/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java index 7eb47026..107bdde3 100644 --- a/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java +++ b/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java @@ -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.>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); } } diff --git a/consumer/s3-consumer/README.adoc b/consumer/s3-consumer/README.adoc index cc24f84c..073c5b59 100644 --- a/consumer/s3-consumer/README.adoc +++ b/consumer/s3-consumer/README.adoc @@ -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. \ No newline at end of file +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. diff --git a/consumer/s3-consumer/pom.xml b/consumer/s3-consumer/pom.xml index 66e66135..6ca9d535 100644 --- a/consumer/s3-consumer/pom.xml +++ b/consumer/s3-consumer/pom.xml @@ -24,6 +24,10 @@ file-common ${project.version} + + software.amazon.awssdk + s3-transfer-manager + org.springframework spring-web diff --git a/consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerConfiguration.java b/consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerConfiguration.java index 7c4f4e4c..8eceaab8 100644 --- a/consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerConfiguration.java +++ b/consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerConfiguration.java @@ -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> s3Consumer() { - return amazonS3MessageHandler(null, null, null)::handleMessage; + public Consumer> 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> 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> 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; } + } diff --git a/consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerProperties.java b/consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerProperties.java index baeaa80c..04dd3388 100644 --- a/consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerProperties.java +++ b/consumer/s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/AwsS3ConsumerProperties.java @@ -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; } diff --git a/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AbstractAwsS3ConsumerMockTests.java b/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AbstractAwsS3ConsumerMockTests.java index aae01e1f..c209f941 100644 --- a/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AbstractAwsS3ConsumerMockTests.java +++ b/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AbstractAwsS3ConsumerMockTests.java @@ -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> 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"); } }; } diff --git a/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadFileTests.java b/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadFileTests.java index d58e79e8..a04a8a82 100644 --- a/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadFileTests.java +++ b/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadFileTests.java @@ -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 putObjectRequestArgumentCaptor = ArgumentCaptor.forClass(PutObjectRequest.class); - verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture()); + ArgumentCaptor 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 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); } + } diff --git a/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadInputStreamTests.java b/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadInputStreamTests.java index 0703e021..8c179965 100644 --- a/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadInputStreamTests.java +++ b/consumer/s3-consumer/src/test/java/org/springframework/cloud/fn/consumer/s3/AmazonS3UploadInputStreamTests.java @@ -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 putObjectRequestArgumentCaptor = ArgumentCaptor.forClass(PutObjectRequest.class); - verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture()); + ArgumentCaptor 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(); } + } diff --git a/supplier/s3-supplier/README.adoc b/supplier/s3-supplier/README.adoc index e187ffc8..760a086e 100644 --- a/supplier/s3-supplier/README.adoc +++ b/supplier/s3-supplier/README.adoc @@ -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>>`. 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` 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. diff --git a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java index bdfc585d..cd670d9e 100644 --- a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java +++ b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java @@ -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 filter(ConcurrentMetadataStore metadataStore) { - ChainFileListFilter chainFilter = new ChainFileListFilter<>(); + public ChainFileListFilter filter(ConcurrentMetadataStore metadataStore) { + ChainFileListFilter 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 filter) { + public S3InboundFileSynchronizer s3InboundFileSynchronizer(ChainFileListFilter 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 listOnlyFilter() { - Predicate predicate = s -> true; + Predicate listOnlyFilter(AwsS3SupplierProperties awsS3SupplierProperties) { + Predicate 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 filter) { + ReactiveMessageSourceProducer s3ListingMessageProducer(S3Client amazonS3, + AwsS3SupplierProperties awsS3SupplierProperties, Predicate filter) { - ListObjectsRequest listObjectsRequest = new ListObjectsRequest(); - listObjectsRequest.setBucketName(awsS3SupplierProperties.getRemoteDir()); return new ReactiveMessageSourceProducer( - (MessageSource>) () -> { - List summaryList = amazonS3.listObjects(listObjectsRequest) - .getObjectSummaries().stream() - .filter(filter).collect(Collectors.toList()); + (MessageSource>) () -> { + List summaryList = + amazonS3.listObjects(ListObjectsRequest.builder() + .bucket(awsS3SupplierProperties.getRemoteDir()) + .build()) + .contents() + .stream() + .filter(filter).collect(Collectors.toList()); return summaryList.isEmpty() ? null : new GenericMessage<>(summaryList); }); } diff --git a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java index 83cd2b03..23b2ba4c 100644 --- a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java +++ b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java @@ -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 S3_OBJECTS; + protected static Map S3_OBJECTS; @Autowired Supplier>> 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 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 : 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; } diff --git a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java index c6305d67..8f2713d3 100644 --- a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java +++ b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java @@ -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 diff --git a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java index 1f4adc69..855bd6f6 100644 --- a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java +++ b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java @@ -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();