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

Make S3 modules as auto-configuration

Browse Source 
* Fix all the Checkstyle violations and compiler warnings
* Fix HTTP Supplier props prefix in the README
This commit is contained in:
Artem Bilan
2024-01-08 13:13:43 -05:00
parent 090ca28233
commit 373fbdd8a2
17 changed files with 93 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
consumer/spring-s3-consumer/README.adoc
View File

@@ -5,7 +5,7 @@ The consumer uses the AWS S3 support from Spring Integration and Spring Cloud AW
== Beans for injection
You can import `AwsS3ConsumerConfiguration` in the application and then inject the following bean.
The `AwsS3ConsumerConfiguration` auto-configuration provides the following bean:
`Consumer<Message<?>> s3Consumer`

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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,14 +19,16 @@ package org.springframework.cloud.fn.consumer.s3;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import io.awspring.cloud.autoconfigure.s3.S3TransferManagerAutoConfiguration;
import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
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.BeanFactory;
import org.springframework.boot.autoconfigure.AutoConfiguration;
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;
@@ -39,7 +41,12 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
@Configuration(proxyBeanMethods = false)
/**
* Auto-configuration for S3 consumer.
*
* @author Artem Bilan
*/
@AutoConfiguration(after = S3TransferManagerAutoConfiguration.class)
@EnableConfigurationProperties(AwsS3ConsumerProperties.class)
public class AwsS3ConsumerConfiguration {
@@ -52,7 +59,7 @@ public class AwsS3ConsumerConfiguration {
public IntegrationFlow s3ConsumerFlow(@Nullable TransferListener transferListener,
MessageHandler amazonS3MessageHandler) {
return flow -> flow.enrichHeaders(headers -> headers.header(AwsHeaders.TRANSFER_LISTENER, transferListener))
return (flow) -> flow.enrichHeaders((headers) -> headers.header(AwsHeaders.TRANSFER_LISTENER, transferListener))
.handle(amazonS3MessageHandler);
}
@@ -62,8 +69,9 @@ public class AwsS3ConsumerConfiguration {
@Nullable BiConsumer<PutObjectRequest.Builder, Message<?>> uploadMetadataProvider) {
Expression bucketExpression = s3ConsumerProperties.getBucketExpression();
if (s3ConsumerProperties.getBucket() != null) {
bucketExpression = new ValueExpression<>(s3ConsumerProperties.getBucket());
String bucket = s3ConsumerProperties.getBucket();
if (bucket != null) {
bucketExpression = new ValueExpression<>(bucket);
}
S3MessageHandler s3MessageHandler = new S3MessageHandler(s3TransferManager, bucketExpression);
@@ -71,8 +79,9 @@ public class AwsS3ConsumerConfiguration {
Expression aclExpression;
if (s3ConsumerProperties.getAcl() != null) {
aclExpression = new ValueExpression<>(s3ConsumerProperties.getAcl());
ObjectCannedACL acl = s3ConsumerProperties.getAcl();
if (acl != null) {
aclExpression = new ValueExpression<>(acl);
}
else {
aclExpression = s3ConsumerProperties.getAclExpression();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -25,6 +25,8 @@ import org.springframework.expression.Expression;
import org.springframework.validation.annotation.Validated;
/**
* Configuration properties for S3 consumer.
*
* @author Artem Bilan
*/
@ConfigurationProperties("s3.consumer")

4
consumer/spring-s3-consumer/src/main/java/org/springframework/cloud/fn/consumer/s3/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The S3 consumer auto-configuration support.
*/
package org.springframework.cloud.fn.consumer.s3;

1
consumer/spring-s3-consumer/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.consumer.s3.AwsS3ConsumerConfiguration

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -51,10 +51,8 @@ public class AmazonS3UploadFileTests extends AbstractAwsS3ConsumerMockTests {
this.s3Consumer.accept(message);
ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor = ArgumentCaptor
.forClass(PutObjectRequest.class);
ArgumentCaptor<AsyncRequestBody> asyncRequestBodyArgumentCaptor = ArgumentCaptor
.forClass(AsyncRequestBody.class);
ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor = ArgumentCaptor.captor();
ArgumentCaptor<AsyncRequestBody> asyncRequestBodyArgumentCaptor = ArgumentCaptor.captor();
verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture(),
asyncRequestBodyArgumentCaptor.capture());
@@ -68,7 +66,7 @@ public class AmazonS3UploadFileTests extends AbstractAwsS3ConsumerMockTests {
AsyncRequestBody asyncRequestBody = asyncRequestBodyArgumentCaptor.getValue();
StepVerifier.create(asyncRequestBody)
.assertNext(buffer -> assertThat(TestUtils.getPropertyValue(buffer, "hb", byte[].class)).isEmpty())
.assertNext((buffer) -> assertThat(TestUtils.getPropertyValue(buffer, "hb", byte[].class)).isEmpty())
.expectComplete()
.verify();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -51,10 +51,8 @@ public class AmazonS3UploadInputStreamTests extends AbstractAwsS3ConsumerMockTes
this.s3Consumer.accept(message);
ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor = ArgumentCaptor
.forClass(PutObjectRequest.class);
ArgumentCaptor<AsyncRequestBody> asyncRequestBodyArgumentCaptor = ArgumentCaptor
.forClass(AsyncRequestBody.class);
ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor = ArgumentCaptor.captor();
ArgumentCaptor<AsyncRequestBody> asyncRequestBodyArgumentCaptor = ArgumentCaptor.captor();
verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture(),
asyncRequestBodyArgumentCaptor.capture());
@@ -67,7 +65,7 @@ public class AmazonS3UploadInputStreamTests extends AbstractAwsS3ConsumerMockTes
assertThat(putObjectRequest.contentDisposition()).isEqualTo("test.json");
AsyncRequestBody asyncRequestBody = asyncRequestBodyArgumentCaptor.getValue();
StepVerifier.create(asyncRequestBody.map(buffer -> StandardCharsets.UTF_8.decode(buffer).toString()))
StepVerifier.create(asyncRequestBody.map((buffer) -> StandardCharsets.UTF_8.decode(buffer).toString()))
.expectNext("a")
.expectComplete()
.verify();

2
supplier/spring-http-supplier/README.adoc
View File

@@ -20,7 +20,7 @@ Once injected, you can use the `get` method of the `Supplier` to invoke it and t
## Configuration Options
All configuration properties are prefixed with `http`.
All configuration properties are prefixed with `http.supplier`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/http/HttpSupplierProperties.java[HttpSupplierProperties].

2
supplier/spring-s3-supplier/README.adoc
View File

@@ -8,7 +8,7 @@ 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.
The `AwsS3SupplierConfiguration` auto-configuration provides the following bean:
`s3Supplier`

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -30,8 +30,10 @@ 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.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.aws.s3.AmazonS3Configuration;
import org.springframework.cloud.fn.common.config.ComponentCustomizer;
import org.springframework.cloud.fn.common.file.FileConsumerProperties;
import org.springframework.cloud.fn.common.file.FileUtils;
@@ -55,10 +57,12 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.StringUtils;
/**
* Auto-configuration for S3 supplier.
*
* @author Artem Bilan
* @author David Turanski
*/
@Configuration(proxyBeanMethods = false)
@AutoConfiguration(after = AmazonS3Configuration.class)
@EnableConfigurationProperties({ AwsS3SupplierProperties.class, FileConsumerProperties.class })
public class AwsS3SupplierConfiguration {
@@ -87,12 +91,12 @@ public class AwsS3SupplierConfiguration {
static class SynchronizingConfiguration extends AwsS3SupplierConfiguration {
@Bean
public Supplier<Flux<Message<?>>> s3Supplier(Publisher<Message<?>> s3SupplierFlow) {
Supplier<Flux<Message<?>>> s3Supplier(Publisher<Message<?>> s3SupplierFlow) {
return () -> Flux.from(s3SupplierFlow);
}
@Bean
public ChainFileListFilter<S3Object> filter(ConcurrentMetadataStore metadataStore) {
ChainFileListFilter<S3Object> filter(ConcurrentMetadataStore metadataStore) {
ChainFileListFilter<S3Object> chainFilter = new ChainFileListFilter<>();
if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) {
chainFilter
@@ -115,19 +119,19 @@ public class AwsS3SupplierConfiguration {
}
@Bean
public Publisher<Message<Object>> s3SupplierFlow(S3InboundFileSynchronizingMessageSource s3MessageSource) {
Publisher<Message<Object>> s3SupplierFlow(S3InboundFileSynchronizingMessageSource s3MessageSource) {
return FileUtils
.enhanceFlowForReadingMode(
IntegrationFlow.from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource)
.doOnSubscribe((s) -> s3MessageSource.start())),
fileConsumerProperties)
this.fileConsumerProperties)
.toReactivePublisher(true);
}
@Bean
public S3InboundFileSynchronizer s3InboundFileSynchronizer(ChainFileListFilter<S3Object> filter) {
S3InboundFileSynchronizer s3InboundFileSynchronizer(ChainFileListFilter<S3Object> filter) {
S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(s3SessionFactory);
S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(this.s3SessionFactory);
synchronizer.setDeleteRemoteFiles(this.awsS3SupplierProperties.isDeleteRemoteFiles());
synchronizer.setPreserveTimestamp(this.awsS3SupplierProperties.isPreserveTimestamp());
String remoteDir = this.awsS3SupplierProperties.getRemoteDir();
@@ -140,8 +144,7 @@ public class AwsS3SupplierConfiguration {
}
@Bean
public S3InboundFileSynchronizingMessageSource s3MessageSource(
S3InboundFileSynchronizer s3InboundFileSynchronizer,
S3InboundFileSynchronizingMessageSource s3MessageSource(S3InboundFileSynchronizer s3InboundFileSynchronizer,
@Nullable ComponentCustomizer<S3InboundFileSynchronizingMessageSource> s3MessageSourceCustomizer) {
S3InboundFileSynchronizingMessageSource s3MessageSource = new S3InboundFileSynchronizingMessageSource(
@@ -170,18 +173,18 @@ public class AwsS3SupplierConfiguration {
}
@Bean
public Supplier<Flux<Message<Object>>> s3Supplier(Publisher<Message<Object>> s3SupplierFlow) {
Supplier<Flux<Message<Object>>> s3Supplier(Publisher<Message<Object>> s3SupplierFlow) {
return () -> Flux.from(s3SupplierFlow);
}
@Bean
public Publisher<Message<Object>> s3SupplierFlow(ReactiveMessageSourceProducer s3ListingProducer) {
Publisher<Message<Object>> s3SupplierFlow(ReactiveMessageSourceProducer s3ListingProducer) {
return IntegrationFlow.from(s3ListingProducer).split().toReactivePublisher(true);
}
@Bean
Predicate<S3Object> listOnlyFilter(AwsS3SupplierProperties awsS3SupplierProperties) {
Predicate<S3Object> predicate = s -> true;
Predicate<S3Object> predicate = (s) -> true;
if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) {
Pattern pattern = Pattern.compile(this.awsS3SupplierProperties.getFilenamePattern());
predicate = (S3Object summary) -> pattern.matcher(summary.key()).matches();
@@ -197,7 +200,7 @@ public class AwsS3SupplierConfiguration {
final String storedLastModified = this.metadataStore.get(key);
boolean result = !lastModified.equals(storedLastModified);
if (result) {
metadataStore.put(key, lastModified);
this.metadataStore.put(key, lastModified);
}
return result;
});
@@ -214,7 +217,7 @@ public class AwsS3SupplierConfiguration {
.contents()
.stream()
.filter(filter)
.map(s3Object -> {
.map((s3Object) -> {
try {
return objectMapper.writeValueAsString(s3Object.toBuilder());
}

22
supplier/spring-s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2024 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.
@@ -28,6 +28,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;
/**
* The configuration properties for S3 supplier.
*
* @author Artem Bilan
*/
@ConfigurationProperties("s3.supplier")
@@ -95,7 +97,7 @@ public class AwsS3SupplierProperties {
@NotBlank
public String getTmpFileSuffix() {
return tmpFileSuffix;
return this.tmpFileSuffix;
}
public void setTmpFileSuffix(String tmpFileSuffix) {
@@ -104,7 +106,7 @@ public class AwsS3SupplierProperties {
@NotBlank
public String getRemoteFileSeparator() {
return remoteFileSeparator;
return this.remoteFileSeparator;
}
public void setRemoteFileSeparator(String remoteFileSeparator) {
@@ -112,7 +114,7 @@ public class AwsS3SupplierProperties {
}
public boolean isAutoCreateLocalDir() {
return autoCreateLocalDir;
return this.autoCreateLocalDir;
}
public void setAutoCreateLocalDir(boolean autoCreateLocalDir) {
@@ -120,7 +122,7 @@ public class AwsS3SupplierProperties {
}
public boolean isDeleteRemoteFiles() {
return deleteRemoteFiles;
return this.deleteRemoteFiles;
}
public void setDeleteRemoteFiles(boolean deleteRemoteFiles) {
@@ -129,7 +131,7 @@ public class AwsS3SupplierProperties {
@NotNull
public File getLocalDir() {
return localDir;
return this.localDir;
}
public final void setLocalDir(File localDir) {
@@ -137,7 +139,7 @@ public class AwsS3SupplierProperties {
}
public String getFilenamePattern() {
return filenamePattern;
return this.filenamePattern;
}
public void setFilenamePattern(String filenamePattern) {
@@ -145,7 +147,7 @@ public class AwsS3SupplierProperties {
}
public Pattern getFilenameRegex() {
return filenameRegex;
return this.filenameRegex;
}
public void setFilenameRegex(Pattern filenameRegex) {
@@ -153,7 +155,7 @@ public class AwsS3SupplierProperties {
}
public boolean isPreserveTimestamp() {
return preserveTimestamp;
return this.preserveTimestamp;
}
public void setPreserveTimestamp(boolean preserveTimestamp) {
@@ -166,7 +168,7 @@ public class AwsS3SupplierProperties {
}
public boolean isListOnly() {
return listOnly;
return this.listOnly;
}
public void setListOnly(boolean listOnly) {

4
supplier/spring-s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* The S3 supplier auto-configuration support.
*/
package org.springframework.cloud.fn.supplier.s3;

1
supplier/spring-s3-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.supplier.s3.AwsS3SupplierConfiguration

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -117,11 +117,11 @@ public abstract class AbstractAwsS3SupplierMockTests {
@AfterAll
public static void tearDown() {
System.clearProperty("s3.supplier.localDir");
S3_OBJECTS.values().forEach(stream -> {
S3_OBJECTS.values().forEach((stream) -> {
try {
stream.close();
}
catch (IOException e) {
catch (IOException ex) {
// Ignore
}
});
@@ -140,10 +140,10 @@ public abstract class AbstractAwsS3SupplierMockTests {
.isTruncated(false)
.build();
willAnswer(invocation -> listObjectsResponse).given(amazonS3).listObjects(any(ListObjectsRequest.class));
willAnswer((invocation) -> listObjectsResponse).given(amazonS3).listObjects(any(ListObjectsRequest.class));
for (Map.Entry<S3Object, InputStream> s3Object : S3_OBJECTS.entrySet()) {
willAnswer(invocation -> new ResponseInputStream<>(GetObjectResponse.builder().build(),
willAnswer((invocation) -> new ResponseInputStream<>(GetObjectResponse.builder().build(),
s3Object.getValue()))
.given(amazonS3)
.getObject(GetObjectRequest.builder().bucket(S3_BUCKET).key(s3Object.getKey().key()).build());

16
supplier/spring-s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3FilesTransferredTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -34,15 +34,15 @@ public class AmazonS3FilesTransferredTests extends AbstractAwsS3SupplierMockTest
@Test
public void test() {
final Flux<Message<?>> messageFlux = s3Supplier.get();
StepVerifier stepVerifier = StepVerifier.create(messageFlux).assertNext((message) -> {
assertThat(new File(message.getPayload().toString().replaceAll("\"", "")))
StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext((message) -> assertThat(new File(message.getPayload().toString().replaceAll("\"", "")))
.isEqualTo(new File(this.awsS3SupplierProperties.getLocalDir() + File.separator + "subdir"
+ File.separator + "1.test"));
}).assertNext((message) -> {
assertThat(new File(message.getPayload().toString().replaceAll("\"", "")))
+ File.separator + "1.test")))
.assertNext((message) -> assertThat(new File(message.getPayload().toString().replaceAll("\"", "")))
.isEqualTo(new File(this.awsS3SupplierProperties.getLocalDir() + File.separator + "subdir"
+ File.separator + "2.test"));
}).thenCancel().verifyLater();
+ File.separator + "2.test")))
.thenCancel()
.verifyLater();
standardIntegrationFlow.start();
stepVerifier.verify(Duration.ofSeconds(10));
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2024 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.
@@ -41,9 +41,10 @@ public class AmazonS3LinesTransferredTests extends AbstractAwsS3SupplierMockTest
assertThat(message.getHeaders()).containsKey(FileHeaders.ORIGINAL_FILE);
assertThat(message.getHeaders())
.containsValue(new File(this.awsS3SupplierProperties.getLocalDir(), "subdir/otherFile"));
}).assertNext((message) -> {
assertThat(message.getPayload().toString()).isEqualTo("Other2");
}).thenCancel().verifyLater();
})
.assertNext((message) -> assertThat(message.getPayload().toString()).isEqualTo("Other2"))
.thenCancel()
.verifyLater();
standardIntegrationFlow.start();
stepVerifier.verify(Duration.ofSeconds(10));

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2023 the original author or authors.
* Copyright 2020-2024 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.
@@ -41,17 +41,17 @@ public class AmazonS3ListOnlyTests extends AbstractAwsS3SupplierMockTests {
keys.add("subdir/1.test");
keys.add("subdir/2.test");
keys.add("subdir/otherFile");
StepVerifier stepVerifier = StepVerifier.create(messageFlux).assertNext(message -> {
StepVerifier stepVerifier = StepVerifier.create(messageFlux).assertNext((message) -> {
String s3Object = (String) message.getPayload();
String key = jsonPathKey(s3Object);
assertThat(keys).contains(key);
keys.remove(key);
}).assertNext(message -> {
}).assertNext((message) -> {
String s3Object = (String) message.getPayload();
String key = jsonPathKey(s3Object);
assertThat(keys).contains(key);
keys.remove(key);
}).assertNext(message -> {
}).assertNext((message) -> {
String s3Object = (String) message.getPayload();
String key = jsonPathKey(s3Object);
assertThat(keys).contains(key);