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/01 12:39:49 UTC
[camel] branch master updated (09d4c81 -> d7e8e1c)
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 09d4c81 Sync deps
new a47bbc3 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new 65cce49 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new 813ab4d CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new 93e20a5 CAMEL-16185 - AWS S3: improve multipart support - streaming upload
new d7e8e1c CAMEL-16185 - AWS S3: improve multipart support - streaming upload
The 5 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 | 8 +
.../camel/catalog/docs/aws2-s3-component.adoc | 12 +-
.../aws2/s3/AWS2S3ComponentConfigurer.java | 24 ++
.../aws2/s3/AWS2S3EndpointConfigurer.java | 24 ++
.../aws2/s3/AWS2S3EndpointUriFactory.java | 6 +-
.../apache/camel/component/aws2/s3/aws2-s3.json | 8 +
.../src/main/docs/aws2-s3-component.adoc | 12 +-
.../component/aws2/s3/AWS2S3Configuration.java | 53 ++++
.../camel/component/aws2/s3/AWS2S3Endpoint.java | 7 +-
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 300 +++++++++++++++++++++
.../aws2/s3/stream/AWSS3NamingStrategyEnum.java} | 8 +-
...=> S3StreamUploadOperationIntegrationTest.java} | 32 +--
.../dsl/Aws2S3ComponentBuilderFactory.java | 66 +++++
.../endpoint/dsl/AWS2S3EndpointBuilderFactory.java | 136 ++++++++++
.../modules/ROOT/pages/aws2-s3-component.adoc | 12 +-
15 files changed, 677 insertions(+), 31 deletions(-)
create mode 100644 components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
copy components/{camel-atmos/src/main/java/org/apache/camel/component/atmos/util/AtmosResultCode.java => camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyEnum.java} (87%)
copy components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/{S3MultipartUploadOperationIntegrationTest.java => S3StreamUploadOperationIntegrationTest.java} (69%)
[camel] 01/05: 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 a47bbc3ec2b3766be516cffd801313492db3655f
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Apr 1 07:10:30 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../camel/catalog/docs/aws2-s3-component.adoc | 6 +-
.../aws2/s3/AWS2S3ComponentConfigurer.java | 6 +
.../aws2/s3/AWS2S3EndpointConfigurer.java | 6 +
.../aws2/s3/AWS2S3EndpointUriFactory.java | 3 +-
.../apache/camel/component/aws2/s3/aws2-s3.json | 2 +
.../src/main/docs/aws2-s3-component.adoc | 6 +-
.../component/aws2/s3/AWS2S3Configuration.java | 13 +
.../camel/component/aws2/s3/AWS2S3Endpoint.java | 6 +-
.../aws2/s3/AWS2S3StreamUploadProducer.java | 289 +++++++++++++++++++++
.../S3StreamUploadOperationIntegrationTest.java | 76 ++++++
.../dsl/Aws2S3ComponentBuilderFactory.java | 16 ++
.../endpoint/dsl/AWS2S3EndpointBuilderFactory.java | 93 +++++++
.../modules/ROOT/pages/aws2-s3-component.adoc | 6 +-
13 files changed, 520 insertions(+), 8 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 78bcbae..e497bde 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 43 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 44 options, which are listed below.
@@ -65,6 +65,7 @@ The AWS 2 S3 Storage Service component supports 43 options, which are listed bel
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
+| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -129,7 +130,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (61 parameters):
+=== Query Parameters (62 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -145,6 +146,7 @@ with the following path and query parameters:
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
+| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | 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 77bb7d0..6f0e69b 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
@@ -98,6 +98,8 @@ 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).setStreamMode(property(camelContext, boolean.class, value)); return true;
case "trustallcertificates":
case "trustAllCertificates": getOrCreateConfiguration(target).setTrustAllCertificates(property(camelContext, boolean.class, value)); return true;
case "uriendpointoverride":
@@ -190,6 +192,8 @@ 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 "trustallcertificates":
case "trustAllCertificates": return boolean.class;
case "uriendpointoverride":
@@ -278,6 +282,8 @@ 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).isStreamMode();
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 30d87c0..255f2bb 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
@@ -121,6 +121,8 @@ 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().setStreamMode(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":
@@ -247,6 +249,8 @@ 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 "timeunit":
case "timeUnit": return java.util.concurrent.TimeUnit.class;
case "trustallcertificates":
@@ -369,6 +373,8 @@ 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().isStreamMode();
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 f88438a..63752dd 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,10 +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<>(62);
+ Set<String> props = new HashSet<>(63);
props.add("customerAlgorithm");
props.add("fileName");
props.add("useCustomerKey");
+ props.add("streamMode");
props.add("bucketNameOrArn");
props.add("customerKeyId");
props.add("prefix");
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 d2cbe25..4aab19d 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
@@ -33,6 +33,7 @@
"proxyPort": { "kind": "property", "displayName": "Proxy Port", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Specify a proxy port to be used inside the client definition." },
"proxyProtocol": { "kind": "property", "displayName": "Proxy Protocol", "group": "common", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "To define a proxy protocol when instantiating th [...]
"region": { "kind": "property", "displayName": "Region", "group": "common", "label": "", "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 region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (f [...]
+ "streamMode": { "kind": "property", "displayName": "Stream Mode", "group": "common", "label": "", "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": "If StreaMode is true a different way of uploading will be provided" },
"trustAllCertificates": { "kind": "property", "displayName": "Trust All Certificates", "group": "common", "label": "", "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": "If we want to trust all certificates in case of overriding the endpoint" },
"uriEndpointOverride": { "kind": "property", "displayName": "Uri Endpoint Override", "group": "common", "label": "", "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": "Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option" },
"useDefaultCredentialsProvider": { "kind": "property", "displayName": "Use Default Credentials Provider", "group": "common", "label": "", "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": "Set whether the S3 client should expect to load credentials through a def [...]
@@ -78,6 +79,7 @@
"proxyPort": { "kind": "parameter", "displayName": "Proxy Port", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Specify a proxy port to be used inside the client definition." },
"proxyProtocol": { "kind": "parameter", "displayName": "Proxy Protocol", "group": "common", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "To define a proxy protocol when instantiating t [...]
"region": { "kind": "parameter", "displayName": "Region", "group": "common", "label": "", "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 region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region ( [...]
+ "streamMode": { "kind": "parameter", "displayName": "Stream Mode", "group": "common", "label": "", "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": "If StreaMode is true a different way of uploading will be provided" },
"trustAllCertificates": { "kind": "parameter", "displayName": "Trust All Certificates", "group": "common", "label": "", "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": "If we want to trust all certificates in case of overriding the endpoint" },
"uriEndpointOverride": { "kind": "parameter", "displayName": "Uri Endpoint Override", "group": "common", "label": "", "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": "Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option" },
"useDefaultCredentialsProvider": { "kind": "parameter", "displayName": "Use Default Credentials Provider", "group": "common", "label": "", "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": "Set whether the S3 client should expect to load credentials through a de [...]
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 78bcbae..e497bde 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 43 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 44 options, which are listed below.
@@ -65,6 +65,7 @@ The AWS 2 S3 Storage Service component supports 43 options, which are listed bel
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
+| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -129,7 +130,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (61 parameters):
+=== Query Parameters (62 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -145,6 +146,7 @@ with the following path and query parameters:
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
+| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
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 439aa56..aec9bf7 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
@@ -110,6 +110,8 @@ public class AWS2S3Configuration implements Cloneable {
private String uriEndpointOverride;
@UriParam(defaultValue = "false")
private boolean pojoRequest;
+ @UriParam(defaultValue = "false")
+ private boolean streamMode;
public long getPartSize() {
return partSize;
@@ -575,6 +577,17 @@ public class AWS2S3Configuration implements Cloneable {
this.amazonS3Presigner = amazonS3Presigner;
}
+ public boolean isStreamMode() {
+ return streamMode;
+ }
+
+ /**
+ * If StreaMode is true a different way of uploading will be provided
+ */
+ public void setStreamMode(boolean streamMode) {
+ this.streamMode = streamMode;
+ }
+
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 9cad4ee..f8aa480 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
@@ -72,7 +72,11 @@ public class AWS2S3Endpoint extends ScheduledPollEndpoint {
@Override
public Producer createProducer() throws Exception {
- return new AWS2S3Producer(this);
+ if (!configuration.isStreamMode()) {
+ return new AWS2S3Producer(this);
+ } else {
+ return new AWS2S3StreamUploadProducer(this);
+ }
}
@Override
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3StreamUploadProducer.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3StreamUploadProducer.java
new file mode 100644
index 0000000..63843d3
--- /dev/null
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3StreamUploadProducer.java
@@ -0,0 +1,289 @@
+/*
+ * 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;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.URISupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.model.*;
+import software.amazon.awssdk.utils.IoUtils;
+
+/**
+ * A Producer which sends messages to the Amazon Web Service Simple Storage Service
+ * <a href="http://aws.amazon.com/s3/">AWS S3</a>
+ */
+public class AWS2S3StreamUploadProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AWS2S3StreamUploadProducer.class);
+
+ private transient String s3ProducerToString;
+
+ ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ CreateMultipartUploadResponse initResponse;
+ int index = 1;
+ List<CompletedPart> completedParts = new ArrayList<CompletedPart>();
+ int part = 0;
+
+ public AWS2S3StreamUploadProducer(final Endpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public void process(final Exchange exchange) throws Exception {
+ streamUpload(exchange);
+ }
+
+ public void streamUpload(final Exchange exchange) throws Exception {
+ File filePayload = null;
+ InputStream is = exchange.getIn().getMandatoryBody(InputStream.class);
+
+ buffer.write(IoUtils.toByteArray(is));
+ final String keyName = determineKey(exchange);
+ String dynamicKeyName;
+ if (part > 0) {
+ dynamicKeyName = keyName + "-" + part;
+ } else {
+ dynamicKeyName = keyName;
+ }
+ CreateMultipartUploadRequest.Builder createMultipartUploadRequest
+ = CreateMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName()).key(dynamicKeyName);
+
+ String storageClass = determineStorageClass(exchange);
+ if (storageClass != null) {
+ createMultipartUploadRequest.storageClass(storageClass);
+ }
+
+ String cannedAcl = exchange.getIn().getHeader(AWS2S3Constants.CANNED_ACL, String.class);
+ if (cannedAcl != null) {
+ ObjectCannedACL objectAcl = ObjectCannedACL.valueOf(cannedAcl);
+ createMultipartUploadRequest.acl(objectAcl);
+ }
+
+ BucketCannedACL acl = exchange.getIn().getHeader(AWS2S3Constants.ACL, BucketCannedACL.class);
+ if (acl != null) {
+ // note: if cannedacl and acl are both specified the last one will
+ // be used. refer to
+ // PutObjectRequest#setAccessControlList for more details
+ createMultipartUploadRequest.acl(acl.toString());
+ }
+
+ if (getConfiguration().isUseAwsKMS()) {
+ createMultipartUploadRequest.ssekmsKeyId(getConfiguration().getAwsKMSKeyId());
+ createMultipartUploadRequest.serverSideEncryption(ServerSideEncryption.AWS_KMS);
+ }
+
+ if (getConfiguration().isUseCustomerKey()) {
+ if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyId())) {
+ createMultipartUploadRequest.sseCustomerKey(getConfiguration().getCustomerKeyId());
+ }
+ if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyMD5())) {
+ createMultipartUploadRequest.sseCustomerKeyMD5(getConfiguration().getCustomerKeyMD5());
+ }
+ if (ObjectHelper.isNotEmpty(getConfiguration().getCustomerAlgorithm())) {
+ createMultipartUploadRequest.sseCustomerAlgorithm(getConfiguration().getCustomerAlgorithm());
+ }
+ }
+
+ LOG.trace("Initiating multipart upload [{}] from exchange [{}]...", createMultipartUploadRequest, exchange);
+ CompleteMultipartUploadResponse uploadResult = null;
+ if (index == 1) {
+ initResponse
+ = getEndpoint().getS3Client().createMultipartUpload(createMultipartUploadRequest.build());
+ //final long contentLength = Long.valueOf(objectMetadata.get("Content-Length"));
+ completedParts = new ArrayList<CompletedPart>();
+ long partSize = getConfiguration().getPartSize();
+ }
+
+ long filePosition = 0;
+
+ try {
+ if (buffer.size() >= 5000000 || index == 100) {
+
+ UploadPartRequest uploadRequest = UploadPartRequest.builder().bucket(getConfiguration().getBucketName())
+ .key(dynamicKeyName).uploadId(initResponse.uploadId())
+ .partNumber(index).build();
+
+ LOG.trace("Uploading part [{}] for {}", index, keyName);
+
+ String etag = getEndpoint().getS3Client()
+ .uploadPart(uploadRequest, RequestBody.fromBytes(buffer.toByteArray())).eTag();
+ CompletedPart partUpload = CompletedPart.builder().partNumber(index).eTag(etag).build();
+ completedParts.add(partUpload);
+ buffer.reset();
+ part++;
+ }
+
+ if (index == 100) {
+ 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);
+ index = 0;
+ }
+
+ } catch (Exception e) {
+ getEndpoint().getS3Client()
+ .abortMultipartUpload(AbortMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName())
+ .key(dynamicKeyName).uploadId(initResponse.uploadId()).build());
+ throw e;
+ }
+
+ index++;
+
+ Message message = getMessageForResponse(exchange);
+
+ }
+
+ private AWS2S3Operations determineOperation(Exchange exchange) {
+ AWS2S3Operations operation = exchange.getIn().getHeader(AWS2S3Constants.S3_OPERATION, AWS2S3Operations.class);
+ if (operation == null) {
+ operation = getConfiguration().getOperation();
+ }
+ return operation;
+ }
+
+ private Map<String, String> determineMetadata(final Exchange exchange) {
+ Map<String, String> objectMetadata = new HashMap<String, String>();
+
+ Long contentLength = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_LENGTH, Long.class);
+ if (contentLength != null) {
+ objectMetadata.put("Content-Length", String.valueOf(contentLength));
+ }
+
+ String contentType = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_TYPE, String.class);
+ if (contentType != null) {
+ objectMetadata.put("Content-Type", String.valueOf(contentType));
+ }
+
+ String cacheControl = exchange.getIn().getHeader(AWS2S3Constants.CACHE_CONTROL, String.class);
+ if (cacheControl != null) {
+ objectMetadata.put("Cache-Control", String.valueOf(cacheControl));
+ }
+
+ String contentDisposition = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_DISPOSITION, String.class);
+ if (contentDisposition != null) {
+ objectMetadata.put("Content-Disposition", String.valueOf(contentDisposition));
+ }
+
+ String contentEncoding = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_ENCODING, String.class);
+ if (contentEncoding != null) {
+ objectMetadata.put("Content-Encoding", String.valueOf(contentEncoding));
+ }
+
+ String contentMD5 = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_MD5, String.class);
+ if (contentMD5 != null) {
+ objectMetadata.put("Content-Md5", String.valueOf(contentMD5));
+ }
+
+ return objectMetadata;
+ }
+
+ /**
+ * Reads the bucket name from the header of the given exchange. If not provided, it's read from the endpoint
+ * configuration.
+ *
+ * @param exchange The exchange to read the header from.
+ * @return The bucket name.
+ * @throws IllegalArgumentException if the header could not be determined.
+ */
+ private String determineBucketName(final Exchange exchange) {
+ String bucketName = exchange.getIn().getHeader(AWS2S3Constants.BUCKET_NAME, String.class);
+
+ if (ObjectHelper.isEmpty(bucketName)) {
+ bucketName = getConfiguration().getBucketName();
+ LOG.trace("AWS S3 Bucket name header is missing, using default one [{}]", bucketName);
+ }
+
+ if (bucketName == null) {
+ throw new IllegalArgumentException("AWS S3 Bucket name header is missing or not configured.");
+ }
+
+ return bucketName;
+ }
+
+ private String determineKey(final Exchange exchange) {
+ String key = exchange.getIn().getHeader(AWS2S3Constants.KEY, String.class);
+ if (ObjectHelper.isEmpty(key)) {
+ key = getConfiguration().getKeyName();
+ }
+ if (key == null) {
+ throw new IllegalArgumentException("AWS S3 Key header missing.");
+ }
+ return key;
+ }
+
+ private String determineStorageClass(final Exchange exchange) {
+ String storageClass = exchange.getIn().getHeader(AWS2S3Constants.STORAGE_CLASS, String.class);
+ if (storageClass == null) {
+ storageClass = getConfiguration().getStorageClass();
+ }
+
+ return storageClass;
+ }
+
+ private ByteArrayOutputStream determineLengthInputStream(InputStream is) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ byte[] bytes = new byte[1024];
+ int count;
+ while ((count = is.read(bytes)) > 0) {
+ out.write(bytes, 0, count);
+ }
+ return out;
+ }
+
+ protected AWS2S3Configuration getConfiguration() {
+ return getEndpoint().getConfiguration();
+ }
+
+ @Override
+ public String toString() {
+ if (s3ProducerToString == null) {
+ s3ProducerToString = "S3Producer[" + URISupport.sanitizeUri(getEndpoint().getEndpointUri()) + "]";
+ }
+ return s3ProducerToString;
+ }
+
+ @Override
+ public AWS2S3Endpoint getEndpoint() {
+ return (AWS2S3Endpoint) super.getEndpoint();
+ }
+
+ public static Message getMessageForResponse(final Exchange exchange) {
+ return exchange.getMessage();
+ }
+
+}
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
new file mode 100644
index 0000000..e567caf
--- /dev/null
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.integration;
+
+import org.apache.camel.*;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+
+@Disabled("Must be manually tested. Provide your own accessKey and secretKey!")
+public class S3StreamUploadOperationIntegrationTest extends CamelTestSupport {
+
+ @BindToRegistry("amazonS3Client")
+ S3Client client
+ = S3Client.builder()
+ .credentialsProvider(StaticCredentialsProvider.create(
+ AwsBasicCredentials.create("xxxx", "yyyy")))
+ .region(Region.EU_WEST_1).build();
+
+ @EndpointInject
+ private ProducerTemplate template;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint result;
+
+ @Test
+ public void sendIn() throws Exception {
+ result.expectedMessageCount(1);
+
+ for (int i = 0; i < 1000; i++) {
+ template.sendBody("direct:stream1", "Andrea\n");
+ }
+ for (int i = 0; i < 1000; i++) {
+ template.sendBody("direct:stream2", "Luca\n");
+ }
+
+ Thread.sleep(30000);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ String awsEndpoint1
+ = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTest";
+ String awsEndpoint2
+ = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTestParallel";
+
+ from("direct:stream1").toD(awsEndpoint1).to("mock:result");
+ from("direct:stream2").toD(awsEndpoint2).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 4cfa061..c5e6b6e 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
@@ -228,6 +228,21 @@ public interface Aws2S3ComponentBuilderFactory {
return this;
}
/**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: common
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default Aws2S3ComponentBuilder streamMode(boolean streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
* If we want to trust all certificates in case of overriding the
* endpoint.
*
@@ -801,6 +816,7 @@ public interface Aws2S3ComponentBuilderFactory {
case "proxyPort": getOrCreateConfiguration((AWS2S3Component) component).setProxyPort((java.lang.Integer) value); return true;
case "proxyProtocol": getOrCreateConfiguration((AWS2S3Component) component).setProxyProtocol((software.amazon.awssdk.core.Protocol) value); return true;
case "region": getOrCreateConfiguration((AWS2S3Component) component).setRegion((java.lang.String) value); return true;
+ case "streamMode": getOrCreateConfiguration((AWS2S3Component) component).setStreamMode((boolean) value); return true;
case "trustAllCertificates": getOrCreateConfiguration((AWS2S3Component) component).setTrustAllCertificates((boolean) value); return true;
case "uriEndpointOverride": getOrCreateConfiguration((AWS2S3Component) component).setUriEndpointOverride((java.lang.String) value); return true;
case "useDefaultCredentialsProvider": getOrCreateConfiguration((AWS2S3Component) component).setUseDefaultCredentialsProvider((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 16396386..0ae939c 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
@@ -328,6 +328,37 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: common
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointConsumerBuilder streamMode(boolean streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option will be converted to a <code>boolean</code>
+ * type.
+ *
+ * Default: false
+ * Group: common
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointConsumerBuilder streamMode(String streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
* If we want to trust all certificates in case of overriding the
* endpoint.
*
@@ -1849,6 +1880,37 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: common
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder streamMode(boolean streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option will be converted to a <code>boolean</code>
+ * type.
+ *
+ * Default: false
+ * Group: common
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder streamMode(String streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
* If we want to trust all certificates in case of overriding the
* endpoint.
*
@@ -2594,6 +2656,37 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: common
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointBuilder streamMode(boolean streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option will be converted to a <code>boolean</code>
+ * type.
+ *
+ * Default: false
+ * Group: common
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointBuilder streamMode(String streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
* If we want to trust all certificates in case of overriding the
* endpoint.
*
diff --git a/docs/components/modules/ROOT/pages/aws2-s3-component.adoc b/docs/components/modules/ROOT/pages/aws2-s3-component.adoc
index d3d355f..579bf3f 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 43 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 44 options, which are listed below.
@@ -67,6 +67,7 @@ The AWS 2 S3 Storage Service component supports 43 options, which are listed bel
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
+| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -131,7 +132,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (61 parameters):
+=== Query Parameters (62 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -147,6 +148,7 @@ with the following path and query parameters:
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
+| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
[camel] 03/05: 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 813ab4d39062e1d5b3ede30e6547db0fde3cca40
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Apr 1 14:18:06 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 58 ++++++++++++----------
.../S3StreamUploadOperationIntegrationTest.java | 5 +-
2 files changed, 35 insertions(+), 28 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 546128e..21d264c 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
@@ -22,6 +22,7 @@ import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
@@ -50,9 +51,10 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
CreateMultipartUploadResponse initResponse;
- int index = 1;
- List<CompletedPart> completedParts = new ArrayList<>();
- int part = 0;
+ AtomicInteger index = new AtomicInteger();
+ List<CompletedPart> completedParts;
+ AtomicInteger part = new AtomicInteger();
+ UUID id = null;
public AWS2S3StreamUploadProducer(final Endpoint endpoint) {
super(endpoint);
@@ -67,7 +69,11 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
final String keyName = getConfiguration().getKeyName();
final String fileName = determineFileName(keyName);
final String extension = determineFileExtension(keyName);
- String dynamicKeyName = fileNameToUpload(fileName, getConfiguration().getNamingStrategy(), extension, part);
+ String dynamicKeyName;
+ if (index.get() == 1 && getConfiguration().getNamingStrategy().equals(AWSS3NamingStrategyEnum.random)) {
+ id = UUID.randomUUID();
+ }
+ dynamicKeyName = fileNameToUpload(fileName, getConfiguration().getNamingStrategy(), extension, part, id);
CreateMultipartUploadRequest.Builder createMultipartUploadRequest
= CreateMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName()).key(dynamicKeyName);
@@ -108,33 +114,32 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
}
LOG.trace("Initiating multipart upload [{}] from exchange [{}]...", createMultipartUploadRequest, exchange);
- CompleteMultipartUploadResponse uploadResult = null;
- if (index == 1) {
+ if (index.get() == 1) {
initResponse
= getEndpoint().getS3Client().createMultipartUpload(createMultipartUploadRequest.build());
//final long contentLength = Long.valueOf(objectMetadata.get("Content-Length"));
- completedParts = new ArrayList<CompletedPart>();
- long partSize = getConfiguration().getPartSize();
+ completedParts = new ArrayList<>();
}
try {
- if (buffer.size() >= getConfiguration().getBatchSize() || index == getConfiguration().getBatchMessageNumber()) {
+ if (buffer.size() >= getConfiguration().getBatchSize()
+ || index.get() == getConfiguration().getBatchMessageNumber()) {
UploadPartRequest uploadRequest = UploadPartRequest.builder().bucket(getConfiguration().getBucketName())
.key(dynamicKeyName).uploadId(initResponse.uploadId())
- .partNumber(index).build();
+ .partNumber(index.get()).build();
LOG.trace("Uploading part [{}] for {}", index, keyName);
String etag = getEndpoint().getS3Client()
.uploadPart(uploadRequest, RequestBody.fromBytes(buffer.toByteArray())).eTag();
- CompletedPart partUpload = CompletedPart.builder().partNumber(index).eTag(etag).build();
+ CompletedPart partUpload = CompletedPart.builder().partNumber(index.get()).eTag(etag).build();
completedParts.add(partUpload);
buffer.reset();
- part++;
+ part.getAndIncrement();
}
- if (index == getConfiguration().getBatchMessageNumber()) {
+ if (index.get() == getConfiguration().getBatchMessageNumber()) {
CompletedMultipartUpload completeMultipartUpload
= CompletedMultipartUpload.builder().parts(completedParts).build();
CompleteMultipartUploadRequest compRequest
@@ -143,8 +148,13 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
.uploadId(initResponse.uploadId())
.build();
- getEndpoint().getS3Client().completeMultipartUpload(compRequest);
- index = 0;
+ CompleteMultipartUploadResponse uploadResult = getEndpoint().getS3Client().completeMultipartUpload(compRequest);
+ Message message = getMessageForResponse(exchange);
+ message.setHeader(AWS2S3Constants.E_TAG, uploadResult.eTag());
+ if (uploadResult.versionId() != null) {
+ message.setHeader(AWS2S3Constants.VERSION_ID, uploadResult.versionId());
+ }
+ index.getAndSet(0);
}
} catch (Exception e) {
@@ -154,16 +164,15 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
throw e;
}
- index++;
-
- Message message = getMessageForResponse(exchange);
+ index.getAndIncrement();
}
- private String fileNameToUpload(String fileName, AWSS3NamingStrategyEnum strategy, String ext, int part) {
+ private String fileNameToUpload(
+ String fileName, AWSS3NamingStrategyEnum strategy, String ext, AtomicInteger part, UUID id) {
String dynamicKeyName;
switch (strategy) {
case progressive:
- if (part > 0) {
+ if (part.get() > 0) {
if (ObjectHelper.isNotEmpty(ext)) {
dynamicKeyName = fileName + "-" + part + ext;
} else {
@@ -178,8 +187,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
}
break;
case random:
- if (part > 0) {
- UUID id = UUID.randomUUID();
+ if (part.get() > 0) {
if (ObjectHelper.isNotEmpty(ext)) {
dynamicKeyName = fileName + "-" + id.toString() + ext;
} else {
@@ -236,8 +244,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
if (extPosition == -1) {
return "";
} else {
- String ext = keyName.substring(extPosition);
- return ext;
+ return keyName.substring(extPosition);
}
}
@@ -246,8 +253,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
if (extPosition == -1) {
return keyName;
} else {
- String prefix = keyName.substring(0, extPosition);
- return prefix;
+ return keyName.substring(0, extPosition);
}
}
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
index 66127c1..9e27023 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
@@ -45,13 +45,14 @@ public class S3StreamUploadOperationIntegrationTest extends CamelTestSupport {
@Test
public void sendIn() throws Exception {
- result.expectedMessageCount(1);
+ result.expectedMessageCount(1000);
for (int i = 0; i < 1000; i++) {
template.sendBody("direct:stream1", "Andrea\n");
}
Thread.sleep(30000);
+ assertMockEndpointsSatisfied();
}
@Override
@@ -60,7 +61,7 @@ public class S3StreamUploadOperationIntegrationTest extends CamelTestSupport {
@Override
public void configure() throws Exception {
String awsEndpoint1
- = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTest&batchMessageNumber=25&namingStrategy=progressive";
+ = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTest.txt&batchMessageNumber=25&namingStrategy=random";
from("direct:stream1").toD(awsEndpoint1).to("mock:result");
}
[camel] 02/05: 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 65cce499533b348b21d35680f12d64eb95b5bdfe
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Apr 1 12:24:37 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../camel/catalog/docs/aws2-s3-component.adoc | 14 +-
.../aws2/s3/AWS2S3ComponentConfigurer.java | 18 ++
.../aws2/s3/AWS2S3EndpointConfigurer.java | 18 ++
.../aws2/s3/AWS2S3EndpointUriFactory.java | 5 +-
.../apache/camel/component/aws2/s3/aws2-s3.json | 10 +-
.../src/main/docs/aws2-s3-component.adoc | 14 +-
.../component/aws2/s3/AWS2S3Configuration.java | 42 +++-
.../camel/component/aws2/s3/AWS2S3Endpoint.java | 1 +
.../{ => stream}/AWS2S3StreamUploadProducer.java | 146 +++++++------
.../aws2/s3/stream/AWSS3NamingStrategyEnum.java | 6 +
.../S3StreamUploadOperationIntegrationTest.java | 8 +-
.../dsl/Aws2S3ComponentBuilderFactory.java | 82 ++++++--
.../endpoint/dsl/AWS2S3EndpointBuilderFactory.java | 229 ++++++++++++---------
.../modules/ROOT/pages/aws2-s3-component.adoc | 14 +-
14 files changed, 400 insertions(+), 207 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 e497bde..16de086 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 44 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 47 options, which are listed below.
@@ -65,7 +65,6 @@ The AWS 2 S3 Storage Service component supports 44 options, which are listed bel
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
-| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -85,13 +84,17 @@ The AWS 2 S3 Storage Service component supports 44 options, which are listed bel
| *moveAfterRead* (consumer) | Move objects from S3 bucket to a different bucket after they have been retrieved. To accomplish the operation the destinationBucket option must be set. The copy bucket operation is only performed if the Exchange is committed. If a rollback occurs, the object is not moved. | false | boolean
| *prefix* (consumer) | The prefix which is used in the com.amazonaws.services.s3.model.ListObjectsRequest to only consume objects we are interested in. | | String
| *autocloseBody* (consumer) | If this option is true and includeBody is false, then the S3Object.close() method will be called on exchange completion. This option is strongly related to includeBody option. In case of setting includeBody to false and autocloseBody to false, it will be up to the caller to close the S3Object stream. Setting autocloseBody to true, will close the S3Object stream automatically. | true | boolean
+| *batchMessageNumber* (producer) | The number of messages composing a batch in stream mode | 10 | int
+| *batchSize* (producer) | The batch size in stream mode | 1000000 | int
| *deleteAfterWrite* (producer) | Delete file object after the S3 file has been uploaded | false | boolean
| *keyName* (producer) | Setting the key name for an element in the bucket through endpoint parameter | | String
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
| *multiPartUpload* (producer) | If it is true, camel will upload the file with multi part format, the part size is decided by the option of partSize | false | boolean
+| *namingStrategy* (producer) | The naming strategy to use. There are 2 enums and the value can be one of: progressive, random | progressive | AWSS3NamingStrategyEnum
| *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) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *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
@@ -130,7 +133,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (62 parameters):
+=== Query Parameters (65 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -146,7 +149,6 @@ with the following path and query parameters:
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
-| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -172,13 +174,17 @@ with the following path and query parameters:
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. There are 3 enums and the value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
| *pollStrategy* (consumer) | A pluggable org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your custom implementation to control error handling usually occurred during the poll operation before an Exchange have been created and being routed in Camel. | | PollingConsumerPollStrategy
+| *batchMessageNumber* (producer) | The number of messages composing a batch in stream mode | 10 | int
+| *batchSize* (producer) | The batch size in stream mode | 1000000 | int
| *deleteAfterWrite* (producer) | Delete file object after the S3 file has been uploaded | false | boolean
| *keyName* (producer) | Setting the key name for an element in the bucket through endpoint parameter | | String
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
| *multiPartUpload* (producer) | If it is true, camel will upload the file with multi part format, the part size is decided by the option of partSize | false | boolean
+| *namingStrategy* (producer) | The naming strategy to use. There are 2 enums and the value can be one of: progressive, random | progressive | AWSS3NamingStrategyEnum
| *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) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *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 6f0e69b..182f1b9 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
@@ -42,6 +42,10 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "autowiredEnabled": target.setAutowiredEnabled(property(camelContext, boolean.class, value)); return true;
case "awskmskeyid":
case "awsKMSKeyId": getOrCreateConfiguration(target).setAwsKMSKeyId(property(camelContext, java.lang.String.class, value)); return true;
+ case "batchmessagenumber":
+ case "batchMessageNumber": getOrCreateConfiguration(target).setBatchMessageNumber(property(camelContext, int.class, value)); return true;
+ case "batchsize":
+ case "batchSize": getOrCreateConfiguration(target).setBatchSize(property(camelContext, int.class, value)); return true;
case "bridgeerrorhandler":
case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
case "configuration": target.setConfiguration(property(camelContext, org.apache.camel.component.aws2.s3.AWS2S3Configuration.class, value)); return true;
@@ -78,6 +82,8 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "moveAfterRead": getOrCreateConfiguration(target).setMoveAfterRead(property(camelContext, boolean.class, value)); return true;
case "multipartupload":
case "multiPartUpload": getOrCreateConfiguration(target).setMultiPartUpload(property(camelContext, boolean.class, value)); return true;
+ case "namingstrategy":
+ case "namingStrategy": getOrCreateConfiguration(target).setNamingStrategy(property(camelContext, org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum.class, value)); return true;
case "operation": getOrCreateConfiguration(target).setOperation(property(camelContext, org.apache.camel.component.aws2.s3.AWS2S3Operations.class, value)); return true;
case "overrideendpoint":
case "overrideEndpoint": getOrCreateConfiguration(target).setOverrideEndpoint(property(camelContext, boolean.class, value)); return true;
@@ -136,6 +142,10 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "autowiredEnabled": return boolean.class;
case "awskmskeyid":
case "awsKMSKeyId": return java.lang.String.class;
+ case "batchmessagenumber":
+ case "batchMessageNumber": return int.class;
+ case "batchsize":
+ case "batchSize": return int.class;
case "bridgeerrorhandler":
case "bridgeErrorHandler": return boolean.class;
case "configuration": return org.apache.camel.component.aws2.s3.AWS2S3Configuration.class;
@@ -172,6 +182,8 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "moveAfterRead": return boolean.class;
case "multipartupload":
case "multiPartUpload": return boolean.class;
+ case "namingstrategy":
+ case "namingStrategy": return org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum.class;
case "operation": return org.apache.camel.component.aws2.s3.AWS2S3Operations.class;
case "overrideendpoint":
case "overrideEndpoint": return boolean.class;
@@ -226,6 +238,10 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "autowiredEnabled": return target.isAutowiredEnabled();
case "awskmskeyid":
case "awsKMSKeyId": return getOrCreateConfiguration(target).getAwsKMSKeyId();
+ case "batchmessagenumber":
+ case "batchMessageNumber": return getOrCreateConfiguration(target).getBatchMessageNumber();
+ case "batchsize":
+ case "batchSize": return getOrCreateConfiguration(target).getBatchSize();
case "bridgeerrorhandler":
case "bridgeErrorHandler": return target.isBridgeErrorHandler();
case "configuration": return target.getConfiguration();
@@ -262,6 +278,8 @@ public class AWS2S3ComponentConfigurer extends PropertyConfigurerSupport impleme
case "moveAfterRead": return getOrCreateConfiguration(target).isMoveAfterRead();
case "multipartupload":
case "multiPartUpload": return getOrCreateConfiguration(target).isMultiPartUpload();
+ case "namingstrategy":
+ case "namingStrategy": return getOrCreateConfiguration(target).getNamingStrategy();
case "operation": return getOrCreateConfiguration(target).getOperation();
case "overrideendpoint":
case "overrideEndpoint": return getOrCreateConfiguration(target).isOverrideEndpoint();
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 255f2bb..19f618f 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
@@ -39,6 +39,10 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "backoffIdleThreshold": target.setBackoffIdleThreshold(property(camelContext, int.class, value)); return true;
case "backoffmultiplier":
case "backoffMultiplier": target.setBackoffMultiplier(property(camelContext, int.class, value)); return true;
+ case "batchmessagenumber":
+ case "batchMessageNumber": target.getConfiguration().setBatchMessageNumber(property(camelContext, int.class, value)); return true;
+ case "batchsize":
+ case "batchSize": target.getConfiguration().setBatchSize(property(camelContext, int.class, value)); return true;
case "bridgeerrorhandler":
case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
case "customeralgorithm":
@@ -86,6 +90,8 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "moveAfterRead": target.getConfiguration().setMoveAfterRead(property(camelContext, boolean.class, value)); return true;
case "multipartupload":
case "multiPartUpload": target.getConfiguration().setMultiPartUpload(property(camelContext, boolean.class, value)); return true;
+ case "namingstrategy":
+ case "namingStrategy": target.getConfiguration().setNamingStrategy(property(camelContext, org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum.class, value)); return true;
case "operation": target.getConfiguration().setOperation(property(camelContext, org.apache.camel.component.aws2.s3.AWS2S3Operations.class, value)); return true;
case "overrideendpoint":
case "overrideEndpoint": target.getConfiguration().setOverrideEndpoint(property(camelContext, boolean.class, value)); return true;
@@ -167,6 +173,10 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "backoffIdleThreshold": return int.class;
case "backoffmultiplier":
case "backoffMultiplier": return int.class;
+ case "batchmessagenumber":
+ case "batchMessageNumber": return int.class;
+ case "batchsize":
+ case "batchSize": return int.class;
case "bridgeerrorhandler":
case "bridgeErrorHandler": return boolean.class;
case "customeralgorithm":
@@ -214,6 +224,8 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "moveAfterRead": return boolean.class;
case "multipartupload":
case "multiPartUpload": return boolean.class;
+ case "namingstrategy":
+ case "namingStrategy": return org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum.class;
case "operation": return org.apache.camel.component.aws2.s3.AWS2S3Operations.class;
case "overrideendpoint":
case "overrideEndpoint": return boolean.class;
@@ -291,6 +303,10 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "backoffIdleThreshold": return target.getBackoffIdleThreshold();
case "backoffmultiplier":
case "backoffMultiplier": return target.getBackoffMultiplier();
+ case "batchmessagenumber":
+ case "batchMessageNumber": return target.getConfiguration().getBatchMessageNumber();
+ case "batchsize":
+ case "batchSize": return target.getConfiguration().getBatchSize();
case "bridgeerrorhandler":
case "bridgeErrorHandler": return target.isBridgeErrorHandler();
case "customeralgorithm":
@@ -338,6 +354,8 @@ public class AWS2S3EndpointConfigurer extends PropertyConfigurerSupport implemen
case "moveAfterRead": return target.getConfiguration().isMoveAfterRead();
case "multipartupload":
case "multiPartUpload": return target.getConfiguration().isMultiPartUpload();
+ case "namingstrategy":
+ case "namingStrategy": return target.getConfiguration().getNamingStrategy();
case "operation": return target.getConfiguration().getOperation();
case "overrideendpoint":
case "overrideEndpoint": return target.getConfiguration().isOverrideEndpoint();
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 63752dd..e6c1ba0 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,9 +20,10 @@ 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<>(63);
+ Set<String> props = new HashSet<>(66);
props.add("customerAlgorithm");
props.add("fileName");
+ props.add("batchMessageNumber");
props.add("useCustomerKey");
props.add("streamMode");
props.add("bucketNameOrArn");
@@ -83,7 +84,9 @@ public class AWS2S3EndpointUriFactory extends org.apache.camel.support.component
props.add("pollStrategy");
props.add("overrideEndpoint");
props.add("deleteAfterRead");
+ props.add("batchSize");
props.add("operation");
+ props.add("namingStrategy");
PROPERTY_NAMES = Collections.unmodifiableSet(props);
Set<String> secretProps = new HashSet<>(2);
secretProps.add("secretKey");
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 4aab19d..09a499e 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
@@ -33,7 +33,6 @@
"proxyPort": { "kind": "property", "displayName": "Proxy Port", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Specify a proxy port to be used inside the client definition." },
"proxyProtocol": { "kind": "property", "displayName": "Proxy Protocol", "group": "common", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "To define a proxy protocol when instantiating th [...]
"region": { "kind": "property", "displayName": "Region", "group": "common", "label": "", "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 region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (f [...]
- "streamMode": { "kind": "property", "displayName": "Stream Mode", "group": "common", "label": "", "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": "If StreaMode is true a different way of uploading will be provided" },
"trustAllCertificates": { "kind": "property", "displayName": "Trust All Certificates", "group": "common", "label": "", "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": "If we want to trust all certificates in case of overriding the endpoint" },
"uriEndpointOverride": { "kind": "property", "displayName": "Uri Endpoint Override", "group": "common", "label": "", "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": "Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option" },
"useDefaultCredentialsProvider": { "kind": "property", "displayName": "Use Default Credentials Provider", "group": "common", "label": "", "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": "Set whether the S3 client should expect to load credentials through a def [...]
@@ -53,13 +52,17 @@
"moveAfterRead": { "kind": "property", "displayName": "Move After Read", "group": "consumer", "label": "consumer", "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": "Move objects from S3 bucket to a different bucket after they have been retrieved. To accomplish [...]
"prefix": { "kind": "property", "displayName": "Prefix", "group": "consumer", "label": "consumer", "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 prefix which is used in the com.amazonaws.services.s3.model.ListObjectsRequest to only consume objects we are interested in." },
"autocloseBody": { "kind": "property", "displayName": "Autoclose Body", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "If this option is true and includeBody is false, then the S3Object.close() met [...]
+ "batchMessageNumber": { "kind": "property", "displayName": "Batch Message Number", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The number of messages composing a batch in stream mode" },
+ "batchSize": { "kind": "property", "displayName": "Batch Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000000, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The batch size in stream mode" },
"deleteAfterWrite": { "kind": "property", "displayName": "Delete After Write", "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": "Delete file object after the S3 file has been uploaded" },
"keyName": { "kind": "property", "displayName": "Key Name", "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": "Setting the key name for an element in the bucket through endpoint parameter" },
"lazyStartProducer": { "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during star [...]
"multiPartUpload": { "kind": "property", "displayName": "Multi Part Upload", "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": "If it is true, camel will upload the file with multi part format, the part size is decided b [...]
+ "namingStrategy": { "kind": "property", "displayName": "Naming Strategy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum", "enum": [ "progressive", "random" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "progressive", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "descript [...]
"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": "If StreaMode is true a different way of uploading will be provided" },
"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" },
@@ -79,7 +82,6 @@
"proxyPort": { "kind": "parameter", "displayName": "Proxy Port", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "Specify a proxy port to be used inside the client definition." },
"proxyProtocol": { "kind": "parameter", "displayName": "Proxy Protocol", "group": "common", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "To define a proxy protocol when instantiating t [...]
"region": { "kind": "parameter", "displayName": "Region", "group": "common", "label": "", "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 region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region ( [...]
- "streamMode": { "kind": "parameter", "displayName": "Stream Mode", "group": "common", "label": "", "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": "If StreaMode is true a different way of uploading will be provided" },
"trustAllCertificates": { "kind": "parameter", "displayName": "Trust All Certificates", "group": "common", "label": "", "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": "If we want to trust all certificates in case of overriding the endpoint" },
"uriEndpointOverride": { "kind": "parameter", "displayName": "Uri Endpoint Override", "group": "common", "label": "", "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": "Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option" },
"useDefaultCredentialsProvider": { "kind": "parameter", "displayName": "Use Default Credentials Provider", "group": "common", "label": "", "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": "Set whether the S3 client should expect to load credentials through a de [...]
@@ -105,13 +107,17 @@
"exceptionHandler": { "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the con [...]
"exchangePattern": { "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut", "InOptionalOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
"pollStrategy": { "kind": "parameter", "displayName": "Poll Strategy", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "A pluggable org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your custom implementation to control error handling usually occurred during the poll operation [...]
+ "batchMessageNumber": { "kind": "parameter", "displayName": "Batch Message Number", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The number of messages composing a batch in stream mode" },
+ "batchSize": { "kind": "parameter", "displayName": "Batch Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000000, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The batch size in stream mode" },
"deleteAfterWrite": { "kind": "parameter", "displayName": "Delete After Write", "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": "Delete file object after the S3 file has been uploaded" },
"keyName": { "kind": "parameter", "displayName": "Key Name", "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": "Setting the key name for an element in the bucket through endpoint parameter" },
"lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during sta [...]
"multiPartUpload": { "kind": "parameter", "displayName": "Multi Part Upload", "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": "If it is true, camel will upload the file with multi part format, the part size is decided [...]
+ "namingStrategy": { "kind": "parameter", "displayName": "Naming Strategy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum", "enum": [ "progressive", "random" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "progressive", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "descrip [...]
"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": "If StreaMode is true a different way of uploading will be provided" },
"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 e497bde..16de086 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 44 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 47 options, which are listed below.
@@ -65,7 +65,6 @@ The AWS 2 S3 Storage Service component supports 44 options, which are listed bel
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
-| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -85,13 +84,17 @@ The AWS 2 S3 Storage Service component supports 44 options, which are listed bel
| *moveAfterRead* (consumer) | Move objects from S3 bucket to a different bucket after they have been retrieved. To accomplish the operation the destinationBucket option must be set. The copy bucket operation is only performed if the Exchange is committed. If a rollback occurs, the object is not moved. | false | boolean
| *prefix* (consumer) | The prefix which is used in the com.amazonaws.services.s3.model.ListObjectsRequest to only consume objects we are interested in. | | String
| *autocloseBody* (consumer) | If this option is true and includeBody is false, then the S3Object.close() method will be called on exchange completion. This option is strongly related to includeBody option. In case of setting includeBody to false and autocloseBody to false, it will be up to the caller to close the S3Object stream. Setting autocloseBody to true, will close the S3Object stream automatically. | true | boolean
+| *batchMessageNumber* (producer) | The number of messages composing a batch in stream mode | 10 | int
+| *batchSize* (producer) | The batch size in stream mode | 1000000 | int
| *deleteAfterWrite* (producer) | Delete file object after the S3 file has been uploaded | false | boolean
| *keyName* (producer) | Setting the key name for an element in the bucket through endpoint parameter | | String
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
| *multiPartUpload* (producer) | If it is true, camel will upload the file with multi part format, the part size is decided by the option of partSize | false | boolean
+| *namingStrategy* (producer) | The naming strategy to use. There are 2 enums and the value can be one of: progressive, random | progressive | AWSS3NamingStrategyEnum
| *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) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *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
@@ -130,7 +133,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (62 parameters):
+=== Query Parameters (65 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -146,7 +149,6 @@ with the following path and query parameters:
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
-| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -172,13 +174,17 @@ with the following path and query parameters:
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. There are 3 enums and the value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
| *pollStrategy* (consumer) | A pluggable org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your custom implementation to control error handling usually occurred during the poll operation before an Exchange have been created and being routed in Camel. | | PollingConsumerPollStrategy
+| *batchMessageNumber* (producer) | The number of messages composing a batch in stream mode | 10 | int
+| *batchSize* (producer) | The batch size in stream mode | 1000000 | int
| *deleteAfterWrite* (producer) | Delete file object after the S3 file has been uploaded | false | boolean
| *keyName* (producer) | Setting the key name for an element in the bucket through endpoint parameter | | String
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
| *multiPartUpload* (producer) | If it is true, camel will upload the file with multi part format, the part size is decided by the option of partSize | false | boolean
+| *namingStrategy* (producer) | The naming strategy to use. There are 2 enums and the value can be one of: progressive, random | progressive | AWSS3NamingStrategyEnum
| *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) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *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/AWS2S3Configuration.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
index aec9bf7..b23e312 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
@@ -17,6 +17,7 @@
package org.apache.camel.component.aws2.s3;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
@@ -110,8 +111,14 @@ public class AWS2S3Configuration implements Cloneable {
private String uriEndpointOverride;
@UriParam(defaultValue = "false")
private boolean pojoRequest;
- @UriParam(defaultValue = "false")
+ @UriParam(defaultValue = "false", label = "producer")
private boolean streamMode;
+ @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;
public long getPartSize() {
return partSize;
@@ -588,6 +595,39 @@ public class AWS2S3Configuration implements Cloneable {
this.streamMode = streamMode;
}
+ public int getBatchMessageNumber() {
+ return batchMessageNumber;
+ }
+
+ /**
+ * The number of messages composing a batch in stream mode
+ */
+ public void setBatchMessageNumber(int batchMessageNumber) {
+ this.batchMessageNumber = batchMessageNumber;
+ }
+
+ public int getBatchSize() {
+ return batchSize;
+ }
+
+ /**
+ * The batch size in stream mode
+ */
+ public void setBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ public AWSS3NamingStrategyEnum getNamingStrategy() {
+ return namingStrategy;
+ }
+
+ /**
+ * The naming strategy to use
+ */
+ public void setNamingStrategy(AWSS3NamingStrategyEnum namingStrategy) {
+ this.namingStrategy = namingStrategy;
+ }
+
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 f8aa480..32d03835 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
@@ -22,6 +22,7 @@ import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.component.aws2.s3.client.AWS2S3ClientFactory;
+import org.apache.camel.component.aws2.s3.stream.AWS2S3StreamUploadProducer;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3StreamUploadProducer.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
similarity index 74%
rename from components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3StreamUploadProducer.java
rename to components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
index 63843d3..546128e 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3StreamUploadProducer.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
@@ -14,20 +14,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.component.aws2.s3;
+package org.apache.camel.component.aws2.s3.stream;
import java.io.ByteArrayOutputStream;
-import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
+import java.util.UUID;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
+import org.apache.camel.component.aws2.s3.AWS2S3Configuration;
+import org.apache.camel.component.aws2.s3.AWS2S3Constants;
+import org.apache.camel.component.aws2.s3.AWS2S3Endpoint;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.URISupport;
@@ -50,7 +51,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
CreateMultipartUploadResponse initResponse;
int index = 1;
- List<CompletedPart> completedParts = new ArrayList<CompletedPart>();
+ List<CompletedPart> completedParts = new ArrayList<>();
int part = 0;
public AWS2S3StreamUploadProducer(final Endpoint endpoint) {
@@ -59,21 +60,14 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
@Override
public void process(final Exchange exchange) throws Exception {
- streamUpload(exchange);
- }
-
- public void streamUpload(final Exchange exchange) throws Exception {
- File filePayload = null;
InputStream is = exchange.getIn().getMandatoryBody(InputStream.class);
buffer.write(IoUtils.toByteArray(is));
- final String keyName = determineKey(exchange);
- String dynamicKeyName;
- if (part > 0) {
- dynamicKeyName = keyName + "-" + part;
- } else {
- dynamicKeyName = keyName;
- }
+
+ final String keyName = getConfiguration().getKeyName();
+ final String fileName = determineFileName(keyName);
+ final String extension = determineFileExtension(keyName);
+ String dynamicKeyName = fileNameToUpload(fileName, getConfiguration().getNamingStrategy(), extension, part);
CreateMultipartUploadRequest.Builder createMultipartUploadRequest
= CreateMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName()).key(dynamicKeyName);
@@ -123,10 +117,8 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
long partSize = getConfiguration().getPartSize();
}
- long filePosition = 0;
-
try {
- if (buffer.size() >= 5000000 || index == 100) {
+ if (buffer.size() >= getConfiguration().getBatchSize() || index == getConfiguration().getBatchMessageNumber()) {
UploadPartRequest uploadRequest = UploadPartRequest.builder().bucket(getConfiguration().getBucketName())
.key(dynamicKeyName).uploadId(initResponse.uploadId())
@@ -142,7 +134,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
part++;
}
- if (index == 100) {
+ if (index == getConfiguration().getBatchMessageNumber()) {
CompletedMultipartUpload completeMultipartUpload
= CompletedMultipartUpload.builder().parts(completedParts).build();
CompleteMultipartUploadRequest compRequest
@@ -151,7 +143,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
.uploadId(initResponse.uploadId())
.build();
- uploadResult = getEndpoint().getS3Client().completeMultipartUpload(compRequest);
+ getEndpoint().getS3Client().completeMultipartUpload(compRequest);
index = 0;
}
@@ -165,51 +157,46 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
index++;
Message message = getMessageForResponse(exchange);
-
- }
-
- private AWS2S3Operations determineOperation(Exchange exchange) {
- AWS2S3Operations operation = exchange.getIn().getHeader(AWS2S3Constants.S3_OPERATION, AWS2S3Operations.class);
- if (operation == null) {
- operation = getConfiguration().getOperation();
- }
- return operation;
}
- private Map<String, String> determineMetadata(final Exchange exchange) {
- Map<String, String> objectMetadata = new HashMap<String, String>();
-
- Long contentLength = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_LENGTH, Long.class);
- if (contentLength != null) {
- objectMetadata.put("Content-Length", String.valueOf(contentLength));
- }
-
- String contentType = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_TYPE, String.class);
- if (contentType != null) {
- objectMetadata.put("Content-Type", String.valueOf(contentType));
- }
-
- String cacheControl = exchange.getIn().getHeader(AWS2S3Constants.CACHE_CONTROL, String.class);
- if (cacheControl != null) {
- objectMetadata.put("Cache-Control", String.valueOf(cacheControl));
- }
-
- String contentDisposition = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_DISPOSITION, String.class);
- if (contentDisposition != null) {
- objectMetadata.put("Content-Disposition", String.valueOf(contentDisposition));
- }
-
- String contentEncoding = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_ENCODING, String.class);
- if (contentEncoding != null) {
- objectMetadata.put("Content-Encoding", String.valueOf(contentEncoding));
- }
-
- String contentMD5 = exchange.getIn().getHeader(AWS2S3Constants.CONTENT_MD5, String.class);
- if (contentMD5 != null) {
- objectMetadata.put("Content-Md5", String.valueOf(contentMD5));
+ private String fileNameToUpload(String fileName, AWSS3NamingStrategyEnum strategy, String ext, int part) {
+ String dynamicKeyName;
+ switch (strategy) {
+ case progressive:
+ if (part > 0) {
+ if (ObjectHelper.isNotEmpty(ext)) {
+ dynamicKeyName = fileName + "-" + part + ext;
+ } else {
+ dynamicKeyName = fileName + "-" + part;
+ }
+ } else {
+ if (ObjectHelper.isNotEmpty(ext)) {
+ dynamicKeyName = fileName + ext;
+ } else {
+ dynamicKeyName = fileName;
+ }
+ }
+ break;
+ case random:
+ if (part > 0) {
+ UUID id = UUID.randomUUID();
+ if (ObjectHelper.isNotEmpty(ext)) {
+ dynamicKeyName = fileName + "-" + id.toString() + ext;
+ } else {
+ dynamicKeyName = fileName + "-" + id.toString();
+ }
+ } else {
+ if (ObjectHelper.isNotEmpty(ext)) {
+ dynamicKeyName = fileName + ext;
+ } else {
+ dynamicKeyName = fileName;
+ }
+ }
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported operation");
}
-
- return objectMetadata;
+ return dynamicKeyName;
}
/**
@@ -235,17 +222,6 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
return bucketName;
}
- private String determineKey(final Exchange exchange) {
- String key = exchange.getIn().getHeader(AWS2S3Constants.KEY, String.class);
- if (ObjectHelper.isEmpty(key)) {
- key = getConfiguration().getKeyName();
- }
- if (key == null) {
- throw new IllegalArgumentException("AWS S3 Key header missing.");
- }
- return key;
- }
-
private String determineStorageClass(final Exchange exchange) {
String storageClass = exchange.getIn().getHeader(AWS2S3Constants.STORAGE_CLASS, String.class);
if (storageClass == null) {
@@ -255,6 +231,26 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
return storageClass;
}
+ private String determineFileExtension(String keyName) {
+ int extPosition = keyName.lastIndexOf(".");
+ if (extPosition == -1) {
+ return "";
+ } else {
+ String ext = keyName.substring(extPosition);
+ return ext;
+ }
+ }
+
+ private String determineFileName(String keyName) {
+ int extPosition = keyName.lastIndexOf(".");
+ if (extPosition == -1) {
+ return keyName;
+ } else {
+ String prefix = keyName.substring(0, extPosition);
+ return prefix;
+ }
+ }
+
private ByteArrayOutputStream determineLengthInputStream(InputStream is) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
byte[] bytes = new byte[1024];
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyEnum.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyEnum.java
new file mode 100644
index 0000000..0ce0787
--- /dev/null
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyEnum.java
@@ -0,0 +1,6 @@
+package org.apache.camel.component.aws2.s3.stream;
+
+public enum AWSS3NamingStrategyEnum {
+ progressive,
+ random
+}
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
index e567caf..66127c1 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
@@ -50,9 +50,6 @@ public class S3StreamUploadOperationIntegrationTest extends CamelTestSupport {
for (int i = 0; i < 1000; i++) {
template.sendBody("direct:stream1", "Andrea\n");
}
- for (int i = 0; i < 1000; i++) {
- template.sendBody("direct:stream2", "Luca\n");
- }
Thread.sleep(30000);
}
@@ -63,12 +60,9 @@ public class S3StreamUploadOperationIntegrationTest extends CamelTestSupport {
@Override
public void configure() throws Exception {
String awsEndpoint1
- = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTest";
- String awsEndpoint2
- = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTestParallel";
+ = "aws2-s3://mycamel-1?autoCreateBucket=true&streamMode=true&keyName=fileTest&batchMessageNumber=25&namingStrategy=progressive";
from("direct:stream1").toD(awsEndpoint1).to("mock:result");
- from("direct:stream2").toD(awsEndpoint2).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 c5e6b6e..d0c7a4b 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
@@ -228,21 +228,6 @@ public interface Aws2S3ComponentBuilderFactory {
return this;
}
/**
- * If StreaMode is true a different way of uploading will be provided.
- *
- * The option is a: <code>boolean</code> type.
- *
- * Default: false
- * Group: common
- *
- * @param streamMode the value to set
- * @return the dsl builder
- */
- default Aws2S3ComponentBuilder streamMode(boolean streamMode) {
- doSetProperty("streamMode", streamMode);
- return this;
- }
- /**
* If we want to trust all certificates in case of overriding the
* endpoint.
*
@@ -572,6 +557,36 @@ public interface Aws2S3ComponentBuilderFactory {
return this;
}
/**
+ * The number of messages composing a batch in stream mode.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 10
+ * Group: producer
+ *
+ * @param batchMessageNumber the value to set
+ * @return the dsl builder
+ */
+ default Aws2S3ComponentBuilder batchMessageNumber(int batchMessageNumber) {
+ doSetProperty("batchMessageNumber", batchMessageNumber);
+ return this;
+ }
+ /**
+ * The batch size in stream mode.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 1000000
+ * Group: producer
+ *
+ * @param batchSize the value to set
+ * @return the dsl builder
+ */
+ default Aws2S3ComponentBuilder batchSize(int batchSize) {
+ doSetProperty("batchSize", batchSize);
+ return this;
+ }
+ /**
* Delete file object after the S3 file has been uploaded.
*
* The option is a: <code>boolean</code> type.
@@ -642,6 +657,23 @@ public interface Aws2S3ComponentBuilderFactory {
return this;
}
/**
+ * The naming strategy to use.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum</code> type.
+ *
+ * Default: progressive
+ * Group: producer
+ *
+ * @param namingStrategy the value to set
+ * @return the dsl builder
+ */
+ default Aws2S3ComponentBuilder namingStrategy(
+ org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum namingStrategy) {
+ doSetProperty("namingStrategy", namingStrategy);
+ return this;
+ }
+ /**
* The operation to do in case the user don't want to do only an upload.
*
* The option is a:
@@ -690,6 +722,21 @@ public interface Aws2S3ComponentBuilderFactory {
return this;
}
/**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default Aws2S3ComponentBuilder streamMode(boolean streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
* Define the id of KMS key to use in case KMS is enabled.
*
* The option is a: <code>java.lang.String</code> type.
@@ -816,7 +863,6 @@ public interface Aws2S3ComponentBuilderFactory {
case "proxyPort": getOrCreateConfiguration((AWS2S3Component) component).setProxyPort((java.lang.Integer) value); return true;
case "proxyProtocol": getOrCreateConfiguration((AWS2S3Component) component).setProxyProtocol((software.amazon.awssdk.core.Protocol) value); return true;
case "region": getOrCreateConfiguration((AWS2S3Component) component).setRegion((java.lang.String) value); return true;
- case "streamMode": getOrCreateConfiguration((AWS2S3Component) component).setStreamMode((boolean) value); return true;
case "trustAllCertificates": getOrCreateConfiguration((AWS2S3Component) component).setTrustAllCertificates((boolean) value); return true;
case "uriEndpointOverride": getOrCreateConfiguration((AWS2S3Component) component).setUriEndpointOverride((java.lang.String) value); return true;
case "useDefaultCredentialsProvider": getOrCreateConfiguration((AWS2S3Component) component).setUseDefaultCredentialsProvider((boolean) value); return true;
@@ -836,13 +882,17 @@ public interface Aws2S3ComponentBuilderFactory {
case "moveAfterRead": getOrCreateConfiguration((AWS2S3Component) component).setMoveAfterRead((boolean) value); return true;
case "prefix": getOrCreateConfiguration((AWS2S3Component) component).setPrefix((java.lang.String) value); return true;
case "autocloseBody": getOrCreateConfiguration((AWS2S3Component) component).setAutocloseBody((boolean) value); return true;
+ case "batchMessageNumber": getOrCreateConfiguration((AWS2S3Component) component).setBatchMessageNumber((int) value); return true;
+ case "batchSize": getOrCreateConfiguration((AWS2S3Component) component).setBatchSize((int) value); return true;
case "deleteAfterWrite": getOrCreateConfiguration((AWS2S3Component) component).setDeleteAfterWrite((boolean) value); return true;
case "keyName": getOrCreateConfiguration((AWS2S3Component) component).setKeyName((java.lang.String) value); return true;
case "lazyStartProducer": ((AWS2S3Component) component).setLazyStartProducer((boolean) value); return true;
case "multiPartUpload": getOrCreateConfiguration((AWS2S3Component) component).setMultiPartUpload((boolean) value); return true;
+ case "namingStrategy": getOrCreateConfiguration((AWS2S3Component) component).setNamingStrategy((org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum) value); return true;
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 "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 0ae939c..1538e46 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
@@ -328,37 +328,6 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
- * If StreaMode is true a different way of uploading will be provided.
- *
- * The option is a: <code>boolean</code> type.
- *
- * Default: false
- * Group: common
- *
- * @param streamMode the value to set
- * @return the dsl builder
- */
- default AWS2S3EndpointConsumerBuilder streamMode(boolean streamMode) {
- doSetProperty("streamMode", streamMode);
- return this;
- }
- /**
- * If StreaMode is true a different way of uploading will be provided.
- *
- * The option will be converted to a <code>boolean</code>
- * type.
- *
- * Default: false
- * Group: common
- *
- * @param streamMode the value to set
- * @return the dsl builder
- */
- default AWS2S3EndpointConsumerBuilder streamMode(String streamMode) {
- doSetProperty("streamMode", streamMode);
- return this;
- }
- /**
* If we want to trust all certificates in case of overriding the
* endpoint.
*
@@ -1880,37 +1849,6 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
- * If StreaMode is true a different way of uploading will be provided.
- *
- * The option is a: <code>boolean</code> type.
- *
- * Default: false
- * Group: common
- *
- * @param streamMode the value to set
- * @return the dsl builder
- */
- default AWS2S3EndpointProducerBuilder streamMode(boolean streamMode) {
- doSetProperty("streamMode", streamMode);
- return this;
- }
- /**
- * If StreaMode is true a different way of uploading will be provided.
- *
- * The option will be converted to a <code>boolean</code>
- * type.
- *
- * Default: false
- * Group: common
- *
- * @param streamMode the value to set
- * @return the dsl builder
- */
- default AWS2S3EndpointProducerBuilder streamMode(String streamMode) {
- doSetProperty("streamMode", streamMode);
- return this;
- }
- /**
* If we want to trust all certificates in case of overriding the
* endpoint.
*
@@ -1999,6 +1937,68 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
+ * The number of messages composing a batch in stream mode.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 10
+ * Group: producer
+ *
+ * @param batchMessageNumber the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder batchMessageNumber(
+ int batchMessageNumber) {
+ doSetProperty("batchMessageNumber", batchMessageNumber);
+ return this;
+ }
+ /**
+ * The number of messages composing a batch in stream mode.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Default: 10
+ * Group: producer
+ *
+ * @param batchMessageNumber the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder batchMessageNumber(
+ String batchMessageNumber) {
+ doSetProperty("batchMessageNumber", batchMessageNumber);
+ return this;
+ }
+ /**
+ * The batch size in stream mode.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 1000000
+ * Group: producer
+ *
+ * @param batchSize the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder batchSize(int batchSize) {
+ doSetProperty("batchSize", batchSize);
+ return this;
+ }
+ /**
+ * The batch size in stream mode.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Default: 1000000
+ * Group: producer
+ *
+ * @param batchSize the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder batchSize(String batchSize) {
+ doSetProperty("batchSize", batchSize);
+ return this;
+ }
+ /**
* Delete file object after the S3 file has been uploaded.
*
* The option is a: <code>boolean</code> type.
@@ -2131,6 +2131,40 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
+ * The naming strategy to use.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum</code> type.
+ *
+ * Default: progressive
+ * Group: producer
+ *
+ * @param namingStrategy the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder namingStrategy(
+ AWSS3NamingStrategyEnum namingStrategy) {
+ doSetProperty("namingStrategy", namingStrategy);
+ return this;
+ }
+ /**
+ * The naming strategy to use.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum</code> type.
+ *
+ * Default: progressive
+ * Group: producer
+ *
+ * @param namingStrategy the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder namingStrategy(
+ String namingStrategy) {
+ doSetProperty("namingStrategy", namingStrategy);
+ return this;
+ }
+ /**
* The operation to do in case the user don't want to do only an upload.
*
* The option is a:
@@ -2209,6 +2243,37 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder streamMode(boolean streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
+ * If StreaMode is true a different way of uploading will be provided.
+ *
+ * The option will be converted to a <code>boolean</code>
+ * type.
+ *
+ * Default: false
+ * Group: producer
+ *
+ * @param streamMode the value to set
+ * @return the dsl builder
+ */
+ default AWS2S3EndpointProducerBuilder streamMode(String streamMode) {
+ doSetProperty("streamMode", streamMode);
+ return this;
+ }
+ /**
* Amazon AWS Access Key.
*
* The option is a: <code>java.lang.String</code> type.
@@ -2656,37 +2721,6 @@ public interface AWS2S3EndpointBuilderFactory {
return this;
}
/**
- * If StreaMode is true a different way of uploading will be provided.
- *
- * The option is a: <code>boolean</code> type.
- *
- * Default: false
- * Group: common
- *
- * @param streamMode the value to set
- * @return the dsl builder
- */
- default AWS2S3EndpointBuilder streamMode(boolean streamMode) {
- doSetProperty("streamMode", streamMode);
- return this;
- }
- /**
- * If StreaMode is true a different way of uploading will be provided.
- *
- * The option will be converted to a <code>boolean</code>
- * type.
- *
- * Default: false
- * Group: common
- *
- * @param streamMode the value to set
- * @return the dsl builder
- */
- default AWS2S3EndpointBuilder streamMode(String streamMode) {
- doSetProperty("streamMode", streamMode);
- return this;
- }
- /**
* If we want to trust all certificates in case of overriding the
* endpoint.
*
@@ -2870,6 +2904,15 @@ public interface AWS2S3EndpointBuilderFactory {
/**
* Proxy enum for
+ * <code>org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum</code> enum.
+ */
+ enum AWSS3NamingStrategyEnum {
+ progressive,
+ random;
+ }
+
+ /**
+ * Proxy enum for
* <code>org.apache.camel.component.aws2.s3.AWS2S3Operations</code> enum.
*/
enum AWS2S3Operations {
diff --git a/docs/components/modules/ROOT/pages/aws2-s3-component.adoc b/docs/components/modules/ROOT/pages/aws2-s3-component.adoc
index 579bf3f..74be95e 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 44 options, which are listed below.
+The AWS 2 S3 Storage Service component supports 47 options, which are listed below.
@@ -67,7 +67,6 @@ The AWS 2 S3 Storage Service component supports 44 options, which are listed bel
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
-| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -87,13 +86,17 @@ The AWS 2 S3 Storage Service component supports 44 options, which are listed bel
| *moveAfterRead* (consumer) | Move objects from S3 bucket to a different bucket after they have been retrieved. To accomplish the operation the destinationBucket option must be set. The copy bucket operation is only performed if the Exchange is committed. If a rollback occurs, the object is not moved. | false | boolean
| *prefix* (consumer) | The prefix which is used in the com.amazonaws.services.s3.model.ListObjectsRequest to only consume objects we are interested in. | | String
| *autocloseBody* (consumer) | If this option is true and includeBody is false, then the S3Object.close() method will be called on exchange completion. This option is strongly related to includeBody option. In case of setting includeBody to false and autocloseBody to false, it will be up to the caller to close the S3Object stream. Setting autocloseBody to true, will close the S3Object stream automatically. | true | boolean
+| *batchMessageNumber* (producer) | The number of messages composing a batch in stream mode | 10 | int
+| *batchSize* (producer) | The batch size in stream mode | 1000000 | int
| *deleteAfterWrite* (producer) | Delete file object after the S3 file has been uploaded | false | boolean
| *keyName* (producer) | Setting the key name for an element in the bucket through endpoint parameter | | String
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
| *multiPartUpload* (producer) | If it is true, camel will upload the file with multi part format, the part size is decided by the option of partSize | false | boolean
+| *namingStrategy* (producer) | The naming strategy to use. There are 2 enums and the value can be one of: progressive, random | progressive | AWSS3NamingStrategyEnum
| *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) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *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
@@ -132,7 +135,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (62 parameters):
+=== Query Parameters (65 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -148,7 +151,6 @@ with the following path and query parameters:
| *proxyPort* (common) | Specify a proxy port to be used inside the client definition. | | Integer
| *proxyProtocol* (common) | To define a proxy protocol when instantiating the S3 client. There are 2 enums and the value can be one of: HTTP, HTTPS | HTTPS | Protocol
| *region* (common) | The region in which S3 client needs to work. When using this parameter, the configuration will expect the lowercase name of the region (for example ap-east-1) You'll need to use the name Region.EU_WEST_1.id() | | String
-| *streamMode* (common) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *trustAllCertificates* (common) | If we want to trust all certificates in case of overriding the endpoint | false | boolean
| *uriEndpointOverride* (common) | Set the overriding uri endpoint. This option needs to be used in combination with overrideEndpoint option | | String
| *useDefaultCredentialsProvider* (common) | Set whether the S3 client should expect to load credentials through a default credentials provider or to expect static credentials to be passed in. | false | boolean
@@ -174,13 +176,17 @@ with the following path and query parameters:
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. There are 3 enums and the value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
| *pollStrategy* (consumer) | A pluggable org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your custom implementation to control error handling usually occurred during the poll operation before an Exchange have been created and being routed in Camel. | | PollingConsumerPollStrategy
+| *batchMessageNumber* (producer) | The number of messages composing a batch in stream mode | 10 | int
+| *batchSize* (producer) | The batch size in stream mode | 1000000 | int
| *deleteAfterWrite* (producer) | Delete file object after the S3 file has been uploaded | false | boolean
| *keyName* (producer) | Setting the key name for an element in the bucket through endpoint parameter | | String
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
| *multiPartUpload* (producer) | If it is true, camel will upload the file with multi part format, the part size is decided by the option of partSize | false | boolean
+| *namingStrategy* (producer) | The naming strategy to use. There are 2 enums and the value can be one of: progressive, random | progressive | AWSS3NamingStrategyEnum
| *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) | If StreaMode is true a different way of uploading will be provided | false | boolean
| *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] 04/05: 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 93e20a5d827db340451e424dbcf45d4253c03659
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Apr 1 14:33:14 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 17 +++++++++++++----
.../aws2/s3/stream/AWSS3NamingStrategyEnum.java | 16 ++++++++++++++++
.../S3StreamUploadOperationIntegrationTest.java | 4 +++-
3 files changed, 32 insertions(+), 5 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 21d264c..20ff25a 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
@@ -36,7 +36,17 @@ import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.core.sync.RequestBody;
-import software.amazon.awssdk.services.s3.model.*;
+import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.BucketCannedACL;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
+import software.amazon.awssdk.services.s3.model.ServerSideEncryption;
+import software.amazon.awssdk.services.s3.model.UploadPartRequest;
import software.amazon.awssdk.utils.IoUtils;
/**
@@ -47,14 +57,13 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
private static final Logger LOG = LoggerFactory.getLogger(AWS2S3StreamUploadProducer.class);
- private transient String s3ProducerToString;
-
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
CreateMultipartUploadResponse initResponse;
AtomicInteger index = new AtomicInteger();
List<CompletedPart> completedParts;
AtomicInteger part = new AtomicInteger();
- UUID id = null;
+ UUID id;
+ private transient String s3ProducerToString;
public AWS2S3StreamUploadProducer(final Endpoint endpoint) {
super(endpoint);
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyEnum.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyEnum.java
index 0ce0787..bd8b94b 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyEnum.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWSS3NamingStrategyEnum.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.stream;
public enum AWSS3NamingStrategyEnum {
diff --git a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
index 9e27023..1693f35 100644
--- a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
+++ b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/integration/S3StreamUploadOperationIntegrationTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.component.aws2.s3.integration;
-import org.apache.camel.*;
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit5.CamelTestSupport;
[camel] 05/05: 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 d7e8e1cf46a7106ffa73d642a4e331791cf92508
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Apr 1 14:39:02 2021 +0200
CAMEL-16185 - AWS S3: improve multipart support - streaming upload
---
.../resources/org/apache/camel/catalog/components/aws2-s3.json | 8 ++++++++
1 file changed, 8 insertions(+)
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 d2cbe25..09a499e 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
@@ -52,13 +52,17 @@
"moveAfterRead": { "kind": "property", "displayName": "Move After Read", "group": "consumer", "label": "consumer", "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": "Move objects from S3 bucket to a different bucket after they have been retrieved. To accomplish [...]
"prefix": { "kind": "property", "displayName": "Prefix", "group": "consumer", "label": "consumer", "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 prefix which is used in the com.amazonaws.services.s3.model.ListObjectsRequest to only consume objects we are interested in." },
"autocloseBody": { "kind": "property", "displayName": "Autoclose Body", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "If this option is true and includeBody is false, then the S3Object.close() met [...]
+ "batchMessageNumber": { "kind": "property", "displayName": "Batch Message Number", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The number of messages composing a batch in stream mode" },
+ "batchSize": { "kind": "property", "displayName": "Batch Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000000, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The batch size in stream mode" },
"deleteAfterWrite": { "kind": "property", "displayName": "Delete After Write", "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": "Delete file object after the S3 file has been uploaded" },
"keyName": { "kind": "property", "displayName": "Key Name", "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": "Setting the key name for an element in the bucket through endpoint parameter" },
"lazyStartProducer": { "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during star [...]
"multiPartUpload": { "kind": "property", "displayName": "Multi Part Upload", "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": "If it is true, camel will upload the file with multi part format, the part size is decided b [...]
+ "namingStrategy": { "kind": "property", "displayName": "Naming Strategy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum", "enum": [ "progressive", "random" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "progressive", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "descript [...]
"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": "If StreaMode is true a different way of uploading will be provided" },
"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" },
@@ -103,13 +107,17 @@
"exceptionHandler": { "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the con [...]
"exchangePattern": { "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut", "InOptionalOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
"pollStrategy": { "kind": "parameter", "displayName": "Poll Strategy", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "A pluggable org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your custom implementation to control error handling usually occurred during the poll operation [...]
+ "batchMessageNumber": { "kind": "parameter", "displayName": "Batch Message Number", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 10, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The number of messages composing a batch in stream mode" },
+ "batchSize": { "kind": "parameter", "displayName": "Batch Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000000, "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "description": "The batch size in stream mode" },
"deleteAfterWrite": { "kind": "parameter", "displayName": "Delete After Write", "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": "Delete file object after the S3 file has been uploaded" },
"keyName": { "kind": "parameter", "displayName": "Key Name", "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": "Setting the key name for an element in the bucket through endpoint parameter" },
"lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during sta [...]
"multiPartUpload": { "kind": "parameter", "displayName": "Multi Part Upload", "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": "If it is true, camel will upload the file with multi part format, the part size is decided [...]
+ "namingStrategy": { "kind": "parameter", "displayName": "Naming Strategy", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum", "enum": [ "progressive", "random" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "progressive", "configurationClass": "org.apache.camel.component.aws2.s3.AWS2S3Configuration", "configurationField": "configuration", "descrip [...]
"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": "If StreaMode is true a different way of uploading will be provided" },
"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" },