You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2025/01/16 08:32:11 UTC
(camel) branch main updated: [CAMEL-18276] : azure-service-bus component support for session enabled entities (#16826)
This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 c35b9579d71 [CAMEL-18276] : azure-service-bus component support for session enabled entities (#16826)
c35b9579d71 is described below
commit c35b9579d7105fd8e9cfb8afbe5a80e054565c76
Author: Pulin <22...@users.noreply.github.com>
AuthorDate: Thu Jan 16 14:02:03 2025 +0530
[CAMEL-18276] : azure-service-bus component support for session enabled entities (#16826)
* Camel ASB: sessionId support added for component and producer
* Camel ASB: test case updated for send operation with sessions
* Camel ASB: enabled sessions for consumer for topic
* Camel ASB: formatting done as per checkProfile plugin
* Camel ASB: producer IT test fixed with session processor
* Camel ASB: producer IT test cases updated for queues and scheduled messages
* Camel ASB: consumerIT cases updated for queues
* Camel ASB: consumerIT cases updated for verifying order for session enabled messages
* Camel ASB: countdown latch timeout updated for added IT cases
* Camel ASB: documentation updated for session support
---------
Co-authored-by: Pulin Kumar <pu...@backbase.com>
---
.../servicebus/ServiceBusComponentConfigurer.java | 12 ++
.../servicebus/ServiceBusEndpointConfigurer.java | 12 ++
.../servicebus/ServiceBusEndpointUriFactory.java | 4 +-
.../azure/servicebus/azure-servicebus.json | 60 +++----
.../src/main/docs/azure-servicebus-component.adoc | 23 +++
.../azure/servicebus/ServiceBusConfiguration.java | 27 +++
.../azure/servicebus/ServiceBusConsumer.java | 10 +-
.../azure/servicebus/ServiceBusProducer.java | 14 +-
.../azure/servicebus/ServiceBusUtils.java | 11 +-
.../servicebus/client/ServiceBusClientFactory.java | 38 +++++
.../operations/ServiceBusSenderOperations.java | 40 +++--
.../azure/servicebus/ServiceBusConsumerTest.java | 28 ++++
.../azure/servicebus/ServiceBusEndpointTest.java | 27 ++-
.../azure/servicebus/ServiceBusUtilsTest.java | 69 +++++++-
.../integration/BaseServiceBusTestSupport.java | 37 +++++
.../integration/ServiceBusConsumerIT.java | 72 +++++++-
.../integration/ServiceBusProducerIT.java | 145 +++++++++++++++-
.../operations/ServiceBusSenderOperationsTest.java | 184 +++++++++++++++++++--
18 files changed, 736 insertions(+), 77 deletions(-)
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
index fd9d5458740..64f26f12ac2 100644
--- 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
@@ -76,6 +76,10 @@ public class ServiceBusComponentConfigurer extends PropertyConfigurerSupport imp
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 "sessionenabled":
+ case "sessionEnabled": getOrCreateConfiguration(target).setSessionEnabled(property(camelContext, boolean.class, value)); return true;
+ case "sessionid":
+ case "sessionId": getOrCreateConfiguration(target).setSessionId(property(camelContext, java.lang.String.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":
@@ -140,6 +144,10 @@ public class ServiceBusComponentConfigurer extends PropertyConfigurerSupport imp
case "serviceBusTransactionContext": return com.azure.messaging.servicebus.ServiceBusTransactionContext.class;
case "servicebustype":
case "serviceBusType": return org.apache.camel.component.azure.servicebus.ServiceBusType.class;
+ case "sessionenabled":
+ case "sessionEnabled": return boolean.class;
+ case "sessionid":
+ case "sessionId": return java.lang.String.class;
case "subqueue":
case "subQueue": return com.azure.messaging.servicebus.models.SubQueue.class;
case "subscriptionname":
@@ -200,6 +208,10 @@ public class ServiceBusComponentConfigurer extends PropertyConfigurerSupport imp
case "serviceBusTransactionContext": return getOrCreateConfiguration(target).getServiceBusTransactionContext();
case "servicebustype":
case "serviceBusType": return getOrCreateConfiguration(target).getServiceBusType();
+ case "sessionenabled":
+ case "sessionEnabled": return getOrCreateConfiguration(target).isSessionEnabled();
+ case "sessionid":
+ case "sessionId": return getOrCreateConfiguration(target).getSessionId();
case "subqueue":
case "subQueue": return getOrCreateConfiguration(target).getSubQueue();
case "subscriptionname":
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
index e0b08731033..4c0952187e0 100644
--- 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
@@ -70,6 +70,10 @@ public class ServiceBusEndpointConfigurer extends PropertyConfigurerSupport impl
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 "sessionenabled":
+ case "sessionEnabled": target.getConfiguration().setSessionEnabled(property(camelContext, boolean.class, value)); return true;
+ case "sessionid":
+ case "sessionId": target.getConfiguration().setSessionId(property(camelContext, java.lang.String.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":
@@ -135,6 +139,10 @@ public class ServiceBusEndpointConfigurer extends PropertyConfigurerSupport impl
case "serviceBusTransactionContext": return com.azure.messaging.servicebus.ServiceBusTransactionContext.class;
case "servicebustype":
case "serviceBusType": return org.apache.camel.component.azure.servicebus.ServiceBusType.class;
+ case "sessionenabled":
+ case "sessionEnabled": return boolean.class;
+ case "sessionid":
+ case "sessionId": return java.lang.String.class;
case "subqueue":
case "subQueue": return com.azure.messaging.servicebus.models.SubQueue.class;
case "subscriptionname":
@@ -196,6 +204,10 @@ public class ServiceBusEndpointConfigurer extends PropertyConfigurerSupport impl
case "serviceBusTransactionContext": return target.getConfiguration().getServiceBusTransactionContext();
case "servicebustype":
case "serviceBusType": return target.getConfiguration().getServiceBusType();
+ case "sessionenabled":
+ case "sessionEnabled": return target.getConfiguration().isSessionEnabled();
+ case "sessionid":
+ case "sessionId": return target.getConfiguration().getSessionId();
case "subqueue":
case "subQueue": return target.getConfiguration().getSubQueue();
case "subscriptionname":
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
index 6a60680f4cb..014844410c8 100644
--- 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
@@ -23,7 +23,7 @@ public class ServiceBusEndpointUriFactory extends org.apache.camel.support.compo
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Set<String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(28);
+ Set<String> props = new HashSet<>(30);
props.add("amqpRetryOptions");
props.add("amqpTransportType");
props.add("binary");
@@ -48,6 +48,8 @@ public class ServiceBusEndpointUriFactory extends org.apache.camel.support.compo
props.add("serviceBusReceiveMode");
props.add("serviceBusTransactionContext");
props.add("serviceBusType");
+ props.add("sessionEnabled");
+ props.add("sessionId");
props.add("subQueue");
props.add("subscriptionName");
props.add("tokenCredential");
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/org/apache/camel/component/azure/servicebus/azure-servicebus.json b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/org/apache/camel/component/azure/servicebus/azure-servicebus.json
index 687b7ae90fc..8e58db902e1 100644
--- a/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/org/apache/camel/component/azure/servicebus/azure-servicebus.json
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/resources/META-INF/org/apache/camel/component/azure/servicebus/azure-servicebus.json
@@ -38,19 +38,21 @@
"prefetchCount": { "index": 11, "kind": "property", "displayName": "Prefetch Count", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the prefetch count of the receiver. For both PEEK_LOCK PEEK_LOCK and RECEIVE_AND_DELETE RECEI [...]
"processorClient": { "index": 12, "kind": "property", "displayName": "Processor Client", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusProcessorClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the processorCl [...]
"serviceBusReceiveMode": { "index": 13, "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": "PEEK_LOCK", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "con [...]
- "subQueue": { "index": 14, "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", "descriptio [...]
- "subscriptionName": { "index": 15, "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 an [...]
- "binary": { "index": 16, "kind": "property", "displayName": "Binary", "group": "producer", "label": "producer", "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": "Set binary mode. If true, message body will be sent as byte. By default, it is false." },
- "lazyStartProducer": { "index": 17, "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 fai [...]
- "producerOperation": { "index": 18, "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.ServiceBusConf [...]
- "scheduledEnqueueTime": { "index": 19, "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 S [...]
- "senderClient": { "index": 20, "kind": "property", "displayName": "Sender Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusSenderClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets senderClient to be used [...]
- "serviceBusTransactionContext": { "index": 21, "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 [...]
- "autowiredEnabled": { "index": 22, "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 [...]
- "connectionString": { "index": 23, "kind": "property", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "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 Service Bus [...]
- "credentialType": { "index": 24, "kind": "property", "displayName": "Credential Type", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.CredentialType", "enum": [ "AZURE_IDENTITY", "CONNECTION_STRING", "TOKEN_CREDENTIAL" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CONNECTION_STRING", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfi [...]
- "fullyQualifiedNamespace": { "index": 25, "kind": "property", "displayName": "Fully Qualified Namespace", "group": "security", "label": "security", "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": "Fully Qualified Namespace of the service bus" },
- "tokenCredential": { "index": 26, "kind": "property", "displayName": "Token Credential", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "com.azure.core.credential.TokenCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "A TokenCredential for Azure AD authentication." }
+ "sessionEnabled": { "index": 14, "kind": "property", "displayName": "Session Enabled", "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": "Enable session support" },
+ "subQueue": { "index": 15, "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", "descriptio [...]
+ "subscriptionName": { "index": 16, "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 an [...]
+ "binary": { "index": 17, "kind": "property", "displayName": "Binary", "group": "producer", "label": "producer", "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": "Set binary mode. If true, message body will be sent as byte. By default, it is false." },
+ "lazyStartProducer": { "index": 18, "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 fai [...]
+ "producerOperation": { "index": 19, "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.ServiceBusConf [...]
+ "scheduledEnqueueTime": { "index": 20, "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 S [...]
+ "senderClient": { "index": 21, "kind": "property", "displayName": "Sender Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusSenderClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets senderClient to be used [...]
+ "serviceBusTransactionContext": { "index": 22, "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 [...]
+ "sessionId": { "index": 23, "kind": "property", "displayName": "Session Id", "group": "producer", "label": "producer", "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": "Session ID for session-enabled queues or topics." },
+ "autowiredEnabled": { "index": 24, "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 [...]
+ "connectionString": { "index": 25, "kind": "property", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "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 Service Bus [...]
+ "credentialType": { "index": 26, "kind": "property", "displayName": "Credential Type", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.CredentialType", "enum": [ "AZURE_IDENTITY", "CONNECTION_STRING", "TOKEN_CREDENTIAL" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CONNECTION_STRING", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfi [...]
+ "fullyQualifiedNamespace": { "index": 27, "kind": "property", "displayName": "Fully Qualified Namespace", "group": "security", "label": "security", "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": "Fully Qualified Namespace of the service bus" },
+ "tokenCredential": { "index": 28, "kind": "property", "displayName": "Token Credential", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "com.azure.core.credential.TokenCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "A TokenCredential for Azure AD authentication." }
},
"headers": {
"CamelAzureServiceBusApplicationProperties": { "index": 0, "kind": "header", "displayName": "", "group": "common", "label": "common", "required": false, "javaType": "Map<String, Object>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The application properties (also known as custom properties) on messages sent and received by the producer and consumer, respectively.", "constantName": "org.apache.camel.component.azure.servicebus.Servi [...]
@@ -93,20 +95,22 @@
"prefetchCount": { "index": 10, "kind": "parameter", "displayName": "Prefetch Count", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the prefetch count of the receiver. For both PEEK_LOCK PEEK_LOCK and RECEIVE_AND_DELETE RECE [...]
"processorClient": { "index": 11, "kind": "parameter", "displayName": "Processor Client", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusProcessorClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets the processorC [...]
"serviceBusReceiveMode": { "index": 12, "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": "PEEK_LOCK", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "co [...]
- "subQueue": { "index": 13, "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", "descripti [...]
- "subscriptionName": { "index": 14, "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 a [...]
- "bridgeErrorHandler": { "index": 15, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "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 (if possible) occurred while the Camel consumer is trying to pickup incoming [...]
- "exceptionHandler": { "index": 16, "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 de [...]
- "exchangePattern": { "index": 17, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
- "binary": { "index": 18, "kind": "parameter", "displayName": "Binary", "group": "producer", "label": "producer", "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": "Set binary mode. If true, message body will be sent as byte. By default, it is false." },
- "producerOperation": { "index": 19, "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.ServiceBusCon [...]
- "scheduledEnqueueTime": { "index": 20, "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 [...]
- "senderClient": { "index": 21, "kind": "parameter", "displayName": "Sender Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusSenderClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets senderClient to be used [...]
- "serviceBusTransactionContext": { "index": 22, "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 [...]
- "lazyStartProducer": { "index": 23, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "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 produ [...]
- "connectionString": { "index": 24, "kind": "parameter", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "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 Service Bus [...]
- "credentialType": { "index": 25, "kind": "parameter", "displayName": "Credential Type", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.CredentialType", "enum": [ "AZURE_IDENTITY", "CONNECTION_STRING", "TOKEN_CREDENTIAL" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CONNECTION_STRING", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConf [...]
- "fullyQualifiedNamespace": { "index": 26, "kind": "parameter", "displayName": "Fully Qualified Namespace", "group": "security", "label": "security", "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": "Fully Qualified Namespace of the service bus" },
- "tokenCredential": { "index": 27, "kind": "parameter", "displayName": "Token Credential", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "com.azure.core.credential.TokenCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "A TokenCredential for Azure AD authentication." }
+ "sessionEnabled": { "index": 13, "kind": "parameter", "displayName": "Session Enabled", "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": "Enable session support" },
+ "subQueue": { "index": 14, "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", "descripti [...]
+ "subscriptionName": { "index": 15, "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 a [...]
+ "bridgeErrorHandler": { "index": 16, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "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 (if possible) occurred while the Camel consumer is trying to pickup incoming [...]
+ "exceptionHandler": { "index": 17, "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 de [...]
+ "exchangePattern": { "index": 18, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
+ "binary": { "index": 19, "kind": "parameter", "displayName": "Binary", "group": "producer", "label": "producer", "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": "Set binary mode. If true, message body will be sent as byte. By default, it is false." },
+ "producerOperation": { "index": 20, "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.ServiceBusCon [...]
+ "scheduledEnqueueTime": { "index": 21, "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 [...]
+ "senderClient": { "index": 22, "kind": "parameter", "displayName": "Sender Client", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "com.azure.messaging.servicebus.ServiceBusSenderClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Sets senderClient to be used [...]
+ "serviceBusTransactionContext": { "index": 23, "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 [...]
+ "sessionId": { "index": 24, "kind": "parameter", "displayName": "Session Id", "group": "producer", "label": "producer", "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": "Session ID for session-enabled queues or topics." },
+ "lazyStartProducer": { "index": 25, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "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 produ [...]
+ "connectionString": { "index": 26, "kind": "parameter", "displayName": "Connection String", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "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 Service Bus [...]
+ "credentialType": { "index": 27, "kind": "parameter", "displayName": "Credential Type", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.component.azure.servicebus.CredentialType", "enum": [ "AZURE_IDENTITY", "CONNECTION_STRING", "TOKEN_CREDENTIAL" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CONNECTION_STRING", "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConf [...]
+ "fullyQualifiedNamespace": { "index": 28, "kind": "parameter", "displayName": "Fully Qualified Namespace", "group": "security", "label": "security", "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": "Fully Qualified Namespace of the service bus" },
+ "tokenCredential": { "index": 29, "kind": "parameter", "displayName": "Token Credential", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "com.azure.core.credential.TokenCredential", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "A TokenCredential for Azure AD authentication." }
}
}
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
index 147f77cc8a8..c8d81e65c3b 100644
--- 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
@@ -125,6 +125,22 @@ from("direct:start")
.to("mock:result");
--------------------------------------------------------------------------------
+[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&sessionId=123")
+ .to("mock:result");
+--------------------------------------------------------------------------------
+
- `scheduleMessages`
@@ -154,6 +170,13 @@ from("azure-servicebus:test//?connectionString=test")
.to("mock:result");
--------------------------------------------------------------------------------
+[source,java]
+--------------------------------------------------------------------------------
+from("azure-servicebus:test//?connectionString=test&sessionEnabled=true")
+ .log("${body}")
+ .to("mock:result");
+--------------------------------------------------------------------------------
+
include::spring-boot:partial$starter.adoc[]
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
index 58e4df9ce58..143bf7dd748 100644
--- 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
@@ -90,6 +90,33 @@ public class ServiceBusConfiguration implements Cloneable, HeaderFilterStrategyA
@UriParam(label = "security", enums = "AZURE_IDENTITY,CONNECTION_STRING,TOKEN_CREDENTIAL",
defaultValue = "CONNECTION_STRING")
private CredentialType credentialType = CONNECTION_STRING;
+ // New fields for session support
+ @UriParam(label = "consumer", defaultValue = "false", description = "Enable session support")
+ private boolean sessionEnabled;
+ @UriParam(label = "producer", description = "Session ID for session-enabled queues or topics.")
+ private String sessionId;
+
+ /**
+ * Flag to enable sessions. Default is false. Used to create processor client for message consumer
+ */
+ public boolean isSessionEnabled() {
+ return sessionEnabled;
+ }
+
+ public void setSessionEnabled(boolean sessionEnabled) {
+ this.sessionEnabled = sessionEnabled;
+ }
+
+ /**
+ * SessionId for the message. To set this field, sessionEnabled should be set to true.
+ */
+ public String getSessionId() {
+ return sessionId;
+ }
+
+ public void setSessionId(String sessionId) {
+ this.sessionId = sessionId;
+ }
/**
* Selected topic name or the queue name, that is depending on serviceBusType config. For example if
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
index 42e7a4e7402..94ba24c83ee 100644
--- 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
@@ -52,8 +52,14 @@ public class ServiceBusConsumer extends DefaultConsumer {
super.doStart();
LOG.debug("Creating connection to Azure ServiceBus");
- client = getEndpoint().getServiceBusClientFactory().createServiceBusProcessorClient(getConfiguration(),
- this::processMessage, this::processError);
+ // create client as per sessions
+ if (Boolean.FALSE.equals(getConfiguration().isSessionEnabled())) {
+ client = getEndpoint().getServiceBusClientFactory().createServiceBusProcessorClient(getConfiguration(),
+ this::processMessage, this::processError);
+ } else {
+ client = getEndpoint().getServiceBusClientFactory().createServiceBusSessionProcessorClient(getConfiguration(),
+ this::processMessage, this::processError);
+ }
client.start();
}
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
index 7b17cad8f6b..9dc3b58f1d2 100644
--- 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
@@ -130,11 +130,13 @@ public class ServiceBusProducer extends DefaultProducer {
}
propagateHeaders(exchange, applicationProperties);
final String correlationId = exchange.getMessage().getHeader(ServiceBusConstants.CORRELATION_ID, String.class);
+ final String sessionId = getConfiguration().getSessionId();
if (inputBody instanceof Iterable<?>) {
serviceBusSenderOperations.sendMessages(convertBodyToList((Iterable<?>) inputBody),
configurationOptionsProxy.getServiceBusTransactionContext(exchange), applicationProperties,
- correlationId);
+ correlationId,
+ sessionId);
} else {
Object convertedBody = inputBody instanceof BinaryData ? inputBody
: getConfiguration().isBinary() ? convertBodyToBinary(exchange)
@@ -142,7 +144,8 @@ public class ServiceBusProducer extends DefaultProducer {
serviceBusSenderOperations.sendMessages(convertedBody,
configurationOptionsProxy.getServiceBusTransactionContext(exchange), applicationProperties,
- correlationId);
+ correlationId,
+ sessionId);
}
};
}
@@ -158,13 +161,15 @@ public class ServiceBusProducer extends DefaultProducer {
}
propagateHeaders(exchange, applicationProperties);
final String correlationId = exchange.getMessage().getHeader(ServiceBusConstants.CORRELATION_ID, String.class);
+ final String sessionId = getConfiguration().getSessionId();
if (inputBody instanceof Iterable<?>) {
serviceBusSenderOperations.scheduleMessages(convertBodyToList((Iterable<?>) inputBody),
configurationOptionsProxy.getScheduledEnqueueTime(exchange),
configurationOptionsProxy.getServiceBusTransactionContext(exchange),
applicationProperties,
- correlationId);
+ correlationId,
+ sessionId);
} else {
Object convertedBody = inputBody instanceof BinaryData ? inputBody
: getConfiguration().isBinary() ? convertBodyToBinary(exchange)
@@ -173,7 +178,8 @@ public class ServiceBusProducer extends DefaultProducer {
configurationOptionsProxy.getScheduledEnqueueTime(exchange),
configurationOptionsProxy.getServiceBusTransactionContext(exchange),
applicationProperties,
- correlationId);
+ correlationId,
+ sessionId);
}
};
}
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
index 120b8c5453d..7d553b94b54 100644
--- 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
@@ -30,7 +30,8 @@ public final class ServiceBusUtils {
}
public static ServiceBusMessage createServiceBusMessage(
- final Object data, final Map<String, Object> applicationProperties, final String correlationId) {
+ final Object data, final Map<String, Object> applicationProperties, final String correlationId,
+ final String sessionId) {
ServiceBusMessage serviceBusMessage;
if (data instanceof String) {
serviceBusMessage = new ServiceBusMessage((String) data);
@@ -47,13 +48,17 @@ public final class ServiceBusUtils {
if (ObjectHelper.isNotEmpty(correlationId)) {
serviceBusMessage.setCorrelationId(correlationId);
}
+ if (ObjectHelper.isNotEmpty(sessionId)) {
+ serviceBusMessage.setSessionId(sessionId);
+ }
return serviceBusMessage;
}
public static Iterable<ServiceBusMessage> createServiceBusMessages(
- final Iterable<?> data, final Map<String, Object> applicationProperties, final String correlationId) {
+ final Iterable<?> data, final Map<String, Object> applicationProperties, final String correlationId,
+ final String sessionId) {
return StreamSupport.stream(data.spliterator(), false)
- .map(obj -> createServiceBusMessage(obj, applicationProperties, correlationId))
+ .map(obj -> createServiceBusMessage(obj, applicationProperties, correlationId, sessionId))
.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
index 802b37eaae2..a245ea2af3a 100644
--- 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
@@ -73,6 +73,24 @@ public final class ServiceBusClientFactory {
return processorClientBuilder;
}
+ private static ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder createBaseServiceBusSessionProcessorClient(
+ final ServiceBusClientBuilder busClientBuilder, final ServiceBusConfiguration configuration) {
+ final ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder processorClientBuilder
+ = busClientBuilder.sessionProcessor();
+
+ // We handle auto-complete in the consumer, since we have no way to propagate errors back to the reactive
+ // pipeline messages are published on so the message would be completed even if an error occurs during Exchange
+ // processing.
+ processorClientBuilder.disableAutoComplete();
+
+ switch (configuration.getServiceBusType()) {
+ case queue -> processorClientBuilder.queueName(configuration.getTopicOrQueueName());
+ case topic -> processorClientBuilder.topicName(configuration.getTopicOrQueueName());
+ }
+
+ return processorClientBuilder;
+ }
+
public ServiceBusSenderClient createServiceBusSenderClient(final ServiceBusConfiguration configuration) {
return createBaseServiceBusSenderClient(createBaseServiceBusClient(configuration), configuration)
.buildClient();
@@ -96,4 +114,24 @@ public final class ServiceBusClientFactory {
return clientBuilder.buildProcessorClient();
}
+
+ public ServiceBusProcessorClient createServiceBusSessionProcessorClient(
+ ServiceBusConfiguration configuration, Consumer<ServiceBusReceivedMessageContext> processMessage,
+ Consumer<ServiceBusErrorContext> processError) {
+
+ ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder clientBuilder
+ = createBaseServiceBusSessionProcessorClient(createBaseServiceBusClient(configuration), configuration);
+
+ clientBuilder
+ .subscriptionName(configuration.getSubscriptionName())
+ .receiveMode(configuration.getServiceBusReceiveMode())
+ .maxAutoLockRenewDuration(configuration.getMaxAutoLockRenewDuration())
+ .prefetchCount(configuration.getPrefetchCount())
+ .subQueue(configuration.getSubQueue())
+ .maxConcurrentCalls(configuration.getMaxConcurrentCalls())
+ .processMessage(processMessage)
+ .processError(processError);
+
+ return clientBuilder.buildProcessorClient();
+ }
}
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
index 7bf7b025de5..68c1ad38b96 100644
--- 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
@@ -42,11 +42,12 @@ public class ServiceBusSenderOperations {
final Object data,
final ServiceBusTransactionContext context,
final Map<String, Object> applicationProperties,
- final String correlationId) {
+ final String correlationId,
+ final String sessionId) {
if (data instanceof Iterable<?>) {
- sendMessages((Iterable<?>) data, context, applicationProperties, correlationId);
+ sendMessages((Iterable<?>) data, context, applicationProperties, correlationId, sessionId);
} else {
- sendMessage(data, context, applicationProperties, correlationId);
+ sendMessage(data, context, applicationProperties, correlationId, sessionId);
}
}
@@ -55,25 +56,30 @@ public class ServiceBusSenderOperations {
final OffsetDateTime scheduledEnqueueTime,
final ServiceBusTransactionContext context,
final Map<String, Object> applicationProperties,
- final String correlationId) {
+ final String correlationId,
+ final String sessionId) {
if (ObjectHelper.isEmpty(scheduledEnqueueTime)) {
throw new IllegalArgumentException("To schedule a message, you need to set scheduledEnqueueTime.");
}
if (data instanceof Iterable<?>) {
- return scheduleMessages((Iterable<?>) data, scheduledEnqueueTime, context, applicationProperties, correlationId);
+ return scheduleMessages((Iterable<?>) data, scheduledEnqueueTime, context, applicationProperties, correlationId,
+ sessionId);
}
- return scheduleMessage(data, scheduledEnqueueTime, context, applicationProperties, correlationId);
+ return scheduleMessage(data, scheduledEnqueueTime, context, applicationProperties, correlationId,
+ sessionId);
}
private void sendMessages(
final Iterable<?> data,
final ServiceBusTransactionContext context,
final Map<String, Object> applicationProperties,
- final String correlationId) {
+ final String correlationId,
+ final String sessionId) {
final Iterable<ServiceBusMessage> messages
- = ServiceBusUtils.createServiceBusMessages(data, applicationProperties, correlationId);
+ = ServiceBusUtils.createServiceBusMessages(data, applicationProperties, correlationId,
+ sessionId);
if (ObjectHelper.isEmpty(context)) {
client.sendMessages(messages);
@@ -86,8 +92,10 @@ public class ServiceBusSenderOperations {
final Object data,
final ServiceBusTransactionContext context,
final Map<String, Object> applicationProperties,
- final String correlationId) {
- final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data, applicationProperties, correlationId);
+ final String correlationId,
+ final String sessionId) {
+ final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data, applicationProperties, correlationId,
+ sessionId);
if (ObjectHelper.isEmpty(context)) {
client.sendMessage(message);
@@ -101,8 +109,10 @@ public class ServiceBusSenderOperations {
final OffsetDateTime scheduledEnqueueTime,
final ServiceBusTransactionContext context,
final Map<String, Object> applicationProperties,
- final String correlationId) {
- final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data, applicationProperties, correlationId);
+ final String correlationId,
+ final String sessionId) {
+ final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data, applicationProperties, correlationId,
+ sessionId);
if (ObjectHelper.isEmpty(context)) {
return Collections.singletonList(client.scheduleMessage(message, scheduledEnqueueTime));
@@ -115,9 +125,11 @@ public class ServiceBusSenderOperations {
final Iterable<?> data, final OffsetDateTime scheduledEnqueueTime,
final ServiceBusTransactionContext context,
final Map<String, Object> applicationProperties,
- final String correlationId) {
+ final String correlationId,
+ final String sessionId) {
final Iterable<ServiceBusMessage> messages
- = ServiceBusUtils.createServiceBusMessages(data, applicationProperties, correlationId);
+ = ServiceBusUtils.createServiceBusMessages(data, applicationProperties, correlationId,
+ sessionId);
if (ObjectHelper.isEmpty(context)) {
return StreamSupport.stream(client.scheduleMessages(messages, scheduledEnqueueTime).spliterator(), false).toList();
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
index bce8d9c3f91..757cc21f107 100644
--- a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/ServiceBusConsumerTest.java
@@ -105,6 +105,9 @@ public class ServiceBusConsumerTest {
when(endpoint.getServiceBusClientFactory()).thenReturn(clientFactory);
when(clientFactory.createServiceBusProcessorClient(any(), processMessageCaptor.capture(), processErrorCaptor.capture()))
.thenReturn(client);
+ when(clientFactory.createServiceBusSessionProcessorClient(any(), processMessageCaptor.capture(),
+ processErrorCaptor.capture()))
+ .thenReturn(client);
when(processor.process(exchangeCaptor.capture(), asyncCallbackCaptor.capture())).thenReturn(true);
when(configuration.getHeaderFilterStrategy()).thenReturn(headerFilterStrategy);
}
@@ -133,6 +136,31 @@ public class ServiceBusConsumerTest {
}
}
+ @Test
+ void consumerSubmitsExchangeToSessionProcessor() throws Exception {
+ when(configuration.isSessionEnabled()).thenReturn(true); // Simulate session ID presence
+ try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint, processor)) {
+ consumer.doStart();
+ verify(client).start();
+ verify(clientFactory).createServiceBusSessionProcessorClient(any(), any(), any());
+
+ when(messageContext.getMessage()).thenReturn(message);
+ configureMockMessage();
+
+ processMessageCaptor.getValue().accept(messageContext);
+
+ verify(processor).process(any(Exchange.class), any(AsyncCallback.class));
+ Exchange exchange = exchangeCaptor.getValue();
+ assertThat(exchange).isNotNull();
+
+ Message inMessage = exchange.getIn();
+ assertThat(inMessage).isNotNull();
+ assertThat(inMessage.getBody()).isInstanceOf(BinaryData.class);
+ assertThat(inMessage.getBody(BinaryData.class).toString()).isEqualTo(MESSAGE_BODY);
+ assertThat(inMessage.getHeaders()).isEqualTo(createExpectedMessageHeaders());
+ }
+ }
+
@Test
void consumerSubmitsDeadLetterExchangeToProcessor() throws Exception {
try (ServiceBusConsumer consumer = new ServiceBusConsumer(endpoint, processor)) {
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
index 58efbd71c03..a2b291f849f 100644
--- 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
@@ -25,9 +25,7 @@ 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.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.*;
class ServiceBusEndpointTest extends CamelTestSupport {
@@ -65,6 +63,29 @@ class ServiceBusEndpointTest extends CamelTestSupport {
assertEquals(true, endpoint.getConfiguration().isBinary());
}
+ @Test
+ void testCreateEndpointWithConfigAndSession() 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");
+ params.put("binary", "true");
+ params.put("sessionId", "session-1");
+
+ 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());
+ assertEquals(true, endpoint.getConfiguration().isBinary());
+ assertEquals("session-1", endpoint.getConfiguration().getSessionId());
+ }
+
@Test
void testCreateEndpointWithFqns() throws Exception {
final String uri = "azure-servicebus://testTopicOrQueue";
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
index 33951cf2584..0a97080d37e 100644
--- 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
@@ -32,18 +32,19 @@ public class ServiceBusUtilsTest {
@Test
void testCreateServiceBusMessage() {
// test string
- final ServiceBusMessage message1 = ServiceBusUtils.createServiceBusMessage("test string", null, null);
+ final ServiceBusMessage message1 = ServiceBusUtils.createServiceBusMessage("test string", null, null, null);
assertEquals("test string", message1.getBody().toString());
// test int
- final ServiceBusMessage message2 = ServiceBusUtils.createServiceBusMessage(String.valueOf(12345), null, null);
+ final ServiceBusMessage message2
+ = ServiceBusUtils.createServiceBusMessage(String.valueOf(12345), null, null, null);
assertEquals("12345", message2.getBody().toString());
//test bytes
byte[] testByteBody = "test string".getBytes(StandardCharsets.UTF_8);
- final ServiceBusMessage message3 = ServiceBusUtils.createServiceBusMessage(testByteBody, null, null);
+ final ServiceBusMessage message3 = ServiceBusUtils.createServiceBusMessage(testByteBody, null, null, null);
assertArrayEquals(testByteBody, message3.getBody().toBytes());
}
@@ -53,7 +54,8 @@ public class ServiceBusUtilsTest {
inputMessages.add("test data");
inputMessages.add(String.valueOf(12345));
- final Iterable<ServiceBusMessage> busMessages = ServiceBusUtils.createServiceBusMessages(inputMessages, null, null);
+ final Iterable<ServiceBusMessage> busMessages
+ = ServiceBusUtils.createServiceBusMessages(inputMessages, null, null, null);
assertTrue(StreamSupport.stream(busMessages.spliterator(), false)
.anyMatch(record -> record.getBody().toString().equals("test data")));
@@ -67,11 +69,68 @@ public class ServiceBusUtilsTest {
inputMessages2.add(byteBody1);
inputMessages2.add(byteBody2);
- final Iterable<ServiceBusMessage> busMessages2 = ServiceBusUtils.createServiceBusMessages(inputMessages2, null, null);
+ final Iterable<ServiceBusMessage> busMessages2
+ = ServiceBusUtils.createServiceBusMessages(inputMessages2, null, null, null);
assertTrue(StreamSupport.stream(busMessages2.spliterator(), false)
.anyMatch(message -> Arrays.equals(message.getBody().toBytes(), byteBody1)));
assertTrue(StreamSupport.stream(busMessages2.spliterator(), false)
.anyMatch(message -> Arrays.equals(message.getBody().toBytes(), byteBody2)));
}
+
+ @Test
+ void testCreateServiceBusMessageWithSession() {
+ // test string
+ final ServiceBusMessage message1
+ = ServiceBusUtils.createServiceBusMessage("test string", null, null, "session-1");
+
+ assertEquals("test string", message1.getBody().toString());
+ assertEquals("session-1", message1.getSessionId());
+
+ // test int
+ final ServiceBusMessage message2
+ = ServiceBusUtils.createServiceBusMessage(String.valueOf(12345), null, null, "session-2");
+
+ assertEquals("12345", message2.getBody().toString());
+ assertEquals("session-2", message2.getSessionId());
+
+ //test bytes
+ byte[] testByteBody = "test string".getBytes(StandardCharsets.UTF_8);
+ final ServiceBusMessage message3 = ServiceBusUtils.createServiceBusMessage(testByteBody, null, null, "session-1");
+ assertArrayEquals(testByteBody, message3.getBody().toBytes());
+ }
+
+ @Test
+ void testCreateServiceBusMessagesWithSession() {
+ final List<String> inputMessages = new LinkedList<>();
+ inputMessages.add("test data");
+ inputMessages.add(String.valueOf(12345));
+
+ final Iterable<ServiceBusMessage> busMessages
+ = ServiceBusUtils.createServiceBusMessages(inputMessages, null, null, "session-1");
+
+ 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")));
+ assertTrue(StreamSupport.stream(busMessages.spliterator(), false)
+ .anyMatch(record -> record.getSessionId().equals("session-1")));
+
+ //Test bytes
+ final List<byte[]> inputMessages2 = new LinkedList<>();
+ byte[] byteBody1 = "test data".getBytes(StandardCharsets.UTF_8);
+ byte[] byteBody2 = "test data2".getBytes(StandardCharsets.UTF_8);
+ inputMessages2.add(byteBody1);
+ inputMessages2.add(byteBody2);
+
+ final Iterable<ServiceBusMessage> busMessages2
+ = ServiceBusUtils.createServiceBusMessages(inputMessages2, null, null, "session-2");
+
+ assertTrue(StreamSupport.stream(busMessages2.spliterator(), false)
+ .anyMatch(message -> Arrays.equals(message.getBody().toBytes(), byteBody1)));
+ assertTrue(StreamSupport.stream(busMessages2.spliterator(), false)
+ .anyMatch(message -> Arrays.equals(message.getBody().toBytes(), byteBody2)));
+ assertTrue(StreamSupport.stream(busMessages2.spliterator(), false)
+ .anyMatch(record -> record.getSessionId().equals("session-2")));
+ }
}
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/BaseServiceBusTestSupport.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/BaseServiceBusTestSupport.java
index 5eac391015e..68a4673e7b1 100644
--- a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/BaseServiceBusTestSupport.java
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/BaseServiceBusTestSupport.java
@@ -25,6 +25,9 @@ import com.azure.messaging.servicebus.ServiceBusProcessorClient;
import com.azure.messaging.servicebus.ServiceBusReceivedMessageContext;
import com.azure.messaging.servicebus.administration.ServiceBusAdministrationClient;
import com.azure.messaging.servicebus.administration.ServiceBusAdministrationClientBuilder;
+import com.azure.messaging.servicebus.administration.models.CreateQueueOptions;
+import com.azure.messaging.servicebus.administration.models.CreateSubscriptionOptions;
+import com.azure.messaging.servicebus.administration.models.CreateTopicOptions;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.azure.servicebus.ServiceBusComponent;
@@ -45,8 +48,11 @@ public abstract class BaseServiceBusTestSupport implements ConfigurableRoute {
public static final String CONNECTION_STRING_PROPERTY_NAME = "camel.component.azure-servicebus.connection-string";
protected static final String CONNECTION_STRING = getProperty(CONNECTION_STRING_PROPERTY_NAME);
protected static final String QUEUE_NAME = "camelTestQueue";
+ protected static final String QUEUE_WITH_SESSIONS_NAME = "camelTestQueueSessions";
protected static final String TOPIC_NAME = "camelTestTopic";
+ protected static final String TOPIC_WITH_SESSIONS_NAME = "camelTestTopicSessions";
protected static final String SUBSCRIPTION_NAME = "camelTestSubscription";
+ protected static final String SUBSCRIPTION_WITH_SESSIONS_NAME = "camelTestSubscriptionSessions";
protected static final String MOCK_RESULT = "mock:result";
private static final Logger LOGGER = LoggerFactory.getLogger(BaseServiceBusTestSupport.class);
@RegisterExtension
@@ -62,16 +68,20 @@ public abstract class BaseServiceBusTestSupport implements ConfigurableRoute {
.buildClient();
try {
serviceBusAdminClient.createQueue(QUEUE_NAME);
+ serviceBusAdminClient.createQueue(QUEUE_WITH_SESSIONS_NAME, new CreateQueueOptions().setSessionRequired(true));
} catch (ResourceExistsException e) {
LOGGER.warn("Test queue already existed", e);
}
try {
serviceBusAdminClient.createTopic(TOPIC_NAME);
+ serviceBusAdminClient.createTopic(TOPIC_WITH_SESSIONS_NAME, new CreateTopicOptions().setSessionRequired(true));
} catch (ResourceExistsException e) {
LOGGER.warn("Test topic already existed", e);
}
try {
serviceBusAdminClient.createSubscription(TOPIC_NAME, SUBSCRIPTION_NAME);
+ serviceBusAdminClient.createSubscription(TOPIC_WITH_SESSIONS_NAME, SUBSCRIPTION_WITH_SESSIONS_NAME,
+ new CreateSubscriptionOptions().setSessionRequired(true));
} catch (ResourceExistsException e) {
LOGGER.warn("Test subscription already existed", e);
}
@@ -80,8 +90,11 @@ public abstract class BaseServiceBusTestSupport implements ConfigurableRoute {
@AfterAll
static void baseAfterAll() {
serviceBusAdminClient.deleteSubscription(TOPIC_NAME, SUBSCRIPTION_NAME);
+ serviceBusAdminClient.deleteSubscription(TOPIC_WITH_SESSIONS_NAME, SUBSCRIPTION_WITH_SESSIONS_NAME);
serviceBusAdminClient.deleteTopic(TOPIC_NAME);
+ serviceBusAdminClient.deleteTopic(TOPIC_WITH_SESSIONS_NAME);
serviceBusAdminClient.deleteQueue(QUEUE_NAME);
+ serviceBusAdminClient.deleteQueue(QUEUE_WITH_SESSIONS_NAME);
}
@ContextFixture
@@ -110,6 +123,17 @@ public abstract class BaseServiceBusTestSupport implements ConfigurableRoute {
.buildProcessorClient();
}
+ protected ServiceBusProcessorClient createSessionQueueProcessorClient() {
+ return new ServiceBusClientBuilder()
+ .connectionString(CONNECTION_STRING)
+ .sessionProcessor()
+ .queueName(QUEUE_WITH_SESSIONS_NAME)
+ .processMessage(this::processMessage)
+ .processError(serviceBusErrorContext -> LOGGER.error("Service Bus client error",
+ serviceBusErrorContext.getException()))
+ .buildProcessorClient();
+ }
+
protected ServiceBusProcessorClient createTopicProcessorClient() {
return new ServiceBusClientBuilder()
.connectionString(CONNECTION_STRING)
@@ -122,6 +146,19 @@ public abstract class BaseServiceBusTestSupport implements ConfigurableRoute {
.buildProcessorClient();
}
+ protected ServiceBusProcessorClient createTopicSessionProcessorClient() {
+ return new ServiceBusClientBuilder()
+ .connectionString(CONNECTION_STRING)
+ .sessionProcessor()
+ .disableAutoComplete()
+ .topicName(TOPIC_WITH_SESSIONS_NAME)
+ .subscriptionName(SUBSCRIPTION_WITH_SESSIONS_NAME)
+ .processMessage(this::processMessage)
+ .processError(serviceBusErrorContext -> LOGGER.error("Service Bus client error",
+ serviceBusErrorContext.getException()))
+ .buildProcessorClient();
+ }
+
private void processMessage(ServiceBusReceivedMessageContext messageContext) {
receivedMessageContexts.add(messageContext);
messageLatch.countDown();
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerIT.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerIT.java
index 0a2f302710f..fa6a823457c 100644
--- a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerIT.java
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerIT.java
@@ -32,8 +32,7 @@ import org.apache.camel.test.infra.core.annotations.RouteFixture;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
-import static org.junit.jupiter.api.Assertions.assertInstanceOf;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.*;
@EnabledIfSystemProperty(named = BaseServiceBusTestSupport.CONNECTION_STRING_PROPERTY_NAME, matches = ".*",
disabledReason = "Service Bus connection string must be supplied to run this test, e.g: mvn verify -D"
@@ -46,6 +45,12 @@ public class ServiceBusConsumerIT extends BaseServiceBusTestSupport {
.queueName(QUEUE_NAME)
.buildClient();
+ private final ServiceBusSenderClient queueWithSessionsSenderClient = new ServiceBusClientBuilder()
+ .connectionString(CONNECTION_STRING)
+ .sender()
+ .queueName(QUEUE_WITH_SESSIONS_NAME)
+ .buildClient();
+
private final ServiceBusSenderClient topicSenderClient = new ServiceBusClientBuilder()
.connectionString(CONNECTION_STRING)
.sender()
@@ -59,8 +64,15 @@ public class ServiceBusConsumerIT extends BaseServiceBusTestSupport {
from("azure-servicebus:" + QUEUE_NAME)
.to(MOCK_RESULT);
+ from("azure-servicebus:" + QUEUE_WITH_SESSIONS_NAME + "?sessionEnabled=true")
+ .to(MOCK_RESULT);
+
from("azure-servicebus:" + TOPIC_NAME + "?serviceBusType=topic&subscriptionName=" + SUBSCRIPTION_NAME)
.to(MOCK_RESULT);
+
+ from("azure-servicebus:" + TOPIC_WITH_SESSIONS_NAME + "?serviceBusType=topic&subscriptionName="
+ + SUBSCRIPTION_WITH_SESSIONS_NAME + "?sessionEnabled=true")
+ .to(MOCK_RESULT);
}
};
}
@@ -120,6 +132,62 @@ public class ServiceBusConsumerIT extends BaseServiceBusTestSupport {
assertBrokerPropertyHeadersPropagated(headers);
}
+ /*
+ Tests for entities with session support
+ */
+
+ @Test
+ public void serviceBusSessionEnabledQueueIsConsumedByCamel() throws InterruptedException {
+ String propagatedHeaderKey = "PropagatedCustomHeader";
+ String propagatedHeaderValue = "propagated header value";
+
+ MockEndpoint to = contextExtension.getMockEndpoint(MOCK_RESULT);
+
+ to.expectedMessageCount(5);
+ to.expectedBodiesReceived("message-0", "message-1", "message-2", "message-3", "message-4");
+ to.expectedHeaderReceived(propagatedHeaderKey, propagatedHeaderValue);
+
+ for (int k = 0; k < 5; k++) {
+ String msg = "message-" + k;
+ ServiceBusMessage serviceBusMessage = new ServiceBusMessage(msg);
+ serviceBusMessage.setSessionId("session-1");
+ serviceBusMessage.getApplicationProperties().put(propagatedHeaderKey, propagatedHeaderValue);
+ queueWithSessionsSenderClient.sendMessage(serviceBusMessage);
+ }
+
+ to.assertIsSatisfied(3000);
+
+ Map<String, Object> headers = to.getExchanges().get(0).getIn().getHeaders();
+ assertTrue(headers.containsKey(propagatedHeaderKey), "Should receive propagated header");
+ assertBrokerPropertyHeadersPropagated(headers);
+ }
+
+ @Test
+ public void serviceBusSessionEnabledTopicIsConsumedByCamel() throws InterruptedException {
+ String propagatedHeaderKey = "PropagatedCustomHeader";
+ String propagatedHeaderValue = "propagated header value";
+
+ MockEndpoint to = contextExtension.getMockEndpoint(MOCK_RESULT);
+
+ to.expectedMessageCount(5);
+ to.expectedBodiesReceived("message-0", "message-1", "message-2", "message-3", "message-4");
+ to.expectedHeaderReceived(propagatedHeaderKey, propagatedHeaderValue);
+
+ for (int k = 0; k < 5; k++) {
+ String msg = "message-" + k;
+ ServiceBusMessage serviceBusMessage = new ServiceBusMessage(msg);
+ serviceBusMessage.setSessionId("session-1");
+ serviceBusMessage.getApplicationProperties().put(propagatedHeaderKey, propagatedHeaderValue);
+ topicSenderClient.sendMessage(serviceBusMessage);
+ }
+
+ to.assertIsSatisfied(3000);
+
+ Map<String, Object> headers = to.getExchanges().get(0).getIn().getHeaders();
+ assertTrue(headers.containsKey(propagatedHeaderKey), "Should receive propagated header");
+ assertBrokerPropertyHeadersPropagated(headers);
+ }
+
private void assertBrokerPropertyHeadersPropagated(Map<String, Object> headers) {
assertInstanceOf(Map.class, headers.get(ServiceBusConstants.APPLICATION_PROPERTIES),
"Should receive application properties");
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusProducerIT.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusProducerIT.java
index 66f01442cfa..97619fda830 100644
--- a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusProducerIT.java
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusProducerIT.java
@@ -42,6 +42,9 @@ public class ServiceBusProducerIT extends BaseServiceBusTestSupport {
private static final String DIRECT_SEND_TO_QUEUE_URI = "direct:sendToQueue";
private static final String DIRECT_SEND_TO_TOPIC_URI = "direct:sendToTopic";
private static final String DIRECT_SEND_SCHEDULED_URI = "direct:sendScheduled";
+ private static final String DIRECT_SEND_SCHEDULED_URI_WITH_SESSIONS = "direct:sendScheduledSessions";
+ private static final String DIRECT_SEND_TO_TOPIC_SESSION_URI = "direct:sendToTopicSessions";
+ private static final String DIRECT_SEND_TO_SESSION_QUEUE_URI = "direct:sendToQueueSessions";
private static final Map<String, Object> PROPAGATED_HEADERS = new HashMap<>();
private static final Pattern MESSAGE_BODY_PATTERN = Pattern.compile("^message-[0-4]$");
@@ -75,11 +78,21 @@ public class ServiceBusProducerIT extends BaseServiceBusTestSupport {
from(DIRECT_SEND_TO_QUEUE_URI)
.to("azure-servicebus:" + QUEUE_NAME);
+ from(DIRECT_SEND_TO_SESSION_QUEUE_URI)
+ .to("azure-servicebus:" + QUEUE_WITH_SESSIONS_NAME + "?sessionId=123");
+
from(DIRECT_SEND_TO_TOPIC_URI)
.to("azure-servicebus:" + TOPIC_NAME + "?serviceBusType=topic");
+ from(DIRECT_SEND_TO_TOPIC_SESSION_URI)
+ .to("azure-servicebus:" + TOPIC_WITH_SESSIONS_NAME + "?serviceBusType=topic&sessionId=123");
+
from(DIRECT_SEND_SCHEDULED_URI)
.to("azure-servicebus:" + QUEUE_NAME + "?producerOperation=scheduleMessages");
+
+ from(DIRECT_SEND_SCHEDULED_URI_WITH_SESSIONS)
+ .to("azure-servicebus:" + QUEUE_WITH_SESSIONS_NAME
+ + "?producerOperation=scheduleMessages&sessionId=123");
}
};
}
@@ -127,7 +140,7 @@ public class ServiceBusProducerIT extends BaseServiceBusTestSupport {
exchange.getIn().setBody(messageBatch);
});
- assertTrue(messageLatch.await(3000, TimeUnit.MILLISECONDS));
+ assertTrue(messageLatch.await(8000, TimeUnit.MILLISECONDS));
assertEquals(5, receivedMessageContexts.size());
receivedMessageContexts.forEach(messageContext -> {
ServiceBusReceivedMessage message = messageContext.getMessage();
@@ -216,4 +229,134 @@ public class ServiceBusProducerIT extends BaseServiceBusTestSupport {
});
}
}
+
+ /*
+ Tests for entities with session support
+ */
+
+ @Test
+ void camelSendsMessageToServiceBusSessionEnabledQueue() throws InterruptedException {
+ messageLatch = new CountDownLatch(5);
+ try (ServiceBusProcessorClient client = createSessionQueueProcessorClient()) {
+ client.start();
+ for (int i = 0; i < 5; i++) {
+ String message = "message-" + i;
+ producerTemplate.sendBodyAndHeaders(DIRECT_SEND_TO_SESSION_QUEUE_URI, message, PROPAGATED_HEADERS);
+ }
+
+ assertTrue(messageLatch.await(8000, TimeUnit.MILLISECONDS));
+ assertEquals(5, receivedMessageContexts.size());
+ receivedMessageContexts.forEach(messageContext -> {
+ ServiceBusReceivedMessage message = messageContext.getMessage();
+ String messageBody = message.getBody().toString();
+ assertTrue(MESSAGE_BODY_PATTERN.matcher(messageBody).matches());
+ Map<String, Object> applicationProperties = message.getApplicationProperties();
+ assertEquals(PROPAGATED_HEADERS, applicationProperties);
+ });
+ }
+ }
+
+ @Test
+ void camelSendsMessageBatchToServiceBusSessionEnabledQueue() throws InterruptedException {
+ messageLatch = new CountDownLatch(5);
+ try (ServiceBusProcessorClient client = createSessionQueueProcessorClient()) {
+ client.start();
+ List<Object> messageBatch = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ String message = "message-" + i;
+ messageBatch.add(message);
+ }
+
+ producerTemplate.send(DIRECT_SEND_TO_SESSION_QUEUE_URI, exchange -> {
+ exchange.getIn().setHeaders(PROPAGATED_HEADERS);
+ exchange.getIn().setBody(messageBatch);
+ });
+
+ assertTrue(messageLatch.await(10000, TimeUnit.MILLISECONDS));
+ assertEquals(5, receivedMessageContexts.size());
+ receivedMessageContexts.forEach(messageContext -> {
+ ServiceBusReceivedMessage message = messageContext.getMessage();
+ String messageBody = message.getBody().toString();
+ assertTrue(MESSAGE_BODY_PATTERN.matcher(messageBody).matches());
+ Map<String, Object> applicationProperties = message.getApplicationProperties();
+ assertEquals(PROPAGATED_HEADERS, applicationProperties);
+ });
+ }
+ }
+
+ @Test
+ void camelSendsMessageWithSessionToServiceBusTopic() throws InterruptedException {
+ messageLatch = new CountDownLatch(5);
+ try (ServiceBusProcessorClient client = createTopicSessionProcessorClient()) {
+ client.start();
+ for (int i = 0; i < 5; i++) {
+ String message = "message-" + i;
+ producerTemplate.sendBodyAndHeaders(DIRECT_SEND_TO_TOPIC_SESSION_URI, message, PROPAGATED_HEADERS);
+ }
+
+ assertTrue(messageLatch.await(8000, TimeUnit.MILLISECONDS));
+ assertEquals(5, receivedMessageContexts.size());
+ receivedMessageContexts.forEach(messageContext -> {
+ ServiceBusReceivedMessage message = messageContext.getMessage();
+ String messageBody = message.getBody().toString();
+ assertTrue(MESSAGE_BODY_PATTERN.matcher(messageBody).matches());
+ });
+ }
+ }
+
+ @Test
+ void camelSchedulesServiceBusMessageWithSessions() throws InterruptedException {
+ messageLatch = new CountDownLatch(5);
+ try (ServiceBusProcessorClient client = createSessionQueueProcessorClient()) {
+ client.start();
+ for (int i = 0; i < 5; i++) {
+ String message = "message-" + i;
+ Map<String, Object> headers = new HashMap<>(PROPAGATED_HEADERS);
+ headers.put(ServiceBusConstants.SCHEDULED_ENQUEUE_TIME, OffsetDateTime.now().plusSeconds(1));
+ producerTemplate.sendBodyAndHeaders(DIRECT_SEND_SCHEDULED_URI_WITH_SESSIONS, message, headers);
+ }
+
+ assertTrue(messageLatch.await(8000, TimeUnit.MILLISECONDS));
+ assertEquals(5, receivedMessageContexts.size());
+ receivedMessageContexts.forEach(messageContext -> {
+ ServiceBusReceivedMessage message = messageContext.getMessage();
+ String messageBody = message.getBody().toString();
+ assertTrue(MESSAGE_BODY_PATTERN.matcher(messageBody).matches());
+ Map<String, Object> applicationProperties = message.getApplicationProperties();
+ assertEquals(PROPAGATED_HEADERS, applicationProperties);
+ assertInstanceOf(OffsetDateTime.class, message.getScheduledEnqueueTime());
+ });
+ }
+ }
+
+ @Test
+ void camelSchedulesServiceBusMessageBatchWIthSessions() throws InterruptedException {
+ messageLatch = new CountDownLatch(5);
+ try (ServiceBusProcessorClient client = createSessionQueueProcessorClient()) {
+ client.start();
+ List<Object> messageBatch = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ String message = "message-" + i;
+ messageBatch.add(message);
+ }
+
+ producerTemplate.send(DIRECT_SEND_SCHEDULED_URI_WITH_SESSIONS, exchange -> {
+ Map<String, Object> headers = new HashMap<>(PROPAGATED_HEADERS);
+ headers.put(ServiceBusConstants.SCHEDULED_ENQUEUE_TIME, OffsetDateTime.now().plusSeconds(1));
+ exchange.getIn().setHeaders(headers);
+ exchange.getIn().setBody(messageBatch);
+ });
+
+ assertTrue(messageLatch.await(8000, TimeUnit.MILLISECONDS));
+ assertEquals(5, receivedMessageContexts.size());
+ receivedMessageContexts.forEach(messageContext -> {
+ ServiceBusReceivedMessage message = messageContext.getMessage();
+ String messageBody = message.getBody().toString();
+ assertTrue(MESSAGE_BODY_PATTERN.matcher(messageBody).matches());
+ Map<String, Object> applicationProperties = message.getApplicationProperties();
+ assertEquals(PROPAGATED_HEADERS, applicationProperties);
+ assertInstanceOf(OffsetDateTime.class, message.getScheduledEnqueueTime());
+ });
+ }
+ }
}
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
index c29a9ee3423..b688cb6a0fb 100644
--- 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
@@ -43,11 +43,20 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class ServiceBusSenderOperationsTest extends BaseServiceBusTestSupport {
private static ServiceBusSenderClient client;
+ private static ServiceBusSenderClient sessionClient;
+ private static ServiceBusSenderClient batchClient;
+ private static ServiceBusSenderClient batchSessionClient;
@BeforeAll
static void prepare() {
client = new ServiceBusClientBuilder().connectionString(CONNECTION_STRING)
.sender().topicName(TOPIC_NAME).buildClient();
+ sessionClient = new ServiceBusClientBuilder().connectionString(CONNECTION_STRING)
+ .sender().topicName(TOPIC_WITH_SESSIONS_NAME).buildClient();
+ batchClient = new ServiceBusClientBuilder().connectionString(CONNECTION_STRING)
+ .sender().queueName(QUEUE_NAME).buildClient();
+ batchSessionClient = new ServiceBusClientBuilder().connectionString(CONNECTION_STRING)
+ .sender().queueName(QUEUE_WITH_SESSIONS_NAME).buildClient();
}
@AfterAll
@@ -67,10 +76,10 @@ public class ServiceBusSenderOperationsTest extends BaseServiceBusTestSupport {
try (ServiceBusProcessorClient processorClient = createTopicProcessorClient()) {
processorClient.start();
- operations.sendMessages("test data", null, Map.of("customKey", "customValue"), null);
+ operations.sendMessages("test data", null, Map.of("customKey", "customValue"), null, null);
//test bytes
byte[] testByteBody = "test data".getBytes(StandardCharsets.UTF_8);
- operations.sendMessages(testByteBody, null, Map.of("customKey", "customValue"), null);
+ operations.sendMessages(testByteBody, null, Map.of("customKey", "customValue"), null, null);
assertTrue(messageLatch.await(3000, TimeUnit.MILLISECONDS));
@@ -85,16 +94,45 @@ public class ServiceBusSenderOperationsTest extends BaseServiceBusTestSupport {
// test if we have something other than string or byte[]
assertThrows(IllegalArgumentException.class, () -> {
- operations.sendMessages(12345, null, null, null);
+ operations.sendMessages(12345, null, null, null, null);
+ });
+ }
+
+ @Test
+ void testSendSingleMessageWithSessions() throws InterruptedException {
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(sessionClient);
+ messageLatch = new CountDownLatch(2);
+
+ try (ServiceBusProcessorClient processorClient = createTopicSessionProcessorClient()) {
+ processorClient.start();
+ operations.sendMessages("test data", null, Map.of("customKey", "customValue"), null, "session-1");
+ //test bytes
+ byte[] testByteBody = "test data".getBytes(StandardCharsets.UTF_8);
+ operations.sendMessages(testByteBody, null, Map.of("customKey", "customValue"), null, "session-1");
+
+ assertTrue(messageLatch.await(5000, TimeUnit.MILLISECONDS));
+
+ final boolean exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> messageContext.getMessage().getBody().toString().equals("test data"));
+ assertTrue(exists, "test message body");
+
+ final boolean exists2 = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> Arrays.equals(messageContext.getMessage().getBody().toBytes(), testByteBody));
+ assertTrue(exists2, "test byte body");
+ }
+
+ // test if we have something other than string or byte[]
+ assertThrows(IllegalArgumentException.class, () -> {
+ operations.sendMessages(12345, null, null, null, "session-1");
});
}
@Test
void testSendingBatchMessages() throws InterruptedException {
- final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(client);
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(batchClient);
messageLatch = new CountDownLatch(5);
- try (ServiceBusProcessorClient processorClient = createTopicProcessorClient()) {
+ try (ServiceBusProcessorClient processorClient = createQueueProcessorClient()) {
processorClient.start();
final List<String> inputBatch = new LinkedList<>();
@@ -102,16 +140,61 @@ public class ServiceBusSenderOperationsTest extends BaseServiceBusTestSupport {
inputBatch.add("test batch 2");
inputBatch.add("test batch 3");
- operations.sendMessages(inputBatch, null, null, null);
+ operations.sendMessages(inputBatch, null, null, null, null);
//test bytes
final List<byte[]> inputBatch2 = new LinkedList<>();
byte[] byteBody1 = "test data".getBytes(StandardCharsets.UTF_8);
byte[] byteBody2 = "test data2".getBytes(StandardCharsets.UTF_8);
inputBatch2.add(byteBody1);
inputBatch2.add(byteBody2);
- operations.sendMessages(inputBatch2, null, null, null);
+ operations.sendMessages(inputBatch2, null, null, null, null);
- assertTrue(messageLatch.await(3000, TimeUnit.MILLISECONDS));
+ assertTrue(messageLatch.await(8000, TimeUnit.MILLISECONDS));
+
+ final boolean batch1Exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> messageContext.getMessage().getBody().toString().equals("test batch 1"));
+ final boolean batch2Exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> messageContext.getMessage().getBody().toString().equals("test batch 2"));
+ final boolean batch3Exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> messageContext.getMessage().getBody().toString().equals("test batch 3"));
+
+ assertTrue(batch1Exists, "test message body 1");
+ assertTrue(batch2Exists, "test message body 2");
+ assertTrue(batch3Exists, "test message body 3");
+
+ final boolean byteBody1Exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> Arrays.equals(messageContext.getMessage().getBody().toBytes(), byteBody1));
+ final boolean byteBody2Exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> Arrays.equals(messageContext.getMessage().getBody().toBytes(), byteBody2));
+
+ assertTrue(byteBody1Exists, "test byte body 1");
+ assertTrue(byteBody2Exists, "test byte body 2");
+ }
+ }
+
+ @Test
+ void testSendingBatchMessagesWithSessions() throws InterruptedException {
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(batchSessionClient);
+ messageLatch = new CountDownLatch(5);
+
+ try (ServiceBusProcessorClient processorClient = createSessionQueueProcessorClient()) {
+ processorClient.start();
+
+ 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, null, null, "session-1");
+ //test bytes
+ final List<byte[]> inputBatch2 = new LinkedList<>();
+ byte[] byteBody1 = "test data".getBytes(StandardCharsets.UTF_8);
+ byte[] byteBody2 = "test data2".getBytes(StandardCharsets.UTF_8);
+ inputBatch2.add(byteBody1);
+ inputBatch2.add(byteBody2);
+ operations.sendMessages(inputBatch2, null, null, null, "session-1");
+
+ assertTrue(messageLatch.await(8000, TimeUnit.MILLISECONDS));
final boolean batch1Exists = receivedMessageContexts.stream()
.anyMatch(messageContext -> messageContext.getMessage().getBody().toString().equals("test batch 1"));
@@ -142,10 +225,39 @@ public class ServiceBusSenderOperationsTest extends BaseServiceBusTestSupport {
try (ServiceBusProcessorClient processorClient = createTopicProcessorClient()) {
processorClient.start();
- operations.scheduleMessages("testScheduleMessage", OffsetDateTime.now(), null, null, null);
+ operations.scheduleMessages("testScheduleMessage", OffsetDateTime.now(), null, null, null, null);
+ //test bytes
+ byte[] testByteBody = "test data".getBytes(StandardCharsets.UTF_8);
+ operations.scheduleMessages(testByteBody, OffsetDateTime.now(), null, null, null, null);
+
+ assertTrue(messageLatch.await(3000, TimeUnit.MILLISECONDS));
+
+ final boolean exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> messageContext.getMessage().getBody().toString().equals("testScheduleMessage"));
+ final boolean exists2 = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> Arrays.equals(messageContext.getMessage().getBody().toBytes(), testByteBody));
+
+ assertTrue(exists, "test message body");
+ assertTrue(exists2, "test byte body");
+ }
+ // test if we have something other than string or byte[]
+ assertThrows(IllegalArgumentException.class, () -> {
+ operations.scheduleMessages(12345, OffsetDateTime.now(), null, null, null, null);
+ });
+ }
+
+ @Test
+ void testScheduleMessageWithSessions() throws InterruptedException {
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(sessionClient);
+ messageLatch = new CountDownLatch(2);
+
+ try (ServiceBusProcessorClient processorClient = createTopicSessionProcessorClient()) {
+ processorClient.start();
+
+ operations.scheduleMessages("testScheduleMessage", OffsetDateTime.now(), null, null, null, "session-1");
//test bytes
byte[] testByteBody = "test data".getBytes(StandardCharsets.UTF_8);
- operations.scheduleMessages(testByteBody, OffsetDateTime.now(), null, null, null);
+ operations.scheduleMessages(testByteBody, OffsetDateTime.now(), null, null, null, "session-1");
assertTrue(messageLatch.await(3000, TimeUnit.MILLISECONDS));
@@ -159,7 +271,7 @@ public class ServiceBusSenderOperationsTest extends BaseServiceBusTestSupport {
}
// test if we have something other than string or byte[]
assertThrows(IllegalArgumentException.class, () -> {
- operations.scheduleMessages(12345, OffsetDateTime.now(), null, null, null);
+ operations.scheduleMessages(12345, OffsetDateTime.now(), null, null, null, "session-1");
});
}
@@ -175,16 +287,60 @@ public class ServiceBusSenderOperationsTest extends BaseServiceBusTestSupport {
inputBatch.add("testSchedulingBatchMessages 1");
inputBatch.add("testSchedulingBatchMessages 2");
inputBatch.add("testSchedulingBatchMessages 3");
- operations.scheduleMessages(inputBatch, OffsetDateTime.now(), null, null, null);
+ operations.scheduleMessages(inputBatch, OffsetDateTime.now(), null, null, null, null);
//test bytes
final List<byte[]> inputBatch2 = new LinkedList<>();
byte[] byteBody1 = "test data".getBytes(StandardCharsets.UTF_8);
byte[] byteBody2 = "test data2".getBytes(StandardCharsets.UTF_8);
inputBatch2.add(byteBody1);
inputBatch2.add(byteBody2);
- operations.scheduleMessages(inputBatch2, OffsetDateTime.now(), null, null, null);
+ operations.scheduleMessages(inputBatch2, OffsetDateTime.now(), null, null, null, null);
- assertTrue(messageLatch.await(3000, TimeUnit.MILLISECONDS));
+ assertTrue(messageLatch.await(5000, TimeUnit.MILLISECONDS));
+
+ final boolean batch1Exists = receivedMessageContexts.stream().anyMatch(
+ messageContext -> messageContext.getMessage().getBody().toString().equals("testSchedulingBatchMessages 1"));
+ final boolean batch2Exists = receivedMessageContexts.stream().anyMatch(
+ messageContext -> messageContext.getMessage().getBody().toString().equals("testSchedulingBatchMessages 2"));
+ final boolean batch3Exists = receivedMessageContexts.stream().anyMatch(
+ messageContext -> messageContext.getMessage().getBody().toString().equals("testSchedulingBatchMessages 3"));
+
+ assertTrue(batch1Exists, "test message body 1");
+ assertTrue(batch2Exists, "test message body 2");
+ assertTrue(batch3Exists, "test message body 3");
+
+ final boolean byteBody1Exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> Arrays.equals(messageContext.getMessage().getBody().toBytes(), byteBody1));
+ final boolean byteBody2Exists = receivedMessageContexts.stream()
+ .anyMatch(messageContext -> Arrays.equals(messageContext.getMessage().getBody().toBytes(), byteBody2));
+
+ assertTrue(byteBody1Exists, "test byte body 1");
+ assertTrue(byteBody2Exists, "test byte body 2");
+ }
+ }
+
+ @Test
+ void testSchedulingBatchMessagesWithSessions() throws InterruptedException {
+ final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(batchSessionClient);
+ messageLatch = new CountDownLatch(5);
+
+ try (ServiceBusProcessorClient processorClient = createSessionQueueProcessorClient()) {
+ processorClient.start();
+
+ final List<String> inputBatch = new LinkedList<>();
+ inputBatch.add("testSchedulingBatchMessages 1");
+ inputBatch.add("testSchedulingBatchMessages 2");
+ inputBatch.add("testSchedulingBatchMessages 3");
+ operations.scheduleMessages(inputBatch, OffsetDateTime.now(), null, null, null, "session-2");
+ //test bytes
+ final List<byte[]> inputBatch2 = new LinkedList<>();
+ byte[] byteBody1 = "test data".getBytes(StandardCharsets.UTF_8);
+ byte[] byteBody2 = "test data2".getBytes(StandardCharsets.UTF_8);
+ inputBatch2.add(byteBody1);
+ inputBatch2.add(byteBody2);
+ operations.scheduleMessages(inputBatch2, OffsetDateTime.now(), null, null, null, "session-2");
+
+ assertTrue(messageLatch.await(8000, TimeUnit.MILLISECONDS));
final boolean batch1Exists = receivedMessageContexts.stream().anyMatch(
messageContext -> messageContext.getMessage().getBody().toString().equals("testSchedulingBatchMessages 1"));