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