You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by oa...@apache.org on 2020/10/13 12:02:53 UTC
[camel] branch master updated: CAMEL-15656: Support batch consumer
in Azure Storage Blob component
This is an automated email from the ASF dual-hosted git repository.
oalsafi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 076bc5a CAMEL-15656: Support batch consumer in Azure Storage Blob component
076bc5a is described below
commit 076bc5ae1764a5c4d193095f8526999d5bdfaca1
Author: Omar Al-Safi <om...@gmail.com>
AuthorDate: Tue Oct 13 12:33:34 2020 +0200
CAMEL-15656: Support batch consumer in Azure Storage Blob component
---
.../catalog/docs/azure-storage-blob-component.adoc | 25 +++-
.../azure/storage/blob/azure-storage-blob.json | 38 +++---
.../main/docs/azure-storage-blob-component.adoc | 25 +++-
.../azure/storage/blob/BlobConfiguration.java | 3 +-
.../blob/BlobConfigurationOptionsProxy.java | 10 +-
.../component/azure/storage/blob/BlobConsumer.java | 137 ++++++++++++++++++---
.../component/azure/storage/blob/BlobEndpoint.java | 3 -
.../azure/storage/blob/BlobExchangeHeaders.java | 3 +-
.../component/azure/storage/blob/BlobUtils.java | 2 +-
.../blob/operations/BlobContainerOperations.java | 16 ---
.../storage/blob/operations/BlobOperations.java | 65 +---------
.../blob/operations/BlobServiceOperations.java | 3 -
.../azure/storage/blob/BlobComponentTest.java | 10 --
.../azure/storage/blob/BlobConsumerIT.java | 61 ++++++++-
.../AzureStorageBlobComponentBuilderFactory.java | 4 +-
.../endpoint/dsl/BlobEndpointBuilderFactory.java | 12 +-
.../ROOT/pages/azure-storage-blob-component.adoc | 25 +++-
17 files changed, 287 insertions(+), 155 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-blob-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-blob-component.adoc
index 3a9944b..2590c33 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-blob-component.adoc
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-blob-component.adoc
@@ -43,7 +43,7 @@ for this component:
azure-storage-blob://accountName[/containerName][?options]
------------------------------
-In case of consumer, accountName, containerName and blobName are required. In case of producer, it depends on the operation that being
+In case of consumer, accountName, containerName are required. In case of producer, it depends on the operation that being
requested, for example if operation is on a container level, e.b: createContainer, accountName and containerName are only required, but in case
of operation being requested in blob level, e.g: getBlob, accountName, containerName and blobName are required.
@@ -73,7 +73,7 @@ The Azure Storage Blob Service component supports 27 options, which are listed b
|===
| Name | Description | Default | Type
| *autoDiscoverClient* (common) | Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in the registry automatically otherwise it will skip that checking. | true | boolean
-| *blobName* (common) | The blob name, required for consumer. However on producer, is only required for the operations on the blob level | | String
+| *blobName* (common) | The blob name, to consume specific blob from a container. However on producer, is only required for the operations on the blob level | | String
| *blobOffset* (common) | Set the blob offset for the upload or download operations, default is 0 | 0 | long
| *blobType* (common) | The blob type in order to initiate the appropriate settings for each blob type. There are 3 enums and the value can be one of: blockblob, appendblob, pageblob | blockblob | BlobType
| *closeStreamAfterRead* (common) | Close the stream after read or keep it open, default is true | true | boolean
@@ -129,7 +129,7 @@ with the following path and query parameters:
|===
| Name | Description | Default | Type
| *autoDiscoverClient* (common) | Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in the registry automatically otherwise it will skip that checking. | true | boolean
-| *blobName* (common) | The blob name, required for consumer. However on producer, is only required for the operations on the blob level | | String
+| *blobName* (common) | The blob name, to consume specific blob from a container. However on producer, is only required for the operations on the blob level | | String
| *blobOffset* (common) | Set the blob offset for the upload or download operations, default is 0 | 0 | long
| *blobServiceClient* (common) | Client to a storage account. This client does not hold any state about a particular storage account but is instead a convenient way of sending off appropriate requests to the resource on the service. It may also be used to construct URLs to blobs and containers. This client contains operations on a service account. Operations on a container are available on BlobContainerClient through getBlobContainerClient(String), and operations on a blob are available [...]
| *blobType* (common) | The blob type in order to initiate the appropriate settings for each blob type. There are 3 enums and the value can be one of: blockblob, appendblob, pageblob | blockblob | BlobType
@@ -164,6 +164,7 @@ with the following path and query parameters:
*Required information options:*
To use this component, you have 3 options in order to provide the required Azure authentication information:
+
- Provide `accountName` and `accessKey` for your Azure account, this is the simplest way to get started. The accessKey can
be generated through your Azure portal.
- Provide a https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/common/StorageSharedKeyCredential.html[StorageSharedKeyCredential] instance which can be
@@ -173,6 +174,15 @@ provided into `blobServiceClient`. Note: You don't need to create a specific cli
can be used to retrieve lower level clients.
+== Batch Consumer
+
+This component implements the Batch Consumer.
+
+This allows you for instance to know how many messages exists in this
+batch and for instance let the Aggregator
+aggregate this number of messages.
+
+
== Usage
=== Message headers evaluated by the component producer
@@ -347,6 +357,15 @@ from("azure-storage-blob:/camelazure/container1?blobName=hello.txt&accountName=y
to("mock:results");
--------------------------------------------------------------------------------
+Also, the component supports batch consumer, hence you can consume multiple blobs with only specifying the container name, the consumer will
+return multiple exchanges depending on the number of the blobs in the container. Example:
+[source,java]
+--------------------------------------------------------------------------------
+from("azure-storage-blob:/camelazure/container1?accountName=yourAccountName&accessKey=yourAccessKey&fileDir=/var/to/awesome/dir").
+to("mock:results");
+--------------------------------------------------------------------------------
+
+
=== Producer Operations Examples
- `listBlobContainers`:
diff --git a/components/camel-azure-storage-blob/src/generated/resources/org/apache/camel/component/azure/storage/blob/azure-storage-blob.json b/components/camel-azure-storage-blob/src/generated/resources/org/apache/camel/component/azure/storage/blob/azure-storage-blob.json
index febbfec..91238af 100644
--- a/components/camel-azure-storage-blob/src/generated/resources/org/apache/camel/component/azure/storage/blob/azure-storage-blob.json
+++ b/components/camel-azure-storage-blob/src/generated/resources/org/apache/camel/component/azure/storage/blob/azure-storage-blob.json
@@ -22,27 +22,27 @@
"lenientProperties": false
},
"componentProperties": {
- "autoDiscoverClient": { "kind": "property", "displayName": "Auto Discover Client", "group": "common", "label": "common", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in th [...]
- "blobName": { "kind": "property", "displayName": "Blob Name", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The blob name, required for consumer. However on producer, is only required for the operations on the blob level" },
- "blobOffset": { "kind": "property", "displayName": "Blob Offset", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Set the blob offset for the upload or download operations, default is 0" },
+ "autoDiscoverClient": { "kind": "property", "displayName": "Auto Discover Client", "group": "common", "label": "common", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in [...]
+ "blobName": { "kind": "property", "displayName": "Blob Name", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The blob name, to consume specific blob from a container. However on producer, is only required for the operations on the blob level" },
+ "blobOffset": { "kind": "property", "displayName": "Blob Offset", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Set the blob offset for the upload or download operations, default is 0" },
"blobType": { "kind": "property", "displayName": "Blob Type", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.storage.blob.BlobType", "enum": [ "blockblob", "appendblob", "pageblob" ], "deprecated": false, "secret": false, "defaultValue": "blockblob", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The blob type in or [...]
- "closeStreamAfterRead": { "kind": "property", "displayName": "Close Stream After Read", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Close the stream after read or keep it open, default is true" },
+ "closeStreamAfterRead": { "kind": "property", "displayName": "Close Stream After Read", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Close the stream after read or keep it open, default is true" },
"configuration": { "kind": "property", "displayName": "Configuration", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "deprecated": false, "secret": false, "description": "The component configurations" },
"credentials": { "kind": "property", "displayName": "Credentials", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.azure.storage.common.StorageSharedKeyCredential", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "StorageSharedKeyCredential can be injected to create the azure client, this holds the important authent [...]
"dataCount": { "kind": "property", "displayName": "Data Count", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "How many bytes to include in the range. Must be greater than or equal to 0 if specified." },
"fileDir": { "kind": "property", "displayName": "File Dir", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The file directory where the downloaded blobs will be saved to, this can be used in both, producer and consumer" },
"maxResultsPerPage": { "kind": "property", "displayName": "Max Results Per Page", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies the maximum number of blobs to return, including all BlobPrefix elements. If the request does not specify [...]
- "maxRetryRequests": { "kind": "property", "displayName": "Max Retry Requests", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies the maximum number of additional HTTP Get requests that will be made while reading the data from a resp [...]
+ "maxRetryRequests": { "kind": "property", "displayName": "Max Retry Requests", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies the maximum number of additional HTTP Get requests that will be made while reading the data from a re [...]
"prefix": { "kind": "property", "displayName": "Prefix", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Filters the results to return only blobs whose names begin with the specified prefix. May be null to return all blobs." },
"serviceClient": { "kind": "property", "displayName": "Service Client", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.azure.storage.blob.BlobServiceClient", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Client to a storage account. This client does not hold any state about a particular storage account but is in [...]
"timeout": { "kind": "property", "displayName": "Timeout", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "An optional timeout value beyond which a RuntimeException will be raised." },
"bridgeErrorHandler": { "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by [...]
"blobSequenceNumber": { "kind": "property", "displayName": "Blob Sequence Number", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "A user-controlled value that you can use to track requests. The value of the sequence number [...]
"blockListType": { "kind": "property", "displayName": "Block List Type", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.storage.blob.models.BlockListType", "enum": [ "committed", "uncommitted", "all" ], "deprecated": false, "secret": false, "defaultValue": "COMMITTED", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies which typ [...]
- "closeStreamAfterWrite": { "kind": "property", "displayName": "Close Stream After Write", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Close the stream after write or keep it open, default is true" },
- "commitBlockListLater": { "kind": "property", "displayName": "Commit Block List Later", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the staged blocks will not be committed directly." },
- "createAppendBlob": { "kind": "property", "displayName": "Create Append Blob", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the append blocks will be created when committing append blocks." },
- "createPageBlob": { "kind": "property", "displayName": "Create Page Blob", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the page blob will be created when uploading page blob." },
+ "closeStreamAfterWrite": { "kind": "property", "displayName": "Close Stream After Write", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Close the stream after write or keep it open, default is true" },
+ "commitBlockListLater": { "kind": "property", "displayName": "Commit Block List Later", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the staged blocks will not be committed directly." },
+ "createAppendBlob": { "kind": "property", "displayName": "Create Append Blob", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the append blocks will be created when committing append blocks." },
+ "createPageBlob": { "kind": "property", "displayName": "Create Page Blob", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the page blob will be created when uploading page blob." },
"downloadLinkExpiration": { "kind": "property", "displayName": "Download Link Expiration", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Override the default expiration (millis) of URL download link." },
"lazyStartProducer": { "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": 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 starting and cause the r [...]
"operation": { "kind": "property", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.storage.blob.BlobOperationsDefinition", "enum": [ "listBlobContainers", "createBlobContainer", "deleteBlobContainer", "listBlobs", "getBlob", "deleteBlob", "downloadBlobToFile", "downloadLink", "uploadBlockBlob", "stageBlockBlobList", "commitBlobBlockList", "getBlobBlockList", "createAppendBlob", "c [...]
@@ -53,17 +53,17 @@
"properties": {
"accountName": { "kind": "path", "displayName": "Account Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Azure account name to be used for authentication with azure blob services" },
"containerName": { "kind": "path", "displayName": "Container Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The blob container name" },
- "autoDiscoverClient": { "kind": "parameter", "displayName": "Auto Discover Client", "group": "common", "label": "common", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in t [...]
- "blobName": { "kind": "parameter", "displayName": "Blob Name", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The blob name, required for consumer. However on producer, is only required for the operations on the blob level" },
- "blobOffset": { "kind": "parameter", "displayName": "Blob Offset", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Set the blob offset for the upload or download operations, default is 0" },
+ "autoDiscoverClient": { "kind": "parameter", "displayName": "Auto Discover Client", "group": "common", "label": "common", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in [...]
+ "blobName": { "kind": "parameter", "displayName": "Blob Name", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The blob name, to consume specific blob from a container. However on producer, is only required for the operations on the blob level" },
+ "blobOffset": { "kind": "parameter", "displayName": "Blob Offset", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Set the blob offset for the upload or download operations, default is 0" },
"blobServiceClient": { "kind": "parameter", "displayName": "Blob Service Client", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.azure.storage.blob.BlobServiceClient", "deprecated": false, "secret": false, "description": "Client to a storage account. This client does not hold any state about a particular storage account but is instead a convenient way of sending off appropriate requests to the resource on the service. It may also be used to cons [...]
"blobType": { "kind": "parameter", "displayName": "Blob Type", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.storage.blob.BlobType", "enum": [ "blockblob", "appendblob", "pageblob" ], "deprecated": false, "secret": false, "defaultValue": "blockblob", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The blob type in o [...]
- "closeStreamAfterRead": { "kind": "parameter", "displayName": "Close Stream After Read", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Close the stream after read or keep it open, default is true" },
+ "closeStreamAfterRead": { "kind": "parameter", "displayName": "Close Stream After Read", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Close the stream after read or keep it open, default is true" },
"credentials": { "kind": "parameter", "displayName": "Credentials", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.azure.storage.common.StorageSharedKeyCredential", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "StorageSharedKeyCredential can be injected to create the azure client, this holds the important authen [...]
"dataCount": { "kind": "parameter", "displayName": "Data Count", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "How many bytes to include in the range. Must be greater than or equal to 0 if specified." },
"fileDir": { "kind": "parameter", "displayName": "File Dir", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "The file directory where the downloaded blobs will be saved to, this can be used in both, producer and consumer" },
"maxResultsPerPage": { "kind": "parameter", "displayName": "Max Results Per Page", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies the maximum number of blobs to return, including all BlobPrefix elements. If the request does not specif [...]
- "maxRetryRequests": { "kind": "parameter", "displayName": "Max Retry Requests", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies the maximum number of additional HTTP Get requests that will be made while reading the data from a res [...]
+ "maxRetryRequests": { "kind": "parameter", "displayName": "Max Retry Requests", "group": "common", "label": "common", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies the maximum number of additional HTTP Get requests that will be made while reading the data from a r [...]
"prefix": { "kind": "parameter", "displayName": "Prefix", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Filters the results to return only blobs whose names begin with the specified prefix. May be null to return all blobs." },
"serviceClient": { "kind": "parameter", "displayName": "Service Client", "group": "common", "label": "", "required": false, "type": "object", "javaType": "com.azure.storage.blob.BlobServiceClient", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Client to a storage account. This client does not hold any state about a particular storage account but is i [...]
"timeout": { "kind": "parameter", "displayName": "Timeout", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "An optional timeout value beyond which a RuntimeException will be raised." },
@@ -72,16 +72,16 @@
"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, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
"blobSequenceNumber": { "kind": "parameter", "displayName": "Blob Sequence Number", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "defaultValue": "0", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "A user-controlled value that you can use to track requests. The value of the sequence numbe [...]
"blockListType": { "kind": "parameter", "displayName": "Block List Type", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.storage.blob.models.BlockListType", "enum": [ "committed", "uncommitted", "all" ], "deprecated": false, "secret": false, "defaultValue": "COMMITTED", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies which ty [...]
- "closeStreamAfterWrite": { "kind": "parameter", "displayName": "Close Stream After Write", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Close the stream after write or keep it open, default is true" },
- "commitBlockListLater": { "kind": "parameter", "displayName": "Commit Block List Later", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the staged blocks will not be committed directly." },
- "createAppendBlob": { "kind": "parameter", "displayName": "Create Append Blob", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the append blocks will be created when committing append blocks." },
- "createPageBlob": { "kind": "parameter", "displayName": "Create Page Blob", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the page blob will be created when uploading page blob." },
+ "closeStreamAfterWrite": { "kind": "parameter", "displayName": "Close Stream After Write", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Close the stream after write or keep it open, default is true" },
+ "commitBlockListLater": { "kind": "parameter", "displayName": "Commit Block List Later", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the staged blocks will not be committed directly." },
+ "createAppendBlob": { "kind": "parameter", "displayName": "Create Append Blob", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the append blocks will be created when committing append blocks." },
+ "createPageBlob": { "kind": "parameter", "displayName": "Create Page Blob", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "When is set to true, the page blob will be created when uploading page blob." },
"downloadLinkExpiration": { "kind": "parameter", "displayName": "Download Link Expiration", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Override the default expiration (millis) of URL download link." },
"lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": 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 starting and cause the [...]
"operation": { "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.storage.blob.BlobOperationsDefinition", "enum": [ "listBlobContainers", "createBlobContainer", "deleteBlobContainer", "listBlobs", "getBlob", "deleteBlob", "downloadBlobToFile", "downloadLink", "uploadBlockBlob", "stageBlockBlobList", "commitBlobBlockList", "getBlobBlockList", "createAppendBlob", " [...]
"pageBlobSize": { "kind": "parameter", "displayName": "Page Blob Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "java.lang.Long", "deprecated": false, "secret": false, "defaultValue": "512", "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Specifies the maximum size for the page blob, up to 8 TB. The page blob size must be aligned to a 512 [...]
"basicPropertyBinding": { "kind": "parameter", "displayName": "Basic Property Binding", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities" },
- "synchronous": { "kind": "parameter", "displayName": "Synchronous", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported)." },
+ "synchronous": { "kind": "parameter", "displayName": "Synchronous", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "description": "Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported)." },
"accessKey": { "kind": "parameter", "displayName": "Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.storage.blob.BlobConfiguration", "configurationField": "configuration", "description": "Access key for the associated azure account name to be used for authentication with azure blob services" }
}
}
diff --git a/components/camel-azure-storage-blob/src/main/docs/azure-storage-blob-component.adoc b/components/camel-azure-storage-blob/src/main/docs/azure-storage-blob-component.adoc
index 3a9944b..2590c33 100644
--- a/components/camel-azure-storage-blob/src/main/docs/azure-storage-blob-component.adoc
+++ b/components/camel-azure-storage-blob/src/main/docs/azure-storage-blob-component.adoc
@@ -43,7 +43,7 @@ for this component:
azure-storage-blob://accountName[/containerName][?options]
------------------------------
-In case of consumer, accountName, containerName and blobName are required. In case of producer, it depends on the operation that being
+In case of consumer, accountName, containerName are required. In case of producer, it depends on the operation that being
requested, for example if operation is on a container level, e.b: createContainer, accountName and containerName are only required, but in case
of operation being requested in blob level, e.g: getBlob, accountName, containerName and blobName are required.
@@ -73,7 +73,7 @@ The Azure Storage Blob Service component supports 27 options, which are listed b
|===
| Name | Description | Default | Type
| *autoDiscoverClient* (common) | Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in the registry automatically otherwise it will skip that checking. | true | boolean
-| *blobName* (common) | The blob name, required for consumer. However on producer, is only required for the operations on the blob level | | String
+| *blobName* (common) | The blob name, to consume specific blob from a container. However on producer, is only required for the operations on the blob level | | String
| *blobOffset* (common) | Set the blob offset for the upload or download operations, default is 0 | 0 | long
| *blobType* (common) | The blob type in order to initiate the appropriate settings for each blob type. There are 3 enums and the value can be one of: blockblob, appendblob, pageblob | blockblob | BlobType
| *closeStreamAfterRead* (common) | Close the stream after read or keep it open, default is true | true | boolean
@@ -129,7 +129,7 @@ with the following path and query parameters:
|===
| Name | Description | Default | Type
| *autoDiscoverClient* (common) | Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in the registry automatically otherwise it will skip that checking. | true | boolean
-| *blobName* (common) | The blob name, required for consumer. However on producer, is only required for the operations on the blob level | | String
+| *blobName* (common) | The blob name, to consume specific blob from a container. However on producer, is only required for the operations on the blob level | | String
| *blobOffset* (common) | Set the blob offset for the upload or download operations, default is 0 | 0 | long
| *blobServiceClient* (common) | Client to a storage account. This client does not hold any state about a particular storage account but is instead a convenient way of sending off appropriate requests to the resource on the service. It may also be used to construct URLs to blobs and containers. This client contains operations on a service account. Operations on a container are available on BlobContainerClient through getBlobContainerClient(String), and operations on a blob are available [...]
| *blobType* (common) | The blob type in order to initiate the appropriate settings for each blob type. There are 3 enums and the value can be one of: blockblob, appendblob, pageblob | blockblob | BlobType
@@ -164,6 +164,7 @@ with the following path and query parameters:
*Required information options:*
To use this component, you have 3 options in order to provide the required Azure authentication information:
+
- Provide `accountName` and `accessKey` for your Azure account, this is the simplest way to get started. The accessKey can
be generated through your Azure portal.
- Provide a https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/common/StorageSharedKeyCredential.html[StorageSharedKeyCredential] instance which can be
@@ -173,6 +174,15 @@ provided into `blobServiceClient`. Note: You don't need to create a specific cli
can be used to retrieve lower level clients.
+== Batch Consumer
+
+This component implements the Batch Consumer.
+
+This allows you for instance to know how many messages exists in this
+batch and for instance let the Aggregator
+aggregate this number of messages.
+
+
== Usage
=== Message headers evaluated by the component producer
@@ -347,6 +357,15 @@ from("azure-storage-blob:/camelazure/container1?blobName=hello.txt&accountName=y
to("mock:results");
--------------------------------------------------------------------------------
+Also, the component supports batch consumer, hence you can consume multiple blobs with only specifying the container name, the consumer will
+return multiple exchanges depending on the number of the blobs in the container. Example:
+[source,java]
+--------------------------------------------------------------------------------
+from("azure-storage-blob:/camelazure/container1?accountName=yourAccountName&accessKey=yourAccessKey&fileDir=/var/to/awesome/dir").
+to("mock:results");
+--------------------------------------------------------------------------------
+
+
=== Producer Operations Examples
- `listBlobContainers`:
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfiguration.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfiguration.java
index a0cec95..d066a88 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfiguration.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfiguration.java
@@ -160,7 +160,8 @@ public class BlobConfiguration implements Cloneable {
}
/**
- * The blob name, required for consumer. However on producer, is only required for the operations on the blob level
+ * The blob name, to consume specific blob from a container. However on producer, is only required for the
+ * operations on the blob level
*/
public String getBlobName() {
return blobName;
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfigurationOptionsProxy.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfigurationOptionsProxy.java
index 5483b4b..9d22760 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfigurationOptionsProxy.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConfigurationOptionsProxy.java
@@ -49,7 +49,7 @@ public class BlobConfigurationOptionsProxy {
}
public ListBlobContainersOptions getListBlobContainersOptions(final Exchange exchange) {
- return BlobExchangeHeaders.getListBlobContainersOptionsFromHeaders(exchange);
+ return getOption(BlobExchangeHeaders::getListBlobContainersOptionsFromHeaders, () -> null, exchange);
}
public Duration getTimeout(final Exchange exchange) {
@@ -57,11 +57,11 @@ public class BlobConfigurationOptionsProxy {
}
public ListBlobsOptions getListBlobsOptions(final Exchange exchange) {
- return BlobExchangeHeaders.getListBlobsOptionsFromHeaders(exchange);
+ return getOption(BlobExchangeHeaders::getListBlobsOptionsFromHeaders, () -> null, exchange);
}
public BlobListDetails getBlobListDetails(final Exchange exchange) {
- return BlobExchangeHeaders.getBlobListDetailsFromHeaders(exchange);
+ return getOption(BlobExchangeHeaders::getBlobListDetailsFromHeaders, () -> null, exchange);
}
public String getPrefix(final Exchange exchange) {
@@ -75,6 +75,10 @@ public class BlobConfigurationOptionsProxy {
public ListBlobsOptions getListBlobOptions(final Exchange exchange) {
ListBlobsOptions blobsOptions = getListBlobsOptions(exchange);
+ if (blobsOptions == null) {
+ blobsOptions = new ListBlobsOptions();
+ }
+
if (!ObjectHelper.isEmpty(blobsOptions)) {
return blobsOptions;
} else {
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java
index e31965a..f46dffb 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java
@@ -16,17 +16,32 @@
*/
package org.apache.camel.component.azure.storage.blob;
-import com.azure.storage.blob.BlobServiceClient;
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.models.BlobItem;
import com.azure.storage.blob.models.BlobStorageException;
import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
import org.apache.camel.Processor;
import org.apache.camel.component.azure.storage.blob.client.BlobClientWrapper;
+import org.apache.camel.component.azure.storage.blob.client.BlobContainerClientWrapper;
+import org.apache.camel.component.azure.storage.blob.operations.BlobContainerOperations;
import org.apache.camel.component.azure.storage.blob.operations.BlobOperationResponse;
import org.apache.camel.component.azure.storage.blob.operations.BlobOperations;
-import org.apache.camel.support.ScheduledPollConsumer;
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.support.ScheduledBatchPollingConsumer;
+import org.apache.camel.util.CastUtils;
import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BlobConsumer extends ScheduledBatchPollingConsumer {
-public class BlobConsumer extends ScheduledPollConsumer {
+ private static final Logger LOG = LoggerFactory.getLogger(BlobConsumer.class);
public BlobConsumer(final BlobEndpoint endpoint, final Processor processor) {
super(endpoint, processor);
@@ -36,26 +51,22 @@ public class BlobConsumer extends ScheduledPollConsumer {
protected int poll() throws Exception {
final String containerName = getEndpoint().getConfiguration().getContainerName();
final String blobName = getEndpoint().getConfiguration().getBlobName();
- final BlobServiceClient serviceClient = getEndpoint().getBlobServiceClient();
- final BlobClientWrapper clientWrapper
- = new BlobClientWrapper(serviceClient.getBlobContainerClient(containerName).getBlobClient(blobName));
- final BlobOperations operations = new BlobOperations(getEndpoint().getConfiguration(), clientWrapper);
- final Exchange exchange = getEndpoint().createExchange();
+ final BlobContainerClient blobContainerClient
+ = getEndpoint().getBlobServiceClient().getBlobContainerClient(containerName);
+
+ Queue<Exchange> exchanges;
try {
- BlobOperationResponse response;
- if (!ObjectHelper.isEmpty(getEndpoint().getConfiguration().getFileDir())) {
- // if we have a fileDir set, we download our content
- response = operations.downloadBlobToFile(exchange);
+ if (ObjectHelper.isNotEmpty(blobName)) {
+ // here we have a blob which means we just download a single blob
+ final Exchange exchange = createExchangeFromBlob(blobName, blobContainerClient);
+ exchanges = new LinkedList<>();
+ exchanges.add(exchange);
} else {
- // otherwise, we rely on the outputstream/inputstream
- response = operations.getBlob(exchange);
+ // download multiple blobs since we only have no blobName set
+ exchanges = createBatchExchangesFromContainer(blobContainerClient);
}
-
- getEndpoint().setResponseOnExchange(response, exchange);
-
- getAsyncProcessor().process(exchange);
- return 1;
+ return processBatch(CastUtils.cast(exchanges));
} catch (BlobStorageException ex) {
if (404 == ex.getStatusCode()) {
return 0;
@@ -65,8 +76,96 @@ public class BlobConsumer extends ScheduledPollConsumer {
}
}
+ private Exchange createExchangeFromBlob(final String blobName, final BlobContainerClient blobContainerClient)
+ throws IOException {
+ final BlobClientWrapper clientWrapper
+ = new BlobClientWrapper(blobContainerClient.getBlobClient(blobName));
+ final BlobOperations operations = new BlobOperations(getEndpoint().getConfiguration(), clientWrapper);
+ final Exchange exchange = getEndpoint().createExchange();
+
+ BlobOperationResponse response;
+ if (!ObjectHelper.isEmpty(getEndpoint().getConfiguration().getFileDir())) {
+ // if we have a fileDir set, we download our content
+ response = operations.downloadBlobToFile(exchange);
+ } else {
+ // otherwise, we rely on the outputstream/inputstream
+ response = operations.getBlob(exchange);
+ }
+
+ getEndpoint().setResponseOnExchange(response, exchange);
+
+ return exchange;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Queue<Exchange> createBatchExchangesFromContainer(final BlobContainerClient blobContainerClient)
+ throws IOException {
+ final BlobContainerClientWrapper containerClientWrapper = new BlobContainerClientWrapper(blobContainerClient);
+ final BlobContainerOperations containerOperations
+ = new BlobContainerOperations(getEndpoint().getConfiguration(), containerClientWrapper);
+
+ final List<BlobItem> blobs = (List<BlobItem>) containerOperations.listBlobs(null).getBody();
+
+ final Queue<Exchange> exchanges = new LinkedList<>();
+
+ for (BlobItem blobItem : blobs) {
+ exchanges.add(createExchangeFromBlob(blobItem.getName(), blobContainerClient));
+ }
+
+ return exchanges;
+ }
+
@Override
public BlobEndpoint getEndpoint() {
return (BlobEndpoint) super.getEndpoint();
}
+
+ @Override
+ public int processBatch(Queue<Object> exchanges) {
+ final int total = exchanges.size();
+
+ for (int index = 0; index < total && isBatchAllowed(); index++) {
+ // only loop if we are started (allowed to run)
+ final Exchange exchange = ObjectHelper.cast(Exchange.class, exchanges.poll());
+
+ // add current index and total as properties
+ exchange.setProperty(Exchange.BATCH_INDEX, index);
+ exchange.setProperty(Exchange.BATCH_SIZE, total);
+ exchange.setProperty(Exchange.BATCH_COMPLETE, index == total - 1);
+
+ // update pending number of exchanges
+ pendingExchanges = total - index - 1;
+
+ // add on completion to handle after work when the exchange is done
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new Synchronization() {
+ @Override
+ public void onComplete(Exchange exchange) {
+ LOG.trace("Complected from processing all exchanges...");
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ processRollback(exchange);
+ }
+ });
+
+ LOG.trace("Processing exchange [{}]...", exchange);
+ getAsyncProcessor().process(exchange, doneSync -> LOG.trace("Processing exchange [{}] done.", exchange));
+ }
+ return total;
+ }
+
+ /**
+ * Strategy when processing the exchange failed.
+ *
+ * @param exchange the exchange
+ */
+ protected void processRollback(Exchange exchange) {
+ Exception cause = exchange.getException();
+ if (cause != null) {
+ LOG.warn("Exchange failed, so rolling back message status: {}", exchange, cause);
+ } else {
+ LOG.warn("Exchange failed, so rolling back message status: {}", exchange);
+ }
+ }
}
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobEndpoint.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobEndpoint.java
index 0cade8f..e8f072f 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobEndpoint.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobEndpoint.java
@@ -62,9 +62,6 @@ public class BlobEndpoint extends DefaultEndpoint {
if (ObjectHelper.isEmpty(configuration.getContainerName())) {
throw new IllegalArgumentException("Container name must be set.");
}
- if (ObjectHelper.isEmpty(configuration.getBlobName())) {
- throw new IllegalArgumentException("BlobName must be set.");
- }
return new BlobConsumer(this, processor);
}
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobExchangeHeaders.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobExchangeHeaders.java
index 44c4f11..04dc931 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobExchangeHeaders.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobExchangeHeaders.java
@@ -45,6 +45,7 @@ import com.azure.storage.blob.models.PageRange;
import com.azure.storage.blob.models.ParallelTransferOptions;
import com.azure.storage.blob.models.PublicAccessType;
import org.apache.camel.Exchange;
+import org.apache.camel.util.ObjectHelper;
public class BlobExchangeHeaders {
@@ -244,7 +245,7 @@ public class BlobExchangeHeaders {
}
private static <T> T getObjectFromHeaders(final Exchange exchange, final String headerName, final Class<T> classType) {
- return exchange.getIn().getHeader(headerName, classType);
+ return ObjectHelper.isEmpty(exchange) ? null : exchange.getIn().getHeader(headerName, classType);
}
public Map<String, Object> toMap() {
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobUtils.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobUtils.java
index 7de4b37..4a2c702 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobUtils.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobUtils.java
@@ -30,7 +30,7 @@ public final class BlobUtils {
}
public static Message getInMessage(final Exchange exchange) {
- return exchange.getIn();
+ return ObjectHelper.isEmpty(exchange) ? null : exchange.getIn();
}
public static Long getInputStreamLength(final InputStream inputStream) throws IOException {
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobContainerOperations.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobContainerOperations.java
index 8d42d9a..f1fea43 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobContainerOperations.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobContainerOperations.java
@@ -45,10 +45,6 @@ public class BlobContainerOperations {
}
public BlobOperationResponse listBlobs(final Exchange exchange) {
- if (exchange == null) {
- return new BlobOperationResponse(client.listBlobs(new ListBlobsOptions(), null));
- }
-
final ListBlobsOptions listBlobOptions = configurationProxy.getListBlobOptions(exchange);
final Duration timeout = configurationProxy.getTimeout(exchange);
@@ -56,12 +52,6 @@ public class BlobContainerOperations {
}
public BlobOperationResponse createContainer(final Exchange exchange) {
- if (exchange == null) {
- final BlobExchangeHeaders blobExchangeHeaders
- = new BlobExchangeHeaders().httpHeaders(client.createContainer(null, null, null));
- return new BlobOperationResponse(true, blobExchangeHeaders.toMap());
- }
-
final Map<String, String> metadata = configurationProxy.getMetadata(exchange);
final PublicAccessType publicAccessType = configurationProxy.getPublicAccessType(exchange);
final Duration timeout = configurationProxy.getTimeout(exchange);
@@ -73,12 +63,6 @@ public class BlobContainerOperations {
}
public BlobOperationResponse deleteContainer(final Exchange exchange) {
- if (exchange == null) {
- final BlobExchangeHeaders blobExchangeHeaders
- = new BlobExchangeHeaders().httpHeaders(client.deleteContainer(null, null));
- return new BlobOperationResponse(true, blobExchangeHeaders.toMap());
- }
-
final BlobRequestConditions blobRequestConditions = configurationProxy.getBlobRequestConditions(exchange);
final Duration timeout = configurationProxy.getTimeout(exchange);
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobOperations.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobOperations.java
index 3c0849c..6080d8e 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobOperations.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobOperations.java
@@ -51,6 +51,7 @@ import com.azure.storage.blob.models.ParallelTransferOptions;
import com.azure.storage.blob.sas.BlobSasPermission;
import com.azure.storage.blob.sas.BlobServiceSasSignatureValues;
import org.apache.camel.Exchange;
+import org.apache.camel.Message;
import org.apache.camel.component.azure.storage.blob.BlobBlock;
import org.apache.camel.component.azure.storage.blob.BlobCommonRequestOptions;
import org.apache.camel.component.azure.storage.blob.BlobConfiguration;
@@ -82,17 +83,10 @@ public class BlobOperations {
}
public BlobOperationResponse getBlob(final Exchange exchange) throws IOException {
- if (exchange == null) {
- final Map<String, Object> blobInputStream = client.openInputStream(new BlobRange(0), null);
- final BlobExchangeHeaders blobExchangeHeaders = BlobExchangeHeaders
- .createBlobExchangeHeadersFromBlobProperties((BlobProperties) blobInputStream.get("properties"));
-
- return new BlobOperationResponse(blobInputStream.get("inputStream"), blobExchangeHeaders.toMap());
- }
-
LOG.trace("Getting a blob [{}] from exchange [{}]...", configurationProxy.getBlobName(exchange), exchange);
- final OutputStream outputStream = BlobUtils.getInMessage(exchange).getBody(OutputStream.class);
+ final Message message = BlobUtils.getInMessage(exchange);
+ final OutputStream outputStream = ObjectHelper.isEmpty(message) ? null : message.getBody(OutputStream.class);
final BlobRange blobRange = configurationProxy.getBlobRange(exchange);
final BlobCommonRequestOptions blobCommonRequestOptions = getCommonRequestOptions(exchange);
@@ -133,18 +127,6 @@ public class BlobOperations {
}
final File fileToDownload = new File(fileDir, client.getBlobName());
-
- if (exchange == null) {
- final Response<BlobProperties> response
- = client.downloadToFileWithResponse(fileToDownload.toString(), null, null, null, null,
- true, null);
- final BlobExchangeHeaders exchangeHeaders
- = BlobExchangeHeaders.createBlobExchangeHeadersFromBlobProperties(response.getValue())
- .httpHeaders(response.getHeaders())
- .fileName(fileToDownload.toString());
-
- return new BlobOperationResponse(fileToDownload, exchangeHeaders.toMap());
- }
final BlobCommonRequestOptions commonRequestOptions = getCommonRequestOptions(exchange);
final BlobRange blobRange = configurationProxy.getBlobRange(exchange);
final ParallelTransferOptions parallelTransferOptions = configurationProxy.getParallelTransferOptions(exchange);
@@ -164,10 +146,6 @@ public class BlobOperations {
}
public BlobOperationResponse deleteBlob(final Exchange exchange) {
- if (exchange == null) {
- return buildResponse(client.delete(null, null, null), true);
- }
-
final BlobCommonRequestOptions commonRequestOptions = getCommonRequestOptions(exchange);
final DeleteSnapshotsOptionType deleteSnapshotsOptionType = configurationProxy.getDeleteSnapshotsOptionType(exchange);
@@ -179,18 +157,8 @@ public class BlobOperations {
final OffsetDateTime offsetDateTime = OffsetDateTime.now();
final long defaultExpirationTime = 60L * 60L; // 1 hour
final BlobSasPermission sasPermission = new BlobSasPermission().setReadPermission(true); // only read access
-
- if (exchange == null) {
- final BlobServiceSasSignatureValues serviceSasSignatureValues
- = new BlobServiceSasSignatureValues(OffsetDateTime.now().plusSeconds(defaultExpirationTime), sasPermission);
- final String url = client.getBlobUrl() + "?" + client.generateSas(serviceSasSignatureValues);
-
- final BlobExchangeHeaders headers = BlobExchangeHeaders.create().downloadLink(url);
-
- return new BlobOperationResponse(true, headers.toMap());
- }
-
final Long expirationMillis = configurationProxy.getDownloadLinkExpiration(exchange);
+
OffsetDateTime offsetDateTimeToSet;
if (expirationMillis != null) {
offsetDateTimeToSet = offsetDateTime.plusSeconds(expirationMillis / 1000);
@@ -304,12 +272,6 @@ public class BlobOperations {
}
public BlobOperationResponse getBlobBlockList(final Exchange exchange) {
- if (exchange == null) {
- final Response<BlockList> response = client.listBlobBlocks(BlockListType.COMMITTED, null, null);
-
- return buildResponse(response, false);
- }
-
LOG.trace("Getting the blob block list [{}] from exchange [{}]...", configurationProxy.getBlobName(exchange), exchange);
final BlockListType blockListType = configurationProxy.getBlockListType(exchange);
@@ -322,12 +284,6 @@ public class BlobOperations {
}
public BlobOperationResponse createAppendBlob(final Exchange exchange) {
- if (exchange == null) {
- final Response<AppendBlobItem> response = client.createAppendBlob(null, null, null, null);
-
- return buildResponse(response, true);
- }
-
LOG.trace("Creating an append blob [{}] from exchange [{}]...", configurationProxy.getBlobName(exchange), exchange);
final BlobCommonRequestOptions commonRequestOptions = getCommonRequestOptions(exchange);
@@ -365,11 +321,6 @@ public class BlobOperations {
}
public BlobOperationResponse createPageBlob(final Exchange exchange) {
- if (exchange == null) {
- final Response<PageBlobItem> response = client.createPageBlob(getPageBlobSize(null), null, null, null, null, null);
- return buildResponse(response, true);
- }
-
LOG.trace("Creating a page blob [{}] from exchange [{}]...", configurationProxy.getBlobName(exchange), exchange);
final Long pageSize = getPageBlobSize(exchange);
@@ -413,11 +364,6 @@ public class BlobOperations {
}
public BlobOperationResponse resizePageBlob(final Exchange exchange) {
- if (exchange == null) {
- final Response<PageBlobItem> response = client.resizePageBlob(getPageBlobSize(null), null, null);
- return buildResponse(response, true);
- }
-
LOG.trace("Resizing a page blob [{}] from exchange [{}]...", configurationProxy.getBlobName(exchange), exchange);
final Long pageSize = getPageBlobSize(exchange);
@@ -501,9 +447,6 @@ public class BlobOperations {
}
private Long getPageBlobSize(final Exchange exchange) {
- if (exchange == null) {
- return BlobConstants.PAGE_BLOB_DEFAULT_SIZE;
- }
// we try to get the size from the page range if exists
final PageRange pageRange = configurationProxy.getPageRange(exchange);
if (pageRange != null) {
diff --git a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobServiceOperations.java b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobServiceOperations.java
index 3439466..631f33d 100644
--- a/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobServiceOperations.java
+++ b/components/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/operations/BlobServiceOperations.java
@@ -41,9 +41,6 @@ public class BlobServiceOperations {
}
public BlobOperationResponse listBlobContainers(final Exchange exchange) {
- if (exchange == null) {
- return new BlobOperationResponse(client.listBlobContainers(null, null));
- }
final ListBlobContainersOptions listBlobContainersOptions = configurationProxy.getListBlobContainersOptions(exchange);
final Duration timeout = configurationProxy.getTimeout(exchange);
diff --git a/components/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobComponentTest.java b/components/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobComponentTest.java
index fe787f6..0fc2066 100644
--- a/components/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobComponentTest.java
+++ b/components/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobComponentTest.java
@@ -127,16 +127,6 @@ class BlobComponentTest extends CamelTestSupport {
assertEquals("blob/sub", endpoint.getConfiguration().getBlobName());
}
- @Test
- void testNoBlobNameConsumer() throws Exception {
- context.getRegistry().bind("creds", storageSharedKeyCredential());
-
- BlobEndpoint endpoint = (BlobEndpoint) context.getEndpoint(
- "azure-storage-blob://camelazure/container?credentials=#creds");
-
- assertThrows(IllegalArgumentException.class, () -> endpoint.createConsumer(null));
- }
-
private StorageSharedKeyCredential storageSharedKeyCredential() {
return new StorageSharedKeyCredential("fakeuser", "fakekey");
}
diff --git a/components/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerIT.java b/components/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerIT.java
index aa763a5..0f2254f 100644
--- a/components/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerIT.java
+++ b/components/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerIT.java
@@ -55,27 +55,33 @@ class BlobConsumerIT extends CamelTestSupport {
@EndpointInject("direct:start")
private ProducerTemplate templateStart;
private String containerName;
+ private String batchContainerName;
private String blobName;
private String blobName2;
private BlobContainerClient containerClient;
+ private BlobContainerClient batchContainerClient;
@BeforeAll
public void prepare() throws Exception {
containerName = RandomStringUtils.randomAlphabetic(5).toLowerCase();
+ batchContainerName = RandomStringUtils.randomAlphabetic(5).toLowerCase();
blobName = RandomStringUtils.randomAlphabetic(5);
blobName2 = RandomStringUtils.randomAlphabetic(5);
BlobConfiguration configuration = new BlobConfiguration();
configuration.setCredentials(storageSharedKeyCredential());
- configuration.setContainerName(containerName);
configuration.setBlobName(blobName);
final BlobServiceClient serviceClient = BlobClientFactory.createBlobServiceClient(configuration);
containerClient = serviceClient.getBlobContainerClient(containerName);
+ batchContainerClient = serviceClient.getBlobContainerClient(batchContainerName);
+
+ configuration.setContainerName(batchContainerName);
// create test container
containerClient.create();
+ batchContainerClient.create();
}
@@ -116,10 +122,57 @@ class BlobConsumerIT extends CamelTestSupport {
assertEquals("Block Blob", bufferedText);
}
+ @Test
+ void testBatchFilePolling() throws InterruptedException, IOException {
+ templateStart.send("direct:createBlob", ExchangePattern.InOnly, exchange -> {
+ exchange.getIn().setBody("Block Batch Blob 1");
+ exchange.getIn().setHeader(BlobConstants.BLOB_CONTAINER_NAME, batchContainerName);
+ exchange.getIn().setHeader(BlobConstants.BLOB_NAME, "test_batch_blob_1");
+ });
+
+ templateStart.send("direct:createBlob", ExchangePattern.InOnly, exchange -> {
+ exchange.getIn().setBody("Block Batch Blob 2");
+ exchange.getIn().setHeader(BlobConstants.BLOB_CONTAINER_NAME, batchContainerName);
+ exchange.getIn().setHeader(BlobConstants.BLOB_NAME, "test_batch_blob_2");
+ });
+
+ // test output stream based
+ final MockEndpoint mockEndpoint = getMockEndpoint("mock:resultBatch");
+ mockEndpoint.expectedMessageCount(2);
+ mockEndpoint.assertIsSatisfied(100);
+
+ final BlobInputStream blobInputStream = mockEndpoint.getExchanges().get(0).getIn().getBody(BlobInputStream.class);
+ final BlobInputStream blobInputStream2 = mockEndpoint.getExchanges().get(1).getIn().getBody(BlobInputStream.class);
+
+ assertNotNull(blobInputStream, "BlobInputStream must be set");
+ assertNotNull(blobInputStream2, "BlobInputStream must be set");
+
+ final String bufferedText = new BufferedReader(new InputStreamReader(blobInputStream)).readLine();
+ final String bufferedText2 = new BufferedReader(new InputStreamReader(blobInputStream2)).readLine();
+
+ assertEquals("Block Batch Blob 1", bufferedText);
+ assertEquals("Block Batch Blob 2", bufferedText2);
+
+ // test file based
+ final MockEndpoint mockEndpointFile = getMockEndpoint("mock:resultBatchFile");
+ mockEndpointFile.expectedMessageCount(2);
+ mockEndpointFile.assertIsSatisfied(100);
+
+ final File file = mockEndpointFile.getExchanges().get(0).getIn().getBody(File.class);
+ final File file2 = mockEndpointFile.getExchanges().get(1).getIn().getBody(File.class);
+
+ assertNotNull(file, "File must be set");
+ assertNotNull(file2, "File must be set");
+
+ assertEquals("Block Batch Blob 1", FileUtils.readFileToString(file, Charset.defaultCharset()));
+ assertEquals("Block Batch Blob 2", FileUtils.readFileToString(file2, Charset.defaultCharset()));
+ }
+
@AfterAll
public void tearDown() {
// delete container
containerClient.delete();
+ batchContainerClient.delete();
}
@Override
@@ -142,6 +195,12 @@ class BlobConsumerIT extends CamelTestSupport {
from("azure-storage-blob://cameldev/" + containerName + "?blobName=" + blobName2 + "&credentials=#creds")
.to("mock:resultOutputStream");
+
+ from("azure-storage-blob://cameldev/" + batchContainerName + "?credentials=#creds")
+ .to("mock:resultBatch");
+
+ from("azure-storage-blob://cameldev/" + batchContainerName + "?credentials=#creds&fileDir="
+ + testDir.toString()).to("mock:resultBatchFile");
}
};
}
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureStorageBlobComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureStorageBlobComponentBuilderFactory.java
index 4420eff..1d8e575 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureStorageBlobComponentBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureStorageBlobComponentBuilderFactory.java
@@ -64,8 +64,8 @@ public interface AzureStorageBlobComponentBuilderFactory {
return this;
}
/**
- * The blob name, required for consumer. However on producer, is only
- * required for the operations on the blob level.
+ * The blob name, to consume specific blob from a container. However on
+ * producer, is only required for the operations on the blob level.
*
* The option is a: <code>java.lang.String</code> type.
*
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BlobEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BlobEndpointBuilderFactory.java
index 963c4c9..eec2f5b 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BlobEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BlobEndpointBuilderFactory.java
@@ -74,8 +74,8 @@ public interface BlobEndpointBuilderFactory {
return this;
}
/**
- * The blob name, required for consumer. However on producer, is only
- * required for the operations on the blob level.
+ * The blob name, to consume specific blob from a container. However on
+ * producer, is only required for the operations on the blob level.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -627,8 +627,8 @@ public interface BlobEndpointBuilderFactory {
return this;
}
/**
- * The blob name, required for consumer. However on producer, is only
- * required for the operations on the blob level.
+ * The blob name, to consume specific blob from a container. However on
+ * producer, is only required for the operations on the blob level.
*
* The option is a: <code>java.lang.String</code> type.
*
@@ -1371,8 +1371,8 @@ public interface BlobEndpointBuilderFactory {
return this;
}
/**
- * The blob name, required for consumer. However on producer, is only
- * required for the operations on the blob level.
+ * The blob name, to consume specific blob from a container. However on
+ * producer, is only required for the operations on the blob level.
*
* The option is a: <code>java.lang.String</code> type.
*
diff --git a/docs/components/modules/ROOT/pages/azure-storage-blob-component.adoc b/docs/components/modules/ROOT/pages/azure-storage-blob-component.adoc
index f2ce79a..dd2d9d9 100644
--- a/docs/components/modules/ROOT/pages/azure-storage-blob-component.adoc
+++ b/docs/components/modules/ROOT/pages/azure-storage-blob-component.adoc
@@ -45,7 +45,7 @@ for this component:
azure-storage-blob://accountName[/containerName][?options]
------------------------------
-In case of consumer, accountName, containerName and blobName are required. In case of producer, it depends on the operation that being
+In case of consumer, accountName, containerName are required. In case of producer, it depends on the operation that being
requested, for example if operation is on a container level, e.b: createContainer, accountName and containerName are only required, but in case
of operation being requested in blob level, e.g: getBlob, accountName, containerName and blobName are required.
@@ -75,7 +75,7 @@ The Azure Storage Blob Service component supports 27 options, which are listed b
|===
| Name | Description | Default | Type
| *autoDiscoverClient* (common) | Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in the registry automatically otherwise it will skip that checking. | true | boolean
-| *blobName* (common) | The blob name, required for consumer. However on producer, is only required for the operations on the blob level | | String
+| *blobName* (common) | The blob name, to consume specific blob from a container. However on producer, is only required for the operations on the blob level | | String
| *blobOffset* (common) | Set the blob offset for the upload or download operations, default is 0 | 0 | long
| *blobType* (common) | The blob type in order to initiate the appropriate settings for each blob type. There are 3 enums and the value can be one of: blockblob, appendblob, pageblob | blockblob | BlobType
| *closeStreamAfterRead* (common) | Close the stream after read or keep it open, default is true | true | boolean
@@ -131,7 +131,7 @@ with the following path and query parameters:
|===
| Name | Description | Default | Type
| *autoDiscoverClient* (common) | Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance in the registry automatically otherwise it will skip that checking. | true | boolean
-| *blobName* (common) | The blob name, required for consumer. However on producer, is only required for the operations on the blob level | | String
+| *blobName* (common) | The blob name, to consume specific blob from a container. However on producer, is only required for the operations on the blob level | | String
| *blobOffset* (common) | Set the blob offset for the upload or download operations, default is 0 | 0 | long
| *blobServiceClient* (common) | Client to a storage account. This client does not hold any state about a particular storage account but is instead a convenient way of sending off appropriate requests to the resource on the service. It may also be used to construct URLs to blobs and containers. This client contains operations on a service account. Operations on a container are available on BlobContainerClient through getBlobContainerClient(String), and operations on a blob are available [...]
| *blobType* (common) | The blob type in order to initiate the appropriate settings for each blob type. There are 3 enums and the value can be one of: blockblob, appendblob, pageblob | blockblob | BlobType
@@ -166,6 +166,7 @@ with the following path and query parameters:
*Required information options:*
To use this component, you have 3 options in order to provide the required Azure authentication information:
+
- Provide `accountName` and `accessKey` for your Azure account, this is the simplest way to get started. The accessKey can
be generated through your Azure portal.
- Provide a https://azuresdkartifacts.blob.core.windows.net/azure-sdk-for-java/staging/apidocs/com/azure/storage/common/StorageSharedKeyCredential.html[StorageSharedKeyCredential] instance which can be
@@ -175,6 +176,15 @@ provided into `blobServiceClient`. Note: You don't need to create a specific cli
can be used to retrieve lower level clients.
+== Batch Consumer
+
+This component implements the Batch Consumer.
+
+This allows you for instance to know how many messages exists in this
+batch and for instance let the Aggregator
+aggregate this number of messages.
+
+
== Usage
=== Message headers evaluated by the component producer
@@ -349,6 +359,15 @@ from("azure-storage-blob:/camelazure/container1?blobName=hello.txt&accountName=y
to("mock:results");
--------------------------------------------------------------------------------
+Also, the component supports batch consumer, hence you can consume multiple blobs with only specifying the container name, the consumer will
+return multiple exchanges depending on the number of the blobs in the container. Example:
+[source,java]
+--------------------------------------------------------------------------------
+from("azure-storage-blob:/camelazure/container1?accountName=yourAccountName&accessKey=yourAccessKey&fileDir=/var/to/awesome/dir").
+to("mock:results");
+--------------------------------------------------------------------------------
+
+
=== Producer Operations Examples
- `listBlobContainers`: