Author: ningjiang
Date: Fri Mar 29 07:49:03 2013
New Revision: 1462424
URL: http://svn.apache.org/r1462424
Log:
CAMEL-6225 Add support for deleting the local file after uploading to S3 with
thanks to Chris
Added:
camel/trunk/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentFileTest.java
Modified:
camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
Modified:
camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java?rev=1462424&r1=1462423&r2=1462424&view=diff
==============================================================================
---
camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
(original)
+++
camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Configuration.java
Fri Mar 29 07:49:03 2013
@@ -32,6 +32,7 @@ public class S3Configuration implements
private String prefix;
private String region;
private boolean deleteAfterRead = true;
+ private boolean deleteAfterWrite;
private String amazonS3Endpoint;
private String policy;
private String storageClass;
@@ -99,6 +100,14 @@ public class S3Configuration implements
public void setDeleteAfterRead(boolean deleteAfterRead) {
this.deleteAfterRead = deleteAfterRead;
}
+
+ public boolean isDeleteAfterWrite() {
+ return deleteAfterWrite;
+ }
+
+ public void setDeleteAfterWrite(boolean deleteAfterWrite) {
+ this.deleteAfterWrite = deleteAfterWrite;
+ }
public String getPolicy() {
return policy;
Modified:
camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java?rev=1462424&r1=1462423&r2=1462424&view=diff
==============================================================================
---
camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
(original)
+++
camel/trunk/components/camel-aws/src/main/java/org/apache/camel/component/aws/s3/S3Producer.java
Fri Mar 29 07:49:03 2013
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.aws.s3;
+import java.io.File;
import java.io.InputStream;
import java.util.Date;
import java.util.Map;
@@ -28,8 +29,10 @@ import com.amazonaws.services.s3.model.P
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.impl.DefaultProducer;
+import org.apache.camel.util.FileUtil;
import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -90,6 +93,12 @@ public class S3Producer extends DefaultP
objectMetadata.setUserMetadata(userMetadata);
}
+ File filePayload = null;
+
+ Object obj = exchange.getIn().getMandatoryBody();
+ if (obj instanceof File) {
+ filePayload = (File)obj;
+ }
PutObjectRequest putObjectRequest = new
PutObjectRequest(getConfiguration().getBucketName(),
determineKey(exchange),
exchange.getIn().getMandatoryBody(InputStream.class), objectMetadata);
@@ -121,6 +130,10 @@ public class S3Producer extends DefaultP
if (putObjectResult.getVersionId() != null) {
message.setHeader(S3Constants.VERSION_ID,
putObjectResult.getVersionId());
}
+
+ if (getConfiguration().isDeleteAfterWrite() && filePayload != null) {
+ FileUtil.deleteFile(filePayload);
+ }
}
private String determineKey(final Exchange exchange) {
Added:
camel/trunk/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentFileTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentFileTest.java?rev=1462424&view=auto
==============================================================================
---
camel/trunk/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentFileTest.java
(added)
+++
camel/trunk/components/camel-aws/src/test/java/org/apache/camel/component/aws/s3/S3ComponentFileTest.java
Fri Mar 29 07:49:03 2013
@@ -0,0 +1,169 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws.s3;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.InputStream;
+
+import com.amazonaws.services.s3.model.PutObjectRequest;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.util.FileUtil;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class S3ComponentFileTest extends CamelTestSupport {
+
+ @EndpointInject(uri = "direct:startKeep")
+ private ProducerTemplate templateKeep;
+
+ @EndpointInject(uri = "direct:startDelete")
+ private ProducerTemplate templateDelete;
+
+ @EndpointInject(uri = "mock:result")
+ private MockEndpoint result;
+
+ private AmazonS3ClientMock client;
+
+ private File testFile;
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ testFile = FileUtil.createTempFile("test", "file");
+
+ FileWriter writer = new FileWriter(testFile);
+ writer.write("This is my bucket content.");
+ writer.close();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ super.tearDown();
+
+ FileUtil.deleteFile(testFile);
+ }
+
+ @Test
+ public void sendFileAndKeep() throws Exception {
+ result.expectedMessageCount(1);
+
+ Exchange exchange = templateKeep.send("direct:startKeep",
ExchangePattern.InOnly, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(S3Constants.KEY, "CamelUnitTest");
+ exchange.getIn().setBody(testFile);
+ }
+ });
+
+ assertMockEndpointsSatisfied();
+
+ assertResultExchange(result.getExchanges().get(0));
+
+ PutObjectRequest putObjectRequest = client.putObjectRequests.get(0);
+ assertEquals("mycamelbucket", putObjectRequest.getBucketName());
+
+ assertResponseMessage(exchange.getIn());
+
+ assertFileExists(testFile.getAbsolutePath());
+ }
+
+ @Test
+ public void sendFileAndDelete() throws Exception {
+ result.expectedMessageCount(1);
+
+ Exchange exchange = templateDelete.send("direct:startDelete",
ExchangePattern.InOnly, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(S3Constants.KEY, "CamelUnitTest");
+ exchange.getIn().setBody(testFile);
+ }
+ });
+
+ assertMockEndpointsSatisfied();
+
+ assertResultExchange(result.getExchanges().get(0));
+
+ PutObjectRequest putObjectRequest = client.putObjectRequests.get(0);
+ assertEquals("mycamelbucket", putObjectRequest.getBucketName());
+
+ assertResponseMessage(exchange.getIn());
+
+ assertFileNotExists(testFile.getAbsolutePath());
+ }
+
+ private void assertResultExchange(Exchange resultExchange) {
+ assertIsInstanceOf(InputStream.class,
resultExchange.getIn().getBody());
+ assertEquals("This is my bucket content.",
resultExchange.getIn().getBody(String.class));
+ assertEquals("mycamelbucket",
resultExchange.getIn().getHeader(S3Constants.BUCKET_NAME));
+ assertEquals("CamelUnitTest",
resultExchange.getIn().getHeader(S3Constants.KEY));
+ assertNull(resultExchange.getIn().getHeader(S3Constants.VERSION_ID));
// not enabled on this bucket
+
assertNull(resultExchange.getIn().getHeader(S3Constants.LAST_MODIFIED));
+ assertNull(resultExchange.getIn().getHeader(S3Constants.E_TAG));
+ assertNull(resultExchange.getIn().getHeader(S3Constants.CONTENT_TYPE));
+
assertNull(resultExchange.getIn().getHeader(S3Constants.CONTENT_ENCODING));
+ assertEquals(0L,
resultExchange.getIn().getHeader(S3Constants.CONTENT_LENGTH));
+
assertNull(resultExchange.getIn().getHeader(S3Constants.CONTENT_DISPOSITION));
+ assertNull(resultExchange.getIn().getHeader(S3Constants.CONTENT_MD5));
+
assertNull(resultExchange.getIn().getHeader(S3Constants.CACHE_CONTROL));
+
assertNull(resultExchange.getIn().getHeader(S3Constants.USER_METADATA));
+ }
+
+ private void assertResponseMessage(Message message) {
+ assertEquals("3a5c8b1ad448bca04584ecb55b836264",
message.getHeader(S3Constants.E_TAG));
+ assertNull(message.getHeader(S3Constants.VERSION_ID));
+ }
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry registry = super.createRegistry();
+
+ client = new AmazonS3ClientMock();
+ registry.bind("amazonS3Client", client);
+
+ return registry;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ String awsEndpoint =
"aws-s3://mycamelbucket?amazonS3Client=#amazonS3Client®ion=us-west-1";
+
+ from("direct:startKeep")
+ .to(awsEndpoint + "&deleteAfterWrite=false");
+
+ from("direct:startDelete")
+ .to(awsEndpoint + "&deleteAfterWrite=true");
+
+ from(awsEndpoint + "&maxMessagesPerPoll=5")
+ .to("mock:result");
+ }
+ };
+ }
+}
