From b5e6cdaa952ae9e3778c18de3ffb25f4f9df7b38 Mon Sep 17 00:00:00 2001 From: John Logan Date: Tue, 12 Jul 2016 21:51:22 -0700 Subject: [PATCH] GH-41: Fix broken S3 File upload Fixes GH-41 (https://github.com/spring-projects/spring-integration-aws/issues/41) Fixes GH-43 (https://github.com/spring-projects/spring-integration-aws/issues/43) - Prior upload() code was trying to reset() a FileInputStream and then upload to S3, which resulted in zero uploaded bytes. - Added dependency on spring-cloud-aws-core to resolve ResourceIdResolver reference. - Fail if InputStream payload does not support mark/reset. - Add metadata for byte array payload. GH-41: Fix assert order per PR comments. --- build.gradle | 1 + .../aws/outbound/S3MessageHandler.java | 93 ++++++++++--------- .../aws/outbound/S3MessageHandlerTests.java | 56 ++++++++++- 3 files changed, 104 insertions(+), 46 deletions(-) diff --git a/build.gradle b/build.gradle index bccb769..0275fb2 100644 --- a/build.gradle +++ b/build.gradle @@ -84,6 +84,7 @@ checkstyle { dependencies { compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion" + compile "org.springframework.cloud:spring-cloud-aws-core:$springCloudAwsVersion" compile("org.springframework.cloud:spring-cloud-aws-messaging:$springCloudAwsVersion", optional) compile("org.springframework.integration:spring-integration-file:$springIntegrationVersion", optional) compile("org.springframework.integration:spring-integration-http:$springIntegrationVersion", optional) diff --git a/src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java index ddfe6de..180a7a3 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/S3MessageHandler.java @@ -18,8 +18,6 @@ package org.springframework.integration.aws.outbound; import java.io.ByteArrayInputStream; import java.io.File; -import java.io.FileInputStream; -import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStream; @@ -35,7 +33,6 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; -import org.springframework.util.StreamUtils; import com.amazonaws.AmazonClientException; import com.amazonaws.event.ProgressEvent; @@ -94,6 +91,7 @@ import com.amazonaws.util.Md5Utils; * {@link #destinationKeyExpression} are required and must not evaluate to {@code null}. * * @author Artem Bilan + * @author John Logan * * @see TransferManager */ @@ -312,39 +310,63 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler { if (this.uploadMetadataProvider != null) { this.uploadMetadataProvider.populateMetadata(metadata, requestMessage); } - InputStream inputStream; - if (payload instanceof InputStream) { - inputStream = (InputStream) payload; - } - else if (payload instanceof File) { - File fileToUpload = (File) payload; - if (key == null) { - key = fileToUpload.getName(); - } - try { - inputStream = new FileInputStream(fileToUpload); + PutObjectRequest putObjectRequest = null; + + try { + if (payload instanceof InputStream) { + InputStream inputStream = (InputStream) payload; + if (metadata.getContentMD5() == null) { + Assert.state(inputStream.markSupported(), + "For an upload InputStream with no MD5 digest metadata, the " + + "markSupported() method must evaluate to true. "); + String contentMd5 = Md5Utils.md5AsBase64(inputStream); + metadata.setContentMD5(contentMd5); + inputStream.reset(); + } + putObjectRequest = new PutObjectRequest(bucketName, key, inputStream, metadata); + } + else if (payload instanceof File) { + File fileToUpload = (File) payload; + if (key == null) { + key = fileToUpload.getName(); + } + if (metadata.getContentMD5() == null) { + String contentMd5 = Md5Utils.md5AsBase64(fileToUpload); + metadata.setContentMD5(contentMd5); + } if (metadata.getContentLength() == 0) { metadata.setContentLength(fileToUpload.length()); } if (metadata.getContentType() == null) { metadata.setContentType(Mimetypes.getInstance().getMimetype(fileToUpload)); } + putObjectRequest = new PutObjectRequest(bucketName, key, fileToUpload).withMetadata(metadata); + } + else if (payload instanceof byte[]) { + byte[] payloadBytes = (byte[]) payload; + InputStream inputStream = new ByteArrayInputStream(payloadBytes); + if (metadata.getContentMD5() == null) { + String contentMd5 = Md5Utils.md5AsBase64(inputStream); + metadata.setContentMD5(contentMd5); + inputStream.reset(); + } + if (metadata.getContentLength() == 0) { + metadata.setContentLength(payloadBytes.length); + } + putObjectRequest = new PutObjectRequest(bucketName, key, inputStream, metadata); + } + else { + throw new IllegalArgumentException("Unsupported payload type: [" + + payload.getClass() + + "]. The only supported payloads for the upload request are " + + "java.io.File, java.io.InputStream, byte[] and PutObjectRequest."); + } + } + catch (IOException e) { + throw new MessageHandlingException(requestMessage, e); + } - } - catch (FileNotFoundException e) { - throw new AmazonClientException(e); - } - } - else if (payload instanceof byte[]) { - inputStream = new ByteArrayInputStream((byte[]) payload); - } - else { - throw new IllegalArgumentException("Unsupported payload type: [" - + payload.getClass() - + "]. The only supported payloads for the upload request are " + - "java.io.File, java.io.InputStream, byte[] and PutObjectRequest."); - } Assert.state(key != null, "The 'keyExpression' must not be null for non-File payloads and can't evaluate to null. " + @@ -361,21 +383,6 @@ public class S3MessageHandler extends AbstractReplyProducingMessageHandler { } } - if (metadata.getContentMD5() == null) { - String contentMd5 = null; - try { - contentMd5 = Md5Utils.md5AsBase64(StreamUtils.copyToByteArray(inputStream)); - if (inputStream.markSupported()) { - inputStream.reset(); - } - metadata.setContentMD5(contentMd5); - } - catch (IOException e) { - throw new MessageHandlingException(requestMessage, e); - } - } - PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, key, inputStream, metadata); - S3ProgressListener progressListener = this.s3ProgressListener; if (this.objectAclExpression != null) { diff --git a/src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java index 1c4d213..a09220d 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java +++ b/src/test/java/org/springframework/integration/aws/outbound/S3MessageHandlerTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.aws.outbound; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Fail.fail; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.Matchers.any; @@ -27,6 +28,7 @@ import static org.mockito.Mockito.verify; import java.io.ByteArrayInputStream; import java.io.File; +import java.io.FileInputStream; import java.io.FileReader; import java.io.IOException; import java.io.InputStream; @@ -60,6 +62,7 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; @@ -98,6 +101,7 @@ import com.amazonaws.util.StringUtils; /** * @author Artem Bilan + * @author John Logan */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration @@ -143,8 +147,8 @@ public class S3MessageHandlerTests { PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue(); assertThat(putObjectRequest.getBucketName()).isEqualTo("myBucket"); assertThat(putObjectRequest.getKey()).isEqualTo("foo.mp3"); - assertThat(putObjectRequest.getFile()).isNull(); - assertThat(putObjectRequest.getInputStream()).isNotNull(); + assertThat(putObjectRequest.getFile()).isNotNull(); + assertThat(putObjectRequest.getInputStream()).isNull(); ObjectMetadata metadata = putObjectRequest.getMetadata(); assertThat(metadata.getContentMD5()).isEqualTo(Md5Utils.md5AsBase64(file)); @@ -196,6 +200,52 @@ public class S3MessageHandlerTests { assertThat(metadata.getContentDisposition()).isEqualTo("test.json"); } + @Test + public void testUploadInputStreamNoMarkSupported() throws IOException, InterruptedException { + File file = this.temporaryFolder.newFile("foo.mp3"); + FileInputStream fileInputStream = new FileInputStream(file); + Message message = MessageBuilder.withPayload(fileInputStream) + .setHeader("s3Command", S3MessageHandler.Command.UPLOAD.name()) + .setHeader("key", "myStream") + .build(); + + try { + this.s3SendChannel.send(message); + fail("Expected send() failure with FileInputStream, got success."); + } + catch (Exception e) { + assertThat(e).isInstanceOf(MessageHandlingException.class); + assertThat(e.getCause()).isInstanceOf(IllegalStateException.class); + } + } + + @Test + public void testUploadByteArray() throws IOException { + byte[] payload = "b".getBytes("UTF-8"); + Message message = MessageBuilder.withPayload(payload) + .setHeader("s3Command", S3MessageHandler.Command.UPLOAD.name()) + .setHeader("key", "myStream") + .build(); + + this.s3SendChannel.send(message); + + ArgumentCaptor putObjectRequestArgumentCaptor = + ArgumentCaptor.forClass(PutObjectRequest.class); + verify(this.amazonS3, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture()); + + PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue(); + assertThat(putObjectRequest.getBucketName()).isEqualTo("myBucket"); + assertThat(putObjectRequest.getKey()).isEqualTo("myStream"); + assertThat(putObjectRequest.getFile()).isNull(); + assertThat(putObjectRequest.getInputStream()).isNotNull(); + + ObjectMetadata metadata = putObjectRequest.getMetadata(); + assertThat(metadata.getContentMD5()).isEqualTo(Md5Utils.md5AsBase64(payload)); + assertThat(metadata.getContentLength()).isEqualTo(1); + assertThat(metadata.getContentType()).isEqualTo(MediaType.APPLICATION_JSON_VALUE); + assertThat(metadata.getContentDisposition()).isEqualTo("test.json"); + } + @Test public void testDownloadDirectory() throws IOException { File directoryForDownload = this.temporaryFolder.newFolder("myFolder"); @@ -368,7 +418,7 @@ public class S3MessageHandlerTests { s3MessageHandler.setKeyExpression(keyExpression); s3MessageHandler.setObjectAclExpression(new ValueExpression<>(CannedAccessControlList.PublicReadWrite)); s3MessageHandler.setUploadMetadataProvider((metadata, message) -> { - if (message.getPayload() instanceof InputStream) { + if (message.getPayload() instanceof InputStream || message.getPayload() instanceof byte[]) { metadata.setContentLength(1); metadata.setContentType(MediaType.APPLICATION_JSON_VALUE); metadata.setContentDisposition("test.json");