Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix S3MessageHandler for upload

Browse Source 
Additional change is made to workaround AWS SDK bug: https://github.com/aws/aws-sdk-java-v2/issues/3839
This commit is contained in:
abilan
2023-03-16 08:59:59 -04:00
parent aee5bdc1af
commit 75469c909a
4 changed files with 34 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
build.gradle
View File

@@ -115,16 +115,17 @@ dependencies {
optionalApi 'io.awspring.cloud:spring-cloud-aws-sns'
optionalApi 'io.awspring.cloud:spring-cloud-aws-sqs'
optionalApi 'io.awspring.cloud:spring-cloud-aws-s3'
optionalApi 'org.springframework.integration:spring-integration-file'
optionalApi 'org.springframework.integration:spring-integration-http'
optionalApi "software.amazon.kinesis:amazon-kinesis-client:$kinesisClientVersion"
optionalApi "com.amazonaws:amazon-kinesis-producer:$kinesisProducerVersion"
optionalApi 'software.amazon.awssdk:kinesis'
optionalApi 'software.amazon.awssdk:dynamodb'
optionalApi 'software.amazon.awssdk:s3-transfer-manager'
optionalApi "software.amazon.kinesis:amazon-kinesis-client:$kinesisClientVersion"
optionalApi "com.amazonaws:amazon-kinesis-producer:$kinesisProducerVersion"
optionalApi "jakarta.servlet:jakarta.servlet-api:$servletApiVersion"
testImplementation('org.springframework.integration:spring-integration-test') {

14
src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java
View File

@@ -24,6 +24,7 @@ import java.util.function.BiConsumer;
import com.amazonaws.util.Base64;
import com.amazonaws.util.Md5Utils;
import org.apache.commons.io.FileUtils;
import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.core.internal.util.Mimetype;
import software.amazon.awssdk.services.s3.S3AsyncClient;
@@ -314,7 +315,8 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler {
if (putObjectRequest.contentType() == null) {
putObjectRequestBuilder.contentType(Mimetype.getInstance().getMimetype(fileToUpload));
}
requestBody = AsyncRequestBody.fromFile(fileToUpload);
// TODO until https://github.com/aws/aws-sdk-java-v2/issues/3839
requestBody = AsyncRequestBody.fromBytes(FileUtils.readFileToByteArray(fileToUpload));
}
else if (payload instanceof byte[] payloadBytes) {
if (putObjectRequest.contentMD5() == null) {
@@ -371,11 +373,7 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler {
String key =
this.keyExpression != null
? this.keyExpression.getValue(this.evaluationContext, requestMessage, String.class)
: targetFile.getName();
Assert.state(key != null,
() -> "The 'keyExpression' must not be null for non-File payloads and can't evaluate to null. "
+ "Root object is: " + requestMessage);
: null;
if (targetFile.isDirectory()) {
DownloadDirectoryRequest.Builder downloadDirectoryRequest =
@@ -393,7 +391,9 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler {
DownloadFileRequest.Builder downloadFileRequest =
DownloadFileRequest.builder()
.destination(targetFile)
.getObjectRequest(request -> request.bucket(bucket).key(key));
.getObjectRequest(request ->
request.bucket(bucket)
.key(key != null ? key : targetFile.getName()));
if (transferListener != null) {
downloadFileRequest.addTransferListener(transferListener);
}

2
src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java
View File

@@ -73,7 +73,7 @@ public interface LocalstackContainerTest {
}
static S3AsyncClient s3Client() {
return applyAwsClientOptions(S3AsyncClient.builder(), LocalStackContainer.Service.CLOUDWATCH);
return applyAwsClientOptions(S3AsyncClient.builder(), LocalStackContainer.Service.S3);
}
static SqsAsyncClient sqsClient() {

38
src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java
View File

@@ -34,7 +34,6 @@ import java.util.concurrent.TimeUnit;
import com.amazonaws.util.StringInputStream;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import software.amazon.awssdk.core.async.AsyncRequestBody;
@@ -42,7 +41,9 @@ import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.CreateBucketResponse;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
import software.amazon.awssdk.services.s3.model.S3Object;
import software.amazon.awssdk.transfer.s3.model.Copy;
import software.amazon.awssdk.transfer.s3.progress.TransferListener;
@@ -111,21 +112,31 @@ public class S3MessageHandlerTests implements LocalstackContainerTest {
@BeforeAll
static void setup() {
S3 = LocalstackContainerTest.s3Client();
S3.createBucket(request -> request.bucket(S3_BUCKET_NAME)).join();
}
@BeforeEach
void prepareBucket() {
try {
S3.deleteBucket(request -> request.bucket(S3_BUCKET_NAME)).get();
}
catch (Exception e) {
// Ignore - assuming no bucket
}
S3.createBucket(request -> request.bucket(S3_BUCKET_NAME)).join();
S3.listObjects(request -> request.bucket(S3_BUCKET_NAME))
.thenCompose(result -> {
if (result.hasContents()) {
return S3.deleteObjects(request -> request.bucket(S3_BUCKET_NAME)
.delete(delete ->
delete.objects(
result.contents()
.stream()
.map(S3Object::key)
.map(key -> ObjectIdentifier.builder().key(key).build())
.toList())));
}
else {
return CompletableFuture.completedFuture(null);
}
})
.join();
}
@Test
@Disabled("The TransferListener.transferComplete is not called")
void testUploadFile() throws IOException, InterruptedException {
File file = new File(temporaryFolder.toFile(), "foo.mp3");
file.createNewFile();
@@ -249,7 +260,6 @@ public class S3MessageHandlerTests implements LocalstackContainerTest {
}
@Test
@Disabled("Unclear why local dir is empty")
void testDownloadDirectory() throws IOException {
CompletableFuture<PutObjectResponse> bb =
S3.putObject(request -> request.bucket(S3_BUCKET_NAME).key(S3_FILE_KEY_BAR),
@@ -293,7 +303,6 @@ public class S3MessageHandlerTests implements LocalstackContainerTest {
}
@Test
@Disabled("The TransferProgressSnapshot does not reflect transferred results")
void testCopy() throws IOException {
byte[] testData = "ff".getBytes();
CompletableFuture<PutObjectResponse> mySource =
@@ -316,9 +325,6 @@ public class S3MessageHandlerTests implements LocalstackContainerTest {
copy.completionFuture().join();
assertThat(copy.progress().snapshot().transferredBytes()).isEqualTo(testData.length);
assertThat(copy.progress().snapshot().remainingBytes().getAsLong()).isEqualTo(0);
File outputFile = new File(temporaryFolder.toFile(), "outputFile");
outputFile.createNewFile();
@@ -340,8 +346,8 @@ public class S3MessageHandlerTests implements LocalstackContainerTest {
public MessageHandler s3MessageHandler() {
S3MessageHandler s3MessageHandler = new S3MessageHandler(S3, S3_BUCKET_NAME);
s3MessageHandler.setCommandExpression(PARSER.parseExpression("headers.s3Command"));
Expression keyExpression = PARSER
.parseExpression("payload instanceof T(java.io.File) ? payload.name : headers.key");
Expression keyExpression = PARSER.parseExpression(
"payload instanceof T(java.io.File) and !payload.directory ? payload.name : headers[key]");
s3MessageHandler.setKeyExpression(keyExpression);
s3MessageHandler.setUploadMetadataProvider((metadata, message) -> {
if (message.getPayload() instanceof InputStream || message.getPayload() instanceof byte[]) {