This is an automated email from the ASF dual-hosted git repository.
oscerd 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 0092223b71ad CAMEL-23419: camel-nats - Fix JetStream consumer pull
subscription mode (#23084)
0092223b71ad is described below
commit 0092223b71ad6a86b752cfafb1b6d369250028ed
Author: Andrea Cosentino <[email protected]>
AuthorDate: Fri May 8 14:27:28 2026 +0200
CAMEL-23419: camel-nats - Fix JetStream consumer pull subscription mode
(#23084)
The JetStream consumer was setting up a pull subscription using the
dispatcher-based subscribe overload but never issued any pull/fetch
requests, so no messages were ever delivered in pull mode (the default).
Switching to push mode via pullSubscription=false was the only workaround.
Fix: use the synchronous pull subscription API and run a fetch loop on the
consumer worker thread. Two new endpoint options control the loop:
- pullBatchSize (default 10): max messages per fetch request
- pullFetchTimeout (default 1000 ms): max wait per fetch request
Also unsubscribe the JetStream subscription on consumer stop.
Adds an integration test covering the pull subscription path.
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../org/apache/camel/catalog/components/nats.json | 70 ++++++++++++----------
.../component/nats/NatsComponentConfigurer.java | 12 ++++
.../component/nats/NatsEndpointConfigurer.java | 12 ++++
.../component/nats/NatsEndpointUriFactory.java | 4 +-
.../org/apache/camel/component/nats/nats.json | 70 ++++++++++++----------
.../camel/component/nats/NatsConfiguration.java | 28 +++++++++
.../apache/camel/component/nats/NatsConsumer.java | 57 ++++++++++++++++--
.../jetstream/NatsJetstreamConsumerPullIT.java | 63 +++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_21.adoc | 12 ++++
.../component/dsl/NatsComponentBuilderFactory.java | 40 +++++++++++++
.../endpoint/dsl/NatsEndpointBuilderFactory.java | 68 +++++++++++++++++++++
11 files changed, 365 insertions(+), 71 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/nats.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/nats.json
index 84e9d2375354..a83b6b4b6373 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/nats.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/nats.json
@@ -49,22 +49,24 @@
"maxMessages": { "index": 22, "kind": "property", "displayName": "Max
Messages", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Stop receiving messages from a topic we are
subscribing to after maxMessages" },
"nackWait": { "index": 23, "kind": "property", "displayName": "Nack Wait",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "long", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "For negative acknowledgements (NAK),
redelivery is delayed by 5 seconds (default). Setting this to 0 or [...]
"poolSize": { "index": 24, "kind": "property", "displayName": "Pool Size",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 10, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Consumer thread pool size (default is 10)" },
- "pullSubscription": { "index": 25, "kind": "property", "displayName":
"Pull Subscription", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Sets the consumer
subscription type for JetStream. Set to true to use a Pull Subscri [...]
- "queueName": { "index": 26, "kind": "property", "displayName": "Queue
Name", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "The Queue name if we are using nats for a
queue configuration" },
- "replyToDisabled": { "index": 27, "kind": "property", "displayName":
"Reply To Disabled", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Can be used to turn off
sending back reply message in the consumer." },
- "consumerConfiguration": { "index": 28, "kind": "property", "displayName":
"Consumer Configuration", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"io.nats.client.api.ConsumerConfiguration", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets a custom ConsumerConfiguration object f
[...]
- "lazyStartProducer": { "index": 29, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Whether the producer should be started lazy (on the first message). By
starting lazy you can use this to allow CamelContext and routes to startup in
situations where a producer may otherwise fai [...]
- "replySubject": { "index": 30, "kind": "property", "displayName": "Reply
Subject", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "the subject to which subscribers should send
response" },
- "requestTimeout": { "index": 31, "kind": "property", "displayName":
"Request Timeout", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 20000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Request timeout in milliseconds" },
- "autowiredEnabled": { "index": 32, "kind": "property", "displayName":
"Autowired Enabled", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether autowiring is enabled. This is used for automatic autowiring options
(the option must be marked as autowired) by looking up in the registry to find
if there is a single instance of matching [...]
- "connection": { "index": 33, "kind": "property", "displayName":
"Connection", "group": "advanced", "label": "advanced", "required": false,
"type": "object", "javaType": "io.nats.client.Connection", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Reference an already instantiated connection
to Nats server" },
- "headerFilterStrategy": { "index": 34, "kind": "property", "displayName":
"Header Filter Strategy", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To use a custom header filter strategy." },
- "jetstreamAsync": { "index": 35, "kind": "property", "displayName":
"Jetstream Async", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets whether to operate JetStream requests
asynchronously." },
- "traceConnection": { "index": 36, "kind": "property", "displayName":
"Trace Connection", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Whether or not
connection trace messages should be printed to standard out for fine g [...]
- "credentialsFilePath": { "index": 37, "kind": "property", "displayName":
"Credentials File Path", "group": "security", "label": "security", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "If we use useCredentialsFile to true we'll
need to set the credentialsFilePath option. It c [...]
- "secure": { "index": 38, "kind": "property", "displayName": "Secure",
"group": "security", "label": "security", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Set secure option indicating TLS is required"
},
- "sslContextParameters": { "index": 39, "kind": "property", "displayName":
"Ssl Context Parameters", "group": "security", "label": "security", "required":
false, "type": "object", "javaType":
"org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To configure security using
SSLContextParameters" },
- "useGlobalSslContextParameters": { "index": 40, "kind": "property",
"displayName": "Use Global Ssl Context Parameters", "group": "security",
"label": "security", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Enable usage of global SSL context
parameters." }
+ "pullBatchSize": { "index": 25, "kind": "property", "displayName": "Pull
Batch Size", "group": "consumer", "label": "consumer", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 10, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Maximum number of messages to fetch per pull
request when using a JetStream Pull Subscription. [...]
+ "pullFetchTimeout": { "index": 26, "kind": "property", "displayName":
"Pull Fetch Timeout", "group": "consumer", "label": "consumer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Maximum time (in milliseconds) to wait for a
batch of messages to be available on the [...]
+ "pullSubscription": { "index": 27, "kind": "property", "displayName":
"Pull Subscription", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Sets the consumer
subscription type for JetStream. Set to true to use a Pull Subscri [...]
+ "queueName": { "index": 28, "kind": "property", "displayName": "Queue
Name", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "The Queue name if we are using nats for a
queue configuration" },
+ "replyToDisabled": { "index": 29, "kind": "property", "displayName":
"Reply To Disabled", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Can be used to turn off
sending back reply message in the consumer." },
+ "consumerConfiguration": { "index": 30, "kind": "property", "displayName":
"Consumer Configuration", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"io.nats.client.api.ConsumerConfiguration", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets a custom ConsumerConfiguration object f
[...]
+ "lazyStartProducer": { "index": 31, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Whether the producer should be started lazy (on the first message). By
starting lazy you can use this to allow CamelContext and routes to startup in
situations where a producer may otherwise fai [...]
+ "replySubject": { "index": 32, "kind": "property", "displayName": "Reply
Subject", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "the subject to which subscribers should send
response" },
+ "requestTimeout": { "index": 33, "kind": "property", "displayName":
"Request Timeout", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 20000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Request timeout in milliseconds" },
+ "autowiredEnabled": { "index": 34, "kind": "property", "displayName":
"Autowired Enabled", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether autowiring is enabled. This is used for automatic autowiring options
(the option must be marked as autowired) by looking up in the registry to find
if there is a single instance of matching [...]
+ "connection": { "index": 35, "kind": "property", "displayName":
"Connection", "group": "advanced", "label": "advanced", "required": false,
"type": "object", "javaType": "io.nats.client.Connection", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Reference an already instantiated connection
to Nats server" },
+ "headerFilterStrategy": { "index": 36, "kind": "property", "displayName":
"Header Filter Strategy", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To use a custom header filter strategy." },
+ "jetstreamAsync": { "index": 37, "kind": "property", "displayName":
"Jetstream Async", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets whether to operate JetStream requests
asynchronously." },
+ "traceConnection": { "index": 38, "kind": "property", "displayName":
"Trace Connection", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Whether or not
connection trace messages should be printed to standard out for fine g [...]
+ "credentialsFilePath": { "index": 39, "kind": "property", "displayName":
"Credentials File Path", "group": "security", "label": "security", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "If we use useCredentialsFile to true we'll
need to set the credentialsFilePath option. It c [...]
+ "secure": { "index": 40, "kind": "property", "displayName": "Secure",
"group": "security", "label": "security", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Set secure option indicating TLS is required"
},
+ "sslContextParameters": { "index": 41, "kind": "property", "displayName":
"Ssl Context Parameters", "group": "security", "label": "security", "required":
false, "type": "object", "javaType":
"org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To configure security using
SSLContextParameters" },
+ "useGlobalSslContextParameters": { "index": 42, "kind": "property",
"displayName": "Use Global Ssl Context Parameters", "group": "security",
"label": "security", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Enable usage of global SSL context
parameters." }
},
"headers": {
"CamelNatsMessageTimestamp": { "index": 0, "kind": "header",
"displayName": "", "group": "common", "label": "", "required": false,
"javaType": "long", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The timestamp of a consumed message.",
"constantName":
"org.apache.camel.component.nats.NatsConstants#NATS_MESSAGE_TIMESTAMP" },
@@ -101,22 +103,24 @@
"maxMessages": { "index": 21, "kind": "parameter", "displayName": "Max
Messages", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Stop receiving messages from a topic we are
subscribing to after maxMessages" },
"nackWait": { "index": 22, "kind": "parameter", "displayName": "Nack
Wait", "group": "consumer", "label": "consumer", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 5000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "For negative acknowledgements (NAK),
redelivery is delayed by 5 seconds (default). Setting this to 0 o [...]
"poolSize": { "index": 23, "kind": "parameter", "displayName": "Pool
Size", "group": "consumer", "label": "consumer", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 10, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Consumer thread pool size (default is 10)" },
- "pullSubscription": { "index": 24, "kind": "parameter", "displayName":
"Pull Subscription", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Sets the consumer
subscription type for JetStream. Set to true to use a Pull Subscr [...]
- "queueName": { "index": 25, "kind": "parameter", "displayName": "Queue
Name", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "The Queue name if we are using nats for a
queue configuration" },
- "replyToDisabled": { "index": 26, "kind": "parameter", "displayName":
"Reply To Disabled", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Can be used to turn off
sending back reply message in the consumer." },
- "bridgeErrorHandler": { "index": 27, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
- "consumerConfiguration": { "index": 28, "kind": "parameter",
"displayName": "Consumer Configuration", "group": "consumer (advanced)",
"label": "consumer,advanced", "required": false, "type": "object", "javaType":
"io.nats.client.api.ConsumerConfiguration", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets a custom ConsumerConfiguration object
[...]
- "exceptionHandler": { "index": 29, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By de [...]
- "exchangePattern": { "index": 30, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
- "replySubject": { "index": 31, "kind": "parameter", "displayName": "Reply
Subject", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "the subject to which subscribers should send
response" },
- "requestTimeout": { "index": 32, "kind": "parameter", "displayName":
"Request Timeout", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 20000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Request timeout in milliseconds" },
- "lazyStartProducer": { "index": 33, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produ [...]
- "connection": { "index": 34, "kind": "parameter", "displayName":
"Connection", "group": "advanced", "label": "advanced", "required": false,
"type": "object", "javaType": "io.nats.client.Connection", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Reference an already instantiated connection
to Nats server" },
- "headerFilterStrategy": { "index": 35, "kind": "parameter", "displayName":
"Header Filter Strategy", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To use a custom header filter strategy." },
- "jetstreamAsync": { "index": 36, "kind": "parameter", "displayName":
"Jetstream Async", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets whether to operate JetStream requests
asynchronously." },
- "traceConnection": { "index": 37, "kind": "parameter", "displayName":
"Trace Connection", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Whether or not
connection trace messages should be printed to standard out for fine [...]
- "credentialsFilePath": { "index": 38, "kind": "parameter", "displayName":
"Credentials File Path", "group": "security", "label": "security", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "If we use useCredentialsFile to true we'll
need to set the credentialsFilePath option. It [...]
- "secure": { "index": 39, "kind": "parameter", "displayName": "Secure",
"group": "security", "label": "security", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Set secure option indicating TLS is required"
},
- "sslContextParameters": { "index": 40, "kind": "parameter", "displayName":
"Ssl Context Parameters", "group": "security", "label": "security", "required":
false, "type": "object", "javaType":
"org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To configure security using
SSLContextParameters" }
+ "pullBatchSize": { "index": 24, "kind": "parameter", "displayName": "Pull
Batch Size", "group": "consumer", "label": "consumer", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 10, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Maximum number of messages to fetch per pull
request when using a JetStream Pull Subscription. [...]
+ "pullFetchTimeout": { "index": 25, "kind": "parameter", "displayName":
"Pull Fetch Timeout", "group": "consumer", "label": "consumer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Maximum time (in milliseconds) to wait for a
batch of messages to be available on the [...]
+ "pullSubscription": { "index": 26, "kind": "parameter", "displayName":
"Pull Subscription", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Sets the consumer
subscription type for JetStream. Set to true to use a Pull Subscr [...]
+ "queueName": { "index": 27, "kind": "parameter", "displayName": "Queue
Name", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "The Queue name if we are using nats for a
queue configuration" },
+ "replyToDisabled": { "index": 28, "kind": "parameter", "displayName":
"Reply To Disabled", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Can be used to turn off
sending back reply message in the consumer." },
+ "bridgeErrorHandler": { "index": 29, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
+ "consumerConfiguration": { "index": 30, "kind": "parameter",
"displayName": "Consumer Configuration", "group": "consumer (advanced)",
"label": "consumer,advanced", "required": false, "type": "object", "javaType":
"io.nats.client.api.ConsumerConfiguration", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets a custom ConsumerConfiguration object
[...]
+ "exceptionHandler": { "index": 31, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By de [...]
+ "exchangePattern": { "index": 32, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
+ "replySubject": { "index": 33, "kind": "parameter", "displayName": "Reply
Subject", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "the subject to which subscribers should send
response" },
+ "requestTimeout": { "index": 34, "kind": "parameter", "displayName":
"Request Timeout", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 20000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Request timeout in milliseconds" },
+ "lazyStartProducer": { "index": 35, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produ [...]
+ "connection": { "index": 36, "kind": "parameter", "displayName":
"Connection", "group": "advanced", "label": "advanced", "required": false,
"type": "object", "javaType": "io.nats.client.Connection", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Reference an already instantiated connection
to Nats server" },
+ "headerFilterStrategy": { "index": 37, "kind": "parameter", "displayName":
"Header Filter Strategy", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To use a custom header filter strategy." },
+ "jetstreamAsync": { "index": 38, "kind": "parameter", "displayName":
"Jetstream Async", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets whether to operate JetStream requests
asynchronously." },
+ "traceConnection": { "index": 39, "kind": "parameter", "displayName":
"Trace Connection", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Whether or not
connection trace messages should be printed to standard out for fine [...]
+ "credentialsFilePath": { "index": 40, "kind": "parameter", "displayName":
"Credentials File Path", "group": "security", "label": "security", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "If we use useCredentialsFile to true we'll
need to set the credentialsFilePath option. It [...]
+ "secure": { "index": 41, "kind": "parameter", "displayName": "Secure",
"group": "security", "label": "security", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Set secure option indicating TLS is required"
},
+ "sslContextParameters": { "index": 42, "kind": "parameter", "displayName":
"Ssl Context Parameters", "group": "security", "label": "security", "required":
false, "type": "object", "javaType":
"org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To configure security using
SSLContextParameters" }
}
}
diff --git
a/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsComponentConfigurer.java
b/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsComponentConfigurer.java
index 96cc901207d3..d0891a119914 100644
---
a/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsComponentConfigurer.java
+++
b/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsComponentConfigurer.java
@@ -81,6 +81,10 @@ public class NatsComponentConfigurer extends
PropertyConfigurerSupport implement
case "pingInterval":
getOrCreateConfiguration(target).setPingInterval(property(camelContext,
int.class, value)); return true;
case "poolsize":
case "poolSize":
getOrCreateConfiguration(target).setPoolSize(property(camelContext, int.class,
value)); return true;
+ case "pullbatchsize":
+ case "pullBatchSize":
getOrCreateConfiguration(target).setPullBatchSize(property(camelContext,
int.class, value)); return true;
+ case "pullfetchtimeout":
+ case "pullFetchTimeout":
getOrCreateConfiguration(target).setPullFetchTimeout(property(camelContext,
long.class, value)); return true;
case "pullsubscription":
case "pullSubscription":
getOrCreateConfiguration(target).setPullSubscription(property(camelContext,
boolean.class, value)); return true;
case "queuename":
@@ -163,6 +167,10 @@ public class NatsComponentConfigurer extends
PropertyConfigurerSupport implement
case "pingInterval": return int.class;
case "poolsize":
case "poolSize": return int.class;
+ case "pullbatchsize":
+ case "pullBatchSize": return int.class;
+ case "pullfetchtimeout":
+ case "pullFetchTimeout": return long.class;
case "pullsubscription":
case "pullSubscription": return boolean.class;
case "queuename":
@@ -246,6 +254,10 @@ public class NatsComponentConfigurer extends
PropertyConfigurerSupport implement
case "pingInterval": return
getOrCreateConfiguration(target).getPingInterval();
case "poolsize":
case "poolSize": return getOrCreateConfiguration(target).getPoolSize();
+ case "pullbatchsize":
+ case "pullBatchSize": return
getOrCreateConfiguration(target).getPullBatchSize();
+ case "pullfetchtimeout":
+ case "pullFetchTimeout": return
getOrCreateConfiguration(target).getPullFetchTimeout();
case "pullsubscription":
case "pullSubscription": return
getOrCreateConfiguration(target).isPullSubscription();
case "queuename":
diff --git
a/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsEndpointConfigurer.java
b/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsEndpointConfigurer.java
index 3436de25552b..923781a8bc8a 100644
---
a/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsEndpointConfigurer.java
+++
b/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsEndpointConfigurer.java
@@ -75,6 +75,10 @@ public class NatsEndpointConfigurer extends
PropertyConfigurerSupport implements
case "pingInterval":
target.getConfiguration().setPingInterval(property(camelContext, int.class,
value)); return true;
case "poolsize":
case "poolSize":
target.getConfiguration().setPoolSize(property(camelContext, int.class,
value)); return true;
+ case "pullbatchsize":
+ case "pullBatchSize":
target.getConfiguration().setPullBatchSize(property(camelContext, int.class,
value)); return true;
+ case "pullfetchtimeout":
+ case "pullFetchTimeout":
target.getConfiguration().setPullFetchTimeout(property(camelContext,
long.class, value)); return true;
case "pullsubscription":
case "pullSubscription":
target.getConfiguration().setPullSubscription(property(camelContext,
boolean.class, value)); return true;
case "queuename":
@@ -156,6 +160,10 @@ public class NatsEndpointConfigurer extends
PropertyConfigurerSupport implements
case "pingInterval": return int.class;
case "poolsize":
case "poolSize": return int.class;
+ case "pullbatchsize":
+ case "pullBatchSize": return int.class;
+ case "pullfetchtimeout":
+ case "pullFetchTimeout": return long.class;
case "pullsubscription":
case "pullSubscription": return boolean.class;
case "queuename":
@@ -238,6 +246,10 @@ public class NatsEndpointConfigurer extends
PropertyConfigurerSupport implements
case "pingInterval": return
target.getConfiguration().getPingInterval();
case "poolsize":
case "poolSize": return target.getConfiguration().getPoolSize();
+ case "pullbatchsize":
+ case "pullBatchSize": return
target.getConfiguration().getPullBatchSize();
+ case "pullfetchtimeout":
+ case "pullFetchTimeout": return
target.getConfiguration().getPullFetchTimeout();
case "pullsubscription":
case "pullSubscription": return
target.getConfiguration().isPullSubscription();
case "queuename":
diff --git
a/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsEndpointUriFactory.java
b/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsEndpointUriFactory.java
index 9252e5d790b3..542551087298 100644
---
a/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsEndpointUriFactory.java
+++
b/components/camel-nats/src/generated/java/org/apache/camel/component/nats/NatsEndpointUriFactory.java
@@ -23,7 +23,7 @@ public class NatsEndpointUriFactory extends
org.apache.camel.support.component.E
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Map<String, String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(41);
+ Set<String> props = new HashSet<>(43);
props.add("ackPolicy");
props.add("ackWait");
props.add("bridgeErrorHandler");
@@ -51,6 +51,8 @@ public class NatsEndpointUriFactory extends
org.apache.camel.support.component.E
props.add("pedantic");
props.add("pingInterval");
props.add("poolSize");
+ props.add("pullBatchSize");
+ props.add("pullFetchTimeout");
props.add("pullSubscription");
props.add("queueName");
props.add("reconnect");
diff --git
a/components/camel-nats/src/generated/resources/META-INF/org/apache/camel/component/nats/nats.json
b/components/camel-nats/src/generated/resources/META-INF/org/apache/camel/component/nats/nats.json
index 84e9d2375354..a83b6b4b6373 100644
---
a/components/camel-nats/src/generated/resources/META-INF/org/apache/camel/component/nats/nats.json
+++
b/components/camel-nats/src/generated/resources/META-INF/org/apache/camel/component/nats/nats.json
@@ -49,22 +49,24 @@
"maxMessages": { "index": 22, "kind": "property", "displayName": "Max
Messages", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Stop receiving messages from a topic we are
subscribing to after maxMessages" },
"nackWait": { "index": 23, "kind": "property", "displayName": "Nack Wait",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "long", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 5000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "For negative acknowledgements (NAK),
redelivery is delayed by 5 seconds (default). Setting this to 0 or [...]
"poolSize": { "index": 24, "kind": "property", "displayName": "Pool Size",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": 10, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Consumer thread pool size (default is 10)" },
- "pullSubscription": { "index": 25, "kind": "property", "displayName":
"Pull Subscription", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Sets the consumer
subscription type for JetStream. Set to true to use a Pull Subscri [...]
- "queueName": { "index": 26, "kind": "property", "displayName": "Queue
Name", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "The Queue name if we are using nats for a
queue configuration" },
- "replyToDisabled": { "index": 27, "kind": "property", "displayName":
"Reply To Disabled", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Can be used to turn off
sending back reply message in the consumer." },
- "consumerConfiguration": { "index": 28, "kind": "property", "displayName":
"Consumer Configuration", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"io.nats.client.api.ConsumerConfiguration", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets a custom ConsumerConfiguration object f
[...]
- "lazyStartProducer": { "index": 29, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Whether the producer should be started lazy (on the first message). By
starting lazy you can use this to allow CamelContext and routes to startup in
situations where a producer may otherwise fai [...]
- "replySubject": { "index": 30, "kind": "property", "displayName": "Reply
Subject", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "the subject to which subscribers should send
response" },
- "requestTimeout": { "index": 31, "kind": "property", "displayName":
"Request Timeout", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 20000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Request timeout in milliseconds" },
- "autowiredEnabled": { "index": 32, "kind": "property", "displayName":
"Autowired Enabled", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether autowiring is enabled. This is used for automatic autowiring options
(the option must be marked as autowired) by looking up in the registry to find
if there is a single instance of matching [...]
- "connection": { "index": 33, "kind": "property", "displayName":
"Connection", "group": "advanced", "label": "advanced", "required": false,
"type": "object", "javaType": "io.nats.client.Connection", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Reference an already instantiated connection
to Nats server" },
- "headerFilterStrategy": { "index": 34, "kind": "property", "displayName":
"Header Filter Strategy", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To use a custom header filter strategy." },
- "jetstreamAsync": { "index": 35, "kind": "property", "displayName":
"Jetstream Async", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets whether to operate JetStream requests
asynchronously." },
- "traceConnection": { "index": 36, "kind": "property", "displayName":
"Trace Connection", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Whether or not
connection trace messages should be printed to standard out for fine g [...]
- "credentialsFilePath": { "index": 37, "kind": "property", "displayName":
"Credentials File Path", "group": "security", "label": "security", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "If we use useCredentialsFile to true we'll
need to set the credentialsFilePath option. It c [...]
- "secure": { "index": 38, "kind": "property", "displayName": "Secure",
"group": "security", "label": "security", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Set secure option indicating TLS is required"
},
- "sslContextParameters": { "index": 39, "kind": "property", "displayName":
"Ssl Context Parameters", "group": "security", "label": "security", "required":
false, "type": "object", "javaType":
"org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To configure security using
SSLContextParameters" },
- "useGlobalSslContextParameters": { "index": 40, "kind": "property",
"displayName": "Use Global Ssl Context Parameters", "group": "security",
"label": "security", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Enable usage of global SSL context
parameters." }
+ "pullBatchSize": { "index": 25, "kind": "property", "displayName": "Pull
Batch Size", "group": "consumer", "label": "consumer", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 10, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Maximum number of messages to fetch per pull
request when using a JetStream Pull Subscription. [...]
+ "pullFetchTimeout": { "index": 26, "kind": "property", "displayName":
"Pull Fetch Timeout", "group": "consumer", "label": "consumer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Maximum time (in milliseconds) to wait for a
batch of messages to be available on the [...]
+ "pullSubscription": { "index": 27, "kind": "property", "displayName":
"Pull Subscription", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Sets the consumer
subscription type for JetStream. Set to true to use a Pull Subscri [...]
+ "queueName": { "index": 28, "kind": "property", "displayName": "Queue
Name", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "The Queue name if we are using nats for a
queue configuration" },
+ "replyToDisabled": { "index": 29, "kind": "property", "displayName":
"Reply To Disabled", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Can be used to turn off
sending back reply message in the consumer." },
+ "consumerConfiguration": { "index": 30, "kind": "property", "displayName":
"Consumer Configuration", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"io.nats.client.api.ConsumerConfiguration", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets a custom ConsumerConfiguration object f
[...]
+ "lazyStartProducer": { "index": 31, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Whether the producer should be started lazy (on the first message). By
starting lazy you can use this to allow CamelContext and routes to startup in
situations where a producer may otherwise fai [...]
+ "replySubject": { "index": 32, "kind": "property", "displayName": "Reply
Subject", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "the subject to which subscribers should send
response" },
+ "requestTimeout": { "index": 33, "kind": "property", "displayName":
"Request Timeout", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 20000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Request timeout in milliseconds" },
+ "autowiredEnabled": { "index": 34, "kind": "property", "displayName":
"Autowired Enabled", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true, "description":
"Whether autowiring is enabled. This is used for automatic autowiring options
(the option must be marked as autowired) by looking up in the registry to find
if there is a single instance of matching [...]
+ "connection": { "index": 35, "kind": "property", "displayName":
"Connection", "group": "advanced", "label": "advanced", "required": false,
"type": "object", "javaType": "io.nats.client.Connection", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Reference an already instantiated connection
to Nats server" },
+ "headerFilterStrategy": { "index": 36, "kind": "property", "displayName":
"Header Filter Strategy", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To use a custom header filter strategy." },
+ "jetstreamAsync": { "index": 37, "kind": "property", "displayName":
"Jetstream Async", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets whether to operate JetStream requests
asynchronously." },
+ "traceConnection": { "index": 38, "kind": "property", "displayName":
"Trace Connection", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Whether or not
connection trace messages should be printed to standard out for fine g [...]
+ "credentialsFilePath": { "index": 39, "kind": "property", "displayName":
"Credentials File Path", "group": "security", "label": "security", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "If we use useCredentialsFile to true we'll
need to set the credentialsFilePath option. It c [...]
+ "secure": { "index": 40, "kind": "property", "displayName": "Secure",
"group": "security", "label": "security", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Set secure option indicating TLS is required"
},
+ "sslContextParameters": { "index": 41, "kind": "property", "displayName":
"Ssl Context Parameters", "group": "security", "label": "security", "required":
false, "type": "object", "javaType":
"org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To configure security using
SSLContextParameters" },
+ "useGlobalSslContextParameters": { "index": 42, "kind": "property",
"displayName": "Use Global Ssl Context Parameters", "group": "security",
"label": "security", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Enable usage of global SSL context
parameters." }
},
"headers": {
"CamelNatsMessageTimestamp": { "index": 0, "kind": "header",
"displayName": "", "group": "common", "label": "", "required": false,
"javaType": "long", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The timestamp of a consumed message.",
"constantName":
"org.apache.camel.component.nats.NatsConstants#NATS_MESSAGE_TIMESTAMP" },
@@ -101,22 +103,24 @@
"maxMessages": { "index": 21, "kind": "parameter", "displayName": "Max
Messages", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Stop receiving messages from a topic we are
subscribing to after maxMessages" },
"nackWait": { "index": 22, "kind": "parameter", "displayName": "Nack
Wait", "group": "consumer", "label": "consumer", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 5000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "For negative acknowledgements (NAK),
redelivery is delayed by 5 seconds (default). Setting this to 0 o [...]
"poolSize": { "index": 23, "kind": "parameter", "displayName": "Pool
Size", "group": "consumer", "label": "consumer", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 10, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Consumer thread pool size (default is 10)" },
- "pullSubscription": { "index": 24, "kind": "parameter", "displayName":
"Pull Subscription", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Sets the consumer
subscription type for JetStream. Set to true to use a Pull Subscr [...]
- "queueName": { "index": 25, "kind": "parameter", "displayName": "Queue
Name", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "The Queue name if we are using nats for a
queue configuration" },
- "replyToDisabled": { "index": 26, "kind": "parameter", "displayName":
"Reply To Disabled", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Can be used to turn off
sending back reply message in the consumer." },
- "bridgeErrorHandler": { "index": 27, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
- "consumerConfiguration": { "index": 28, "kind": "parameter",
"displayName": "Consumer Configuration", "group": "consumer (advanced)",
"label": "consumer,advanced", "required": false, "type": "object", "javaType":
"io.nats.client.api.ConsumerConfiguration", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets a custom ConsumerConfiguration object
[...]
- "exceptionHandler": { "index": 29, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By de [...]
- "exchangePattern": { "index": 30, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
- "replySubject": { "index": 31, "kind": "parameter", "displayName": "Reply
Subject", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "the subject to which subscribers should send
response" },
- "requestTimeout": { "index": 32, "kind": "parameter", "displayName":
"Request Timeout", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 20000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Request timeout in milliseconds" },
- "lazyStartProducer": { "index": 33, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produ [...]
- "connection": { "index": 34, "kind": "parameter", "displayName":
"Connection", "group": "advanced", "label": "advanced", "required": false,
"type": "object", "javaType": "io.nats.client.Connection", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Reference an already instantiated connection
to Nats server" },
- "headerFilterStrategy": { "index": 35, "kind": "parameter", "displayName":
"Header Filter Strategy", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To use a custom header filter strategy." },
- "jetstreamAsync": { "index": 36, "kind": "parameter", "displayName":
"Jetstream Async", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets whether to operate JetStream requests
asynchronously." },
- "traceConnection": { "index": 37, "kind": "parameter", "displayName":
"Trace Connection", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Whether or not
connection trace messages should be printed to standard out for fine [...]
- "credentialsFilePath": { "index": 38, "kind": "parameter", "displayName":
"Credentials File Path", "group": "security", "label": "security", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "If we use useCredentialsFile to true we'll
need to set the credentialsFilePath option. It [...]
- "secure": { "index": 39, "kind": "parameter", "displayName": "Secure",
"group": "security", "label": "security", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Set secure option indicating TLS is required"
},
- "sslContextParameters": { "index": 40, "kind": "parameter", "displayName":
"Ssl Context Parameters", "group": "security", "label": "security", "required":
false, "type": "object", "javaType":
"org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To configure security using
SSLContextParameters" }
+ "pullBatchSize": { "index": 24, "kind": "parameter", "displayName": "Pull
Batch Size", "group": "consumer", "label": "consumer", "required": false,
"type": "integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 10, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Maximum number of messages to fetch per pull
request when using a JetStream Pull Subscription. [...]
+ "pullFetchTimeout": { "index": 25, "kind": "parameter", "displayName":
"Pull Fetch Timeout", "group": "consumer", "label": "consumer", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Maximum time (in milliseconds) to wait for a
batch of messages to be available on the [...]
+ "pullSubscription": { "index": 26, "kind": "parameter", "displayName":
"Pull Subscription", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": true,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Sets the consumer
subscription type for JetStream. Set to true to use a Pull Subscr [...]
+ "queueName": { "index": 27, "kind": "parameter", "displayName": "Queue
Name", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "The Queue name if we are using nats for a
queue configuration" },
+ "replyToDisabled": { "index": 28, "kind": "parameter", "displayName":
"Reply To Disabled", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Can be used to turn off
sending back reply message in the consumer." },
+ "bridgeErrorHandler": { "index": 29, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
+ "consumerConfiguration": { "index": 30, "kind": "parameter",
"displayName": "Consumer Configuration", "group": "consumer (advanced)",
"label": "consumer,advanced", "required": false, "type": "object", "javaType":
"io.nats.client.api.ConsumerConfiguration", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets a custom ConsumerConfiguration object
[...]
+ "exceptionHandler": { "index": 31, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By de [...]
+ "exchangePattern": { "index": 32, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
+ "replySubject": { "index": 33, "kind": "parameter", "displayName": "Reply
Subject", "group": "producer", "label": "producer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "the subject to which subscribers should send
response" },
+ "requestTimeout": { "index": 34, "kind": "parameter", "displayName":
"Request Timeout", "group": "producer", "label": "producer", "required": false,
"type": "integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 20000, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Request timeout in milliseconds" },
+ "lazyStartProducer": { "index": 35, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produ [...]
+ "connection": { "index": 36, "kind": "parameter", "displayName":
"Connection", "group": "advanced", "label": "advanced", "required": false,
"type": "object", "javaType": "io.nats.client.Connection", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Reference an already instantiated connection
to Nats server" },
+ "headerFilterStrategy": { "index": 37, "kind": "parameter", "displayName":
"Header Filter Strategy", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.spi.HeaderFilterStrategy", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To use a custom header filter strategy." },
+ "jetstreamAsync": { "index": 38, "kind": "parameter", "displayName":
"Jetstream Async", "group": "advanced", "label": "advanced", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": true, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Sets whether to operate JetStream requests
asynchronously." },
+ "traceConnection": { "index": 39, "kind": "parameter", "displayName":
"Trace Connection", "group": "advanced", "label": "advanced", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false,
"configurationClass": "org.apache.camel.component.nats.NatsConfiguration",
"configurationField": "configuration", "description": "Whether or not
connection trace messages should be printed to standard out for fine [...]
+ "credentialsFilePath": { "index": 40, "kind": "parameter", "displayName":
"Credentials File Path", "group": "security", "label": "security", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "If we use useCredentialsFile to true we'll
need to set the credentialsFilePath option. It [...]
+ "secure": { "index": 41, "kind": "parameter", "displayName": "Secure",
"group": "security", "label": "security", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "Set secure option indicating TLS is required"
},
+ "sslContextParameters": { "index": 42, "kind": "parameter", "displayName":
"Ssl Context Parameters", "group": "security", "label": "security", "required":
false, "type": "object", "javaType":
"org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.nats.NatsConfiguration", "configurationField":
"configuration", "description": "To configure security using
SSLContextParameters" }
}
}
diff --git
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
index fee6e509fe4b..ea2fa55136a5 100644
---
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
+++
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
@@ -99,6 +99,10 @@ public class NatsConfiguration implements Cloneable {
private ConsumerConfiguration consumerConfiguration;
@UriParam(label = "consumer", defaultValue = "true")
private boolean pullSubscription = true;
+ @UriParam(label = "consumer", defaultValue = "10")
+ private int pullBatchSize = 10;
+ @UriParam(label = "consumer", defaultValue = "1000")
+ private long pullFetchTimeout = 1000;
@UriParam(label = "consumer")
private String durableName;
@UriParam(label = "security")
@@ -568,6 +572,30 @@ public class NatsConfiguration implements Cloneable {
this.pullSubscription = pullSubscription;
}
+ public int getPullBatchSize() {
+ return pullBatchSize;
+ }
+
+ /**
+ * Maximum number of messages to fetch per pull request when using a
JetStream Pull Subscription. Only used when
+ * {@code pullSubscription=true}.
+ */
+ public void setPullBatchSize(int pullBatchSize) {
+ this.pullBatchSize = pullBatchSize;
+ }
+
+ public long getPullFetchTimeout() {
+ return pullFetchTimeout;
+ }
+
+ /**
+ * Maximum time (in milliseconds) to wait for a batch of messages to be
available on the server during a single
+ * fetch when using a JetStream Pull Subscription. Only used when {@code
pullSubscription=true}.
+ */
+ public void setPullFetchTimeout(long pullFetchTimeout) {
+ this.pullFetchTimeout = pullFetchTimeout;
+ }
+
/**
* The name to assign to the JetStream durable consumer.
* <p>
diff --git
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
index fcedd2c1f2df..c36d0be1fdcc 100644
---
a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
+++
b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.nats;
import java.io.IOException;
import java.time.Duration;
+import java.util.List;
import java.util.concurrent.ExecutorService;
import io.nats.client.Connection;
@@ -25,6 +26,7 @@ import io.nats.client.Connection.Status;
import io.nats.client.Dispatcher;
import io.nats.client.JetStreamApiException;
import io.nats.client.JetStreamManagement;
+import io.nats.client.JetStreamStatusException;
import io.nats.client.JetStreamSubscription;
import io.nats.client.Message;
import io.nats.client.MessageHandler;
@@ -96,6 +98,14 @@ public class NatsConsumer extends DefaultConsumer {
}
}
+ if (ObjectHelper.isNotEmpty(this.jetStreamSubscription) &&
this.jetStreamSubscription.isActive()) {
+ try {
+ this.jetStreamSubscription.unsubscribe();
+ } catch (final Exception e) {
+ this.getExceptionHandler().handleException("Error during
unsubscribing", e);
+ }
+ }
+
LOG.debug("Stopping Nats Consumer");
if (this.executor != null) {
if (this.getEndpoint() != null &&
this.getEndpoint().getCamelContext() != null) {
@@ -201,19 +211,26 @@ public class NatsConsumer extends DefaultConsumer {
}
CamelNatsMessageHandler messageHandler = new
CamelNatsMessageHandler();
- NatsConsumer.this.dispatcher =
this.connection.createDispatcher(messageHandler);
if (this.configuration.isPullSubscription()) {
PullSubscribeOptions pullOptions =
PullSubscribeOptions.builder()
.configuration(cc)
.build();
+ LOG.info(
+ "Subscribing topic: {} using AckPolicy: {} (ackWait:{}
nackWait:{} pullBatchSize:{} pullFetchTimeout:{})",
+ configuration.getTopic(),
configuration.getAckPolicy(), configuration.getAckWait(),
+ configuration.getNackWait(),
configuration.getPullBatchSize(), configuration.getPullFetchTimeout());
+
NatsConsumer.this.jetStreamSubscription =
this.connection.jetStream().subscribe(
NatsConsumer.this.getEndpoint().getConfiguration().getTopic(),
- dispatcher,
- messageHandler,
pullOptions);
+
+ NatsConsumer.this.setActive(true);
+ runPullFetchLoop(NatsConsumer.this.jetStreamSubscription,
messageHandler);
} else {
+ NatsConsumer.this.dispatcher =
this.connection.createDispatcher(messageHandler);
+
PushSubscribeOptions pushOptions =
PushSubscribeOptions.builder()
.configuration(cc)
.deliverGroup(queueName)
@@ -232,9 +249,41 @@ public class NatsConsumer extends DefaultConsumer {
messageHandler,
autoAck,
pushOptions);
+
+ NatsConsumer.this.setActive(true);
}
+ }
- NatsConsumer.this.setActive(true);
+ private void runPullFetchLoop(JetStreamSubscription sub,
CamelNatsMessageHandler messageHandler) {
+ int batchSize = configuration.getPullBatchSize();
+ Duration fetchTimeout =
Duration.ofMillis(configuration.getPullFetchTimeout());
+ while (NatsConsumer.this.isRunAllowed() &&
!Thread.currentThread().isInterrupted()) {
+ try {
+ List<Message> messages = sub.fetch(batchSize,
fetchTimeout);
+ for (Message msg : messages) {
+ if (!NatsConsumer.this.isRunAllowed()) {
+ break;
+ }
+ try {
+ messageHandler.onMessage(msg);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ } catch (JetStreamStatusException e) {
+ // Synchronous pull calls may throw
JetStreamStatusException for transient
+ // statuses such as 409 (consumer deleted/push-based). Log
and continue so the
+ // consumer can recover when the consumer state is
restored.
+
NatsConsumer.this.getExceptionHandler().handleException("JetStream status
during pull fetch", e);
+ } catch (IllegalStateException e) {
+ // Subscription was closed (e.g. consumer is stopping),
exit the loop
+ LOG.debug("Pull subscription closed, stopping fetch loop",
e);
+ return;
+ } catch (Exception e) {
+
NatsConsumer.this.getExceptionHandler().handleException("Error during JetStream
pull fetch", e);
+ }
+ }
}
private void setupStandardNatsConsumer(String topic, String queueName,
Integer maxMessages) {
diff --git
a/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullIT.java
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullIT.java
new file mode 100644
index 000000000000..f1a1b23f00d1
--- /dev/null
+++
b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullIT.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.nats.jetstream;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.nats.NatsConstants;
+import org.apache.camel.component.nats.integration.NatsITSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Isolated;
+
+@Isolated
+public class NatsJetstreamConsumerPullIT extends NatsITSupport {
+
+ @EndpointInject("mock:result")
+ protected MockEndpoint mockResultEndpoint;
+
+ @Test
+ public void testPullConsumer() throws Exception {
+ mockResultEndpoint.expectedBodiesReceived("Hello Pull");
+ mockResultEndpoint.expectedHeaderReceived(NatsConstants.NATS_SUBJECT,
"mytopic-pull");
+
+ template.sendBody("direct:send", "Hello Pull");
+
+ mockResultEndpoint.setAssertPeriod(5000);
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ String producerUri
+ =
"nats:mytopic-pull?jetstreamEnabled=true&jetstreamName=mystream-pull&jetstreamAsync=false";
+ String consumerUri
+ =
"nats:mytopic-pull?jetstreamEnabled=true&jetstreamName=mystream-pull&jetstreamAsync=false&durableName=camel-pull&pullSubscription=true&pullFetchTimeout=500";
+
+ from("direct:send")
+ // when running full test suite then send can fail due
to nats server setup/teardown
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(5))
+ .to(producerUri);
+
+ from(consumerUri).to(mockResultEndpoint);
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_21.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_21.adoc
index 8e70f77d4738..02af0fb2e1fc 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_21.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_21.adoc
@@ -268,6 +268,18 @@ What it means is that, from now on, the final user or any
third party dependency
This is just an informative note, there is not action expected by the final
user.
+=== camel-nats
+
+Fixed the JetStream consumer pull subscription mode (which is the default) so
that messages are
+actually consumed from the server. Previously the consumer would set up a pull
subscription but never
+issue any pull/fetch requests, so no messages were delivered until the user
explicitly switched to
+push mode with `pullSubscription=false`.
+
+Two new endpoint options have been added to control the pull fetch loop:
+
+* `pullBatchSize` (default `10`) — maximum number of messages to fetch per
pull request.
+* `pullFetchTimeout` (default `1000` ms) — maximum time to wait for a batch on
each fetch.
+
=== Deprecation of camel-ironmq
The component camel-ironmq is deprecated. The official library used has been
unmaintained since 2017
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/NatsComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/NatsComponentBuilderFactory.java
index b78c037b7c72..74051188ea21 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/NatsComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/NatsComponentBuilderFactory.java
@@ -509,6 +509,44 @@ public interface NatsComponentBuilderFactory {
}
+ /**
+ * Maximum number of messages to fetch per pull request when using a
+ * JetStream Pull Subscription. Only used when {code
+ * pullSubscription=true}.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 10
+ * Group: consumer
+ *
+ * @param pullBatchSize the value to set
+ * @return the dsl builder
+ */
+ default NatsComponentBuilder pullBatchSize(int pullBatchSize) {
+ doSetProperty("pullBatchSize", pullBatchSize);
+ return this;
+ }
+
+
+ /**
+ * Maximum time (in milliseconds) to wait for a batch of messages to be
+ * available on the server during a single fetch when using a JetStream
+ * Pull Subscription. Only used when {code pullSubscription=true}.
+ *
+ * The option is a: <code>long</code> type.
+ *
+ * Default: 1000
+ * Group: consumer
+ *
+ * @param pullFetchTimeout the value to set
+ * @return the dsl builder
+ */
+ default NatsComponentBuilder pullFetchTimeout(long pullFetchTimeout) {
+ doSetProperty("pullFetchTimeout", pullFetchTimeout);
+ return this;
+ }
+
+
/**
* Sets the consumer subscription type for JetStream. Set to true to
use
* a Pull Subscription (consumer explicitly requests messages). Set to
@@ -841,6 +879,8 @@ public interface NatsComponentBuilderFactory {
case "maxMessages": getOrCreateConfiguration((NatsComponent)
component).setMaxMessages((java.lang.String) value); return true;
case "nackWait": getOrCreateConfiguration((NatsComponent)
component).setNackWait((long) value); return true;
case "poolSize": getOrCreateConfiguration((NatsComponent)
component).setPoolSize((int) value); return true;
+ case "pullBatchSize": getOrCreateConfiguration((NatsComponent)
component).setPullBatchSize((int) value); return true;
+ case "pullFetchTimeout": getOrCreateConfiguration((NatsComponent)
component).setPullFetchTimeout((long) value); return true;
case "pullSubscription": getOrCreateConfiguration((NatsComponent)
component).setPullSubscription((boolean) value); return true;
case "queueName": getOrCreateConfiguration((NatsComponent)
component).setQueueName((java.lang.String) value); return true;
case "replyToDisabled": getOrCreateConfiguration((NatsComponent)
component).setReplyToDisabled((boolean) value); return true;
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/NatsEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/NatsEndpointBuilderFactory.java
index 004f94329b6b..85a1442fb647 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/NatsEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/NatsEndpointBuilderFactory.java
@@ -728,6 +728,74 @@ public interface NatsEndpointBuilderFactory {
doSetProperty("poolSize", poolSize);
return this;
}
+ /**
+ * Maximum number of messages to fetch per pull request when using a
+ * JetStream Pull Subscription. Only used when {code
+ * pullSubscription=true}.
+ *
+ * The option is a: <code>int</code> type.
+ *
+ * Default: 10
+ * Group: consumer
+ *
+ * @param pullBatchSize the value to set
+ * @return the dsl builder
+ */
+ default NatsEndpointConsumerBuilder pullBatchSize(int pullBatchSize) {
+ doSetProperty("pullBatchSize", pullBatchSize);
+ return this;
+ }
+ /**
+ * Maximum number of messages to fetch per pull request when using a
+ * JetStream Pull Subscription. Only used when {code
+ * pullSubscription=true}.
+ *
+ * The option will be converted to a <code>int</code> type.
+ *
+ * Default: 10
+ * Group: consumer
+ *
+ * @param pullBatchSize the value to set
+ * @return the dsl builder
+ */
+ default NatsEndpointConsumerBuilder pullBatchSize(String
pullBatchSize) {
+ doSetProperty("pullBatchSize", pullBatchSize);
+ return this;
+ }
+ /**
+ * Maximum time (in milliseconds) to wait for a batch of messages to be
+ * available on the server during a single fetch when using a JetStream
+ * Pull Subscription. Only used when {code pullSubscription=true}.
+ *
+ * The option is a: <code>long</code> type.
+ *
+ * Default: 1000
+ * Group: consumer
+ *
+ * @param pullFetchTimeout the value to set
+ * @return the dsl builder
+ */
+ default NatsEndpointConsumerBuilder pullFetchTimeout(long
pullFetchTimeout) {
+ doSetProperty("pullFetchTimeout", pullFetchTimeout);
+ return this;
+ }
+ /**
+ * Maximum time (in milliseconds) to wait for a batch of messages to be
+ * available on the server during a single fetch when using a JetStream
+ * Pull Subscription. Only used when {code pullSubscription=true}.
+ *
+ * The option will be converted to a <code>long</code> type.
+ *
+ * Default: 1000
+ * Group: consumer
+ *
+ * @param pullFetchTimeout the value to set
+ * @return the dsl builder
+ */
+ default NatsEndpointConsumerBuilder pullFetchTimeout(String
pullFetchTimeout) {
+ doSetProperty("pullFetchTimeout", pullFetchTimeout);
+ return this;
+ }
/**
* Sets the consumer subscription type for JetStream. Set to true to
use
* a Pull Subscription (consumer explicitly requests messages). Set to
