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:01 UTC
[camel] branch master updated (d944320 -> 5721592)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from d944320 CAMEL-16405 Added statement query timeout in sql-stored component (#5294)
new c0f45fb CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new 844da25 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new fb64227 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new 624a259 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new a49b814 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new 4821425 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new 578dfe9 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new c701c66 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new 5721592 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/camel/catalog/components/aws2-s3.json | 6 +-
.../camel/catalog/docs/aws2-s3-component.adoc | 16 +--
.../aws2/s3/AWS2S3ComponentConfigurer.java | 18 ++--
.../aws2/s3/AWS2S3EndpointConfigurer.java | 18 ++--
.../aws2/s3/AWS2S3EndpointUriFactory.java | 5 +-
.../apache/camel/component/aws2/s3/aws2-s3.json | 6 +-
.../src/main/docs/aws2-s3-component.adoc | 16 +--
.../component/aws2/s3/AWS2S3Configuration.java | 23 ++++-
.../camel/component/aws2/s3/AWS2S3Endpoint.java | 2 +-
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 114 ++++++++++++++++-----
.../S3StreamUploadOperationLocalstackTest.java | 2 +-
...va => S3StreamUploadTimeoutLocalstackTest.java} | 11 +-
.../dsl/Aws2S3ComponentBuilderFactory.java | 26 ++++-
.../endpoint/dsl/AWS2S3EndpointBuilderFactory.java | 46 +++++++--
.../modules/ROOT/pages/aws2-s3-component.adoc | 16 +--
15 files changed, 237 insertions(+), 88 deletions(-)
copy components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/{S3StreamUploadOperationLocalstackTest.java => S3StreamUploadTimeoutLocalstackTest.java} (88%)
[camel] 06/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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 4821425d27fa10a3ec8d418f9204f97606f5e1d1
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Apr 6 07:39:09 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../aws2/s3/localstack/S3StreamUploadOperationLocalstackTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
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 d463732..454e30b 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
@@ -56,7 +56,7 @@ public class S3StreamUploadOperationLocalstackTest extends Aws2S3BaseTest {
exchange.getIn().setHeader(AWS2S3Constants.S3_OPERATION, AWS2S3Operations.listObjects);
}
});
-
+
List<S3Object> resp = ex.getMessage().getBody(List.class);
assertEquals(40, resp.size());
}
[camel] 03/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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");
[camel] 08/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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 c701c6662d03432ef30e5a595946b64756ea07e6
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Apr 6 08:23:31 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc b/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc
index dc3fd8f..4489346 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc
+++ b/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc
@@ -555,7 +555,7 @@ This operation will return an S3Object instance related to the camelKey object i
This operation will return a download link url for the file camel-key in the bucket mycamelbucket and region region
-== Stream mode
+== Streaming Upload mode
With the stream mode enabled users will be able to upload data to S3 without knowing ahead of time the dimension of the data, by leveraging multipart upload.
The upload will be completed when: the batchSize has been completed or the batchMessageNumber has been reached.
@@ -567,11 +567,11 @@ As an example:
--------------------------------------------------------------------------------
from(kafka("topic1").brokers("localhost:9092"))
.log("Kafka Message is: ${body}")
- .to(aws2S3("camel-bucket").streamMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));
+ .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));
from(kafka("topic2").brokers("localhost:9092"))
.log("Kafka Message is: ${body}")
- .to(aws2S3("camel-bucket").streamMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic2}}/{{kafkaTopic2}}.txt"));
+ .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic2}}/{{kafkaTopic2}}.txt"));
--------------------------------------------------------------------------------
The default size for a batch is 1 Mb, but you can adjust it according to your requirements.
[camel] 02/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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 844da25b4ebf690d2625a543dd154eb9fdabc37d
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Sun Apr 4 16:37:24 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 89 ++++++++++++++--------
.../S3StreamUploadOperationLocalstackTest.java | 3 +
...va => S3StreamUploadTimeoutLocalstackTest.java} | 29 ++-----
3 files changed, 68 insertions(+), 53 deletions(-)
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 32c6f28..f4d544d 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
@@ -66,9 +66,9 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
AtomicInteger part = new AtomicInteger();
UUID id;
String dynamicKeyName;
+ CompleteMultipartUploadResponse uploadResult;
private transient String s3ProducerToString;
private ScheduledExecutorService timeoutCheckerExecutorService;
- private boolean timeout;
@Override
protected void doStart() throws Exception {
@@ -76,7 +76,21 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
timeoutCheckerExecutorService
= getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
"timeout_checker");
- timeoutCheckerExecutorService.scheduleAtFixedRate(new AggregationIntervalTask(), 1, 1, TimeUnit.SECONDS);
+ timeoutCheckerExecutorService.scheduleAtFixedRate(new AggregationIntervalTask(), 10, 10, TimeUnit.SECONDS);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (ObjectHelper.isNotEmpty(initResponse)) {
+ if (ObjectHelper.isNotEmpty(initResponse.uploadId())) {
+ if (index.get() > 0) {
+ uploadPart();
+ completeUpload();
+ }
+ }
+ }
+ super.doStop();
+
}
/**
@@ -86,8 +100,14 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
@Override
public void run() {
- timeout = true;
- LOG.info("timeout triggered");
+ if (ObjectHelper.isNotEmpty(initResponse)) {
+ if (ObjectHelper.isNotEmpty(initResponse.uploadId())) {
+ if (index.get() > 0) {
+ uploadPart();
+ completeUpload();
+ }
+ }
+ }
}
}
@@ -157,41 +177,15 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
try {
if (buffer.size() >= getConfiguration().getBatchSize()
|| index.get() == getConfiguration().getBatchMessageNumber()) {
- LOG.info("Timeout " + timeout);
-
- UploadPartRequest uploadRequest = UploadPartRequest.builder().bucket(getConfiguration().getBucketName())
- .key(dynamicKeyName).uploadId(initResponse.uploadId())
- .partNumber(index.get()).build();
- LOG.trace("Uploading part {} at index {} for {}", part, index, keyName);
+ uploadPart();
+ completeUpload();
- String etag = getEndpoint().getS3Client()
- .uploadPart(uploadRequest, RequestBody.fromBytes(buffer.toByteArray())).eTag();
- CompletedPart partUpload = CompletedPart.builder().partNumber(index.get()).eTag(etag).build();
- completedParts.add(partUpload);
- buffer.reset();
- part.getAndIncrement();
- }
-
- if (index.get() == getConfiguration().getBatchMessageNumber() || timeout) {
- CompletedMultipartUpload completeMultipartUpload
- = CompletedMultipartUpload.builder().parts(completedParts).build();
- CompleteMultipartUploadRequest compRequest
- = CompleteMultipartUploadRequest.builder().multipartUpload(completeMultipartUpload)
- .bucket(getConfiguration().getBucketName()).key(dynamicKeyName)
- .uploadId(initResponse.uploadId())
- .build();
-
- CompleteMultipartUploadResponse uploadResult = getEndpoint().getS3Client().completeMultipartUpload(compRequest);
- LOG.info("Completed upload for the part {} with etag {} at index {}", part, uploadResult.eTag(),
- index);
Message message = getMessageForResponse(exchange);
message.setHeader(AWS2S3Constants.E_TAG, uploadResult.eTag());
if (uploadResult.versionId() != null) {
message.setHeader(AWS2S3Constants.VERSION_ID, uploadResult.versionId());
}
- timeout = false;
- index.getAndSet(0);
}
} catch (Exception e) {
@@ -204,6 +198,37 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
index.getAndIncrement();
}
+ private void completeUpload() {
+ CompletedMultipartUpload completeMultipartUpload
+ = CompletedMultipartUpload.builder().parts(completedParts).build();
+ CompleteMultipartUploadRequest compRequest
+ = CompleteMultipartUploadRequest.builder().multipartUpload(completeMultipartUpload)
+ .bucket(getConfiguration().getBucketName()).key(dynamicKeyName)
+ .uploadId(initResponse.uploadId())
+ .build();
+
+ uploadResult = getEndpoint().getS3Client().completeMultipartUpload(compRequest);
+ LOG.info("Completed upload for the part {} with etag {} at index {}", part, uploadResult.eTag(),
+ index);
+
+ index.getAndSet(0);
+ }
+
+ private void uploadPart() {
+ UploadPartRequest uploadRequest = UploadPartRequest.builder().bucket(getConfiguration().getBucketName())
+ .key(dynamicKeyName).uploadId(initResponse.uploadId())
+ .partNumber(index.get()).build();
+
+ LOG.trace("Uploading part {} at index {} for {}", part, index, getConfiguration().getKeyName());
+
+ String etag = getEndpoint().getS3Client()
+ .uploadPart(uploadRequest, RequestBody.fromBytes(buffer.toByteArray())).eTag();
+ CompletedPart partUpload = CompletedPart.builder().partNumber(index.get()).eTag(etag).build();
+ completedParts.add(partUpload);
+ buffer.reset();
+ part.getAndIncrement();
+ }
+
private String fileNameToUpload(
String fileName, AWSS3NamingStrategyEnum strategy, String ext, AtomicInteger part, UUID id) {
String dynamicKeyName;
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 6a37b9ad..0db9740 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
@@ -60,6 +60,9 @@ public class S3StreamUploadOperationLocalstackTest extends Aws2S3BaseTest {
Thread.sleep(30000);
List<S3Object> resp = ex.getMessage().getBody(List.class);
assertEquals(40, resp.size());
+ for (S3Object s3Object : resp) {
+ System.err.println(s3Object.key());
+ }
}
@Override
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/S3StreamUploadTimeoutLocalstackTest.java
similarity index 65%
copy from components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadOperationLocalstackTest.java
copy to components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/localstack/S3StreamUploadTimeoutLocalstackTest.java
index 6a37b9ad..f42a153 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/S3StreamUploadTimeoutLocalstackTest.java
@@ -1,19 +1,3 @@
-/*
- * 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.aws2.s3.localstack;
import java.util.List;
@@ -31,7 +15,7 @@ import software.amazon.awssdk.services.s3.model.S3Object;
import static org.junit.jupiter.api.Assertions.assertEquals;
-public class S3StreamUploadOperationLocalstackTest extends Aws2S3BaseTest {
+public class S3StreamUploadTimeoutLocalstackTest extends Aws2S3BaseTest {
@EndpointInject
private ProducerTemplate template;
@@ -41,12 +25,13 @@ public class S3StreamUploadOperationLocalstackTest extends Aws2S3BaseTest {
@Test
public void sendIn() throws Exception {
- result.expectedMessageCount(1000);
+ result.expectedMessageCount(23);
- for (int i = 0; i < 1000; i++) {
+ for (int i = 0; i < 23; i++) {
template.sendBody("direct:stream1", "Andrea\n");
}
+ Thread.sleep(11000);
assertMockEndpointsSatisfied();
Exchange ex = template.request("direct:listObjects", new Processor() {
@@ -57,9 +42,11 @@ public class S3StreamUploadOperationLocalstackTest extends Aws2S3BaseTest {
}
});
- Thread.sleep(30000);
List<S3Object> resp = ex.getMessage().getBody(List.class);
- assertEquals(40, resp.size());
+ assertEquals(1, resp.size());
+ for (S3Object s3Object : resp) {
+ System.err.println(s3Object.key());
+ }
}
@Override
[camel] 05/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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 a49b814d6ccde0a951e06c47f373d538cc724a9e
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Apr 6 07:21:37 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../aws2/s3/localstack/S3StreamUploadOperationLocalstackTest.java | 6 +-----
.../aws2/s3/localstack/S3StreamUploadTimeoutLocalstackTest.java | 3 ---
2 files changed, 1 insertion(+), 8 deletions(-)
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 7330813..d463732 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
@@ -56,13 +56,9 @@ public class S3StreamUploadOperationLocalstackTest extends Aws2S3BaseTest {
exchange.getIn().setHeader(AWS2S3Constants.S3_OPERATION, AWS2S3Operations.listObjects);
}
});
-
- Thread.sleep(30000);
+
List<S3Object> resp = ex.getMessage().getBody(List.class);
assertEquals(40, resp.size());
- for (S3Object s3Object : resp) {
- System.err.println(s3Object.key());
- }
}
@Override
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 79f1d54..1601256 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
@@ -44,9 +44,6 @@ public class S3StreamUploadTimeoutLocalstackTest extends Aws2S3BaseTest {
List<S3Object> resp = ex.getMessage().getBody(List.class);
assertEquals(1, resp.size());
- for (S3Object s3Object : resp) {
- System.err.println(s3Object.key());
- }
}
@Override
[camel] 01/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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 c0f45fb324398da9307344496c098d65ba2de178
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Fri Apr 2 19:22:27 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 31 ++++++++++++++++++++--
.../S3StreamUploadOperationLocalstackTest.java | 1 +
2 files changed, 30 insertions(+), 2 deletions(-)
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 4a4785f..32c6f28 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
@@ -21,6 +21,8 @@ import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.Endpoint;
@@ -65,6 +67,29 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
UUID id;
String dynamicKeyName;
private transient String s3ProducerToString;
+ private ScheduledExecutorService timeoutCheckerExecutorService;
+ private boolean timeout;
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ timeoutCheckerExecutorService
+ = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
+ "timeout_checker");
+ timeoutCheckerExecutorService.scheduleAtFixedRate(new AggregationIntervalTask(), 1, 1, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Background task that triggers completion based on interval.
+ */
+ private final class AggregationIntervalTask implements Runnable {
+
+ @Override
+ public void run() {
+ timeout = true;
+ LOG.info("timeout triggered");
+ }
+ }
public AWS2S3StreamUploadProducer(final Endpoint endpoint) {
super(endpoint);
@@ -132,6 +157,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
try {
if (buffer.size() >= getConfiguration().getBatchSize()
|| index.get() == getConfiguration().getBatchMessageNumber()) {
+ LOG.info("Timeout " + timeout);
UploadPartRequest uploadRequest = UploadPartRequest.builder().bucket(getConfiguration().getBucketName())
.key(dynamicKeyName).uploadId(initResponse.uploadId())
@@ -147,7 +173,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
part.getAndIncrement();
}
- if (index.get() == getConfiguration().getBatchMessageNumber()) {
+ if (index.get() == getConfiguration().getBatchMessageNumber() || timeout) {
CompletedMultipartUpload completeMultipartUpload
= CompletedMultipartUpload.builder().parts(completedParts).build();
CompleteMultipartUploadRequest compRequest
@@ -157,13 +183,14 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
.build();
CompleteMultipartUploadResponse uploadResult = getEndpoint().getS3Client().completeMultipartUpload(compRequest);
- LOG.trace("Completed upload for the part {} with etag {} at index {}", part, uploadResult.eTag(),
+ LOG.info("Completed upload for the part {} with etag {} at index {}", part, uploadResult.eTag(),
index);
Message message = getMessageForResponse(exchange);
message.setHeader(AWS2S3Constants.E_TAG, uploadResult.eTag());
if (uploadResult.versionId() != null) {
message.setHeader(AWS2S3Constants.VERSION_ID, uploadResult.versionId());
}
+ timeout = false;
index.getAndSet(0);
}
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 87cca16..6a37b9ad 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
@@ -57,6 +57,7 @@ public class S3StreamUploadOperationLocalstackTest extends Aws2S3BaseTest {
}
});
+ Thread.sleep(30000);
List<S3Object> resp = ex.getMessage().getBody(List.class);
assertEquals(40, resp.size());
}
[camel] 04/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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 624a2598a715dba0b1e5e3935b20e51fe9f1cd17
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Apr 6 07:12:33 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../camel/catalog/docs/aws2-s3-component.adoc | 10 +++--
.../aws2/s3/AWS2S3ComponentConfigurer.java | 23 ++++++++---
.../aws2/s3/AWS2S3EndpointConfigurer.java | 23 ++++++++---
.../aws2/s3/AWS2S3EndpointUriFactory.java | 5 ++-
.../apache/camel/component/aws2/s3/aws2-s3.json | 6 ++-
.../src/main/docs/aws2-s3-component.adoc | 10 +++--
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 8 ++--
.../S3StreamUploadTimeoutLocalstackTest.java | 2 +-
.../dsl/Aws2S3ComponentBuilderFactory.java | 26 ++++++++++--
.../endpoint/dsl/AWS2S3EndpointBuilderFactory.java | 46 +++++++++++++++++++---
.../modules/ROOT/pages/aws2-s3-component.adoc | 10 +++--
11 files changed, 127 insertions(+), 42 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc
index 5d5ea99..dc3fd8f 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc
@@ -47,7 +47,7 @@ from("aws2-s3://helloBucket?accessKey=yourAccessKey&secretKey=yourSecretKey&pref
// component options: START
-The AWS 2 S3 Storage Service component supports 47 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 48 options, which are listed below.
@@ -94,7 +94,8 @@ The AWS 2 S3 Storage Service component supports 47 options, which are listed bel
| *operation* (producer) | The operation to do in case the user don't want to do only an upload. There are 7 enums and the value can be one of: copyObject, listObjects, deleteObject, deleteBucket, listBuckets, getObject, getObjectRange | | AWS2S3Operations
| *partSize* (producer) | Setup the partSize which is used in multi part upload, the default size is 25M. | 26214400 | long
| *storageClass* (producer) | The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request. | | String
-| *streamMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadTimeout* (producer) | While streaming upload mode is true, this option set the timeout to complete upload | | long
| *awsKMSKeyId* (producer) | Define the id of KMS key to use in case KMS is enabled | | String
| *useAwsKMS* (producer) | Define if KMS must be used or not | false | boolean
| *useCustomerKey* (producer) | Define if Customer Key must be used or not | false | boolean
@@ -133,7 +134,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (65 parameters):
+=== Query Parameters (66 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -184,7 +185,8 @@ with the following path and query parameters:
| *operation* (producer) | The operation to do in case the user don't want to do only an upload. There are 7 enums and the value can be one of: copyObject, listObjects, deleteObject, deleteBucket, listBuckets, getObject, getObjectRange | | AWS2S3Operations
| *partSize* (producer) | Setup the partSize which is used in multi part upload, the default size is 25M. | 26214400 | long
| *storageClass* (producer) | The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request. | | String
-| *streamMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadTimeout* (producer) | While streaming upload mode is true, this option set the timeout to complete upload | | long
| *awsKMSKeyId* (producer) | Define the id of KMS key to use in case KMS is enabled | | String
| *useAwsKMS* (producer) | Define if KMS must be used or not | false | boolean
| *useCustomerKey* (producer) | Define if Customer Key must be used or not | false | boolean
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 89a4c80..4920a0a 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,9 +1,14 @@
/* 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;
/**
@@ -99,8 +104,10 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "secretKey": getOrCreateConfiguration(target).setSecretKey(property(camelContext, java.lang.String.class, value)); return true;
case "storageclass":
case "storageClass": getOrCreateConfiguration(target).setStorageClass(property(camelContext, java.lang.String.class, value)); return true;
- case "streammode":
- case "streamMode": getOrCreateConfiguration(target).setStreamingUploadMode(property(camelContext, boolean.class, value)); return true;
+ case "streaminguploadmode":
+ case "streamingUploadMode": getOrCreateConfiguration(target).setStreamingUploadMode(property(camelContext, boolean.class, value)); return true;
+ case "streaminguploadtimeout":
+ case "streamingUploadTimeout": getOrCreateConfiguration(target).setStreamingUploadTimeout(property(camelContext, long.class, value)); return true;
case "trustallcertificates":
case "trustAllCertificates": getOrCreateConfiguration(target).setTrustAllCertificates(property(camelContext, boolean.class, value)); return true;
case "uriendpointoverride":
@@ -199,8 +206,10 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "secretKey": return java.lang.String.class;
case "storageclass":
case "storageClass": return java.lang.String.class;
- case "streammode":
- case "streamMode": return boolean.class;
+ case "streaminguploadmode":
+ case "streamingUploadMode": return boolean.class;
+ case "streaminguploadtimeout":
+ case "streamingUploadTimeout": return long.class;
case "trustallcertificates":
case "trustAllCertificates": return boolean.class;
case "uriendpointoverride":
@@ -295,8 +304,10 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "secretKey": return getOrCreateConfiguration(target).getSecretKey();
case "storageclass":
case "storageClass": return getOrCreateConfiguration(target).getStorageClass();
- case "streammode":
- case "streamMode": return getOrCreateConfiguration(target).isStreamingUploadMode();
+ case "streaminguploadmode":
+ case "streamingUploadMode": return getOrCreateConfiguration(target).isStreamingUploadMode();
+ case "streaminguploadtimeout":
+ case "streamingUploadTimeout": return getOrCreateConfiguration(target).getStreamingUploadTimeout();
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 bd98c61..9b57190 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,9 +1,14 @@
/* 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;
/**
@@ -122,8 +127,10 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "startScheduler": target.setStartScheduler(property(camelContext, boolean.class, value)); return true;
case "storageclass":
case "storageClass": target.getConfiguration().setStorageClass(property(camelContext, java.lang.String.class, value)); return true;
- case "streammode":
- case "streamMode": target.getConfiguration().setStreamingUploadMode(property(camelContext, boolean.class, value)); return true;
+ case "streaminguploadmode":
+ case "streamingUploadMode": target.getConfiguration().setStreamingUploadMode(property(camelContext, boolean.class, value)); return true;
+ case "streaminguploadtimeout":
+ case "streamingUploadTimeout": target.getConfiguration().setStreamingUploadTimeout(property(camelContext, long.class, value)); return true;
case "timeunit":
case "timeUnit": target.setTimeUnit(property(camelContext, java.util.concurrent.TimeUnit.class, value)); return true;
case "trustallcertificates":
@@ -256,8 +263,10 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "startScheduler": return boolean.class;
case "storageclass":
case "storageClass": return java.lang.String.class;
- case "streammode":
- case "streamMode": return boolean.class;
+ case "streaminguploadmode":
+ case "streamingUploadMode": return boolean.class;
+ case "streaminguploadtimeout":
+ case "streamingUploadTimeout": return long.class;
case "timeunit":
case "timeUnit": return java.util.concurrent.TimeUnit.class;
case "trustallcertificates":
@@ -386,8 +395,10 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "startScheduler": return target.isStartScheduler();
case "storageclass":
case "storageClass": return target.getConfiguration().getStorageClass();
- case "streammode":
- case "streamMode": return target.getConfiguration().isStreamingUploadMode();
+ case "streaminguploadmode":
+ case "streamingUploadMode": return target.getConfiguration().isStreamingUploadMode();
+ case "streaminguploadtimeout":
+ case "streamingUploadTimeout": return target.getConfiguration().getStreamingUploadTimeout();
case "timeunit":
case "timeUnit": return target.getTimeUnit();
case "trustallcertificates":
diff --git a/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3EndpointUriFactory.java b/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3EndpointUriFactory.java
index e6c1ba0..6eb9073 100644
--- a/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3EndpointUriFactory.java
+++ b/components/camel-aws/camel-aws2-s3/src/generated/java/org/apache/camel/component/aws2/s3/AWS2S3EndpointUriFactory.java
@@ -20,12 +20,11 @@ public class AWS2S3EndpointUriFactory extends org.apache.camel.support.component
private static final Set<String> PROPERTY_NAMES;
private static final Set<String> SECRET_PROPERTY_NAMES;
static {
- Set<String> props = new HashSet<>(66);
+ Set<String> props = new HashSet<>(67);
props.add("customerAlgorithm");
props.add("fileName");
props.add("batchMessageNumber");
props.add("useCustomerKey");
- props.add("streamMode");
props.add("bucketNameOrArn");
props.add("customerKeyId");
props.add("prefix");
@@ -37,6 +36,7 @@ public class AWS2S3EndpointUriFactory extends org.apache.camel.support.component
props.add("bridgeErrorHandler");
props.add("awsKMSKeyId");
props.add("delimiter");
+ props.add("streamingUploadMode");
props.add("amazonS3Client");
props.add("greedy");
props.add("maxMessagesPerPoll");
@@ -45,6 +45,7 @@ public class AWS2S3EndpointUriFactory extends org.apache.camel.support.component
props.add("moveAfterRead");
props.add("sendEmptyMessageWhenIdle");
props.add("schedulerProperties");
+ props.add("streamingUploadTimeout");
props.add("proxyHost");
props.add("backoffIdleThreshold");
props.add("trustAllCertificates");
diff --git a/components/camel-aws/camel-aws2-s3/src/generated/resources/org/apache/camel/component/aws2/s3/aws2-s3.json b/components/camel-aws/camel-aws2-s3/src/generated/resources/org/apache/camel/component/aws2/s3/aws2-s3.json
index feced29..584eec8 100644
--- a/components/camel-aws/camel-aws2-s3/src/generated/resources/org/apache/camel/component/aws2/s3/aws2-s3.json
+++ b/components/camel-aws/camel-aws2-s3/src/generated/resources/org/apache/camel/component/aws2/s3/aws2-s3.json
@@ -62,7 +62,8 @@
"operation": { "kind": "property", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.AWS2S3Operations", "enum": [ "copyObject", "listObjects", "deleteObject", "deleteBucket", "listBuckets", "getObject", "getObjectRange" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": [...]
"partSize": { "kind": "property", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Setup the partSize which is used in multi part upload, the default size is 25M." },
"storageClass": { "kind": "property", "displayName": "Storage Class", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request." },
- "streamMode": { "kind": "property", "displayName": "Stream Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true the upload to bucket will be done in streaming" },
+ "streamingUploadMode": { "kind": "property", "displayName": "Streaming Upload Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true the upload to bucket will be done in streaming" },
+ "streamingUploadTimeout": { "kind": "property", "displayName": "Streaming Upload Timeout", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "While streaming upload mode is true, this option set the timeout to complete upload" },
"awsKMSKeyId": { "kind": "property", "displayName": "Aws KMSKey Id", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define the id of KMS key to use in case KMS is enabled" },
"useAwsKMS": { "kind": "property", "displayName": "Use Aws KMS", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define if KMS must be used or not" },
"useCustomerKey": { "kind": "property", "displayName": "Use Customer Key", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define if Customer Key must be used or not" },
@@ -117,7 +118,8 @@
"operation": { "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.AWS2S3Operations", "enum": [ "copyObject", "listObjects", "deleteObject", "deleteBucket", "listBuckets", "getObject", "getObjectRange" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": [...]
"partSize": { "kind": "parameter", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Setup the partSize which is used in multi part upload, the default size is 25M." },
"storageClass": { "kind": "parameter", "displayName": "Storage Class", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request." },
- "streamMode": { "kind": "parameter", "displayName": "Stream Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true the upload to bucket will be done in streaming" },
+ "streamingUploadMode": { "kind": "parameter", "displayName": "Streaming Upload Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true the upload to bucket will be done in streaming" },
+ "streamingUploadTimeout": { "kind": "parameter", "displayName": "Streaming Upload Timeout", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "While streaming upload mode is true, this option set the timeout to complete upload" },
"awsKMSKeyId": { "kind": "parameter", "displayName": "Aws KMSKey Id", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define the id of KMS key to use in case KMS is enabled" },
"useAwsKMS": { "kind": "parameter", "displayName": "Use Aws KMS", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define if KMS must be used or not" },
"useCustomerKey": { "kind": "parameter", "displayName": "Use Customer Key", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define if Customer Key must be used or not" },
diff --git a/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc b/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc
index 5d5ea99..dc3fd8f 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc
+++ b/components/camel-aws/camel-aws2-s3/src/main/docs/aws2-s3-component.adoc
@@ -47,7 +47,7 @@ from("aws2-s3://helloBucket?accessKey=yourAccessKey&secretKey=yourSecretKey&pref
// component options: START
-The AWS 2 S3 Storage Service component supports 47 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 48 options, which are listed below.
@@ -94,7 +94,8 @@ The AWS 2 S3 Storage Service component supports 47 options, which are listed bel
| *operation* (producer) | The operation to do in case the user don't want to do only an upload. There are 7 enums and the value can be one of: copyObject, listObjects, deleteObject, deleteBucket, listBuckets, getObject, getObjectRange | | AWS2S3Operations
| *partSize* (producer) | Setup the partSize which is used in multi part upload, the default size is 25M. | 26214400 | long
| *storageClass* (producer) | The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request. | | String
-| *streamMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadTimeout* (producer) | While streaming upload mode is true, this option set the timeout to complete upload | | long
| *awsKMSKeyId* (producer) | Define the id of KMS key to use in case KMS is enabled | | String
| *useAwsKMS* (producer) | Define if KMS must be used or not | false | boolean
| *useCustomerKey* (producer) | Define if Customer Key must be used or not | false | boolean
@@ -133,7 +134,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (65 parameters):
+=== Query Parameters (66 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -184,7 +185,8 @@ with the following path and query parameters:
| *operation* (producer) | The operation to do in case the user don't want to do only an upload. There are 7 enums and the value can be one of: copyObject, listObjects, deleteObject, deleteBucket, listBuckets, getObject, getObjectRange | | AWS2S3Operations
| *partSize* (producer) | Setup the partSize which is used in multi part upload, the default size is 25M. | 26214400 | long
| *storageClass* (producer) | The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request. | | String
-| *streamMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadTimeout* (producer) | While streaming upload mode is true, this option set the timeout to complete upload | | long
| *awsKMSKeyId* (producer) | Define the id of KMS key to use in case KMS is enabled | | String
| *useAwsKMS* (producer) | Define if KMS must be used or not | false | boolean
| *useCustomerKey* (producer) | Define if Customer Key must be used or not | false | boolean
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 21e8b53c..412df8f 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,11 +73,13 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
@Override
protected void doStart() throws Exception {
super.doStart();
- if (ObjectHelper.isNotEmpty(getConfiguration().getStreamingUploadTimeout())) {
+ if (getConfiguration().getStreamingUploadTimeout() > 0) {
timeoutCheckerExecutorService
= getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
- "timeout_checker");
- timeoutCheckerExecutorService.scheduleAtFixedRate(new AggregationIntervalTask(), getConfiguration().getStreamingUploadTimeout(), getConfiguration().getStreamingUploadTimeout(), TimeUnit.MILLISECONDS);
+ "timeout_checker");
+ timeoutCheckerExecutorService.scheduleAtFixedRate(new AggregationIntervalTask(),
+ getConfiguration().getStreamingUploadTimeout(), getConfiguration().getStreamingUploadTimeout(),
+ TimeUnit.MILLISECONDS);
}
}
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 d958e1c..79f1d54 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&streamingUploadMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random";
+ = "aws2-s3://mycamel-1?autoCreateBucket=true&streamingUploadMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random&streamingUploadTimeout=10000";
from("direct:stream1").to(awsEndpoint1).to("mock:result");
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2S3ComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2S3ComponentBuilderFactory.java
index f31f288..9218000 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2S3ComponentBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2S3ComponentBuilderFactory.java
@@ -730,11 +730,28 @@ public interface Aws2S3ComponentBuilderFactory {
* Default: false
* Group: producer
*
- * @param streamMode the value to set
+ * @param streamingUploadMode the value to set
* @return the dsl builder
*/
- default Aws2S3ComponentBuilder streamMode(boolean streamMode) {
- doSetProperty("streamMode", streamMode);
+ default Aws2S3ComponentBuilder streamingUploadMode(
+ boolean streamingUploadMode) {
+ doSetProperty("streamingUploadMode", streamingUploadMode);
+ return this;
+ }
+ /**
+ * While streaming upload mode is true, this option set the timeout to
+ * complete upload.
+ *
+ * The option is a: <code>long</code> type.
+ *
+ * Group: producer
+ *
+ * @param streamingUploadTimeout the value to set
+ * @return the dsl builder
+ */
+ default Aws2S3ComponentBuilder streamingUploadTimeout(
+ long streamingUploadTimeout) {
+ doSetProperty("streamingUploadTimeout", streamingUploadTimeout);
return this;
}
/**
@@ -893,7 +910,8 @@ public interface Aws2S3ComponentBuilderFactory {
case "operation": getOrCreateConfiguration((AWS2S3Component) component).setOperation((org.apache.camel.component.aws2.s3.AWS2S3Operations) value); return true;
case "partSize": getOrCreateConfiguration((AWS2S3Component) component).setPartSize((long) value); return true;
case "storageClass": getOrCreateConfiguration((AWS2S3Component) component).setStorageClass((java.lang.String) value); return true;
- case "streamMode": getOrCreateConfiguration((AWS2S3Component) component).setStreamMode((boolean) value); return true;
+ case "streamingUploadMode": getOrCreateConfiguration((AWS2S3Component) component).setStreamingUploadMode((boolean) value); return true;
+ case "streamingUploadTimeout": getOrCreateConfiguration((AWS2S3Component) component).setStreamingUploadTimeout((long) value); return true;
case "awsKMSKeyId": getOrCreateConfiguration((AWS2S3Component) component).setAwsKMSKeyId((java.lang.String) value); return true;
case "useAwsKMS": getOrCreateConfiguration((AWS2S3Component) component).setUseAwsKMS((boolean) value); return true;
case "useCustomerKey": getOrCreateConfiguration((AWS2S3Component) component).setUseCustomerKey((boolean) value); return true;
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AWS2S3EndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AWS2S3EndpointBuilderFactory.java
index 0f77798..bcc7ff6 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AWS2S3EndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/AWS2S3EndpointBuilderFactory.java
@@ -2251,11 +2251,12 @@ public interface AWS2S3EndpointBuilderFactory {
* Default: false
* Group: producer
*
- * @param streamMode the value to set
+ * @param streamingUploadMode the value to set
* @return the dsl builder
*/
- default AWS2S3EndpointProducerBuilder streamMode(boolean streamMode) {
- doSetProperty("streamMode", streamMode);
+ default AWS2S3EndpointProducerBuilder streamingUploadMode(
+ boolean streamingUploadMode) {
+ doSetProperty("streamingUploadMode", streamingUploadMode);
return this;
}
/**
@@ -2268,11 +2269,44 @@ public interface AWS2S3EndpointBuilderFactory {
* Default: false
* Group: producer
*
- * @param streamMode the value to set
+ * @param streamingUploadMode the value to set
* @return the dsl builder
*/
- default AWS2S3EndpointProducerBuilder streamMode(String streamMode) {
- doSetProperty("streamMode", streamMode);
+ default AWS2S3EndpointProducerBuilder streamingUploadMode(
+ String streamingUploadMode) {
+ doSetProperty("streamingUploadMode", streamingUploadMode);
+ return this;
+ }
+ /**
+ * While streaming upload mode is true, this option set the timeout to
+ * complete upload.
+ *
+ * The option is a: <code>long</code> type.
+ *
+ * Group: producer
+ *
+ * @param streamingUploadTimeout the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder streamingUploadTimeout(
+ long streamingUploadTimeout) {
+ doSetProperty("streamingUploadTimeout", streamingUploadTimeout);
+ return this;
+ }
+ /**
+ * While streaming upload mode is true, this option set the timeout to
+ * complete upload.
+ *
+ * The option will be converted to a <code>long</code> type.
+ *
+ * Group: producer
+ *
+ * @param streamingUploadTimeout the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder streamingUploadTimeout(
+ String streamingUploadTimeout) {
+ doSetProperty("streamingUploadTimeout", streamingUploadTimeout);
return this;
}
/**
diff --git a/docs/components/modules/ROOT/pages/aws2-s3-component.adoc b/docs/components/modules/ROOT/pages/aws2-s3-component.adoc
index d9b4856..f20ba1f 100644
--- a/docs/components/modules/ROOT/pages/aws2-s3-component.adoc
+++ b/docs/components/modules/ROOT/pages/aws2-s3-component.adoc
@@ -49,7 +49,7 @@ from("aws2-s3://helloBucket?accessKey=yourAccessKey&secretKey=yourSecretKey&pref
// component options: START
-The AWS 2 S3 Storage Service component supports 47 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 48 options, which are listed below.
@@ -96,7 +96,8 @@ The AWS 2 S3 Storage Service component supports 47 options, which are listed bel
| *operation* (producer) | The operation to do in case the user don't want to do only an upload. There are 7 enums and the value can be one of: copyObject, listObjects, deleteObject, deleteBucket, listBuckets, getObject, getObjectRange | | AWS2S3Operations
| *partSize* (producer) | Setup the partSize which is used in multi part upload, the default size is 25M. | 26214400 | long
| *storageClass* (producer) | The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request. | | String
-| *streamMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadTimeout* (producer) | While streaming upload mode is true, this option set the timeout to complete upload | | long
| *awsKMSKeyId* (producer) | Define the id of KMS key to use in case KMS is enabled | | String
| *useAwsKMS* (producer) | Define if KMS must be used or not | false | boolean
| *useCustomerKey* (producer) | Define if Customer Key must be used or not | false | boolean
@@ -135,7 +136,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (65 parameters):
+=== Query Parameters (66 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -186,7 +187,8 @@ with the following path and query parameters:
| *operation* (producer) | The operation to do in case the user don't want to do only an upload. There are 7 enums and the value can be one of: copyObject, listObjects, deleteObject, deleteBucket, listBuckets, getObject, getObjectRange | | AWS2S3Operations
| *partSize* (producer) | Setup the partSize which is used in multi part upload, the default size is 25M. | 26214400 | long
| *storageClass* (producer) | The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request. | | String
-| *streamMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadMode* (producer) | When stream mode is true the upload to bucket will be done in streaming | false | boolean
+| *streamingUploadTimeout* (producer) | While streaming upload mode is true, this option set the timeout to complete upload | | long
| *awsKMSKeyId* (producer) | Define the id of KMS key to use in case KMS is enabled | | String
| *useAwsKMS* (producer) | Define if KMS must be used or not | false | boolean
| *useCustomerKey* (producer) | Define if Customer Key must be used or not | false | boolean
[camel] 09/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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 57215927777f79d6611246e30d0dd077d22af717
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Apr 6 08:37:39 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../resources/org/apache/camel/catalog/components/aws2-s3.json | 6 ++++--
.../resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc | 6 +++---
docs/components/modules/ROOT/pages/aws2-s3-component.adoc | 6 +++---
3 files changed, 10 insertions(+), 8 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-s3.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-s3.json
index feced29..584eec8 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-s3.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-s3.json
@@ -62,7 +62,8 @@
"operation": { "kind": "property", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.AWS2S3Operations", "enum": [ "copyObject", "listObjects", "deleteObject", "deleteBucket", "listBuckets", "getObject", "getObjectRange" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": [...]
"partSize": { "kind": "property", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Setup the partSize which is used in multi part upload, the default size is 25M." },
"storageClass": { "kind": "property", "displayName": "Storage Class", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request." },
- "streamMode": { "kind": "property", "displayName": "Stream Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true the upload to bucket will be done in streaming" },
+ "streamingUploadMode": { "kind": "property", "displayName": "Streaming Upload Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true the upload to bucket will be done in streaming" },
+ "streamingUploadTimeout": { "kind": "property", "displayName": "Streaming Upload Timeout", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "While streaming upload mode is true, this option set the timeout to complete upload" },
"awsKMSKeyId": { "kind": "property", "displayName": "Aws KMSKey Id", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define the id of KMS key to use in case KMS is enabled" },
"useAwsKMS": { "kind": "property", "displayName": "Use Aws KMS", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define if KMS must be used or not" },
"useCustomerKey": { "kind": "property", "displayName": "Use Customer Key", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define if Customer Key must be used or not" },
@@ -117,7 +118,8 @@
"operation": { "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.AWS2S3Operations", "enum": [ "copyObject", "listObjects", "deleteObject", "deleteBucket", "listBuckets", "getObject", "getObjectRange" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": [...]
"partSize": { "kind": "parameter", "displayName": "Part Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 26214400, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Setup the partSize which is used in multi part upload, the default size is 25M." },
"storageClass": { "kind": "parameter", "displayName": "Storage Class", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The storage class to set in the com.amazonaws.services.s3.model.PutObjectRequest request." },
- "streamMode": { "kind": "parameter", "displayName": "Stream Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true the upload to bucket will be done in streaming" },
+ "streamingUploadMode": { "kind": "parameter", "displayName": "Streaming Upload Mode", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "When stream mode is true the upload to bucket will be done in streaming" },
+ "streamingUploadTimeout": { "kind": "parameter", "displayName": "Streaming Upload Timeout", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "While streaming upload mode is true, this option set the timeout to complete upload" },
"awsKMSKeyId": { "kind": "parameter", "displayName": "Aws KMSKey Id", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define the id of KMS key to use in case KMS is enabled" },
"useAwsKMS": { "kind": "parameter", "displayName": "Use Aws KMS", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define if KMS must be used or not" },
"useCustomerKey": { "kind": "parameter", "displayName": "Use Customer Key", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Define if Customer Key must be used or not" },
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc
index dc3fd8f..4489346 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-s3-component.adoc
@@ -555,7 +555,7 @@ This operation will return an S3Object instance related to the camelKey object i
This operation will return a download link url for the file camel-key in the bucket mycamelbucket and region region
-== Stream mode
+== Streaming Upload mode
With the stream mode enabled users will be able to upload data to S3 without knowing ahead of time the dimension of the data, by leveraging multipart upload.
The upload will be completed when: the batchSize has been completed or the batchMessageNumber has been reached.
@@ -567,11 +567,11 @@ As an example:
--------------------------------------------------------------------------------
from(kafka("topic1").brokers("localhost:9092"))
.log("Kafka Message is: ${body}")
- .to(aws2S3("camel-bucket").streamMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));
+ .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));
from(kafka("topic2").brokers("localhost:9092"))
.log("Kafka Message is: ${body}")
- .to(aws2S3("camel-bucket").streamMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic2}}/{{kafkaTopic2}}.txt"));
+ .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic2}}/{{kafkaTopic2}}.txt"));
--------------------------------------------------------------------------------
The default size for a batch is 1 Mb, but you can adjust it according to your requirements.
diff --git a/docs/components/modules/ROOT/pages/aws2-s3-component.adoc b/docs/components/modules/ROOT/pages/aws2-s3-component.adoc
index f20ba1f..c0a137d 100644
--- a/docs/components/modules/ROOT/pages/aws2-s3-component.adoc
+++ b/docs/components/modules/ROOT/pages/aws2-s3-component.adoc
@@ -557,7 +557,7 @@ This operation will return an S3Object instance related to the camelKey object i
This operation will return a download link url for the file camel-key in the bucket mycamelbucket and region region
-== Stream mode
+== Streaming Upload mode
With the stream mode enabled users will be able to upload data to S3 without knowing ahead of time the dimension of the data, by leveraging multipart upload.
The upload will be completed when: the batchSize has been completed or the batchMessageNumber has been reached.
@@ -569,11 +569,11 @@ As an example:
--------------------------------------------------------------------------------
from(kafka("topic1").brokers("localhost:9092"))
.log("Kafka Message is: ${body}")
- .to(aws2S3("camel-bucket").streamMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));
+ .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));
from(kafka("topic2").brokers("localhost:9092"))
.log("Kafka Message is: ${body}")
- .to(aws2S3("camel-bucket").streamMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic2}}/{{kafkaTopic2}}.txt"));
+ .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic2}}/{{kafkaTopic2}}.txt"));
--------------------------------------------------------------------------------
The default size for a batch is 1 Mb, but you can adjust it according to your requirements.
[camel] 07/09: CAMEL-16185 - AWS S3: improve multipart support -
streaming upload
Posted by ac...@apache.org.
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 578dfe94fbdcb4719198e2cfc6779ee01063a15a
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Apr 6 07:47:11 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 12 ++++++------
.../localstack/S3StreamUploadTimeoutLocalstackTest.java | 16 ++++++++++++++++
2 files changed, 22 insertions(+), 6 deletions(-)
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 412df8f..b96a59b 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
@@ -70,6 +70,10 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
private transient String s3ProducerToString;
private ScheduledExecutorService timeoutCheckerExecutorService;
+ public AWS2S3StreamUploadProducer(final Endpoint endpoint) {
+ super(endpoint);
+ }
+
@Override
protected void doStart() throws Exception {
super.doStart();
@@ -77,7 +81,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
timeoutCheckerExecutorService
= getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
"timeout_checker");
- timeoutCheckerExecutorService.scheduleAtFixedRate(new AggregationIntervalTask(),
+ timeoutCheckerExecutorService.scheduleAtFixedRate(new StreamingUploadTimeoutTask(),
getConfiguration().getStreamingUploadTimeout(), getConfiguration().getStreamingUploadTimeout(),
TimeUnit.MILLISECONDS);
}
@@ -104,7 +108,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
/**
* Background task that triggers completion based on interval.
*/
- private final class AggregationIntervalTask implements Runnable {
+ private final class StreamingUploadTimeoutTask implements Runnable {
@Override
public void run() {
@@ -119,10 +123,6 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
}
}
- public AWS2S3StreamUploadProducer(final Endpoint endpoint) {
- super(endpoint);
- }
-
@Override
public void process(final Exchange exchange) throws Exception {
InputStream is = exchange.getIn().getMandatoryBody(InputStream.class);
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 1601256..b84445c 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
@@ -1,3 +1,19 @@
+/*
+ * 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.aws2.s3.localstack;
import java.util.List;