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 f866bb3604e5 CAMEL-24961: Catalog body types for cloud messaging
(#28275)
f866bb3604e5 is described below
commit f866bb3604e50892436dcb5f4b855ee3dee46d13
Author: Claus Ibsen <[email protected]>
AuthorDate: Sat Oct 24 09:08:31 2026 +0200
CAMEL-24961: Catalog body types for cloud messaging (#28275)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
---
.../catalog/components/aws2-kinesis-firehose.json | 4 ++
.../camel/catalog/components/aws2-kinesis.json | 5 +++
.../apache/camel/catalog/components/aws2-sns.json | 4 ++
.../apache/camel/catalog/components/aws2-sqs.json | 2 +-
.../camel/catalog/components/azure-eventhubs.json | 5 +++
.../camel/catalog/components/azure-servicebus.json | 5 +++
.../catalog/components/azure-storage-queue.json | 5 +++
.../camel/catalog/components/google-pubsub.json | 5 +++
.../aws2/firehose/aws2-kinesis-firehose.json | 4 ++
.../camel/component/aws2/kinesis/aws2-kinesis.json | 5 +++
.../aws2/firehose/KinesisFirehose2Endpoint.java | 42 +++++++++++++++++
.../component/aws2/kinesis/Kinesis2Endpoint.java | 18 ++++++++
.../apache/camel/component/aws2/sns/aws2-sns.json | 4 ++
.../camel/component/aws2/sns/Sns2Endpoint.java | 12 +++++
.../apache/camel/component/aws2/sqs/aws2-sqs.json | 2 +-
.../camel/component/aws2/sqs/Sqs2Endpoint.java | 11 ++---
.../component/azure/eventhubs/azure-eventhubs.json | 5 +++
.../azure/eventhubs/EventHubsEndpoint.java | 12 +++++
.../azure/servicebus/azure-servicebus.json | 5 +++
.../azure/servicebus/ServiceBusEndpoint.java | 14 ++++++
.../azure/storage/queue/azure-storage-queue.json | 5 +++
.../azure/storage/queue/QueueEndpoint.java | 39 ++++++++++++++++
.../component/google/pubsub/google-pubsub.json | 5 +++
.../google/pubsub/GooglePubsubEndpoint.java | 12 +++++
.../camel/dsl/yaml/validator/TextBodyFlowTest.java | 52 ++++++++++++++++++++++
25 files changed, 275 insertions(+), 7 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis-firehose.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis-firehose.json
index eb7e0ba1ffc5..448eba5c7a09 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis-firehose.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis-firehose.json
@@ -54,6 +54,10 @@
"CamelAwsKinesisFirehoseEncrypted": { "index": 5, "kind": "header",
"displayName": "", "group": "sendBatchRecord", "label": "sendBatchRecord",
"required": false, "javaType": "Boolean", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description":
"Whether the batch operation was encrypted.", "constantName":
"org.apache.camel.component.aws2.firehose.KinesisFirehose2Constants#ENCRYPTED"
},
"CamelAwsKinesisFirehoseDeliveryStreamStatus": { "index": 6, "kind":
"header", "displayName": "", "group": "describeDeliveryStream", "label":
"describeDeliveryStream", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The status of the delivery stream.", "constantName":
"org.apache.camel.component.aws2.firehose.KinesisFirehose2Constants#DELIVERY_STREAM_STATUS"
}
},
+ "bodyTypes": {
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "operation", "value": "sendBatchRecord", "carriers":
[ "list", "pojo" ] }, { "option": "operation", "value": "createDeliveryStream",
"carriers": [ "pojo" ], "javaType":
"software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamRequest" },
{ "option": "operation", "value": "updateDestination", "carriers": [ "pojo" ],
"javaType": "software.amazon.awssdk.services.firehose.model.UpdateD [...]
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "sendBatchRecord", "carriers": [ "pojo" ], "javaType":
"software.amazon.awssdk.services.firehose.model.PutRecordBatchResponse" }, {
"option": "operation", "value": "createDeliveryStream", "carriers": [ "pojo" ],
"javaType":
"software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamResponse"
}, { "option": "operation", "value": "deleteDeliveryStream", "carriers": [
"pojo" ], "javaTy [...]
+ },
"properties": {
"streamName": { "index": 0, "kind": "path", "displayName": "Stream Name",
"group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration",
"configurationField": "configuration", "description": "Name of the stream" },
"cborEnabled": { "index": 1, "kind": "parameter", "displayName": "Cbor
Enabled", "group": "common", "label": "common", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration",
"configurationField": "configuration", "description": "This option will set the
CBOR_ENABLED property during the execution" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
index 40e1eb9595bb..7bc42d67463f 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
@@ -72,6 +72,11 @@
"CamelAwsKinesisFailedRecordCount": { "index": 7, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The number of records that failed in a
batch put operation.", "constantName":
"org.apache.camel.component.aws2.kinesis.Kinesis2Constants#FAILED_RECORD_COUNT"
},
"CamelAwsKinesisRecordCount": { "index": 8, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The total number of records in a batch
put operation.", "constantName":
"org.apache.camel.component.aws2.kinesis.Kinesis2Constants#RECORD_COUNT" }
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"byOption": [ { "option": "useKclConsumers", "value": "true", "carriers": [
"pojo" ], "javaType": "java.nio.ByteBuffer" } ], "description": "The data of
the record as a byte array, one message per record, with its partition key,
sequence number and shard id in headers. With useKclConsumers=true a
java.nio.ByteBuffer: convert it to a String or bytes for the steps that read
text" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "list" ],
"description": "The data of the record: text, bytes, a stream or a file. A List
or another Iterable is sent as a batch, one record per element, and each
element must be text, bytes, a ByteBuffer or a stream. A Map or a POJO fails:
marshal it first" },
+ "producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged,
with the sequence number and the shard id of the record in headers, or the
number of records of a batch" }
+ },
"properties": {
"streamName": { "index": 0, "kind": "path", "displayName": "Stream Name",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Name of the stream" },
"cborEnabled": { "index": 1, "kind": "parameter", "displayName": "Cbor
Enabled", "group": "common", "label": "common", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "This option will set the
CBOR_ENABLED property during the execution" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sns.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sns.json
index 7329d7204a10..a95e468d1634 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sns.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sns.json
@@ -64,6 +64,10 @@
"CamelAwsSnsFailedMessageCount": { "index": 4, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The number of failed messages in a
batch publish operation.", "constantName":
"org.apache.camel.component.aws2.sns.Sns2Constants#FAILED_MESSAGE_COUNT" },
"CamelAwsSnsSuccessfulMessageCount": { "index": 5, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The number of successful messages in a
batch publish operation.", "constantName":
"org.apache.camel.component.aws2.sns.Sns2Constants#SUCCESSFUL_MESSAGE_COUNT" }
},
+ "bodyTypes": {
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "batchEnabled", "value": "true", "carriers": [ "list"
] } ], "description": "The message to publish, as text: bytes, a stream or a
file are read as text, and a Map or a POJO is sent as its toString(): marshal
it first. With messageStructure=json, a JSON object with a message per
protocol; with batchEnabled=true, a List of the PublishBatchRequestEntry of the
AWS SDK" },
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"batchEnabled", "value": "true", "carriers": [ "pojo" ], "javaType":
"software.amazon.awssdk.services.sns.model.PublishBatchResponse" } ],
"description": "Unchanged after a publish, with the id of the message in a
header. With batchEnabled=true, the PublishBatchResponse of the AWS SDK" }
+ },
"properties": {
"topicNameOrArn": { "index": 0, "kind": "path", "displayName": "Topic Name
Or Arn", "group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "Topic name or ARN" },
"autoCreateTopic": { "index": 1, "kind": "parameter", "displayName": "Auto
Create Topic", "group": "producer", "label": "", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.aws2.sns.Sns2Configuration", "configurationField":
"configuration", "description": "Setting the auto-creation of the topic" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sqs.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sqs.json
index 8ff172f1e1f3..3f8288b43e5b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sqs.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-sqs.json
@@ -95,7 +95,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "text" ], "javaType": "java.lang.String",
"description": "The body of the SQS message" },
- "producerIn": { "carriers": [ "text" ], "byOption": [ { "option":
"operation", "value": "sendBatchMessage", "carriers": [ "list", "text" ] } ],
"description": "The body of the message to send, as text, so a Map or a POJO is
sent as its toString(): marshal it first. sendBatchMessage takes a List of
text, or text that batchSeparator splits" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "operation", "value": "sendBatchMessage", "carriers":
[ "list", "text", "pojo" ] } ], "description": "The body of the message to
send, as text: bytes, a stream or a file are read as text, and a Map or a POJO
is sent as its toString(): marshal it first. sendBatchMessage takes a List of
text, text that batchSeparator splits, or a SendMessageBatchRequest" },
"producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "sendBatchMessage", "carriers": [ "pojo" ], "javaType":
"software.amazon.awssdk.services.sqs.model.SendMessageBatchResponse" }, {
"option": "operation", "value": "deleteMessage", "carriers": [ "pojo" ],
"javaType": "software.amazon.awssdk.services.sqs.model.DeleteMessageResponse"
}, { "option": "operation", "value": "listQueues", "carriers": [ "pojo" ],
"javaType": "software.amazon.awssdk.se [...]
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-eventhubs.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-eventhubs.json
index 77145836e81e..a81217609b1a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-eventhubs.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-eventhubs.json
@@ -60,6 +60,11 @@
"CamelMessageTimestamp": { "index": 6, "kind": "header", "displayName":
"", "group": "consumer", "label": "consumer", "required": false, "javaType":
"long", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The timestamp of the message", "constantName":
"org.apache.camel.component.azure.eventhubs.EventHubsConstants#MESSAGE_TIMESTAMP"
},
"CamelAzureEventHubsCheckpointUpdatedBy": { "index": 7, "kind": "header",
"displayName": "", "group": "consumer", "label": "consumer", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "It sets the reason for the checkpoint
to have been updated. This is only present on a received EventData.",
"constantName":
"org.apache.camel.component.azure.eventhubs.EventHubsConstants#CHECKPOINT_UPDATED_BY"
}
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"description": "The body of the event, one message per event" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "list" ],
"description": "What converts to a byte array: text, bytes, a stream or a file,
sent as one event; a Map, a number or a POJO fails: marshal it first. A List is
sent as one event per element, in one batch, and an element that is an Exchange
or a Message is sent with its headers" },
+ "producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged: a
send gives no result, and no header is set" }
+ },
"properties": {
"namespace": { "index": 0, "kind": "path", "displayName": "Namespace",
"group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.azure.eventhubs.EventHubsConfiguration",
"configurationField": "configuration", "description": "EventHubs namespace
created in Azure Portal." },
"eventHubName": { "index": 1, "kind": "path", "displayName": "Event Hub
Name", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.azure.eventhubs.EventHubsConfiguration",
"configurationField": "configuration", "description": "EventHubs name under a
specific namespace." },
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 80ca2d410803..9fd52d36a0e0 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
@@ -82,6 +82,11 @@
"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"
}
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "pojo" ], "javaType":
"com.azure.core.util.BinaryData", "description": "The payload of the Service
Bus message as an Azure BinaryData, which converts to a String but not to bytes
or a stream: convert it to a String before an unmarshal, a file or another
producer" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "list" ],
"description": "Sent as text by default: bytes, a stream or a file are read as
text, so binary content needs binary=true. A BinaryData, as the consumer gives
it, is sent as it is, and a Map or a POJO is sent as its toString(): marshal it
first. A List is sent as one message per element" },
+ "producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged,
also after scheduleMessages" }
+ },
"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 [...]
"amqpRetryOptions": { "index": 1, "kind": "parameter", "displayName":
"Amqp Retry Options", "group": "common", "label": "common", "required": false,
"type": "object", "javaType": "com.azure.core.amqp.AmqpRetryOptions",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.azure.servicebus.ServiceBusConfiguration",
"configurationField": "configuration", "description": "Sets the retry options
for Service Bus clients. If not specifie [...]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-storage-queue.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-storage-queue.json
index dbae457091d6..6f772e736274 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-storage-queue.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-storage-queue.json
@@ -61,6 +61,11 @@
"CamelAzureStorageQueueTimeToLive": { "index": 14, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Duration", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "(sendMessage) How long the
message will stay alive in the queue. If unset the value will default to 7
days, if -1 is passed the message will not expire. The time to live must be -1
or any positive number.", "constantNa [...]
"CamelAzureStorageQueueCreateQueue": { "index": 15, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "(sendMessage) When is set to true, the
queue will be automatically created when sending messages to the queue.",
"constantName":
"org.apache.camel.component.azure.storage.queue.QueueConstants#CREATE_QUEUE" }
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "stream", "null" ], "description": "The
text of the queue message as a stream, with its id and pop receipt in headers;
no body for a message without text" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "operation", "value": "updateMessage", "carriers": [
"text", "bytes", "stream", "file", "null" ] }, { "option": "operation",
"value": "listQueues", "carriers": [ "any" ] }, { "option": "operation",
"value": "createQueue", "carriers": [ "any" ] }, { "option": "operation",
"value": "deleteQueue", "carriers": [ "any" ] }, { "option": "operation",
"value": "clearQueue", "carriers": [ "any" ] }, { [...]
+ "producerOut": { "carriers": [ "value" ], "javaType": "java.lang.Boolean",
"byOption": [ { "option": "operation", "value": "listQueues", "carriers": [
"list" ], "javaType":
"java.util.List<com.azure.storage.queue.models.QueueItem>" }, { "option":
"operation", "value": "receiveMessages", "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.storage.queue.models.QueueMessageItem>" }, {
"option": "operation", "value": "peekMessages", "carriers": [ "list" ],
"javaType": "java.uti [...]
+ },
"properties": {
"accountName": { "index": 0, "kind": "path", "displayName": "Account
Name", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.azure.storage.queue.QueueConfiguration",
"configurationField": "configuration", "description": "Azure account name to be
used for authentication with azure queue services" },
"queueName": { "index": 1, "kind": "path", "displayName": "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.storage.queue.QueueConfiguration",
"configurationField": "configuration", "description": "The queue resource name"
},
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/google-pubsub.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/google-pubsub.json
index 2da993e22ea3..765633f3cd31 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/google-pubsub.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/google-pubsub.json
@@ -45,6 +45,11 @@
"CamelGooglePubsubAcknowledge": { "index": 5, "kind": "header",
"displayName": "", "group": "consumer", "label": "consumer", "required": false,
"javaType":
"org.apache.camel.component.google.pubsub.consumer.GooglePubsubAcknowledge",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Can be used to manually acknowledge or
negative-acknowledge a message when ackMode=NONE.", "constantName":
"org.apache.camel.component.google.pubsub.GooglePub [...]
"CamelGooglePubsubDeliveryAttempt": { "index": 6, "kind": "header",
"displayName": "", "group": "consumer", "label": "consumer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The delivery attempt counter received
from PubSub. This is the approximate number of times the message has been
delivered. This will be 1 for the first delivery. This feature requires a
dead-letter policy to be configure [...]
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"description": "The data of the Pub\/Sub message as a byte array, with its
attributes in headers: convert or unmarshal it" },
+ "producerIn": { "carriers": [ "text", "bytes" ], "dependsOn": [
"serializer" ], "description": "Text is sent as UTF-8 and bytes as they are; a
List of the exchanges of a grouped aggregation is sent as one message per
exchange. Any other body goes to the serializer, by default Java serialization,
so a Map or a POJO is sent as Java-serialized bytes, and a stream or a file
fails: marshal or convert it first" },
+ "producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged,
with the id of the published message in a header" }
+ },
"properties": {
"projectId": { "index": 0, "kind": "path", "displayName": "Project Id",
"group": "common", "label": "common", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The Google Cloud PubSub
Project Id" },
"destinationName": { "index": 1, "kind": "path", "displayName":
"Destination Name", "group": "common", "label": "common", "required": true,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
Destination Name. For the consumer this will be the subscription name, while
for the producer this will be the topic name." },
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json
b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json
index eb7e0ba1ffc5..448eba5c7a09 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json
+++
b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json
@@ -54,6 +54,10 @@
"CamelAwsKinesisFirehoseEncrypted": { "index": 5, "kind": "header",
"displayName": "", "group": "sendBatchRecord", "label": "sendBatchRecord",
"required": false, "javaType": "Boolean", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description":
"Whether the batch operation was encrypted.", "constantName":
"org.apache.camel.component.aws2.firehose.KinesisFirehose2Constants#ENCRYPTED"
},
"CamelAwsKinesisFirehoseDeliveryStreamStatus": { "index": 6, "kind":
"header", "displayName": "", "group": "describeDeliveryStream", "label":
"describeDeliveryStream", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The status of the delivery stream.", "constantName":
"org.apache.camel.component.aws2.firehose.KinesisFirehose2Constants#DELIVERY_STREAM_STATUS"
}
},
+ "bodyTypes": {
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "operation", "value": "sendBatchRecord", "carriers":
[ "list", "pojo" ] }, { "option": "operation", "value": "createDeliveryStream",
"carriers": [ "pojo" ], "javaType":
"software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamRequest" },
{ "option": "operation", "value": "updateDestination", "carriers": [ "pojo" ],
"javaType": "software.amazon.awssdk.services.firehose.model.UpdateD [...]
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "sendBatchRecord", "carriers": [ "pojo" ], "javaType":
"software.amazon.awssdk.services.firehose.model.PutRecordBatchResponse" }, {
"option": "operation", "value": "createDeliveryStream", "carriers": [ "pojo" ],
"javaType":
"software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamResponse"
}, { "option": "operation", "value": "deleteDeliveryStream", "carriers": [
"pojo" ], "javaTy [...]
+ },
"properties": {
"streamName": { "index": 0, "kind": "path", "displayName": "Stream Name",
"group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration",
"configurationField": "configuration", "description": "Name of the stream" },
"cborEnabled": { "index": 1, "kind": "parameter", "displayName": "Cbor
Enabled", "group": "common", "label": "common", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration",
"configurationField": "configuration", "description": "This option will set the
CBOR_ENABLED property during the execution" },
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
index 40e1eb9595bb..7bc42d67463f 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
+++
b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
@@ -72,6 +72,11 @@
"CamelAwsKinesisFailedRecordCount": { "index": 7, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The number of records that failed in a
batch put operation.", "constantName":
"org.apache.camel.component.aws2.kinesis.Kinesis2Constants#FAILED_RECORD_COUNT"
},
"CamelAwsKinesisRecordCount": { "index": 8, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The total number of records in a batch
put operation.", "constantName":
"org.apache.camel.component.aws2.kinesis.Kinesis2Constants#RECORD_COUNT" }
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"byOption": [ { "option": "useKclConsumers", "value": "true", "carriers": [
"pojo" ], "javaType": "java.nio.ByteBuffer" } ], "description": "The data of
the record as a byte array, one message per record, with its partition key,
sequence number and shard id in headers. With useKclConsumers=true a
java.nio.ByteBuffer: convert it to a String or bytes for the steps that read
text" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "list" ],
"description": "The data of the record: text, bytes, a stream or a file. A List
or another Iterable is sent as a batch, one record per element, and each
element must be text, bytes, a ByteBuffer or a stream. A Map or a POJO fails:
marshal it first" },
+ "producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged,
with the sequence number and the shard id of the record in headers, or the
number of records of a batch" }
+ },
"properties": {
"streamName": { "index": 0, "kind": "path", "displayName": "Stream Name",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Name of the stream" },
"cborEnabled": { "index": 1, "kind": "parameter", "displayName": "Cbor
Enabled", "group": "common", "label": "common", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "This option will set the
CBOR_ENABLED property during the execution" },
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
index 35e73ae69161..9cb690c5b8c2 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
@@ -23,6 +23,9 @@ import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import
org.apache.camel.component.aws2.firehose.client.KinesisFirehoseClientFactory;
+import org.apache.camel.spi.BodyType;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
import org.apache.camel.spi.EndpointServiceLocation;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
@@ -40,6 +43,45 @@ import static
software.amazon.awssdk.core.SdkSystemSetting.CBOR_ENABLED;
Category.CLOUD,
Category.MESSAGING },
headersClass = KinesisFirehose2Constants.class)
+@BodyTypes(producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE },
+ byOption = {
+ @BodyType.When(option = "operation",
value = "sendBatchRecord",
+ carriers = {
Carrier.LIST, Carrier.POJO }),
+ @BodyType.When(option = "operation",
value = "createDeliveryStream",
+ carriers =
Carrier.POJO,
+ javaType =
"software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamRequest"),
+ @BodyType.When(option = "operation",
value = "updateDestination",
+ carriers =
Carrier.POJO,
+ javaType =
"software.amazon.awssdk.services.firehose.model.UpdateDestinationRequest"),
+ @BodyType.When(option = "operation",
value = "deleteDeliveryStream",
+ carriers =
Carrier.ANY),
+ @BodyType.When(option = "operation",
value = "describeDeliveryStream",
+ carriers =
Carrier.ANY) },
+ description = "The data of the record, when
no operation is set: text, bytes, a stream or a"
+ + " file. A Map, a List or a
POJO fails, and a number is sent as its binary"
+ + " value: marshal it or
convert it to text first. sendBatchRecord takes a List"
+ + " of the Record of the
Firehose SDK or a PutRecordBatchRequest, and the other"
+ + " operations their request
of the AWS SDK, or for deleteDeliveryStream and"
+ + " describeDeliveryStream the
CamelAwsKinesisFirehoseDeliveryStreamName header"),
+ producerOut = @BodyType(value = Carrier.UNCHANGED,
+ byOption = {
+ @BodyType.When(option =
"operation", value = "sendBatchRecord",
+ carriers =
Carrier.POJO,
+ javaType =
"software.amazon.awssdk.services.firehose.model.PutRecordBatchResponse"),
+ @BodyType.When(option =
"operation", value = "createDeliveryStream",
+ carriers =
Carrier.POJO,
+ javaType =
"software.amazon.awssdk.services.firehose.model.CreateDeliveryStreamResponse"),
+ @BodyType.When(option =
"operation", value = "deleteDeliveryStream",
+ carriers =
Carrier.POJO,
+ javaType =
"software.amazon.awssdk.services.firehose.model.DeleteDeliveryStreamResponse"),
+ @BodyType.When(option =
"operation", value = "describeDeliveryStream",
+ carriers =
Carrier.POJO,
+ javaType =
"software.amazon.awssdk.services.firehose.model.DescribeDeliveryStreamResponse"),
+ @BodyType.When(option =
"operation", value = "updateDestination",
+ carriers =
Carrier.POJO,
+ javaType =
"software.amazon.awssdk.services.firehose.model.UpdateDestinationResponse") },
+ description = "Unchanged after a send, with
the id of the record in a header; the response"
+ + " of the AWS SDK for the
operations"))
public class KinesisFirehose2Endpoint extends DefaultEndpoint implements
EndpointServiceLocation {
@UriParam
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
index 7bef9ecfb57e..85746c41217b 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
@@ -23,6 +23,9 @@ import org.apache.camel.Category;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
+import org.apache.camel.spi.BodyType;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
import org.apache.camel.spi.EndpointServiceLocation;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
@@ -39,6 +42,21 @@ import static
software.amazon.awssdk.core.SdkSystemSetting.CBOR_ENABLED;
*/
@UriEndpoint(firstVersion = "3.2.0", scheme = "aws2-kinesis", title = "AWS
Kinesis", syntax = "aws2-kinesis:streamName",
category = { Category.CLOUD, Category.MESSAGING }, headersClass =
Kinesis2Constants.class)
+@BodyTypes(consumerOut = @BodyType(value = Carrier.BYTES, javaType = "byte[]",
+ byOption = @BodyType.When(option =
"useKclConsumers", value = "true",
+ carriers =
Carrier.POJO, javaType = "java.nio.ByteBuffer"),
+ description = "The data of the record as a
byte array, one message per record, with its"
+ + " partition key, sequence
number and shard id in headers. With"
+ + " useKclConsumers=true a
java.nio.ByteBuffer: convert it to a String or"
+ + " bytes for the steps that
read text"),
+ producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE, Carrier.LIST },
+ description = "The data of the record: text,
bytes, a stream or a file. A List or another"
+ + " Iterable is sent as a
batch, one record per element, and each element"
+ + " must be text, bytes, a
ByteBuffer or a stream. A Map or a POJO fails:"
+ + " marshal it first"),
+ producerOut = @BodyType(value = Carrier.UNCHANGED,
+ description = "Unchanged, with the sequence
number and the shard id of the record in"
+ + " headers, or the number of
records of a batch"))
public class Kinesis2Endpoint extends ScheduledPollEndpoint implements
EndpointServiceLocation {
@UriParam
diff --git
a/components/camel-aws/camel-aws2-sns/src/generated/resources/META-INF/org/apache/camel/component/aws2/sns/aws2-sns.json
b/components/camel-aws/camel-aws2-sns/src/generated/resources/META-INF/org/apache/camel/component/aws2/sns/aws2-sns.json
index 7329d7204a10..a95e468d1634 100644
---
a/components/camel-aws/camel-aws2-sns/src/generated/resources/META-INF/org/apache/camel/component/aws2/sns/aws2-sns.json
+++
b/components/camel-aws/camel-aws2-sns/src/generated/resources/META-INF/org/apache/camel/component/aws2/sns/aws2-sns.json
@@ -64,6 +64,10 @@
"CamelAwsSnsFailedMessageCount": { "index": 4, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The number of failed messages in a
batch publish operation.", "constantName":
"org.apache.camel.component.aws2.sns.Sns2Constants#FAILED_MESSAGE_COUNT" },
"CamelAwsSnsSuccessfulMessageCount": { "index": 5, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The number of successful messages in a
batch publish operation.", "constantName":
"org.apache.camel.component.aws2.sns.Sns2Constants#SUCCESSFUL_MESSAGE_COUNT" }
},
+ "bodyTypes": {
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "batchEnabled", "value": "true", "carriers": [ "list"
] } ], "description": "The message to publish, as text: bytes, a stream or a
file are read as text, and a Map or a POJO is sent as its toString(): marshal
it first. With messageStructure=json, a JSON object with a message per
protocol; with batchEnabled=true, a List of the PublishBatchRequestEntry of the
AWS SDK" },
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"batchEnabled", "value": "true", "carriers": [ "pojo" ], "javaType":
"software.amazon.awssdk.services.sns.model.PublishBatchResponse" } ],
"description": "Unchanged after a publish, with the id of the message in a
header. With batchEnabled=true, the PublishBatchResponse of the AWS SDK" }
+ },
"properties": {
"topicNameOrArn": { "index": 0, "kind": "path", "displayName": "Topic Name
Or Arn", "group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "Topic name or ARN" },
"autoCreateTopic": { "index": 1, "kind": "parameter", "displayName": "Auto
Create Topic", "group": "producer", "label": "", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.aws2.sns.Sns2Configuration", "configurationField":
"configuration", "description": "Setting the auto-creation of the topic" },
diff --git
a/components/camel-aws/camel-aws2-sns/src/main/java/org/apache/camel/component/aws2/sns/Sns2Endpoint.java
b/components/camel-aws/camel-aws2-sns/src/main/java/org/apache/camel/component/aws2/sns/Sns2Endpoint.java
index e5cdb788e7d5..ac9f04e97c37 100644
---
a/components/camel-aws/camel-aws2-sns/src/main/java/org/apache/camel/component/aws2/sns/Sns2Endpoint.java
+++
b/components/camel-aws/camel-aws2-sns/src/main/java/org/apache/camel/component/aws2/sns/Sns2Endpoint.java
@@ -28,6 +28,7 @@ import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.component.aws2.sns.client.Sns2ClientFactory;
import org.apache.camel.spi.*;
+import org.apache.camel.spi.BodyType.Carrier;
import org.apache.camel.support.DefaultEndpoint;
import org.apache.camel.support.ResourceHelper;
import org.apache.camel.util.ObjectHelper;
@@ -51,6 +52,17 @@ import software.amazon.awssdk.services.sns.model.Topic;
@UriEndpoint(firstVersion = "3.1.0", scheme = "aws2-sns", title = "AWS Simple
Notification System (SNS)",
syntax = "aws2-sns:topicNameOrArn", producerOnly = true,
category = { Category.CLOUD, Category.MESSAGING, Category.MOBILE
}, headersClass = Sns2Constants.class)
+@BodyTypes(producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE },
+ byOption = @BodyType.When(option =
"batchEnabled", value = "true", carriers = Carrier.LIST),
+ description = "The message to publish, as
text: bytes, a stream or a file are read as text,"
+ + " and a Map or a POJO is
sent as its toString(): marshal it first. With"
+ + " messageStructure=json, a
JSON object with a message per protocol; with"
+ + " batchEnabled=true, a List
of the PublishBatchRequestEntry of the AWS SDK"),
+ producerOut = @BodyType(value = Carrier.UNCHANGED,
+ byOption = @BodyType.When(option =
"batchEnabled", value = "true", carriers = Carrier.POJO,
+ javaType =
"software.amazon.awssdk.services.sns.model.PublishBatchResponse"),
+ description = "Unchanged after a publish,
with the id of the message in a header. With"
+ + " batchEnabled=true, the
PublishBatchResponse of the AWS SDK"))
public class Sns2Endpoint extends DefaultEndpoint implements
HeaderFilterStrategyAware, EndpointServiceLocation {
private static final Logger LOG =
LoggerFactory.getLogger(Sns2Endpoint.class);
diff --git
a/components/camel-aws/camel-aws2-sqs/src/generated/resources/META-INF/org/apache/camel/component/aws2/sqs/aws2-sqs.json
b/components/camel-aws/camel-aws2-sqs/src/generated/resources/META-INF/org/apache/camel/component/aws2/sqs/aws2-sqs.json
index 8ff172f1e1f3..3f8288b43e5b 100644
---
a/components/camel-aws/camel-aws2-sqs/src/generated/resources/META-INF/org/apache/camel/component/aws2/sqs/aws2-sqs.json
+++
b/components/camel-aws/camel-aws2-sqs/src/generated/resources/META-INF/org/apache/camel/component/aws2/sqs/aws2-sqs.json
@@ -95,7 +95,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "text" ], "javaType": "java.lang.String",
"description": "The body of the SQS message" },
- "producerIn": { "carriers": [ "text" ], "byOption": [ { "option":
"operation", "value": "sendBatchMessage", "carriers": [ "list", "text" ] } ],
"description": "The body of the message to send, as text, so a Map or a POJO is
sent as its toString(): marshal it first. sendBatchMessage takes a List of
text, or text that batchSeparator splits" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "operation", "value": "sendBatchMessage", "carriers":
[ "list", "text", "pojo" ] } ], "description": "The body of the message to
send, as text: bytes, a stream or a file are read as text, and a Map or a POJO
is sent as its toString(): marshal it first. sendBatchMessage takes a List of
text, text that batchSeparator splits, or a SendMessageBatchRequest" },
"producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "sendBatchMessage", "carriers": [ "pojo" ], "javaType":
"software.amazon.awssdk.services.sqs.model.SendMessageBatchResponse" }, {
"option": "operation", "value": "deleteMessage", "carriers": [ "pojo" ],
"javaType": "software.amazon.awssdk.services.sqs.model.DeleteMessageResponse"
}, { "option": "operation", "value": "listQueues", "carriers": [ "pojo" ],
"javaType": "software.amazon.awssdk.se [...]
},
"properties": {
diff --git
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
index 961073d19076..680fc3ecae45 100644
---
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
+++
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
@@ -63,12 +63,13 @@ import
software.amazon.awssdk.services.sqs.model.SqsException;
headersClass = Sqs2Constants.class)
@BodyTypes(consumerOut = @BodyType(value = Carrier.TEXT, javaType =
"java.lang.String",
description = "The body of the SQS
message"),
- producerIn = @BodyType(value = Carrier.TEXT,
+ producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE },
byOption = @BodyType.When(option =
"operation", value = "sendBatchMessage",
- carriers = {
Carrier.LIST, Carrier.TEXT }),
- description = "The body of the message to
send, as text, so a Map or a POJO is sent as"
- + " its toString(): marshal it
first. sendBatchMessage takes a List of"
- + " text, or text that
batchSeparator splits"),
+ carriers = {
Carrier.LIST, Carrier.TEXT, Carrier.POJO }),
+ description = "The body of the message to
send, as text: bytes, a stream or a file are read"
+ + " as text, and a Map or a
POJO is sent as its toString(): marshal it first."
+ + " sendBatchMessage takes a
List of text, text that batchSeparator splits, or"
+ + " a
SendMessageBatchRequest"),
producerOut = @BodyType(value = Carrier.UNCHANGED,
byOption = {
@BodyType.When(option =
"operation", value = "sendBatchMessage",
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
b/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
index 77145836e81e..a81217609b1a 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
+++
b/components/camel-azure/camel-azure-eventhubs/src/generated/resources/META-INF/org/apache/camel/component/azure/eventhubs/azure-eventhubs.json
@@ -60,6 +60,11 @@
"CamelMessageTimestamp": { "index": 6, "kind": "header", "displayName":
"", "group": "consumer", "label": "consumer", "required": false, "javaType":
"long", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The timestamp of the message", "constantName":
"org.apache.camel.component.azure.eventhubs.EventHubsConstants#MESSAGE_TIMESTAMP"
},
"CamelAzureEventHubsCheckpointUpdatedBy": { "index": 7, "kind": "header",
"displayName": "", "group": "consumer", "label": "consumer", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "It sets the reason for the checkpoint
to have been updated. This is only present on a received EventData.",
"constantName":
"org.apache.camel.component.azure.eventhubs.EventHubsConstants#CHECKPOINT_UPDATED_BY"
}
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"description": "The body of the event, one message per event" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "list" ],
"description": "What converts to a byte array: text, bytes, a stream or a file,
sent as one event; a Map, a number or a POJO fails: marshal it first. A List is
sent as one event per element, in one batch, and an element that is an Exchange
or a Message is sent with its headers" },
+ "producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged: a
send gives no result, and no header is set" }
+ },
"properties": {
"namespace": { "index": 0, "kind": "path", "displayName": "Namespace",
"group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.azure.eventhubs.EventHubsConfiguration",
"configurationField": "configuration", "description": "EventHubs namespace
created in Azure Portal." },
"eventHubName": { "index": 1, "kind": "path", "displayName": "Event Hub
Name", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.azure.eventhubs.EventHubsConfiguration",
"configurationField": "configuration", "description": "EventHubs name under a
specific namespace." },
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpoint.java
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpoint.java
index 623cd418df6a..dbd6c5678252 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpoint.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsEndpoint.java
@@ -23,6 +23,9 @@ import org.apache.camel.Component;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
+import org.apache.camel.spi.BodyType;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
import org.apache.camel.spi.EndpointServiceLocation;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
@@ -36,6 +39,15 @@ import org.apache.camel.util.ObjectHelper;
syntax = "azure-eventhubs:namespace/eventHubName", category = {
Category.CLOUD, Category.MESSAGING },
headersClass = EventHubsConstants.class)
+@BodyTypes(consumerOut = @BodyType(value = Carrier.BYTES, javaType = "byte[]",
+ description = "The body of the event, one
message per event"),
+ producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE, Carrier.LIST },
+ description = "What converts to a byte
array: text, bytes, a stream or a file, sent as one"
+ + " event; a Map, a number or
a POJO fails: marshal it first. A List is sent"
+ + " as one event per element,
in one batch, and an element that is an Exchange"
+ + " or a Message is sent with
its headers"),
+ producerOut = @BodyType(value = Carrier.UNCHANGED,
+ description = "Unchanged: a send gives no
result, and no header is set"))
public class EventHubsEndpoint extends DefaultEndpoint implements
EndpointServiceLocation {
@UriParam
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 80ca2d410803..9fd52d36a0e0 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
@@ -82,6 +82,11 @@
"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"
}
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "pojo" ], "javaType":
"com.azure.core.util.BinaryData", "description": "The payload of the Service
Bus message as an Azure BinaryData, which converts to a String but not to bytes
or a stream: convert it to a String before an unmarshal, a file or another
producer" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "list" ],
"description": "Sent as text by default: bytes, a stream or a file are read as
text, so binary content needs binary=true. A BinaryData, as the consumer gives
it, is sent as it is, and a Map or a POJO is sent as its toString(): marshal it
first. A List is sent as one message per element" },
+ "producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged,
also after scheduleMessages" }
+ },
"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 [...]
"amqpRetryOptions": { "index": 1, "kind": "parameter", "displayName":
"Amqp Retry Options", "group": "common", "label": "common", "required": false,
"type": "object", "javaType": "com.azure.core.amqp.AmqpRetryOptions",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.azure.servicebus.ServiceBusConfiguration",
"configurationField": "configuration", "description": "Sets the retry options
for Service Bus clients. If not specifie [...]
diff --git
a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpoint.java
b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpoint.java
index 454f081ceba1..bf5a8f8d08a7 100644
---
a/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpoint.java
+++
b/components/camel-azure/camel-azure-servicebus/src/main/java/org/apache/camel/component/azure/servicebus/ServiceBusEndpoint.java
@@ -24,6 +24,9 @@ import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import
org.apache.camel.component.azure.servicebus.client.ServiceBusClientFactory;
+import org.apache.camel.spi.BodyType;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
import org.apache.camel.spi.EndpointServiceLocation;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
@@ -37,6 +40,17 @@ import org.apache.camel.util.ObjectHelper;
syntax = "azure-servicebus:topicOrQueueName", category = {
Category.CLOUD, Category.MESSAGING },
headersClass = ServiceBusConstants.class)
+@BodyTypes(consumerOut = @BodyType(value = Carrier.POJO, javaType =
"com.azure.core.util.BinaryData",
+ description = "The payload of the Service
Bus message as an Azure BinaryData, which"
+ + " converts to a String but
not to bytes or a stream: convert it to a"
+ + " String before an
unmarshal, a file or another producer"),
+ producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE, Carrier.LIST },
+ description = "Sent as text by default:
bytes, a stream or a file are read as text, so"
+ + " binary content needs
binary=true. A BinaryData, as the consumer gives it, is"
+ + " sent as it is, and a Map
or a POJO is sent as its toString(): marshal it"
+ + " first. A List is sent as
one message per element"),
+ producerOut = @BodyType(value = Carrier.UNCHANGED,
+ description = "Unchanged, also after
scheduleMessages"))
public class ServiceBusEndpoint extends DefaultEndpoint implements
EndpointServiceLocation {
@UriParam
diff --git
a/components/camel-azure/camel-azure-storage-queue/src/generated/resources/META-INF/org/apache/camel/component/azure/storage/queue/azure-storage-queue.json
b/components/camel-azure/camel-azure-storage-queue/src/generated/resources/META-INF/org/apache/camel/component/azure/storage/queue/azure-storage-queue.json
index dbae457091d6..6f772e736274 100644
---
a/components/camel-azure/camel-azure-storage-queue/src/generated/resources/META-INF/org/apache/camel/component/azure/storage/queue/azure-storage-queue.json
+++
b/components/camel-azure/camel-azure-storage-queue/src/generated/resources/META-INF/org/apache/camel/component/azure/storage/queue/azure-storage-queue.json
@@ -61,6 +61,11 @@
"CamelAzureStorageQueueTimeToLive": { "index": 14, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Duration", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "(sendMessage) How long the
message will stay alive in the queue. If unset the value will default to 7
days, if -1 is passed the message will not expire. The time to live must be -1
or any positive number.", "constantNa [...]
"CamelAzureStorageQueueCreateQueue": { "index": 15, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "(sendMessage) When is set to true, the
queue will be automatically created when sending messages to the queue.",
"constantName":
"org.apache.camel.component.azure.storage.queue.QueueConstants#CREATE_QUEUE" }
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "stream", "null" ], "description": "The
text of the queue message as a stream, with its id and pop receipt in headers;
no body for a message without text" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"byOption": [ { "option": "operation", "value": "updateMessage", "carriers": [
"text", "bytes", "stream", "file", "null" ] }, { "option": "operation",
"value": "listQueues", "carriers": [ "any" ] }, { "option": "operation",
"value": "createQueue", "carriers": [ "any" ] }, { "option": "operation",
"value": "deleteQueue", "carriers": [ "any" ] }, { "option": "operation",
"value": "clearQueue", "carriers": [ "any" ] }, { [...]
+ "producerOut": { "carriers": [ "value" ], "javaType": "java.lang.Boolean",
"byOption": [ { "option": "operation", "value": "listQueues", "carriers": [
"list" ], "javaType":
"java.util.List<com.azure.storage.queue.models.QueueItem>" }, { "option":
"operation", "value": "receiveMessages", "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.storage.queue.models.QueueMessageItem>" }, {
"option": "operation", "value": "peekMessages", "carriers": [ "list" ],
"javaType": "java.uti [...]
+ },
"properties": {
"accountName": { "index": 0, "kind": "path", "displayName": "Account
Name", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.azure.storage.queue.QueueConfiguration",
"configurationField": "configuration", "description": "Azure account name to be
used for authentication with azure queue services" },
"queueName": { "index": 1, "kind": "path", "displayName": "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.storage.queue.QueueConfiguration",
"configurationField": "configuration", "description": "The queue resource name"
},
diff --git
a/components/camel-azure/camel-azure-storage-queue/src/main/java/org/apache/camel/component/azure/storage/queue/QueueEndpoint.java
b/components/camel-azure/camel-azure-storage-queue/src/main/java/org/apache/camel/component/azure/storage/queue/QueueEndpoint.java
index 1230622682a5..f54464a4537c 100644
---
a/components/camel-azure/camel-azure-storage-queue/src/main/java/org/apache/camel/component/azure/storage/queue/QueueEndpoint.java
+++
b/components/camel-azure/camel-azure-storage-queue/src/main/java/org/apache/camel/component/azure/storage/queue/QueueEndpoint.java
@@ -25,6 +25,9 @@ import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import
org.apache.camel.component.azure.storage.queue.client.QueueClientFactory;
+import org.apache.camel.spi.BodyType;
+import org.apache.camel.spi.BodyType.Carrier;
+import org.apache.camel.spi.BodyTypes;
import org.apache.camel.spi.EndpointServiceLocation;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
@@ -37,6 +40,42 @@ import org.apache.camel.util.ObjectHelper;
@UriEndpoint(firstVersion = "3.3.0", scheme = "azure-storage-queue", title =
"Azure Storage Queue Service",
syntax = "azure-storage-queue:accountName/queueName", category =
{ Category.CLOUD, Category.MESSAGING },
headersClass = QueueConstants.class)
+@BodyTypes(consumerOut = @BodyType(value = { Carrier.STREAM, Carrier.NULL },
+ description = "The text of the queue
message as a stream, with its id and pop receipt"
+ + " in headers; no body for a
message without text"),
+ producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE },
+ byOption = {
+ @BodyType.When(option = "operation",
value = "updateMessage",
+ carriers = {
+ Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
+ Carrier.NULL
}),
+ @BodyType.When(option = "operation",
value = "listQueues", carriers = Carrier.ANY),
+ @BodyType.When(option = "operation",
value = "createQueue", carriers = Carrier.ANY),
+ @BodyType.When(option = "operation",
value = "deleteQueue", carriers = Carrier.ANY),
+ @BodyType.When(option = "operation",
value = "clearQueue", carriers = Carrier.ANY),
+ @BodyType.When(option = "operation",
value = "deleteMessage", carriers = Carrier.ANY),
+ @BodyType.When(option = "operation",
value = "receiveMessages",
+ carriers =
Carrier.ANY),
+ @BodyType.When(option = "operation",
value = "peekMessages",
+ carriers =
Carrier.ANY) },
+ description = "The text of the message for
sendMessage, the default, and updateMessage: bytes,"
+ + " a stream or a file are
read as text, and a Map or a POJO is sent as its"
+ + " toString(): marshal it
first. sendMessage fails without a body, and"
+ + " updateMessage then changes
only the visibility. The other operations do not"
+ + " use the body"),
+ producerOut = @BodyType(value = Carrier.VALUE, javaType =
"java.lang.Boolean",
+ byOption = {
+ @BodyType.When(option =
"operation", value = "listQueues", carriers = Carrier.LIST,
+ javaType =
"java.util.List<com.azure.storage.queue.models.QueueItem>"),
+ @BodyType.When(option =
"operation", value = "receiveMessages",
+ carriers =
Carrier.LIST,
+ javaType =
"java.util.List<com.azure.storage.queue.models.QueueMessageItem>"),
+ @BodyType.When(option =
"operation", value = "peekMessages", carriers = Carrier.LIST,
+ javaType =
"java.util.List<com.azure.storage.queue.models.PeekedMessageItem>") },
+ description = "true after sendMessage, the
default, and the operations that create, clear,"
+ + " update or delete, with
their results in headers, such as the id and the pop"
+ + " receipt of the message
sent. A List of the items of the SDK for listQueues,"
+ + " receiveMessages and
peekMessages"))
public class QueueEndpoint extends ScheduledPollEndpoint implements
EndpointServiceLocation {
private QueueServiceClient queueServiceClient;
diff --git
a/components/camel-google/camel-google-pubsub/src/generated/resources/META-INF/org/apache/camel/component/google/pubsub/google-pubsub.json
b/components/camel-google/camel-google-pubsub/src/generated/resources/META-INF/org/apache/camel/component/google/pubsub/google-pubsub.json
index 2da993e22ea3..765633f3cd31 100644
---
a/components/camel-google/camel-google-pubsub/src/generated/resources/META-INF/org/apache/camel/component/google/pubsub/google-pubsub.json
+++
b/components/camel-google/camel-google-pubsub/src/generated/resources/META-INF/org/apache/camel/component/google/pubsub/google-pubsub.json
@@ -45,6 +45,11 @@
"CamelGooglePubsubAcknowledge": { "index": 5, "kind": "header",
"displayName": "", "group": "consumer", "label": "consumer", "required": false,
"javaType":
"org.apache.camel.component.google.pubsub.consumer.GooglePubsubAcknowledge",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Can be used to manually acknowledge or
negative-acknowledge a message when ackMode=NONE.", "constantName":
"org.apache.camel.component.google.pubsub.GooglePub [...]
"CamelGooglePubsubDeliveryAttempt": { "index": 6, "kind": "header",
"displayName": "", "group": "consumer", "label": "consumer", "required": false,
"javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The delivery attempt counter received
from PubSub. This is the approximate number of times the message has been
delivered. This will be 1 for the first delivery. This feature requires a
dead-letter policy to be configure [...]
},
+ "bodyTypes": {
+ "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"description": "The data of the Pub\/Sub message as a byte array, with its
attributes in headers: convert or unmarshal it" },
+ "producerIn": { "carriers": [ "text", "bytes" ], "dependsOn": [
"serializer" ], "description": "Text is sent as UTF-8 and bytes as they are; a
List of the exchanges of a grouped aggregation is sent as one message per
exchange. Any other body goes to the serializer, by default Java serialization,
so a Map or a POJO is sent as Java-serialized bytes, and a stream or a file
fails: marshal or convert it first" },
+ "producerOut": { "carriers": [ "unchanged" ], "description": "Unchanged,
with the id of the published message in a header" }
+ },
"properties": {
"projectId": { "index": 0, "kind": "path", "displayName": "Project Id",
"group": "common", "label": "common", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The Google Cloud PubSub
Project Id" },
"destinationName": { "index": 1, "kind": "path", "displayName":
"Destination Name", "group": "common", "label": "common", "required": true,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
Destination Name. For the consumer this will be the subscription name, while
for the producer this will be the topic name." },
diff --git
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubEndpoint.java
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubEndpoint.java
index cbdacf9ff85e..9b111e27f3f8 100644
---
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubEndpoint.java
+++
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubEndpoint.java
@@ -30,6 +30,7 @@ import
org.apache.camel.component.google.common.GoogleCommonConfiguration;
import
org.apache.camel.component.google.pubsub.serializer.DefaultGooglePubsubSerializer;
import
org.apache.camel.component.google.pubsub.serializer.GooglePubsubSerializer;
import org.apache.camel.spi.*;
+import org.apache.camel.spi.BodyType.Carrier;
import org.apache.camel.support.DefaultEndpoint;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
@@ -43,6 +44,17 @@ import org.slf4j.LoggerFactory;
@UriEndpoint(firstVersion = "2.19.0", scheme = "google-pubsub", title =
"Google Pubsub",
syntax = "google-pubsub:projectId:destinationName", category = {
Category.CLOUD, Category.MESSAGING },
headersClass = GooglePubsubConstants.class)
+@BodyTypes(consumerOut = @BodyType(value = Carrier.BYTES, javaType = "byte[]",
+ description = "The data of the Pub/Sub
message as a byte array, with its attributes in"
+ + " headers: convert or
unmarshal it"),
+ producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES },
dependsOn = "serializer",
+ description = "Text is sent as UTF-8 and
bytes as they are; a List of the exchanges of a"
+ + " grouped aggregation is
sent as one message per exchange. Any other body"
+ + " goes to the serializer, by
default Java serialization, so a Map or a POJO"
+ + " is sent as Java-serialized
bytes, and a stream or a file fails: marshal or"
+ + " convert it first"),
+ producerOut = @BodyType(value = Carrier.UNCHANGED,
+ description = "Unchanged, with the id of
the published message in a header"))
public class GooglePubsubEndpoint extends DefaultEndpoint
implements EndpointServiceLocation, HeaderFilterStrategyAware,
GoogleCommonConfiguration {
diff --git
a/dsl/camel-yaml-dsl/camel-yaml-dsl-validator/src/test/java/org/apache/camel/dsl/yaml/validator/TextBodyFlowTest.java
b/dsl/camel-yaml-dsl/camel-yaml-dsl-validator/src/test/java/org/apache/camel/dsl/yaml/validator/TextBodyFlowTest.java
index 2422508a73f5..27c6c2b9e64a 100644
---
a/dsl/camel-yaml-dsl/camel-yaml-dsl-validator/src/test/java/org/apache/camel/dsl/yaml/validator/TextBodyFlowTest.java
+++
b/dsl/camel-yaml-dsl/camel-yaml-dsl-validator/src/test/java/org/apache/camel/dsl/yaml/validator/TextBodyFlowTest.java
@@ -1134,6 +1134,58 @@ public class TextBodyFlowTest {
.contains("the azure-storage-blob consumer gives it as
a stream or a file: unmarshal it first"));
}
+ @Test
+ public void testTheBodyOfCloudMessaging() {
+ // the event hubs consumer gives bytes and the queue consumer a
stream, so a field read needs an unmarshal;
+ // google-pubsub sends text or bytes, so parsed data must be
marshalled first, while aws2-kinesis sends a List
+ // as a batch of records
+ String yaml = """
+ - route:
+ id: events
+ from:
+ uri: azure-eventhubs:namespace/orders
+ steps:
+ - setHeader:
+ name: id
+ simple: ${body[id]}
+ - route:
+ id: queue
+ from:
+ uri: azure-storage-queue:account/orders
+ steps:
+ - setHeader:
+ name: id
+ simple: ${body[id]}
+ - route:
+ id: publish
+ from:
+ uri: timer:tick
+ steps:
+ - setBody:
+ constant: '{"id": 1}'
+ - unmarshal:
+ json: {}
+ - to: google-pubsub:project:orders
+ - route:
+ id: batch
+ from:
+ uri: timer:tick
+ steps:
+ - setBody:
+ constant: '[1, 2]'
+ - unmarshal:
+ json: {}
+ - to: aws2-kinesis:orders
+ """;
+ assertThat(lines(yaml)).hasSize(3).anySatisfy(m -> assertThat(m)
+ .contains("route events: simple reads a field of the body
(${body[id]})")
+ .contains("the azure-eventhubs consumer gives it as bytes"))
+ .anySatisfy(m -> assertThat(m)
+ .contains("route queue: simple reads a field of the
body (${body[id]})")
+ .contains("the azure-storage-queue consumer gives it
as a stream"))
+ .anySatisfy(m -> assertThat(m).contains("to: google-pubsub
needs the body as text or bytes"));
+ }
+
@Test
public void testOneReportForAStepThatFails() {
// the first write fails at runtime, so the second does not run with
this body
