You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2024/02/21 10:33:49 UTC
(camel) branch CAMEL-20444-4.4.x created (now 9c5a59f0cca)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a change to branch CAMEL-20444-4.4.x
in repository https://gitbox.apache.org/repos/asf/camel.git
at 9c5a59f0cca CAMEL-20444 - Camel-Azure-Servicebus: Support setting of CorrelationId on producer
This branch includes the following new commits:
new 80ebb3562dc CAMEL-20444 - Camel-Azure-Servicebus: Support setting of CorrelationId on producer
new 9c5a59f0cca CAMEL-20444 - Camel-Azure-Servicebus: Support setting of CorrelationId on producer
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 02/02: CAMEL-20444 - Camel-Azure-Servicebus: Support setting of CorrelationId on producer
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch CAMEL-20444-4.4.x
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 9c5a59f0cca02a9c9a9a06625ace6923d950eab9
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Wed Feb 21 11:30:45 2024 +0100
CAMEL-20444 - Camel-Azure-Servicebus: Support setting of CorrelationId on producer
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../camel/catalog/components/azure-servicebus.json | 46 +++++++++++-----------
1 file changed, 23 insertions(+), 23 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-servicebus.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-servicebus.json
index 634bc4f8881..de0dd313deb 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-servicebus.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-servicebus.json
@@ -55,29 +55,29 @@
"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 [...]
"CamelAzureServiceBusContentType": { "index": 1, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the content type of the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#CONTENT_TYPE" },
- "CamelAzureServiceBusCorrelationId": { "index": 2, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets a correlation identifier.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#CORRELATION_ID" },
- "CamelAzureServiceBusDeadLetterErrorDescription": { "index": 3, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the description for a message that has been dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_ERROR_DESCRIPTION" },
- "CamelAzureServiceBusDeadLetterReason": { "index": 4, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the reason a message was dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_REASON" },
- "CamelAzureServiceBusDeadLetterSource": { "index": 5, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the name of the queue or subscription that this message was enqueued on, before it was dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_SOURCE" },
- "CamelAzureServiceBusDeliveryCount": { "index": 6, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the number of the times this message was delivered to clients.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DELIVERY_COUNT" },
- "CamelAzureServiceBusEnqueuedSequenceNumber": { "index": 7, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the enqueued sequence number assigned to a message by Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#ENQUEUED_SEQUENCE_NUMBER" },
- "CamelAzureServiceBusEnqueuedTime": { "index": 8, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which this message was enqueued in Azure Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#ENQUEUED_TIME" },
- "CamelAzureServiceBusExpiresAt": { "index": 9, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which this message will expire.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#EXPIRES_AT" },
- "CamelAzureServiceBusLockToken": { "index": 10, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the lock token for the current message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#LOCK_TOKEN" },
- "CamelAzureServiceBusLockedUntil": { "index": 11, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which the lock of this message expires.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#LOCKED_UNTIL" },
- "CamelAzureServiceBusMessageId": { "index": 12, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the identifier for the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#MESSAGE_ID" },
- "CamelAzureServiceBusPartitionKey": { "index": 13, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the partition key for sending a message to a partitioned entity.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#PARTITION_KEY" },
- "CamelAzureServiceBusRawAmqpMessage": { "index": 14, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "AmqpAnnotatedMessage", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The representation of message as defined by AMQP protocol.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#RAW_AMQP_MESSAGE" },
- "CamelAzureServiceBusReplyTo": { "index": 15, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the address of an entity to send replies to.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#REPLY_TO" },
- "CamelAzureServiceBusReplyToSessionId": { "index": 16, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets or sets a session identifier augmenting the ReplyTo address.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#REPLY_TO_SESSION_ID" },
- "CamelAzureServiceBusSequenceNumber": { "index": 17, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the unique number assigned to a message by Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SEQUENCE_NUMBER" },
- "CamelAzureServiceBusSessionId": { "index": 18, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the session id of the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SESSION_ID" },
- "CamelAzureServiceBusSubject": { "index": 19, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the subject for the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SUBJECT" },
- "CamelAzureServiceBusTimeToLive": { "index": 20, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "Duration", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the duration before this message expires.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#TIME_TO_LIVE" },
- "CamelAzureServiceBusTo": { "index": 21, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the to address.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#TO" },
- "CamelAzureServiceBusScheduledEnqueueTime": { "index": 22, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "(producer)Overrides the OffsetDateTime at which the message should appear in the Service Bus queue or topic. (consumer) Gets the scheduled enqueue time of this message.", "constantName": "org.apache.camel.component [...]
- "CamelAzureServiceBusServiceBusTransactionContext": { "index": 23, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "ServiceBusTransactionContext", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Overrides the transaction in service. This object just contains transaction id.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SERVICE_BUS_TRANSAC [...]
- "CamelAzureServiceBusProducerOperation": { "index": 24, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition", "enum": [ "sendMessages", "scheduleMessages" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Overrides the desired operation to be used in the producer.", "constantName": "org.apache.ca [...]
+ "CamelAzureServiceBusDeadLetterErrorDescription": { "index": 2, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the description for a message that has been dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_ERROR_DESCRIPTION" },
+ "CamelAzureServiceBusDeadLetterReason": { "index": 3, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the reason a message was dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_REASON" },
+ "CamelAzureServiceBusDeadLetterSource": { "index": 4, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the name of the queue or subscription that this message was enqueued on, before it was dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_SOURCE" },
+ "CamelAzureServiceBusDeliveryCount": { "index": 5, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the number of the times this message was delivered to clients.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DELIVERY_COUNT" },
+ "CamelAzureServiceBusEnqueuedSequenceNumber": { "index": 6, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the enqueued sequence number assigned to a message by Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#ENQUEUED_SEQUENCE_NUMBER" },
+ "CamelAzureServiceBusEnqueuedTime": { "index": 7, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which this message was enqueued in Azure Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#ENQUEUED_TIME" },
+ "CamelAzureServiceBusExpiresAt": { "index": 8, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which this message will expire.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#EXPIRES_AT" },
+ "CamelAzureServiceBusLockToken": { "index": 9, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the lock token for the current message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#LOCK_TOKEN" },
+ "CamelAzureServiceBusLockedUntil": { "index": 10, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which the lock of this message expires.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#LOCKED_UNTIL" },
+ "CamelAzureServiceBusMessageId": { "index": 11, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the identifier for the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#MESSAGE_ID" },
+ "CamelAzureServiceBusPartitionKey": { "index": 12, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the partition key for sending a message to a partitioned entity.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#PARTITION_KEY" },
+ "CamelAzureServiceBusRawAmqpMessage": { "index": 13, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "AmqpAnnotatedMessage", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The representation of message as defined by AMQP protocol.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#RAW_AMQP_MESSAGE" },
+ "CamelAzureServiceBusReplyTo": { "index": 14, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the address of an entity to send replies to.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#REPLY_TO" },
+ "CamelAzureServiceBusReplyToSessionId": { "index": 15, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets or sets a session identifier augmenting the ReplyTo address.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#REPLY_TO_SESSION_ID" },
+ "CamelAzureServiceBusSequenceNumber": { "index": 16, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the unique number assigned to a message by Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SEQUENCE_NUMBER" },
+ "CamelAzureServiceBusSessionId": { "index": 17, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the session id of the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SESSION_ID" },
+ "CamelAzureServiceBusSubject": { "index": 18, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the subject for the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SUBJECT" },
+ "CamelAzureServiceBusTimeToLive": { "index": 19, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "Duration", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the duration before this message expires.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#TIME_TO_LIVE" },
+ "CamelAzureServiceBusTo": { "index": 20, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the to address.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#TO" },
+ "CamelAzureServiceBusScheduledEnqueueTime": { "index": 21, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "(producer)Overrides the OffsetDateTime at which the message should appear in the Service Bus queue or topic. (consumer) Gets the scheduled enqueue time of this message.", "constantName": "org.apache.camel.component [...]
+ "CamelAzureServiceBusServiceBusTransactionContext": { "index": 22, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "ServiceBusTransactionContext", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Overrides the transaction in service. This object just contains transaction id.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SERVICE_BUS_TRANSAC [...]
+ "CamelAzureServiceBusProducerOperation": { "index": 23, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition", "enum": [ "sendMessages", "scheduleMessages" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Overrides the desired operation to be used in the producer.", "constantName": "org.apache.ca [...]
+ "CamelAzureServiceBusCorrelationId": { "index": 24, "kind": "header", "displayName": "", "group": "common", "label": "common", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets or Sets a correlation identifier.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#CORRELATION_ID" }
},
"properties": {
"topicOrQueueName": { "index": 0, "kind": "path", "displayName": "Topic Or Queue Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Selected topic name or the queue name, that is depending on serviceBusType config. For exampl [...]
(camel) 01/02: CAMEL-20444 - Camel-Azure-Servicebus: Support setting of CorrelationId on producer
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch CAMEL-20444-4.4.x
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 80ebb3562dc90218bd1d53321f74377e6b93029d
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Wed Feb 21 11:24:46 2024 +0100
CAMEL-20444 - Camel-Azure-Servicebus: Support setting of CorrelationId on producer
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../azure/servicebus/azure-servicebus.json | 46 +++++++++++-----------
.../azure/servicebus/ServiceBusConstants.java | 6 ++-
.../azure/servicebus/ServiceBusProducer.java | 14 +++++--
.../azure/servicebus/ServiceBusUtils.java | 10 +++--
.../operations/ServiceBusSenderOperations.java | 36 ++++++++++-------
.../azure/servicebus/ServiceBusUtilsTest.java | 10 ++---
.../integration/ServiceBusConsumerTest.java | 4 +-
.../operations/ServiceBusSenderOperationsTest.java | 20 +++++-----
8 files changed, 83 insertions(+), 63 deletions(-)
diff --git a/components/camel-azure/camel-azure-servicebus/src/generated/resources/org/apache/camel/component/azure/servicebus/azure-servicebus.json b/components/camel-azure/camel-azure-servicebus/src/generated/resources/org/apache/camel/component/azure/servicebus/azure-servicebus.json
index 634bc4f8881..de0dd313deb 100644
--- a/components/camel-azure/camel-azure-servicebus/src/generated/resources/org/apache/camel/component/azure/servicebus/azure-servicebus.json
+++ b/components/camel-azure/camel-azure-servicebus/src/generated/resources/org/apache/camel/component/azure/servicebus/azure-servicebus.json
@@ -55,29 +55,29 @@
"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 [...]
"CamelAzureServiceBusContentType": { "index": 1, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the content type of the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#CONTENT_TYPE" },
- "CamelAzureServiceBusCorrelationId": { "index": 2, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets a correlation identifier.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#CORRELATION_ID" },
- "CamelAzureServiceBusDeadLetterErrorDescription": { "index": 3, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the description for a message that has been dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_ERROR_DESCRIPTION" },
- "CamelAzureServiceBusDeadLetterReason": { "index": 4, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the reason a message was dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_REASON" },
- "CamelAzureServiceBusDeadLetterSource": { "index": 5, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the name of the queue or subscription that this message was enqueued on, before it was dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_SOURCE" },
- "CamelAzureServiceBusDeliveryCount": { "index": 6, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the number of the times this message was delivered to clients.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DELIVERY_COUNT" },
- "CamelAzureServiceBusEnqueuedSequenceNumber": { "index": 7, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the enqueued sequence number assigned to a message by Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#ENQUEUED_SEQUENCE_NUMBER" },
- "CamelAzureServiceBusEnqueuedTime": { "index": 8, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which this message was enqueued in Azure Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#ENQUEUED_TIME" },
- "CamelAzureServiceBusExpiresAt": { "index": 9, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which this message will expire.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#EXPIRES_AT" },
- "CamelAzureServiceBusLockToken": { "index": 10, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the lock token for the current message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#LOCK_TOKEN" },
- "CamelAzureServiceBusLockedUntil": { "index": 11, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which the lock of this message expires.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#LOCKED_UNTIL" },
- "CamelAzureServiceBusMessageId": { "index": 12, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the identifier for the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#MESSAGE_ID" },
- "CamelAzureServiceBusPartitionKey": { "index": 13, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the partition key for sending a message to a partitioned entity.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#PARTITION_KEY" },
- "CamelAzureServiceBusRawAmqpMessage": { "index": 14, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "AmqpAnnotatedMessage", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The representation of message as defined by AMQP protocol.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#RAW_AMQP_MESSAGE" },
- "CamelAzureServiceBusReplyTo": { "index": 15, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the address of an entity to send replies to.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#REPLY_TO" },
- "CamelAzureServiceBusReplyToSessionId": { "index": 16, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets or sets a session identifier augmenting the ReplyTo address.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#REPLY_TO_SESSION_ID" },
- "CamelAzureServiceBusSequenceNumber": { "index": 17, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the unique number assigned to a message by Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SEQUENCE_NUMBER" },
- "CamelAzureServiceBusSessionId": { "index": 18, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the session id of the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SESSION_ID" },
- "CamelAzureServiceBusSubject": { "index": 19, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the subject for the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SUBJECT" },
- "CamelAzureServiceBusTimeToLive": { "index": 20, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "Duration", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the duration before this message expires.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#TIME_TO_LIVE" },
- "CamelAzureServiceBusTo": { "index": 21, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the to address.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#TO" },
- "CamelAzureServiceBusScheduledEnqueueTime": { "index": 22, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "(producer)Overrides the OffsetDateTime at which the message should appear in the Service Bus queue or topic. (consumer) Gets the scheduled enqueue time of this message.", "constantName": "org.apache.camel.component [...]
- "CamelAzureServiceBusServiceBusTransactionContext": { "index": 23, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "ServiceBusTransactionContext", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Overrides the transaction in service. This object just contains transaction id.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SERVICE_BUS_TRANSAC [...]
- "CamelAzureServiceBusProducerOperation": { "index": 24, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition", "enum": [ "sendMessages", "scheduleMessages" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Overrides the desired operation to be used in the producer.", "constantName": "org.apache.ca [...]
+ "CamelAzureServiceBusDeadLetterErrorDescription": { "index": 2, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the description for a message that has been dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_ERROR_DESCRIPTION" },
+ "CamelAzureServiceBusDeadLetterReason": { "index": 3, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the reason a message was dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_REASON" },
+ "CamelAzureServiceBusDeadLetterSource": { "index": 4, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the name of the queue or subscription that this message was enqueued on, before it was dead-lettered.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DEAD_LETTER_SOURCE" },
+ "CamelAzureServiceBusDeliveryCount": { "index": 5, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the number of the times this message was delivered to clients.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#DELIVERY_COUNT" },
+ "CamelAzureServiceBusEnqueuedSequenceNumber": { "index": 6, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the enqueued sequence number assigned to a message by Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#ENQUEUED_SEQUENCE_NUMBER" },
+ "CamelAzureServiceBusEnqueuedTime": { "index": 7, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which this message was enqueued in Azure Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#ENQUEUED_TIME" },
+ "CamelAzureServiceBusExpiresAt": { "index": 8, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which this message will expire.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#EXPIRES_AT" },
+ "CamelAzureServiceBusLockToken": { "index": 9, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the lock token for the current message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#LOCK_TOKEN" },
+ "CamelAzureServiceBusLockedUntil": { "index": 10, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the datetime at which the lock of this message expires.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#LOCKED_UNTIL" },
+ "CamelAzureServiceBusMessageId": { "index": 11, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the identifier for the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#MESSAGE_ID" },
+ "CamelAzureServiceBusPartitionKey": { "index": 12, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the partition key for sending a message to a partitioned entity.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#PARTITION_KEY" },
+ "CamelAzureServiceBusRawAmqpMessage": { "index": 13, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "AmqpAnnotatedMessage", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The representation of message as defined by AMQP protocol.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#RAW_AMQP_MESSAGE" },
+ "CamelAzureServiceBusReplyTo": { "index": 14, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the address of an entity to send replies to.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#REPLY_TO" },
+ "CamelAzureServiceBusReplyToSessionId": { "index": 15, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets or sets a session identifier augmenting the ReplyTo address.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#REPLY_TO_SESSION_ID" },
+ "CamelAzureServiceBusSequenceNumber": { "index": 16, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the unique number assigned to a message by Service Bus.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SEQUENCE_NUMBER" },
+ "CamelAzureServiceBusSessionId": { "index": 17, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the session id of the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SESSION_ID" },
+ "CamelAzureServiceBusSubject": { "index": 18, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the subject for the message.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SUBJECT" },
+ "CamelAzureServiceBusTimeToLive": { "index": 19, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "Duration", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the duration before this message expires.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#TIME_TO_LIVE" },
+ "CamelAzureServiceBusTo": { "index": 20, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets the to address.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#TO" },
+ "CamelAzureServiceBusScheduledEnqueueTime": { "index": 21, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "OffsetDateTime", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "(producer)Overrides the OffsetDateTime at which the message should appear in the Service Bus queue or topic. (consumer) Gets the scheduled enqueue time of this message.", "constantName": "org.apache.camel.component [...]
+ "CamelAzureServiceBusServiceBusTransactionContext": { "index": 22, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "ServiceBusTransactionContext", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Overrides the transaction in service. This object just contains transaction id.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#SERVICE_BUS_TRANSAC [...]
+ "CamelAzureServiceBusProducerOperation": { "index": 23, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition", "enum": [ "sendMessages", "scheduleMessages" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Overrides the desired operation to be used in the producer.", "constantName": "org.apache.ca [...]
+ "CamelAzureServiceBusCorrelationId": { "index": 24, "kind": "header", "displayName": "", "group": "common", "label": "common", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Gets or Sets a correlation identifier.", "constantName": "org.apache.camel.component.azure.servicebus.ServiceBusConstants#CORRELATION_ID" }
},
"properties": {
"topicOrQueueName": { "index": 0, "kind": "path", "displayName": "Topic Or Queue Name", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.azure.servicebus.ServiceBusConfiguration", "configurationField": "configuration", "description": "Selected topic name or the queue name, that is depending on serviceBusType config. For exampl [...]
diff --git a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConstants.java b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConstants.java
index cde4ff54b16..5ce1b5a9348 100644
--- a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConstants.java
+++ b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusConstants.java
@@ -29,8 +29,6 @@ public final class ServiceBusConstants {
public static final String APPLICATION_PROPERTIES = HEADER_PREFIX + "ApplicationProperties";
@Metadata(label = "consumer", description = "Gets the content type of the message.", javaType = "String")
public static final String CONTENT_TYPE = HEADER_PREFIX + "ContentType";
- @Metadata(label = "consumer", description = "Gets a correlation identifier.", javaType = "String")
- public static final String CORRELATION_ID = HEADER_PREFIX + "CorrelationId";
@Metadata(label = "consumer", description = "Gets the description for a message that has been dead-lettered.",
javaType = "String")
public static final String DEAD_LETTER_ERROR_DESCRIPTION = HEADER_PREFIX + "DeadLetterErrorDescription";
@@ -98,6 +96,10 @@ public final class ServiceBusConstants {
javaType = "org.apache.camel.component.azure.servicebus.ServiceBusProducerOperationDefinition")
public static final String PRODUCER_OPERATION = HEADER_PREFIX + "ProducerOperation";
+ // headers evaluated by the producer and consumer
+ @Metadata(label = "common", description = "Gets or Sets a correlation identifier.", javaType = "String")
+ public static final String CORRELATION_ID = HEADER_PREFIX + "CorrelationId";
+
private ServiceBusConstants() {
}
}
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 e6300d342f2..b390a6a6cc5 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
@@ -149,20 +149,23 @@ public class ServiceBusProducer extends DefaultAsyncProducer {
final Object inputBody = exchange.getMessage().getBody();
final Map<String, Object> applicationProperties
= exchange.getMessage().getHeader(ServiceBusConstants.APPLICATION_PROPERTIES, Map.class);
+ final String correlationId = exchange.getMessage().getHeader(ServiceBusConstants.CORRELATION_ID, String.class);
Mono<Void> sendMessageAsync;
if (inputBody instanceof Iterable<?>) {
sendMessageAsync
= serviceBusSenderOperations.sendMessages(convertBodyToList((Iterable<?>) inputBody),
- configurationOptionsProxy.getServiceBusTransactionContext(exchange), applicationProperties);
+ configurationOptionsProxy.getServiceBusTransactionContext(exchange), applicationProperties,
+ correlationId);
} else {
Object convertedBody = inputBody instanceof BinaryData ? inputBody
: getConfiguration().isBinary() ? convertBodyToBinary(exchange)
: exchange.getMessage().getBody(String.class);
sendMessageAsync = serviceBusSenderOperations.sendMessages(convertedBody,
- configurationOptionsProxy.getServiceBusTransactionContext(exchange), applicationProperties);
+ configurationOptionsProxy.getServiceBusTransactionContext(exchange), applicationProperties,
+ correlationId);
}
subscribeToMono(sendMessageAsync, exchange, noop -> {
@@ -176,6 +179,7 @@ public class ServiceBusProducer extends DefaultAsyncProducer {
final Object inputBody = exchange.getMessage().getBody();
final Map<String, Object> applicationProperties
= exchange.getMessage().getHeader(ServiceBusConstants.APPLICATION_PROPERTIES, Map.class);
+ final String correlationId = exchange.getMessage().getHeader(ServiceBusConstants.CORRELATION_ID, String.class);
Mono<List<Long>> scheduleMessagesAsync;
@@ -184,7 +188,8 @@ public class ServiceBusProducer extends DefaultAsyncProducer {
= serviceBusSenderOperations.scheduleMessages(convertBodyToList((Iterable<?>) inputBody),
configurationOptionsProxy.getScheduledEnqueueTime(exchange),
configurationOptionsProxy.getServiceBusTransactionContext(exchange),
- applicationProperties);
+ applicationProperties,
+ correlationId);
} else {
Object convertedBody = inputBody instanceof BinaryData ? inputBody
: getConfiguration().isBinary() ? convertBodyToBinary(exchange)
@@ -193,7 +198,8 @@ public class ServiceBusProducer extends DefaultAsyncProducer {
= serviceBusSenderOperations.scheduleMessages(convertedBody,
configurationOptionsProxy.getScheduledEnqueueTime(exchange),
configurationOptionsProxy.getServiceBusTransactionContext(exchange),
- applicationProperties);
+ applicationProperties,
+ correlationId);
}
subscribeToMono(scheduleMessagesAsync, exchange,
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 4f26dbd9d5f..120b8c5453d 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
@@ -22,6 +22,7 @@ import java.util.stream.StreamSupport;
import com.azure.core.util.BinaryData;
import com.azure.messaging.servicebus.ServiceBusMessage;
+import org.apache.camel.util.ObjectHelper;
public final class ServiceBusUtils {
@@ -29,7 +30,7 @@ public final class ServiceBusUtils {
}
public static ServiceBusMessage createServiceBusMessage(
- final Object data, final Map<String, Object> applicationProperties) {
+ final Object data, final Map<String, Object> applicationProperties, final String correlationId) {
ServiceBusMessage serviceBusMessage;
if (data instanceof String) {
serviceBusMessage = new ServiceBusMessage((String) data);
@@ -43,13 +44,16 @@ public final class ServiceBusUtils {
if (applicationProperties != null) {
serviceBusMessage.getRawAmqpMessage().getApplicationProperties().putAll(applicationProperties);
}
+ if (ObjectHelper.isNotEmpty(correlationId)) {
+ serviceBusMessage.setCorrelationId(correlationId);
+ }
return serviceBusMessage;
}
public static Iterable<ServiceBusMessage> createServiceBusMessages(
- final Iterable<?> data, final Map<String, Object> applicationProperties) {
+ final Iterable<?> data, final Map<String, Object> applicationProperties, final String correlationId) {
return StreamSupport.stream(data.spliterator(), false)
- .map(obj -> createServiceBusMessage(obj, applicationProperties))
+ .map(obj -> createServiceBusMessage(obj, applicationProperties, correlationId))
.collect(Collectors.toList());
}
}
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 612168261f5..c299e9cad32 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
@@ -41,35 +41,39 @@ public class ServiceBusSenderOperations {
public Mono<Void> sendMessages(
final Object data,
final ServiceBusTransactionContext context,
- final Map<String, Object> applicationProperties) {
+ final Map<String, Object> applicationProperties,
+ final String correlationId) {
if (data instanceof Iterable<?>) {
- return sendMessages((Iterable<?>) data, context, applicationProperties);
+ return sendMessages((Iterable<?>) data, context, applicationProperties, correlationId);
}
- return sendMessage(data, context, applicationProperties);
+ return sendMessage(data, context, applicationProperties, correlationId);
}
public Mono<List<Long>> scheduleMessages(
final Object data,
final OffsetDateTime scheduledEnqueueTime,
final ServiceBusTransactionContext context,
- final Map<String, Object> applicationProperties) {
+ final Map<String, Object> applicationProperties,
+ final String correlationId) {
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);
+ return scheduleMessages((Iterable<?>) data, scheduledEnqueueTime, context, applicationProperties, correlationId);
}
- return scheduleMessage(data, scheduledEnqueueTime, context, applicationProperties);
+ return scheduleMessage(data, scheduledEnqueueTime, context, applicationProperties, correlationId);
}
private Mono<Void> sendMessages(
final Iterable<?> data,
final ServiceBusTransactionContext context,
- final Map<String, Object> applicationProperties) {
- final Iterable<ServiceBusMessage> messages = ServiceBusUtils.createServiceBusMessages(data, applicationProperties);
+ final Map<String, Object> applicationProperties,
+ final String correlationId) {
+ final Iterable<ServiceBusMessage> messages
+ = ServiceBusUtils.createServiceBusMessages(data, applicationProperties, correlationId);
if (ObjectHelper.isEmpty(context)) {
return client.sendMessages(messages);
@@ -81,8 +85,9 @@ public class ServiceBusSenderOperations {
private Mono<Void> sendMessage(
final Object data,
final ServiceBusTransactionContext context,
- final Map<String, Object> applicationProperties) {
- final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data, applicationProperties);
+ final Map<String, Object> applicationProperties,
+ final String correlationId) {
+ final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data, applicationProperties, correlationId);
if (ObjectHelper.isEmpty(context)) {
return client.sendMessage(message);
@@ -95,8 +100,9 @@ public class ServiceBusSenderOperations {
final Object data,
final OffsetDateTime scheduledEnqueueTime,
final ServiceBusTransactionContext context,
- final Map<String, Object> applicationProperties) {
- final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data, applicationProperties);
+ final Map<String, Object> applicationProperties,
+ final String correlationId) {
+ final ServiceBusMessage message = ServiceBusUtils.createServiceBusMessage(data, applicationProperties, correlationId);
if (ObjectHelper.isEmpty(context)) {
return client.scheduleMessage(message, scheduledEnqueueTime)
@@ -110,8 +116,10 @@ public class ServiceBusSenderOperations {
private Mono<List<Long>> scheduleMessages(
final Iterable<?> data, final OffsetDateTime scheduledEnqueueTime,
final ServiceBusTransactionContext context,
- final Map<String, Object> applicationProperties) {
- final Iterable<ServiceBusMessage> messages = ServiceBusUtils.createServiceBusMessages(data, applicationProperties);
+ final Map<String, Object> applicationProperties,
+ final String correlationId) {
+ final Iterable<ServiceBusMessage> messages
+ = ServiceBusUtils.createServiceBusMessages(data, applicationProperties, correlationId);
if (ObjectHelper.isEmpty(context)) {
return client.scheduleMessages(messages, scheduledEnqueueTime)
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 f952bc81514..33951cf2584 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,18 @@ public class ServiceBusUtilsTest {
@Test
void testCreateServiceBusMessage() {
// test string
- final ServiceBusMessage message1 = ServiceBusUtils.createServiceBusMessage("test string", null);
+ final ServiceBusMessage message1 = ServiceBusUtils.createServiceBusMessage("test string", null, null);
assertEquals("test string", message1.getBody().toString());
// test int
- final ServiceBusMessage message2 = ServiceBusUtils.createServiceBusMessage(String.valueOf(12345), null);
+ final ServiceBusMessage message2 = ServiceBusUtils.createServiceBusMessage(String.valueOf(12345), null, null);
assertEquals("12345", message2.getBody().toString());
//test bytes
byte[] testByteBody = "test string".getBytes(StandardCharsets.UTF_8);
- final ServiceBusMessage message3 = ServiceBusUtils.createServiceBusMessage(testByteBody, null);
+ final ServiceBusMessage message3 = ServiceBusUtils.createServiceBusMessage(testByteBody, null, null);
assertArrayEquals(testByteBody, message3.getBody().toBytes());
}
@@ -53,7 +53,7 @@ public class ServiceBusUtilsTest {
inputMessages.add("test data");
inputMessages.add(String.valueOf(12345));
- final Iterable<ServiceBusMessage> busMessages = ServiceBusUtils.createServiceBusMessages(inputMessages, null);
+ final Iterable<ServiceBusMessage> busMessages = ServiceBusUtils.createServiceBusMessages(inputMessages, null, null);
assertTrue(StreamSupport.stream(busMessages.spliterator(), false)
.anyMatch(record -> record.getBody().toString().equals("test data")));
@@ -67,7 +67,7 @@ public class ServiceBusUtilsTest {
inputMessages2.add(byteBody1);
inputMessages2.add(byteBody2);
- final Iterable<ServiceBusMessage> busMessages2 = ServiceBusUtils.createServiceBusMessages(inputMessages2, null);
+ final Iterable<ServiceBusMessage> busMessages2 = ServiceBusUtils.createServiceBusMessages(inputMessages2, null, null);
assertTrue(StreamSupport.stream(busMessages2.spliterator(), false)
.anyMatch(message -> Arrays.equals(message.getBody().toBytes(), byteBody1)));
diff --git a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerTest.java b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerTest.java
index dbb34aab4c0..9ff5ebb89ef 100644
--- a/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerTest.java
+++ b/components/camel-azure/camel-azure-servicebus/src/test/java/org/apache/camel/component/azure/servicebus/integration/ServiceBusConsumerTest.java
@@ -53,7 +53,7 @@ class ServiceBusConsumerTest extends BaseCamelServiceBusTestSupport {
inputBatch.add("test batch 3");
new ServiceBusSenderOperations(new ServiceBusSenderAsyncClientWrapper(senderAsyncClient))
- .sendMessages(inputBatch, null, null)
+ .sendMessages(inputBatch, null, null, null)
.block();
// test the data now
@@ -83,7 +83,7 @@ class ServiceBusConsumerTest extends BaseCamelServiceBusTestSupport {
inputBatch.add("peek test batch 3");
new ServiceBusSenderOperations(new ServiceBusSenderAsyncClientWrapper(senderAsyncClient))
- .sendMessages(inputBatch, null, null)
+ .sendMessages(inputBatch, null, null, null)
.block();
// test the data now
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 f8a53540b5b..b131609adc5 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
@@ -88,7 +88,7 @@ public class ServiceBusSenderOperationsTest {
void testSendSingleMessage() {
final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(clientSenderWrapper);
- operations.sendMessages("test data", null, Map.of("customKey", "customValue")).block();
+ operations.sendMessages("test data", null, Map.of("customKey", "customValue"), null).block();
final boolean exists = StreamSupport.stream(clientReceiverWrapper.receiveMessages().toIterable().spliterator(), false)
.anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString().equals("test data"));
@@ -97,7 +97,7 @@ public class ServiceBusSenderOperationsTest {
//test bytes
byte[] testByteBody = "test data".getBytes(StandardCharsets.UTF_8);
- operations.sendMessages(testByteBody, null, Map.of("customKey", "customValue")).block();
+ operations.sendMessages(testByteBody, null, Map.of("customKey", "customValue"), null).block();
final boolean exists2 = StreamSupport.stream(clientReceiverWrapper.receiveMessages().toIterable().spliterator(), false)
.anyMatch(serviceBusReceivedMessage -> Arrays.equals(serviceBusReceivedMessage.getBody().toBytes(),
testByteBody));
@@ -105,7 +105,7 @@ public class ServiceBusSenderOperationsTest {
// test if we have something other than string or byte[]
assertThrows(IllegalArgumentException.class, () -> {
- operations.sendMessages(12345, null, null).block();
+ operations.sendMessages(12345, null, null, null).block();
});
}
@@ -118,7 +118,7 @@ public class ServiceBusSenderOperationsTest {
inputBatch.add("test batch 2");
inputBatch.add("test batch 3");
- operations.sendMessages(inputBatch, null, null).block();
+ operations.sendMessages(inputBatch, null, null, null).block();
final Spliterator<ServiceBusReceivedMessage> receivedMessages
= clientReceiverWrapper.receiveMessages().toIterable().spliterator();
@@ -143,7 +143,7 @@ public class ServiceBusSenderOperationsTest {
inputBatch2.add(byteBody1);
inputBatch2.add(byteBody2);
- operations.sendMessages(inputBatch2, null, null).block();
+ operations.sendMessages(inputBatch2, null, null, null).block();
final Spliterator<ServiceBusReceivedMessage> receivedMessages2
= clientReceiverWrapper.receiveMessages().toIterable().spliterator();
@@ -161,7 +161,7 @@ public class ServiceBusSenderOperationsTest {
void testScheduleMessage() {
final ServiceBusSenderOperations operations = new ServiceBusSenderOperations(clientSenderWrapper);
- operations.scheduleMessages("testScheduleMessage", OffsetDateTime.now(), null, null).block();
+ operations.scheduleMessages("testScheduleMessage", OffsetDateTime.now(), null, null, null).block();
final boolean exists = StreamSupport.stream(clientReceiverWrapper.receiveMessages().toIterable().spliterator(), false)
.anyMatch(serviceBusReceivedMessage -> serviceBusReceivedMessage.getBody().toString()
@@ -171,7 +171,7 @@ public class ServiceBusSenderOperationsTest {
//test bytes
byte[] testByteBody = "test data".getBytes(StandardCharsets.UTF_8);
- operations.scheduleMessages(testByteBody, OffsetDateTime.now(), null, null).block();
+ operations.scheduleMessages(testByteBody, OffsetDateTime.now(), null, null, null).block();
final boolean exists2 = StreamSupport.stream(clientReceiverWrapper.receiveMessages().toIterable().spliterator(), false)
.anyMatch(serviceBusReceivedMessage -> Arrays.equals(serviceBusReceivedMessage.getBody().toBytes(),
testByteBody));
@@ -179,7 +179,7 @@ public class ServiceBusSenderOperationsTest {
// test if we have something other than string or byte[]
assertThrows(IllegalArgumentException.class, () -> {
- operations.scheduleMessages(12345, OffsetDateTime.now(), null, null).block();
+ operations.scheduleMessages(12345, OffsetDateTime.now(), null, null, null).block();
});
}
@@ -192,7 +192,7 @@ public class ServiceBusSenderOperationsTest {
inputBatch.add("testSchedulingBatchMessages 2");
inputBatch.add("testSchedulingBatchMessages 3");
- operations.scheduleMessages(inputBatch, OffsetDateTime.now(), null, null).block();
+ operations.scheduleMessages(inputBatch, OffsetDateTime.now(), null, null, null).block();
final Spliterator<ServiceBusReceivedMessage> receivedMessages
= clientReceiverWrapper.receiveMessages().toIterable().spliterator();
@@ -220,7 +220,7 @@ public class ServiceBusSenderOperationsTest {
inputBatch2.add(byteBody1);
inputBatch2.add(byteBody2);
- operations.scheduleMessages(inputBatch2, OffsetDateTime.now(), null, null).block();
+ operations.scheduleMessages(inputBatch2, OffsetDateTime.now(), null, null, null).block();
final Spliterator<ServiceBusReceivedMessage> receivedMessages2
= clientReceiverWrapper.receiveMessages().toIterable().spliterator();