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/08/17 09:05:39 UTC
[camel] branch master updated: CAMEL-15356: Add Azure EventHubs
component (#4091)
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 0b4a7c4 CAMEL-15356: Add Azure EventHubs component (#4091)
0b4a7c4 is described below
commit 0b4a7c411739132cf6bfbaf70c0c222c5666d6d6
Author: Omar Al-Safi <om...@gmail.com>
AuthorDate: Mon Aug 17 11:05:18 2020 +0200
CAMEL-15356: Add Azure EventHubs component (#4091)
* CAMEL-15356: Add initial client configurations
* CAMEL-15356: Add initial client configurations
* CAMEL-15356: Add client and unit tests
* CAMEL-15356: Add producer operation
* CAMEL-15356: Add consumer operation
* CAMEL-15356: Finish up the consumer with unit/IT tests
* CAMEL-15356: Fix CS and polish it up
* CAMEL-15356: Add docs
* CAMEL-15356: Improve async callbacks
* CAMEL-15356: Improve async callbacks
---
bom/camel-bom/pom.xml | 5 +
components/camel-azure-eventhubs/pom.xml | 119 ++
.../eventhubs/EventHubsComponentConfigurer.java | 150 +++
.../eventhubs/EventHubsEndpointConfigurer.java | 153 +++
.../services/org/apache/camel/component.properties | 7 +
.../org/apache/camel/component/azure-eventhubs | 2 +
.../camel/configurer/azure-eventhubs-component | 2 +
.../camel/configurer/azure-eventhubs-endpoint | 2 +
.../component/azure/eventhubs/azure-eventhubs.json | 73 ++
.../src/main/docs/azure-eventhubs-component.adoc | 240 +++++
.../azure/eventhubs/EventHubsComponent.java | 123 +++
.../azure/eventhubs/EventHubsConfiguration.java | 334 ++++++
.../EventHubsConfigurationOptionsProxy.java | 58 +
.../azure/eventhubs/EventHubsConstants.java | 31 +
.../azure/eventhubs/EventHubsConsumer.java | 129 +++
.../azure/eventhubs/EventHubsEndpoint.java | 100 ++
.../azure/eventhubs/EventHubsProducer.java | 77 ++
.../eventhubs/client/EventHubsClientFactory.java | 140 +++
.../operations/EventHubsProducerOperations.java | 106 ++
.../azure/eventhubs/EventHubsComponentTest.java | 123 +++
.../azure/eventhubs/EventHubsConsumerIT.java | 142 +++
.../azure/eventhubs/EventHubsProducerIT.java | 138 +++
.../azure/eventhubs/EventProcessorIT.java | 106 ++
.../azure/eventhubs/EventProcessorTest.java | 53 +
.../camel/component/azure/eventhubs/TestUtils.java | 59 +
.../operations/EventHubsProducerOperationsIT.java | 86 ++
.../src/test/resources/log4j2.properties | 27 +
components/pom.xml | 1 +
core/camel-allcomponents/pom.xml | 4 +
.../component/ComponentsBuilderFactory.java | 14 +
.../dsl/AzureEventhubsComponentBuilderFactory.java | 438 ++++++++
.../src/generated/resources/metadata.json | 21 +
.../builder/endpoint/EndpointBuilderFactory.java | 1 +
.../camel/builder/endpoint/EndpointBuilders.java | 1 +
.../builder/endpoint/StaticEndpointBuilders.java | 53 +
.../dsl/EventHubsEndpointBuilderFactory.java | 1137 ++++++++++++++++++++
parent/pom.xml | 7 +
37 files changed, 4262 insertions(+)
diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml
index 5c243810..0118577 100644
--- a/bom/camel-bom/pom.xml
+++ b/bom/camel-bom/pom.xml
@@ -323,6 +323,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-azure-eventhubs</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-azure-storage-blob</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/components/camel-azure-eventhubs/pom.xml b/components/camel-azure-eventhubs/pom.xml
new file mode 100644
index 0000000..bd06e37
--- /dev/null
+++ b/components/camel-azure-eventhubs/pom.xml
@@ -0,0 +1,119 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>3.5.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-azure-eventhubs</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel :: Azure Event Hubs</name>
+ <description>Camel Azure Event Hubs Component</description>
+
+ <properties>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-support</artifactId>
+ </dependency>
+
+ <!-- azure sdk -->
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-messaging-eventhubs</artifactId>
+ <version>${azure-eventhubs-java-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
+ <version>${azure-eventhubs-checkpointstore-blob-version}</version>
+ </dependency>
+
+ <!-- extras -->
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ </dependency>
+
+ <!-- for testing -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-junit5</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>${commons-lang3-version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <profiles>
+ <profile>
+ <id>fullTests</id>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>${maven-surefire-plugin-version}</version>
+ <executions>
+ <execution>
+ <phase>integration-test</phase>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <excludes>
+ <exclude>none</exclude>
+ </excludes>
+ <includes>
+ <include>**/*IT</include>
+ </includes>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+</project>
diff --git a/components/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentConfigurer.java b/components/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentConfigurer.java
new file mode 100644
index 0000000..df06c08
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentConfigurer.java
@@ -0,0 +1,150 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.azure.eventhubs;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class EventHubsComponentConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ private org.apache.camel.component.azure.eventhubs.EventHubsConfiguration getOrCreateConfiguration(EventHubsComponent target) {
+ if (target.getConfiguration() == null) {
+ target.setConfiguration(new org.apache.camel.component.azure.eventhubs.EventHubsConfiguration());
+ }
+ return target.getConfiguration();
+ }
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ EventHubsComponent target = (EventHubsComponent) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amqpretryoptions":
+ case "amqpRetryOptions": getOrCreateConfiguration(target).setAmqpRetryOptions(property(camelContext, com.azure.core.amqp.AmqpRetryOptions.class, value)); return true;
+ case "amqptransporttype":
+ case "amqpTransportType": getOrCreateConfiguration(target).setAmqpTransportType(property(camelContext, com.azure.core.amqp.AmqpTransportType.class, value)); return true;
+ case "autodiscoverclient":
+ case "autoDiscoverClient": getOrCreateConfiguration(target).setAutoDiscoverClient(property(camelContext, boolean.class, value)); return true;
+ case "basicpropertybinding":
+ case "basicPropertyBinding": target.setBasicPropertyBinding(property(camelContext, boolean.class, value)); return true;
+ case "blobaccesskey":
+ case "blobAccessKey": getOrCreateConfiguration(target).setBlobAccessKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "blobaccountname":
+ case "blobAccountName": getOrCreateConfiguration(target).setBlobAccountName(property(camelContext, java.lang.String.class, value)); return true;
+ case "blobcontainername":
+ case "blobContainerName": getOrCreateConfiguration(target).setBlobContainerName(property(camelContext, java.lang.String.class, value)); return true;
+ case "blobstoragesharedkeycredential":
+ case "blobStorageSharedKeyCredential": getOrCreateConfiguration(target).setBlobStorageSharedKeyCredential(property(camelContext, com.azure.storage.common.StorageSharedKeyCredential.class, value)); return true;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+ case "checkpointstore":
+ case "checkpointStore": getOrCreateConfiguration(target).setCheckpointStore(property(camelContext, com.azure.messaging.eventhubs.CheckpointStore.class, value)); return true;
+ case "configuration": target.setConfiguration(property(camelContext, org.apache.camel.component.azure.eventhubs.EventHubsConfiguration.class, value)); return true;
+ case "connectionstring":
+ case "connectionString": getOrCreateConfiguration(target).setConnectionString(property(camelContext, java.lang.String.class, value)); return true;
+ case "consumergroupname":
+ case "consumerGroupName": getOrCreateConfiguration(target).setConsumerGroupName(property(camelContext, java.lang.String.class, value)); return true;
+ case "eventposition":
+ case "eventPosition": getOrCreateConfiguration(target).setEventPosition(property(camelContext, java.util.Map.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "partitionid":
+ case "partitionId": getOrCreateConfiguration(target).setPartitionId(property(camelContext, java.lang.String.class, value)); return true;
+ case "partitionkey":
+ case "partitionKey": getOrCreateConfiguration(target).setPartitionKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "prefetchcount":
+ case "prefetchCount": getOrCreateConfiguration(target).setPrefetchCount(property(camelContext, int.class, value)); return true;
+ case "producerasyncclient":
+ case "producerAsyncClient": getOrCreateConfiguration(target).setProducerAsyncClient(property(camelContext, com.azure.messaging.eventhubs.EventHubProducerAsyncClient.class, value)); return true;
+ case "sharedaccesskey":
+ case "sharedAccessKey": getOrCreateConfiguration(target).setSharedAccessKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "sharedaccessname":
+ case "sharedAccessName": getOrCreateConfiguration(target).setSharedAccessName(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Map<String, Object> getAllOptions(Object target) {
+ Map<String, Object> answer = new CaseInsensitiveMap();
+ answer.put("amqpRetryOptions", com.azure.core.amqp.AmqpRetryOptions.class);
+ answer.put("amqpTransportType", com.azure.core.amqp.AmqpTransportType.class);
+ answer.put("autoDiscoverClient", boolean.class);
+ answer.put("basicPropertyBinding", boolean.class);
+ answer.put("blobAccessKey", java.lang.String.class);
+ answer.put("blobAccountName", java.lang.String.class);
+ answer.put("blobContainerName", java.lang.String.class);
+ answer.put("blobStorageSharedKeyCredential", com.azure.storage.common.StorageSharedKeyCredential.class);
+ answer.put("bridgeErrorHandler", boolean.class);
+ answer.put("checkpointStore", com.azure.messaging.eventhubs.CheckpointStore.class);
+ answer.put("configuration", org.apache.camel.component.azure.eventhubs.EventHubsConfiguration.class);
+ answer.put("connectionString", java.lang.String.class);
+ answer.put("consumerGroupName", java.lang.String.class);
+ answer.put("eventPosition", java.util.Map.class);
+ answer.put("lazyStartProducer", boolean.class);
+ answer.put("partitionId", java.lang.String.class);
+ answer.put("partitionKey", java.lang.String.class);
+ answer.put("prefetchCount", int.class);
+ answer.put("producerAsyncClient", com.azure.messaging.eventhubs.EventHubProducerAsyncClient.class);
+ answer.put("sharedAccessKey", java.lang.String.class);
+ answer.put("sharedAccessName", java.lang.String.class);
+ return answer;
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ EventHubsComponent target = (EventHubsComponent) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amqpretryoptions":
+ case "amqpRetryOptions": return getOrCreateConfiguration(target).getAmqpRetryOptions();
+ case "amqptransporttype":
+ case "amqpTransportType": return getOrCreateConfiguration(target).getAmqpTransportType();
+ case "autodiscoverclient":
+ case "autoDiscoverClient": return getOrCreateConfiguration(target).isAutoDiscoverClient();
+ case "basicpropertybinding":
+ case "basicPropertyBinding": return target.isBasicPropertyBinding();
+ case "blobaccesskey":
+ case "blobAccessKey": return getOrCreateConfiguration(target).getBlobAccessKey();
+ case "blobaccountname":
+ case "blobAccountName": return getOrCreateConfiguration(target).getBlobAccountName();
+ case "blobcontainername":
+ case "blobContainerName": return getOrCreateConfiguration(target).getBlobContainerName();
+ case "blobstoragesharedkeycredential":
+ case "blobStorageSharedKeyCredential": return getOrCreateConfiguration(target).getBlobStorageSharedKeyCredential();
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return target.isBridgeErrorHandler();
+ case "checkpointstore":
+ case "checkpointStore": return getOrCreateConfiguration(target).getCheckpointStore();
+ case "configuration": return target.getConfiguration();
+ case "connectionstring":
+ case "connectionString": return getOrCreateConfiguration(target).getConnectionString();
+ case "consumergroupname":
+ case "consumerGroupName": return getOrCreateConfiguration(target).getConsumerGroupName();
+ case "eventposition":
+ case "eventPosition": return getOrCreateConfiguration(target).getEventPosition();
+ case "lazystartproducer":
+ case "lazyStartProducer": return target.isLazyStartProducer();
+ case "partitionid":
+ case "partitionId": return getOrCreateConfiguration(target).getPartitionId();
+ case "partitionkey":
+ case "partitionKey": return getOrCreateConfiguration(target).getPartitionKey();
+ case "prefetchcount":
+ case "prefetchCount": return getOrCreateConfiguration(target).getPrefetchCount();
+ case "producerasyncclient":
+ case "producerAsyncClient": return getOrCreateConfiguration(target).getProducerAsyncClient();
+ case "sharedaccesskey":
+ case "sharedAccessKey": return getOrCreateConfiguration(target).getSharedAccessKey();
+ case "sharedaccessname":
+ case "sharedAccessName": return getOrCreateConfiguration(target).getSharedAccessName();
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointConfigurer.java b/components/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointConfigurer.java
new file mode 100644
index 0000000..c3f213e
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/generated/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpointConfigurer.java
@@ -0,0 +1,153 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.azure.eventhubs;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class EventHubsEndpointConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ EventHubsEndpoint target = (EventHubsEndpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amqpretryoptions":
+ case "amqpRetryOptions": target.getConfiguration().setAmqpRetryOptions(property(camelContext, com.azure.core.amqp.AmqpRetryOptions.class, value)); return true;
+ case "amqptransporttype":
+ case "amqpTransportType": target.getConfiguration().setAmqpTransportType(property(camelContext, com.azure.core.amqp.AmqpTransportType.class, value)); return true;
+ case "autodiscoverclient":
+ case "autoDiscoverClient": target.getConfiguration().setAutoDiscoverClient(property(camelContext, boolean.class, value)); return true;
+ case "basicpropertybinding":
+ case "basicPropertyBinding": target.setBasicPropertyBinding(property(camelContext, boolean.class, value)); return true;
+ case "blobaccesskey":
+ case "blobAccessKey": target.getConfiguration().setBlobAccessKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "blobaccountname":
+ case "blobAccountName": target.getConfiguration().setBlobAccountName(property(camelContext, java.lang.String.class, value)); return true;
+ case "blobcontainername":
+ case "blobContainerName": target.getConfiguration().setBlobContainerName(property(camelContext, java.lang.String.class, value)); return true;
+ case "blobstoragesharedkeycredential":
+ case "blobStorageSharedKeyCredential": target.getConfiguration().setBlobStorageSharedKeyCredential(property(camelContext, com.azure.storage.common.StorageSharedKeyCredential.class, value)); return true;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+ case "checkpointstore":
+ case "checkpointStore": target.getConfiguration().setCheckpointStore(property(camelContext, com.azure.messaging.eventhubs.CheckpointStore.class, value)); return true;
+ case "connectionstring":
+ case "connectionString": target.getConfiguration().setConnectionString(property(camelContext, java.lang.String.class, value)); return true;
+ case "consumergroupname":
+ case "consumerGroupName": target.getConfiguration().setConsumerGroupName(property(camelContext, java.lang.String.class, value)); return true;
+ case "eventposition":
+ case "eventPosition": target.getConfiguration().setEventPosition(property(camelContext, java.util.Map.class, value)); return true;
+ case "exceptionhandler":
+ case "exceptionHandler": target.setExceptionHandler(property(camelContext, org.apache.camel.spi.ExceptionHandler.class, value)); return true;
+ case "exchangepattern":
+ case "exchangePattern": target.setExchangePattern(property(camelContext, org.apache.camel.ExchangePattern.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "partitionid":
+ case "partitionId": target.getConfiguration().setPartitionId(property(camelContext, java.lang.String.class, value)); return true;
+ case "partitionkey":
+ case "partitionKey": target.getConfiguration().setPartitionKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "prefetchcount":
+ case "prefetchCount": target.getConfiguration().setPrefetchCount(property(camelContext, int.class, value)); return true;
+ case "producerasyncclient":
+ case "producerAsyncClient": target.getConfiguration().setProducerAsyncClient(property(camelContext, com.azure.messaging.eventhubs.EventHubProducerAsyncClient.class, value)); return true;
+ case "sharedaccesskey":
+ case "sharedAccessKey": target.getConfiguration().setSharedAccessKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "sharedaccessname":
+ case "sharedAccessName": target.getConfiguration().setSharedAccessName(property(camelContext, java.lang.String.class, value)); return true;
+ case "synchronous": target.setSynchronous(property(camelContext, boolean.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Map<String, Object> getAllOptions(Object target) {
+ Map<String, Object> answer = new CaseInsensitiveMap();
+ answer.put("amqpRetryOptions", com.azure.core.amqp.AmqpRetryOptions.class);
+ answer.put("amqpTransportType", com.azure.core.amqp.AmqpTransportType.class);
+ answer.put("autoDiscoverClient", boolean.class);
+ answer.put("basicPropertyBinding", boolean.class);
+ answer.put("blobAccessKey", java.lang.String.class);
+ answer.put("blobAccountName", java.lang.String.class);
+ answer.put("blobContainerName", java.lang.String.class);
+ answer.put("blobStorageSharedKeyCredential", com.azure.storage.common.StorageSharedKeyCredential.class);
+ answer.put("bridgeErrorHandler", boolean.class);
+ answer.put("checkpointStore", com.azure.messaging.eventhubs.CheckpointStore.class);
+ answer.put("connectionString", java.lang.String.class);
+ answer.put("consumerGroupName", java.lang.String.class);
+ answer.put("eventPosition", java.util.Map.class);
+ answer.put("exceptionHandler", org.apache.camel.spi.ExceptionHandler.class);
+ answer.put("exchangePattern", org.apache.camel.ExchangePattern.class);
+ answer.put("lazyStartProducer", boolean.class);
+ answer.put("partitionId", java.lang.String.class);
+ answer.put("partitionKey", java.lang.String.class);
+ answer.put("prefetchCount", int.class);
+ answer.put("producerAsyncClient", com.azure.messaging.eventhubs.EventHubProducerAsyncClient.class);
+ answer.put("sharedAccessKey", java.lang.String.class);
+ answer.put("sharedAccessName", java.lang.String.class);
+ answer.put("synchronous", boolean.class);
+ return answer;
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ EventHubsEndpoint target = (EventHubsEndpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amqpretryoptions":
+ case "amqpRetryOptions": return target.getConfiguration().getAmqpRetryOptions();
+ case "amqptransporttype":
+ case "amqpTransportType": return target.getConfiguration().getAmqpTransportType();
+ case "autodiscoverclient":
+ case "autoDiscoverClient": return target.getConfiguration().isAutoDiscoverClient();
+ case "basicpropertybinding":
+ case "basicPropertyBinding": return target.isBasicPropertyBinding();
+ case "blobaccesskey":
+ case "blobAccessKey": return target.getConfiguration().getBlobAccessKey();
+ case "blobaccountname":
+ case "blobAccountName": return target.getConfiguration().getBlobAccountName();
+ case "blobcontainername":
+ case "blobContainerName": return target.getConfiguration().getBlobContainerName();
+ case "blobstoragesharedkeycredential":
+ case "blobStorageSharedKeyCredential": return target.getConfiguration().getBlobStorageSharedKeyCredential();
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return target.isBridgeErrorHandler();
+ case "checkpointstore":
+ case "checkpointStore": return target.getConfiguration().getCheckpointStore();
+ case "connectionstring":
+ case "connectionString": return target.getConfiguration().getConnectionString();
+ case "consumergroupname":
+ case "consumerGroupName": return target.getConfiguration().getConsumerGroupName();
+ case "eventposition":
+ case "eventPosition": return target.getConfiguration().getEventPosition();
+ case "exceptionhandler":
+ case "exceptionHandler": return target.getExceptionHandler();
+ case "exchangepattern":
+ case "exchangePattern": return target.getExchangePattern();
+ case "lazystartproducer":
+ case "lazyStartProducer": return target.isLazyStartProducer();
+ case "partitionid":
+ case "partitionId": return target.getConfiguration().getPartitionId();
+ case "partitionkey":
+ case "partitionKey": return target.getConfiguration().getPartitionKey();
+ case "prefetchcount":
+ case "prefetchCount": return target.getConfiguration().getPrefetchCount();
+ case "producerasyncclient":
+ case "producerAsyncClient": return target.getConfiguration().getProducerAsyncClient();
+ case "sharedaccesskey":
+ case "sharedAccessKey": return target.getConfiguration().getSharedAccessKey();
+ case "sharedaccessname":
+ case "sharedAccessName": return target.getConfiguration().getSharedAccessName();
+ case "synchronous": return target.isSynchronous();
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component.properties b/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component.properties
new file mode 100644
index 0000000..576af6d
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component.properties
@@ -0,0 +1,7 @@
+# Generated by camel build tools - do NOT edit this file!
+components=azure-eventhubs
+groupId=org.apache.camel
+artifactId=camel-azure-eventhubs
+version=3.5.0-SNAPSHOT
+projectName=Camel :: Azure Event Hubs
+projectDescription=Camel Azure Event Hubs Component
diff --git a/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component/azure-eventhubs b/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component/azure-eventhubs
new file mode 100644
index 0000000..bf24c2f
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/component/azure-eventhubs
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.azure.eventhubs.EventHubsComponent
diff --git a/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-eventhubs-component b/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-eventhubs-component
new file mode 100644
index 0000000..927b28a
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-eventhubs-component
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.azure.eventhubs.EventHubsComponentConfigurer
diff --git a/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-eventhubs-endpoint b/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-eventhubs-endpoint
new file mode 100644
index 0000000..d500a7e
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-eventhubs-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.azure.eventhubs.EventHubsEndpointConfigurer
diff --git a/components/camel-azure-eventhubs/src/generated/resources/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json b/components/camel-azure-eventhubs/src/generated/resources/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
new file mode 100644
index 0000000..8b50393
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/generated/resources/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
@@ -0,0 +1,73 @@
+{
+ "component": {
+ "kind": "component",
+ "name": "azure-eventhubs",
+ "title": "Azure Event Hubs",
+ "description": "The azure-eventhubs component that integrates Azure Event Hubs using AMQP protocol. Azure EventHubs is a highly scalable publish-subscribe service that can ingest millions of events per second and stream them to multiple consumers.",
+ "deprecated": false,
+ "firstVersion": "3.5.0",
+ "label": "cloud,messaging",
+ "javaType": "org.apache.camel.component.azure.eventhubs.EventHubsComponent",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-azure-eventhubs",
+ "version": "3.5.0-SNAPSHOT",
+ "scheme": "azure-eventhubs",
+ "extendsScheme": "",
+ "syntax": "azure-eventhubs:namespace\/eventHubName",
+ "async": false,
+ "consumerOnly": false,
+ "producerOnly": false,
+ "lenientProperties": false
+ },
+ "componentProperties": {
+ "amqpRetryOptions": { "kind": "property", "displayName": "Amqp Retry Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpRetryOptions", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the retry policy for EventHubAsyncClient. If not specified, the default retry options are used." },
+ "amqpTransportType": { "kind": "property", "displayName": "Amqp Transport Type", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpTransportType", "enum": [ "Amqp", "AmqpWebSockets" ], "deprecated": false, "secret": false, "defaultValue": "Amqp", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the transport type by which [...]
+ "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.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance i [...]
+ "configuration": { "kind": "property", "displayName": "Configuration", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "deprecated": false, "secret": false, "description": "The component configurations" },
+ "blobAccessKey": { "kind": "property", "displayName": "Blob Access Key", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets access key for the associated azure account name to be used [...]
+ "blobAccountName": { "kind": "property", "displayName": "Blob Account Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets Azure account name to be used for authentication with a [...]
+ "blobContainerName": { "kind": "property", "displayName": "Blob Container Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets the blob container that shall be used by the BlobCh [...]
+ "blobStorageSharedKeyCredential": { "kind": "property", "displayName": "Blob Storage Shared Key Credential", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.storage.common.StorageSharedKeyCredential", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, [...]
+ "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 [...]
+ "checkpointStore": { "kind": "property", "displayName": "Checkpoint Store", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.CheckpointStore", "deprecated": false, "secret": false, "defaultValue": "BlobCheckpointStore", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the CheckpointStore the EventProcessorClient [...]
+ "consumerGroupName": { "kind": "property", "displayName": "Consumer Group Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "$Default", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the name of the consumer group this consumer is associated with. Events are rea [...]
+ "eventPosition": { "kind": "property", "displayName": "Event Position", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, com.azure.messaging.eventhubs.models.EventPosition>", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the map containing the event position to use for [...]
+ "prefetchCount": { "kind": "property", "displayName": "Prefetch Count", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the count used by the receiver to control the number of events the Event Hub consumer will actively recei [...]
+ "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 [...]
+ "partitionId": { "kind": "property", "displayName": "Partition Id", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the identifier of the Event Hub partition that the {link EventData events} will be sent to. If the identifier is not spe [...]
+ "partitionKey": { "kind": "property", "displayName": "Partition Key", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key to a spec [...]
+ "producerAsyncClient": { "kind": "property", "displayName": "Producer Async Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.EventHubProducerAsyncClient", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the EventHubProducerAsyncClient.An asynchronous producer respo [...]
+ "basicPropertyBinding": { "kind": "property", "displayName": "Basic Property Binding", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities" },
+ "connectionString": { "kind": "property", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Instead of supplying namespace, sharedAccessKey, sharedAccessName ... etc, you can just supply the connection string [...]
+ "sharedAccessKey": { "kind": "property", "displayName": "Shared Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The generated value for the SharedAccessName" },
+ "sharedAccessName": { "kind": "property", "displayName": "Shared Access Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The name you chose for your EventHubs SAS keys" }
+ },
+ "properties": {
+ "namespace": { "kind": "path", "displayName": "Namespace", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "EventHubs namespace created in Azure Portal" },
+ "eventHubName": { "kind": "path", "displayName": "Event Hub Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "EventHubs name under a specific namcespace" },
+ "amqpRetryOptions": { "kind": "parameter", "displayName": "Amqp Retry Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpRetryOptions", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the retry policy for EventHubAsyncClient. If not specified, the default retry options are used." },
+ "amqpTransportType": { "kind": "parameter", "displayName": "Amqp Transport Type", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpTransportType", "enum": [ "Amqp", "AmqpWebSockets" ], "deprecated": false, "secret": false, "defaultValue": "Amqp", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the transport type by which [...]
+ "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.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Setting the autoDiscoverClient mechanism, if true, the component will look for a client instance [...]
+ "blobAccessKey": { "kind": "parameter", "displayName": "Blob Access Key", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets access key for the associated azure account name to be used [...]
+ "blobAccountName": { "kind": "parameter", "displayName": "Blob Account Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets Azure account name to be used for authentication with [...]
+ "blobContainerName": { "kind": "parameter", "displayName": "Blob Container Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore, this sets the blob container that shall be used by the BlobC [...]
+ "blobStorageSharedKeyCredential": { "kind": "parameter", "displayName": "Blob Storage Shared Key Credential", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.storage.common.StorageSharedKeyCredential", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "In case you chose the default BlobCheckpointStore [...]
+ "bridgeErrorHandler": { "kind": "parameter", "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 b [...]
+ "checkpointStore": { "kind": "parameter", "displayName": "Checkpoint Store", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.CheckpointStore", "deprecated": false, "secret": false, "defaultValue": "BlobCheckpointStore", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the CheckpointStore the EventProcessorClien [...]
+ "consumerGroupName": { "kind": "parameter", "displayName": "Consumer Group Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "defaultValue": "$Default", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the name of the consumer group this consumer is associated with. Events are re [...]
+ "eventPosition": { "kind": "parameter", "displayName": "Event Position", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, com.azure.messaging.eventhubs.models.EventPosition>", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the map containing the event position to use for [...]
+ "prefetchCount": { "kind": "parameter", "displayName": "Prefetch Count", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "500", "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the count used by the receiver to control the number of events the Event Hub consumer will actively rece [...]
+ "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, "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 consumer will deal with [...]
+ "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." },
+ "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 [...]
+ "partitionId": { "kind": "parameter", "displayName": "Partition Id", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the identifier of the Event Hub partition that the {link EventData events} will be sent to. If the identifier is not sp [...]
+ "partitionKey": { "kind": "parameter", "displayName": "Partition Key", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key to a spe [...]
+ "producerAsyncClient": { "kind": "parameter", "displayName": "Producer Async Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.eventhubs.EventHubProducerAsyncClient", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Sets the EventHubProducerAsyncClient.An asynchronous producer resp [...]
+ "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)." },
+ "connectionString": { "kind": "parameter", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "Instead of supplying namespace, sharedAccessKey, sharedAccessName ... etc, you can just supply the connection strin [...]
+ "sharedAccessKey": { "kind": "parameter", "displayName": "Shared Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The generated value for the SharedAccessName" },
+ "sharedAccessName": { "kind": "parameter", "displayName": "Shared Access Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.eventhubs.EventHubsConfiguration", "configurationField": "configuration", "description": "The name you chose for your EventHubs SAS keys" }
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/main/docs/azure-eventhubs-component.adoc b/components/camel-azure-eventhubs/src/main/docs/azure-eventhubs-component.adoc
new file mode 100644
index 0000000..9320bad
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/docs/azure-eventhubs-component.adoc
@@ -0,0 +1,240 @@
+[[azure-eventhubs-component]]
+= Azure Event Hubs Component
+:docTitle: Azure Event Hubs
+:artifactId: camel-azure-eventhubs
+:description: The azure-eventhubs component that integrates Azure Event Hubs using AMQP protocol. Azure EventHubs is a highly scalable publish-subscribe service that can ingest millions of events per second and stream them to multiple consumers.
+:since: 3.5
+:supportLevel: Preview
+:component-header: Both producer and consumer are supported
+
+*Since Camel {since}*
+
+*{component-header}*
+
+The Azure Event Hubs used to integrate https://azure.microsoft.com/en-us/services/event-hubs/[Azure Event Hubs] using https://en.wikipedia.org/wiki/Advanced_Message_Queuing_Protocol[AMQP protocol].
+Azure EventHubs is a highly scalable publish-subscribe service that can ingest millions of events per second and stream them to multiple consumers.
+
+NOTE: Besides AMQP protocol support, Event Hubs as well supports Kafka and HTTPS protocols. Therefore, you can use as well xref:components::kafka-component.adoc[Camel Kafka] component to produce and consume to Azure Event Hubs. You can lean more https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-quickstart-kafka-enabled-event-hubs[here].
+
+
+Prerequisites
+
+You must have a valid Windows Azure Event Hubs account. More information is available at
+https://docs.microsoft.com/azure/[Azure Documentation Portal].
+
+Maven users will need to add the following dependency to their `pom.xml`
+for this component:
+
+[source,xml]
+------------------------------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-azure-eventhubs</artifactId>
+ <version>x.x.x</version>
+ <!-- use the same version as your Camel core version -->
+</dependency>
+------------------------------------------------------------
+
+== URI Format
+
+[source,text]
+------------------------------
+azure-eventhubs://[namespace/eventHubName][?options]
+------------------------------
+
+In case you supply the `connectionString`, `namespace` and `eventHubName` are not required as these options already included
+in the `connectionString`
+
+For example in order consume event from EventHub, use the following snippet:
+[source,java]
+--------------------------------------------------------------------------------
+from("azure-eventhubs:/camel/camelHub?sharedAccessName=SASaccountName&sharedAccessKey=SASaccessKey&blobAccountName=accountName&blobAccessKey=accessKey&blobContainerName=containerName").
+to("file://queuedirectory");
+--------------------------------------------------------------------------------
+
+== URI Options
+
+// endpoint options: START
+The Azure Event Hubs endpoint is configured using URI syntax:
+
+----
+azure-eventhubs:namespace/eventHubName
+----
+
+with the following path and query parameters:
+
+=== Path Parameters (2 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *namespace* | EventHubs namespace created in Azure Portal | | String
+| *eventHubName* | EventHubs name under a specific namcespace | | String
+|===
+
+
+=== Query Parameters (23 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *amqpRetryOptions* (common) | Sets the retry policy for EventHubAsyncClient. If not specified, the default retry options are used. | | AmqpRetryOptions
+| *amqpTransportType* (common) | Sets the transport type by which all the communication with Azure Event Hubs occurs. Default value is AmqpTransportType#AMQP. The value can be one of: Amqp, AmqpWebSockets | Amqp | AmqpTransportType
+| *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
+| *blobAccessKey* (consumer) | In case you chose the default BlobCheckpointStore, this sets access key for the associated azure account name to be used for authentication with azure blob services | | String
+| *blobAccountName* (consumer) | In case you chose the default BlobCheckpointStore, this sets Azure account name to be used for authentication with azure blob services. | | String
+| *blobContainerName* (consumer) | In case you chose the default BlobCheckpointStore, this sets the blob container that shall be used by the BlobCheckpointStore to store the checkpoint offsets | | String
+| *blobStorageSharedKeyCredential* (consumer) | In case you chose the default BlobCheckpointStore, StorageSharedKeyCredential can be injected to create the azure client, this holds the important authentication information | | StorageSharedKeyCredential
+| *bridgeErrorHandler* (consumer) | 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 the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
+| *checkpointStore* (consumer) | Sets the CheckpointStore the EventProcessorClient will use for storing partition ownership and checkpoint information. Users can, optionally, provide their own implementation of CheckpointStore which will store ownership and checkpoint information. By default it set to use com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore which stores all checkpoint offsets into Azure Blob Storage | BlobCheckpointStore | CheckpointStore
+| *consumerGroupName* (consumer) | Sets the name of the consumer group this consumer is associated with. Events are read in the context of this group. The name of the consumer group that is created by default is {link #DEFAULT_CONSUMER_GROUP_NAME \$Default}. | $Default | String
+| *eventPosition* (consumer) | Sets the map containing the event position to use for each partition if a checkpoint for the partition does not exist in CheckpointStore. This map is keyed off of the partition id. If there is no checkpoint in CheckpointStore and there is no entry in this map, the processing of the partition will start from {link EventPosition#latest() latest} position. | | Map
+| *prefetchCount* (consumer) | Sets the count used by the receiver to control the number of events the Event Hub consumer will actively receive and queue locally without regard to whether a receive operation is currently active. | 500 | int
+| *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. The value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
+| *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 [...]
+| *partitionId* (producer) | Sets the identifier of the Event Hub partition that the {link EventData events} will be sent to. If the identifier is not specified, the Event Hubs service will be responsible for routing events that are sent to an available partition. | | String
+| *partitionKey* (producer) | Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key to a specific partition. The selection of a partition is stable for a given partition hashing key. Should any other batches of events be sent using the same exact partition hashing key, the Event Hubs service will route them all to the same partition. This should be specified only when there is a need to group events by partition, but there is fl [...]
+| *producerAsyncClient* (producer) | Sets the EventHubProducerAsyncClient.An asynchronous producer responsible for transmitting EventData to a specific Event Hub, grouped together in batches. Depending on the {link CreateBatchOptions options} specified when creating an \{linkEventDataBatch\}, the events may be automatically routed to an available partition or specific to a partition. Use by this component to produce the data in camel producer. | | EventHubProducerAsyncClient
+| *basicPropertyBinding* (advanced) | Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
+| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+| *connectionString* (security) | Instead of supplying namespace, sharedAccessKey, sharedAccessName ... etc, you can just supply the connection string for your eventHub. The connection string for EventHubs already include all the necessary information to connection to your EventHub. To learn on how to generate the connection string, take a look at this documentation: \https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string | | String
+| *sharedAccessKey* (security) | The generated value for the SharedAccessName | | String
+| *sharedAccessName* (security) | The name you chose for your EventHubs SAS keys | | String
+|===
+// endpoint options: END
+
+// component options: START
+The Azure Event Hubs component supports 21 options, which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *amqpRetryOptions* (common) | Sets the retry policy for EventHubAsyncClient. If not specified, the default retry options are used. | | AmqpRetryOptions
+| *amqpTransportType* (common) | Sets the transport type by which all the communication with Azure Event Hubs occurs. Default value is AmqpTransportType#AMQP. The value can be one of: Amqp, AmqpWebSockets | Amqp | AmqpTransportType
+| *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
+| *configuration* (common) | The component configurations | | EventHubsConfiguration
+| *blobAccessKey* (consumer) | In case you chose the default BlobCheckpointStore, this sets access key for the associated azure account name to be used for authentication with azure blob services | | String
+| *blobAccountName* (consumer) | In case you chose the default BlobCheckpointStore, this sets Azure account name to be used for authentication with azure blob services. | | String
+| *blobContainerName* (consumer) | In case you chose the default BlobCheckpointStore, this sets the blob container that shall be used by the BlobCheckpointStore to store the checkpoint offsets | | String
+| *blobStorageSharedKeyCredential* (consumer) | In case you chose the default BlobCheckpointStore, StorageSharedKeyCredential can be injected to create the azure client, this holds the important authentication information | | StorageSharedKeyCredential
+| *bridgeErrorHandler* (consumer) | 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 the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
+| *checkpointStore* (consumer) | Sets the CheckpointStore the EventProcessorClient will use for storing partition ownership and checkpoint information. Users can, optionally, provide their own implementation of CheckpointStore which will store ownership and checkpoint information. By default it set to use com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore which stores all checkpoint offsets into Azure Blob Storage | BlobCheckpointStore | CheckpointStore
+| *consumerGroupName* (consumer) | Sets the name of the consumer group this consumer is associated with. Events are read in the context of this group. The name of the consumer group that is created by default is {link #DEFAULT_CONSUMER_GROUP_NAME \$Default}. | $Default | String
+| *eventPosition* (consumer) | Sets the map containing the event position to use for each partition if a checkpoint for the partition does not exist in CheckpointStore. This map is keyed off of the partition id. If there is no checkpoint in CheckpointStore and there is no entry in this map, the processing of the partition will start from {link EventPosition#latest() latest} position. | | Map
+| *prefetchCount* (consumer) | Sets the count used by the receiver to control the number of events the Event Hub consumer will actively receive and queue locally without regard to whether a receive operation is currently active. | 500 | int
+| *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 [...]
+| *partitionId* (producer) | Sets the identifier of the Event Hub partition that the {link EventData events} will be sent to. If the identifier is not specified, the Event Hubs service will be responsible for routing events that are sent to an available partition. | | String
+| *partitionKey* (producer) | Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key to a specific partition. The selection of a partition is stable for a given partition hashing key. Should any other batches of events be sent using the same exact partition hashing key, the Event Hubs service will route them all to the same partition. This should be specified only when there is a need to group events by partition, but there is fl [...]
+| *producerAsyncClient* (producer) | Sets the EventHubProducerAsyncClient.An asynchronous producer responsible for transmitting EventData to a specific Event Hub, grouped together in batches. Depending on the {link CreateBatchOptions options} specified when creating an \{linkEventDataBatch\}, the events may be automatically routed to an available partition or specific to a partition. Use by this component to produce the data in camel producer. | | EventHubProducerAsyncClient
+| *basicPropertyBinding* (advanced) | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
+| *connectionString* (security) | Instead of supplying namespace, sharedAccessKey, sharedAccessName ... etc, you can just supply the connection string for your eventHub. The connection string for EventHubs already include all the necessary information to connection to your EventHub. To learn on how to generate the connection string, take a look at this documentation: \https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string | | String
+| *sharedAccessKey* (security) | The generated value for the SharedAccessName | | String
+| *sharedAccessName* (security) | The name you chose for your EventHubs SAS keys | | String
+|===
+// component options: END
+
+
+== Authentication Information
+
+To use this component, you have 3 options in order to provide the required Azure authentication information:
+
+- Provide `sharedAccessName` and `sharedAccessKey` for your Azure Event Hubs account. The sharedAccessKey can
+be generated through your Event Hubs Azure portal.
+- Provide `connectionString` string, if you provide the connection string, you don't supply `namespace`, `eventHubName`, `sharedAccessKey` and `sharedAccessName`
+as these data already included in the `connectionString`, therefore is the simplest option to get started. Learn more https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string[here] on how to generate the connection string.
+- Provide a https://docs.microsoft.com/en-us/java/api/com.azure.messaging.eventhubs.eventhubproducerasyncclient?view=azure-java-stable[EventHubProducerAsyncClient] instance which can be
+provided into `producerAsyncClient`. However, this is *only possible for camel producer*, for the camel consumer, is not possible to inject the client due to some design constrain by the `EventProcessorClient`.
+
+
+== Checkpoint Store Information
+A checkpoint store stores and retrieves partition ownership information and checkpoint details for each partition in a given consumer group of an event hub instance. Users are not meant to implement an CheckpointStore.
+Users are expected to choose existing implementations of this interface, instantiate it, and pass it to the component through `checkpointStore` option.
+Users are not expected to use any of the methods on a checkpoint store, these are used internally by the client.
+
+Having said that, if the user does not pass any `CheckpointStore` implementation, the component will fallback to use https://docs.microsoft.com/en-us/javascript/api/@azure/eventhubs-checkpointstore-blob/blobcheckpointstore?view=azure-node-latest[`BlobCheckpointStore`] to store the checkpoint info in Azure Blob Storage account.
+If you chose to use the default `BlobCheckpointStore`, you will need to supply the following options:
+
+- `blobAccountName`: It sets Azure account name to be used for authentication with azure blob services.
+- `blobAccessKey` : It sets access key for the associated azure account name to be used for authentication with azure blob services.
+- `blobContainerName` : It sets the blob container that shall be used by the BlobCheckpointStore to store the checkpoint offsets.
+
+
+== Async Consumer and Producer
+
+This component implements the async Consumer and producer.
+
+This allows camel route to consume and produce events asynchronously without blocking any threads.
+
+
+== Usage
+
+=== Message headers evaluated by the component producer
+[width="100%",cols="10%,10%,10%,70%",options="header",]
+|=======================================================================
+|Header |Variable Name |Type |Description
+
+|`CamelAzureEventHubsPartitionKey`| `EventHubsConstants.PARTITION_KEY`|`String`| Overrides the hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key to a specific partition.
+|`CamelAzureEventHubsPartitionId`| `EventHubsConstants.PARTITION_ID`|`String`| Overrides the identifier of the Event Hub partition that the {link EventData events} will be sent to.
+|=======================================================================
+
+
+=== Message headers set by the component consumer
+[width="100%",cols="10%,10%,10%,70%",options="header",]
+|=======================================================================
+|Header |Variable Name |Type |Description
+
+|`CamelAzureEventHubsPartitionKey`| `EventHubsConstants.PARTITION_KEY`|`String`| It sets the partition hashing key if it was set when originally publishing the event. If it exists, this value was used to compute a hash to select a partition to send the message to. This is only present on a received {@link EventData}.
+|`CamelAzureEventHubsPartitionId`| `EventHubsConstants.PARTITION_ID`|`String`| It sets the partition id of the Event Hub.
+|`CamelAzureEventHubsOffset`| `EventHubsConstants.OFFSET`|`Long`| It sets the offset of the event when it was received from the associated Event Hub partition. This is only present on a received {@link EventData}.
+|`CamelAzureEventHubsEnqueuedTime`| `EventHubsConstants.ENQUEUED_TIME`|`Instant`| It sets the instant, in UTC, of when the event was enqueued in the Event Hub partition. This is only present on a received {@link EventData}.
+|`CamelAzureEventHubsSequenceNumber`| `EventHubsConstants.SEQUENCE_NUMBER`|`Long`| It sets the sequence number assigned to the event when it was enqueued in the associated Event Hub partition. This is unique for every message received in the Event Hub partition. This is only present on a received {@link EventData}.
+|=======================================================================
+
+=== Message body type
+The component's producer expects the data in the message body to be in `byte[]`. This allows the user to utilize Camel TypeConverter to marshal/unmarshal data with ease.
+The same goes as well for the component's consumer, it will set the encoded data as `byte[]` in the message body.
+
+
+=== Automatic detection of EventHubProducerAsyncClient client in registry
+
+The component is capable of detecting the presence of an EventHubProducerAsyncClient bean into the registry.
+If it's the only instance of that type it will be used as client and you won't have to define it as uri parameter, like the example above.
+This may be really useful for smarter configuration of the endpoint.
+
+=== Consumer Example
+The example below will unmarshal the events that was originally produced in JSON:
+```
+from("azure-eventhubs:?connectionString=RAW({{connectionString}})"&blobContainerName=containerTest&eventPosition=#eventPosition"
+ +"&blobAccountName={{blobAccountName}}&blobAccessKey=RAW({{blobAccessKey}})")
+.unmarshal().json(JsonLibrary.Jackson)
+.to(result);
+```
+
+=== Producer Example
+The example below will send events as String to EventHubs:
+```
+from("direct:start")
+.process(exchange -> {
+ exchange.getIn().setHeader(EventHubsConstants.PARTITION_ID, firstPartition);
+ exchange.getIn().setBody("test event");
+})
+.to("azure-eventhubs:?connectionString=RAW({{connectionString}})"
+```
+
+=== Development Notes (Important)
+When developing on this component, you will need to obtain your Azure accessKey in order to run the integration tests. In addition to the mocked unit tests
+you *will need to run the integration tests with every change you make or even client upgrade as the Azure client can break things even on minor versions upgrade.*
+To run the integration tests, on this component directory, run the following maven command:
+----
+mvn verify -PfullTests -DconnectionString=string -DblobAccountName=blob -DblobAccessKey=key
+----
+Whereby `blobAccountName` is your Azure account name and `blobAccessKey` is the access key being generated from Azure portal and `connectionString` is the eventHub connection string.
+
+
+include::camel-spring-boot::page$azure-eventhubs-starter.adoc[]
+
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsComponent.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsComponent.java
new file mode 100644
index 0000000..40aaacc
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsComponent.java
@@ -0,0 +1,123 @@
+/*
+ * 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.azure.eventhubs;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import org.apache.camel.Endpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Azure EventHubs component
+ */
+@Component("azure-eventhubs")
+public class EventHubsComponent extends DefaultComponent {
+
+ private static final Logger LOG = LoggerFactory.getLogger(EventHubsComponent.class);
+
+ @Metadata
+ private EventHubsConfiguration configuration = new EventHubsConfiguration();
+
+ public EventHubsComponent() {
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+
+ final EventHubsConfiguration configuration = this.configuration != null ? this.configuration.copy() : new EventHubsConfiguration();
+
+ final EventHubsEndpoint endpoint = new EventHubsEndpoint(uri, this, configuration);
+ setProperties(endpoint, parameters);
+
+ if (configuration.isAutoDiscoverClient()) {
+ checkAndSetRegistryClient(configuration::setProducerAsyncClient, configuration::getProducerAsyncClient, EventHubProducerAsyncClient.class);
+ }
+
+ // if we don't have client nor connectionString, we check for params
+ if (areAzureClientsNotSet(configuration) && ObjectHelper.isEmpty(configuration.getConnectionString())) {
+ checkAndSetNamespaceAndHubName(configuration, remaining);
+ validateConfigurations(configuration);
+ }
+
+ return endpoint;
+ }
+
+ /**
+ * The component configurations
+ */
+ public EventHubsConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(EventHubsConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ private <C> void checkAndSetRegistryClient(final Consumer<C> setClientFn, final Supplier<C> getClientFn, final Class<C> clientType) {
+ if (ObjectHelper.isEmpty(getClientFn.get())) {
+ final Set<C> clients = getCamelContext().getRegistry().findByType(clientType);
+ if (clients.size() == 1) {
+ setClientFn.accept(clients.stream().findFirst().get());
+ } else if (clients.size() > 1) {
+ LOG.info(String.format("More than one %s instance in the registry, make sure to have only one instance", clientType.getSimpleName()));
+ } else {
+ LOG.info(String.format("No %s instance in the registry", clientType.getSimpleName()));
+ }
+ } else {
+ LOG.info(String.format("%s instance is already set at endpoint level: skipping the check in the registry", clientType.getSimpleName()));
+ }
+ }
+
+ private void validateConfigurations(final EventHubsConfiguration configuration) {
+ if (!isAccessKeyAndAccessNameSet(configuration)) {
+ throw new IllegalArgumentException("Azure EventHubs SharedAccessName/SharedAccessKey, ConsumerAsyncClient/ProducerAsyncClient "
+ + "or connectionString must be specified.");
+ }
+ }
+
+ private boolean isAccessKeyAndAccessNameSet(final EventHubsConfiguration configuration) {
+ return ObjectHelper.isNotEmpty(configuration.getSharedAccessName()) && ObjectHelper.isNotEmpty(configuration.getSharedAccessKey());
+ }
+
+ private boolean areAzureClientsNotSet(final EventHubsConfiguration configuration) {
+ return ObjectHelper.isEmpty(configuration.getProducerAsyncClient());
+ }
+
+ private void checkAndSetNamespaceAndHubName(final EventHubsConfiguration configuration, final String remaining) {
+ // only set if clients are empty and remaining exists
+ if (ObjectHelper.isEmpty(remaining)) {
+ throw new IllegalArgumentException("ConnectionString, AzureClients or Namespace and EventHub name must be set");
+ }
+
+ final String[] parts = remaining.split("/");
+
+ if (parts.length < 2) {
+ throw new IllegalArgumentException("ConnectionString, AzureClients or Namespace and EventHub name must be set");
+ }
+ configuration.setNamespace(parts[0]);
+ configuration.setEventHubName(parts[1]);
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfiguration.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfiguration.java
new file mode 100644
index 0000000..6bc4ac9
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfiguration.java
@@ -0,0 +1,334 @@
+/*
+ * 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.azure.eventhubs;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.azure.core.amqp.AmqpRetryOptions;
+import com.azure.core.amqp.AmqpTransportType;
+import com.azure.messaging.eventhubs.CheckpointStore;
+import com.azure.messaging.eventhubs.EventData;
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import com.azure.messaging.eventhubs.EventProcessorClient;
+import com.azure.messaging.eventhubs.models.EventPosition;
+import com.azure.storage.common.StorageSharedKeyCredential;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.spi.UriPath;
+
+@UriParams
+public class EventHubsConfiguration implements Cloneable {
+
+ @UriPath
+ private String namespace;
+ @UriPath
+ private String eventHubName;
+ @UriParam(label = "security")
+ private String sharedAccessName;
+ @UriParam(label = "security", secret = true)
+ private String sharedAccessKey;
+ @UriParam(label = "security", secret = true)
+ private String connectionString;
+ @UriParam(label = "common", defaultValue = "Amqp")
+ private AmqpTransportType amqpTransportType = AmqpTransportType.AMQP;
+ @UriParam(label = "common")
+ private AmqpRetryOptions amqpRetryOptions;
+ @UriParam(label = "common", defaultValue = "true")
+ private boolean autoDiscoverClient = true;
+ @UriParam(label = "consumer", defaultValue = "$Default")
+ private String consumerGroupName = "$Default";
+ @UriParam(label = "consumer", defaultValue = "500")
+ private int prefetchCount = 500;
+ @UriParam(label = "consumer", defaultValue = "BlobCheckpointStore")
+ private CheckpointStore checkpointStore;
+ @UriParam(label = "consumer")
+ private String blobAccountName;
+ @UriParam(label = "consumer", secret = true)
+ private String blobAccessKey;
+ @UriParam(label = "consumer")
+ private String blobContainerName;
+ @UriParam(label = "consumer", secret = true)
+ private StorageSharedKeyCredential blobStorageSharedKeyCredential;
+ @UriParam(label = "consumer")
+ private Map<String, EventPosition> eventPosition = new HashMap<>();
+ @UriParam(label = "producer")
+ private EventHubProducerAsyncClient producerAsyncClient;
+ @UriParam(label = "producer")
+ private String partitionKey;
+ @UriParam(label = "producer")
+ private String partitionId;
+
+ /**
+ * EventHubs namespace created in Azure Portal
+ */
+ public String getNamespace() {
+ return namespace;
+ }
+
+ public void setNamespace(String namespace) {
+ this.namespace = namespace;
+ }
+
+ /**
+ * EventHubs name under a specific namcespace
+ */
+ public String getEventHubName() {
+ return eventHubName;
+ }
+
+ public void setEventHubName(String eventHubName) {
+ this.eventHubName = eventHubName;
+ }
+
+ /**
+ * The name you chose for your EventHubs SAS keys
+ */
+ public String getSharedAccessName() {
+ return sharedAccessName;
+ }
+
+ public void setSharedAccessName(String sharedAccessName) {
+ this.sharedAccessName = sharedAccessName;
+ }
+
+ /**
+ * The generated value for the SharedAccessName
+ */
+ public String getSharedAccessKey() {
+ return sharedAccessKey;
+ }
+
+ public void setSharedAccessKey(String sharedAccessKey) {
+ this.sharedAccessKey = sharedAccessKey;
+ }
+
+ /**
+ * Instead of supplying namespace, sharedAccessKey, sharedAccessName ... etc, you can just supply the connection string
+ * for your eventHub. The connection string for EventHubs already include all the necessary information to connection to
+ * your EventHub. To learn on how to generate the connection string, take a look at this documentation:
+ * https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string
+ */
+ public String getConnectionString() {
+ return connectionString;
+ }
+
+ public void setConnectionString(String connectionString) {
+ this.connectionString = connectionString;
+ }
+
+ /**
+ * Sets the transport type by which all the communication with Azure Event Hubs occurs. Default value is {@link
+ * AmqpTransportType#AMQP}.
+ */
+ public AmqpTransportType getAmqpTransportType() {
+ return amqpTransportType;
+ }
+
+ public void setAmqpTransportType(AmqpTransportType amqpTransportType) {
+ this.amqpTransportType = amqpTransportType;
+ }
+
+ /**
+ * Sets the retry policy for {@link EventHubAsyncClient}. If not specified, the default retry options are used.
+ */
+ public AmqpRetryOptions getAmqpRetryOptions() {
+ return amqpRetryOptions;
+ }
+
+ public void setAmqpRetryOptions(AmqpRetryOptions amqpRetryOptions) {
+ this.amqpRetryOptions = amqpRetryOptions;
+ }
+
+ /**
+ * Sets the name of the consumer group this consumer is associated with. Events are read in the context of this
+ * group. The name of the consumer group that is created by default is {@link #DEFAULT_CONSUMER_GROUP_NAME
+ * "$Default"}.
+ */
+ public String getConsumerGroupName() {
+ return consumerGroupName;
+ }
+
+ public void setConsumerGroupName(String consumerGroupName) {
+ this.consumerGroupName = consumerGroupName;
+ }
+
+ /**
+ * Sets the count used by the receiver to control the number of events the Event Hub consumer will actively receive
+ * and queue locally without regard to whether a receive operation is currently active.
+ */
+ public int getPrefetchCount() {
+ return prefetchCount;
+ }
+
+ public void setPrefetchCount(int prefetchCount) {
+ this.prefetchCount = prefetchCount;
+ }
+
+ /**
+ * Sets the {@link EventHubProducerAsyncClient}.An asynchronous producer responsible for transmitting {@link EventData} to a specific Event Hub, grouped
+ * together in batches. Depending on the {@link CreateBatchOptions options} specified when creating an {@linkEventDataBatch},
+ * the events may be automatically routed to an available partition or specific to a partition. Use by this component to produce the data in camel
+ * producer.
+ */
+ public EventHubProducerAsyncClient getProducerAsyncClient() {
+ return producerAsyncClient;
+ }
+
+ public void setProducerAsyncClient(EventHubProducerAsyncClient producerAsyncClient) {
+ this.producerAsyncClient = producerAsyncClient;
+ }
+
+ /**
+ * Setting the autoDiscoverClient mechanism, if true, the component will
+ * look for a client instance in the registry automatically otherwise it
+ * will skip that checking.
+ */
+ public boolean isAutoDiscoverClient() {
+ return autoDiscoverClient;
+ }
+
+ public void setAutoDiscoverClient(boolean autoDiscoverClient) {
+ this.autoDiscoverClient = autoDiscoverClient;
+ }
+
+ /**
+ * Sets the identifier of the Event Hub partition that the {@link EventData events} will be sent to. If the
+ * identifier is not specified, the Event Hubs service will be responsible for routing events that are sent to an
+ * available partition.
+ */
+ public String getPartitionId() {
+ return partitionId;
+ }
+
+ public void setPartitionId(String partitionId) {
+ this.partitionId = partitionId;
+ }
+
+ /**
+ * Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key
+ * to a specific partition.
+ *
+ * The selection of a partition is stable for a given partition hashing key. Should any other batches of events
+ * be sent using the same exact partition hashing key, the Event Hubs service will route them all to the same
+ * partition.
+ *
+ * This should be specified only when there is a need to group events by partition, but there is flexibility into
+ * which partition they are routed. If ensuring that a batch of events is sent only to a specific partition, it is
+ * recommended that the {@link #setPartitionId(String) identifier of the position be specified directly} when
+ * sending the batch.
+ */
+ public String getPartitionKey() {
+ return partitionKey;
+ }
+
+ public void setPartitionKey(String partitionKey) {
+ this.partitionKey = partitionKey;
+ }
+
+ /**
+ * Sets the {@link CheckpointStore} the {@link EventProcessorClient} will use for storing partition ownership and
+ * checkpoint information.
+ *
+ * <p>
+ * Users can, optionally, provide their own implementation of {@link CheckpointStore} which will store ownership and
+ * checkpoint information.
+ * </p>
+ *
+ * By default it set to use {@link com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore} which stores
+ * all checkpoint offsets into Azure Blob Storage
+ */
+ public CheckpointStore getCheckpointStore() {
+ return checkpointStore;
+ }
+
+ public void setCheckpointStore(CheckpointStore checkpointStore) {
+ this.checkpointStore = checkpointStore;
+ }
+
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets Azure account name to be used for authentication with azure blob services.
+ */
+ public String getBlobAccountName() {
+ return blobAccountName;
+ }
+
+ public void setBlobAccountName(String blobAccountName) {
+ this.blobAccountName = blobAccountName;
+ }
+
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets access key for the associated azure account name to be used for authentication with azure blob services
+ */
+ public String getBlobAccessKey() {
+ return blobAccessKey;
+ }
+
+ public void setBlobAccessKey(String blobAccessKey) {
+ this.blobAccessKey = blobAccessKey;
+ }
+
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets the blob container that shall be used by the BlobCheckpointStore
+ * to store the checkpoint offsets
+ */
+ public String getBlobContainerName() {
+ return blobContainerName;
+ }
+
+ public void setBlobContainerName(String blobContainerName) {
+ this.blobContainerName = blobContainerName;
+ }
+
+ /**
+ * In case you chose the default BlobCheckpointStore, StorageSharedKeyCredential can be injected to create the azure client, this holds the important authentication information
+ */
+ public StorageSharedKeyCredential getBlobStorageSharedKeyCredential() {
+ return blobStorageSharedKeyCredential;
+ }
+
+ public void setBlobStorageSharedKeyCredential(StorageSharedKeyCredential blobStorageSharedKeyCredential) {
+ this.blobStorageSharedKeyCredential = blobStorageSharedKeyCredential;
+ }
+
+ /**
+ * Sets the map containing the event position to use for each partition if a checkpoint for the partition does not
+ * exist in {@link CheckpointStore}. This map is keyed off of the partition id. If there is no checkpoint in {@link
+ * CheckpointStore} and there is no entry in this map, the processing of the partition will start from {@link
+ * EventPosition#latest() latest} position.
+ */
+ public Map<String, EventPosition> getEventPosition() {
+ return eventPosition;
+ }
+
+ public void setEventPosition(Map<String, EventPosition> eventPosition) {
+ this.eventPosition = eventPosition;
+ }
+
+ // *************************************************
+ //
+ // *************************************************
+
+ public EventHubsConfiguration copy() {
+ try {
+ return (EventHubsConfiguration) super.clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfigurationOptionsProxy.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfigurationOptionsProxy.java
new file mode 100644
index 0000000..a2e0d3d1a
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConfigurationOptionsProxy.java
@@ -0,0 +1,58 @@
+/*
+ * 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.azure.eventhubs;
+
+import java.util.function.Supplier;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * A proxy class for {@link EventHubsConfiguration} and {@link EventHubsConstants}. Ideally this
+ * is responsible to obtain the correct configurations options either from configs or exchange headers
+ */
+public class EventHubsConfigurationOptionsProxy {
+
+ private final EventHubsConfiguration configuration;
+
+ public EventHubsConfigurationOptionsProxy(final EventHubsConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ private static <T> T getObjectFromHeaders(final Exchange exchange, final String headerName, final Class<T> classType) {
+ return exchange.getIn().getHeader(headerName, classType);
+ }
+
+ public String getPartitionKey(final Exchange exchange) {
+ return getOption(exchange, EventHubsConstants.PARTITION_KEY, configuration::getPartitionKey, String.class);
+ }
+
+ public String getPartitionId(final Exchange exchange) {
+ return getOption(exchange, EventHubsConstants.PARTITION_ID, configuration::getPartitionId, String.class);
+ }
+
+ public EventHubsConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ private <R> R getOption(final Exchange exchange, final String headerName, final Supplier<R> fallbackFn, final Class<R> type) {
+ // we first try to look if our value in exchange otherwise fallback to fallbackFn which could be either a function or constant
+ return ObjectHelper.isEmpty(exchange) || ObjectHelper.isEmpty(getObjectFromHeaders(exchange, headerName, type)) ? fallbackFn.get()
+ : getObjectFromHeaders(exchange, headerName, type);
+ }
+
+}
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConstants.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConstants.java
new file mode 100644
index 0000000..f0b24ea
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConstants.java
@@ -0,0 +1,31 @@
+/*
+ * 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.azure.eventhubs;
+
+public final class EventHubsConstants {
+ private static final String HEADER_PREFIX = "CamelAzureEventHubs";
+ // common headers, set by consumer and evaluated by producer
+ public static final String PARTITION_KEY = HEADER_PREFIX + "PartitionKey";
+ public static final String PARTITION_ID = HEADER_PREFIX + "PartitionId";
+ // headers set by the consumer only
+ public static final String OFFSET = HEADER_PREFIX + "Offset";
+ public static final String ENQUEUED_TIME = HEADER_PREFIX + "EnqueuedTime";
+ public static final String SEQUENCE_NUMBER = HEADER_PREFIX + "SequenceNumber";
+
+ private EventHubsConstants() {
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
new file mode 100644
index 0000000..cff9ba7
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
@@ -0,0 +1,129 @@
+/*
+ * 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.azure.eventhubs;
+
+import com.azure.messaging.eventhubs.EventProcessorClient;
+import com.azure.messaging.eventhubs.models.ErrorContext;
+import com.azure.messaging.eventhubs.models.EventContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.support.DefaultConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EventHubsConsumer extends DefaultConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(EventHubsConsumer.class);
+
+ // we use the EventProcessorClient as recommended by Azure docs to consume from all partitions
+ private EventProcessorClient processorClient;
+
+ public EventHubsConsumer(final EventHubsEndpoint endpoint, final Processor processor) {
+ super(endpoint, processor);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ // create the client
+ processorClient = EventHubsClientFactory.createEventProcessorClient(getConfiguration(),
+ this::onEventListener, this::onErrorListener);
+
+ // start the client but we will rely on the Azure Client Scheduler for thread management
+ processorClient.start();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (processorClient != null) {
+ // shutdown the client
+ processorClient.stop();
+ }
+
+ // shutdown camel consumer
+ super.doStop();
+ }
+
+ public EventHubsConfiguration getConfiguration() {
+ return getEndpoint().getConfiguration();
+ }
+
+ @Override
+ public EventHubsEndpoint getEndpoint() {
+ return (EventHubsEndpoint) super.getEndpoint();
+ }
+
+ private void onEventListener(final EventContext eventContext) {
+ final Exchange exchange = getEndpoint().createAzureEventHubExchange(eventContext);
+
+ // add exchange callback
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(new Synchronization() {
+ @Override
+ public void onComplete(Exchange exchange) {
+ // we update the consumer offsets
+ processCommit(exchange, eventContext);
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ // we do nothing here
+ processRollback(exchange);
+ }
+ });
+ // send message to next processor in the route
+ getAsyncProcessor().process(exchange, doneSync -> LOG.trace("Processing exchange [{}] done.", exchange));
+ }
+
+ private void onErrorListener(final ErrorContext errorContext) {
+ final Exchange exchange = getEndpoint().createAzureEventHubExchange(errorContext);
+
+ // log exception if an exception occurred and was not handled
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange", exchange,
+ exchange.getException());
+ }
+ }
+
+ /**
+ * Strategy to commit the offset after message being processed successfully.
+ *
+ * @param exchange the exchange
+ */
+ private void processCommit(final Exchange exchange, final EventContext eventContext) {
+ try {
+ eventContext.updateCheckpoint();
+ } catch (Exception ex) {
+ getExceptionHandler().handleException("Error occurred during updating the checkpoint. This exception is ignored.", exchange, ex);
+ }
+ }
+
+ /**
+ * Strategy when processing the exchange failed.
+ *
+ * @param exchange the exchange
+ */
+ private void processRollback(Exchange exchange) {
+ final Exception cause = exchange.getException();
+ if (cause != null) {
+ getExceptionHandler().handleException("Error during processing exchange.", exchange, cause);
+ }
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpoint.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpoint.java
new file mode 100644
index 0000000..43e6f37
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpoint.java
@@ -0,0 +1,100 @@
+/*
+ * 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.azure.eventhubs;
+
+import com.azure.messaging.eventhubs.models.ErrorContext;
+import com.azure.messaging.eventhubs.models.EventContext;
+import org.apache.camel.Category;
+import org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.support.DefaultEndpoint;
+
+/**
+ * The azure-eventhubs component that integrates Azure Event Hubs using AMQP protocol. Azure EventHubs is a highly scalable publish-subscribe service that
+ * can ingest millions of events per second and stream them to multiple consumers.
+ */
+@UriEndpoint(firstVersion = "3.5.0", scheme = "azure-eventhubs", title = "Azure Event Hubs", syntax = "azure-eventhubs:namespace/eventHubName", category = {
+ Category.CLOUD, Category.MESSAGING })
+public class EventHubsEndpoint extends DefaultEndpoint {
+
+ @UriParam
+ private EventHubsConfiguration configuration;
+
+ public EventHubsEndpoint(final String uri, final Component component, final EventHubsConfiguration configuration) {
+ super(uri, component);
+ this.configuration = configuration;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new EventHubsProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ final Consumer eventHubConsumer = new EventHubsConsumer(this, processor);
+ configureConsumer(eventHubConsumer);
+
+ return eventHubConsumer;
+ }
+
+ /**
+ * The component configurations
+ */
+ public EventHubsConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(EventHubsConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ public Exchange createAzureEventHubExchange(final EventContext eventContext) {
+ final Exchange exchange = createExchange();
+ final Message message = exchange.getIn();
+
+ // set body as byte[] and let camel typeConverters do the job to convert
+ message.setBody(eventContext.getEventData().getBody());
+ // set headers
+ message.setHeader(EventHubsConstants.PARTITION_ID, eventContext.getPartitionContext().getPartitionId());
+ message.setHeader(EventHubsConstants.PARTITION_KEY, eventContext.getEventData().getPartitionKey());
+ message.setHeader(EventHubsConstants.OFFSET, eventContext.getEventData().getOffset());
+ message.setHeader(EventHubsConstants.ENQUEUED_TIME, eventContext.getEventData().getEnqueuedTime());
+ message.setHeader(EventHubsConstants.SEQUENCE_NUMBER, eventContext.getEventData().getSequenceNumber());
+
+ return exchange;
+ }
+
+ public Exchange createAzureEventHubExchange(final ErrorContext errorContext) {
+ final Exchange exchange = createExchange();
+ final Message message = exchange.getIn();
+
+ // set headers
+ message.setHeader(EventHubsConstants.PARTITION_ID, errorContext.getPartitionContext().getPartitionId());
+
+ // set exception
+ exchange.setException(errorContext.getThrowable());
+
+ return exchange;
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsProducer.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsProducer.java
new file mode 100644
index 0000000..1164156
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsProducer.java
@@ -0,0 +1,77 @@
+/*
+ * 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.azure.eventhubs;
+
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import org.apache.camel.component.azure.eventhubs.operations.EventHubsProducerOperations;
+import org.apache.camel.support.DefaultAsyncProducer;
+
+public class EventHubsProducer extends DefaultAsyncProducer {
+
+ private EventHubProducerAsyncClient producerAsyncClient;
+ private EventHubsProducerOperations producerOperations;
+
+ public EventHubsProducer(final Endpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ // create the client
+ producerAsyncClient = EventHubsClientFactory.createEventHubProducerAsyncClient(getEndpoint().getConfiguration());
+
+ // create our operations
+ producerOperations = new EventHubsProducerOperations(producerAsyncClient, getConfiguration());
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ try {
+ return producerOperations.sendEvents(exchange, callback);
+ } catch (Exception e) {
+ exchange.setException(e);
+ callback.done(true);
+ return true;
+ }
+
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (producerAsyncClient != null) {
+ // shutdown async client
+ producerAsyncClient.close();
+ }
+
+ super.doStop();
+ }
+
+ @Override
+ public EventHubsEndpoint getEndpoint() {
+ return (EventHubsEndpoint) super.getEndpoint();
+ }
+
+ public EventHubsConfiguration getConfiguration() {
+ return getEndpoint().getConfiguration();
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/client/EventHubsClientFactory.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/client/EventHubsClientFactory.java
new file mode 100644
index 0000000..5f1295a
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/client/EventHubsClientFactory.java
@@ -0,0 +1,140 @@
+/*
+ * 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.azure.eventhubs.client;
+
+import java.util.Locale;
+import java.util.function.Consumer;
+
+import com.azure.messaging.eventhubs.CheckpointStore;
+import com.azure.messaging.eventhubs.EventHubClientBuilder;
+import com.azure.messaging.eventhubs.EventHubConsumerAsyncClient;
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import com.azure.messaging.eventhubs.EventProcessorClient;
+import com.azure.messaging.eventhubs.EventProcessorClientBuilder;
+import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
+import com.azure.messaging.eventhubs.models.ErrorContext;
+import com.azure.messaging.eventhubs.models.EventContext;
+import com.azure.storage.blob.BlobContainerAsyncClient;
+import com.azure.storage.blob.BlobContainerClientBuilder;
+import com.azure.storage.common.StorageSharedKeyCredential;
+import org.apache.camel.component.azure.eventhubs.EventHubsConfiguration;
+import org.apache.camel.util.ObjectHelper;
+
+public final class EventHubsClientFactory {
+
+ private static final String SERVICE_URI_SEGMENT = "servicebus.windows.net";
+ private static final String BLOB_SERVICE_URI_SEGMENT = ".blob.core.windows.net";
+
+ private EventHubsClientFactory() {
+ }
+
+ public static EventHubProducerAsyncClient createEventHubProducerAsyncClient(final EventHubsConfiguration configuration) {
+ return new EventHubClientBuilder()
+ .connectionString(buildConnectionString(configuration))
+ .transportType(configuration.getAmqpTransportType())
+ .retry(configuration.getAmqpRetryOptions())
+ .buildAsyncProducerClient();
+ }
+
+ public static EventHubConsumerAsyncClient createEventHubConsumerAsyncClient(final EventHubsConfiguration configuration) {
+ return new EventHubClientBuilder()
+ .connectionString(buildConnectionString(configuration))
+ .consumerGroup(configuration.getConsumerGroupName())
+ .prefetchCount(configuration.getPrefetchCount())
+ .transportType(configuration.getAmqpTransportType())
+ .retry(configuration.getAmqpRetryOptions())
+ .buildAsyncConsumerClient();
+ }
+
+ public static EventProcessorClient createEventProcessorClient(final EventHubsConfiguration configuration, final Consumer<EventContext> processEvent,
+ final Consumer<ErrorContext> processError) {
+ return new EventProcessorClientBuilder()
+ .initialPartitionEventPosition(configuration.getEventPosition())
+ .connectionString(buildConnectionString(configuration))
+ .checkpointStore(createCheckpointStore(configuration))
+ .consumerGroup(configuration.getConsumerGroupName())
+ .retry(configuration.getAmqpRetryOptions())
+ .transportType(configuration.getAmqpTransportType())
+ .processError(processError)
+ .processEvent(processEvent)
+ .buildEventProcessorClient();
+
+ }
+
+ // public for testing purposes
+ public static BlobContainerAsyncClient createBlobContainerClient(final EventHubsConfiguration configuration) {
+ return new BlobContainerClientBuilder()
+ .endpoint(buildAzureEndpointUri(configuration))
+ .containerName(configuration.getBlobContainerName())
+ .credential(getCredentialForClient(configuration))
+ .buildAsyncClient();
+ }
+
+ private static CheckpointStore createCheckpointStore(final EventHubsConfiguration configuration) {
+ if (ObjectHelper.isNotEmpty(configuration.getCheckpointStore())) {
+ return configuration.getCheckpointStore();
+ }
+ // so we have no checkpoint store, we fallback to default BlobCheckpointStore
+ // first we check if we have all required params for BlobCheckpointStore
+ if (ObjectHelper.isEmpty(configuration.getBlobContainerName())
+ || !isCredentialsSet(configuration)) {
+ throw new IllegalArgumentException("Since there is no provided CheckpointStore, you will need to set blobAccountName, blobAccessName"
+ + " or blobContainerName in order to use the default BlobCheckpointStore");
+ }
+
+ // second build the BlobContainerAsyncClient
+ return new BlobCheckpointStore(createBlobContainerClient(configuration));
+ }
+
+ private static boolean isCredentialsSet(final EventHubsConfiguration configuration) {
+ if (ObjectHelper.isNotEmpty(configuration.getBlobStorageSharedKeyCredential())) {
+ return true;
+ }
+
+ return ObjectHelper.isNotEmpty(configuration.getBlobAccessKey()) && ObjectHelper.isNotEmpty(configuration.getBlobAccountName());
+ }
+
+ private static String buildConnectionString(final EventHubsConfiguration configuration) {
+ if (ObjectHelper.isNotEmpty(configuration.getConnectionString())) {
+ return configuration.getConnectionString();
+ }
+
+ return String.format(Locale.ROOT, "Endpoint=sb://%s.%s/;SharedAccessKeyName=%s;SharedAccessKey=%s;EntityPath=%s",
+ configuration.getNamespace(), SERVICE_URI_SEGMENT, configuration.getSharedAccessName(), configuration.getSharedAccessKey(),
+ configuration.getEventHubName());
+ }
+
+ private static String buildAzureEndpointUri(final EventHubsConfiguration configuration) {
+ return String.format(Locale.ROOT, "https://%s" + BLOB_SERVICE_URI_SEGMENT, getAccountName(configuration));
+ }
+
+ private static StorageSharedKeyCredential getCredentialForClient(final EventHubsConfiguration configuration) {
+ final StorageSharedKeyCredential storageSharedKeyCredential = configuration.getBlobStorageSharedKeyCredential();
+
+ if (storageSharedKeyCredential != null) {
+ return storageSharedKeyCredential;
+ }
+
+ return new StorageSharedKeyCredential(configuration.getBlobAccountName(), configuration.getBlobAccessKey());
+ }
+
+ private static String getAccountName(final EventHubsConfiguration configuration) {
+ return ObjectHelper.isNotEmpty(configuration.getBlobStorageSharedKeyCredential())
+ ? configuration.getBlobStorageSharedKeyCredential().getAccountName()
+ : configuration.getBlobAccountName();
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperations.java b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperations.java
new file mode 100644
index 0000000..292259d
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperations.java
@@ -0,0 +1,106 @@
+/*
+ * 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.azure.eventhubs.operations;
+
+import java.util.Collections;
+
+import com.azure.messaging.eventhubs.EventData;
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import com.azure.messaging.eventhubs.models.SendOptions;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.azure.eventhubs.EventHubsConfiguration;
+import org.apache.camel.component.azure.eventhubs.EventHubsConfigurationOptionsProxy;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
+public class EventHubsProducerOperations {
+
+ private static final Logger LOG = LoggerFactory.getLogger(EventHubsProducerOperations.class);
+
+ private final EventHubProducerAsyncClient producerAsyncClient;
+ private final EventHubsConfigurationOptionsProxy configurationOptionsProxy;
+
+ public EventHubsProducerOperations(final EventHubProducerAsyncClient producerAsyncClient, final EventHubsConfiguration configuration) {
+ ObjectHelper.notNull(producerAsyncClient, "client cannot be null");
+
+ this.producerAsyncClient = producerAsyncClient;
+ configurationOptionsProxy = new EventHubsConfigurationOptionsProxy(configuration);
+ }
+
+ public boolean sendEvents(final Exchange exchange, final AsyncCallback callback) {
+ ObjectHelper.notNull(exchange, "exchange cannot be null");
+ ObjectHelper.notNull(callback, "callback cannot be null");
+
+ final SendOptions sendOptions = createSendOptions(configurationOptionsProxy.getPartitionKey(exchange), configurationOptionsProxy.getPartitionId(exchange));
+ final Iterable<EventData> eventData = createEventData(exchange);
+
+ return sendAsyncEvents(eventData, sendOptions, exchange, callback);
+ }
+
+ private boolean sendAsyncEvents(final Iterable<EventData> eventData, final SendOptions sendOptions, final Exchange exchange, final AsyncCallback asyncCallback) {
+ sendAsyncEventsWithSuitableMethod(eventData, sendOptions)
+ .subscribe(unused -> LOG.debug("Processed one event..."), error -> {
+ // error but we continue
+ LOG.debug("Error processing async exchange with error:" + error.getMessage());
+ exchange.setException(error);
+ asyncCallback.done(false);
+ }, () -> {
+ // we are done from everything, so mark it as sync done
+ LOG.debug("All events with exchange have been sent successfully.");
+ asyncCallback.done(false);
+ });
+
+ return false;
+ }
+
+ private Mono<Void> sendAsyncEventsWithSuitableMethod(final Iterable<EventData> eventData, final SendOptions sendOptions) {
+ if (ObjectHelper.isEmpty(sendOptions)) {
+ return producerAsyncClient.send(eventData);
+ }
+
+ return producerAsyncClient.send(eventData, sendOptions);
+ }
+
+ private SendOptions createSendOptions(final String partitionKey, final String partitionId) {
+ // if both are set, we don't want that
+ if (ObjectHelper.isNotEmpty(partitionKey) && ObjectHelper.isNotEmpty(partitionId)) {
+ throw new IllegalArgumentException("Both partitionKey and partitionId are set. Only one or the other can be set.");
+ }
+ // if both are not set, we return null and let EventHubs handle the partition assigning
+ if (ObjectHelper.isEmpty(partitionKey) && ObjectHelper.isEmpty(partitionId)) {
+ return null;
+ }
+
+ return new SendOptions()
+ .setPartitionId(partitionId)
+ .setPartitionKey(partitionKey);
+ }
+
+ private Iterable<EventData> createEventData(final Exchange exchange) {
+ final byte[] data = exchange.getIn().getBody(byte[].class);
+
+ if (ObjectHelper.isEmpty(data)) {
+ throw new IllegalArgumentException(String.format("Cannot convert message body %s to byte[]. You will need "
+ + "to make sure the data encoded in byte[] or add a Camel TypeConverter to convert the data to byte[]", exchange.getIn().getBody()));
+ }
+ // for now we only support single event
+ return Collections.singletonList(new EventData(data));
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentTest.java b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentTest.java
new file mode 100644
index 0000000..7a9537e
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsComponentTest.java
@@ -0,0 +1,123 @@
+/*
+ * 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.azure.eventhubs;
+
+import com.azure.messaging.eventhubs.EventHubConsumerAsyncClient;
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import org.apache.camel.ResolveEndpointFailedException;
+import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class EventHubsComponentTest extends CamelTestSupport {
+
+ @Test
+ public void testCreateEndpointWithNoEventHubsNameOrNameSpace() throws Exception {
+ ResolveEndpointFailedException exception = assertThrows(ResolveEndpointFailedException.class, () ->
+ context.getEndpoint("azure-eventhubs:?sharedAccessKey=string&sharedAccessName=name"));
+
+ assertTrue(exception.getMessage().contains("ConnectionString, AzureClients or Namespace and EventHub name must be set"));
+
+ ResolveEndpointFailedException exception2 = assertThrows(ResolveEndpointFailedException.class, () ->
+ context.getEndpoint("azure-eventhubs:name?sharedAccessKey=string&sharedAccessName=name"));
+
+ assertTrue(exception2.getMessage().contains("ConnectionString, AzureClients or Namespace and EventHub name must be set"));
+ }
+
+ @Test
+ public void testCreateEndpointWithNoSuppliedClientsOrKeysOrConnectionString() {
+ final String expectedErrorMessage = "Azure EventHubs SharedAccessName/SharedAccessKey, ConsumerAsyncClient/ProducerAsyncClient "
+ + "or connectionString must be specified";
+
+ // first case: with no client or key or connectionstring
+ assertTrue(getErrorMessage("azure-eventhubs:name/hubName?").contains(expectedErrorMessage));
+
+ // second case: connectionString set
+ assertNotNull(context.getEndpoint("azure-eventhubs:?connectionString=string"));
+
+ // third case: either access key or access name set
+ assertTrue(getErrorMessage("azure-eventhubs:name/hubName?sharedAccessName=test").contains(expectedErrorMessage));
+ assertTrue(getErrorMessage("azure-eventhubs:name/hubName?sharedAccessKey=test").contains(expectedErrorMessage));
+ assertNotNull(context.getEndpoint("azure-eventhubs:name/hubName?sharedAccessName=test&sharedAccessKey=test"));
+
+ // forth case: with client set
+ final EventHubsConfiguration configuration = new EventHubsConfiguration();
+ configuration.setNamespace("test");
+ configuration.setConsumerGroupName("testGroup");
+ configuration.setSharedAccessKey("dummyKey");
+ configuration.setSharedAccessName("dummyUser");
+
+ final EventHubProducerAsyncClient producerAsyncClient = EventHubsClientFactory.createEventHubProducerAsyncClient(configuration);
+
+ context.getRegistry().bind("producerClient", producerAsyncClient);
+
+ assertNotNull(context.getEndpoint("azure-eventhubs:name/hubName?autoDiscoverClient=false&producerAsyncClient=#producerClient"));
+ }
+
+ @Test
+ public void testClientAutoDiscovery() {
+ final EventHubsConfiguration configuration = new EventHubsConfiguration();
+ configuration.setNamespace("test");
+ configuration.setConsumerGroupName("testGroup");
+ configuration.setSharedAccessKey("dummyKey");
+ configuration.setSharedAccessName("dummyUser");
+
+ final EventHubConsumerAsyncClient consumerAsyncClient = EventHubsClientFactory.createEventHubConsumerAsyncClient(configuration);
+ final EventHubConsumerAsyncClient consumerAsyncClient2 = EventHubsClientFactory.createEventHubConsumerAsyncClient(configuration);
+ final EventHubProducerAsyncClient producerAsyncClient = EventHubsClientFactory.createEventHubProducerAsyncClient(configuration);
+
+ // we dont allow more than one instance
+ context.getRegistry().bind("consumerClient", consumerAsyncClient);
+ context.getRegistry().bind("consumerClient2", consumerAsyncClient2);
+
+ assertThrows(ResolveEndpointFailedException.class, () -> context.getEndpoint("azure-eventhubs:name/hubName"));
+
+ context.getRegistry().bind("producerClient", producerAsyncClient);
+
+ final EventHubsEndpoint endpoint = (EventHubsEndpoint) context.getEndpoint("azure-eventhubs:name/hubName");
+
+ assertEquals(producerAsyncClient, endpoint.getConfiguration().getProducerAsyncClient());
+ }
+
+ @Test
+ public void testCreateEndpointWithConfig() {
+ final String uri = "azure-eventhubs:namespace/hubName?sharedAccessName=DummyAccessKeyName"
+ + "&sharedAccessKey=DummyKey"
+ + "&consumerGroupName=testConsumer&prefetchCount=100";
+
+ final EventHubsEndpoint endpoint = (EventHubsEndpoint) context.getEndpoint(uri);
+
+ assertEquals("namespace", endpoint.getConfiguration().getNamespace());
+ assertEquals("hubName", endpoint.getConfiguration().getEventHubName());
+ assertEquals("testConsumer", endpoint.getConfiguration().getConsumerGroupName());
+ assertEquals("DummyAccessKeyName", endpoint.getConfiguration().getSharedAccessName());
+ assertEquals("DummyKey", endpoint.getConfiguration().getSharedAccessKey());
+ assertEquals(100, endpoint.getConfiguration().getPrefetchCount());
+ assertTrue(endpoint.getConfiguration().isAutoDiscoverClient());
+ }
+
+ private String getErrorMessage(final String uri) {
+ ResolveEndpointFailedException exception = assertThrows(ResolveEndpointFailedException.class, () -> context.getEndpoint(uri));
+ return exception.getMessage();
+ }
+
+}
\ No newline at end of file
diff --git a/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerIT.java b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerIT.java
new file mode 100644
index 0000000..c98dc15
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerIT.java
@@ -0,0 +1,142 @@
+/*
+ * 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.azure.eventhubs;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+
+import com.azure.messaging.eventhubs.EventData;
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import com.azure.messaging.eventhubs.models.EventPosition;
+import com.azure.messaging.eventhubs.models.SendOptions;
+import com.azure.storage.blob.BlobContainerAsyncClient;
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class EventHubsConsumerIT extends CamelTestSupport {
+
+ @EndpointInject("mock:result")
+ private MockEndpoint result;
+
+ private String containerName;
+ private BlobContainerAsyncClient containerAsyncClient;
+ private EventHubsConfiguration configuration;
+
+ @BeforeAll
+ public void prepare() throws Exception {
+ containerName = RandomStringUtils.randomAlphabetic(5).toLowerCase();
+
+ final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
+
+ configuration = new EventHubsConfiguration();
+ configuration.setBlobAccessKey(properties.getProperty(TestUtils.BLOB_ACCESS_KEY));
+ configuration.setBlobAccountName(properties.getProperty(TestUtils.BLOB_ACCOUNT_NAME));
+ configuration.setBlobContainerName(containerName);
+ configuration.setConnectionString(properties.getProperty(TestUtils.CONNECTION_STRING));
+
+ containerAsyncClient = EventHubsClientFactory.createBlobContainerClient(configuration);
+
+ // create test container
+ containerAsyncClient.create().block();
+ }
+
+ @Test
+ public void testConsumerEvents() throws InterruptedException {
+ // send test data
+ final EventHubProducerAsyncClient producerAsyncClient = EventHubsClientFactory.createEventHubProducerAsyncClient(configuration);
+
+ final String messageBody = RandomStringUtils.randomAlphabetic(30);
+ final String messageKey = RandomStringUtils.randomAlphabetic(5);
+
+ producerAsyncClient.send(Collections.singletonList(new EventData(messageBody)), new SendOptions().setPartitionKey(messageKey)).block();
+
+ result.expectedMinimumMessageCount(1);
+ result.setAssertPeriod(20000);
+
+ final List<Exchange> exchanges = result.getExchanges();
+ result.assertIsSatisfied();
+
+ // now we check our messages
+ final Exchange returnedMessage = exchanges.stream()
+ .filter(Objects::nonNull)
+ .filter(exchange -> exchange.getMessage().getHeader(EventHubsConstants.PARTITION_KEY)
+ != null && exchange.getMessage().getHeader(EventHubsConstants.PARTITION_KEY).equals(messageKey))
+ .findFirst()
+ .orElse(null);
+
+ assertNotNull(returnedMessage);
+
+ assertEquals(messageKey, returnedMessage.getMessage().getHeader(EventHubsConstants.PARTITION_KEY));
+
+ assertNotNull(returnedMessage.getMessage().getBody());
+ assertNotNull(returnedMessage.getMessage().getHeader(EventHubsConstants.PARTITION_ID));
+ assertNotNull(returnedMessage.getMessage().getHeader(EventHubsConstants.SEQUENCE_NUMBER));
+ assertNotNull(returnedMessage.getMessage().getHeader(EventHubsConstants.OFFSET));
+ assertNotNull(returnedMessage.getMessage().getHeader(EventHubsConstants.ENQUEUED_TIME));
+ }
+
+ @AfterAll
+ public void tearDown() {
+ // delete testing container
+ containerAsyncClient.delete().block();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("azure-eventhubs:?"
+ + "connectionString=RAW({{connectionString}})"
+ + "&blobContainerName=" + containerName + "&eventPosition=#eventPosition"
+ + "&blobAccountName={{blobAccountName}}&blobAccessKey=RAW({{blobAccessKey}})")
+ .to(result);
+
+ }
+ };
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ final Map<String, EventPosition> positionMap = new HashMap<>();
+ positionMap.put("0", EventPosition.earliest());
+ positionMap.put("1", EventPosition.earliest());
+
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("eventPosition", positionMap);
+
+ return context;
+ }
+}
\ No newline at end of file
diff --git a/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsProducerIT.java b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsProducerIT.java
new file mode 100644
index 0000000..e1f22ac
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsProducerIT.java
@@ -0,0 +1,138 @@
+/*
+ * 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.azure.eventhubs;
+
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.azure.messaging.eventhubs.EventHubClientBuilder;
+import com.azure.messaging.eventhubs.EventHubConsumerAsyncClient;
+import com.azure.messaging.eventhubs.models.EventPosition;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class EventHubsProducerIT extends CamelTestSupport {
+
+ @EndpointInject
+ private ProducerTemplate template;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint result;
+
+ private EventHubConsumerAsyncClient consumerAsyncClient;
+
+ @BeforeAll
+ public void prepare() throws Exception {
+ final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
+
+ final EventHubsConfiguration configuration = new EventHubsConfiguration();
+ configuration.setConnectionString(properties.getProperty(TestUtils.CONNECTION_STRING));
+ configuration.setConsumerGroupName(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME);
+
+ consumerAsyncClient = EventHubsClientFactory.createEventHubConsumerAsyncClient(configuration);
+ }
+
+ @Test
+ public void testSendEventWithSpecificPartition() throws InterruptedException {
+
+ final String messageBody = RandomStringUtils.randomAlphabetic(30);
+ final String firstPartition = "0";
+
+ final AtomicBoolean eventExists = new AtomicBoolean(false);
+
+ final CompletableFuture<Exchange> resultAsync = template.asyncSend("direct:sendAsync", exchange -> {
+ exchange.getIn().setHeader(EventHubsConstants.PARTITION_ID, firstPartition);
+ exchange.getIn().setBody(messageBody);
+ });
+
+ resultAsync.whenComplete((exchange, throwable) -> {
+ // we sent our exchange, let's check it out
+ final Boolean eventFlag = consumerAsyncClient.receiveFromPartition(firstPartition, EventPosition.earliest())
+ .any(partitionEvent -> partitionEvent.getPartitionContext().getPartitionId().equals(firstPartition) && partitionEvent.getData().getBodyAsString()
+ .contains(messageBody))
+ .block();
+
+ if (eventFlag == null) {
+ eventExists.set(false);
+ }
+
+ eventExists.set(eventFlag);
+ });
+
+ result.expectedMinimumMessageCount(1);
+ result.setAssertPeriod(20000);
+ result.assertIsSatisfied();
+
+ assertTrue(eventExists.get());
+ }
+
+ @Test
+ public void testSendingNonValidData() throws InterruptedException {
+
+ final String messageBody = RandomStringUtils.randomAlphabetic(30);
+ final String firstPartition = "0";
+
+ final AtomicReference<Exchange> resultExchange = new AtomicReference<>();
+
+ final CompletableFuture<Exchange> resultAsync = template.asyncSend("direct:sendAsync", exchange -> {
+ exchange.getIn().setHeader(EventHubsConstants.PARTITION_ID, firstPartition);
+ exchange.getIn().setHeader(EventHubsConstants.PARTITION_KEY, "testKey");
+ exchange.getIn().setBody(messageBody);
+ });
+
+ resultAsync.whenComplete((exchange, throwable) -> resultExchange.set(exchange));
+
+ result.setAssertPeriod(100);
+ result.assertIsSatisfied();
+
+ assertNotNull(resultExchange.get());
+ assertNotNull(resultExchange.get().getException());
+ }
+
+ @AfterAll
+ public void tearDown() {
+ consumerAsyncClient.close();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:sendAsync")
+ .to("azure-eventhubs:?connectionString=RAW({{connectionString}})")
+ .to(result);
+ }
+ };
+ }
+}
\ No newline at end of file
diff --git a/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorIT.java b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorIT.java
new file mode 100644
index 0000000..ce5c8e5
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorIT.java
@@ -0,0 +1,106 @@
+/*
+ * 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.azure.eventhubs;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+import com.azure.messaging.eventhubs.EventData;
+import com.azure.messaging.eventhubs.EventHubClientBuilder;
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import com.azure.messaging.eventhubs.EventProcessorClient;
+import com.azure.messaging.eventhubs.models.ErrorContext;
+import com.azure.messaging.eventhubs.models.EventContext;
+import com.azure.messaging.eventhubs.models.EventPosition;
+import com.azure.messaging.eventhubs.models.SendOptions;
+import com.azure.storage.blob.BlobContainerAsyncClient;
+import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class EventProcessorIT {
+
+ private EventHubsConfiguration configuration;
+ private BlobContainerAsyncClient containerAsyncClient;
+
+ @BeforeAll
+ public void prepare() throws Exception {
+ final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
+ final String containerName = RandomStringUtils.randomAlphabetic(5).toLowerCase();
+
+ configuration = new EventHubsConfiguration();
+ configuration.setConnectionString(properties.getProperty("connectionString"));
+ configuration.setConsumerGroupName(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME);
+ configuration.setBlobAccessKey(properties.getProperty("blobAccessKey"));
+ configuration.setBlobAccountName(properties.getProperty("blobAccountName"));
+
+ Map<String, EventPosition> positionMap = new HashMap<>();
+ positionMap.put("0", EventPosition.earliest());
+
+ configuration.setEventPosition(positionMap);
+ configuration.setBlobContainerName(containerName);
+
+ containerAsyncClient = EventHubsClientFactory.createBlobContainerClient(configuration);
+
+ // create test container
+ containerAsyncClient.create().block();
+ }
+
+ @Test
+ public void testEventProcessingWithBlobCheckpointStore() {
+ final AtomicBoolean doneAsync = new AtomicBoolean(false);
+ final EventHubProducerAsyncClient producerAsyncClient = EventHubsClientFactory.createEventHubProducerAsyncClient(configuration);
+ final Consumer<EventContext> onEvent = eventContext -> {
+ final String body = eventContext.getEventData().getBodyAsString();
+ if (eventContext.getPartitionContext().getPartitionId().equals("0") && body.contains("Testing Event Consumer With BlobStore")) {
+ assertTrue(true);
+ doneAsync.set(true);
+ }
+ };
+ final Consumer<ErrorContext> onError = errorContext -> {
+ };
+ final EventProcessorClient processorClient = EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError);
+
+ processorClient.start();
+
+ producerAsyncClient.send(Collections.singletonList(new EventData("Testing Event Consumer With BlobStore")), new SendOptions().setPartitionId("0")).block();
+
+ Awaitility.await()
+ .timeout(30, TimeUnit.SECONDS)
+ .untilTrue(doneAsync);
+
+ processorClient.stop();
+ producerAsyncClient.close();
+ }
+
+ @AfterAll
+ public void tearDown() {
+ containerAsyncClient.delete().block();
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorTest.java b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorTest.java
new file mode 100644
index 0000000..780cbe8
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventProcessorTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.azure.eventhubs;
+
+import java.util.function.Consumer;
+
+import com.azure.messaging.eventhubs.models.ErrorContext;
+import com.azure.messaging.eventhubs.models.EventContext;
+import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class EventProcessorTest {
+
+ @Test
+ public void testCreateEventProcessorWithNonValidOptions() {
+ final EventHubsConfiguration configuration = new EventHubsConfiguration();
+ final Consumer<EventContext> onEvent = event -> {
+ };
+ final Consumer<ErrorContext> onError = error -> {
+ };
+
+ assertThrows(IllegalArgumentException.class, () -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+
+ configuration.setBlobContainerName("testContainer");
+ assertThrows(IllegalArgumentException.class, () -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+
+ configuration.setBlobAccountName("testAcc");
+ assertThrows(IllegalArgumentException.class, () -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+
+ configuration.setBlobAccessKey("testAccess");
+ assertNotNull(EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+
+ configuration.setBlobContainerName(null);
+ assertThrows(IllegalArgumentException.class, () -> EventHubsClientFactory.createEventProcessorClient(configuration, onEvent, onError));
+ }
+}
diff --git a/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/TestUtils.java b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/TestUtils.java
new file mode 100644
index 0000000..d8f49a8
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/TestUtils.java
@@ -0,0 +1,59 @@
+/*
+ * 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.azure.eventhubs;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+import java.util.Properties;
+
+public final class TestUtils {
+
+ public static final String CONNECTION_STRING = "connectionString";
+ public static final String BLOB_ACCOUNT_NAME = "blobAccountName";
+ public static final String BLOB_ACCESS_KEY = "blobAccessKey";
+
+ private TestUtils() {
+ }
+
+ public static Properties loadAzurePropertiesFile() throws IOException {
+ final Properties properties = new Properties();
+ final String fileName = "azure_key.properties";
+
+ final InputStream inputStream = Objects.requireNonNull(TestUtils.class.getClassLoader().getResourceAsStream(fileName));
+
+ properties.load(inputStream);
+
+ return properties;
+ }
+
+ public static Properties loadAzureAccessFromJvmEnv() throws Exception {
+ final Properties properties = new Properties();
+ if (System.getProperty(CONNECTION_STRING) == null
+ || System.getProperty(BLOB_ACCOUNT_NAME) == null
+ || System.getProperty(BLOB_ACCESS_KEY) == null) {
+ throw new Exception("Make sure to supply azure eventHubs connectionString, e.g: mvn verify -PfullTests -DconnectionString=string"
+ + " -DblobAccountName=blob -DblobAccessKey=key");
+ }
+ properties.setProperty(CONNECTION_STRING, System.getProperty(CONNECTION_STRING));
+ properties.setProperty(BLOB_ACCOUNT_NAME, System.getProperty(BLOB_ACCOUNT_NAME));
+ properties.setProperty(BLOB_ACCESS_KEY, System.getProperty(BLOB_ACCESS_KEY));
+
+ return properties;
+ }
+
+}
diff --git a/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperationsIT.java b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperationsIT.java
new file mode 100644
index 0000000..be5d16e
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/operations/EventHubsProducerOperationsIT.java
@@ -0,0 +1,86 @@
+/*
+ * 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.azure.eventhubs.operations;
+
+import java.time.Duration;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import com.azure.messaging.eventhubs.EventHubClientBuilder;
+import com.azure.messaging.eventhubs.EventHubConsumerAsyncClient;
+import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
+import com.azure.messaging.eventhubs.models.EventPosition;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.azure.eventhubs.EventHubsConfiguration;
+import org.apache.camel.component.azure.eventhubs.EventHubsConstants;
+import org.apache.camel.component.azure.eventhubs.TestUtils;
+import org.apache.camel.component.azure.eventhubs.client.EventHubsClientFactory;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class EventHubsProducerOperationsIT extends CamelTestSupport {
+
+ private EventHubsConfiguration configuration;
+
+ @BeforeAll
+ public void prepare() throws Exception {
+ final Properties properties = TestUtils.loadAzureAccessFromJvmEnv();
+
+ configuration = new EventHubsConfiguration();
+ configuration.setConnectionString(properties.getProperty("connectionString"));
+ configuration.setConsumerGroupName(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME);
+ }
+
+ @Test
+ public void testSendEventWithSpecificPartition() {
+ final EventHubProducerAsyncClient producerAsyncClient = EventHubsClientFactory.createEventHubProducerAsyncClient(configuration);
+ final EventHubsProducerOperations operations = new EventHubsProducerOperations(producerAsyncClient, configuration);
+ final EventHubConsumerAsyncClient consumerAsyncClient = EventHubsClientFactory.createEventHubConsumerAsyncClient(configuration);
+ final String firstPartition = producerAsyncClient.getPartitionIds().blockLast();
+ final Exchange exchange = new DefaultExchange(context);
+
+ exchange.getIn().setHeader(EventHubsConstants.PARTITION_ID, firstPartition);
+ exchange.getIn().setBody("test should be in firstPartition");
+
+ operations.sendEvents(exchange, doneSync -> { });
+
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollDelay(Duration.ofSeconds(2))
+ .pollInterval(Duration.ofSeconds(2))
+ .until(() -> {
+ final Boolean eventExists = consumerAsyncClient.receiveFromPartition(firstPartition, EventPosition.earliest())
+ .any(partitionEvent -> partitionEvent.getPartitionContext().getPartitionId().equals(firstPartition) && partitionEvent.getData().getBodyAsString()
+ .contains("test should be in firstPartition"))
+ .block();
+
+ if (eventExists == null) {
+ return false;
+ }
+
+ return eventExists;
+ });
+
+ producerAsyncClient.close();
+ consumerAsyncClient.close();
+ }
+}
\ No newline at end of file
diff --git a/components/camel-azure-eventhubs/src/test/resources/log4j2.properties b/components/camel-azure-eventhubs/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..5ec8843
--- /dev/null
+++ b/components/camel-azure-eventhubs/src/test/resources/log4j2.properties
@@ -0,0 +1,27 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-eventhubs-test.log
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+appender.out.type = Console
+appender.out.name = out
+appender.out.layout.type = PatternLayout
+appender.out.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
\ No newline at end of file
diff --git a/components/pom.xml b/components/pom.xml
index 3ea9642..d9d7e9d 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -145,6 +145,7 @@
<module>camel-azure</module>
<module>camel-azure-storage-blob</module>
<module>camel-azure-storage-queue</module>
+ <module>camel-azure-eventhubs</module>
<module>camel-barcode</module>
<module>camel-base64</module>
<module>camel-bean-validator</module>
diff --git a/core/camel-allcomponents/pom.xml b/core/camel-allcomponents/pom.xml
index cd9fd04..dc5721c 100644
--- a/core/camel-allcomponents/pom.xml
+++ b/core/camel-allcomponents/pom.xml
@@ -244,6 +244,10 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-azure-eventhubs</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-azure-storage-blob</artifactId>
</dependency>
<dependency>
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
index 193d343..b0195da 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
@@ -696,6 +696,20 @@ public interface ComponentsBuilderFactory {
return org.apache.camel.builder.component.dsl.AzureBlobComponentBuilderFactory.azureBlob();
}
/**
+ * Azure Event Hubs (camel-azure-eventhubs)
+ * The azure-eventhubs component that integrates Azure Event Hubs using AMQP
+ * protocol. Azure EventHubs is a highly scalable publish-subscribe service
+ * that can ingest millions of events per second and stream them to multiple
+ * consumers.
+ *
+ * Category: cloud,messaging
+ * Since: 3.5
+ * Maven coordinates: org.apache.camel:camel-azure-eventhubs
+ */
+ static org.apache.camel.builder.component.dsl.AzureEventhubsComponentBuilderFactory.AzureEventhubsComponentBuilder azureEventhubs() {
+ return org.apache.camel.builder.component.dsl.AzureEventhubsComponentBuilderFactory.azureEventhubs();
+ }
+ /**
* Azure Storage Queue Service (Deprecated) (camel-azure)
* Store and retrieve messages from Azure Storage Queue Service.
*
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureEventhubsComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureEventhubsComponentBuilderFactory.java
new file mode 100644
index 0000000..2ca2d51
--- /dev/null
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureEventhubsComponentBuilderFactory.java
@@ -0,0 +1,438 @@
+/*
+ * 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.builder.component.dsl;
+
+import javax.annotation.Generated;
+import org.apache.camel.Component;
+import org.apache.camel.builder.component.AbstractComponentBuilder;
+import org.apache.camel.builder.component.ComponentBuilder;
+import org.apache.camel.component.azure.eventhubs.EventHubsComponent;
+
+/**
+ * The azure-eventhubs component that integrates Azure Event Hubs using AMQP
+ * protocol. Azure EventHubs is a highly scalable publish-subscribe service that
+ * can ingest millions of events per second and stream them to multiple
+ * consumers.
+ *
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.ComponentDslMojo")
+public interface AzureEventhubsComponentBuilderFactory {
+
+ /**
+ * Azure Event Hubs (camel-azure-eventhubs)
+ * The azure-eventhubs component that integrates Azure Event Hubs using AMQP
+ * protocol. Azure EventHubs is a highly scalable publish-subscribe service
+ * that can ingest millions of events per second and stream them to multiple
+ * consumers.
+ *
+ * Category: cloud,messaging
+ * Since: 3.5
+ * Maven coordinates: org.apache.camel:camel-azure-eventhubs
+ */
+ static AzureEventhubsComponentBuilder azureEventhubs() {
+ return new AzureEventhubsComponentBuilderImpl();
+ }
+
+ /**
+ * Builder for the Azure Event Hubs component.
+ */
+ interface AzureEventhubsComponentBuilder
+ extends
+ ComponentBuilder<EventHubsComponent> {
+ /**
+ * Sets the retry policy for EventHubAsyncClient. If not specified, the
+ * default retry options are used.
+ *
+ * The option is a: <code>com.azure.core.amqp.AmqpRetryOptions</code>
+ * type.
+ *
+ * Group: common
+ */
+ default AzureEventhubsComponentBuilder amqpRetryOptions(
+ com.azure.core.amqp.AmqpRetryOptions amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Event Hubs occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option is a: <code>com.azure.core.amqp.AmqpTransportType</code>
+ * type.
+ *
+ * Default: Amqp
+ * Group: common
+ */
+ default AzureEventhubsComponentBuilder amqpTransportType(
+ com.azure.core.amqp.AmqpTransportType amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Setting the autoDiscoverClient mechanism, if true, the component will
+ * look for a client instance in the registry automatically otherwise it
+ * will skip that checking.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: common
+ */
+ default AzureEventhubsComponentBuilder autoDiscoverClient(
+ boolean autoDiscoverClient) {
+ doSetProperty("autoDiscoverClient", autoDiscoverClient);
+ return this;
+ }
+ /**
+ * The component configurations.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.eventhubs.EventHubsConfiguration</code> type.
+ *
+ * Group: common
+ */
+ default AzureEventhubsComponentBuilder configuration(
+ org.apache.camel.component.azure.eventhubs.EventHubsConfiguration configuration) {
+ doSetProperty("configuration", configuration);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets access
+ * key for the associated azure account name to be used for
+ * authentication with azure blob services.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder blobAccessKey(
+ java.lang.String blobAccessKey) {
+ doSetProperty("blobAccessKey", blobAccessKey);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets Azure
+ * account name to be used for authentication with azure blob services.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder blobAccountName(
+ java.lang.String blobAccountName) {
+ doSetProperty("blobAccountName", blobAccountName);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets the blob
+ * container that shall be used by the BlobCheckpointStore to store the
+ * checkpoint offsets.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder blobContainerName(
+ java.lang.String blobContainerName) {
+ doSetProperty("blobContainerName", blobContainerName);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore,
+ * StorageSharedKeyCredential can be injected to create the azure
+ * client, this holds the important authentication information.
+ *
+ * The option is a:
+ * <code>com.azure.storage.common.StorageSharedKeyCredential</code>
+ * type.
+ *
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder blobStorageSharedKeyCredential(
+ com.azure.storage.common.StorageSharedKeyCredential blobStorageSharedKeyCredential) {
+ doSetProperty("blobStorageSharedKeyCredential", blobStorageSharedKeyCredential);
+ return this;
+ }
+ /**
+ * 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 the routing Error Handler. By default the
+ * consumer will use the org.apache.camel.spi.ExceptionHandler to deal
+ * with exceptions, that will be logged at WARN or ERROR level and
+ * ignored.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder bridgeErrorHandler(
+ boolean bridgeErrorHandler) {
+ doSetProperty("bridgeErrorHandler", bridgeErrorHandler);
+ return this;
+ }
+ /**
+ * Sets the CheckpointStore the EventProcessorClient will use for
+ * storing partition ownership and checkpoint information. Users can,
+ * optionally, provide their own implementation of CheckpointStore which
+ * will store ownership and checkpoint information. By default it set to
+ * use
+ * com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore which stores all checkpoint offsets into Azure Blob Storage.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.eventhubs.CheckpointStore</code> type.
+ *
+ * Default: BlobCheckpointStore
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder checkpointStore(
+ com.azure.messaging.eventhubs.CheckpointStore checkpointStore) {
+ doSetProperty("checkpointStore", checkpointStore);
+ return this;
+ }
+ /**
+ * Sets the name of the consumer group this consumer is associated with.
+ * Events are read in the context of this group. The name of the
+ * consumer group that is created by default is {link
+ * #DEFAULT_CONSUMER_GROUP_NAME $Default}.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: $Default
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder consumerGroupName(
+ java.lang.String consumerGroupName) {
+ doSetProperty("consumerGroupName", consumerGroupName);
+ return this;
+ }
+ /**
+ * Sets the map containing the event position to use for each partition
+ * if a checkpoint for the partition does not exist in CheckpointStore.
+ * This map is keyed off of the partition id. If there is no checkpoint
+ * in CheckpointStore and there is no entry in this map, the processing
+ * of the partition will start from {link EventPosition#latest() latest}
+ * position.
+ *
+ * The option is a: <code>java.util.Map<java.lang.String,
+ * com.azure.messaging.eventhubs.models.EventPosition></code> type.
+ *
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder eventPosition(
+ java.util.Map<java.lang.String, com.azure.messaging.eventhubs.models.EventPosition> eventPosition) {
+ doSetProperty("eventPosition", eventPosition);
+ return this;
+ }
+ /**
+ * Sets the count used by the receiver to control the number of events
+ * the Event Hub consumer will actively receive and queue locally
+ * without regard to whether a receive operation is currently active.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 500
+ * Group: consumer
+ */
+ default AzureEventhubsComponentBuilder prefetchCount(int prefetchCount) {
+ doSetProperty("prefetchCount", prefetchCount);
+ return this;
+ }
+ /**
+ * 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 starting the
+ * producer may take a little time and prolong the total processing time
+ * of the processing.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer
+ */
+ default AzureEventhubsComponentBuilder lazyStartProducer(
+ boolean lazyStartProducer) {
+ doSetProperty("lazyStartProducer", lazyStartProducer);
+ return this;
+ }
+ /**
+ * Sets the identifier of the Event Hub partition that the {link
+ * EventData events} will be sent to. If the identifier is not
+ * specified, the Event Hubs service will be responsible for routing
+ * events that are sent to an available partition.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: producer
+ */
+ default AzureEventhubsComponentBuilder partitionId(
+ java.lang.String partitionId) {
+ doSetProperty("partitionId", partitionId);
+ return this;
+ }
+ /**
+ * Sets a hashing key to be provided for the batch of events, which
+ * instructs the Event Hubs service to map this key to a specific
+ * partition. The selection of a partition is stable for a given
+ * partition hashing key. Should any other batches of events be sent
+ * using the same exact partition hashing key, the Event Hubs service
+ * will route them all to the same partition. This should be specified
+ * only when there is a need to group events by partition, but there is
+ * flexibility into which partition they are routed. If ensuring that a
+ * batch of events is sent only to a specific partition, it is
+ * recommended that the {link #setPartitionId(String) identifier of the
+ * position be specified directly} when sending the batch.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: producer
+ */
+ default AzureEventhubsComponentBuilder partitionKey(
+ java.lang.String partitionKey) {
+ doSetProperty("partitionKey", partitionKey);
+ return this;
+ }
+ /**
+ * Sets the EventHubProducerAsyncClient.An asynchronous producer
+ * responsible for transmitting EventData to a specific Event Hub,
+ * grouped together in batches. Depending on the {link
+ * CreateBatchOptions options} specified when creating an
+ * {linkEventDataBatch}, the events may be automatically routed to an
+ * available partition or specific to a partition. Use by this component
+ * to produce the data in camel producer.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.eventhubs.EventHubProducerAsyncClient</code> type.
+ *
+ * Group: producer
+ */
+ default AzureEventhubsComponentBuilder producerAsyncClient(
+ com.azure.messaging.eventhubs.EventHubProducerAsyncClient producerAsyncClient) {
+ doSetProperty("producerAsyncClient", producerAsyncClient);
+ return this;
+ }
+ /**
+ * Whether the component should use basic property binding (Camel 2.x)
+ * or the newer property binding with additional capabilities.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AzureEventhubsComponentBuilder basicPropertyBinding(
+ boolean basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * Instead of supplying namespace, sharedAccessKey, sharedAccessName ...
+ * etc, you can just supply the connection string for your eventHub. The
+ * connection string for EventHubs already include all the necessary
+ * information to connection to your EventHub. To learn on how to
+ * generate the connection string, take a look at this documentation:
+ * https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default AzureEventhubsComponentBuilder connectionString(
+ java.lang.String connectionString) {
+ doSetProperty("connectionString", connectionString);
+ return this;
+ }
+ /**
+ * The generated value for the SharedAccessName.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default AzureEventhubsComponentBuilder sharedAccessKey(
+ java.lang.String sharedAccessKey) {
+ doSetProperty("sharedAccessKey", sharedAccessKey);
+ return this;
+ }
+ /**
+ * The name you chose for your EventHubs SAS keys.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default AzureEventhubsComponentBuilder sharedAccessName(
+ java.lang.String sharedAccessName) {
+ doSetProperty("sharedAccessName", sharedAccessName);
+ return this;
+ }
+ }
+
+ class AzureEventhubsComponentBuilderImpl
+ extends
+ AbstractComponentBuilder<EventHubsComponent>
+ implements
+ AzureEventhubsComponentBuilder {
+ @Override
+ protected EventHubsComponent buildConcreteComponent() {
+ return new EventHubsComponent();
+ }
+ private org.apache.camel.component.azure.eventhubs.EventHubsConfiguration getOrCreateConfiguration(
+ org.apache.camel.component.azure.eventhubs.EventHubsComponent component) {
+ if (component.getConfiguration() == null) {
+ component.setConfiguration(new org.apache.camel.component.azure.eventhubs.EventHubsConfiguration());
+ }
+ return component.getConfiguration();
+ }
+ @Override
+ protected boolean setPropertyOnComponent(
+ Component component,
+ String name,
+ Object value) {
+ switch (name) {
+ case "amqpRetryOptions": getOrCreateConfiguration((EventHubsComponent) component).setAmqpRetryOptions((com.azure.core.amqp.AmqpRetryOptions) value); return true;
+ case "amqpTransportType": getOrCreateConfiguration((EventHubsComponent) component).setAmqpTransportType((com.azure.core.amqp.AmqpTransportType) value); return true;
+ case "autoDiscoverClient": getOrCreateConfiguration((EventHubsComponent) component).setAutoDiscoverClient((boolean) value); return true;
+ case "configuration": ((EventHubsComponent) component).setConfiguration((org.apache.camel.component.azure.eventhubs.EventHubsConfiguration) value); return true;
+ case "blobAccessKey": getOrCreateConfiguration((EventHubsComponent) component).setBlobAccessKey((java.lang.String) value); return true;
+ case "blobAccountName": getOrCreateConfiguration((EventHubsComponent) component).setBlobAccountName((java.lang.String) value); return true;
+ case "blobContainerName": getOrCreateConfiguration((EventHubsComponent) component).setBlobContainerName((java.lang.String) value); return true;
+ case "blobStorageSharedKeyCredential": getOrCreateConfiguration((EventHubsComponent) component).setBlobStorageSharedKeyCredential((com.azure.storage.common.StorageSharedKeyCredential) value); return true;
+ case "bridgeErrorHandler": ((EventHubsComponent) component).setBridgeErrorHandler((boolean) value); return true;
+ case "checkpointStore": getOrCreateConfiguration((EventHubsComponent) component).setCheckpointStore((com.azure.messaging.eventhubs.CheckpointStore) value); return true;
+ case "consumerGroupName": getOrCreateConfiguration((EventHubsComponent) component).setConsumerGroupName((java.lang.String) value); return true;
+ case "eventPosition": getOrCreateConfiguration((EventHubsComponent) component).setEventPosition((java.util.Map) value); return true;
+ case "prefetchCount": getOrCreateConfiguration((EventHubsComponent) component).setPrefetchCount((int) value); return true;
+ case "lazyStartProducer": ((EventHubsComponent) component).setLazyStartProducer((boolean) value); return true;
+ case "partitionId": getOrCreateConfiguration((EventHubsComponent) component).setPartitionId((java.lang.String) value); return true;
+ case "partitionKey": getOrCreateConfiguration((EventHubsComponent) component).setPartitionKey((java.lang.String) value); return true;
+ case "producerAsyncClient": getOrCreateConfiguration((EventHubsComponent) component).setProducerAsyncClient((com.azure.messaging.eventhubs.EventHubProducerAsyncClient) value); return true;
+ case "basicPropertyBinding": ((EventHubsComponent) component).setBasicPropertyBinding((boolean) value); return true;
+ case "connectionString": getOrCreateConfiguration((EventHubsComponent) component).setConnectionString((java.lang.String) value); return true;
+ case "sharedAccessKey": getOrCreateConfiguration((EventHubsComponent) component).setSharedAccessKey((java.lang.String) value); return true;
+ case "sharedAccessName": getOrCreateConfiguration((EventHubsComponent) component).setSharedAccessName((java.lang.String) value); return true;
+ default: return false;
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/core/camel-componentdsl/src/generated/resources/metadata.json b/core/camel-componentdsl/src/generated/resources/metadata.json
index 4d81868..de33e89 100644
--- a/core/camel-componentdsl/src/generated/resources/metadata.json
+++ b/core/camel-componentdsl/src/generated/resources/metadata.json
@@ -1262,6 +1262,27 @@
"producerOnly": false,
"lenientProperties": false
},
+ "AzureEventhubsComponentBuilderFactory": {
+ "kind": "component",
+ "name": "azure-eventhubs",
+ "title": "Azure Event Hubs",
+ "description": "The azure-eventhubs component that integrates Azure Event Hubs using AMQP protocol. Azure EventHubs is a highly scalable publish-subscribe service that can ingest millions of events per second and stream them to multiple consumers.",
+ "deprecated": false,
+ "firstVersion": "3.5.0",
+ "label": "cloud,messaging",
+ "javaType": "org.apache.camel.component.azure.eventhubs.EventHubsComponent",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-azure-eventhubs",
+ "version": "3.5.0-SNAPSHOT",
+ "scheme": "azure-eventhubs",
+ "extendsScheme": "",
+ "syntax": "azure-eventhubs:namespace\/eventHubName",
+ "async": false,
+ "consumerOnly": false,
+ "producerOnly": false,
+ "lenientProperties": false
+ },
"AzureQueueComponentBuilderFactory": {
"kind": "component",
"name": "azure-queue",
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointBuilderFactory.java
index 57b6931..e91bbfc 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointBuilderFactory.java
@@ -115,6 +115,7 @@ public interface EndpointBuilderFactory
org.apache.camel.builder.endpoint.dsl.EtcdStatsEndpointBuilderFactory.EtcdStatsBuilders,
org.apache.camel.builder.endpoint.dsl.EtcdWatchEndpointBuilderFactory.EtcdWatchBuilders,
org.apache.camel.builder.endpoint.dsl.EventEndpointBuilderFactory.EventBuilders,
+ org.apache.camel.builder.endpoint.dsl.EventHubsEndpointBuilderFactory.EventHubsBuilders,
org.apache.camel.builder.endpoint.dsl.ExecEndpointBuilderFactory.ExecBuilders,
org.apache.camel.builder.endpoint.dsl.FacebookEndpointBuilderFactory.FacebookBuilders,
org.apache.camel.builder.endpoint.dsl.FhirEndpointBuilderFactory.FhirBuilders,
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointBuilders.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointBuilders.java
index 75f63cf..8927626 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointBuilders.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointBuilders.java
@@ -112,6 +112,7 @@ public interface EndpointBuilders
org.apache.camel.builder.endpoint.dsl.EtcdStatsEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.EtcdWatchEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.EventEndpointBuilderFactory,
+ org.apache.camel.builder.endpoint.dsl.EventHubsEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.ExecEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.FacebookEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.FhirEndpointBuilderFactory,
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
index 4f06e16..8ad03a8 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java
@@ -2519,6 +2519,59 @@ public class StaticEndpointBuilders {
return org.apache.camel.builder.endpoint.dsl.BlobServiceEndpointBuilderFactory.endpointBuilder(componentName, path);
}
/**
+ * Azure Event Hubs (camel-azure-eventhubs)
+ * The azure-eventhubs component that integrates Azure Event Hubs using AMQP
+ * protocol. Azure EventHubs is a highly scalable publish-subscribe service
+ * that can ingest millions of events per second and stream them to multiple
+ * consumers.
+ *
+ * Category: cloud,messaging
+ * Since: 3.5
+ * Maven coordinates: org.apache.camel:camel-azure-eventhubs
+ *
+ * Syntax: <code>azure-eventhubs:namespace/eventHubName</code>
+ *
+ * Path parameter: namespace
+ * EventHubs namespace created in Azure Portal
+ *
+ * Path parameter: eventHubName
+ * EventHubs name under a specific namcespace
+ *
+ * @param path namespace/eventHubName
+ */
+ public static org.apache.camel.builder.endpoint.dsl.EventHubsEndpointBuilderFactory.EventHubsEndpointBuilder azureEventhubs(
+ String path) {
+ return org.apache.camel.builder.endpoint.dsl.EventHubsEndpointBuilderFactory.endpointBuilder("azure-eventhubs", path);
+ }
+ /**
+ * Azure Event Hubs (camel-azure-eventhubs)
+ * The azure-eventhubs component that integrates Azure Event Hubs using AMQP
+ * protocol. Azure EventHubs is a highly scalable publish-subscribe service
+ * that can ingest millions of events per second and stream them to multiple
+ * consumers.
+ *
+ * Category: cloud,messaging
+ * Since: 3.5
+ * Maven coordinates: org.apache.camel:camel-azure-eventhubs
+ *
+ * Syntax: <code>azure-eventhubs:namespace/eventHubName</code>
+ *
+ * Path parameter: namespace
+ * EventHubs namespace created in Azure Portal
+ *
+ * Path parameter: eventHubName
+ * EventHubs name under a specific namcespace
+ *
+ * @param componentName to use a custom component name for the endpoint
+ * instead of the default name
+ * @param path namespace/eventHubName
+ */
+ public static org.apache.camel.builder.endpoint.dsl.EventHubsEndpointBuilderFactory.EventHubsEndpointBuilder azureEventhubs(
+ String componentName,
+ String path) {
+ return org.apache.camel.builder.endpoint.dsl.EventHubsEndpointBuilderFactory.endpointBuilder(componentName, path);
+ }
+ /**
* Azure Storage Queue Service (Deprecated) (camel-azure)
* Store and retrieve messages from Azure Storage Queue Service.
*
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/EventHubsEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/EventHubsEndpointBuilderFactory.java
new file mode 100644
index 0000000..eec7f4d
--- /dev/null
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/EventHubsEndpointBuilderFactory.java
@@ -0,0 +1,1137 @@
+/*
+ * 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.builder.endpoint.dsl;
+
+import java.util.Map;
+import javax.annotation.Generated;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.builder.EndpointConsumerBuilder;
+import org.apache.camel.builder.EndpointProducerBuilder;
+import org.apache.camel.builder.endpoint.AbstractEndpointBuilder;
+import org.apache.camel.spi.ExceptionHandler;
+
+/**
+ * The azure-eventhubs component that integrates Azure Event Hubs using AMQP
+ * protocol. Azure EventHubs is a highly scalable publish-subscribe service that
+ * can ingest millions of events per second and stream them to multiple
+ * consumers.
+ *
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.EndpointDslMojo")
+public interface EventHubsEndpointBuilderFactory {
+
+
+ /**
+ * Builder for endpoint consumers for the Azure Event Hubs component.
+ */
+ public interface EventHubsEndpointConsumerBuilder
+ extends
+ EndpointConsumerBuilder {
+ default AdvancedEventHubsEndpointConsumerBuilder advanced() {
+ return (AdvancedEventHubsEndpointConsumerBuilder) this;
+ }
+ /**
+ * Sets the retry policy for EventHubAsyncClient. If not specified, the
+ * default retry options are used.
+ *
+ * The option is a: <code>com.azure.core.amqp.AmqpRetryOptions</code>
+ * type.
+ *
+ * Group: common
+ */
+ default EventHubsEndpointConsumerBuilder amqpRetryOptions(
+ Object amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the retry policy for EventHubAsyncClient. If not specified, the
+ * default retry options are used.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.AmqpRetryOptions</code> type.
+ *
+ * Group: common
+ */
+ default EventHubsEndpointConsumerBuilder amqpRetryOptions(
+ String amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Event Hubs occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option is a: <code>com.azure.core.amqp.AmqpTransportType</code>
+ * type.
+ *
+ * Default: Amqp
+ * Group: common
+ */
+ default EventHubsEndpointConsumerBuilder amqpTransportType(
+ AmqpTransportType amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Event Hubs occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.AmqpTransportType</code> type.
+ *
+ * Default: Amqp
+ * Group: common
+ */
+ default EventHubsEndpointConsumerBuilder amqpTransportType(
+ String amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Setting the autoDiscoverClient mechanism, if true, the component will
+ * look for a client instance in the registry automatically otherwise it
+ * will skip that checking.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: common
+ */
+ default EventHubsEndpointConsumerBuilder autoDiscoverClient(
+ boolean autoDiscoverClient) {
+ doSetProperty("autoDiscoverClient", autoDiscoverClient);
+ return this;
+ }
+ /**
+ * Setting the autoDiscoverClient mechanism, if true, the component will
+ * look for a client instance in the registry automatically otherwise it
+ * will skip that checking.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: common
+ */
+ default EventHubsEndpointConsumerBuilder autoDiscoverClient(
+ String autoDiscoverClient) {
+ doSetProperty("autoDiscoverClient", autoDiscoverClient);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets access
+ * key for the associated azure account name to be used for
+ * authentication with azure blob services.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder blobAccessKey(
+ String blobAccessKey) {
+ doSetProperty("blobAccessKey", blobAccessKey);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets Azure
+ * account name to be used for authentication with azure blob services.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder blobAccountName(
+ String blobAccountName) {
+ doSetProperty("blobAccountName", blobAccountName);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore, this sets the blob
+ * container that shall be used by the BlobCheckpointStore to store the
+ * checkpoint offsets.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder blobContainerName(
+ String blobContainerName) {
+ doSetProperty("blobContainerName", blobContainerName);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore,
+ * StorageSharedKeyCredential can be injected to create the azure
+ * client, this holds the important authentication information.
+ *
+ * The option is a:
+ * <code>com.azure.storage.common.StorageSharedKeyCredential</code>
+ * type.
+ *
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder blobStorageSharedKeyCredential(
+ Object blobStorageSharedKeyCredential) {
+ doSetProperty("blobStorageSharedKeyCredential", blobStorageSharedKeyCredential);
+ return this;
+ }
+ /**
+ * In case you chose the default BlobCheckpointStore,
+ * StorageSharedKeyCredential can be injected to create the azure
+ * client, this holds the important authentication information.
+ *
+ * The option will be converted to a
+ * <code>com.azure.storage.common.StorageSharedKeyCredential</code>
+ * type.
+ *
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder blobStorageSharedKeyCredential(
+ String blobStorageSharedKeyCredential) {
+ doSetProperty("blobStorageSharedKeyCredential", blobStorageSharedKeyCredential);
+ return this;
+ }
+ /**
+ * 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 the routing Error Handler. By default the
+ * consumer will use the org.apache.camel.spi.ExceptionHandler to deal
+ * with exceptions, that will be logged at WARN or ERROR level and
+ * ignored.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder bridgeErrorHandler(
+ boolean bridgeErrorHandler) {
+ doSetProperty("bridgeErrorHandler", bridgeErrorHandler);
+ return this;
+ }
+ /**
+ * 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 the routing Error Handler. By default the
+ * consumer will use the org.apache.camel.spi.ExceptionHandler to deal
+ * with exceptions, that will be logged at WARN or ERROR level and
+ * ignored.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder bridgeErrorHandler(
+ String bridgeErrorHandler) {
+ doSetProperty("bridgeErrorHandler", bridgeErrorHandler);
+ return this;
+ }
+ /**
+ * Sets the CheckpointStore the EventProcessorClient will use for
+ * storing partition ownership and checkpoint information. Users can,
+ * optionally, provide their own implementation of CheckpointStore which
+ * will store ownership and checkpoint information. By default it set to
+ * use
+ * com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore which stores all checkpoint offsets into Azure Blob Storage.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.eventhubs.CheckpointStore</code> type.
+ *
+ * Default: BlobCheckpointStore
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder checkpointStore(
+ Object checkpointStore) {
+ doSetProperty("checkpointStore", checkpointStore);
+ return this;
+ }
+ /**
+ * Sets the CheckpointStore the EventProcessorClient will use for
+ * storing partition ownership and checkpoint information. Users can,
+ * optionally, provide their own implementation of CheckpointStore which
+ * will store ownership and checkpoint information. By default it set to
+ * use
+ * com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore which stores all checkpoint offsets into Azure Blob Storage.
+ *
+ * The option will be converted to a
+ * <code>com.azure.messaging.eventhubs.CheckpointStore</code> type.
+ *
+ * Default: BlobCheckpointStore
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder checkpointStore(
+ String checkpointStore) {
+ doSetProperty("checkpointStore", checkpointStore);
+ return this;
+ }
+ /**
+ * Sets the name of the consumer group this consumer is associated with.
+ * Events are read in the context of this group. The name of the
+ * consumer group that is created by default is {link
+ * #DEFAULT_CONSUMER_GROUP_NAME $Default}.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Default: $Default
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder consumerGroupName(
+ String consumerGroupName) {
+ doSetProperty("consumerGroupName", consumerGroupName);
+ return this;
+ }
+ /**
+ * Sets the map containing the event position to use for each partition
+ * if a checkpoint for the partition does not exist in CheckpointStore.
+ * This map is keyed off of the partition id. If there is no checkpoint
+ * in CheckpointStore and there is no entry in this map, the processing
+ * of the partition will start from {link EventPosition#latest() latest}
+ * position.
+ *
+ * The option is a: <code>java.util.Map<java.lang.String,
+ * com.azure.messaging.eventhubs.models.EventPosition></code> type.
+ *
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder eventPosition(
+ Map<String, Object> eventPosition) {
+ doSetProperty("eventPosition", eventPosition);
+ return this;
+ }
+ /**
+ * Sets the map containing the event position to use for each partition
+ * if a checkpoint for the partition does not exist in CheckpointStore.
+ * This map is keyed off of the partition id. If there is no checkpoint
+ * in CheckpointStore and there is no entry in this map, the processing
+ * of the partition will start from {link EventPosition#latest() latest}
+ * position.
+ *
+ * The option will be converted to a
+ * <code>java.util.Map<java.lang.String,
+ * com.azure.messaging.eventhubs.models.EventPosition></code> type.
+ *
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder eventPosition(
+ String eventPosition) {
+ doSetProperty("eventPosition", eventPosition);
+ return this;
+ }
+ /**
+ * Sets the count used by the receiver to control the number of events
+ * the Event Hub consumer will actively receive and queue locally
+ * without regard to whether a receive operation is currently active.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 500
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder prefetchCount(int prefetchCount) {
+ doSetProperty("prefetchCount", prefetchCount);
+ return this;
+ }
+ /**
+ * Sets the count used by the receiver to control the number of events
+ * the Event Hub consumer will actively receive and queue locally
+ * without regard to whether a receive operation is currently active.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Default: 500
+ * Group: consumer
+ */
+ default EventHubsEndpointConsumerBuilder prefetchCount(
+ String prefetchCount) {
+ doSetProperty("prefetchCount", prefetchCount);
+ return this;
+ }
+ /**
+ * Instead of supplying namespace, sharedAccessKey, sharedAccessName ...
+ * etc, you can just supply the connection string for your eventHub. The
+ * connection string for EventHubs already include all the necessary
+ * information to connection to your EventHub. To learn on how to
+ * generate the connection string, take a look at this documentation:
+ * https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointConsumerBuilder connectionString(
+ String connectionString) {
+ doSetProperty("connectionString", connectionString);
+ return this;
+ }
+ /**
+ * The generated value for the SharedAccessName.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointConsumerBuilder sharedAccessKey(
+ String sharedAccessKey) {
+ doSetProperty("sharedAccessKey", sharedAccessKey);
+ return this;
+ }
+ /**
+ * The name you chose for your EventHubs SAS keys.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointConsumerBuilder sharedAccessName(
+ String sharedAccessName) {
+ doSetProperty("sharedAccessName", sharedAccessName);
+ return this;
+ }
+ }
+
+ /**
+ * Advanced builder for endpoint consumers for the Azure Event Hubs
+ * component.
+ */
+ public interface AdvancedEventHubsEndpointConsumerBuilder
+ extends
+ EndpointConsumerBuilder {
+ default EventHubsEndpointConsumerBuilder basic() {
+ return (EventHubsEndpointConsumerBuilder) this;
+ }
+ /**
+ * 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.
+ *
+ * The option is a: <code>org.apache.camel.spi.ExceptionHandler</code>
+ * type.
+ *
+ * Group: consumer (advanced)
+ */
+ default AdvancedEventHubsEndpointConsumerBuilder exceptionHandler(
+ ExceptionHandler exceptionHandler) {
+ doSetProperty("exceptionHandler", exceptionHandler);
+ return this;
+ }
+ /**
+ * 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.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.spi.ExceptionHandler</code> type.
+ *
+ * Group: consumer (advanced)
+ */
+ default AdvancedEventHubsEndpointConsumerBuilder exceptionHandler(
+ String exceptionHandler) {
+ doSetProperty("exceptionHandler", exceptionHandler);
+ return this;
+ }
+ /**
+ * Sets the exchange pattern when the consumer creates an exchange.
+ *
+ * The option is a: <code>org.apache.camel.ExchangePattern</code> type.
+ *
+ * Group: consumer (advanced)
+ */
+ default AdvancedEventHubsEndpointConsumerBuilder exchangePattern(
+ ExchangePattern exchangePattern) {
+ doSetProperty("exchangePattern", exchangePattern);
+ return this;
+ }
+ /**
+ * Sets the exchange pattern when the consumer creates an exchange.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.ExchangePattern</code> type.
+ *
+ * Group: consumer (advanced)
+ */
+ default AdvancedEventHubsEndpointConsumerBuilder exchangePattern(
+ String exchangePattern) {
+ doSetProperty("exchangePattern", exchangePattern);
+ return this;
+ }
+ /**
+ * Whether the endpoint should use basic property binding (Camel 2.x) or
+ * the newer property binding with additional capabilities.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointConsumerBuilder basicPropertyBinding(
+ boolean basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * Whether the endpoint should use basic property binding (Camel 2.x) or
+ * the newer property binding with additional capabilities.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointConsumerBuilder basicPropertyBinding(
+ String basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * Sets whether synchronous processing should be strictly used, or Camel
+ * is allowed to use asynchronous processing (if supported).
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointConsumerBuilder synchronous(
+ boolean synchronous) {
+ doSetProperty("synchronous", synchronous);
+ return this;
+ }
+ /**
+ * Sets whether synchronous processing should be strictly used, or Camel
+ * is allowed to use asynchronous processing (if supported).
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointConsumerBuilder synchronous(
+ String synchronous) {
+ doSetProperty("synchronous", synchronous);
+ return this;
+ }
+ }
+
+ /**
+ * Builder for endpoint producers for the Azure Event Hubs component.
+ */
+ public interface EventHubsEndpointProducerBuilder
+ extends
+ EndpointProducerBuilder {
+ default AdvancedEventHubsEndpointProducerBuilder advanced() {
+ return (AdvancedEventHubsEndpointProducerBuilder) this;
+ }
+ /**
+ * Sets the retry policy for EventHubAsyncClient. If not specified, the
+ * default retry options are used.
+ *
+ * The option is a: <code>com.azure.core.amqp.AmqpRetryOptions</code>
+ * type.
+ *
+ * Group: common
+ */
+ default EventHubsEndpointProducerBuilder amqpRetryOptions(
+ Object amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the retry policy for EventHubAsyncClient. If not specified, the
+ * default retry options are used.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.AmqpRetryOptions</code> type.
+ *
+ * Group: common
+ */
+ default EventHubsEndpointProducerBuilder amqpRetryOptions(
+ String amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Event Hubs occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option is a: <code>com.azure.core.amqp.AmqpTransportType</code>
+ * type.
+ *
+ * Default: Amqp
+ * Group: common
+ */
+ default EventHubsEndpointProducerBuilder amqpTransportType(
+ AmqpTransportType amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Event Hubs occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.AmqpTransportType</code> type.
+ *
+ * Default: Amqp
+ * Group: common
+ */
+ default EventHubsEndpointProducerBuilder amqpTransportType(
+ String amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Setting the autoDiscoverClient mechanism, if true, the component will
+ * look for a client instance in the registry automatically otherwise it
+ * will skip that checking.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: common
+ */
+ default EventHubsEndpointProducerBuilder autoDiscoverClient(
+ boolean autoDiscoverClient) {
+ doSetProperty("autoDiscoverClient", autoDiscoverClient);
+ return this;
+ }
+ /**
+ * Setting the autoDiscoverClient mechanism, if true, the component will
+ * look for a client instance in the registry automatically otherwise it
+ * will skip that checking.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: common
+ */
+ default EventHubsEndpointProducerBuilder autoDiscoverClient(
+ String autoDiscoverClient) {
+ doSetProperty("autoDiscoverClient", autoDiscoverClient);
+ return this;
+ }
+ /**
+ * 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 starting the
+ * producer may take a little time and prolong the total processing time
+ * of the processing.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer
+ */
+ default EventHubsEndpointProducerBuilder lazyStartProducer(
+ boolean lazyStartProducer) {
+ doSetProperty("lazyStartProducer", lazyStartProducer);
+ return this;
+ }
+ /**
+ * 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 starting the
+ * producer may take a little time and prolong the total processing time
+ * of the processing.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer
+ */
+ default EventHubsEndpointProducerBuilder lazyStartProducer(
+ String lazyStartProducer) {
+ doSetProperty("lazyStartProducer", lazyStartProducer);
+ return this;
+ }
+ /**
+ * Sets the identifier of the Event Hub partition that the {link
+ * EventData events} will be sent to. If the identifier is not
+ * specified, the Event Hubs service will be responsible for routing
+ * events that are sent to an available partition.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: producer
+ */
+ default EventHubsEndpointProducerBuilder partitionId(String partitionId) {
+ doSetProperty("partitionId", partitionId);
+ return this;
+ }
+ /**
+ * Sets a hashing key to be provided for the batch of events, which
+ * instructs the Event Hubs service to map this key to a specific
+ * partition. The selection of a partition is stable for a given
+ * partition hashing key. Should any other batches of events be sent
+ * using the same exact partition hashing key, the Event Hubs service
+ * will route them all to the same partition. This should be specified
+ * only when there is a need to group events by partition, but there is
+ * flexibility into which partition they are routed. If ensuring that a
+ * batch of events is sent only to a specific partition, it is
+ * recommended that the {link #setPartitionId(String) identifier of the
+ * position be specified directly} when sending the batch.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: producer
+ */
+ default EventHubsEndpointProducerBuilder partitionKey(
+ String partitionKey) {
+ doSetProperty("partitionKey", partitionKey);
+ return this;
+ }
+ /**
+ * Sets the EventHubProducerAsyncClient.An asynchronous producer
+ * responsible for transmitting EventData to a specific Event Hub,
+ * grouped together in batches. Depending on the {link
+ * CreateBatchOptions options} specified when creating an
+ * {linkEventDataBatch}, the events may be automatically routed to an
+ * available partition or specific to a partition. Use by this component
+ * to produce the data in camel producer.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.eventhubs.EventHubProducerAsyncClient</code> type.
+ *
+ * Group: producer
+ */
+ default EventHubsEndpointProducerBuilder producerAsyncClient(
+ Object producerAsyncClient) {
+ doSetProperty("producerAsyncClient", producerAsyncClient);
+ return this;
+ }
+ /**
+ * Sets the EventHubProducerAsyncClient.An asynchronous producer
+ * responsible for transmitting EventData to a specific Event Hub,
+ * grouped together in batches. Depending on the {link
+ * CreateBatchOptions options} specified when creating an
+ * {linkEventDataBatch}, the events may be automatically routed to an
+ * available partition or specific to a partition. Use by this component
+ * to produce the data in camel producer.
+ *
+ * The option will be converted to a
+ * <code>com.azure.messaging.eventhubs.EventHubProducerAsyncClient</code> type.
+ *
+ * Group: producer
+ */
+ default EventHubsEndpointProducerBuilder producerAsyncClient(
+ String producerAsyncClient) {
+ doSetProperty("producerAsyncClient", producerAsyncClient);
+ return this;
+ }
+ /**
+ * Instead of supplying namespace, sharedAccessKey, sharedAccessName ...
+ * etc, you can just supply the connection string for your eventHub. The
+ * connection string for EventHubs already include all the necessary
+ * information to connection to your EventHub. To learn on how to
+ * generate the connection string, take a look at this documentation:
+ * https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointProducerBuilder connectionString(
+ String connectionString) {
+ doSetProperty("connectionString", connectionString);
+ return this;
+ }
+ /**
+ * The generated value for the SharedAccessName.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointProducerBuilder sharedAccessKey(
+ String sharedAccessKey) {
+ doSetProperty("sharedAccessKey", sharedAccessKey);
+ return this;
+ }
+ /**
+ * The name you chose for your EventHubs SAS keys.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointProducerBuilder sharedAccessName(
+ String sharedAccessName) {
+ doSetProperty("sharedAccessName", sharedAccessName);
+ return this;
+ }
+ }
+
+ /**
+ * Advanced builder for endpoint producers for the Azure Event Hubs
+ * component.
+ */
+ public interface AdvancedEventHubsEndpointProducerBuilder
+ extends
+ EndpointProducerBuilder {
+ default EventHubsEndpointProducerBuilder basic() {
+ return (EventHubsEndpointProducerBuilder) this;
+ }
+ /**
+ * Whether the endpoint should use basic property binding (Camel 2.x) or
+ * the newer property binding with additional capabilities.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointProducerBuilder basicPropertyBinding(
+ boolean basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * Whether the endpoint should use basic property binding (Camel 2.x) or
+ * the newer property binding with additional capabilities.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointProducerBuilder basicPropertyBinding(
+ String basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * Sets whether synchronous processing should be strictly used, or Camel
+ * is allowed to use asynchronous processing (if supported).
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointProducerBuilder synchronous(
+ boolean synchronous) {
+ doSetProperty("synchronous", synchronous);
+ return this;
+ }
+ /**
+ * Sets whether synchronous processing should be strictly used, or Camel
+ * is allowed to use asynchronous processing (if supported).
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointProducerBuilder synchronous(
+ String synchronous) {
+ doSetProperty("synchronous", synchronous);
+ return this;
+ }
+ }
+
+ /**
+ * Builder for endpoint for the Azure Event Hubs component.
+ */
+ public interface EventHubsEndpointBuilder
+ extends
+ EventHubsEndpointConsumerBuilder,
+ EventHubsEndpointProducerBuilder {
+ default AdvancedEventHubsEndpointBuilder advanced() {
+ return (AdvancedEventHubsEndpointBuilder) this;
+ }
+ /**
+ * Sets the retry policy for EventHubAsyncClient. If not specified, the
+ * default retry options are used.
+ *
+ * The option is a: <code>com.azure.core.amqp.AmqpRetryOptions</code>
+ * type.
+ *
+ * Group: common
+ */
+ default EventHubsEndpointBuilder amqpRetryOptions(
+ Object amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the retry policy for EventHubAsyncClient. If not specified, the
+ * default retry options are used.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.AmqpRetryOptions</code> type.
+ *
+ * Group: common
+ */
+ default EventHubsEndpointBuilder amqpRetryOptions(
+ String amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Event Hubs occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option is a: <code>com.azure.core.amqp.AmqpTransportType</code>
+ * type.
+ *
+ * Default: Amqp
+ * Group: common
+ */
+ default EventHubsEndpointBuilder amqpTransportType(
+ AmqpTransportType amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Event Hubs occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.AmqpTransportType</code> type.
+ *
+ * Default: Amqp
+ * Group: common
+ */
+ default EventHubsEndpointBuilder amqpTransportType(
+ String amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Setting the autoDiscoverClient mechanism, if true, the component will
+ * look for a client instance in the registry automatically otherwise it
+ * will skip that checking.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: common
+ */
+ default EventHubsEndpointBuilder autoDiscoverClient(
+ boolean autoDiscoverClient) {
+ doSetProperty("autoDiscoverClient", autoDiscoverClient);
+ return this;
+ }
+ /**
+ * Setting the autoDiscoverClient mechanism, if true, the component will
+ * look for a client instance in the registry automatically otherwise it
+ * will skip that checking.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: common
+ */
+ default EventHubsEndpointBuilder autoDiscoverClient(
+ String autoDiscoverClient) {
+ doSetProperty("autoDiscoverClient", autoDiscoverClient);
+ return this;
+ }
+ /**
+ * Instead of supplying namespace, sharedAccessKey, sharedAccessName ...
+ * etc, you can just supply the connection string for your eventHub. The
+ * connection string for EventHubs already include all the necessary
+ * information to connection to your EventHub. To learn on how to
+ * generate the connection string, take a look at this documentation:
+ * https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointBuilder connectionString(
+ String connectionString) {
+ doSetProperty("connectionString", connectionString);
+ return this;
+ }
+ /**
+ * The generated value for the SharedAccessName.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointBuilder sharedAccessKey(String sharedAccessKey) {
+ doSetProperty("sharedAccessKey", sharedAccessKey);
+ return this;
+ }
+ /**
+ * The name you chose for your EventHubs SAS keys.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ */
+ default EventHubsEndpointBuilder sharedAccessName(
+ String sharedAccessName) {
+ doSetProperty("sharedAccessName", sharedAccessName);
+ return this;
+ }
+ }
+
+ /**
+ * Advanced builder for endpoint for the Azure Event Hubs component.
+ */
+ public interface AdvancedEventHubsEndpointBuilder
+ extends
+ AdvancedEventHubsEndpointConsumerBuilder,
+ AdvancedEventHubsEndpointProducerBuilder {
+ default EventHubsEndpointBuilder basic() {
+ return (EventHubsEndpointBuilder) this;
+ }
+ /**
+ * Whether the endpoint should use basic property binding (Camel 2.x) or
+ * the newer property binding with additional capabilities.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointBuilder basicPropertyBinding(
+ boolean basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * Whether the endpoint should use basic property binding (Camel 2.x) or
+ * the newer property binding with additional capabilities.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointBuilder basicPropertyBinding(
+ String basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * Sets whether synchronous processing should be strictly used, or Camel
+ * is allowed to use asynchronous processing (if supported).
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointBuilder synchronous(boolean synchronous) {
+ doSetProperty("synchronous", synchronous);
+ return this;
+ }
+ /**
+ * Sets whether synchronous processing should be strictly used, or Camel
+ * is allowed to use asynchronous processing (if supported).
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default AdvancedEventHubsEndpointBuilder synchronous(String synchronous) {
+ doSetProperty("synchronous", synchronous);
+ return this;
+ }
+ }
+
+ /**
+ * Proxy enum for <code>com.azure.core.amqp.AmqpTransportType</code> enum.
+ */
+ enum AmqpTransportType {
+ AMQP,
+ AMQP_WEB_SOCKETS;
+ }
+
+ public interface EventHubsBuilders {
+ /**
+ * Azure Event Hubs (camel-azure-eventhubs)
+ * The azure-eventhubs component that integrates Azure Event Hubs using
+ * AMQP protocol. Azure EventHubs is a highly scalable publish-subscribe
+ * service that can ingest millions of events per second and stream them
+ * to multiple consumers.
+ *
+ * Category: cloud,messaging
+ * Since: 3.5
+ * Maven coordinates: org.apache.camel:camel-azure-eventhubs
+ *
+ * Syntax: <code>azure-eventhubs:namespace/eventHubName</code>
+ *
+ * Path parameter: namespace
+ * EventHubs namespace created in Azure Portal
+ *
+ * Path parameter: eventHubName
+ * EventHubs name under a specific namcespace
+ *
+ * @param path namespace/eventHubName
+ */
+ default EventHubsEndpointBuilder azureEventhubs(String path) {
+ return EventHubsEndpointBuilderFactory.endpointBuilder("azure-eventhubs", path);
+ }
+ /**
+ * Azure Event Hubs (camel-azure-eventhubs)
+ * The azure-eventhubs component that integrates Azure Event Hubs using
+ * AMQP protocol. Azure EventHubs is a highly scalable publish-subscribe
+ * service that can ingest millions of events per second and stream them
+ * to multiple consumers.
+ *
+ * Category: cloud,messaging
+ * Since: 3.5
+ * Maven coordinates: org.apache.camel:camel-azure-eventhubs
+ *
+ * Syntax: <code>azure-eventhubs:namespace/eventHubName</code>
+ *
+ * Path parameter: namespace
+ * EventHubs namespace created in Azure Portal
+ *
+ * Path parameter: eventHubName
+ * EventHubs name under a specific namcespace
+ *
+ * @param componentName to use a custom component name for the endpoint
+ * instead of the default name
+ * @param path namespace/eventHubName
+ */
+ default EventHubsEndpointBuilder azureEventhubs(
+ String componentName,
+ String path) {
+ return EventHubsEndpointBuilderFactory.endpointBuilder(componentName, path);
+ }
+ }
+ static EventHubsEndpointBuilder endpointBuilder(
+ String componentName,
+ String path) {
+ class EventHubsEndpointBuilderImpl extends AbstractEndpointBuilder implements EventHubsEndpointBuilder, AdvancedEventHubsEndpointBuilder {
+ public EventHubsEndpointBuilderImpl(String path) {
+ super(componentName, path);
+ }
+ }
+ return new EventHubsEndpointBuilderImpl(path);
+ }
+}
\ No newline at end of file
diff --git a/parent/pom.xml b/parent/pom.xml
index ec9e33e..e3ef4ca 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -85,6 +85,8 @@
<azure-storage-java-sdk-version>8.6.0</azure-storage-java-sdk-version>
<azure-storage-blob-java-sdk12-version>12.7.0</azure-storage-blob-java-sdk12-version>
<azure-storage-queue-java-sdk12-version>12.5.2</azure-storage-queue-java-sdk12-version>
+ <azure-eventhubs-java-version>5.1.2</azure-eventhubs-java-version>
+ <azure-eventhubs-checkpointstore-blob-version>1.1.2</azure-eventhubs-checkpointstore-blob-version>
<beanio-version>2.1.0</beanio-version>
<beanstalkd-client-version>1.4.6</beanstalkd-client-version>
<bouncycastle-version>1.66</bouncycastle-version>
@@ -1042,6 +1044,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-azure-eventhubs</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-barcode</artifactId>
<version>${project.version}</version>
</dependency>