diff --git a/common/aws-s3-common/pom.xml b/common/aws-s3-common/pom.xml index 172c0ee4..1b2f1d4a 100644 --- a/common/aws-s3-common/pom.xml +++ b/common/aws-s3-common/pom.xml @@ -25,8 +25,8 @@ - software.amazon.awssdk.crt - aws-crt + software.amazon.awssdk + aws-crt-client org.springframework.integration 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 cd670d9e..5ad44a6b 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,6 +22,8 @@ import java.util.function.Supplier; import java.util.regex.Pattern; import java.util.stream.Collectors; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import software.amazon.awssdk.services.s3.S3Client; @@ -206,18 +208,27 @@ public class AwsS3SupplierConfiguration { } @Bean - ReactiveMessageSourceProducer s3ListingMessageProducer(S3Client amazonS3, + ReactiveMessageSourceProducer s3ListingMessageProducer(S3Client amazonS3, ObjectMapper objectMapper, AwsS3SupplierProperties awsS3SupplierProperties, Predicate filter) { return new ReactiveMessageSourceProducer( - (MessageSource>) () -> { - List summaryList = + (MessageSource>) () -> { + List summaryList = amazonS3.listObjects(ListObjectsRequest.builder() .bucket(awsS3SupplierProperties.getRemoteDir()) .build()) .contents() .stream() - .filter(filter).collect(Collectors.toList()); + .filter(filter) + .map(s3Object -> { + try { + return objectMapper.writeValueAsString(s3Object.toBuilder()); + } + catch (JsonProcessingException ex) { + throw new RuntimeException(ex); + } + }) + .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/AmazonS3ListOnlyTests.java b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java index 855bd6f6..bb8d324c 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 @@ -16,14 +16,16 @@ package org.springframework.cloud.fn.supplier.s3; +import java.io.IOException; +import java.io.UncheckedIOException; import java.time.Duration; import java.util.HashSet; 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.integration.json.JsonPathUtils; import org.springframework.messaging.Message; import org.springframework.test.context.TestPropertySource; @@ -43,19 +45,22 @@ public class AmazonS3ListOnlyTests extends AbstractAwsS3SupplierMockTests { keys.add("subdir/otherFile"); StepVerifier stepVerifier = StepVerifier.create(messageFlux) .assertNext(message -> { - S3Object s3Object = (S3Object) message.getPayload(); - assertThat(keys).contains(s3Object.key()); - keys.remove(s3Object.key()); + String s3Object = (String) message.getPayload(); + String key = jsonPathKey(s3Object); + assertThat(keys).contains(key); + keys.remove(key); }) .assertNext(message -> { - S3Object s3Object = (S3Object) message.getPayload(); - assertThat(keys).contains(s3Object.key()); - keys.remove(s3Object.key()); + String s3Object = (String) message.getPayload(); + String key = jsonPathKey(s3Object); + assertThat(keys).contains(key); + keys.remove(key); }) .assertNext(message -> { - S3Object s3Object = (S3Object) message.getPayload(); - assertThat(keys).contains(s3Object.key()); - keys.remove(s3Object.key()); + String s3Object = (String) message.getPayload(); + String key = jsonPathKey(s3Object); + assertThat(keys).contains(key); + keys.remove(key); }) .thenCancel() .verifyLater(); @@ -63,4 +68,13 @@ public class AmazonS3ListOnlyTests extends AbstractAwsS3SupplierMockTests { stepVerifier.verify(Duration.ofSeconds(10)); } + private static String jsonPathKey(String s3Object) { + try { + return JsonPathUtils.evaluate(s3Object, "$.key"); + } + catch (IOException ex) { + throw new UncheckedIOException(ex); + } + } + }