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

Upgrade to latest SI-AWS for S3

Browse Source 
* Fix `AwsS3SupplierConfiguration` to
serialize `S3Object` to JSON
This commit is contained in:
Artem Bilan
2023-10-20 10:28:32 -04:00
parent cf9d8cd79c
commit 595f962128
3 changed files with 41 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,8 +25,8 @@
</dependency>
<dependency>
<groupId>software.amazon.awssdk.crt</groupId>
<artifactId>aws-crt</artifactId>
<groupId>software.amazon.awssdk</groupId>
<artifactId>aws-crt-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>

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

@@ -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<S3Object> filter) {
return new ReactiveMessageSourceProducer(
(MessageSource<List<S3Object>>) () -> {
List<S3Object> summaryList =
(MessageSource<List<String>>) () -> {
List<String> 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);
});
}

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

@@ -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);
}
}
}