You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ni...@apache.org on 2014/11/03 15:20:21 UTC
[3/4] git commit: CAMEL-7991 Added MultiPartUpload functionality to
S3Producer with thanks to Andreas
CAMEL-7991 Added MultiPartUpload functionality to S3Producer with thanks to Andreas
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/aa73efb1
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/aa73efb1
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/aa73efb1
Branch: refs/heads/master
Commit: aa73efb1b88ec216399c8064dd7f6be4c80e6744
Parents: cf6959e
Author: Willem Jiang <wi...@gmail.com>
Authored: Mon Nov 3 11:29:21 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Mon Nov 3 22:19:49 2014 +0800
----------------------------------------------------------------------
.../camel/component/aws/s3/S3Configuration.java | 31 +++-
.../camel/component/aws/s3/S3Endpoint.java | 2 +-
.../camel/component/aws/s3/S3Producer.java | 174 ++++++++++++++++---
3 files changed, 171 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/aa73efb1/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
index 8cd52e8..42faf71 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
@@ -20,32 +20,49 @@ import com.amazonaws.services.s3.AmazonS3;
/**
* The AWS S3 component configuration properties
- *
*/
public class S3Configuration implements Cloneable {
private String accessKey;
private String secretKey;
private AmazonS3 amazonS3Client;
-
+
private String bucketName;
private String fileName;
private String prefix;
private String region;
private boolean deleteAfterRead = true;
private boolean deleteAfterWrite;
+ private boolean multiPartUpload;
+ private long partSize = 25 * 1024 * 1024;
private String amazonS3Endpoint;
private String policy;
private String storageClass;
-
+
+ public long getPartSize() {
+ return partSize;
+ }
+
+ public void setPartSize(long partSize) {
+ this.partSize = partSize;
+ }
+
+ public boolean isMultiPartUpload() {
+ return multiPartUpload;
+ }
+
+ public void setMultiPartUpload(boolean multiPartUpload) {
+ this.multiPartUpload = multiPartUpload;
+ }
+
public void setAmazonS3Endpoint(String amazonS3Endpoint) {
this.amazonS3Endpoint = amazonS3Endpoint;
}
-
+
public String getAmazonS3Endpoint() {
return amazonS3Endpoint;
}
-
+
public String getAccessKey() {
return accessKey;
}
@@ -61,7 +78,7 @@ public class S3Configuration implements Cloneable {
public void setSecretKey(String secretKey) {
this.secretKey = secretKey;
}
-
+
public AmazonS3 getAmazonS3Client() {
return amazonS3Client;
}
@@ -118,7 +135,7 @@ public class S3Configuration implements Cloneable {
public void setDeleteAfterWrite(boolean deleteAfterWrite) {
this.deleteAfterWrite = deleteAfterWrite;
}
-
+
public String getPolicy() {
return policy;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/aa73efb1/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Endpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Endpoint.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Endpoint.java
index f150bf2..e960c8a 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Endpoint.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Endpoint.java
@@ -96,7 +96,7 @@ public class S3Endpoint extends ScheduledPollEndpoint {
}
String bucketName = getConfiguration().getBucketName();
- LOG.trace("Quering whether bucket [{}] already exists...", bucketName);
+ LOG.trace("Querying whether bucket [{}] already exists...", bucketName);
try {
s3Client.listObjects(new ListObjectsRequest(bucketName, null, null, null, 0));
http://git-wip-us.apache.org/repos/asf/camel/blob/aa73efb1/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
----------------------------------------------------------------------
diff --git a/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java b/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
index 825af36..38db307 100644
--- a/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
+++ b/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
@@ -18,19 +18,31 @@ package org.apache.camel.component.aws.s3;
import java.io.File;
import java.io.InputStream;
+import java.util.ArrayList;
import java.util.Date;
+import java.util.List;
import java.util.Map;
+import com.amazonaws.services.cloudfront.model.InvalidArgumentException;
+import com.amazonaws.services.s3.model.AbortMultipartUploadRequest;
import com.amazonaws.services.s3.model.AccessControlList;
import com.amazonaws.services.s3.model.CannedAccessControlList;
+import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
+import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
+import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
+import com.amazonaws.services.s3.model.InitiateMultipartUploadResult;
import com.amazonaws.services.s3.model.ObjectMetadata;
+import com.amazonaws.services.s3.model.PartETag;
import com.amazonaws.services.s3.model.PutObjectRequest;
import com.amazonaws.services.s3.model.PutObjectResult;
+import com.amazonaws.services.s3.model.StorageClass;
+import com.amazonaws.services.s3.model.UploadPartRequest;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.impl.DefaultProducer;
+import org.apache.camel.util.CastUtils;
import org.apache.camel.util.FileUtil;
import org.apache.camel.util.IOHelper;
import org.apache.camel.util.URISupport;
@@ -49,55 +61,114 @@ public class S3Producer extends DefaultProducer {
super(endpoint);
}
+
@Override
public void process(final Exchange exchange) throws Exception {
- ObjectMetadata objectMetadata = new ObjectMetadata();
+ if (getConfiguration().isMultiPartUpload()) {
+ processMultiPart(exchange);
+ } else {
+ processSingleOp(exchange);
+ }
- Long contentLength = exchange.getIn().getHeader(S3Constants.CONTENT_LENGTH, Long.class);
- if (contentLength != null) {
- objectMetadata.setContentLength(contentLength);
+
+ }
+
+ public void processMultiPart(final Exchange exchange) throws Exception {
+ File filePayload = null;
+ final Object obj = exchange.getIn().getMandatoryBody();
+ if (obj instanceof File) {
+ filePayload = (File) obj;
+ } else {
+ throw new InvalidArgumentException("aws-s3: MultiPart upload requires a File input.");
}
- String contentType = exchange.getIn().getHeader(S3Constants.CONTENT_TYPE, String.class);
- if (contentType != null) {
- objectMetadata.setContentType(contentType);
+ ObjectMetadata objectMetadata = determineMetadata(exchange);
+ if (objectMetadata.getContentLength() == 0) {
+ objectMetadata.setContentLength(filePayload.length());
}
- String cacheControl = exchange.getIn().getHeader(S3Constants.CACHE_CONTROL, String.class);
- if (cacheControl != null) {
- objectMetadata.setCacheControl(cacheControl);
+ final String keyName = determineKey(exchange);
+ final InitiateMultipartUploadRequest initRequest = new InitiateMultipartUploadRequest(getConfiguration().getBucketName(),
+ keyName, objectMetadata);
+
+ String storageClass = determineStorageClass(exchange);
+ if (storageClass != null) {
+ initRequest.setStorageClass(StorageClass.fromValue(storageClass));
}
- String contentDisposition = exchange.getIn().getHeader(S3Constants.CONTENT_DISPOSITION, String.class);
- if (contentDisposition != null) {
- objectMetadata.setContentDisposition(contentDisposition);
+ String cannedAcl = exchange.getIn().getHeader(S3Constants.CANNED_ACL, String.class);
+ if (cannedAcl != null) {
+ CannedAccessControlList objectAcl = CannedAccessControlList.valueOf(cannedAcl);
+ initRequest.setCannedACL(objectAcl);
}
- String contentEncoding = exchange.getIn().getHeader(S3Constants.CONTENT_ENCODING, String.class);
- if (contentEncoding != null) {
- objectMetadata.setContentEncoding(contentEncoding);
+ AccessControlList acl = exchange.getIn().getHeader(S3Constants.ACL, AccessControlList.class);
+ if (acl != null) {
+ // note: if cannedacl and acl are both specified the last one will be used. refer to
+ // PutObjectRequest#setAccessControlList for more details
+ initRequest.setAccessControlList(acl);
}
- String contentMD5 = exchange.getIn().getHeader(S3Constants.CONTENT_MD5, String.class);
- if (contentMD5 != null) {
- objectMetadata.setContentMD5(contentMD5);
+ LOG.trace("Initiating multipart upload [{}] from exchange [{}]...", initRequest, exchange);
+
+ final InitiateMultipartUploadResult initResponse = getEndpoint().getS3Client().initiateMultipartUpload(initRequest);
+ final long contentLength = objectMetadata.getContentLength();
+ final List<PartETag> partETags = new ArrayList<PartETag>();
+ long partSize = getConfiguration().getPartSize();
+ CompleteMultipartUploadResult uploadResult = null;
+
+ long filePosition = 0;
+
+
+ try {
+ for (int part = 1; filePosition < contentLength; part++) {
+ partSize = Math.min(partSize, contentLength - filePosition);
+
+ UploadPartRequest uploadRequest = new UploadPartRequest()
+ .withBucketName(getConfiguration().getBucketName()).withKey(keyName)
+ .withUploadId(initResponse.getUploadId()).withPartNumber(part)
+ .withFileOffset(filePosition)
+ .withFile(filePayload)
+ .withPartSize(partSize);
+
+ LOG.trace("Uploading part [{}] for {}", part, keyName);
+ partETags.add(getEndpoint().getS3Client().uploadPart(uploadRequest).getPartETag());
+
+ filePosition += partSize;
+ }
+ CompleteMultipartUploadRequest compRequest = new
+ CompleteMultipartUploadRequest(getConfiguration().getBucketName(),
+ keyName,
+ initResponse.getUploadId(),
+ partETags);
+
+ uploadResult = getEndpoint().getS3Client().completeMultipartUpload(compRequest);
+
+ } catch (Exception e) {
+ getEndpoint().getS3Client().abortMultipartUpload(new AbortMultipartUploadRequest(
+ getConfiguration().getBucketName(), keyName, initResponse.getUploadId()));
+ throw e;
}
- Date lastModified = exchange.getIn().getHeader(S3Constants.LAST_MODIFIED, Date.class);
- if (lastModified != null) {
- objectMetadata.setLastModified(lastModified);
+ Message message = getMessageForResponse(exchange);
+ message.setHeader(S3Constants.E_TAG, uploadResult.getETag());
+ if (uploadResult.getVersionId() != null) {
+ message.setHeader(S3Constants.VERSION_ID, uploadResult.getVersionId());
}
- Map<String, String> userMetadata = exchange.getIn().getHeader(S3Constants.USER_METADATA, Map.class);
- if (userMetadata != null) {
- objectMetadata.setUserMetadata(userMetadata);
+ if (getConfiguration().isDeleteAfterWrite() && filePayload != null) {
+ FileUtil.deleteFile(filePayload);
}
+ }
- File filePayload = null;
+ public void processSingleOp(final Exchange exchange) throws Exception {
+ ObjectMetadata objectMetadata = determineMetadata(exchange);
+
+ File filePayload = null;
Object obj = exchange.getIn().getMandatoryBody();
if (obj instanceof File) {
- filePayload = (File)obj;
+ filePayload = (File) obj;
}
PutObjectRequest putObjectRequest = new PutObjectRequest(getConfiguration().getBucketName(),
determineKey(exchange), exchange.getIn().getMandatoryBody(InputStream.class), objectMetadata);
@@ -137,6 +208,52 @@ public class S3Producer extends DefaultProducer {
}
}
+ private ObjectMetadata determineMetadata(final Exchange exchange) {
+ ObjectMetadata objectMetadata = new ObjectMetadata();
+
+ Long contentLength = exchange.getIn().getHeader(S3Constants.CONTENT_LENGTH, Long.class);
+ if (contentLength != null) {
+ objectMetadata.setContentLength(contentLength);
+ }
+
+ String contentType = exchange.getIn().getHeader(S3Constants.CONTENT_TYPE, String.class);
+ if (contentType != null) {
+ objectMetadata.setContentType(contentType);
+ }
+
+ String cacheControl = exchange.getIn().getHeader(S3Constants.CACHE_CONTROL, String.class);
+ if (cacheControl != null) {
+ objectMetadata.setCacheControl(cacheControl);
+ }
+
+ String contentDisposition = exchange.getIn().getHeader(S3Constants.CONTENT_DISPOSITION, String.class);
+ if (contentDisposition != null) {
+ objectMetadata.setContentDisposition(contentDisposition);
+ }
+
+ String contentEncoding = exchange.getIn().getHeader(S3Constants.CONTENT_ENCODING, String.class);
+ if (contentEncoding != null) {
+ objectMetadata.setContentEncoding(contentEncoding);
+ }
+
+ String contentMD5 = exchange.getIn().getHeader(S3Constants.CONTENT_MD5, String.class);
+ if (contentMD5 != null) {
+ objectMetadata.setContentMD5(contentMD5);
+ }
+
+ Date lastModified = exchange.getIn().getHeader(S3Constants.LAST_MODIFIED, Date.class);
+ if (lastModified != null) {
+ objectMetadata.setLastModified(lastModified);
+ }
+
+ Map<String, String> userMetadata = CastUtils.cast(exchange.getIn().getHeader(S3Constants.USER_METADATA, Map.class));
+ if (userMetadata != null) {
+ objectMetadata.setUserMetadata(userMetadata);
+ }
+
+ return objectMetadata;
+ }
+
private String determineKey(final Exchange exchange) {
String key = exchange.getIn().getHeader(S3Constants.KEY, String.class);
if (key == null) {
@@ -177,4 +294,5 @@ public class S3Producer extends DefaultProducer {
public S3Endpoint getEndpoint() {
return (S3Endpoint) super.getEndpoint();
}
-}
\ No newline at end of file
+
+}