You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2021/04/06 06:40:04 UTC
[camel] 03/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit fb64227308f57b0f500b2aef65e7c0c051db58c3
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Apr 6 06:55:38 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../aws2/s3/AWS2S3ComponentConfigurer.java | 9 ++-------
.../aws2/s3/AWS2S3EndpointConfigurer.java | 9 ++-------
.../component/aws2/s3/AWS2S3Configuration.java | 23 +++++++++++++++++-----
.../camel/component/aws2/s3/AWS2S3Endpoint.java | 2 +-
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 16 ++++++++++-----
.../S3StreamUploadOperationLocalstackTest.java | 2 +-
.../S3StreamUploadTimeoutLocalstackTest.java | 2 +-
7 files changed, 36 insertions(+), 27 deletions(-)
diff --git a/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3ComponentConfigurer.java b/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3ComponentConfigurer.java
index 182f1b9..89a4c80 100644
--- a/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3ComponentConfigurer.java
+++ b/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3ComponentConfigurer.java
@@ -1,14 +1,9 @@
/* Generated by camel build tools - do NOT edit this file! */
package org.apache.camel.component.aws2.s3;
-import java.util.Map;
-
import org.apache.camel.CamelContext;
-import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
import org.apache.camel.spi.PropertyConfigurerGetter;
-import org.apache.camel.spi.ConfigurerStrategy;
import org.apache.camel.spi.GeneratedPropertyConfigurer;
-import org.apache.camel.util.CaseInsensitiveMap;
import org.apache.camel.support.component.PropertyConfigurerSupport;
/**
@@ -105,7 +100,7 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "storageclass":
case "storageClass": getOrCreateConfiguration(target).setStorageClass(property(camelContext, java.lang.String.class, value)); return true;
case "streammode":
- case "streamMode": getOrCreateConfiguration(target).setStreamMode(property(camelContext, boolean.class, value)); return true;
+ case "streamMode": getOrCreateConfiguration(target).setStreamingUploadMode(property(camelContext, boolean.class, value)); return true;
case "trustallcertificates":
case "trustAllCertificates": getOrCreateConfiguration(target).setTrustAllCertificates(property(camelContext, boolean.class, value)); return true;
case "uriendpointoverride":
@@ -301,7 +296,7 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "storageclass":
case "storageClass": return getOrCreateConfiguration(target).getStorageClass();
case "streammode":
- case "streamMode": return getOrCreateConfiguration(target).isStreamMode();
+ case "streamMode": return getOrCreateConfiguration(target).isStreamingUploadMode();
case "trustallcertificates":
case "trustAllCertificates": return getOrCreateConfiguration(target).isTrustAllCertificates();
case "uriendpointoverride":
diff --git a/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3EndpointConfigurer.java b/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3EndpointConfigurer.java
index 19f618f..bd98c61 100644
--- a/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3EndpointConfigurer.java
+++ b/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3EndpointConfigurer.java
@@ -1,14 +1,9 @@
/* Generated by camel build tools - do NOT edit this file! */
package org.apache.camel.component.aws2.s3;
-import java.util.Map;
-
import org.apache.camel.CamelContext;
-import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
import org.apache.camel.spi.PropertyConfigurerGetter;
-import org.apache.camel.spi.ConfigurerStrategy;
import org.apache.camel.spi.GeneratedPropertyConfigurer;
-import org.apache.camel.util.CaseInsensitiveMap;
import org.apache.camel.support.component.PropertyConfigurerSupport;
/**
@@ -128,7 +123,7 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "storageclass":
case "storageClass": target.getConfiguration().setStorageClass(property(camelContext, java.lang.String.class, value)); return true;
case "streammode":
- case "streamMode": target.getConfiguration().setStreamMode(property(camelContext, boolean.class, value)); return true;
+ case "streamMode": target.getConfiguration().setStreamingUploadMode(property(camelContext, boolean.class, value)); return true;
case "timeunit":
case "timeUnit": target.setTimeUnit(property(camelContext, java.util.concurrent.TimeUnit.class, value)); return true;
case "trustallcertificates":
@@ -392,7 +387,7 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "storageclass":
case "storageClass": return target.getConfiguration().getStorageClass();
case "streammode":
- case "streamMode": return target.getConfiguration().isStreamMode();
+ case "streamMode": return target.getConfiguration().isStreamingUploadMode();
case "timeunit":
case "timeUnit": return target.getTimeUnit();
case "trustallcertificates":
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
index d135306..0f48a66 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
@@ -112,13 +112,15 @@ public class AWS2S3Configuration implements Cloneable {
@UriParam(defaultValue = "false")
private boolean pojoRequest;
@UriParam(defaultValue = "false", label = "producer")
- private boolean streamMode;
+ private boolean streamingUploadMode;
@UriParam(defaultValue = "10", label = "producer")
private int batchMessageNumber = 10;
@UriParam(defaultValue = "1000000", label = "producer")
private int batchSize = 1000000;
@UriParam(defaultValue = "progressive", label = "producer")
private AWSS3NamingStrategyEnum namingStrategy = AWSS3NamingStrategyEnum.progressive;
+ @UriParam(label = "producer")
+ private long streamingUploadTimeout;
public long getPartSize() {
return partSize;
@@ -584,15 +586,15 @@ public class AWS2S3Configuration implements Cloneable {
this.amazonS3Presigner = amazonS3Presigner;
}
- public boolean isStreamMode() {
- return streamMode;
+ public boolean isStreamingUploadMode() {
+ return streamingUploadMode;
}
/**
* When stream mode is true the upload to bucket will be done in streaming
*/
- public void setStreamMode(boolean streamMode) {
- this.streamMode = streamMode;
+ public void setStreamingUploadMode(boolean streamingUploadMode) {
+ this.streamingUploadMode = streamingUploadMode;
}
public int getBatchMessageNumber() {
@@ -628,6 +630,17 @@ public class AWS2S3Configuration implements Cloneable {
this.namingStrategy = namingStrategy;
}
+ public long getStreamingUploadTimeout() {
+ return streamingUploadTimeout;
+ }
+
+ /**
+ * While streaming upload mode is true, this option set the timeout to complete upload
+ */
+ public void setStreamingUploadTimeout(long streamingUploadTimeout) {
+ this.streamingUploadTimeout = streamingUploadTimeout;
+ }
+
public AWS2S3Configuration copy() {
try {
return (AWS2S3Configuration) super.clone();
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Endpoint.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Endpoint.java
index 32d03835..f0bf82a 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Endpoint.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Endpoint.java
@@ -73,7 +73,7 @@ public class AWS2S3Endpoint extends ScheduledPollEndpoint {
@Override
public Producer createProducer() throws Exception {
- if (!configuration.isStreamMode()) {
+ if (!configuration.isStreamingUploadMode()) {
return new AWS2S3Producer(this);
} else {
return new AWS2S3StreamUploadProducer(this);
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
index f4d544d..21e8b53c 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
@@ -73,10 +73,12 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
@Override
protected void doStart() throws Exception {
super.doStart();
- timeoutCheckerExecutorService
- = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
- "timeout_checker");
- timeoutCheckerExecutorService.scheduleAtFixedRate(new AggregationIntervalTask(), 10, 10, TimeUnit.SECONDS);
+ if (ObjectHelper.isNotEmpty(getConfiguration().getStreamingUploadTimeout())) {
+ timeoutCheckerExecutorService
+ = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
+ "timeout_checker");
+ timeoutCheckerExecutorService.scheduleAtFixedRate(new AggregationIntervalTask(), getConfiguration().getStreamingUploadTimeout(), getConfiguration().getStreamingUploadTimeout(), TimeUnit.MILLISECONDS);
+ }
}
@Override
@@ -89,6 +91,10 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
}
}
}
+ if (timeoutCheckerExecutorService != null) {
+ getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(timeoutCheckerExecutorService);
+ timeoutCheckerExecutorService = null;
+ }
super.doStop();
}
@@ -276,7 +282,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
@Override
public String toString() {
if (s3ProducerToString == null) {
- s3ProducerToString = "S3Producer[" + URISupport.sanitizeUri(getEndpoint().getEndpointUri()) + "]";
+ s3ProducerToString = "AWS2S3StreamUploadProducer[" + URISupport.sanitizeUri(getEndpoint().getEndpointUri()) + "]";
}
return s3ProducerToString;
}
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadOperationLocalstackTest.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadOperationLocalstackTest.java
index 0db9740..7330813 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadOperationLocalstackTest.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadOperationLocalstackTest.java
@@ -71,7 +71,7 @@ public class S3StreamUploadOperationLocalstackTest extends Aws2S3BaseTest {
@Override
public void configure() throws Exception {
String awsEndpoint1
- = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random";
+ = "aws2-s3://mycamel-1?autoCreateBucket=true&streamingUploadMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random";
from("direct:stream1").to(awsEndpoint1).to("mock:result");
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadTimeoutLocalstackTest.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadTimeoutLocalstackTest.java
index f42a153..d958e1c 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadTimeoutLocalstackTest.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadTimeoutLocalstackTest.java
@@ -55,7 +55,7 @@ public class S3StreamUploadTimeoutLocalstackTest extends Aws2S3BaseTest {
@Override
public void configure() throws Exception {
String awsEndpoint1
- = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random";
+ = "aws2-s3://mycamel-1?autoCreateBucket=true&streamingUploadMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random";
from("direct:stream1").to(awsEndpoint1).to("mock:result");