You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by cm...@apache.org on 2011/11/24 23:55:27 UTC
svn commit: r1206010 - in
/camel/branches/camel-2.8.x/components/camel-aws/src:
main/java/org/apache/camel/component/aws/s3/
test/java/org/apache/camel/component/aws/s3/
Author: cmueller
Date: Thu Nov 24 22:55:26 2011
New Revision: 1206010
URL: http://svn.apache.org/viewvc?rev=1206010&view=rev
Log:
CAMEL-4267: Camel-AWS : Add support for Storage Class / Reduced Redundancy support
Thank you Bilgin for the patch
Modified:
camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Constants.java
camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/AmazonS3ClientMock.java
camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentConfigurationTest.java
camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentExistingBucketTest.java
camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentNonExistingBucketTest.java
Modified: camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java?rev=1206010&r1=1206009&r2=1206010&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java (original)
+++ camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java Thu Nov 24 22:55:26 2011
@@ -33,6 +33,7 @@ public class S3Configuration implements
private boolean deleteAfterRead = true;
private String amazonS3Endpoint;
private String policy;
+ private String storageClass;
public void setAmazonS3Endpoint(String amazonS3Endpoint) {
this.amazonS3Endpoint = amazonS3Endpoint;
@@ -97,4 +98,12 @@ public class S3Configuration implements
public void setPolicy(String policy) {
this.policy = policy;
}
+
+ public String getStorageClass() {
+ return storageClass;
+ }
+
+ public void setStorageClass(String storageClass) {
+ this.storageClass = storageClass;
+ }
}
\ No newline at end of file
Modified: camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Constants.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Constants.java?rev=1206010&r1=1206009&r2=1206010&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Constants.java (original)
+++ camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Constants.java Thu Nov 24 22:55:26 2011
@@ -32,5 +32,6 @@ public interface S3Constants {
String E_TAG = "CamelAwsS3ETag";
String KEY = "CamelAwsS3Key";
String LAST_MODIFIED = "CamelAwsS3LastModified";
+ String STORAGE_CLASS = "CamelAwsS3StorageClass";
String VERSION_ID = "CamelAwsS3VersionId";
}
\ No newline at end of file
Modified: camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java?rev=1206010&r1=1206009&r2=1206010&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java (original)
+++ camel/branches/camel-2.8.x/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java Thu Nov 24 22:55:26 2011
@@ -86,7 +86,12 @@ public class S3Producer extends DefaultP
determineKey(exchange),
exchange.getIn().getMandatoryBody(InputStream.class),
objectMetadata);
-
+
+ String storageClass = determineStorageClass(exchange);
+ if (storageClass != null) {
+ putObjectRequest.setStorageClass(storageClass);
+ }
+
LOG.trace("Put object [{}] from exchange [{}]...", putObjectRequest, exchange);
PutObjectResult putObjectResult = getEndpoint().getS3Client().putObject(putObjectRequest);
@@ -107,6 +112,15 @@ public class S3Producer extends DefaultP
}
return key;
}
+
+ private String determineStorageClass(Exchange exchange) {
+ String storageClass = exchange.getIn().getHeader(S3Constants.STORAGE_CLASS, String.class);
+ if (storageClass == null) {
+ storageClass = getConfiguration().getStorageClass();
+ }
+
+ return storageClass;
+ }
private Message getMessageForResponse(Exchange exchange) {
if (exchange.getPattern().isOutCapable()) {
Modified: camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/AmazonS3ClientMock.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/AmazonS3ClientMock.java?rev=1206010&r1=1206009&r2=1206010&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/AmazonS3ClientMock.java (original)
+++ camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/AmazonS3ClientMock.java Thu Nov 24 22:55:26 2011
@@ -79,6 +79,7 @@ import com.amazonaws.services.s3.model.V
public class AmazonS3ClientMock extends AmazonS3Client {
List<S3Object> objects = new ArrayList<S3Object>();
+ List<PutObjectRequest> putObjectRequests = new ArrayList<PutObjectRequest>();
private boolean nonExistingBucketCreated;
@@ -300,6 +301,8 @@ public class AmazonS3ClientMock extends
@Override
public PutObjectResult putObject(PutObjectRequest putObjectRequest) throws AmazonClientException, AmazonServiceException {
+ putObjectRequests.add(putObjectRequest);
+
S3Object s3Object = new S3Object();
s3Object.setBucketName(putObjectRequest.getBucketName());
s3Object.setKey(putObjectRequest.getKey());
Modified: camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentConfigurationTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentConfigurationTest.java?rev=1206010&r1=1206009&r2=1206010&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentConfigurationTest.java (original)
+++ camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentConfigurationTest.java Thu Nov 24 22:55:26 2011
@@ -58,14 +58,14 @@ public class S3ComponentConfigurationTes
assertNull(endpoint.getConfiguration().getAmazonS3Endpoint());
assertNull(endpoint.getConfiguration().getPolicy());
}
-
+
@Test
public void createEndpointWithMaximalConfiguration() throws Exception {
S3Component component = new S3Component(context);
S3Endpoint endpoint = (S3Endpoint) component.createEndpoint("aws-s3://MyBucket?amazonS3Endpoint=sns.eu-west-1.amazonaws.com"
+ "&accessKey=xxx&secretKey=yyy®ion=us-west-1&deleteAfterRead=false&maxMessagesPerPoll=1&policy=%7B%22Version%22%3A%222008-10-17%22,%22Id%22%3A%22Policy4324355464%22,"
+ "%22Statement%22%3A%5B%7B%22Sid%22%3A%22Stmt456464646477%22,%22Action%22%3A%5B%22s3%3AGetObject%22%5D,%22Effect%22%3A%22Allow%22,"
- + "%22Resource%22%3A%5B%22arn%3Aaws%3As3%3A%3A%3Amybucket/some/path/*%22%5D,%22Principal%22%3A%7B%22AWS%22%3A%5B%22*%22%5D%7D%7D%5D%7D");
+ + "%22Resource%22%3A%5B%22arn%3Aaws%3As3%3A%3A%3Amybucket/some/path/*%22%5D,%22Principal%22%3A%7B%22AWS%22%3A%5B%22*%22%5D%7D%7D%5D%7D&storageClass=REDUCED_REDUNDANCY");
assertEquals("MyBucket", endpoint.getConfiguration().getBucketName());
assertEquals("xxx", endpoint.getConfiguration().getAccessKey());
@@ -77,6 +77,7 @@ public class S3ComponentConfigurationTes
assertEquals("sns.eu-west-1.amazonaws.com", endpoint.getConfiguration().getAmazonS3Endpoint());
assertEquals("{\"Version\":\"2008-10-17\",\"Id\":\"Policy4324355464\",\"Statement\":[{\"Sid\":\"Stmt456464646477\",\"Action\":[\"s3:GetObject\"],\"Effect\":\"Allow\",\"Resource\":"
+ "[\"arn:aws:s3:::mybucket/some/path/*\"],\"Principal\":{\"AWS\":[\"*\"]}}]}", endpoint.getConfiguration().getPolicy());
+ assertEquals("REDUCED_REDUNDANCY", endpoint.getConfiguration().getStorageClass());
}
@Test(expected = IllegalArgumentException.class)
Modified: camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentExistingBucketTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentExistingBucketTest.java?rev=1206010&r1=1206009&r2=1206010&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentExistingBucketTest.java (original)
+++ camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentExistingBucketTest.java Thu Nov 24 22:55:26 2011
@@ -17,6 +17,9 @@
package org.apache.camel.component.aws.s3;
import java.io.InputStream;
+import java.util.Date;
+
+import com.amazonaws.services.s3.model.PutObjectRequest;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
@@ -38,6 +41,8 @@ public class S3ComponentExistingBucketTe
@EndpointInject(uri = "mock:result")
private MockEndpoint result;
+ private AmazonS3ClientMock client;
+
@Test
public void sendIn() throws Exception {
result.expectedMessageCount(1);
@@ -53,6 +58,10 @@ public class S3ComponentExistingBucketTe
assertResultExchange(result.getExchanges().get(0));
+ PutObjectRequest putObjectRequest = client.putObjectRequests.get(0);
+ assertEquals("REDUCED_REDUNDANCY", putObjectRequest.getStorageClass());
+ assertEquals("mycamelbucket", putObjectRequest.getBucketName());
+
assertResponseMessage(exchange.getIn());
}
@@ -71,9 +80,52 @@ public class S3ComponentExistingBucketTe
assertResultExchange(result.getExchanges().get(0));
+ PutObjectRequest putObjectRequest = client.putObjectRequests.get(0);
+ assertEquals("REDUCED_REDUNDANCY", putObjectRequest.getStorageClass());
+ assertEquals("mycamelbucket", putObjectRequest.getBucketName());
+
assertResponseMessage(exchange.getOut());
}
+ @Test
+ public void sendCustomHeaderValues() throws Exception {
+ result.expectedMessageCount(1);
+ final Date now = new Date();
+
+ Exchange exchange = template.send("direct:start", ExchangePattern.InOnly, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(S3Constants.STORAGE_CLASS, "STANDARD");
+ exchange.getIn().setHeader(S3Constants.KEY, "CamelUnitTest");
+ exchange.getIn().setHeader(S3Constants.CONTENT_LENGTH, 2L);
+ exchange.getIn().setHeader(S3Constants.CONTENT_TYPE, "text/html");
+ exchange.getIn().setHeader(S3Constants.CACHE_CONTROL, "no-cache");
+ exchange.getIn().setHeader(S3Constants.CONTENT_DISPOSITION, "attachment;");
+ exchange.getIn().setHeader(S3Constants.CONTENT_ENCODING, "gzip");
+ exchange.getIn().setHeader(S3Constants.CONTENT_MD5, "TWF");
+ exchange.getIn().setHeader(S3Constants.LAST_MODIFIED, now);
+
+ exchange.getIn().setBody("This is my bucket content.");
+ }
+ });
+
+ assertMockEndpointsSatisfied();
+
+ assertResultExchange(result.getExchanges().get(0));
+
+ PutObjectRequest putObjectRequest = client.putObjectRequests.get(0);
+ assertEquals("STANDARD", putObjectRequest.getStorageClass());
+ assertEquals("mycamelbucket", putObjectRequest.getBucketName());
+ assertEquals(2L, putObjectRequest.getMetadata().getContentLength());
+ assertEquals("text/html", putObjectRequest.getMetadata().getContentType());
+ assertEquals("no-cache", putObjectRequest.getMetadata().getCacheControl());
+ assertEquals("attachment;", putObjectRequest.getMetadata().getContentDisposition());
+ assertEquals("gzip", putObjectRequest.getMetadata().getContentEncoding());
+ assertEquals("TWF", putObjectRequest.getMetadata().getContentMD5());
+ assertEquals(now, putObjectRequest.getMetadata().getLastModified());
+
+ assertResponseMessage(exchange.getIn());
+ }
+
private void assertResultExchange(Exchange resultExchange) {
assertIsInstanceOf(InputStream.class, resultExchange.getIn().getBody());
assertEquals("This is my bucket content.", resultExchange.getIn().getBody(String.class));
@@ -98,7 +150,9 @@ public class S3ComponentExistingBucketTe
@Override
protected JndiRegistry createRegistry() throws Exception {
JndiRegistry registry = super.createRegistry();
- registry.bind("amazonS3Client", new AmazonS3ClientMock());
+
+ client = new AmazonS3ClientMock();
+ registry.bind("amazonS3Client", client);
return registry;
}
@@ -111,7 +165,7 @@ public class S3ComponentExistingBucketTe
String awsEndpoint = "aws-s3://mycamelbucket?amazonS3Client=#amazonS3Client®ion=us-west-1";
from("direct:start")
- .to(awsEndpoint);
+ .to(awsEndpoint + "&storageClass=REDUCED_REDUNDANCY");
from(awsEndpoint + "&maxMessagesPerPoll=5")
.to("mock:result");
Modified: camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentNonExistingBucketTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentNonExistingBucketTest.java?rev=1206010&r1=1206009&r2=1206010&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentNonExistingBucketTest.java (original)
+++ camel/branches/camel-2.8.x/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentNonExistingBucketTest.java Thu Nov 24 22:55:26 2011
@@ -17,6 +17,9 @@
package org.apache.camel.component.aws.s3;
import java.io.InputStream;
+import java.util.Date;
+
+import com.amazonaws.services.s3.model.PutObjectRequest;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
@@ -38,6 +41,8 @@ public class S3ComponentNonExistingBucke
@EndpointInject(uri = "mock:result")
private MockEndpoint result;
+ private AmazonS3ClientMock client;
+
@Test
public void sendInOnly() throws Exception {
result.expectedMessageCount(1);
@@ -53,6 +58,10 @@ public class S3ComponentNonExistingBucke
assertResultExchange(result.getExchanges().get(0));
+ PutObjectRequest putObjectRequest = client.putObjectRequests.get(0);
+ assertEquals("REDUCED_REDUNDANCY", putObjectRequest.getStorageClass());
+ assertEquals("nonExistingBucket", putObjectRequest.getBucketName());
+
assertResponseMessage(exchange.getIn());
}
@@ -71,9 +80,52 @@ public class S3ComponentNonExistingBucke
assertResultExchange(result.getExchanges().get(0));
+ PutObjectRequest putObjectRequest = client.putObjectRequests.get(0);
+ assertEquals("REDUCED_REDUNDANCY", putObjectRequest.getStorageClass());
+ assertEquals("nonExistingBucket", putObjectRequest.getBucketName());
+
assertResponseMessage(exchange.getOut());
}
+ @Test
+ public void sendCustomHeaderValues() throws Exception {
+ result.expectedMessageCount(1);
+ final Date now = new Date();
+
+ Exchange exchange = template.send("direct:start", ExchangePattern.InOnly, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(S3Constants.STORAGE_CLASS, "STANDARD");
+ exchange.getIn().setHeader(S3Constants.KEY, "CamelUnitTest");
+ exchange.getIn().setHeader(S3Constants.CONTENT_LENGTH, 2L);
+ exchange.getIn().setHeader(S3Constants.CONTENT_TYPE, "text/html");
+ exchange.getIn().setHeader(S3Constants.CACHE_CONTROL, "no-cache");
+ exchange.getIn().setHeader(S3Constants.CONTENT_DISPOSITION, "attachment;");
+ exchange.getIn().setHeader(S3Constants.CONTENT_ENCODING, "gzip");
+ exchange.getIn().setHeader(S3Constants.CONTENT_MD5, "TWF");
+ exchange.getIn().setHeader(S3Constants.LAST_MODIFIED, now);
+
+ exchange.getIn().setBody("This is my bucket content.");
+ }
+ });
+
+ assertMockEndpointsSatisfied();
+
+ assertResultExchange(result.getExchanges().get(0));
+
+ PutObjectRequest putObjectRequest = client.putObjectRequests.get(0);
+ assertEquals("STANDARD", putObjectRequest.getStorageClass());
+ assertEquals("nonExistingBucket", putObjectRequest.getBucketName());
+ assertEquals(2L, putObjectRequest.getMetadata().getContentLength());
+ assertEquals("text/html", putObjectRequest.getMetadata().getContentType());
+ assertEquals("no-cache", putObjectRequest.getMetadata().getCacheControl());
+ assertEquals("attachment;", putObjectRequest.getMetadata().getContentDisposition());
+ assertEquals("gzip", putObjectRequest.getMetadata().getContentEncoding());
+ assertEquals("TWF", putObjectRequest.getMetadata().getContentMD5());
+ assertEquals(now, putObjectRequest.getMetadata().getLastModified());
+
+ assertResponseMessage(exchange.getIn());
+ }
+
private void assertResultExchange(Exchange resultExchange) {
assertIsInstanceOf(InputStream.class, resultExchange.getIn().getBody());
assertEquals("This is my bucket content.", resultExchange.getIn().getBody(String.class));
@@ -98,7 +150,9 @@ public class S3ComponentNonExistingBucke
@Override
protected JndiRegistry createRegistry() throws Exception {
JndiRegistry registry = super.createRegistry();
- registry.bind("amazonS3Client", new AmazonS3ClientMock());
+
+ client = new AmazonS3ClientMock();
+ registry.bind("amazonS3Client", client);
return registry;
}
@@ -111,7 +165,7 @@ public class S3ComponentNonExistingBucke
String awsEndpoint = "aws-s3://nonExistingBucket?amazonS3Client=#amazonS3Client®ion=us-west-1&policy=xxx";
from("direct:start")
- .to(awsEndpoint);
+ .to(awsEndpoint + "&storageClass=REDUCED_REDUNDANCY");
from(awsEndpoint + "&maxMessagesPerPoll=5")
.to("mock:result");