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 2021/07/06 08:10:51 UTC
[camel] branch main updated: CAMEL-16589: Add Azure ServiceBus
component (#5793)
This is an automated email from the ASF dual-hosted git repository.
oalsafi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a9d6af0 CAMEL-16589: Add Azure ServiceBus component (#5793)
a9d6af0 is described below
commit a9d6af024513b4468ff97e577b94f26112cdbe4b
Author: Omar Al-Safi <om...@gmail.com>
AuthorDate: Tue Jul 6 10:10:26 2021 +0200
CAMEL-16589: Add Azure ServiceBus component (#5793)
* Initial commit
* CAMEL-16589: Add the wrapper clients
* CAMEL-16589: Add the producer
* CAMEL-16589: Add the producer IT tests
* CAMEL-16589: Add the ServiceBusConsumerOperations
* CAMEL-16589: Add configs
* CAMEL-16589: Add configs
* CAMEL-16589: Add builder
* CAMEL-16589: Work on the producer
* CAMEL-16589: Progress on the producer
* Rebuild
* CAMEL-16589: Build against 3.12.0-SNAPSHOT
* CAMEL-16589: Complete the producer
* CAMEL-16589: Complete the producer
* CAMEL-16589: Complete the consumer
* CAMEL-16589: Complete the component
* CAMEL-16589: Add the docs
---
bom/camel-bom/pom.xml | 5 +
.../catalog/docs/azure-servicebus-component.adoc | 29 +
.../camel-azure/camel-azure-servicebus/pom.xml | 90 ++
.../servicebus/ServiceBusComponentConfigurer.java | 190 +++
.../servicebus/ServiceBusEndpointConfigurer.java | 186 +++
.../servicebus/ServiceBusEndpointUriFactory.java | 86 ++
.../servicebus/ServiceBusTypeConverterLoader.java | 54 +
.../services/org/apache/camel/TypeConverterLoader | 2 +
.../services/org/apache/camel/component.properties | 7 +
.../org/apache/camel/component/azure-servicebus | 2 +
.../camel/configurer/azure-servicebus-component | 2 +
.../camel/configurer/azure-servicebus-endpoint | 2 +
.../camel/urifactory/azure-servicebus-endpoint | 2 +
.../azure/servicebus/azure-servicebus.json | 75 +
.../src/main/docs/azure-servicebus-component.adoc | 255 ++++
.../azure/servicebus/ServiceBusComponent.java | 86 ++
.../azure/servicebus/ServiceBusConfiguration.java | 333 +++++
.../ServiceBusConfigurationOptionsProxy.java | 69 +
.../azure/servicebus/ServiceBusConstants.java | 56 +
.../azure/servicebus/ServiceBusConsumer.java | 213 +++
.../ServiceBusConsumerOperationDefinition.java | 22 +
.../azure/servicebus/ServiceBusEndpoint.java | 70 +
.../azure/servicebus/ServiceBusProducer.java | 208 +++
.../ServiceBusProducerOperationDefinition.java | 22 +
.../component/azure/servicebus/ServiceBusType.java | 22 +
.../azure/servicebus/ServiceBusTypeConverter.java | 44 +
.../azure/servicebus/ServiceBusUtils.java | 47 +
.../servicebus/client/ServiceBusClientFactory.java | 80 ++
.../ServiceBusReceiverAsyncClientWrapper.java | 158 +++
.../client/ServiceBusSenderAsyncClientWrapper.java | 133 ++
.../operations/ServiceBusReceiverOperations.java | 46 +
.../operations/ServiceBusSenderOperations.java | 107 ++
.../azure/servicebus/ServiceBusEndpointTest.java | 59 +
.../azure/servicebus/ServiceBusTestUtils.java | 98 ++
.../azure/servicebus/ServiceBusUtilsTest.java | 56 +
.../BaseCamelServiceBusTestSupport.java | 63 +
.../integration/ServiceBusConsumerTest.java | 115 ++
.../integration/ServiceBusProducerTest.java | 172 +++
.../operations/ServiceBusSenderOperationsTest.java | 181 +++
.../src/test/resources/log4j2.properties | 27 +
components/camel-azure/pom.xml | 1 +
core/camel-allcomponents/pom.xml | 4 +
.../component/ComponentsBuilderFactory.java | 18 +
.../AzureServicebusComponentBuilderFactory.java | 532 ++++++++
.../src/generated/resources/metadata.json | 22 +
.../builder/endpoint/EndpointBuilderFactory.java | 1 +
.../camel/builder/endpoint/EndpointBuilders.java | 1 +
.../builder/endpoint/StaticEndpointBuilders.java | 53 +
.../dsl/ServiceBusEndpointBuilderFactory.java | 1440 ++++++++++++++++++++
.../ROOT/pages/azure-servicebus-component.adoc | 31 +
parent/pom.xml | 5 +
51 files changed, 5582 insertions(+)
diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml
index f5f84ef..b396f38c 100644
--- a/bom/camel-bom/pom.xml
+++ b/bom/camel-bom/pom.xml
@@ -278,6 +278,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-azure-servicebus</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-azure-parent</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-servicebus-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-servicebus-component.adoc
new file mode 100644
index 0000000..161ed3b
--- /dev/null
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-servicebus-component.adoc
@@ -0,0 +1,29 @@
+[[azure-servicebus-component]]
+= Azure ServiceBus Component
+:docTitle: Azure ServiceBus
+:artifactId: camel-azure-servicebus
+:description: The azure-servicebus component that integrates Azure ServiceBus. Azure ServiceBus is a fully managed enterprise integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and secure platform for asynchronous transfer of data and state. Data is transferred between different applications and services using messages.
+:since: 3.12
+:supportLevel: Preview
+:component-header: Both producer and consumer are supported
+include::{cq-version}@camel-quarkus:ROOT:partial$reference/components/azure-servicebus.adoc[opts=optional]
+//Manually maintained attributes
+:group: Azure
+
+*Since Camel {since}*
+
+*{component-header}*
+
+```
+
+=== 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::{page-component-version}@camel-spring-boot::page$azure-servicebus-starter.adoc[]
diff --git a/components/camel-azure/camel-azure-servicebus/pom.xml b/components/camel-azure/camel-azure-servicebus/pom.xml
new file mode 100644
index 0000000..faad2e9
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/pom.xml
@@ -0,0 +1,90 @@
+<?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>camel-azure-parent</artifactId>
+ <version>3.12.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-azure-servicebus</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel :: Azure ServiceBus</name>
+ <description>Camel Azure ServiceBus Component</description>
+
+ <properties>
+ <camel.osgi.import.pkg>
+ !org.apache.camel.component.azure.eventhubs*,
+ reactor*;version="[3,4)",
+ ${camel.osgi.import.defaults},
+ *
+ </camel.osgi.import.pkg>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-support</artifactId>
+ </dependency>
+
+ <!-- azure sdk -->
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-messaging-servicebus</artifactId>
+ </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>
+</project>
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusComponentConfigurer.java b/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusComponentConfigurer.java
new file mode 100644
index 0000000..8f288ed
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusComponentConfigurer.java
@@ -0,0 +1,190 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.azure.servicebus;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class ServiceBusComponentConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ private org.apache.camel.component.azure.servicebus.ServiceBusConfiguration getOrCreateConfiguration(ServiceBusComponent target) {
+ if (target.getConfiguration() == null) {
+ target.setConfiguration(new org.apache.camel.component.azure.servicebus.ServiceBusConfiguration());
+ }
+ return target.getConfiguration();
+ }
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ ServiceBusComponent target = (ServiceBusComponent) 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 "autowiredenabled":
+ case "autowiredEnabled": target.setAutowiredEnabled(property(camelContext, boolean.class, value)); return true;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+ case "clientoptions":
+ case "clientOptions": getOrCreateConfiguration(target).setClientOptions(property(camelContext, com.azure.core.util.ClientOptions.class, value)); return true;
+ case "configuration": target.setConfiguration(property(camelContext, org.apache.camel.component.azure.servicebus.ServiceBusConfiguration.class, value)); return true;
+ case "connectionstring":
+ case "connectionString": getOrCreateConfiguration(target).setConnectionString(property(camelContext, java.lang.String.class, value)); return true;
+ case "consumeroperation":
+ case "consumerOperation": getOrCreateConfiguration(target).setConsumerOperation(property(camelContext, org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition.class, value)); return true;
+ case "disableautocomplete":
+ case "disableAutoComplete": getOrCreateConfiguration(target).setDisableAutoComplete(property(camelContext, boolean.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "maxautolockrenewduration":
+ case "maxAutoLockRenewDuration": getOrCreateConfiguration(target).setMaxAutoLockRenewDuration(property(camelContext, java.time.Duration.class, value)); return true;
+ case "peeknummaxmessages":
+ case "peekNumMaxMessages": getOrCreateConfiguration(target).setPeekNumMaxMessages(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "prefetchcount":
+ case "prefetchCount": getOrCreateConfiguration(target).setPrefetchCount(property(camelContext, int.class, value)); return true;
+ case "produceroperation":
+ case "producerOperation": getOrCreateConfiguration(target).setProducerOperation(property(camelContext, org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition.class, value)); return true;
+ case "proxyoptions":
+ case "proxyOptions": getOrCreateConfiguration(target).setProxyOptions(property(camelContext, com.azure.core.amqp.ProxyOptions.class, value)); return true;
+ case "receiverasyncclient":
+ case "receiverAsyncClient": getOrCreateConfiguration(target).setReceiverAsyncClient(property(camelContext, com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient.class, value)); return true;
+ case "scheduledenqueuetime":
+ case "scheduledEnqueueTime": getOrCreateConfiguration(target).setScheduledEnqueueTime(property(camelContext, java.time.OffsetDateTime.class, value)); return true;
+ case "senderasyncclient":
+ case "senderAsyncClient": getOrCreateConfiguration(target).setSenderAsyncClient(property(camelContext, com.azure.messaging.servicebus.ServiceBusSenderAsyncClient.class, value)); return true;
+ case "servicebusreceivemode":
+ case "serviceBusReceiveMode": getOrCreateConfiguration(target).setServiceBusReceiveMode(property(camelContext, com.azure.messaging.servicebus.models.ServiceBusReceiveMode.class, value)); return true;
+ case "servicebustransactioncontext":
+ case "serviceBusTransactionContext": getOrCreateConfiguration(target).setServiceBusTransactionContext(property(camelContext, com.azure.messaging.servicebus.ServiceBusTransactionContext.class, value)); return true;
+ case "servicebustype":
+ case "serviceBusType": getOrCreateConfiguration(target).setServiceBusType(property(camelContext, org.apache.camel.component.azure.servicebus.ServiceBusType.class, value)); return true;
+ case "subqueue":
+ case "subQueue": getOrCreateConfiguration(target).setSubQueue(property(camelContext, com.azure.messaging.servicebus.models.SubQueue.class, value)); return true;
+ case "subscriptionname":
+ case "subscriptionName": getOrCreateConfiguration(target).setSubscriptionName(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public String[] getAutowiredNames() {
+ return new String[]{"receiverAsyncClient","senderAsyncClient"};
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amqpretryoptions":
+ case "amqpRetryOptions": return com.azure.core.amqp.AmqpRetryOptions.class;
+ case "amqptransporttype":
+ case "amqpTransportType": return com.azure.core.amqp.AmqpTransportType.class;
+ case "autowiredenabled":
+ case "autowiredEnabled": return boolean.class;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return boolean.class;
+ case "clientoptions":
+ case "clientOptions": return com.azure.core.util.ClientOptions.class;
+ case "configuration": return org.apache.camel.component.azure.servicebus.ServiceBusConfiguration.class;
+ case "connectionstring":
+ case "connectionString": return java.lang.String.class;
+ case "consumeroperation":
+ case "consumerOperation": return org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition.class;
+ case "disableautocomplete":
+ case "disableAutoComplete": return boolean.class;
+ case "lazystartproducer":
+ case "lazyStartProducer": return boolean.class;
+ case "maxautolockrenewduration":
+ case "maxAutoLockRenewDuration": return java.time.Duration.class;
+ case "peeknummaxmessages":
+ case "peekNumMaxMessages": return java.lang.Integer.class;
+ case "prefetchcount":
+ case "prefetchCount": return int.class;
+ case "produceroperation":
+ case "producerOperation": return org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition.class;
+ case "proxyoptions":
+ case "proxyOptions": return com.azure.core.amqp.ProxyOptions.class;
+ case "receiverasyncclient":
+ case "receiverAsyncClient": return com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient.class;
+ case "scheduledenqueuetime":
+ case "scheduledEnqueueTime": return java.time.OffsetDateTime.class;
+ case "senderasyncclient":
+ case "senderAsyncClient": return com.azure.messaging.servicebus.ServiceBusSenderAsyncClient.class;
+ case "servicebusreceivemode":
+ case "serviceBusReceiveMode": return com.azure.messaging.servicebus.models.ServiceBusReceiveMode.class;
+ case "servicebustransactioncontext":
+ case "serviceBusTransactionContext": return com.azure.messaging.servicebus.ServiceBusTransactionContext.class;
+ case "servicebustype":
+ case "serviceBusType": return org.apache.camel.component.azure.servicebus.ServiceBusType.class;
+ case "subqueue":
+ case "subQueue": return com.azure.messaging.servicebus.models.SubQueue.class;
+ case "subscriptionname":
+ case "subscriptionName": return java.lang.String.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ ServiceBusComponent target = (ServiceBusComponent) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amqpretryoptions":
+ case "amqpRetryOptions": return getOrCreateConfiguration(target).getAmqpRetryOptions();
+ case "amqptransporttype":
+ case "amqpTransportType": return getOrCreateConfiguration(target).getAmqpTransportType();
+ case "autowiredenabled":
+ case "autowiredEnabled": return target.isAutowiredEnabled();
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return target.isBridgeErrorHandler();
+ case "clientoptions":
+ case "clientOptions": return getOrCreateConfiguration(target).getClientOptions();
+ case "configuration": return target.getConfiguration();
+ case "connectionstring":
+ case "connectionString": return getOrCreateConfiguration(target).getConnectionString();
+ case "consumeroperation":
+ case "consumerOperation": return getOrCreateConfiguration(target).getConsumerOperation();
+ case "disableautocomplete":
+ case "disableAutoComplete": return getOrCreateConfiguration(target).isDisableAutoComplete();
+ case "lazystartproducer":
+ case "lazyStartProducer": return target.isLazyStartProducer();
+ case "maxautolockrenewduration":
+ case "maxAutoLockRenewDuration": return getOrCreateConfiguration(target).getMaxAutoLockRenewDuration();
+ case "peeknummaxmessages":
+ case "peekNumMaxMessages": return getOrCreateConfiguration(target).getPeekNumMaxMessages();
+ case "prefetchcount":
+ case "prefetchCount": return getOrCreateConfiguration(target).getPrefetchCount();
+ case "produceroperation":
+ case "producerOperation": return getOrCreateConfiguration(target).getProducerOperation();
+ case "proxyoptions":
+ case "proxyOptions": return getOrCreateConfiguration(target).getProxyOptions();
+ case "receiverasyncclient":
+ case "receiverAsyncClient": return getOrCreateConfiguration(target).getReceiverAsyncClient();
+ case "scheduledenqueuetime":
+ case "scheduledEnqueueTime": return getOrCreateConfiguration(target).getScheduledEnqueueTime();
+ case "senderasyncclient":
+ case "senderAsyncClient": return getOrCreateConfiguration(target).getSenderAsyncClient();
+ case "servicebusreceivemode":
+ case "serviceBusReceiveMode": return getOrCreateConfiguration(target).getServiceBusReceiveMode();
+ case "servicebustransactioncontext":
+ case "serviceBusTransactionContext": return getOrCreateConfiguration(target).getServiceBusTransactionContext();
+ case "servicebustype":
+ case "serviceBusType": return getOrCreateConfiguration(target).getServiceBusType();
+ case "subqueue":
+ case "subQueue": return getOrCreateConfiguration(target).getSubQueue();
+ case "subscriptionname":
+ case "subscriptionName": return getOrCreateConfiguration(target).getSubscriptionName();
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointConfigurer.java b/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointConfigurer.java
new file mode 100644
index 0000000..07bcf0d
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointConfigurer.java
@@ -0,0 +1,186 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.azure.servicebus;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class ServiceBusEndpointConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ ServiceBusEndpoint target = (ServiceBusEndpoint) 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 "bridgeerrorhandler":
+ case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+ case "clientoptions":
+ case "clientOptions": target.getConfiguration().setClientOptions(property(camelContext, com.azure.core.util.ClientOptions.class, value)); return true;
+ case "connectionstring":
+ case "connectionString": target.getConfiguration().setConnectionString(property(camelContext, java.lang.String.class, value)); return true;
+ case "consumeroperation":
+ case "consumerOperation": target.getConfiguration().setConsumerOperation(property(camelContext, org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition.class, value)); return true;
+ case "disableautocomplete":
+ case "disableAutoComplete": target.getConfiguration().setDisableAutoComplete(property(camelContext, boolean.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 "maxautolockrenewduration":
+ case "maxAutoLockRenewDuration": target.getConfiguration().setMaxAutoLockRenewDuration(property(camelContext, java.time.Duration.class, value)); return true;
+ case "peeknummaxmessages":
+ case "peekNumMaxMessages": target.getConfiguration().setPeekNumMaxMessages(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "prefetchcount":
+ case "prefetchCount": target.getConfiguration().setPrefetchCount(property(camelContext, int.class, value)); return true;
+ case "produceroperation":
+ case "producerOperation": target.getConfiguration().setProducerOperation(property(camelContext, org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition.class, value)); return true;
+ case "proxyoptions":
+ case "proxyOptions": target.getConfiguration().setProxyOptions(property(camelContext, com.azure.core.amqp.ProxyOptions.class, value)); return true;
+ case "receiverasyncclient":
+ case "receiverAsyncClient": target.getConfiguration().setReceiverAsyncClient(property(camelContext, com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient.class, value)); return true;
+ case "scheduledenqueuetime":
+ case "scheduledEnqueueTime": target.getConfiguration().setScheduledEnqueueTime(property(camelContext, java.time.OffsetDateTime.class, value)); return true;
+ case "senderasyncclient":
+ case "senderAsyncClient": target.getConfiguration().setSenderAsyncClient(property(camelContext, com.azure.messaging.servicebus.ServiceBusSenderAsyncClient.class, value)); return true;
+ case "servicebusreceivemode":
+ case "serviceBusReceiveMode": target.getConfiguration().setServiceBusReceiveMode(property(camelContext, com.azure.messaging.servicebus.models.ServiceBusReceiveMode.class, value)); return true;
+ case "servicebustransactioncontext":
+ case "serviceBusTransactionContext": target.getConfiguration().setServiceBusTransactionContext(property(camelContext, com.azure.messaging.servicebus.ServiceBusTransactionContext.class, value)); return true;
+ case "servicebustype":
+ case "serviceBusType": target.getConfiguration().setServiceBusType(property(camelContext, org.apache.camel.component.azure.servicebus.ServiceBusType.class, value)); return true;
+ case "subqueue":
+ case "subQueue": target.getConfiguration().setSubQueue(property(camelContext, com.azure.messaging.servicebus.models.SubQueue.class, value)); return true;
+ case "subscriptionname":
+ case "subscriptionName": target.getConfiguration().setSubscriptionName(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public String[] getAutowiredNames() {
+ return new String[]{"receiverAsyncClient","senderAsyncClient"};
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amqpretryoptions":
+ case "amqpRetryOptions": return com.azure.core.amqp.AmqpRetryOptions.class;
+ case "amqptransporttype":
+ case "amqpTransportType": return com.azure.core.amqp.AmqpTransportType.class;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return boolean.class;
+ case "clientoptions":
+ case "clientOptions": return com.azure.core.util.ClientOptions.class;
+ case "connectionstring":
+ case "connectionString": return java.lang.String.class;
+ case "consumeroperation":
+ case "consumerOperation": return org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition.class;
+ case "disableautocomplete":
+ case "disableAutoComplete": return boolean.class;
+ case "exceptionhandler":
+ case "exceptionHandler": return org.apache.camel.spi.ExceptionHandler.class;
+ case "exchangepattern":
+ case "exchangePattern": return org.apache.camel.ExchangePattern.class;
+ case "lazystartproducer":
+ case "lazyStartProducer": return boolean.class;
+ case "maxautolockrenewduration":
+ case "maxAutoLockRenewDuration": return java.time.Duration.class;
+ case "peeknummaxmessages":
+ case "peekNumMaxMessages": return java.lang.Integer.class;
+ case "prefetchcount":
+ case "prefetchCount": return int.class;
+ case "produceroperation":
+ case "producerOperation": return org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition.class;
+ case "proxyoptions":
+ case "proxyOptions": return com.azure.core.amqp.ProxyOptions.class;
+ case "receiverasyncclient":
+ case "receiverAsyncClient": return com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient.class;
+ case "scheduledenqueuetime":
+ case "scheduledEnqueueTime": return java.time.OffsetDateTime.class;
+ case "senderasyncclient":
+ case "senderAsyncClient": return com.azure.messaging.servicebus.ServiceBusSenderAsyncClient.class;
+ case "servicebusreceivemode":
+ case "serviceBusReceiveMode": return com.azure.messaging.servicebus.models.ServiceBusReceiveMode.class;
+ case "servicebustransactioncontext":
+ case "serviceBusTransactionContext": return com.azure.messaging.servicebus.ServiceBusTransactionContext.class;
+ case "servicebustype":
+ case "serviceBusType": return org.apache.camel.component.azure.servicebus.ServiceBusType.class;
+ case "subqueue":
+ case "subQueue": return com.azure.messaging.servicebus.models.SubQueue.class;
+ case "subscriptionname":
+ case "subscriptionName": return java.lang.String.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ ServiceBusEndpoint target = (ServiceBusEndpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amqpretryoptions":
+ case "amqpRetryOptions": return target.getConfiguration().getAmqpRetryOptions();
+ case "amqptransporttype":
+ case "amqpTransportType": return target.getConfiguration().getAmqpTransportType();
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": return target.isBridgeErrorHandler();
+ case "clientoptions":
+ case "clientOptions": return target.getConfiguration().getClientOptions();
+ case "connectionstring":
+ case "connectionString": return target.getConfiguration().getConnectionString();
+ case "consumeroperation":
+ case "consumerOperation": return target.getConfiguration().getConsumerOperation();
+ case "disableautocomplete":
+ case "disableAutoComplete": return target.getConfiguration().isDisableAutoComplete();
+ case "exceptionhandler":
+ case "exceptionHandler": return target.getExceptionHandler();
+ case "exchangepattern":
+ case "exchangePattern": return target.getExchangePattern();
+ case "lazystartproducer":
+ case "lazyStartProducer": return target.isLazyStartProducer();
+ case "maxautolockrenewduration":
+ case "maxAutoLockRenewDuration": return target.getConfiguration().getMaxAutoLockRenewDuration();
+ case "peeknummaxmessages":
+ case "peekNumMaxMessages": return target.getConfiguration().getPeekNumMaxMessages();
+ case "prefetchcount":
+ case "prefetchCount": return target.getConfiguration().getPrefetchCount();
+ case "produceroperation":
+ case "producerOperation": return target.getConfiguration().getProducerOperation();
+ case "proxyoptions":
+ case "proxyOptions": return target.getConfiguration().getProxyOptions();
+ case "receiverasyncclient":
+ case "receiverAsyncClient": return target.getConfiguration().getReceiverAsyncClient();
+ case "scheduledenqueuetime":
+ case "scheduledEnqueueTime": return target.getConfiguration().getScheduledEnqueueTime();
+ case "senderasyncclient":
+ case "senderAsyncClient": return target.getConfiguration().getSenderAsyncClient();
+ case "servicebusreceivemode":
+ case "serviceBusReceiveMode": return target.getConfiguration().getServiceBusReceiveMode();
+ case "servicebustransactioncontext":
+ case "serviceBusTransactionContext": return target.getConfiguration().getServiceBusTransactionContext();
+ case "servicebustype":
+ case "serviceBusType": return target.getConfiguration().getServiceBusType();
+ case "subqueue":
+ case "subQueue": return target.getConfiguration().getSubQueue();
+ case "subscriptionname":
+ case "subscriptionName": return target.getConfiguration().getSubscriptionName();
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointUriFactory.java b/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointUriFactory.java
new file mode 100644
index 0000000..b4df751
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointUriFactory.java
@@ -0,0 +1,86 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.azure.servicebus;
+
+import java.net.URISyntaxException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.spi.EndpointUriFactory;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+public class ServiceBusEndpointUriFactory extends org.apache.camel.support.component.EndpointUriFactorySupport implements EndpointUriFactory {
+
+ private static final String BASE = ":queueNameOrTopicName";
+
+ private static final Set<String> PROPERTY_NAMES;
+ private static final Set<String> SECRET_PROPERTY_NAMES;
+ static {
+ Set<String> props = new HashSet<>(24);
+ props.add("clientOptions");
+ props.add("serviceBusType");
+ props.add("connectionString");
+ props.add("prefetchCount");
+ props.add("receiverAsyncClient");
+ props.add("serviceBusTransactionContext");
+ props.add("scheduledEnqueueTime");
+ props.add("exchangePattern");
+ props.add("producerOperation");
+ props.add("amqpTransportType");
+ props.add("consumerOperation");
+ props.add("serviceBusReceiveMode");
+ props.add("senderAsyncClient");
+ props.add("lazyStartProducer");
+ props.add("bridgeErrorHandler");
+ props.add("peekNumMaxMessages");
+ props.add("subQueue");
+ props.add("topicOrQueueName");
+ props.add("subscriptionName");
+ props.add("amqpRetryOptions");
+ props.add("proxyOptions");
+ props.add("disableAutoComplete");
+ props.add("exceptionHandler");
+ props.add("maxAutoLockRenewDuration");
+ PROPERTY_NAMES = Collections.unmodifiableSet(props);
+ Set<String> secretProps = new HashSet<>(1);
+ secretProps.add("connectionString");
+ SECRET_PROPERTY_NAMES = Collections.unmodifiableSet(secretProps);
+ }
+
+ @Override
+ public boolean isEnabled(String scheme) {
+ return "azure-servicebus".equals(scheme);
+ }
+
+ @Override
+ public String buildUri(String scheme, Map<String, Object> properties, boolean encode) throws URISyntaxException {
+ String syntax = scheme + BASE;
+ String uri = syntax;
+
+ Map<String, Object> copy = new HashMap<>(properties);
+
+ uri = buildPathParameter(syntax, uri, "topicOrQueueName", null, false, copy);
+ uri = buildQueryParameters(uri, copy, encode);
+ return uri;
+ }
+
+ @Override
+ public Set<String> propertyNames() {
+ return PROPERTY_NAMES;
+ }
+
+ @Override
+ public Set<String> secretPropertyNames() {
+ return SECRET_PROPERTY_NAMES;
+ }
+
+ @Override
+ public boolean isLenientProperties() {
+ return false;
+ }
+}
+
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusTypeConverterLoader.java b/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusTypeConverterLoader.java
new file mode 100644
index 0000000..1e94093
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/java/org/apache/camel/component/azure/servicebus/ServiceBusTypeConverterLoader.java
@@ -0,0 +1,54 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.azure.servicebus;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.DeferredContextBinding;
+import org.apache.camel.Exchange;
+import org.apache.camel.TypeConversionException;
+import org.apache.camel.TypeConverterLoaderException;
+import org.apache.camel.spi.TypeConverterLoader;
+import org.apache.camel.spi.TypeConverterRegistry;
+import org.apache.camel.support.SimpleTypeConverter;
+import org.apache.camel.support.TypeConverterSupport;
+import org.apache.camel.util.DoubleMap;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+@DeferredContextBinding
+public final class ServiceBusTypeConverterLoader implements TypeConverterLoader, CamelContextAware {
+
+ private CamelContext camelContext;
+
+ public ServiceBusTypeConverterLoader() {
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ @Override
+ public void load(TypeConverterRegistry registry) throws TypeConverterLoaderException {
+ registerConverters(registry);
+ }
+
+ private void registerConverters(TypeConverterRegistry registry) {
+ addTypeConverter(registry, com.azure.core.util.BinaryData.class, java.lang.String.class, false,
+ (type, exchange, value) -> org.apache.camel.component.azure.servicebus.ServiceBusTypeConverter.toBinaryData((java.lang.String) value));
+ addTypeConverter(registry, java.lang.String.class, com.azure.core.util.BinaryData.class, false,
+ (type, exchange, value) -> org.apache.camel.component.azure.servicebus.ServiceBusTypeConverter.toString((com.azure.core.util.BinaryData) value));
+ }
+
+ private static void addTypeConverter(TypeConverterRegistry registry, Class<?> toType, Class<?> fromType, boolean allowNull, SimpleTypeConverter.ConversionMethod method) {
+ registry.addTypeConverter(toType, fromType, new SimpleTypeConverter(allowNull, method));
+ }
+
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader
new file mode 100644
index 0000000..8feda5b
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+org.apache.camel.component.azure.servicebus.ServiceBusTypeConverterLoader
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/component.properties b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/component.properties
new file mode 100644
index 0000000..293e069
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/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-servicebus
+groupId=org.apache.camel
+artifactId=camel-azure-servicebus
+version=3.12.0-SNAPSHOT
+projectName=Camel :: Azure ServiceBus
+projectDescription=Camel Azure ServiceBus Component
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/component/azure-servicebus b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/component/azure-servicebus
new file mode 100644
index 0000000..dd59f48
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/component/azure-servicebus
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.azure.servicebus.ServiceBusComponent
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-servicebus-component b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-servicebus-component
new file mode 100644
index 0000000..3cc0bc4
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-servicebus-component
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.azure.servicebus.ServiceBusComponentConfigurer
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-servicebus-endpoint b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-servicebus-endpoint
new file mode 100644
index 0000000..ef75c17
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/configurer/azure-servicebus-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.azure.servicebus.ServiceBusEndpointConfigurer
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/urifactory/azure-servicebus-endpoint b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/urifactory/azure-servicebus-endpoint
new file mode 100644
index 0000000..3d53f23
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/services/org/apache/camel/urifactory/azure-servicebus-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.azure.servicebus.ServiceBusEndpointUriFactory
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/org/apache/camel/component/azure/servicebus/azure-servicebus.json b/components/camel-azure/camel-azure-servicebus/src/generated/resources/org/apache/camel/component/azure/servicebus/azure-servicebus.json
new file mode 100644
index 0000000..1a96c0e
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/resources/org/apache/camel/component/azure/servicebus/azure-servicebus.json
@@ -0,0 +1,75 @@
+{
+ "component": {
+ "kind": "component",
+ "name": "azure-servicebus",
+ "title": "Azure ServiceBus",
+ "description": "The azure-servicebus component that integrates Azure ServiceBus. Azure ServiceBus is a fully managed enterprise integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and secure platform for asynchronous transfer of data and state. Data is transferred between different applications and services using messages.",
+ "deprecated": false,
+ "firstVersion": "3.12.0",
+ "label": "cloud,messaging",
+ "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusComponent",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-azure-servicebus",
+ "version": "3.12.0-SNAPSHOT",
+ "scheme": "azure-servicebus",
+ "extendsScheme": "",
+ "syntax": "azure-servicebus:queueNameOrTopicName",
+ "async": false,
+ "api": 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, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the retry options for Service Bus clients. If not specified, the defaul [...]
+ "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, "autowired": false, "secret": false, "defaultValue": "AMQP", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the tr [...]
+ "clientOptions": { "kind": "property", "displayName": "Client Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.util.ClientOptions", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the ClientOptions to be sent from the client built from this builder, enabling cu [...]
+ "configuration": { "kind": "property", "displayName": "Configuration", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "deprecated": false, "autowired": false, "secret": false, "description": "The component configurations" },
+ "proxyOptions": { "kind": "property", "displayName": "Proxy Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.ProxyOptions", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the proxy configuration to use for ServiceBusSenderAsyncClient. When a proxy is conf [...]
+ "serviceBusType": { "kind": "property", "displayName": "Service Bus Type", "group": "common", "label": "common", "required": true, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusType", "enum": [ "queue", "topic" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "queue", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration [...]
+ "bridgeErrorHandler": { "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": 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 me [...]
+ "consumerOperation": { "kind": "property", "displayName": "Consumer Operation", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition", "enum": [ "receiveMessages", "peekMessages" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "receiveMessages", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", [...]
+ "disableAutoComplete": { "kind": "property", "displayName": "Disable Auto Complete", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Disables auto-complete and auto-abandon of received messages. By defaul [...]
+ "maxAutoLockRenewDuration": { "kind": "property", "displayName": "Max Auto Lock Renew Duration", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "5min", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the amount of time to continue auto-renewin [...]
+ "peekNumMaxMessages": { "kind": "property", "displayName": "Peek Num Max Messages", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Set the max number of messages to be peeked during the peek operation." },
+ "prefetchCount": { "kind": "property", "displayName": "Prefetch Count", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the prefetch count of the receiver. For both {link ServiceBusReceiveMode#PEEK_LOCK PEEK [...]
+ "receiverAsyncClient": { "kind": "property", "displayName": "Receiver Async Client", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the receiverAsy [...]
+ "serviceBusReceiveMode": { "kind": "property", "displayName": "Service Bus Receive Mode", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.models.ServiceBusReceiveMode", "enum": [ "PEEK_LOCK", "RECEIVE_AND_DELETE" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PEER_LOCK", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationFie [...]
+ "subQueue": { "kind": "property", "displayName": "Sub Queue", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.models.SubQueue", "enum": [ "NONE", "DEAD_LETTER_QUEUE", "TRANSFER_DEAD_LETTER_QUEUE" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the [...]
+ "subscriptionName": { "kind": "property", "displayName": "Subscription Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the name of the subscription in the topic to listen to. topicOrQueueName and serviceBusT [...]
+ "lazyStartProducer": { "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during star [...]
+ "producerOperation": { "kind": "property", "displayName": "Producer Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition", "enum": [ "sendMessages", "scheduleMessages" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "sendMessages", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", " [...]
+ "scheduledEnqueueTime": { "kind": "property", "displayName": "Scheduled Enqueue Time", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "java.time.OffsetDateTime", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets OffsetDateTime at which the message should appear in the Service Bus qu [...]
+ "senderAsyncClient": { "kind": "property", "displayName": "Sender Async Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusSenderAsyncClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets SenderAsyncClient to [...]
+ "serviceBusTransactionContext": { "kind": "property", "displayName": "Service Bus Transaction Context", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusTransactionContext", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Represents transaction i [...]
+ "autowiredEnabled": { "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which t [...]
+ "connectionString": { "kind": "property", "displayName": "Connection String", "group": "security", "label": "security", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the connection string for a Service Bus namespace or a specific Ser [...]
+ },
+ "properties": {
+ "topicOrQueueName": { "kind": "path", "displayName": "Topic Or Queue Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Selected topic name or the queue name, that is depending on serviceBusType config. For example if service [...]
+ "amqpRetryOptions": { "kind": "parameter", "displayName": "Amqp Retry Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.AmqpRetryOptions", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the retry options for Service Bus clients. If not specified, the defau [...]
+ "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, "autowired": false, "secret": false, "defaultValue": "AMQP", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the t [...]
+ "clientOptions": { "kind": "parameter", "displayName": "Client Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.util.ClientOptions", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the ClientOptions to be sent from the client built from this builder, enabling c [...]
+ "proxyOptions": { "kind": "parameter", "displayName": "Proxy Options", "group": "common", "label": "common", "required": false, "type": "object", "javaType": "com.azure.core.amqp.ProxyOptions", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the proxy configuration to use for ServiceBusSenderAsyncClient. When a proxy is con [...]
+ "serviceBusType": { "kind": "parameter", "displayName": "Service Bus Type", "group": "common", "label": "common", "required": true, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusType", "enum": [ "queue", "topic" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "queue", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuratio [...]
+ "bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": 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 m [...]
+ "consumerOperation": { "kind": "parameter", "displayName": "Consumer Operation", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition", "enum": [ "receiveMessages", "peekMessages" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "receiveMessages", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration" [...]
+ "disableAutoComplete": { "kind": "parameter", "displayName": "Disable Auto Complete", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Disables auto-complete and auto-abandon of received messages. By defau [...]
+ "maxAutoLockRenewDuration": { "kind": "parameter", "displayName": "Max Auto Lock Renew Duration", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "java.time.Duration", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "5min", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the amount of time to continue auto-renewi [...]
+ "peekNumMaxMessages": { "kind": "parameter", "displayName": "Peek Num Max Messages", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Set the max number of messages to be peeked during the peek operation." },
+ "prefetchCount": { "kind": "parameter", "displayName": "Prefetch Count", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the prefetch count of the receiver. For both {link ServiceBusReceiveMode#PEEK_LOCK PEE [...]
+ "receiverAsyncClient": { "kind": "parameter", "displayName": "Receiver Async Client", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the receiverAs [...]
+ "serviceBusReceiveMode": { "kind": "parameter", "displayName": "Service Bus Receive Mode", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.models.ServiceBusReceiveMode", "enum": [ "PEEK_LOCK", "RECEIVE_AND_DELETE" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "PEER_LOCK", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationFi [...]
+ "subQueue": { "kind": "parameter", "displayName": "Sub Queue", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.models.SubQueue", "enum": [ "NONE", "DEAD_LETTER_QUEUE", "TRANSFER_DEAD_LETTER_QUEUE" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets th [...]
+ "subscriptionName": { "kind": "parameter", "displayName": "Subscription Name", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the name of the subscription in the topic to listen to. topicOrQueueName and serviceBus [...]
+ "exceptionHandler": { "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the con [...]
+ "exchangePattern": { "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut", "InOptionalOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
+ "lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during sta [...]
+ "producerOperation": { "kind": "parameter", "displayName": "Producer Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition", "enum": [ "sendMessages", "scheduleMessages" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "sendMessages", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", [...]
+ "scheduledEnqueueTime": { "kind": "parameter", "displayName": "Scheduled Enqueue Time", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "java.time.OffsetDateTime", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets OffsetDateTime at which the message should appear in the Service Bus q [...]
+ "senderAsyncClient": { "kind": "parameter", "displayName": "Sender Async Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusSenderAsyncClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets SenderAsyncClient to [...]
+ "serviceBusTransactionContext": { "kind": "parameter", "displayName": "Service Bus Transaction Context", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusTransactionContext", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Represents transaction [...]
+ "connectionString": { "kind": "parameter", "displayName": "Connection String", "group": "security", "label": "security", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the connection string for a Service Bus namespace or a specific Se [...]
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/docs/azure-servicebus-component.adoc b/components/camel-azure/camel-azure-servicebus/src/main/docs/azure-servicebus-component.adoc
new file mode 100644
index 0000000..cc225e8
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/docs/azure-servicebus-component.adoc
@@ -0,0 +1,255 @@
+[[azure-servicebus-component]]
+= Azure ServiceBus Component
+:docTitle: Azure ServiceBus
+:artifactId: camel-azure-servicebus
+:description: The azure-servicebus component that integrates Azure ServiceBus. Azure ServiceBus is a fully managed enterprise integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and secure platform for asynchronous transfer of data and state. Data is transferred between different applications and services using messages.
+:since: 3.12
+:supportLevel: Preview
+:component-header: Both producer and consumer are supported
+include::{cq-version}@camel-quarkus:ROOT:partial$reference/components/azure-servicebus.adoc[opts=optional]
+//Manually maintained attributes
+:group: Azure
+
+*Since Camel {since}*
+
+*{component-header}*
+
+The azure-servicebus component that integrates https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview[Azure ServiceBus]. Azure ServiceBus is a fully managed enterprise
+integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and secure platform for asynchronous transfer of data and state. Data is transferred between different applications and
+ services using messages.
+
+Prerequisites
+
+You must have a valid Windows Azure Storage account. More information is available at
+https://docs.microsoft.com/azure/[Azure Documentation Portal].
+
+[source,xml]
+------------------------------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-azure-servicebus</artifactId>
+ <version>x.x.x</version>
+ <!-- use the same version as your Camel core version -->
+</dependency>
+------------------------------------------------------------
+
+== URI Options
+// endpoint options: START
+The Azure ServiceBus endpoint is configured using URI syntax:
+
+----
+azure-servicebus:queueNameOrTopicName
+----
+
+with the following path and query parameters:
+
+=== Path Parameters (1 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *topicOrQueueName* | Selected topic name or the queue name, that is depending on serviceBusType config. For example if serviceBusType=queue, then this will be the queue name and if serviceBusType=topic, this will be the topic name. | | String
+|===
+
+
+=== Query Parameters (23 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *amqpRetryOptions* (common) | Sets the retry options for Service Bus clients. If not specified, the default retry options are used. | | AmqpRetryOptions
+| *amqpTransportType* (common) | Sets the transport type by which all the communication with Azure Service Bus occurs. Default value is AmqpTransportType#AMQP. There are 2 enums and the value can be one of: Amqp, AmqpWebSockets | AMQP | AmqpTransportType
+| *clientOptions* (common) | Sets the ClientOptions to be sent from the client built from this builder, enabling customization of certain properties, as well as support the addition of custom header information. Refer to the ClientOptions documentation for more information. | | ClientOptions
+| *proxyOptions* (common) | Sets the proxy configuration to use for ServiceBusSenderAsyncClient. When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must be used for the transport type. | | ProxyOptions
+| *serviceBusType* (common) | *Required* The service bus type of connection to execute. Queue is for typical queue option and topic for subscription based model. There are 2 enums and the value can be one of: queue, topic | queue | ServiceBusType
+| *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
+| *consumerOperation* (consumer) | Sets the desired operation to be used in the consumer. There are 2 enums and the value can be one of: receiveMessages, peekMessages | receiveMessages | ServiceBusConsumerOperationDefinition
+| *disableAutoComplete* (consumer) | Disables auto-complete and auto-abandon of received messages. By default, a successfully processed message is {link ServiceBusReceiverAsyncClient#complete(ServiceBusReceivedMessage) completed}. If an error happens when the message is processed, it is {link ServiceBusReceiverAsyncClient#abandon(ServiceBusReceivedMessage) abandoned}. | false | boolean
+| *maxAutoLockRenewDuration* (consumer) | Sets the amount of time to continue auto-renewing the lock. Setting Duration#ZERO or null disables auto-renewal. For {link ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} mode, auto-renewal is disabled. | 5min | Duration
+| *peekNumMaxMessages* (consumer) | Set the max number of messages to be peeked during the peek operation. | | Integer
+| *prefetchCount* (consumer) | Sets the prefetch count of the receiver. For both {link ServiceBusReceiveMode#PEEK_LOCK PEEK_LOCK} and {link ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} modes the default value is 1. Prefetch speeds up the message flow by aiming to have a message readily available for local retrieval when and before the application asks for one using ServiceBusReceiverAsyncClient#receiveMessages(). Setting a non-zero value will prefetch that number of messa [...]
+| *receiverAsyncClient* (consumer) | *Autowired* Sets the receiverAsyncClient in order to consume messages in the Consumer | | ServiceBusReceiverAsyncClient
+| *serviceBusReceiveMode* (consumer) | Sets the receive mode for the receiver. There are 2 enums and the value can be one of: PEEK_LOCK, RECEIVE_AND_DELETE | PEER_LOCK | ServiceBusReceiveMode
+| *subQueue* (consumer) | Sets the type of the SubQueue to connect to. There are 3 enums and the value can be one of: NONE, DEAD_LETTER_QUEUE, TRANSFER_DEAD_LETTER_QUEUE | | SubQueue
+| *subscriptionName* (consumer) | Sets the name of the subscription in the topic to listen to. topicOrQueueName and serviceBusType=topic must also be set. | | String
+| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
+| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. There are 3 enums and the value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
+| *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 [...]
+| *producerOperation* (producer) | Sets the desired operation to be used in the producer. There are 2 enums and the value can be one of: sendMessages, scheduleMessages | sendMessages | ServiceBusProducerOperationDefinition
+| *scheduledEnqueueTime* (producer) | Sets OffsetDateTime at which the message should appear in the Service Bus queue or topic. | | OffsetDateTime
+| *senderAsyncClient* (producer) | *Autowired* Sets SenderAsyncClient to be used in the producer. | | ServiceBusSenderAsyncClient
+| *serviceBusTransactionContext* (producer) | Represents transaction in service. This object just contains transaction id. | | ServiceBusTransactionContext
+| *connectionString* (security) | *Required* Sets the connection string for a Service Bus namespace or a specific Service Bus resource. | | String
+|===
+// endpoint options: END
+
+== Component Options
+// component options: START
+The Azure ServiceBus component supports 23 options, which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *amqpRetryOptions* (common) | Sets the retry options for Service Bus clients. If not specified, the default retry options are used. | | AmqpRetryOptions
+| *amqpTransportType* (common) | Sets the transport type by which all the communication with Azure Service Bus occurs. Default value is AmqpTransportType#AMQP. There are 2 enums and the value can be one of: Amqp, AmqpWebSockets | AMQP | AmqpTransportType
+| *clientOptions* (common) | Sets the ClientOptions to be sent from the client built from this builder, enabling customization of certain properties, as well as support the addition of custom header information. Refer to the ClientOptions documentation for more information. | | ClientOptions
+| *configuration* (common) | The component configurations | | ServiceBusConfiguration
+| *proxyOptions* (common) | Sets the proxy configuration to use for ServiceBusSenderAsyncClient. When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must be used for the transport type. | | ProxyOptions
+| *serviceBusType* (common) | *Required* The service bus type of connection to execute. Queue is for typical queue option and topic for subscription based model. There are 2 enums and the value can be one of: queue, topic | queue | ServiceBusType
+| *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
+| *consumerOperation* (consumer) | Sets the desired operation to be used in the consumer. There are 2 enums and the value can be one of: receiveMessages, peekMessages | receiveMessages | ServiceBusConsumerOperationDefinition
+| *disableAutoComplete* (consumer) | Disables auto-complete and auto-abandon of received messages. By default, a successfully processed message is {link ServiceBusReceiverAsyncClient#complete(ServiceBusReceivedMessage) completed}. If an error happens when the message is processed, it is {link ServiceBusReceiverAsyncClient#abandon(ServiceBusReceivedMessage) abandoned}. | false | boolean
+| *maxAutoLockRenewDuration* (consumer) | Sets the amount of time to continue auto-renewing the lock. Setting Duration#ZERO or null disables auto-renewal. For {link ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} mode, auto-renewal is disabled. | 5min | Duration
+| *peekNumMaxMessages* (consumer) | Set the max number of messages to be peeked during the peek operation. | | Integer
+| *prefetchCount* (consumer) | Sets the prefetch count of the receiver. For both {link ServiceBusReceiveMode#PEEK_LOCK PEEK_LOCK} and {link ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} modes the default value is 1. Prefetch speeds up the message flow by aiming to have a message readily available for local retrieval when and before the application asks for one using ServiceBusReceiverAsyncClient#receiveMessages(). Setting a non-zero value will prefetch that number of messa [...]
+| *receiverAsyncClient* (consumer) | *Autowired* Sets the receiverAsyncClient in order to consume messages in the Consumer | | ServiceBusReceiverAsyncClient
+| *serviceBusReceiveMode* (consumer) | Sets the receive mode for the receiver. There are 2 enums and the value can be one of: PEEK_LOCK, RECEIVE_AND_DELETE | PEER_LOCK | ServiceBusReceiveMode
+| *subQueue* (consumer) | Sets the type of the SubQueue to connect to. There are 3 enums and the value can be one of: NONE, DEAD_LETTER_QUEUE, TRANSFER_DEAD_LETTER_QUEUE | | SubQueue
+| *subscriptionName* (consumer) | Sets the name of the subscription in the topic to listen to. topicOrQueueName and serviceBusType=topic must also be set. | | String
+| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
+| *producerOperation* (producer) | Sets the desired operation to be used in the producer. There are 2 enums and the value can be one of: sendMessages, scheduleMessages | sendMessages | ServiceBusProducerOperationDefinition
+| *scheduledEnqueueTime* (producer) | Sets OffsetDateTime at which the message should appear in the Service Bus queue or topic. | | OffsetDateTime
+| *senderAsyncClient* (producer) | *Autowired* Sets SenderAsyncClient to be used in the producer. | | ServiceBusSenderAsyncClient
+| *serviceBusTransactionContext* (producer) | Represents transaction in service. This object just contains transaction id. | | ServiceBusTransactionContext
+| *autowiredEnabled* (advanced) | Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc. | true | boolean
+| *connectionString* (security) | *Required* Sets the connection string for a Service Bus namespace or a specific Service Bus resource. | | String
+|===
+// component options: END
+
+== 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
+
+|`CamelAzureServiceBusScheduledEnqueueTime`| `ServiceBusConstants.SCHEDULED_ENQUEUE_TIME`|`OffsetDateTime`| Overrides the OffsetDateTime at which the message should appear in the Service Bus queue or topic.
+|`CamelAzureServiceBusServiceBusTransactionContext`| `ServiceBusConstants.SERVICE_BUS_TRANSACTION_CONTEXT`|`ServiceBusTransactionContext`| Overrides the transaction in service. This object just contains transaction id.
+|`CamelAzureServiceBusProducerOperation`| `ServiceBusConstants.PRODUCER_OPERATION`|`ServiceBusProducerOperationDefinition`| Overrides the desired operation to be used in the producer.
+|=======================================================================
+
+
+=== Message headers set by the component consumer
+[width="100%",cols="10%,10%,10%,70%",options="header",]
+|=======================================================================
+|Header |Variable Name |Type |Description
+
+|`CamelAzureServiceBusApplicationProperties`| `ServiceBusConstants.APPLICATION_PROPERTIES`|`Map<String, Object>`| Gets the set of free-form {@link ServiceBusReceivedMessage} properties which may be used for passing metadata associated with the {@link ServiceBusReceivedMessage} during Service Bus operations.
+|`CamelAzureServiceBusContentType`| `ServiceBusConstants.CONTENT_TYPE`|`String`| Gets the content type of the message.
+|`CamelAzureServiceBusMessageId`| `ServiceBusConstants.MESSAGE_ID`|`String`| Gets the identifier for the message.
+|`CamelAzureServiceBusCorrelationId`| `ServiceBusConstants.CORRELATION_ID`|`String`| Gets a correlation identifier.
+|`CamelAzureServiceBusDeadLetterErrorDescription`| `ServiceBusConstants.DEAD_LETTER_ERROR_DESCRIPTION`|`String`| Gets the description for a message that has been dead-lettered.
+|`CamelAzureServiceBusDeadLetterReason`| `ServiceBusConstants.DEAD_LETTER_REASON`|`String`| Gets the reason a message was dead-lettered.
+|`CamelAzureServiceBusDeadLetterSource`| `ServiceBusConstants.DEAD_LETTER_SOURCE`|`String`| Gets the name of the queue or subscription that this message was enqueued on, before it was dead-lettered.
+|`CamelAzureServiceBusDeliveryCount`| `ServiceBusConstants.DELIVERY_COUNT`|`long`| Gets the number of the times this message was delivered to clients.
+|`CamelAzureServiceBusScheduledEnqueueTime`| `ServiceBusConstants.SCHEDULED_ENQUEUE_TIME`|`OffsetDateTime`| Gets the scheduled enqueue time of this message.
+|`CamelAzureServiceBusEnqueuedSequenceNumber`| `ServiceBusConstants.ENQUEUED_SEQUENCE_NUMBER`|`long`| Gets the enqueued sequence number assigned to a message by Service Bus.
+|`CamelAzureServiceBusEnqueuedTime`| `ServiceBusConstants.ENQUEUED_TIME`|`OffsetDateTime`| Gets the datetime at which this message was enqueued in Azure Service Bus.
+|`CamelAzureServiceBusExpiresAt`| `ServiceBusConstants.EXPIRES_AT`|`OffsetDateTime`| Gets the datetime at which this message will expire.
+|`CamelAzureServiceBusLockToken`| `ServiceBusConstants.LOCK_TOKEN`|`String`| Gets the lock token for the current message.
+|`CamelAzureServiceBusLockedUntil`| `ServiceBusConstants.LOCKED_UNTIL`|`OffsetDateTime`| Gets the datetime at which the lock of this message expires.
+|`CamelAzureServiceBusPartitionKey`| `ServiceBusConstants.PARTITION_KEY`|`String`| Gets the partition key for sending a message to a partitioned entity.
+|`CamelAzureServiceBusRawAmqpMessage`| `ServiceBusConstants.RAW_AMQP_MESSAGE`|`AmqpAnnotatedMessage`| The representation of message as defined by AMQP protocol.
+|`CamelAzureServiceBusReplyTo`| `ServiceBusConstants.REPLY_TO`|`String`| Gets the address of an entity to send replies to.
+|`CamelAzureServiceBusReplyToSessionId`| `ServiceBusConstants.REPLY_TO_SESSION_ID`|`String`| Gets or sets a session identifier augmenting the {@link #getReplyTo() ReplyTo} address.
+|`CamelAzureServiceBusSequenceNumber`| `ServiceBusConstants.SEQUENCE_NUMBER`|`long`| Gets the unique number assigned to a message by Service Bus.
+|`CamelAzureServiceBusSessionId`| `ServiceBusConstants.SESSION_ID`|`String`| Gets the session id of the message.
+|`CamelAzureServiceBusSubject`| `ServiceBusConstants.SUBJECT`|`String`| Gets the subject for the message.
+|`CamelAzureServiceBusTimeToLive`| `ServiceBusConstants.TIME_TO_LIVE`|`Duration`| Gets the duration before this message expires.
+|`CamelAzureServiceBusTimeTo`| `ServiceBusConstants.TO`|`String`| Gets the "to" address.
+|=======================================================================
+
+
+=== Message Body
+In the producer, this component accepts message body of `String` type or `List<String>` to send batch messages.
+
+In the consumer, the returned message body will be of type `String.
+
+
+=== Azure ServiceBus Producer operations
+[width="100%",cols="10%,90%",options="header",]
+|===
+|Operation |Description
+|`sendMessages` | Sends a set of messages to a Service Bus queue or topic using a batched approach.
+|`scheduleMessages` | Sends a scheduled message to the Azure Service Bus entity this sender is connected to. A scheduled message is enqueued and made available to receivers only at the scheduled enqueue time.
+|===
+
+
+=== Azure ServiceBus Consumer operations
+[width="100%",cols="10%,90%",options="header",]
+|===
+|Operation |Description
+|`receiveMessages` | Receives an <b>infinite</b> stream of messages from the Service Bus entity.
+|`peekMessages` | Reads the next batch of active messages without changing the state of the receiver or the message source.
+|===
+
+
+==== Examples
+- `sendMessages`
+
+[source,java]
+--------------------------------------------------------------------------------
+from("direct:start")
+ .process(exchange -> {
+ final List<Object> inputBatch = new LinkedList<>();
+ inputBatch.add("test batch 1");
+ inputBatch.add("test batch 2");
+ inputBatch.add("test batch 3");
+ inputBatch.add(123456);
+
+ exchange.getIn().setBody(inputBatch);
+ })
+ .to("azure-servicebus:test//?connectionString=test")
+ .to("mock:result");
+--------------------------------------------------------------------------------
+
+
+- `scheduleMessages`
+
+[source,java]
+--------------------------------------------------------------------------------
+from("direct:start")
+ .process(exchange -> {
+ final List<Object> inputBatch = new LinkedList<>();
+ inputBatch.add("test batch 1");
+ inputBatch.add("test batch 2");
+ inputBatch.add("test batch 3");
+ inputBatch.add(123456);
+
+ exchange.getIn().setHeader(ServiceBusConstants.SCHEDULED_ENQUEUE_TIME, OffsetDateTime.now());
+ exchange.getIn().setBody(inputBatch);
+ })
+ .to("azure-servicebus:test//?connectionString=test&producerOperation=scheduleMessages")
+ .to("mock:result");
+--------------------------------------------------------------------------------
+
+- `receiveMessages`
+
+[source,java]
+--------------------------------------------------------------------------------
+from("azure-servicebus:test//?connectionString=test")
+ .log("${body}")
+ .to("mock:result");
+--------------------------------------------------------------------------------
+
+- `peekMessages`
+
+[source,java]
+--------------------------------------------------------------------------------
+from("azure-servicebus:test//?connectionString=test&consumerOperation=peekMessages&peekNumMaxMessages=3")
+ .log("${body}")
+ .to("mock:result");
+--------------------------------------------------------------------------------
+
+
+
+include::{page-component-version}@camel-spring-boot::page$azure-servicebus-starter.adoc[]
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusComponent.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusComponent.java
new file mode 100644
index 0000000..725d827
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusComponent.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.servicebus;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+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 ServiceBus component
+ */
+@Component("azure-servicebus")
+public class ServiceBusComponent extends DefaultComponent {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ServiceBusComponent.class);
+
+ @Metadata
+ private ServiceBusConfiguration configuration = new ServiceBusConfiguration();
+
+ public ServiceBusComponent() {
+ }
+
+ public ServiceBusComponent(final CamelContext context) {
+ super(context);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+
+ if (remaining == null || remaining.trim().length() == 0) {
+ throw new IllegalArgumentException("A queue or topic name must be specified.");
+ }
+
+ final ServiceBusConfiguration configuration
+ = this.configuration != null ? this.configuration.copy() : new ServiceBusConfiguration();
+
+ // set account or topic name
+ configuration.setTopicOrQueueName(remaining);
+
+ final ServiceBusEndpoint endpoint = new ServiceBusEndpoint(uri, this, configuration);
+ setProperties(endpoint, parameters);
+ validateConfigurations(configuration);
+
+ return endpoint;
+ }
+
+ /**
+ * The component configurations
+ */
+ public ServiceBusConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(ServiceBusConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ private void validateConfigurations(final ServiceBusConfiguration configuration) {
+ if (configuration.getReceiverAsyncClient() == null || configuration.getSenderAsyncClient() == null) {
+ if (ObjectHelper.isEmpty(configuration.getConnectionString())) {
+ throw new IllegalArgumentException("Azure ServiceBus ConnectionString must be specified.");
+ }
+ }
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConfiguration.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConfiguration.java
new file mode 100644
index 0000000..3d5ae81
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConfiguration.java
@@ -0,0 +1,333 @@
+/*
+ * 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.servicebus;
+
+import java.time.Duration;
+import java.time.OffsetDateTime;
+
+import com.azure.core.amqp.AmqpRetryOptions;
+import com.azure.core.amqp.AmqpTransportType;
+import com.azure.core.amqp.ProxyOptions;
+import com.azure.core.util.ClientOptions;
+import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
+import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
+import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+import com.azure.messaging.servicebus.ServiceBusTransactionContext;
+import com.azure.messaging.servicebus.models.ServiceBusReceiveMode;
+import com.azure.messaging.servicebus.models.SubQueue;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.spi.UriPath;
+
+@UriParams
+public class ServiceBusConfiguration implements Cloneable {
+
+ @UriPath
+ private String topicOrQueueName;
+ @UriParam(label = "security", secret = true)
+ @Metadata(required = true)
+ private String connectionString;
+ @UriParam(label = "common", defaultValue = "queue")
+ @Metadata(required = true)
+ private ServiceBusType serviceBusType = ServiceBusType.queue;
+ @UriParam(label = "common")
+ private ClientOptions clientOptions;
+ @UriParam(label = "common")
+ private ProxyOptions proxyOptions;
+ @UriParam(label = "common")
+ private AmqpRetryOptions amqpRetryOptions;
+ @UriParam(label = "common", defaultValue = "AMQP")
+ private AmqpTransportType amqpTransportType = AmqpTransportType.AMQP;
+ @UriParam(label = "consumer", defaultValue = "receiveMessages")
+ private ServiceBusConsumerOperationDefinition consumerOperation = ServiceBusConsumerOperationDefinition.receiveMessages;
+ @UriParam(label = "consumer")
+ @Metadata(autowired = true)
+ private ServiceBusReceiverAsyncClient receiverAsyncClient;
+ @UriParam(label = "consumer")
+ private String subscriptionName;
+ @UriParam(label = "consumer", defaultValue = "false")
+ private boolean disableAutoComplete;
+ @UriParam(label = "consumer", defaultValue = "PEER_LOCK")
+ private ServiceBusReceiveMode serviceBusReceiveMode = ServiceBusReceiveMode.PEEK_LOCK;
+ @UriParam(label = "consumer", defaultValue = "5min")
+ private Duration maxAutoLockRenewDuration = Duration.ofMinutes(5);
+ @UriParam(label = "consumer", defaultValue = "0")
+ private int prefetchCount;
+ @UriParam(label = "consumer")
+ private SubQueue subQueue;
+ @UriParam(label = "consumer")
+ private Integer peekNumMaxMessages;
+ @UriParam(label = "producer", defaultValue = "sendMessages")
+ private ServiceBusProducerOperationDefinition producerOperation = ServiceBusProducerOperationDefinition.sendMessages;
+ @UriParam(label = "producer")
+ @Metadata(autowired = true)
+ private ServiceBusSenderAsyncClient senderAsyncClient;
+ @UriParam(label = "producer")
+ private ServiceBusTransactionContext serviceBusTransactionContext;
+ @UriParam(label = "producer")
+ private OffsetDateTime scheduledEnqueueTime;
+
+ /**
+ * Selected topic name or the queue name, that is depending on serviceBusType config. For example if
+ * serviceBusType=queue, then this will be the queue name and if serviceBusType=topic, this will be the topic name.
+ */
+ public String getTopicOrQueueName() {
+ return topicOrQueueName;
+ }
+
+ public void setTopicOrQueueName(String topicOrQueueName) {
+ this.topicOrQueueName = topicOrQueueName;
+ }
+
+ /**
+ * The service bus type of connection to execute. Queue is for typical queue option and topic for subscription based
+ * model.
+ */
+ public ServiceBusType getServiceBusType() {
+ return serviceBusType;
+ }
+
+ public void setServiceBusType(ServiceBusType serviceBusType) {
+ this.serviceBusType = serviceBusType;
+ }
+
+ /**
+ * Sets the connection string for a Service Bus namespace or a specific Service Bus resource.
+ */
+ public String getConnectionString() {
+ return connectionString;
+ }
+
+ public void setConnectionString(String connectionString) {
+ this.connectionString = connectionString;
+ }
+
+ /**
+ * Sets the name of the subscription in the topic to listen to. topicOrQueueName and serviceBusType=topic must also
+ * be set.
+ */
+ public String getSubscriptionName() {
+ return subscriptionName;
+ }
+
+ public void setSubscriptionName(String subscriptionName) {
+ this.subscriptionName = subscriptionName;
+ }
+
+ /**
+ * Sets the {@link ClientOptions} to be sent from the client built from this builder, enabling customization of
+ * certain properties, as well as support the addition of custom header information. Refer to the
+ * {@link ClientOptions} documentation for more information.
+ */
+ public ClientOptions getClientOptions() {
+ return clientOptions;
+ }
+
+ public void setClientOptions(ClientOptions clientOptions) {
+ this.clientOptions = clientOptions;
+ }
+
+ /**
+ * Sets the proxy configuration to use for {@link ServiceBusSenderAsyncClient}. When a proxy is configured,
+ * {@link AmqpTransportType#AMQP_WEB_SOCKETS} must be used for the transport type.
+ */
+ public ProxyOptions getProxyOptions() {
+ return proxyOptions;
+ }
+
+ public void setProxyOptions(ProxyOptions proxyOptions) {
+ this.proxyOptions = proxyOptions;
+ }
+
+ /**
+ * Sets the retry options for Service Bus clients. If not specified, the default retry options are used.
+ */
+ public AmqpRetryOptions getAmqpRetryOptions() {
+ return amqpRetryOptions;
+ }
+
+ public void setAmqpRetryOptions(AmqpRetryOptions amqpRetryOptions) {
+ this.amqpRetryOptions = amqpRetryOptions;
+ }
+
+ /**
+ * Sets the transport type by which all the communication with Azure Service Bus occurs. Default value is
+ * {@link AmqpTransportType#AMQP}.
+ */
+ public AmqpTransportType getAmqpTransportType() {
+ return amqpTransportType;
+ }
+
+ public void setAmqpTransportType(AmqpTransportType amqpTransportType) {
+ this.amqpTransportType = amqpTransportType;
+ }
+
+ /**
+ * Sets the receiverAsyncClient in order to consume messages in the Consumer
+ */
+ public ServiceBusReceiverAsyncClient getReceiverAsyncClient() {
+ return receiverAsyncClient;
+ }
+
+ public void setReceiverAsyncClient(ServiceBusReceiverAsyncClient receiverAsyncClient) {
+ this.receiverAsyncClient = receiverAsyncClient;
+ }
+
+ /**
+ * Disables auto-complete and auto-abandon of received messages. By default, a successfully processed message is
+ * {@link ServiceBusReceiverAsyncClient#complete(ServiceBusReceivedMessage) completed}. If an error happens when the
+ * message is processed, it is {@link ServiceBusReceiverAsyncClient#abandon(ServiceBusReceivedMessage) abandoned}.
+ */
+ public boolean isDisableAutoComplete() {
+ return disableAutoComplete;
+ }
+
+ public void setDisableAutoComplete(boolean disableAutoComplete) {
+ this.disableAutoComplete = disableAutoComplete;
+ }
+
+ /**
+ * Sets the receive mode for the receiver.
+ */
+ public ServiceBusReceiveMode getServiceBusReceiveMode() {
+ return serviceBusReceiveMode;
+ }
+
+ public void setServiceBusReceiveMode(ServiceBusReceiveMode serviceBusReceiveMode) {
+ this.serviceBusReceiveMode = serviceBusReceiveMode;
+ }
+
+ /**
+ * Sets the amount of time to continue auto-renewing the lock. Setting {@link Duration#ZERO} or {@code null}
+ * disables auto-renewal. For {@link ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} mode, auto-renewal
+ * is disabled.
+ */
+ public Duration getMaxAutoLockRenewDuration() {
+ return maxAutoLockRenewDuration;
+ }
+
+ public void setMaxAutoLockRenewDuration(Duration maxAutoLockRenewDuration) {
+ this.maxAutoLockRenewDuration = maxAutoLockRenewDuration;
+ }
+
+ /**
+ * Sets the prefetch count of the receiver. For both {@link ServiceBusReceiveMode#PEEK_LOCK PEEK_LOCK} and
+ * {@link ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} modes the default value is 1.
+ *
+ * Prefetch speeds up the message flow by aiming to have a message readily available for local retrieval when and
+ * before the application asks for one using {@link ServiceBusReceiverAsyncClient#receiveMessages()}. Setting a
+ * non-zero value will prefetch that number of messages. Setting the value to zero turns prefetch off.
+ */
+ public int getPrefetchCount() {
+ return prefetchCount;
+ }
+
+ public void setPrefetchCount(int prefetchCount) {
+ this.prefetchCount = prefetchCount;
+ }
+
+ /**
+ * Sets the type of the {@link SubQueue} to connect to.
+ */
+ public SubQueue getSubQueue() {
+ return subQueue;
+ }
+
+ public void setSubQueue(SubQueue subQueue) {
+ this.subQueue = subQueue;
+ }
+
+ /**
+ * Sets SenderAsyncClient to be used in the producer.
+ */
+ public ServiceBusSenderAsyncClient getSenderAsyncClient() {
+ return senderAsyncClient;
+ }
+
+ public void setSenderAsyncClient(ServiceBusSenderAsyncClient senderAsyncClient) {
+ this.senderAsyncClient = senderAsyncClient;
+ }
+
+ /**
+ * Sets the desired operation to be used in the consumer
+ */
+ public ServiceBusConsumerOperationDefinition getConsumerOperation() {
+ return consumerOperation;
+ }
+
+ public void setConsumerOperation(ServiceBusConsumerOperationDefinition consumerOperation) {
+ this.consumerOperation = consumerOperation;
+ }
+
+ /**
+ * Sets the desired operation to be used in the producer
+ */
+ public ServiceBusProducerOperationDefinition getProducerOperation() {
+ return producerOperation;
+ }
+
+ public void setProducerOperation(ServiceBusProducerOperationDefinition producerOperation) {
+ this.producerOperation = producerOperation;
+ }
+
+ /**
+ * Represents transaction in service. This object just contains transaction id.
+ */
+ public ServiceBusTransactionContext getServiceBusTransactionContext() {
+ return serviceBusTransactionContext;
+ }
+
+ public void setServiceBusTransactionContext(ServiceBusTransactionContext serviceBusTransactionContext) {
+ this.serviceBusTransactionContext = serviceBusTransactionContext;
+ }
+
+ /**
+ * Sets OffsetDateTime at which the message should appear in the Service Bus queue or topic.
+ */
+ public OffsetDateTime getScheduledEnqueueTime() {
+ return scheduledEnqueueTime;
+ }
+
+ public void setScheduledEnqueueTime(OffsetDateTime scheduledEnqueueTime) {
+ this.scheduledEnqueueTime = scheduledEnqueueTime;
+ }
+
+ /**
+ * Set the max number of messages to be peeked during the peek operation.
+ */
+ public Integer getPeekNumMaxMessages() {
+ return peekNumMaxMessages;
+ }
+
+ public void setPeekNumMaxMessages(Integer peekNumMaxMessages) {
+ this.peekNumMaxMessages = peekNumMaxMessages;
+ }
+
+ // *************************************************
+ //
+ // *************************************************
+
+ public org.apache.camel.component.azure.servicebus.ServiceBusConfiguration copy() {
+ try {
+ return (org.apache.camel.component.azure.servicebus.ServiceBusConfiguration) super.clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConfigurationOptionsProxy.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConfigurationOptionsProxy.java
new file mode 100644
index 0000000..8d4be32
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConfigurationOptionsProxy.java
@@ -0,0 +1,69 @@
+/*
+ * 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.servicebus;
+
+import java.time.OffsetDateTime;
+import java.util.function.Supplier;
+
+import com.azure.messaging.servicebus.ServiceBusTransactionContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * A proxy class for {@link ServiceBusConfiguration} and {@link ServiceBusConstants}. Ideally this is responsible to
+ * obtain the correct configurations options either from configs or exchange headers
+ */
+public class ServiceBusConfigurationOptionsProxy {
+
+ private final ServiceBusConfiguration configuration;
+
+ public ServiceBusConfigurationOptionsProxy(final ServiceBusConfiguration 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 ServiceBusConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public ServiceBusTransactionContext getServiceBusTransactionContext(final Exchange exchange) {
+ return getOption(exchange, ServiceBusConstants.SERVICE_BUS_TRANSACTION_CONTEXT,
+ configuration::getServiceBusTransactionContext, ServiceBusTransactionContext.class);
+ }
+
+ public OffsetDateTime getScheduledEnqueueTime(final Exchange exchange) {
+ return getOption(exchange, ServiceBusConstants.SCHEDULED_ENQUEUE_TIME, configuration::getScheduledEnqueueTime,
+ OffsetDateTime.class);
+ }
+
+ public ServiceBusProducerOperationDefinition getServiceBusProducerOperationDefinition(final Exchange exchange) {
+ return getOption(exchange, ServiceBusConstants.PRODUCER_OPERATION, configuration::getProducerOperation,
+ ServiceBusProducerOperationDefinition.class);
+ }
+
+ 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/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConstants.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConstants.java
new file mode 100644
index 0000000..0e98675
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConstants.java
@@ -0,0 +1,56 @@
+/*
+ * 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.servicebus;
+
+public final class ServiceBusConstants {
+ private static final String HEADER_PREFIX = "CamelAzureServiceBus";
+ // common headers, set by consumer and evaluated by producer
+
+ // headers set by the consumer only
+ public static final String APPLICATION_PROPERTIES = HEADER_PREFIX + "ApplicationProperties";
+ public static final String CONTENT_TYPE = HEADER_PREFIX + "ContentType";
+ public static final String CORRELATION_ID = HEADER_PREFIX + "CorrelationId";
+ public static final String DEAD_LETTER_ERROR_DESCRIPTION = HEADER_PREFIX + "DeadLetterErrorDescription";
+ public static final String DEAD_LETTER_REASON = HEADER_PREFIX + "DeadLetterReason";
+ public static final String DEAD_LETTER_SOURCE = HEADER_PREFIX + "DeadLetterSource";
+ public static final String DELIVERY_COUNT = HEADER_PREFIX + "DeliveryCount";
+ public static final String ENQUEUED_SEQUENCE_NUMBER = HEADER_PREFIX + "EnqueuedSequenceNumber";
+ public static final String ENQUEUED_TIME = HEADER_PREFIX + "EnqueuedTime";
+ public static final String EXPIRES_AT = HEADER_PREFIX + "ExpiresAt";
+ public static final String LOCK_TOKEN = HEADER_PREFIX + "LockToken";
+ public static final String LOCKED_UNTIL = HEADER_PREFIX + "LockedUntil";
+ public static final String MESSAGE_ID = HEADER_PREFIX + "MessageId";
+ public static final String PARTITION_KEY = HEADER_PREFIX + "PartitionKey";
+ public static final String RAW_AMQP_MESSAGE = HEADER_PREFIX + "RawAmqpMessage";
+ public static final String REPLY_TO = HEADER_PREFIX + "ReplyTo";
+ public static final String REPLY_TO_SESSION_ID = HEADER_PREFIX + "ReplyToSessionId";
+ public static final String SEQUENCE_NUMBER = HEADER_PREFIX + "SequenceNumber";
+ public static final String SESSION_ID = HEADER_PREFIX + "SessionId";
+ public static final String SUBJECT = HEADER_PREFIX + "Subject";
+ public static final String TIME_TO_LIVE = HEADER_PREFIX + "TimeToLive";
+ public static final String TO = HEADER_PREFIX + "To";
+
+ // headers set by consumer and evaluated by producer
+ public static final String SCHEDULED_ENQUEUE_TIME = HEADER_PREFIX + "ScheduledEnqueueTime";
+
+ // headers evaluated by the producer
+ public static final String SERVICE_BUS_TRANSACTION_CONTEXT = HEADER_PREFIX + "ServiceBusTransactionContext";
+ public static final String PRODUCER_OPERATION = HEADER_PREFIX + "ProducerOperation";
+
+ private ServiceBusConstants() {
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java
new file mode 100644
index 0000000..af6e520
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumer.java
@@ -0,0 +1,213 @@
+/*
+ * 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.servicebus;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
+import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedExchange;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusClientFactory;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusReceiverAsyncClientWrapper;
+import org.apache.camel.component.azure.servicebus.operations.ServiceBusReceiverOperations;
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.SynchronizationAdapter;
+import org.apache.camel.util.ObjectHelper;
+
+public class ServiceBusConsumer extends DefaultConsumer {
+
+ private Synchronization onCompletion;
+ private ServiceBusReceiverAsyncClientWrapper clientWrapper;
+ private ServiceBusReceiverOperations operations;
+
+ private final Map<ServiceBusConsumerOperationDefinition, Runnable> operationsToExecute = new HashMap<>();
+
+ {
+ bind(ServiceBusConsumerOperationDefinition.peekMessages, this::peekMessages);
+ bind(ServiceBusConsumerOperationDefinition.receiveMessages, this::receiveMessages);
+ }
+
+ public ServiceBusConsumer(final ServiceBusEndpoint endpoint, final Processor processor) {
+ super(endpoint, processor);
+ }
+
+ @Override
+ protected void doInit() throws Exception {
+ super.doInit();
+ onCompletion = new ConsumerOnCompletion();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ // create the client
+ final ServiceBusReceiverAsyncClient client = getConfiguration().getReceiverAsyncClient() != null
+ ? getConfiguration().getReceiverAsyncClient()
+ : ServiceBusClientFactory.createServiceBusReceiverAsyncClient(getConfiguration());
+
+ // create the wrapper
+ clientWrapper = new ServiceBusReceiverAsyncClientWrapper(client);
+
+ // create the operations
+ operations = new ServiceBusReceiverOperations(clientWrapper);
+
+ // get the operation that we want to invoke
+ final ServiceBusConsumerOperationDefinition chosenOperation = getConfiguration().getConsumerOperation();
+
+ // invoke the operation and run it
+ invokeOperation(chosenOperation);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (clientWrapper != null) {
+ // shutdown the client
+ clientWrapper.close();
+ }
+
+ // shutdown camel consumer
+ super.doStop();
+ }
+
+ public ServiceBusConfiguration getConfiguration() {
+ return getEndpoint().getConfiguration();
+ }
+
+ @Override
+ public ServiceBusEndpoint getEndpoint() {
+ return (ServiceBusEndpoint) super.getEndpoint();
+ }
+
+ private void bind(final ServiceBusConsumerOperationDefinition operation, Runnable fn) {
+ operationsToExecute.put(operation, fn);
+ }
+
+ /**
+ * Entry method that selects the appropriate operation and executes it
+ */
+ private void invokeOperation(final ServiceBusConsumerOperationDefinition operation) {
+ final ServiceBusConsumerOperationDefinition operationsToInvoke;
+
+ if (ObjectHelper.isEmpty(operation)) {
+ operationsToInvoke = ServiceBusConsumerOperationDefinition.receiveMessages;
+ } else {
+ operationsToInvoke = operation;
+ }
+
+ final Runnable fnToInvoke = operationsToExecute.get(operationsToInvoke);
+
+ if (fnToInvoke != null) {
+ fnToInvoke.run();
+ } else {
+ throw new RuntimeException("Operation not supported. Value: " + operationsToInvoke);
+ }
+ }
+
+ private void receiveMessages() {
+ operations.receiveMessages()
+ .subscribe(this::onEventListener, this::onErrorListener, () -> {
+ });
+ }
+
+ private void peekMessages() {
+ operations.peekMessages(getConfiguration().getPeekNumMaxMessages())
+ .subscribe(this::onEventListener, this::onErrorListener, () -> {
+ });
+ }
+
+ private void onEventListener(final ServiceBusReceivedMessage message) {
+ final Exchange exchange = createServiceBusExchange(message);
+
+ // add exchange callback
+ exchange.adapt(ExtendedExchange.class).addOnCompletion(onCompletion);
+ // use default consumer callback
+ AsyncCallback cb = defaultConsumerCallback(exchange, true);
+ getAsyncProcessor().process(exchange, cb);
+ }
+
+ private Exchange createServiceBusExchange(final ServiceBusReceivedMessage receivedMessage) {
+ final Exchange exchange = createExchange(true);
+ final Message message = exchange.getIn();
+
+ // set body
+ message.setBody(receivedMessage.getBody());
+
+ // set headers
+ message.setHeader(ServiceBusConstants.APPLICATION_PROPERTIES, receivedMessage.getApplicationProperties());
+ message.setHeader(ServiceBusConstants.CONTENT_TYPE, receivedMessage.getContentType());
+ message.setHeader(ServiceBusConstants.MESSAGE_ID, receivedMessage.getMessageId());
+ message.setHeader(ServiceBusConstants.CORRELATION_ID, receivedMessage.getCorrelationId());
+ message.setHeader(ServiceBusConstants.DEAD_LETTER_ERROR_DESCRIPTION, receivedMessage.getDeadLetterErrorDescription());
+ message.setHeader(ServiceBusConstants.DEAD_LETTER_REASON, receivedMessage.getDeadLetterReason());
+ message.setHeader(ServiceBusConstants.DEAD_LETTER_SOURCE, receivedMessage.getDeadLetterSource());
+ message.setHeader(ServiceBusConstants.DELIVERY_COUNT, receivedMessage.getDeliveryCount());
+ message.setHeader(ServiceBusConstants.SCHEDULED_ENQUEUE_TIME, receivedMessage.getScheduledEnqueueTime());
+ message.setHeader(ServiceBusConstants.ENQUEUED_SEQUENCE_NUMBER, receivedMessage.getEnqueuedSequenceNumber());
+ message.setHeader(ServiceBusConstants.ENQUEUED_TIME, receivedMessage.getEnqueuedTime());
+ message.setHeader(ServiceBusConstants.EXPIRES_AT, receivedMessage.getExpiresAt());
+ message.setHeader(ServiceBusConstants.LOCK_TOKEN, receivedMessage.getLockToken());
+ message.setHeader(ServiceBusConstants.LOCKED_UNTIL, receivedMessage.getLockedUntil());
+ message.setHeader(ServiceBusConstants.PARTITION_KEY, receivedMessage.getPartitionKey());
+ message.setHeader(ServiceBusConstants.RAW_AMQP_MESSAGE, receivedMessage.getRawAmqpMessage());
+ message.setHeader(ServiceBusConstants.REPLY_TO, receivedMessage.getReplyTo());
+ message.setHeader(ServiceBusConstants.REPLY_TO_SESSION_ID, receivedMessage.getReplyToSessionId());
+ message.setHeader(ServiceBusConstants.SEQUENCE_NUMBER, receivedMessage.getSequenceNumber());
+ message.setHeader(ServiceBusConstants.SESSION_ID, receivedMessage.getSessionId());
+ message.setHeader(ServiceBusConstants.SUBJECT, receivedMessage.getSubject());
+ message.setHeader(ServiceBusConstants.TIME_TO_LIVE, receivedMessage.getTimeToLive());
+ message.setHeader(ServiceBusConstants.TO, receivedMessage.getTo());
+
+ return exchange;
+ }
+
+ private void onErrorListener(final Throwable errorContext) {
+ final Exchange exchange = createServiceBusExchange(errorContext);
+
+ // log exception if an exception occurred and was not handled
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange", exchange,
+ exchange.getException());
+ }
+ }
+
+ private Exchange createServiceBusExchange(final Throwable errorContext) {
+ final Exchange exchange = createExchange(true);
+
+ // set exception
+ exchange.setException(errorContext);
+
+ return exchange;
+ }
+
+ private class ConsumerOnCompletion extends SynchronizationAdapter {
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ final Exception cause = exchange.getException();
+ if (cause != null) {
+ getExceptionHandler().handleException("Error during processing exchange.", exchange, cause);
+ }
+ }
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerOperationDefinition.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerOperationDefinition.java
new file mode 100644
index 0000000..89946fa
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerOperationDefinition.java
@@ -0,0 +1,22 @@
+/*
+ * 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.servicebus;
+
+public enum ServiceBusConsumerOperationDefinition {
+ receiveMessages,
+ peekMessages
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpoint.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpoint.java
new file mode 100644
index 0000000..0ac3df6
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpoint.java
@@ -0,0 +1,70 @@
+/*
+ * 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.servicebus;
+
+import org.apache.camel.Category;
+import org.apache.camel.Component;
+import org.apache.camel.Consumer;
+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-servicebus component that integrates Azure ServiceBus. Azure ServiceBus is a fully managed enterprise
+ * integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and
+ * secure platform for asynchronous transfer of data and state. Data is transferred between different applications and
+ * services using messages.
+ */
+@UriEndpoint(firstVersion = "3.12.0", scheme = "azure-servicebus", title = "Azure ServiceBus",
+ syntax = "azure-servicebus:queueNameOrTopicName", category = {
+ Category.CLOUD, Category.MESSAGING })
+public class ServiceBusEndpoint extends DefaultEndpoint {
+
+ @UriParam
+ private ServiceBusConfiguration configuration;
+
+ public ServiceBusEndpoint(final String uri, final Component component, final ServiceBusConfiguration configuration) {
+ super(uri, component);
+ this.configuration = configuration;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new ServiceBusProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ final Consumer eventHubConsumer = new ServiceBusConsumer(this, processor);
+ configureConsumer(eventHubConsumer);
+
+ return eventHubConsumer;
+ }
+
+ /**
+ * The component configurations
+ */
+ public ServiceBusConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(ServiceBusConfiguration configuration) {
+ this.configuration = configuration;
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusProducer.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusProducer.java
new file mode 100644
index 0000000..444f0f4
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusProducer.java
@@ -0,0 +1,208 @@
+/*
+ * 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.servicebus;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusClientFactory;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusSenderAsyncClientWrapper;
+import org.apache.camel.component.azure.servicebus.operations.ServiceBusSenderOperations;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
+public class ServiceBusProducer extends DefaultAsyncProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ServiceBusProducer.class);
+
+ private ServiceBusSenderAsyncClientWrapper senderClientWrapper;
+ private ServiceBusConfigurationOptionsProxy configurationOptionsProxy;
+ private ServiceBusSenderOperations serviceBusSenderOperations;
+
+ private final Map<ServiceBusProducerOperationDefinition, BiConsumer<Exchange, AsyncCallback>> operationsToExecute
+ = new HashMap<>();
+
+ {
+ bind(ServiceBusProducerOperationDefinition.sendMessages, sendMessages());
+ bind(ServiceBusProducerOperationDefinition.scheduleMessages, scheduleMessages());
+ }
+
+ public ServiceBusProducer(final Endpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ protected void doInit() throws Exception {
+ super.doInit();
+ configurationOptionsProxy = new ServiceBusConfigurationOptionsProxy(getConfiguration());
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ // create the senderClient
+ final ServiceBusSenderAsyncClient senderClient = getConfiguration().getSenderAsyncClient() != null
+ ? getConfiguration().getSenderAsyncClient()
+ : ServiceBusClientFactory.createServiceBusSenderAsyncClient(getConfiguration());
+
+ // create the wrapper
+ senderClientWrapper = new ServiceBusSenderAsyncClientWrapper(senderClient);
+
+ // create the operations
+ serviceBusSenderOperations = new ServiceBusSenderOperations(senderClientWrapper);
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ try {
+ invokeOperation(configurationOptionsProxy.getServiceBusProducerOperationDefinition(exchange), exchange, callback);
+ return false;
+ } catch (Exception e) {
+ exchange.setException(e);
+ callback.done(true);
+ return true;
+ }
+
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (senderClientWrapper != null) {
+ // shutdown async client
+ senderClientWrapper.close();
+ }
+
+ super.doStop();
+ }
+
+ @Override
+ public ServiceBusEndpoint getEndpoint() {
+ return (ServiceBusEndpoint) super.getEndpoint();
+ }
+
+ public ServiceBusConfiguration getConfiguration() {
+ return getEndpoint().getConfiguration();
+ }
+
+ private void bind(ServiceBusProducerOperationDefinition operation, BiConsumer<Exchange, AsyncCallback> fn) {
+ operationsToExecute.put(operation, fn);
+ }
+
+ /**
+ * Entry method that selects the appropriate ServiceBusProducerOperationDefinition operation and executes it
+ */
+ private void invokeOperation(
+ final ServiceBusProducerOperationDefinition operation, final Exchange exchange, final AsyncCallback callback) {
+ final ServiceBusProducerOperationDefinition operationsToInvoke;
+
+ // we put sendMessage operation as default in case no operation has been selected
+ if (ObjectHelper.isEmpty(operation)) {
+ operationsToInvoke = ServiceBusProducerOperationDefinition.sendMessages;
+ } else {
+ operationsToInvoke = operation;
+ }
+
+ final BiConsumer<Exchange, AsyncCallback> fnToInvoke = operationsToExecute.get(operationsToInvoke);
+
+ if (fnToInvoke != null) {
+ fnToInvoke.accept(exchange, callback);
+ } else {
+ throw new RuntimeException("Operation not supported. Value: " + operationsToInvoke);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private BiConsumer<Exchange, AsyncCallback> sendMessages() {
+ return (exchange, callback) -> {
+ final Object inputBody = exchange.getMessage().getBody();
+
+ Mono<Void> sendMessageAsync;
+
+ if (exchange.getMessage().getBody() instanceof Iterable) {
+ sendMessageAsync
+ = serviceBusSenderOperations.sendMessages(convertBodyToList((Iterable<Object>) inputBody),
+ configurationOptionsProxy.getServiceBusTransactionContext(exchange));
+ } else {
+ sendMessageAsync = serviceBusSenderOperations.sendMessages(exchange.getMessage().getBody(String.class),
+ configurationOptionsProxy.getServiceBusTransactionContext(exchange));
+ }
+
+ subscribeToMono(sendMessageAsync, exchange, noop -> {
+ }, callback);
+ };
+ }
+
+ @SuppressWarnings("unchecked")
+ private BiConsumer<Exchange, AsyncCallback> scheduleMessages() {
+ return (exchange, callback) -> {
+ final Object inputBody = exchange.getMessage().getBody();
+
+ Mono<List<Long>> scheduleMessagesAsync;
+
+ if (exchange.getMessage().getBody() instanceof Iterable) {
+ scheduleMessagesAsync
+ = serviceBusSenderOperations.scheduleMessages(convertBodyToList((Iterable<Object>) inputBody),
+ configurationOptionsProxy.getScheduledEnqueueTime(exchange),
+ configurationOptionsProxy.getServiceBusTransactionContext(exchange));
+ } else {
+ scheduleMessagesAsync
+ = serviceBusSenderOperations.scheduleMessages(exchange.getMessage().getBody(String.class),
+ configurationOptionsProxy.getScheduledEnqueueTime(exchange),
+ configurationOptionsProxy.getServiceBusTransactionContext(exchange));
+ }
+
+ subscribeToMono(scheduleMessagesAsync, exchange,
+ sequenceNumbers -> exchange.getMessage().setBody(sequenceNumbers), callback);
+ };
+ }
+
+ private List<String> convertBodyToList(final Iterable<Object> inputBody) {
+ return StreamSupport.stream(inputBody.spliterator(), false)
+ .map(body -> getEndpoint().getCamelContext().getTypeConverter().convertTo(String.class, body))
+ .collect(Collectors.toList());
+ }
+
+ private <T> void subscribeToMono(
+ final Mono<T> inputMono, final Exchange exchange, final Consumer<T> resultsCallback, final AsyncCallback callback) {
+ inputMono
+ .subscribe(resultsCallback, error -> {
+ // error but we continue
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Error processing async exchange with error: {}", error.getMessage());
+ }
+ exchange.setException(error);
+ callback.done(false);
+ }, () -> {
+ // we are done from everything, so mark it as sync done
+ LOG.trace("All events with exchange have been sent successfully.");
+ callback.done(false);
+ });
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusProducerOperationDefinition.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusProducerOperationDefinition.java
new file mode 100644
index 0000000..8f38101
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusProducerOperationDefinition.java
@@ -0,0 +1,22 @@
+/*
+ * 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.servicebus;
+
+public enum ServiceBusProducerOperationDefinition {
+ sendMessages,
+ scheduleMessages
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusType.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusType.java
new file mode 100644
index 0000000..b3a5c00
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusType.java
@@ -0,0 +1,22 @@
+/*
+ * 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.servicebus;
+
+public enum ServiceBusType {
+ queue,
+ topic;
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusTypeConverter.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusTypeConverter.java
new file mode 100644
index 0000000..da1621a
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusTypeConverter.java
@@ -0,0 +1,44 @@
+/*
+ * 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.servicebus;
+
+import com.azure.core.util.BinaryData;
+import org.apache.camel.Converter;
+import org.apache.camel.util.ObjectHelper;
+
+@Converter(generateLoader = true)
+public final class ServiceBusTypeConverter {
+
+ private ServiceBusTypeConverter() {
+ }
+
+ @Converter
+ public static String toString(final BinaryData binaryData) {
+ if (ObjectHelper.isNotEmpty(binaryData)) {
+ return binaryData.toString();
+ }
+ return null;
+ }
+
+ @Converter
+ public static BinaryData toBinaryData(final String data) {
+ if (ObjectHelper.isNotEmpty(data)) {
+ return BinaryData.fromString(data);
+ }
+ return null;
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusUtils.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusUtils.java
new file mode 100644
index 0000000..a0d5640
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusUtils.java
@@ -0,0 +1,47 @@
+/*
+ * 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.servicebus;
+
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import com.azure.core.util.BinaryData;
+import com.azure.messaging.servicebus.ServiceBusMessage;
+
+public final class ServiceBusUtils {
+
+ private ServiceBusUtils() {
+ }
+
+ public static ServiceBusMessage createServiceBusMessage(final Object data) {
+ if (data instanceof String) {
+ return new ServiceBusMessage((String) data);
+ } else if (data instanceof byte[]) {
+ return new ServiceBusMessage((byte[]) data);
+ } else if (data instanceof BinaryData) {
+ return new ServiceBusMessage((BinaryData) data);
+ } else {
+ throw new IllegalArgumentException("Make sure your message data is in String, byte[] or BinaryData");
+ }
+ }
+
+ public static Iterable<ServiceBusMessage> createServiceBusMessages(final Iterable<Object> data) {
+ return StreamSupport.stream(data.spliterator(), false)
+ .map(ServiceBusUtils::createServiceBusMessage)
+ .collect(Collectors.toList());
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusClientFactory.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusClientFactory.java
new file mode 100644
index 0000000..d1b8485
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusClientFactory.java
@@ -0,0 +1,80 @@
+/*
+ * 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.servicebus.client;
+
+import com.azure.messaging.servicebus.ServiceBusClientBuilder;
+import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
+import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+import org.apache.camel.component.azure.servicebus.ServiceBusConfiguration;
+import org.apache.camel.component.azure.servicebus.ServiceBusType;
+
+public final class ServiceBusClientFactory {
+
+ private ServiceBusClientFactory() {
+ }
+
+ public static ServiceBusSenderAsyncClient createServiceBusSenderAsyncClient(final ServiceBusConfiguration configuration) {
+ return createBaseServiceBusSenderClient(createBaseServiceBusClient(configuration), configuration)
+ .buildAsyncClient();
+ }
+
+ public static ServiceBusReceiverAsyncClient createServiceBusReceiverAsyncClient(
+ final ServiceBusConfiguration configuration) {
+ return createBaseServiceBusReceiverClient(createBaseServiceBusClient(configuration), configuration)
+ .prefetchCount(configuration.getPrefetchCount())
+ .receiveMode(configuration.getServiceBusReceiveMode())
+ .subQueue(configuration.getSubQueue())
+ .maxAutoLockRenewDuration(configuration.getMaxAutoLockRenewDuration())
+ .subscriptionName(configuration.getSubscriptionName())
+ .buildAsyncClient();
+ }
+
+ private static ServiceBusClientBuilder createBaseServiceBusClient(final ServiceBusConfiguration configuration) {
+ return new ServiceBusClientBuilder()
+ .transportType(configuration.getAmqpTransportType())
+ .clientOptions(configuration.getClientOptions())
+ .retryOptions(configuration.getAmqpRetryOptions())
+ .proxyOptions(configuration.getProxyOptions())
+ .connectionString(configuration.getConnectionString());
+ }
+
+ private static ServiceBusClientBuilder.ServiceBusSenderClientBuilder createBaseServiceBusSenderClient(
+ final ServiceBusClientBuilder busClientBuilder, final ServiceBusConfiguration configuration) {
+ if (configuration.getServiceBusType() == ServiceBusType.queue) {
+ return busClientBuilder.sender()
+ .queueName(configuration.getTopicOrQueueName());
+ } else {
+ return busClientBuilder.sender()
+ .topicName(configuration.getTopicOrQueueName());
+ }
+ }
+
+ private static ServiceBusClientBuilder.ServiceBusReceiverClientBuilder createBaseServiceBusReceiverClient(
+ final ServiceBusClientBuilder busClientBuilder, final ServiceBusConfiguration configuration) {
+ final ServiceBusClientBuilder.ServiceBusReceiverClientBuilder receiverClientBuilder = busClientBuilder.receiver();
+
+ if (configuration.isDisableAutoComplete()) {
+ receiverClientBuilder.disableAutoComplete();
+ }
+
+ if (configuration.getServiceBusType() == ServiceBusType.queue) {
+ return receiverClientBuilder.queueName(configuration.getTopicOrQueueName());
+ } else {
+ return receiverClientBuilder.topicName(configuration.getTopicOrQueueName());
+ }
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusReceiverAsyncClientWrapper.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusReceiverAsyncClientWrapper.java
new file mode 100644
index 0000000..4261ecb
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusReceiverAsyncClientWrapper.java
@@ -0,0 +1,158 @@
+/*
+ * 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.servicebus.client;
+
+import java.time.Duration;
+import java.time.OffsetDateTime;
+
+import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
+import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
+import com.azure.messaging.servicebus.ServiceBusTransactionContext;
+import com.azure.messaging.servicebus.models.AbandonOptions;
+import com.azure.messaging.servicebus.models.CompleteOptions;
+import com.azure.messaging.servicebus.models.DeadLetterOptions;
+import com.azure.messaging.servicebus.models.DeferOptions;
+import org.apache.camel.util.ObjectHelper;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class ServiceBusReceiverAsyncClientWrapper {
+
+ private ServiceBusReceiverAsyncClient client;
+
+ public ServiceBusReceiverAsyncClientWrapper(final ServiceBusReceiverAsyncClient client) {
+ ObjectHelper.isNotEmpty(client);
+
+ this.client = client;
+ }
+
+ public String getFullyQualifiedNamespace() {
+ return client.getFullyQualifiedNamespace();
+ }
+
+ public String getEntityPath() {
+ return client.getEntityPath();
+ }
+
+ public Mono<Void> abandon(ServiceBusReceivedMessage message) {
+ return client.abandon(message);
+ }
+
+ public Mono<Void> abandon(
+ ServiceBusReceivedMessage message,
+ AbandonOptions options) {
+ return client.abandon(message, options);
+ }
+
+ public Mono<Void> complete(ServiceBusReceivedMessage message) {
+ return client.complete(message);
+ }
+
+ public Mono<Void> complete(
+ ServiceBusReceivedMessage message,
+ CompleteOptions options) {
+ return client.complete(message, options);
+ }
+
+ public Mono<Void> defer(ServiceBusReceivedMessage message) {
+ return client.defer(message);
+ }
+
+ public Mono<Void> defer(
+ ServiceBusReceivedMessage message,
+ DeferOptions options) {
+ return client.defer(message, options);
+ }
+
+ public Mono<Void> deadLetter(ServiceBusReceivedMessage message) {
+ return client.deadLetter(message);
+ }
+
+ public Mono<Void> deadLetter(
+ ServiceBusReceivedMessage message,
+ DeadLetterOptions options) {
+ return client.deadLetter(message, options);
+ }
+
+ public Mono<byte[]> getSessionState() {
+ return client.getSessionState();
+ }
+
+ public Mono<ServiceBusReceivedMessage> peekMessage() {
+ return client.peekMessage();
+ }
+
+ public Mono<ServiceBusReceivedMessage> peekMessage(long sequenceNumber) {
+ return client.peekMessage(sequenceNumber);
+ }
+
+ public Flux<ServiceBusReceivedMessage> peekMessages(int maxMessages) {
+ return client.peekMessages(maxMessages);
+ }
+
+ public Flux<ServiceBusReceivedMessage> peekMessages(int maxMessages, long sequenceNumber) {
+ return client.peekMessages(maxMessages, sequenceNumber);
+ }
+
+ public Flux<ServiceBusReceivedMessage> receiveMessages() {
+ return client.receiveMessages();
+ }
+
+ public Mono<ServiceBusReceivedMessage> receiveDeferredMessage(long sequenceNumber) {
+ return client.receiveDeferredMessage(sequenceNumber);
+ }
+
+ public Flux<ServiceBusReceivedMessage> receiveDeferredMessages(Iterable<Long> sequenceNumbers) {
+ return client.receiveDeferredMessages(sequenceNumbers);
+ }
+
+ public Mono<OffsetDateTime> renewMessageLock(ServiceBusReceivedMessage message) {
+ return client.renewMessageLock(message);
+ }
+
+ public Mono<Void> renewMessageLock(ServiceBusReceivedMessage message, Duration maxLockRenewalDuration) {
+ return client.renewMessageLock(message, maxLockRenewalDuration);
+ }
+
+ public Mono<OffsetDateTime> renewSessionLock() {
+ return client.renewSessionLock();
+ }
+
+ public Mono<Void> renewSessionLock(Duration maxLockRenewalDuration) {
+ return client.renewSessionLock(maxLockRenewalDuration);
+ }
+
+ public Mono<Void> setSessionState(byte[] sessionState) {
+ return client.setSessionState(sessionState);
+ }
+
+ public Mono<ServiceBusTransactionContext> createTransaction() {
+ return client.createTransaction();
+ }
+
+ public Mono<Void> commitTransaction(ServiceBusTransactionContext transactionContext) {
+ return client.commitTransaction(transactionContext);
+ }
+
+ public Mono<Void> rollbackTransaction(ServiceBusTransactionContext transactionContext) {
+ return client.rollbackTransaction(transactionContext);
+ }
+
+ public void close() {
+ client.close();
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusSenderAsyncClientWrapper.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusSenderAsyncClientWrapper.java
new file mode 100644
index 0000000..be34176
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/client/ServiceBusSenderAsyncClientWrapper.java
@@ -0,0 +1,133 @@
+/*
+ * 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.servicebus.client;
+
+import java.time.OffsetDateTime;
+
+import com.azure.messaging.servicebus.ServiceBusMessage;
+import com.azure.messaging.servicebus.ServiceBusMessageBatch;
+import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+import com.azure.messaging.servicebus.ServiceBusTransactionContext;
+import com.azure.messaging.servicebus.models.CreateMessageBatchOptions;
+import org.apache.camel.util.ObjectHelper;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class ServiceBusSenderAsyncClientWrapper {
+
+ private final ServiceBusSenderAsyncClient client;
+
+ public ServiceBusSenderAsyncClientWrapper(final ServiceBusSenderAsyncClient client) {
+ ObjectHelper.isNotEmpty(client);
+
+ this.client = client;
+ }
+
+ public String getFullyQualifiedNamespace() {
+ return client.getFullyQualifiedNamespace();
+ }
+
+ public String getEntityPath() {
+ return client.getEntityPath();
+ }
+
+ public Mono<Void> sendMessage(ServiceBusMessage message) {
+ return client.sendMessage(message);
+ }
+
+ public Mono<Void> sendMessage(
+ ServiceBusMessage message,
+ ServiceBusTransactionContext transactionContext) {
+ return client.sendMessage(message, transactionContext);
+ }
+
+ public Mono<Void> sendMessages(
+ Iterable<ServiceBusMessage> messages,
+ ServiceBusTransactionContext transactionContext) {
+ return client.sendMessages(messages, transactionContext);
+ }
+
+ public Mono<Void> sendMessages(Iterable<ServiceBusMessage> messages) {
+ return client.sendMessages(messages);
+ }
+
+ public Mono<Void> sendMessages(ServiceBusMessageBatch batch) {
+ return client.sendMessages(batch);
+ }
+
+ public Mono<Void> sendMessages(
+ ServiceBusMessageBatch batch,
+ ServiceBusTransactionContext transactionContext) {
+ return client.sendMessages(batch, transactionContext);
+ }
+
+ public Mono<ServiceBusMessageBatch> createMessageBatch() {
+ return client.createMessageBatch();
+ }
+
+ public Mono<ServiceBusMessageBatch> createMessageBatch(
+ CreateMessageBatchOptions options) {
+ return client.createMessageBatch(options);
+ }
+
+ public Mono<Long> scheduleMessage(
+ ServiceBusMessage message, OffsetDateTime scheduledEnqueueTime,
+ ServiceBusTransactionContext transactionContext) {
+ return client.scheduleMessage(message, scheduledEnqueueTime, transactionContext);
+ }
+
+ public Mono<Long> scheduleMessage(ServiceBusMessage message, OffsetDateTime scheduledEnqueueTime) {
+ return client.scheduleMessage(message, scheduledEnqueueTime);
+ }
+
+ public Flux<Long> scheduleMessages(
+ Iterable<ServiceBusMessage> messages,
+ OffsetDateTime scheduledEnqueueTime) {
+ return client.scheduleMessages(messages, scheduledEnqueueTime);
+ }
+
+ public Flux<Long> scheduleMessages(
+ Iterable<ServiceBusMessage> messages,
+ OffsetDateTime scheduledEnqueueTime,
+ ServiceBusTransactionContext transactionContext) {
+ return client.scheduleMessages(messages, scheduledEnqueueTime, transactionContext);
+ }
+
+ public Mono<Void> cancelScheduledMessage(long sequenceNumber) {
+ return client.cancelScheduledMessage(sequenceNumber);
+ }
+
+ public Mono<Void> cancelScheduledMessages(Iterable<Long> sequenceNumbers) {
+ return client.cancelScheduledMessages(sequenceNumbers);
+ }
+
+ public Mono<ServiceBusTransactionContext> createTransaction() {
+ return client.createTransaction();
+ }
+
+ public Mono<Void> commitTransaction(ServiceBusTransactionContext transactionContext) {
+ return client.commitTransaction(transactionContext);
+ }
+
+ public Mono<Void> rollbackTransaction(ServiceBusTransactionContext transactionContext) {
+ return client.rollbackTransaction(transactionContext);
+ }
+
+ public void close() {
+ client.close();
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/operations/ServiceBusReceiverOperations.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/operations/ServiceBusReceiverOperations.java
new file mode 100644
index 0000000..27c8741
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/operations/ServiceBusReceiverOperations.java
@@ -0,0 +1,46 @@
+/*
+ * 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.servicebus.operations;
+
+import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusReceiverAsyncClientWrapper;
+import org.apache.camel.util.ObjectHelper;
+import reactor.core.publisher.Flux;
+
+public class ServiceBusReceiverOperations {
+
+ private final ServiceBusReceiverAsyncClientWrapper client;
+
+ public ServiceBusReceiverOperations(final ServiceBusReceiverAsyncClientWrapper client) {
+ ObjectHelper.notNull(client, "client");
+
+ this.client = client;
+ }
+
+ public Flux<ServiceBusReceivedMessage> receiveMessages() {
+ return client.receiveMessages();
+ }
+
+ public Flux<ServiceBusReceivedMessage> peekMessages(final Integer numMaxMessages) {
+ if (ObjectHelper.isEmpty(numMaxMessages)) {
+ return client.peekMessage()
+ .flux();
+ }
+
+ return client.peekMessages(numMaxMessages);
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/operations/ServiceBusSenderOperations.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/operations/ServiceBusSenderOperations.java
new file mode 100644
index 0000000..64e2aba
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/operations/ServiceBusSenderOperations.java
@@ -0,0 +1,107 @@
+/*
+ * 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.servicebus.operations;
+
+import java.time.OffsetDateTime;
+import java.util.Collections;
+import java.util.List;
+
+import com.azure.messaging.servicebus.ServiceBusMessage;
+import com.azure.messaging.servicebus.ServiceBusTransactionContext;
+import org.apache.camel.component.azure.servicebus.ServiceBusUtils;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusSenderAsyncClientWrapper;
+import org.apache.camel.util.ObjectHelper;
+import reactor.core.publisher.Mono;
+
+public class ServiceBusSenderOperations {
+
+ private final ServiceBusSenderAsyncClientWrapper client;
+
+ public ServiceBusSenderOperations(ServiceBusSenderAsyncClientWrapper client) {
+ ObjectHelper.notNull(client, "client");
+
+ this.client = client;
+ }
+
+ public Mono<Void> sendMessages(final Object data, final ServiceBusTransactionContext context) {
+ if (data instanceof Iterable) {
+ return sendMessages((Iterable<Object>) data, context);
+ }
+
+ return sendMessage(data, context);
+ }
+
+ public Mono<List<Long>> scheduleMessages(
+ final Object data, final OffsetDateTime scheduledEnqueueTime, final ServiceBusTransactionContext context) {
+ if (ObjectHelper.isEmpty(scheduledEnqueueTime)) {
+ throw new IllegalArgumentException("To schedule a message, you need to set scheduledEnqueueTime.");
+ }
+
+ if (data instanceof Iterable) {
+ return scheduleMessages((Iterable<Object>) data, scheduledEnqueueTime, context);
+ }
+
+ return scheduleMessage(data, scheduledEnqueueTime, context);
+ }
+
+ private Mono<Void> sendMessages(final Iterable<Object> data, final ServiceBusTransactionContext context) {
+ final Iterable<ServiceBusMessage> messages = ServiceBusUtils.createServiceBusMessages(data);
+
+ if (ObjectHelper.isEmpty(context)) {
+ return client.sendMessages(messages);
+ }
+
+ return client.sendMessages(messages, context);
+ }
+
+ private Mono<Void> sendMessage(final Object data, final ServiceBusTransactionContext context) {
+ final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data);
+
+ if (ObjectHelper.isEmpty(context)) {
+ return client.sendMessage(message);
+ }
+
+ return client.sendMessage(message, context);
+ }
+
+ private Mono<List<Long>> scheduleMessage(
+ final Object data, final OffsetDateTime scheduledEnqueueTime, final ServiceBusTransactionContext context) {
+ final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data);
+
+ if (ObjectHelper.isEmpty(context)) {
+ return client.scheduleMessage(message, scheduledEnqueueTime)
+ .map(Collections::singletonList);
+ }
+
+ return client.scheduleMessage(message, scheduledEnqueueTime, context)
+ .map(Collections::singletonList);
+ }
+
+ private Mono<List<Long>> scheduleMessages(
+ final Iterable<Object> data, final OffsetDateTime scheduledEnqueueTime,
+ final ServiceBusTransactionContext context) {
+ final Iterable<ServiceBusMessage> messages = ServiceBusUtils.createServiceBusMessages(data);
+
+ if (ObjectHelper.isEmpty(context)) {
+ return client.scheduleMessages(messages, scheduledEnqueueTime)
+ .collectList();
+ }
+
+ return client.scheduleMessages(messages, scheduledEnqueueTime, context)
+ .collectList();
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointTest.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointTest.java
new file mode 100644
index 0000000..94af271
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpointTest.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.servicebus;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.ResolveEndpointFailedException;
+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.assertThrows;
+
+class ServiceBusEndpointTest extends CamelTestSupport {
+
+ @Test
+ void testCreateWithInvalidData() throws Exception {
+ assertThrows(ResolveEndpointFailedException.class,
+ () -> context.getEndpoint("azure-servicebus:test//?"));
+
+ assertThrows(ResolveEndpointFailedException.class,
+ () -> context.getEndpoint("azure-servicebus://?connectionString=test"));
+ }
+
+ @Test
+ void testCreateEndpointWithConfig() throws Exception {
+ final String uri = "azure-servicebus://testTopicOrQueue";
+ final String remaining = "testTopicOrQueue";
+ final Map<String, Object> params = new HashMap<>();
+ params.put("serviceBusType", ServiceBusType.topic);
+ params.put("prefetchCount", 10);
+ params.put("connectionString", "testString");
+
+ final ServiceBusEndpoint endpoint
+ = (ServiceBusEndpoint) context.getComponent("azure-servicebus", ServiceBusComponent.class)
+ .createEndpoint(uri, remaining, params);
+
+ assertEquals(ServiceBusType.topic, endpoint.getConfiguration().getServiceBusType());
+ assertEquals("testTopicOrQueue", endpoint.getConfiguration().getTopicOrQueueName());
+ assertEquals(10, endpoint.getConfiguration().getPrefetchCount());
+ assertEquals("testString", endpoint.getConfiguration().getConnectionString());
+ }
+
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusTestUtils.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusTestUtils.java
new file mode 100644
index 0000000..57771f1
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusTestUtils.java
@@ -0,0 +1,98 @@
+/*
+ * 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.servicebus;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+import java.util.Properties;
+
+import com.azure.messaging.servicebus.ServiceBusClientBuilder;
+import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
+import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+
+public final class ServiceBusTestUtils {
+
+ public static final String CONNECTION_STRING = "connectionString";
+ public static final String TOPIC_NAME = "topicName";
+ public static final String SUBSCRIPTION_NAME = "subscriptionName";
+ public static final String QUEUE_NAME = "queueName";
+
+ private ServiceBusTestUtils() {
+ }
+
+ public static Properties loadAzurePropertiesFile() throws IOException {
+ final Properties properties = new Properties();
+ final String fileName = "azure_key.properties";
+
+ final InputStream inputStream
+ = Objects.requireNonNull(ServiceBusTestUtils.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) {
+ throw new Exception(
+ "Make sure to supply azure servicebus connectionString, e.g: mvn verify -PfullTests -DconnectionString=string");
+ }
+ properties.setProperty(CONNECTION_STRING, System.getProperty(CONNECTION_STRING));
+ properties.setProperty(TOPIC_NAME, System.getProperty(TOPIC_NAME));
+ properties.setProperty(SUBSCRIPTION_NAME, System.getProperty(SUBSCRIPTION_NAME));
+ //properties.setProperty(QUEUE_NAME, System.getProperty(QUEUE_NAME));
+
+ return properties;
+ }
+
+ public static ServiceBusReceiverAsyncClient createServiceBusReceiverAsyncClient(final ServiceBusType type)
+ throws Exception {
+ final Properties properties = loadAzureAccessFromJvmEnv();
+
+ final ServiceBusClientBuilder.ServiceBusReceiverClientBuilder clientBuilder = new ServiceBusClientBuilder()
+ .connectionString(properties.getProperty(CONNECTION_STRING))
+ .receiver()
+ .subscriptionName(properties.getProperty(SUBSCRIPTION_NAME));
+
+ if (type == ServiceBusType.queue) {
+ clientBuilder.queueName(properties.getProperty(QUEUE_NAME));
+ } else {
+ clientBuilder.topicName(properties.getProperty(TOPIC_NAME));
+ }
+
+ return clientBuilder.buildAsyncClient();
+ }
+
+ public static ServiceBusSenderAsyncClient createServiceBusSenderAsyncClient(final ServiceBusType type) throws Exception {
+ final Properties properties = loadAzureAccessFromJvmEnv();
+
+ final ServiceBusClientBuilder.ServiceBusSenderClientBuilder clientBuilder = new ServiceBusClientBuilder()
+ .connectionString(properties.getProperty(CONNECTION_STRING))
+ .sender();
+
+ if (type == ServiceBusType.queue) {
+ clientBuilder.queueName(properties.getProperty(QUEUE_NAME));
+ } else {
+ clientBuilder.topicName(properties.getProperty(TOPIC_NAME));
+ }
+
+ return clientBuilder.buildAsyncClient();
+ }
+
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusUtilsTest.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusUtilsTest.java
new file mode 100644
index 0000000..ee7d4ff
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusUtilsTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.servicebus;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.StreamSupport;
+
+import com.azure.messaging.servicebus.ServiceBusMessage;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class ServiceBusUtilsTest {
+
+ @Test
+ void testCreateServiceBusMessage() {
+ // test string
+ final ServiceBusMessage message1 = ServiceBusUtils.createServiceBusMessage("test string");
+
+ assertEquals("test string", message1.getBody().toString());
+
+ // test int
+ final ServiceBusMessage message2 = ServiceBusUtils.createServiceBusMessage(String.valueOf(12345));
+
+ assertEquals("12345", message2.getBody().toString());
+ }
+
+ @Test
+ void testCreateServiceBusMessages() {
+ final List<Object> inputMessages = new LinkedList<>();
+ inputMessages.add("test data");
+ inputMessages.add(String.valueOf(12345));
+
+ final Iterable<ServiceBusMessage> busMessages = ServiceBusUtils.createServiceBusMessages(inputMessages);
+
+ assertTrue(StreamSupport.stream(busMessages.spliterator(), false)
+ .anyMatch(record -> record.getBody().toString().equals("test data")));
+ assertTrue(StreamSupport.stream(busMessages.spliterator(), false)
+ .anyMatch(record -> record.getBody().toString().equals("12345")));
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/BaseCamelServiceBusTestSupport.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/BaseCamelServiceBusTestSupport.java
new file mode 100644
index 0000000..5e84c47
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/BaseCamelServiceBusTestSupport.java
@@ -0,0 +1,63 @@
+/*
+ * 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.servicebus.integration;
+
+import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
+import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.azure.servicebus.ServiceBusComponent;
+import org.apache.camel.component.azure.servicebus.ServiceBusTestUtils;
+import org.apache.camel.component.azure.servicebus.ServiceBusType;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+
+public class BaseCamelServiceBusTestSupport extends CamelTestSupport {
+
+ protected ServiceBusSenderAsyncClient senderAsyncClient;
+ protected ServiceBusReceiverAsyncClient receiverAsyncClient;
+
+ @BeforeEach
+ void prepareClient() throws Exception {
+ senderAsyncClient = ServiceBusTestUtils.createServiceBusSenderAsyncClient(ServiceBusType.topic);
+ receiverAsyncClient = ServiceBusTestUtils.createServiceBusReceiverAsyncClient(ServiceBusType.topic);
+ }
+
+ @AfterEach
+ void closeClient() {
+ senderAsyncClient.close();
+ receiverAsyncClient.close();
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ final ServiceBusSenderAsyncClient injectedSenderAsyncClient
+ = ServiceBusTestUtils.createServiceBusSenderAsyncClient(ServiceBusType.topic);
+ final ServiceBusReceiverAsyncClient injectedReceiverAsyncClient
+ = ServiceBusTestUtils.createServiceBusReceiverAsyncClient(ServiceBusType.topic);
+
+ final CamelContext context = super.createCamelContext();
+ final ServiceBusComponent component = new ServiceBusComponent(context);
+
+ component.init();
+ component.getConfiguration().setReceiverAsyncClient(injectedReceiverAsyncClient);
+ component.getConfiguration().setSenderAsyncClient(injectedSenderAsyncClient);
+ context.addComponent("azure-servicebus", component);
+
+ return context;
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerTest.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerTest.java
new file mode 100644
index 0000000..531412e
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerTest.java
@@ -0,0 +1,115 @@
+/*
+ * 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.servicebus.integration;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.azure.servicebus.ServiceBusConstants;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusSenderAsyncClientWrapper;
+import org.apache.camel.component.azure.servicebus.operations.ServiceBusSenderOperations;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+@EnabledIfSystemProperty(named = "connectionString", matches = ".*",
+ disabledReason = "Make sure to supply azure servicebus connectionString, e.g: mvn verify -DconnectionString=string")
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class ServiceBusConsumerTest extends BaseCamelServiceBusTestSupport {
+
+ @EndpointInject("mock:receiveMessagesResult")
+ private MockEndpoint receiveMessagesResult;
+
+ @EndpointInject("mock:peekMessagesResult")
+ private MockEndpoint peekMessagesResult;
+
+ @Test
+ void testReceiveMessages() throws InterruptedException {
+ // send test data
+ final List<Object> inputBatch = new LinkedList<>();
+ inputBatch.add("test batch 1");
+ inputBatch.add("test batch 2");
+ inputBatch.add("test batch 3");
+
+ new ServiceBusSenderOperations(new ServiceBusSenderAsyncClientWrapper(senderAsyncClient)).sendMessages(inputBatch, null)
+ .block();
+
+ // test the data now
+ receiveMessagesResult.expectedMessageCount(3);
+ receiveMessagesResult.expectedBodiesReceived("test batch 1", "test batch 2", "test batch 3");
+ receiveMessagesResult.assertIsSatisfied();
+
+ final List<Exchange> exchanges = receiveMessagesResult.getExchanges();
+
+ assertNotNull(exchanges.get(0).getMessage().getHeaders());
+ assertNotNull(exchanges.get(1).getMessage().getHeaders());
+ assertNotNull(exchanges.get(2).getMessage().getHeaders());
+
+ // we test headers
+ assertNotNull(exchanges.get(0).getMessage().getHeader(ServiceBusConstants.MESSAGE_ID));
+ assertNotNull(exchanges.get(1).getMessage().getHeader(ServiceBusConstants.MESSAGE_ID));
+ assertNotNull(exchanges.get(2).getMessage().getHeader(ServiceBusConstants.MESSAGE_ID));
+ }
+
+ @Test
+ @Disabled
+ void testPeekMessages() throws InterruptedException {
+ // send test data
+ final List<Object> inputBatch = new LinkedList<>();
+ inputBatch.add("peek test batch 1");
+ inputBatch.add("peek test batch 2");
+ inputBatch.add("peek test batch 3");
+
+ new ServiceBusSenderOperations(new ServiceBusSenderAsyncClientWrapper(senderAsyncClient)).sendMessages(inputBatch, null)
+ .block();
+
+ // test the data now
+ peekMessagesResult.expectedMessageCount(3);
+ peekMessagesResult.expectedBodiesReceived("peek test batch 1", "peek test batch 2", "peek test batch 3");
+ peekMessagesResult.assertIsSatisfied();
+
+ final List<Exchange> exchanges = peekMessagesResult.getExchanges();
+
+ assertNotNull(exchanges.get(0).getMessage().getHeaders());
+ assertNotNull(exchanges.get(1).getMessage().getHeaders());
+ assertNotNull(exchanges.get(2).getMessage().getHeaders());
+
+ // we test headers
+ assertNotNull(exchanges.get(0).getMessage().getHeader(ServiceBusConstants.MESSAGE_ID));
+ assertNotNull(exchanges.get(1).getMessage().getHeader(ServiceBusConstants.MESSAGE_ID));
+ assertNotNull(exchanges.get(2).getMessage().getHeader(ServiceBusConstants.MESSAGE_ID));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("azure-servicebus:test//?connectionString=test").to(receiveMessagesResult);
+ from("azure-servicebus:test//?connectionString=test&consumerOperation=peekMessages&peekNumMaxMessages=3")
+ .to(peekMessagesResult);
+ }
+ };
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusProducerTest.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusProducerTest.java
new file mode 100644
index 0000000..b63df97
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusProducerTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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.servicebus.integration;
+
+import java.time.OffsetDateTime;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Spliterator;
+import java.util.stream.StreamSupport;
+
+import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.azure.servicebus.ServiceBusConstants;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@EnabledIfSystemProperty(named = "connectionString", matches = ".*",
+ disabledReason = "Make sure to supply azure servicebus connectionString, e.g: mvn verify -DconnectionString=string")
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class ServiceBusProducerTest extends BaseCamelServiceBusTestSupport {
+
+ @EndpointInject
+ private ProducerTemplate template;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint result;
+
+ @Test
+ public void testSendMessage() throws InterruptedException {
+ template.send("direct:sendMessage", exchange -> {
+ exchange.getIn().setBody(123456789);
+ });
+
+ Thread.sleep(1000);
+
+ // let's check our data
+ final Spliterator<ServiceBusReceivedMessage> receivedMessages
+ = receiverAsyncClient.receiveMessages().toIterable().spliterator();
+
+ final boolean batch1Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("123456789"));
+
+ assertTrue(batch1Exists);
+ }
+
+ @Test
+ public void testSendBatchMessages() throws InterruptedException {
+ template.send("direct:sendBatchMessages", exchange -> {
+ final List<Object> inputBatch = new LinkedList<>();
+ inputBatch.add("test batch 1");
+ inputBatch.add("test batch 2");
+ inputBatch.add("test batch 3");
+ inputBatch.add(123456);
+
+ exchange.getIn().setBody(inputBatch);
+ });
+
+ Thread.sleep(1000);
+
+ // let's check our data
+ final Spliterator<ServiceBusReceivedMessage> receivedMessages
+ = receiverAsyncClient.receiveMessages().toIterable().spliterator();
+
+ final boolean batch1Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 1"));
+
+ final boolean batch2Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 2"));
+
+ final boolean batch3Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 3"));
+
+ final boolean batch4Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("123456"));
+
+ assertTrue(batch1Exists, "test message body 1");
+ assertTrue(batch2Exists, "test message body 2");
+ assertTrue(batch3Exists, "test message body 3");
+ assertTrue(batch4Exists, "test message body 4");
+ }
+
+ @Test
+ void testScheduleMessage() throws InterruptedException {
+ template.send("direct:scheduleMessage", exchange -> {
+ exchange.getIn().setHeader(ServiceBusConstants.SCHEDULED_ENQUEUE_TIME, OffsetDateTime.now());
+ exchange.getIn().setBody("test message");
+ });
+
+ Thread.sleep(1000);
+
+ // let's check our data
+ final Spliterator<ServiceBusReceivedMessage> receivedMessages
+ = receiverAsyncClient.receiveMessages().toIterable().spliterator();
+
+ final boolean batch1Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test message"));
+
+ assertTrue(batch1Exists);
+ }
+
+ @Test
+ public void testScheduleBatchMessages() throws InterruptedException {
+ template.send("direct:scheduleBatchMessages", exchange -> {
+ final List<Object> inputBatch = new LinkedList<>();
+ inputBatch.add("test batch 1");
+ inputBatch.add("test batch 2");
+ inputBatch.add("test batch 3");
+ inputBatch.add(123456);
+
+ exchange.getIn().setHeader(ServiceBusConstants.SCHEDULED_ENQUEUE_TIME, OffsetDateTime.now());
+ exchange.getIn().setBody(inputBatch);
+ });
+
+ Thread.sleep(1000);
+
+ // let's check our data
+ final Spliterator<ServiceBusReceivedMessage> receivedMessages
+ = receiverAsyncClient.receiveMessages().toIterable().spliterator();
+
+ final boolean batch1Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 1"));
+
+ final boolean batch2Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 2"));
+
+ final boolean batch3Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 3"));
+
+ final boolean batch4Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("123456"));
+
+ assertTrue(batch1Exists, "test message body 1");
+ assertTrue(batch2Exists, "test message body 2");
+ assertTrue(batch3Exists, "test message body 3");
+ assertTrue(batch4Exists, "test message body 4");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:sendBatchMessages").to("azure-servicebus:test//?connectionString=test").to(result);
+ from("direct:sendMessage").to("azure-servicebus:test//?connectionString=test").to(result);
+ from("direct:scheduleMessage")
+ .to("azure-servicebus:test//?connectionString=test&producerOperation=scheduleMessages").to(result);
+ from("direct:scheduleBatchMessages")
+ .to("azure-servicebus:test//?connectionString=test&producerOperation=scheduleMessages").to(result);
+ }
+ };
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/operations/ServiceBusSenderOperationsTest.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/operations/ServiceBusSenderOperationsTest.java
new file mode 100644
index 0000000..f90503c
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/operations/ServiceBusSenderOperationsTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.servicebus.integration.operations;
+
+import java.time.OffsetDateTime;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import java.util.Spliterator;
+import java.util.stream.StreamSupport;
+
+import com.azure.messaging.servicebus.ServiceBusClientBuilder;
+import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
+import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
+import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+import org.apache.camel.component.azure.servicebus.ServiceBusTestUtils;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusReceiverAsyncClientWrapper;
+import org.apache.camel.component.azure.servicebus.client.ServiceBusSenderAsyncClientWrapper;
+import org.apache.camel.component.azure.servicebus.operations.ServiceBusSenderOperations;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@EnabledIfSystemProperty(named = "connectionString", matches = ".*",
+ disabledReason = "Make sure to supply azure servicebus connectionString, e.g: mvn verify -DconnectionString=string")
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class ServiceBusSenderOperationsTest {
+
+ private ServiceBusSenderAsyncClientWrapper clientSenderWrapper;
+ private ServiceBusReceiverAsyncClientWrapper clientReceiverWrapper;
+
+ @BeforeAll
+ void prepare() throws Exception {
+ final Properties properties = ServiceBusTestUtils.loadAzureAccessFromJvmEnv();
+
+ final ServiceBusSenderAsyncClient senderClient = new ServiceBusClientBuilder()
+ .connectionString(properties.getProperty(ServiceBusTestUtils.CONNECTION_STRING))
+ .sender()
+ .buildAsyncClient();
+
+ clientSenderWrapper = new ServiceBusSenderAsyncClientWrapper(senderClient);
+ }
+
+ @BeforeEach
+ void prepareReceiver() throws Exception {
+ final Properties properties = ServiceBusTestUtils.loadAzureAccessFromJvmEnv();
+
+ final ServiceBusReceiverAsyncClient receiverClient = new ServiceBusClientBuilder()
+ .connectionString(properties.getProperty(ServiceBusTestUtils.CONNECTION_STRING))
+ .receiver()
+ .topicName(properties.getProperty(ServiceBusTestUtils.TOPIC_NAME))
+ .subscriptionName(properties.getProperty(ServiceBusTestUtils.SUBSCRIPTION_NAME))
+ .buildAsyncClient();
+
+ clientReceiverWrapper = new ServiceBusReceiverAsyncClientWrapper(receiverClient);
+ }
+
+ @AfterAll
+ void closeClient() {
+ clientSenderWrapper.close();
+ }
+
+ @AfterEach
+ void closeSubscriber() {
+ clientReceiverWrapper.close();
+ }
+
+ @Test
+ void testSendSingleMessage() {
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(clientSenderWrapper);
+
+ operations.sendMessages("test data", null).block();
+
+ final boolean exists = StreamSupport.stream(clientReceiverWrapper.receiveMessages().toIterable().spliterator(), false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test data"));
+
+ assertTrue(exists, "test message body");
+
+ // test if we have something other than string or byte[]
+ assertThrows(IllegalArgumentException.class, () -> {
+ operations.sendMessages(12345, null).block();
+ });
+ }
+
+ @Test
+ void testSendingBatchMessages() {
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(clientSenderWrapper);
+
+ final List<String> inputBatch = new LinkedList<>();
+ inputBatch.add("test batch 1");
+ inputBatch.add("test batch 2");
+ inputBatch.add("test batch 3");
+
+ operations.sendMessages(inputBatch, null).block();
+
+ final Spliterator<ServiceBusReceivedMessage> receivedMessages
+ = clientReceiverWrapper.receiveMessages().toIterable().spliterator();
+
+ final boolean batch1Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 1"));
+
+ final boolean batch2Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 2"));
+
+ final boolean batch3Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test batch 3"));
+
+ assertTrue(batch1Exists, "test message body 1");
+ assertTrue(batch2Exists, "test message body 2");
+ assertTrue(batch3Exists, "test message body 3");
+ }
+
+ @Test
+ void testScheduleMessage() {
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(clientSenderWrapper);
+
+ operations.scheduleMessages("testScheduleMessage", OffsetDateTime.now(), null).block();
+
+ final boolean exists = StreamSupport.stream(clientReceiverWrapper.receiveMessages().toIterable().spliterator(), false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString()
+ .equals("testScheduleMessage"));
+
+ assertTrue(exists, "test message body");
+
+ // test if we have something other than string or byte[]
+ assertThrows(IllegalArgumentException.class, () -> {
+ operations.scheduleMessages(12345, OffsetDateTime.now(), null).block();
+ });
+ }
+
+ @Test
+ void testSchedulingBatchMessages() {
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(clientSenderWrapper);
+
+ final List<String> inputBatch = new LinkedList<>();
+ inputBatch.add("testSchedulingBatchMessages 1");
+ inputBatch.add("testSchedulingBatchMessages 2");
+ inputBatch.add("testSchedulingBatchMessages 3");
+
+ operations.scheduleMessages(inputBatch, OffsetDateTime.now(), null).block();
+
+ final Spliterator<ServiceBusReceivedMessage> receivedMessages
+ = clientReceiverWrapper.receiveMessages().toIterable().spliterator();
+
+ final boolean batch1Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString()
+ .equals("testSchedulingBatchMessages 1"));
+
+ final boolean batch2Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString()
+ .equals("testSchedulingBatchMessages 2"));
+
+ final boolean batch3Exists = StreamSupport.stream(receivedMessages, false)
+ .anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString()
+ .equals("testSchedulingBatchMessages 3"));
+
+ assertTrue(batch1Exists, "test message body 1");
+ assertTrue(batch2Exists, "test message body 2");
+ assertTrue(batch3Exists, "test message body 3");
+ }
+}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/resources/log4j2.properties b/components/camel-azure/camel-azure-servicebus/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..91c0226
--- /dev/null
+++ b/components/camel-azure/camel-azure-servicebus/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-servicebus-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
diff --git a/components/camel-azure/pom.xml b/components/camel-azure/pom.xml
index aa5c17c..4efdaf1 100644
--- a/components/camel-azure/pom.xml
+++ b/components/camel-azure/pom.xml
@@ -38,6 +38,7 @@
<module>camel-azure-storage-datalake</module>
<module>camel-azure-storage-queue</module>
<module>camel-azure-cosmosdb</module>
+ <module>camel-azure-servicebus</module>
</modules>
</project>
diff --git a/core/camel-allcomponents/pom.xml b/core/camel-allcomponents/pom.xml
index 4f164c0..69018c3 100644
--- a/core/camel-allcomponents/pom.xml
+++ b/core/camel-allcomponents/pom.xml
@@ -207,6 +207,10 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-azure-servicebus</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 e5acea1..fd49e66 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
@@ -615,6 +615,24 @@ public interface ComponentsBuilderFactory {
return org.apache.camel.builder.component.dsl.AzureEventhubsComponentBuilderFactory.azureEventhubs();
}
/**
+ * Azure ServiceBus (camel-azure-servicebus)
+ * The azure-servicebus component that integrates Azure ServiceBus. Azure
+ * ServiceBus is a fully managed enterprise integration message broker.
+ * Service Bus can decouple applications and services. Service Bus offers a
+ * reliable and secure platform for asynchronous transfer of data and state.
+ * Data is transferred between different applications and services using
+ * messages.
+ *
+ * Category: cloud,messaging
+ * Since: 3.12
+ * Maven coordinates: org.apache.camel:camel-azure-servicebus
+ *
+ * @return the dsl builder
+ */
+ static org.apache.camel.builder.component.dsl.AzureServicebusComponentBuilderFactory.AzureServicebusComponentBuilder azureServicebus() {
+ return org.apache.camel.builder.component.dsl.AzureServicebusComponentBuilderFactory.azureServicebus();
+ }
+ /**
* Azure Storage Blob Service (camel-azure-storage-blob)
* Store and retrieve blobs from Azure Storage Blob Service using SDK v12.
*
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureServicebusComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureServicebusComponentBuilderFactory.java
new file mode 100644
index 0000000..0000b25
--- /dev/null
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/AzureServicebusComponentBuilderFactory.java
@@ -0,0 +1,532 @@
+/*
+ * 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.servicebus.ServiceBusComponent;
+
+/**
+ * The azure-servicebus component that integrates Azure ServiceBus. Azure
+ * ServiceBus is a fully managed enterprise integration message broker. Service
+ * Bus can decouple applications and services. Service Bus offers a reliable and
+ * secure platform for asynchronous transfer of data and state. Data is
+ * transferred between different applications and services using messages.
+ *
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.ComponentDslMojo")
+public interface AzureServicebusComponentBuilderFactory {
+
+ /**
+ * Azure ServiceBus (camel-azure-servicebus)
+ * The azure-servicebus component that integrates Azure ServiceBus. Azure
+ * ServiceBus is a fully managed enterprise integration message broker.
+ * Service Bus can decouple applications and services. Service Bus offers a
+ * reliable and secure platform for asynchronous transfer of data and state.
+ * Data is transferred between different applications and services using
+ * messages.
+ *
+ * Category: cloud,messaging
+ * Since: 3.12
+ * Maven coordinates: org.apache.camel:camel-azure-servicebus
+ *
+ * @return the dsl builder
+ */
+ static AzureServicebusComponentBuilder azureServicebus() {
+ return new AzureServicebusComponentBuilderImpl();
+ }
+
+ /**
+ * Builder for the Azure ServiceBus component.
+ */
+ interface AzureServicebusComponentBuilder
+ extends
+ ComponentBuilder<ServiceBusComponent> {
+ /**
+ * Sets the retry options for Service Bus clients. If not specified, the
+ * default retry options are used.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.AmqpRetryOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param amqpRetryOptions the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder amqpRetryOptions(
+ com.azure.core.amqp.AmqpRetryOptions amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Service Bus occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.AmqpTransportType</code> type.
+ *
+ * Default: AMQP
+ * Group: common
+ *
+ * @param amqpTransportType the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder amqpTransportType(
+ com.azure.core.amqp.AmqpTransportType amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the ClientOptions to be sent from the client built from this
+ * builder, enabling customization of certain properties, as well as
+ * support the addition of custom header information. Refer to the
+ * ClientOptions documentation for more information.
+ *
+ * The option is a:
+ * <code>com.azure.core.util.ClientOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param clientOptions the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder clientOptions(
+ com.azure.core.util.ClientOptions clientOptions) {
+ doSetProperty("clientOptions", clientOptions);
+ return this;
+ }
+ /**
+ * The component configurations.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusConfiguration</code> type.
+ *
+ * Group: common
+ *
+ * @param configuration the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder configuration(
+ org.apache.camel.component.azure.servicebus.ServiceBusConfiguration configuration) {
+ doSetProperty("configuration", configuration);
+ return this;
+ }
+ /**
+ * Sets the proxy configuration to use for ServiceBusSenderAsyncClient.
+ * When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must
+ * be used for the transport type.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.ProxyOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param proxyOptions the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder proxyOptions(
+ com.azure.core.amqp.ProxyOptions proxyOptions) {
+ doSetProperty("proxyOptions", proxyOptions);
+ return this;
+ }
+ /**
+ * The service bus type of connection to execute. Queue is for typical
+ * queue option and topic for subscription based model.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusType</code> type.
+ *
+ * Default: queue
+ * Group: common
+ *
+ * @param serviceBusType the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder serviceBusType(
+ org.apache.camel.component.azure.servicebus.ServiceBusType serviceBusType) {
+ doSetProperty("serviceBusType", serviceBusType);
+ 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
+ *
+ * @param bridgeErrorHandler the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder bridgeErrorHandler(
+ boolean bridgeErrorHandler) {
+ doSetProperty("bridgeErrorHandler", bridgeErrorHandler);
+ return this;
+ }
+ /**
+ * Sets the desired operation to be used in the consumer.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition</code> type.
+ *
+ * Default: receiveMessages
+ * Group: consumer
+ *
+ * @param consumerOperation the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder consumerOperation(
+ org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition consumerOperation) {
+ doSetProperty("consumerOperation", consumerOperation);
+ return this;
+ }
+ /**
+ * Disables auto-complete and auto-abandon of received messages. By
+ * default, a successfully processed message is {link
+ * ServiceBusReceiverAsyncClient#complete(ServiceBusReceivedMessage)
+ * completed}. If an error happens when the message is processed, it is
+ * {link
+ * ServiceBusReceiverAsyncClient#abandon(ServiceBusReceivedMessage)
+ * abandoned}.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ *
+ * @param disableAutoComplete the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder disableAutoComplete(
+ boolean disableAutoComplete) {
+ doSetProperty("disableAutoComplete", disableAutoComplete);
+ return this;
+ }
+ /**
+ * Sets the amount of time to continue auto-renewing the lock. Setting
+ * Duration#ZERO or null disables auto-renewal. For {link
+ * ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} mode,
+ * auto-renewal is disabled.
+ *
+ * The option is a: <code>java.time.Duration</code> type.
+ *
+ * Default: 5min
+ * Group: consumer
+ *
+ * @param maxAutoLockRenewDuration the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder maxAutoLockRenewDuration(
+ java.time.Duration maxAutoLockRenewDuration) {
+ doSetProperty("maxAutoLockRenewDuration", maxAutoLockRenewDuration);
+ return this;
+ }
+ /**
+ * Set the max number of messages to be peeked during the peek
+ * operation.
+ *
+ * The option is a: <code>java.lang.Integer</code> type.
+ *
+ * Group: consumer
+ *
+ * @param peekNumMaxMessages the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder peekNumMaxMessages(
+ java.lang.Integer peekNumMaxMessages) {
+ doSetProperty("peekNumMaxMessages", peekNumMaxMessages);
+ return this;
+ }
+ /**
+ * Sets the prefetch count of the receiver. For both {link
+ * ServiceBusReceiveMode#PEEK_LOCK PEEK_LOCK} and {link
+ * ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} modes
+ * the default value is 1. Prefetch speeds up the message flow by aiming
+ * to have a message readily available for local retrieval when and
+ * before the application asks for one using
+ * ServiceBusReceiverAsyncClient#receiveMessages(). Setting a non-zero
+ * value will prefetch that number of messages. Setting the value to
+ * zero turns prefetch off.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 0
+ * Group: consumer
+ *
+ * @param prefetchCount the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder prefetchCount(int prefetchCount) {
+ doSetProperty("prefetchCount", prefetchCount);
+ return this;
+ }
+ /**
+ * Sets the receiverAsyncClient in order to consume messages in the
+ * Consumer.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient</code> type.
+ *
+ * Group: consumer
+ *
+ * @param receiverAsyncClient the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder receiverAsyncClient(
+ com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient receiverAsyncClient) {
+ doSetProperty("receiverAsyncClient", receiverAsyncClient);
+ return this;
+ }
+ /**
+ * Sets the receive mode for the receiver.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.models.ServiceBusReceiveMode</code> type.
+ *
+ * Default: PEER_LOCK
+ * Group: consumer
+ *
+ * @param serviceBusReceiveMode the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder serviceBusReceiveMode(
+ com.azure.messaging.servicebus.models.ServiceBusReceiveMode serviceBusReceiveMode) {
+ doSetProperty("serviceBusReceiveMode", serviceBusReceiveMode);
+ return this;
+ }
+ /**
+ * Sets the type of the SubQueue to connect to.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.models.SubQueue</code> type.
+ *
+ * Group: consumer
+ *
+ * @param subQueue the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder subQueue(
+ com.azure.messaging.servicebus.models.SubQueue subQueue) {
+ doSetProperty("subQueue", subQueue);
+ return this;
+ }
+ /**
+ * Sets the name of the subscription in the topic to listen to.
+ * topicOrQueueName and serviceBusType=topic must also be set.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ *
+ * @param subscriptionName the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder subscriptionName(
+ java.lang.String subscriptionName) {
+ doSetProperty("subscriptionName", subscriptionName);
+ 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
+ *
+ * @param lazyStartProducer the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder lazyStartProducer(
+ boolean lazyStartProducer) {
+ doSetProperty("lazyStartProducer", lazyStartProducer);
+ return this;
+ }
+ /**
+ * Sets the desired operation to be used in the producer.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition</code> type.
+ *
+ * Default: sendMessages
+ * Group: producer
+ *
+ * @param producerOperation the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder producerOperation(
+ org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition producerOperation) {
+ doSetProperty("producerOperation", producerOperation);
+ return this;
+ }
+ /**
+ * Sets OffsetDateTime at which the message should appear in the Service
+ * Bus queue or topic.
+ *
+ * The option is a: <code>java.time.OffsetDateTime</code>
+ * type.
+ *
+ * Group: producer
+ *
+ * @param scheduledEnqueueTime the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder scheduledEnqueueTime(
+ java.time.OffsetDateTime scheduledEnqueueTime) {
+ doSetProperty("scheduledEnqueueTime", scheduledEnqueueTime);
+ return this;
+ }
+ /**
+ * Sets SenderAsyncClient to be used in the producer.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.ServiceBusSenderAsyncClient</code> type.
+ *
+ * Group: producer
+ *
+ * @param senderAsyncClient the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder senderAsyncClient(
+ com.azure.messaging.servicebus.ServiceBusSenderAsyncClient senderAsyncClient) {
+ doSetProperty("senderAsyncClient", senderAsyncClient);
+ return this;
+ }
+ /**
+ * Represents transaction in service. This object just contains
+ * transaction id.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.ServiceBusTransactionContext</code> type.
+ *
+ * Group: producer
+ *
+ * @param serviceBusTransactionContext the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder serviceBusTransactionContext(
+ com.azure.messaging.servicebus.ServiceBusTransactionContext serviceBusTransactionContext) {
+ doSetProperty("serviceBusTransactionContext", serviceBusTransactionContext);
+ return this;
+ }
+ /**
+ * Whether autowiring is enabled. This is used for automatic autowiring
+ * options (the option must be marked as autowired) by looking up in the
+ * registry to find if there is a single instance of matching type,
+ * which then gets configured on the component. This can be used for
+ * automatic configuring JDBC data sources, JMS connection factories,
+ * AWS Clients, etc.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: true
+ * Group: advanced
+ *
+ * @param autowiredEnabled the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder autowiredEnabled(
+ boolean autowiredEnabled) {
+ doSetProperty("autowiredEnabled", autowiredEnabled);
+ return this;
+ }
+ /**
+ * Sets the connection string for a Service Bus namespace or a specific
+ * Service Bus resource.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: security
+ *
+ * @param connectionString the value to set
+ * @return the dsl builder
+ */
+ default AzureServicebusComponentBuilder connectionString(
+ java.lang.String connectionString) {
+ doSetProperty("connectionString", connectionString);
+ return this;
+ }
+ }
+
+ class AzureServicebusComponentBuilderImpl
+ extends
+ AbstractComponentBuilder<ServiceBusComponent>
+ implements
+ AzureServicebusComponentBuilder {
+ @Override
+ protected ServiceBusComponent buildConcreteComponent() {
+ return new ServiceBusComponent();
+ }
+ private org.apache.camel.component.azure.servicebus.ServiceBusConfiguration getOrCreateConfiguration(
+ org.apache.camel.component.azure.servicebus.ServiceBusComponent component) {
+ if (component.getConfiguration() == null) {
+ component.setConfiguration(new org.apache.camel.component.azure.servicebus.ServiceBusConfiguration());
+ }
+ return component.getConfiguration();
+ }
+ @Override
+ protected boolean setPropertyOnComponent(
+ Component component,
+ String name,
+ Object value) {
+ switch (name) {
+ case "amqpRetryOptions": getOrCreateConfiguration((ServiceBusComponent) component).setAmqpRetryOptions((com.azure.core.amqp.AmqpRetryOptions) value); return true;
+ case "amqpTransportType": getOrCreateConfiguration((ServiceBusComponent) component).setAmqpTransportType((com.azure.core.amqp.AmqpTransportType) value); return true;
+ case "clientOptions": getOrCreateConfiguration((ServiceBusComponent) component).setClientOptions((com.azure.core.util.ClientOptions) value); return true;
+ case "configuration": ((ServiceBusComponent) component).setConfiguration((org.apache.camel.component.azure.servicebus.ServiceBusConfiguration) value); return true;
+ case "proxyOptions": getOrCreateConfiguration((ServiceBusComponent) component).setProxyOptions((com.azure.core.amqp.ProxyOptions) value); return true;
+ case "serviceBusType": getOrCreateConfiguration((ServiceBusComponent) component).setServiceBusType((org.apache.camel.component.azure.servicebus.ServiceBusType) value); return true;
+ case "bridgeErrorHandler": ((ServiceBusComponent) component).setBridgeErrorHandler((boolean) value); return true;
+ case "consumerOperation": getOrCreateConfiguration((ServiceBusComponent) component).setConsumerOperation((org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition) value); return true;
+ case "disableAutoComplete": getOrCreateConfiguration((ServiceBusComponent) component).setDisableAutoComplete((boolean) value); return true;
+ case "maxAutoLockRenewDuration": getOrCreateConfiguration((ServiceBusComponent) component).setMaxAutoLockRenewDuration((java.time.Duration) value); return true;
+ case "peekNumMaxMessages": getOrCreateConfiguration((ServiceBusComponent) component).setPeekNumMaxMessages((java.lang.Integer) value); return true;
+ case "prefetchCount": getOrCreateConfiguration((ServiceBusComponent) component).setPrefetchCount((int) value); return true;
+ case "receiverAsyncClient": getOrCreateConfiguration((ServiceBusComponent) component).setReceiverAsyncClient((com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient) value); return true;
+ case "serviceBusReceiveMode": getOrCreateConfiguration((ServiceBusComponent) component).setServiceBusReceiveMode((com.azure.messaging.servicebus.models.ServiceBusReceiveMode) value); return true;
+ case "subQueue": getOrCreateConfiguration((ServiceBusComponent) component).setSubQueue((com.azure.messaging.servicebus.models.SubQueue) value); return true;
+ case "subscriptionName": getOrCreateConfiguration((ServiceBusComponent) component).setSubscriptionName((java.lang.String) value); return true;
+ case "lazyStartProducer": ((ServiceBusComponent) component).setLazyStartProducer((boolean) value); return true;
+ case "producerOperation": getOrCreateConfiguration((ServiceBusComponent) component).setProducerOperation((org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition) value); return true;
+ case "scheduledEnqueueTime": getOrCreateConfiguration((ServiceBusComponent) component).setScheduledEnqueueTime((java.time.OffsetDateTime) value); return true;
+ case "senderAsyncClient": getOrCreateConfiguration((ServiceBusComponent) component).setSenderAsyncClient((com.azure.messaging.servicebus.ServiceBusSenderAsyncClient) value); return true;
+ case "serviceBusTransactionContext": getOrCreateConfiguration((ServiceBusComponent) component).setServiceBusTransactionContext((com.azure.messaging.servicebus.ServiceBusTransactionContext) value); return true;
+ case "autowiredEnabled": ((ServiceBusComponent) component).setAutowiredEnabled((boolean) value); return true;
+ case "connectionString": getOrCreateConfiguration((ServiceBusComponent) component).setConnectionString((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 deef460..02b95e8 100644
--- a/core/camel-componentdsl/src/generated/resources/metadata.json
+++ b/core/camel-componentdsl/src/generated/resources/metadata.json
@@ -950,6 +950,28 @@
"producerOnly": false,
"lenientProperties": false
},
+ "AzureServicebusComponentBuilderFactory": {
+ "kind": "component",
+ "name": "azure-servicebus",
+ "title": "Azure ServiceBus",
+ "description": "The azure-servicebus component that integrates Azure ServiceBus. Azure ServiceBus is a fully managed enterprise integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and secure platform for asynchronous transfer of data and state. Data is transferred between different applications and services using messages.",
+ "deprecated": false,
+ "firstVersion": "3.12.0",
+ "label": "cloud,messaging",
+ "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusComponent",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-azure-servicebus",
+ "version": "3.12.0-SNAPSHOT",
+ "scheme": "azure-servicebus",
+ "extendsScheme": "",
+ "syntax": "azure-servicebus:queueNameOrTopicName",
+ "async": false,
+ "api": false,
+ "consumerOnly": false,
+ "producerOnly": false,
+ "lenientProperties": false
+ },
"AzureStorageBlobComponentBuilderFactory": {
"kind": "component",
"name": "azure-storage-blob",
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 2e6bff8..1ea0291 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
@@ -293,6 +293,7 @@ public interface EndpointBuilderFactory
org.apache.camel.builder.endpoint.dsl.SecretsManagerEndpointBuilderFactory.SecretsManagerBuilders,
org.apache.camel.builder.endpoint.dsl.SedaEndpointBuilderFactory.SedaBuilders,
org.apache.camel.builder.endpoint.dsl.ServerEndpointBuilderFactory.ServerBuilders,
+ org.apache.camel.builder.endpoint.dsl.ServiceBusEndpointBuilderFactory.ServiceBusBuilders,
org.apache.camel.builder.endpoint.dsl.ServiceEndpointBuilderFactory.ServiceBuilders,
org.apache.camel.builder.endpoint.dsl.ServiceNowEndpointBuilderFactory.ServiceNowBuilders,
org.apache.camel.builder.endpoint.dsl.ServletEndpointBuilderFactory.ServletBuilders,
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 772bedf..215fe6e 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
@@ -290,6 +290,7 @@ public interface EndpointBuilders
org.apache.camel.builder.endpoint.dsl.SecretsManagerEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.SedaEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.ServerEndpointBuilderFactory,
+ org.apache.camel.builder.endpoint.dsl.ServiceBusEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.ServiceEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.ServiceNowEndpointBuilderFactory,
org.apache.camel.builder.endpoint.dsl.ServletEndpointBuilderFactory,
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 a6d9115..4b7e089 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
@@ -1946,6 +1946,59 @@ public class StaticEndpointBuilders {
return org.apache.camel.builder.endpoint.dsl.EventHubsEndpointBuilderFactory.endpointBuilder(componentName, path);
}
/**
+ * Azure ServiceBus (camel-azure-servicebus)
+ * The azure-servicebus component that integrates Azure ServiceBus. Azure
+ * ServiceBus is a fully managed enterprise integration message broker.
+ * Service Bus can decouple applications and services. Service Bus offers a
+ * reliable and secure platform for asynchronous transfer of data and state.
+ * Data is transferred between different applications and services using
+ * messages.
+ *
+ * Category: cloud,messaging
+ * Since: 3.12
+ * Maven coordinates: org.apache.camel:camel-azure-servicebus
+ *
+ * Syntax: <code>azure-servicebus:queueNameOrTopicName</code>
+ *
+ * Path parameter: topicOrQueueName
+ * d
+ *
+ * @param path queueNameOrTopicName
+ * @return the dsl builder
+ */
+ public static org.apache.camel.builder.endpoint.dsl.ServiceBusEndpointBuilderFactory.ServiceBusEndpointBuilder azureServicebus(
+ String path) {
+ return org.apache.camel.builder.endpoint.dsl.ServiceBusEndpointBuilderFactory.endpointBuilder("azure-servicebus", path);
+ }
+ /**
+ * Azure ServiceBus (camel-azure-servicebus)
+ * The azure-servicebus component that integrates Azure ServiceBus. Azure
+ * ServiceBus is a fully managed enterprise integration message broker.
+ * Service Bus can decouple applications and services. Service Bus offers a
+ * reliable and secure platform for asynchronous transfer of data and state.
+ * Data is transferred between different applications and services using
+ * messages.
+ *
+ * Category: cloud,messaging
+ * Since: 3.12
+ * Maven coordinates: org.apache.camel:camel-azure-servicebus
+ *
+ * Syntax: <code>azure-servicebus:queueNameOrTopicName</code>
+ *
+ * Path parameter: topicOrQueueName
+ * d
+ *
+ * @param componentName to use a custom component name for the endpoint
+ * instead of the default name
+ * @param path queueNameOrTopicName
+ * @return the dsl builder
+ */
+ public static org.apache.camel.builder.endpoint.dsl.ServiceBusEndpointBuilderFactory.ServiceBusEndpointBuilder azureServicebus(
+ String componentName,
+ String path) {
+ return org.apache.camel.builder.endpoint.dsl.ServiceBusEndpointBuilderFactory.endpointBuilder(componentName, path);
+ }
+ /**
* Azure Storage Blob Service (camel-azure-storage-blob)
* Store and retrieve blobs from Azure Storage Blob Service using SDK v12.
*
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ServiceBusEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ServiceBusEndpointBuilderFactory.java
new file mode 100644
index 0000000..4dbfd3b
--- /dev/null
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ServiceBusEndpointBuilderFactory.java
@@ -0,0 +1,1440 @@
+/*
+ * 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.time.Duration;
+import java.time.OffsetDateTime;
+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-servicebus component that integrates Azure ServiceBus. Azure
+ * ServiceBus is a fully managed enterprise integration message broker. Service
+ * Bus can decouple applications and services. Service Bus offers a reliable and
+ * secure platform for asynchronous transfer of data and state. Data is
+ * transferred between different applications and services using messages.
+ *
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.EndpointDslMojo")
+public interface ServiceBusEndpointBuilderFactory {
+
+
+ /**
+ * Builder for endpoint consumers for the Azure ServiceBus component.
+ */
+ public interface ServiceBusEndpointConsumerBuilder
+ extends
+ EndpointConsumerBuilder {
+ default AdvancedServiceBusEndpointConsumerBuilder advanced() {
+ return (AdvancedServiceBusEndpointConsumerBuilder) this;
+ }
+ /**
+ * Sets the retry options for Service Bus clients. If not specified, the
+ * default retry options are used.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.AmqpRetryOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param amqpRetryOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder amqpRetryOptions(
+ Object amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the retry options for Service Bus clients. 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
+ *
+ * @param amqpRetryOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder amqpRetryOptions(
+ String amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Service Bus occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.AmqpTransportType</code> type.
+ *
+ * Default: AMQP
+ * Group: common
+ *
+ * @param amqpTransportType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder amqpTransportType(
+ AmqpTransportType amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Service Bus 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
+ *
+ * @param amqpTransportType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder amqpTransportType(
+ String amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the ClientOptions to be sent from the client built from this
+ * builder, enabling customization of certain properties, as well as
+ * support the addition of custom header information. Refer to the
+ * ClientOptions documentation for more information.
+ *
+ * The option is a:
+ * <code>com.azure.core.util.ClientOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param clientOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder clientOptions(
+ Object clientOptions) {
+ doSetProperty("clientOptions", clientOptions);
+ return this;
+ }
+ /**
+ * Sets the ClientOptions to be sent from the client built from this
+ * builder, enabling customization of certain properties, as well as
+ * support the addition of custom header information. Refer to the
+ * ClientOptions documentation for more information.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.util.ClientOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param clientOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder clientOptions(
+ String clientOptions) {
+ doSetProperty("clientOptions", clientOptions);
+ return this;
+ }
+ /**
+ * Sets the proxy configuration to use for ServiceBusSenderAsyncClient.
+ * When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must
+ * be used for the transport type.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.ProxyOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param proxyOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder proxyOptions(
+ Object proxyOptions) {
+ doSetProperty("proxyOptions", proxyOptions);
+ return this;
+ }
+ /**
+ * Sets the proxy configuration to use for ServiceBusSenderAsyncClient.
+ * When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must
+ * be used for the transport type.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.ProxyOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param proxyOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder proxyOptions(
+ String proxyOptions) {
+ doSetProperty("proxyOptions", proxyOptions);
+ return this;
+ }
+ /**
+ * The service bus type of connection to execute. Queue is for typical
+ * queue option and topic for subscription based model.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusType</code> type.
+ *
+ * Required: true
+ * Default: queue
+ * Group: common
+ *
+ * @param serviceBusType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder serviceBusType(
+ ServiceBusType serviceBusType) {
+ doSetProperty("serviceBusType", serviceBusType);
+ return this;
+ }
+ /**
+ * The service bus type of connection to execute. Queue is for typical
+ * queue option and topic for subscription based model.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusType</code> type.
+ *
+ * Required: true
+ * Default: queue
+ * Group: common
+ *
+ * @param serviceBusType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder serviceBusType(
+ String serviceBusType) {
+ doSetProperty("serviceBusType", serviceBusType);
+ 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
+ *
+ * @param bridgeErrorHandler the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder 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
+ *
+ * @param bridgeErrorHandler the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder bridgeErrorHandler(
+ String bridgeErrorHandler) {
+ doSetProperty("bridgeErrorHandler", bridgeErrorHandler);
+ return this;
+ }
+ /**
+ * Sets the desired operation to be used in the consumer.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition</code> type.
+ *
+ * Default: receiveMessages
+ * Group: consumer
+ *
+ * @param consumerOperation the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder consumerOperation(
+ ServiceBusConsumerOperationDefinition consumerOperation) {
+ doSetProperty("consumerOperation", consumerOperation);
+ return this;
+ }
+ /**
+ * Sets the desired operation to be used in the consumer.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition</code> type.
+ *
+ * Default: receiveMessages
+ * Group: consumer
+ *
+ * @param consumerOperation the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder consumerOperation(
+ String consumerOperation) {
+ doSetProperty("consumerOperation", consumerOperation);
+ return this;
+ }
+ /**
+ * Disables auto-complete and auto-abandon of received messages. By
+ * default, a successfully processed message is {link
+ * ServiceBusReceiverAsyncClient#complete(ServiceBusReceivedMessage)
+ * completed}. If an error happens when the message is processed, it is
+ * {link
+ * ServiceBusReceiverAsyncClient#abandon(ServiceBusReceivedMessage)
+ * abandoned}.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ *
+ * @param disableAutoComplete the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder disableAutoComplete(
+ boolean disableAutoComplete) {
+ doSetProperty("disableAutoComplete", disableAutoComplete);
+ return this;
+ }
+ /**
+ * Disables auto-complete and auto-abandon of received messages. By
+ * default, a successfully processed message is {link
+ * ServiceBusReceiverAsyncClient#complete(ServiceBusReceivedMessage)
+ * completed}. If an error happens when the message is processed, it is
+ * {link
+ * ServiceBusReceiverAsyncClient#abandon(ServiceBusReceivedMessage)
+ * abandoned}.
+ *
+ * The option will be converted to a <code>boolean</code>
+ * type.
+ *
+ * Default: false
+ * Group: consumer
+ *
+ * @param disableAutoComplete the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder disableAutoComplete(
+ String disableAutoComplete) {
+ doSetProperty("disableAutoComplete", disableAutoComplete);
+ return this;
+ }
+ /**
+ * Sets the amount of time to continue auto-renewing the lock. Setting
+ * Duration#ZERO or null disables auto-renewal. For {link
+ * ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} mode,
+ * auto-renewal is disabled.
+ *
+ * The option is a: <code>java.time.Duration</code> type.
+ *
+ * Default: 5min
+ * Group: consumer
+ *
+ * @param maxAutoLockRenewDuration the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder maxAutoLockRenewDuration(
+ Duration maxAutoLockRenewDuration) {
+ doSetProperty("maxAutoLockRenewDuration", maxAutoLockRenewDuration);
+ return this;
+ }
+ /**
+ * Sets the amount of time to continue auto-renewing the lock. Setting
+ * Duration#ZERO or null disables auto-renewal. For {link
+ * ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} mode,
+ * auto-renewal is disabled.
+ *
+ * The option will be converted to a
+ * <code>java.time.Duration</code> type.
+ *
+ * Default: 5min
+ * Group: consumer
+ *
+ * @param maxAutoLockRenewDuration the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder maxAutoLockRenewDuration(
+ String maxAutoLockRenewDuration) {
+ doSetProperty("maxAutoLockRenewDuration", maxAutoLockRenewDuration);
+ return this;
+ }
+ /**
+ * Set the max number of messages to be peeked during the peek
+ * operation.
+ *
+ * The option is a: <code>java.lang.Integer</code> type.
+ *
+ * Group: consumer
+ *
+ * @param peekNumMaxMessages the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder peekNumMaxMessages(
+ Integer peekNumMaxMessages) {
+ doSetProperty("peekNumMaxMessages", peekNumMaxMessages);
+ return this;
+ }
+ /**
+ * Set the max number of messages to be peeked during the peek
+ * operation.
+ *
+ * The option will be converted to a
+ * <code>java.lang.Integer</code> type.
+ *
+ * Group: consumer
+ *
+ * @param peekNumMaxMessages the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder peekNumMaxMessages(
+ String peekNumMaxMessages) {
+ doSetProperty("peekNumMaxMessages", peekNumMaxMessages);
+ return this;
+ }
+ /**
+ * Sets the prefetch count of the receiver. For both {link
+ * ServiceBusReceiveMode#PEEK_LOCK PEEK_LOCK} and {link
+ * ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} modes
+ * the default value is 1. Prefetch speeds up the message flow by aiming
+ * to have a message readily available for local retrieval when and
+ * before the application asks for one using
+ * ServiceBusReceiverAsyncClient#receiveMessages(). Setting a non-zero
+ * value will prefetch that number of messages. Setting the value to
+ * zero turns prefetch off.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 0
+ * Group: consumer
+ *
+ * @param prefetchCount the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder prefetchCount(
+ int prefetchCount) {
+ doSetProperty("prefetchCount", prefetchCount);
+ return this;
+ }
+ /**
+ * Sets the prefetch count of the receiver. For both {link
+ * ServiceBusReceiveMode#PEEK_LOCK PEEK_LOCK} and {link
+ * ServiceBusReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} modes
+ * the default value is 1. Prefetch speeds up the message flow by aiming
+ * to have a message readily available for local retrieval when and
+ * before the application asks for one using
+ * ServiceBusReceiverAsyncClient#receiveMessages(). Setting a non-zero
+ * value will prefetch that number of messages. Setting the value to
+ * zero turns prefetch off.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Default: 0
+ * Group: consumer
+ *
+ * @param prefetchCount the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder prefetchCount(
+ String prefetchCount) {
+ doSetProperty("prefetchCount", prefetchCount);
+ return this;
+ }
+ /**
+ * Sets the receiverAsyncClient in order to consume messages in the
+ * Consumer.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient</code> type.
+ *
+ * Group: consumer
+ *
+ * @param receiverAsyncClient the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder receiverAsyncClient(
+ Object receiverAsyncClient) {
+ doSetProperty("receiverAsyncClient", receiverAsyncClient);
+ return this;
+ }
+ /**
+ * Sets the receiverAsyncClient in order to consume messages in the
+ * Consumer.
+ *
+ * The option will be converted to a
+ * <code>com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient</code> type.
+ *
+ * Group: consumer
+ *
+ * @param receiverAsyncClient the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder receiverAsyncClient(
+ String receiverAsyncClient) {
+ doSetProperty("receiverAsyncClient", receiverAsyncClient);
+ return this;
+ }
+ /**
+ * Sets the receive mode for the receiver.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.models.ServiceBusReceiveMode</code> type.
+ *
+ * Default: PEER_LOCK
+ * Group: consumer
+ *
+ * @param serviceBusReceiveMode the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder serviceBusReceiveMode(
+ ServiceBusReceiveMode serviceBusReceiveMode) {
+ doSetProperty("serviceBusReceiveMode", serviceBusReceiveMode);
+ return this;
+ }
+ /**
+ * Sets the receive mode for the receiver.
+ *
+ * The option will be converted to a
+ * <code>com.azure.messaging.servicebus.models.ServiceBusReceiveMode</code> type.
+ *
+ * Default: PEER_LOCK
+ * Group: consumer
+ *
+ * @param serviceBusReceiveMode the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder serviceBusReceiveMode(
+ String serviceBusReceiveMode) {
+ doSetProperty("serviceBusReceiveMode", serviceBusReceiveMode);
+ return this;
+ }
+ /**
+ * Sets the type of the SubQueue to connect to.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.models.SubQueue</code> type.
+ *
+ * Group: consumer
+ *
+ * @param subQueue the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder subQueue(SubQueue subQueue) {
+ doSetProperty("subQueue", subQueue);
+ return this;
+ }
+ /**
+ * Sets the type of the SubQueue to connect to.
+ *
+ * The option will be converted to a
+ * <code>com.azure.messaging.servicebus.models.SubQueue</code> type.
+ *
+ * Group: consumer
+ *
+ * @param subQueue the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder subQueue(String subQueue) {
+ doSetProperty("subQueue", subQueue);
+ return this;
+ }
+ /**
+ * Sets the name of the subscription in the topic to listen to.
+ * topicOrQueueName and serviceBusType=topic must also be set.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ *
+ * @param subscriptionName the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder subscriptionName(
+ String subscriptionName) {
+ doSetProperty("subscriptionName", subscriptionName);
+ return this;
+ }
+ /**
+ * Sets the connection string for a Service Bus namespace or a specific
+ * Service Bus resource.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Required: true
+ * Group: security
+ *
+ * @param connectionString the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointConsumerBuilder connectionString(
+ String connectionString) {
+ doSetProperty("connectionString", connectionString);
+ return this;
+ }
+ }
+
+ /**
+ * Advanced builder for endpoint consumers for the Azure ServiceBus
+ * component.
+ */
+ public interface AdvancedServiceBusEndpointConsumerBuilder
+ extends
+ EndpointConsumerBuilder {
+ default ServiceBusEndpointConsumerBuilder basic() {
+ return (ServiceBusEndpointConsumerBuilder) 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)
+ *
+ * @param exceptionHandler the value to set
+ * @return the dsl builder
+ */
+ default AdvancedServiceBusEndpointConsumerBuilder 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)
+ *
+ * @param exceptionHandler the value to set
+ * @return the dsl builder
+ */
+ default AdvancedServiceBusEndpointConsumerBuilder 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)
+ *
+ * @param exchangePattern the value to set
+ * @return the dsl builder
+ */
+ default AdvancedServiceBusEndpointConsumerBuilder 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)
+ *
+ * @param exchangePattern the value to set
+ * @return the dsl builder
+ */
+ default AdvancedServiceBusEndpointConsumerBuilder exchangePattern(
+ String exchangePattern) {
+ doSetProperty("exchangePattern", exchangePattern);
+ return this;
+ }
+ }
+
+ /**
+ * Builder for endpoint producers for the Azure ServiceBus component.
+ */
+ public interface ServiceBusEndpointProducerBuilder
+ extends
+ EndpointProducerBuilder {
+ default AdvancedServiceBusEndpointProducerBuilder advanced() {
+ return (AdvancedServiceBusEndpointProducerBuilder) this;
+ }
+ /**
+ * Sets the retry options for Service Bus clients. If not specified, the
+ * default retry options are used.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.AmqpRetryOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param amqpRetryOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder amqpRetryOptions(
+ Object amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the retry options for Service Bus clients. 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
+ *
+ * @param amqpRetryOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder amqpRetryOptions(
+ String amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Service Bus occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.AmqpTransportType</code> type.
+ *
+ * Default: AMQP
+ * Group: common
+ *
+ * @param amqpTransportType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder amqpTransportType(
+ AmqpTransportType amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Service Bus 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
+ *
+ * @param amqpTransportType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder amqpTransportType(
+ String amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the ClientOptions to be sent from the client built from this
+ * builder, enabling customization of certain properties, as well as
+ * support the addition of custom header information. Refer to the
+ * ClientOptions documentation for more information.
+ *
+ * The option is a:
+ * <code>com.azure.core.util.ClientOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param clientOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder clientOptions(
+ Object clientOptions) {
+ doSetProperty("clientOptions", clientOptions);
+ return this;
+ }
+ /**
+ * Sets the ClientOptions to be sent from the client built from this
+ * builder, enabling customization of certain properties, as well as
+ * support the addition of custom header information. Refer to the
+ * ClientOptions documentation for more information.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.util.ClientOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param clientOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder clientOptions(
+ String clientOptions) {
+ doSetProperty("clientOptions", clientOptions);
+ return this;
+ }
+ /**
+ * Sets the proxy configuration to use for ServiceBusSenderAsyncClient.
+ * When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must
+ * be used for the transport type.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.ProxyOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param proxyOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder proxyOptions(
+ Object proxyOptions) {
+ doSetProperty("proxyOptions", proxyOptions);
+ return this;
+ }
+ /**
+ * Sets the proxy configuration to use for ServiceBusSenderAsyncClient.
+ * When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must
+ * be used for the transport type.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.ProxyOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param proxyOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder proxyOptions(
+ String proxyOptions) {
+ doSetProperty("proxyOptions", proxyOptions);
+ return this;
+ }
+ /**
+ * The service bus type of connection to execute. Queue is for typical
+ * queue option and topic for subscription based model.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusType</code> type.
+ *
+ * Required: true
+ * Default: queue
+ * Group: common
+ *
+ * @param serviceBusType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder serviceBusType(
+ ServiceBusType serviceBusType) {
+ doSetProperty("serviceBusType", serviceBusType);
+ return this;
+ }
+ /**
+ * The service bus type of connection to execute. Queue is for typical
+ * queue option and topic for subscription based model.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusType</code> type.
+ *
+ * Required: true
+ * Default: queue
+ * Group: common
+ *
+ * @param serviceBusType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder serviceBusType(
+ String serviceBusType) {
+ doSetProperty("serviceBusType", serviceBusType);
+ 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
+ *
+ * @param lazyStartProducer the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder 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
+ *
+ * @param lazyStartProducer the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder lazyStartProducer(
+ String lazyStartProducer) {
+ doSetProperty("lazyStartProducer", lazyStartProducer);
+ return this;
+ }
+ /**
+ * Sets the desired operation to be used in the producer.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition</code> type.
+ *
+ * Default: sendMessages
+ * Group: producer
+ *
+ * @param producerOperation the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder producerOperation(
+ ServiceBusProducerOperationDefinition producerOperation) {
+ doSetProperty("producerOperation", producerOperation);
+ return this;
+ }
+ /**
+ * Sets the desired operation to be used in the producer.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition</code> type.
+ *
+ * Default: sendMessages
+ * Group: producer
+ *
+ * @param producerOperation the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder producerOperation(
+ String producerOperation) {
+ doSetProperty("producerOperation", producerOperation);
+ return this;
+ }
+ /**
+ * Sets OffsetDateTime at which the message should appear in the Service
+ * Bus queue or topic.
+ *
+ * The option is a: <code>java.time.OffsetDateTime</code>
+ * type.
+ *
+ * Group: producer
+ *
+ * @param scheduledEnqueueTime the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder scheduledEnqueueTime(
+ OffsetDateTime scheduledEnqueueTime) {
+ doSetProperty("scheduledEnqueueTime", scheduledEnqueueTime);
+ return this;
+ }
+ /**
+ * Sets OffsetDateTime at which the message should appear in the Service
+ * Bus queue or topic.
+ *
+ * The option will be converted to a
+ * <code>java.time.OffsetDateTime</code> type.
+ *
+ * Group: producer
+ *
+ * @param scheduledEnqueueTime the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder scheduledEnqueueTime(
+ String scheduledEnqueueTime) {
+ doSetProperty("scheduledEnqueueTime", scheduledEnqueueTime);
+ return this;
+ }
+ /**
+ * Sets SenderAsyncClient to be used in the producer.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.ServiceBusSenderAsyncClient</code> type.
+ *
+ * Group: producer
+ *
+ * @param senderAsyncClient the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder senderAsyncClient(
+ Object senderAsyncClient) {
+ doSetProperty("senderAsyncClient", senderAsyncClient);
+ return this;
+ }
+ /**
+ * Sets SenderAsyncClient to be used in the producer.
+ *
+ * The option will be converted to a
+ * <code>com.azure.messaging.servicebus.ServiceBusSenderAsyncClient</code> type.
+ *
+ * Group: producer
+ *
+ * @param senderAsyncClient the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder senderAsyncClient(
+ String senderAsyncClient) {
+ doSetProperty("senderAsyncClient", senderAsyncClient);
+ return this;
+ }
+ /**
+ * Represents transaction in service. This object just contains
+ * transaction id.
+ *
+ * The option is a:
+ * <code>com.azure.messaging.servicebus.ServiceBusTransactionContext</code> type.
+ *
+ * Group: producer
+ *
+ * @param serviceBusTransactionContext the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder serviceBusTransactionContext(
+ Object serviceBusTransactionContext) {
+ doSetProperty("serviceBusTransactionContext", serviceBusTransactionContext);
+ return this;
+ }
+ /**
+ * Represents transaction in service. This object just contains
+ * transaction id.
+ *
+ * The option will be converted to a
+ * <code>com.azure.messaging.servicebus.ServiceBusTransactionContext</code> type.
+ *
+ * Group: producer
+ *
+ * @param serviceBusTransactionContext the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder serviceBusTransactionContext(
+ String serviceBusTransactionContext) {
+ doSetProperty("serviceBusTransactionContext", serviceBusTransactionContext);
+ return this;
+ }
+ /**
+ * Sets the connection string for a Service Bus namespace or a specific
+ * Service Bus resource.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Required: true
+ * Group: security
+ *
+ * @param connectionString the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointProducerBuilder connectionString(
+ String connectionString) {
+ doSetProperty("connectionString", connectionString);
+ return this;
+ }
+ }
+
+ /**
+ * Advanced builder for endpoint producers for the Azure ServiceBus
+ * component.
+ */
+ public interface AdvancedServiceBusEndpointProducerBuilder
+ extends
+ EndpointProducerBuilder {
+ default ServiceBusEndpointProducerBuilder basic() {
+ return (ServiceBusEndpointProducerBuilder) this;
+ }
+ }
+
+ /**
+ * Builder for endpoint for the Azure ServiceBus component.
+ */
+ public interface ServiceBusEndpointBuilder
+ extends
+ ServiceBusEndpointConsumerBuilder,
+ ServiceBusEndpointProducerBuilder {
+ default AdvancedServiceBusEndpointBuilder advanced() {
+ return (AdvancedServiceBusEndpointBuilder) this;
+ }
+ /**
+ * Sets the retry options for Service Bus clients. If not specified, the
+ * default retry options are used.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.AmqpRetryOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param amqpRetryOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder amqpRetryOptions(
+ Object amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the retry options for Service Bus clients. 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
+ *
+ * @param amqpRetryOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder amqpRetryOptions(
+ String amqpRetryOptions) {
+ doSetProperty("amqpRetryOptions", amqpRetryOptions);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Service Bus occurs. Default value is AmqpTransportType#AMQP.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.AmqpTransportType</code> type.
+ *
+ * Default: AMQP
+ * Group: common
+ *
+ * @param amqpTransportType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder amqpTransportType(
+ AmqpTransportType amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the transport type by which all the communication with Azure
+ * Service Bus 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
+ *
+ * @param amqpTransportType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder amqpTransportType(
+ String amqpTransportType) {
+ doSetProperty("amqpTransportType", amqpTransportType);
+ return this;
+ }
+ /**
+ * Sets the ClientOptions to be sent from the client built from this
+ * builder, enabling customization of certain properties, as well as
+ * support the addition of custom header information. Refer to the
+ * ClientOptions documentation for more information.
+ *
+ * The option is a:
+ * <code>com.azure.core.util.ClientOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param clientOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder clientOptions(Object clientOptions) {
+ doSetProperty("clientOptions", clientOptions);
+ return this;
+ }
+ /**
+ * Sets the ClientOptions to be sent from the client built from this
+ * builder, enabling customization of certain properties, as well as
+ * support the addition of custom header information. Refer to the
+ * ClientOptions documentation for more information.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.util.ClientOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param clientOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder clientOptions(String clientOptions) {
+ doSetProperty("clientOptions", clientOptions);
+ return this;
+ }
+ /**
+ * Sets the proxy configuration to use for ServiceBusSenderAsyncClient.
+ * When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must
+ * be used for the transport type.
+ *
+ * The option is a:
+ * <code>com.azure.core.amqp.ProxyOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param proxyOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder proxyOptions(Object proxyOptions) {
+ doSetProperty("proxyOptions", proxyOptions);
+ return this;
+ }
+ /**
+ * Sets the proxy configuration to use for ServiceBusSenderAsyncClient.
+ * When a proxy is configured, AmqpTransportType#AMQP_WEB_SOCKETS must
+ * be used for the transport type.
+ *
+ * The option will be converted to a
+ * <code>com.azure.core.amqp.ProxyOptions</code> type.
+ *
+ * Group: common
+ *
+ * @param proxyOptions the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder proxyOptions(String proxyOptions) {
+ doSetProperty("proxyOptions", proxyOptions);
+ return this;
+ }
+ /**
+ * The service bus type of connection to execute. Queue is for typical
+ * queue option and topic for subscription based model.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusType</code> type.
+ *
+ * Required: true
+ * Default: queue
+ * Group: common
+ *
+ * @param serviceBusType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder serviceBusType(
+ ServiceBusType serviceBusType) {
+ doSetProperty("serviceBusType", serviceBusType);
+ return this;
+ }
+ /**
+ * The service bus type of connection to execute. Queue is for typical
+ * queue option and topic for subscription based model.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusType</code> type.
+ *
+ * Required: true
+ * Default: queue
+ * Group: common
+ *
+ * @param serviceBusType the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder serviceBusType(String serviceBusType) {
+ doSetProperty("serviceBusType", serviceBusType);
+ return this;
+ }
+ /**
+ * Sets the connection string for a Service Bus namespace or a specific
+ * Service Bus resource.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Required: true
+ * Group: security
+ *
+ * @param connectionString the value to set
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder connectionString(
+ String connectionString) {
+ doSetProperty("connectionString", connectionString);
+ return this;
+ }
+ }
+
+ /**
+ * Advanced builder for endpoint for the Azure ServiceBus component.
+ */
+ public interface AdvancedServiceBusEndpointBuilder
+ extends
+ AdvancedServiceBusEndpointConsumerBuilder,
+ AdvancedServiceBusEndpointProducerBuilder {
+ default ServiceBusEndpointBuilder basic() {
+ return (ServiceBusEndpointBuilder) this;
+ }
+ }
+
+ /**
+ * Proxy enum for <code>com.azure.core.amqp.AmqpTransportType</code> enum.
+ */
+ enum AmqpTransportType {
+ AMQP,
+ AMQP_WEB_SOCKETS;
+ }
+
+ /**
+ * Proxy enum for
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusType</code>
+ * enum.
+ */
+ enum ServiceBusType {
+ queue,
+ topic;
+ }
+
+ /**
+ * Proxy enum for
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusConsumerOperationDefinition</code> enum.
+ */
+ enum ServiceBusConsumerOperationDefinition {
+ receiveMessages,
+ peekMessages;
+ }
+
+ /**
+ * Proxy enum for
+ * <code>com.azure.messaging.servicebus.models.ServiceBusReceiveMode</code>
+ * enum.
+ */
+ enum ServiceBusReceiveMode {
+ PEEK_LOCK,
+ RECEIVE_AND_DELETE;
+ }
+
+ /**
+ * Proxy enum for
+ * <code>com.azure.messaging.servicebus.models.SubQueue</code> enum.
+ */
+ enum SubQueue {
+ NONE,
+ DEAD_LETTER_QUEUE,
+ TRANSFER_DEAD_LETTER_QUEUE;
+ }
+
+ /**
+ * Proxy enum for
+ * <code>org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition</code> enum.
+ */
+ enum ServiceBusProducerOperationDefinition {
+ sendMessages,
+ scheduleMessages;
+ }
+
+ public interface ServiceBusBuilders {
+ /**
+ * Azure ServiceBus (camel-azure-servicebus)
+ * The azure-servicebus component that integrates Azure ServiceBus.
+ * Azure ServiceBus is a fully managed enterprise integration message
+ * broker. Service Bus can decouple applications and services. Service
+ * Bus offers a reliable and secure platform for asynchronous transfer
+ * of data and state. Data is transferred between different applications
+ * and services using messages.
+ *
+ * Category: cloud,messaging
+ * Since: 3.12
+ * Maven coordinates: org.apache.camel:camel-azure-servicebus
+ *
+ * Syntax: <code>azure-servicebus:queueNameOrTopicName</code>
+ *
+ * Path parameter: topicOrQueueName
+ * Selected topic name or the queue name, that is depending on
+ * serviceBusType config. For example if serviceBusType=queue, then this
+ * will be the queue name and if serviceBusType=topic, this will be the
+ * topic name.
+ *
+ * @param path queueNameOrTopicName
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder azureServicebus(String path) {
+ return ServiceBusEndpointBuilderFactory.endpointBuilder("azure-servicebus", path);
+ }
+ /**
+ * Azure ServiceBus (camel-azure-servicebus)
+ * The azure-servicebus component that integrates Azure ServiceBus.
+ * Azure ServiceBus is a fully managed enterprise integration message
+ * broker. Service Bus can decouple applications and services. Service
+ * Bus offers a reliable and secure platform for asynchronous transfer
+ * of data and state. Data is transferred between different applications
+ * and services using messages.
+ *
+ * Category: cloud,messaging
+ * Since: 3.12
+ * Maven coordinates: org.apache.camel:camel-azure-servicebus
+ *
+ * Syntax: <code>azure-servicebus:queueNameOrTopicName</code>
+ *
+ * Path parameter: topicOrQueueName
+ * Selected topic name or the queue name, that is depending on
+ * serviceBusType config. For example if serviceBusType=queue, then this
+ * will be the queue name and if serviceBusType=topic, this will be the
+ * topic name.
+ *
+ * @param componentName to use a custom component name for the endpoint
+ * instead of the default name
+ * @param path queueNameOrTopicName
+ * @return the dsl builder
+ */
+ default ServiceBusEndpointBuilder azureServicebus(
+ String componentName,
+ String path) {
+ return ServiceBusEndpointBuilderFactory.endpointBuilder(componentName, path);
+ }
+ }
+ static ServiceBusEndpointBuilder endpointBuilder(
+ String componentName,
+ String path) {
+ class ServiceBusEndpointBuilderImpl extends AbstractEndpointBuilder implements ServiceBusEndpointBuilder, AdvancedServiceBusEndpointBuilder {
+ public ServiceBusEndpointBuilderImpl(String path) {
+ super(componentName, path);
+ }
+ }
+ return new ServiceBusEndpointBuilderImpl(path);
+ }
+}
\ No newline at end of file
diff --git a/docs/components/modules/ROOT/pages/azure-servicebus-component.adoc b/docs/components/modules/ROOT/pages/azure-servicebus-component.adoc
new file mode 100644
index 0000000..a4da533
--- /dev/null
+++ b/docs/components/modules/ROOT/pages/azure-servicebus-component.adoc
@@ -0,0 +1,31 @@
+[[azure-servicebus-component]]
+= Azure ServiceBus Component
+//THIS FILE IS COPIED: EDIT THE SOURCE FILE:
+:page-source: components/camel-azure/camel-azure-servicebus/src/main/docs/azure-servicebus-component.adoc
+:docTitle: Azure ServiceBus
+:artifactId: camel-azure-servicebus
+:description: The azure-servicebus component that integrates Azure ServiceBus. Azure ServiceBus is a fully managed enterprise integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and secure platform for asynchronous transfer of data and state. Data is transferred between different applications and services using messages.
+:since: 3.12
+:supportLevel: Preview
+:component-header: Both producer and consumer are supported
+include::{cq-version}@camel-quarkus:ROOT:partial$reference/components/azure-servicebus.adoc[opts=optional]
+//Manually maintained attributes
+:group: Azure
+
+*Since Camel {since}*
+
+*{component-header}*
+
+```
+
+=== 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::{page-component-version}@camel-spring-boot::page$azure-servicebus-starter.adoc[]
diff --git a/parent/pom.xml b/parent/pom.xml
index 32c5eae..a388fd0 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -1005,6 +1005,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-azure-servicebus</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-azure-storage-blob</artifactId>
<version>${project.version}</version>
</dependency>