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

GH-41: Fix broken S3 File upload

Browse Source 
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.
This commit is contained in:
John Logan
2016-07-12 21:51:22 -07:00
committed by Artem Bilan
parent 07e8d59c1f
commit b5e6cdaa95
3 changed files with 104 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

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

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

@@ -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) {

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

@@ -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<PutObjectRequest> 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");