You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by fm...@apache.org on 2026/05/12 16:32:52 UTC
(camel) branch main updated: CAMEL-23483: add manual acknowledgment support for camel-nats JetStream consumer
This is an automated email from the ASF dual-hosted git repository.
Croway 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 17a72793ef57 CAMEL-23483: add manual acknowledgment support for camel-nats JetStream consumer
17a72793ef57 is described below
commit 17a72793ef57c14c5a61c223c641788957dcf571
Author: Croway <fe...@gmail.com>
AuthorDate: Tue May 12 12:03:16 2026 +0200
CAMEL-23483: add manual acknowledgment support for camel-nats JetStream consumer
---
.../org/apache/camel/catalog/components/nats.json | 95 +++++-----
.../component/nats/NatsComponentConfigurer.java | 6 +
.../component/nats/NatsEndpointConfigurer.java | 6 +
.../component/nats/NatsEndpointUriFactory.java | 3 +-
.../org/apache/camel/component/nats/nats.json | 95 +++++-----
.../camel-nats/src/main/docs/nats-component.adoc | 23 +++
.../camel/component/nats/DefaultNatsManualAck.java | 55 ++++++
.../camel/component/nats/NatsConfiguration.java | 23 +++
.../apache/camel/component/nats/NatsConstants.java | 4 +
.../apache/camel/component/nats/NatsConsumer.java | 53 ++++--
.../apache/camel/component/nats/NatsManualAck.java | 53 ++++++
.../NatsJetstreamConsumerManualAckIT.java | 196 +++++++++++++++++++++
.../NatsJetstreamConsumerManualAckPullIT.java | 122 +++++++++++++
.../component/dsl/NatsComponentBuilderFactory.java | 28 +++
.../endpoint/dsl/NatsEndpointBuilderFactory.java | 64 +++++++
.../services/NatsLocalContainerAuthService.java | 1 +
.../NatsLocalContainerAuthTokenService.java | 1 +
.../services/NatsLocalContainerInfraService.java | 1 +
.../services/NatsLocalContainerTLSAuthService.java | 1 +
19 files changed, 718 insertions(+), 112 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 a83b6b4b6373..16a6c71fa2bb 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
@@ -45,28 +45,29 @@
"ackWait": { "index": 18, "kind": "property", "displayName": "Ack Wait", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "After a message is delivered to a consumer, the server waits 30 seconds (default) for an acknowledgement [...]
"bridgeErrorHandler": { "index": 19, "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the lik [...]
"durableName": { "index": 20, "kind": "property", "displayName": "Durable 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": "Sets the name to assign to the JetStream durable consumer. Setting this value makes the consumer durable. Th [...]
- "maxDeliver": { "index": 21, "kind": "property", "displayName": "Max Deliver", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Maximum number of attempts to deliver a message from Nats to a consumer. Once MaxDeliver is reached, the NATS server stop [...]
- "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)" },
- "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." }
+ "manualAck": { "index": 21, "kind": "property", "displayName": "Manual Ack", "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": "Whether to allow doing manual acknowledgment via NatsManualAck. If this option is enabled then an [...]
+ "maxDeliver": { "index": 22, "kind": "property", "displayName": "Max Deliver", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Maximum number of attempts to deliver a message from Nats to a consumer. Once MaxDeliver is reached, the NATS server stop [...]
+ "maxMessages": { "index": 23, "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": 24, "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": 25, "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)" },
+ "pullBatchSize": { "index": 26, "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": 27, "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": 28, "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": 29, "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": 30, "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": 31, "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": 32, "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": 33, "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": 34, "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": 35, "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": 36, "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": 37, "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": 38, "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": 39, "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": 40, "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": 41, "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": 42, "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": 43, "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" },
@@ -76,7 +77,8 @@
"CamelNatsQueueName": { "index": 4, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The Queue name of a consumed message (may be null).", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_QUEUE_NAME" },
"CamelNatsStatusCode": { "index": 5, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "int", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "important": true, "description": "Status message code", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_STATUS_CODE" },
"CamelNatsStatusError": { "index": 6, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "important": true, "description": "Status message error message", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_STATUS_ERROR" },
- "CamelNatsDeliveryCounter": { "index": 7, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "important": true, "description": "Number of times this message has been delivered (1 = first, 1 then message has been redelivered)", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_DELIVERY_COUNTER" }
+ "CamelNatsDeliveryCounter": { "index": 7, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "important": true, "description": "Number of times this message has been delivered (1 = first, 1 then message has been redelivered)", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_DELIVERY_COUNTER" },
+ "CamelNatsManualAck": { "index": 8, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "org.apache.camel.component.nats.NatsManualAck", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The manual acknowledgment handle for JetStream messages (only set when manualAck=true).", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_MANUAL_ACK" }
},
"properties": {
"topic": { "index": 0, "kind": "path", "displayName": "Topic", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "The name of topic we want to use" },
@@ -99,28 +101,29 @@
"ackPolicy": { "index": 17, "kind": "parameter", "displayName": "Ack Policy", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "io.nats.client.api.AckPolicy", "enum": [ "None", "All", "Explicit" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "Explicit", "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Acknowledgement mode. none = Messa [...]
"ackWait": { "index": 18, "kind": "parameter", "displayName": "Ack Wait", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "After a message is delivered to a consumer, the server waits 30 seconds (default) for an acknowledgemen [...]
"durableName": { "index": 19, "kind": "parameter", "displayName": "Durable 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": "Sets the name to assign to the JetStream durable consumer. Setting this value makes the consumer durable. T [...]
- "maxDeliver": { "index": 20, "kind": "parameter", "displayName": "Max Deliver", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Maximum number of attempts to deliver a message from Nats to a consumer. Once MaxDeliver is reached, the NATS server sto [...]
- "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)" },
- "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" }
+ "manualAck": { "index": 20, "kind": "parameter", "displayName": "Manual Ack", "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": "Whether to allow doing manual acknowledgment via NatsManualAck. If this option is enabled then a [...]
+ "maxDeliver": { "index": 21, "kind": "parameter", "displayName": "Max Deliver", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Maximum number of attempts to deliver a message from Nats to a consumer. Once MaxDeliver is reached, the NATS server sto [...]
+ "maxMessages": { "index": 22, "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": 23, "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": 24, "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)" },
+ "pullBatchSize": { "index": 25, "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": 26, "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": 27, "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": 28, "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": 29, "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": 30, "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": 31, "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": 32, "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": 33, "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": 34, "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": 35, "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": 36, "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": 37, "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": 38, "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": 39, "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": 40, "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": 41, "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": 42, "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": 43, "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 d0891a119914..b9f2658bec57 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
@@ -62,6 +62,8 @@ public class NatsComponentConfigurer extends PropertyConfigurerSupport implement
case "jetstreamName": getOrCreateConfiguration(target).setJetstreamName(property(camelContext, java.lang.String.class, value)); return true;
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "manualack":
+ case "manualAck": getOrCreateConfiguration(target).setManualAck(property(camelContext, boolean.class, value)); return true;
case "maxdeliver":
case "maxDeliver": getOrCreateConfiguration(target).setMaxDeliver(property(camelContext, long.class, value)); return true;
case "maxmessages":
@@ -148,6 +150,8 @@ public class NatsComponentConfigurer extends PropertyConfigurerSupport implement
case "jetstreamName": return java.lang.String.class;
case "lazystartproducer":
case "lazyStartProducer": return boolean.class;
+ case "manualack":
+ case "manualAck": return boolean.class;
case "maxdeliver":
case "maxDeliver": return long.class;
case "maxmessages":
@@ -235,6 +239,8 @@ public class NatsComponentConfigurer extends PropertyConfigurerSupport implement
case "jetstreamName": return getOrCreateConfiguration(target).getJetstreamName();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
+ case "manualack":
+ case "manualAck": return getOrCreateConfiguration(target).isManualAck();
case "maxdeliver":
case "maxDeliver": return getOrCreateConfiguration(target).getMaxDeliver();
case "maxmessages":
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 923781a8bc8a..162644fb2792 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
@@ -56,6 +56,8 @@ public class NatsEndpointConfigurer extends PropertyConfigurerSupport implements
case "jetstreamName": target.getConfiguration().setJetstreamName(property(camelContext, java.lang.String.class, value)); return true;
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "manualack":
+ case "manualAck": target.getConfiguration().setManualAck(property(camelContext, boolean.class, value)); return true;
case "maxdeliver":
case "maxDeliver": target.getConfiguration().setMaxDeliver(property(camelContext, long.class, value)); return true;
case "maxmessages":
@@ -141,6 +143,8 @@ public class NatsEndpointConfigurer extends PropertyConfigurerSupport implements
case "jetstreamName": return java.lang.String.class;
case "lazystartproducer":
case "lazyStartProducer": return boolean.class;
+ case "manualack":
+ case "manualAck": return boolean.class;
case "maxdeliver":
case "maxDeliver": return long.class;
case "maxmessages":
@@ -227,6 +231,8 @@ public class NatsEndpointConfigurer extends PropertyConfigurerSupport implements
case "jetstreamName": return target.getConfiguration().getJetstreamName();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
+ case "manualack":
+ case "manualAck": return target.getConfiguration().isManualAck();
case "maxdeliver":
case "maxDeliver": return target.getConfiguration().getMaxDeliver();
case "maxmessages":
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 542551087298..fd6be14547dd 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<>(43);
+ Set<String> props = new HashSet<>(44);
props.add("ackPolicy");
props.add("ackWait");
props.add("bridgeErrorHandler");
@@ -41,6 +41,7 @@ public class NatsEndpointUriFactory extends org.apache.camel.support.component.E
props.add("jetstreamEnabled");
props.add("jetstreamName");
props.add("lazyStartProducer");
+ props.add("manualAck");
props.add("maxDeliver");
props.add("maxMessages");
props.add("maxPingsOut");
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 a83b6b4b6373..16a6c71fa2bb 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
@@ -45,28 +45,29 @@
"ackWait": { "index": 18, "kind": "property", "displayName": "Ack Wait", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "After a message is delivered to a consumer, the server waits 30 seconds (default) for an acknowledgement [...]
"bridgeErrorHandler": { "index": 19, "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the lik [...]
"durableName": { "index": 20, "kind": "property", "displayName": "Durable 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": "Sets the name to assign to the JetStream durable consumer. Setting this value makes the consumer durable. Th [...]
- "maxDeliver": { "index": 21, "kind": "property", "displayName": "Max Deliver", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Maximum number of attempts to deliver a message from Nats to a consumer. Once MaxDeliver is reached, the NATS server stop [...]
- "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)" },
- "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." }
+ "manualAck": { "index": 21, "kind": "property", "displayName": "Manual Ack", "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": "Whether to allow doing manual acknowledgment via NatsManualAck. If this option is enabled then an [...]
+ "maxDeliver": { "index": 22, "kind": "property", "displayName": "Max Deliver", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Maximum number of attempts to deliver a message from Nats to a consumer. Once MaxDeliver is reached, the NATS server stop [...]
+ "maxMessages": { "index": 23, "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": 24, "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": 25, "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)" },
+ "pullBatchSize": { "index": 26, "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": 27, "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": 28, "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": 29, "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": 30, "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": 31, "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": 32, "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": 33, "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": 34, "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": 35, "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": 36, "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": 37, "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": 38, "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": 39, "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": 40, "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": 41, "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": 42, "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": 43, "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" },
@@ -76,7 +77,8 @@
"CamelNatsQueueName": { "index": 4, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The Queue name of a consumed message (may be null).", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_QUEUE_NAME" },
"CamelNatsStatusCode": { "index": 5, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "int", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "important": true, "description": "Status message code", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_STATUS_CODE" },
"CamelNatsStatusError": { "index": 6, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "important": true, "description": "Status message error message", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_STATUS_ERROR" },
- "CamelNatsDeliveryCounter": { "index": 7, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "important": true, "description": "Number of times this message has been delivered (1 = first, 1 then message has been redelivered)", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_DELIVERY_COUNTER" }
+ "CamelNatsDeliveryCounter": { "index": 7, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "important": true, "description": "Number of times this message has been delivered (1 = first, 1 then message has been redelivered)", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_DELIVERY_COUNTER" },
+ "CamelNatsManualAck": { "index": 8, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "org.apache.camel.component.nats.NatsManualAck", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The manual acknowledgment handle for JetStream messages (only set when manualAck=true).", "constantName": "org.apache.camel.component.nats.NatsConstants#NATS_MANUAL_ACK" }
},
"properties": {
"topic": { "index": 0, "kind": "path", "displayName": "Topic", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "The name of topic we want to use" },
@@ -99,28 +101,29 @@
"ackPolicy": { "index": 17, "kind": "parameter", "displayName": "Ack Policy", "group": "consumer", "label": "consumer", "required": false, "type": "enum", "javaType": "io.nats.client.api.AckPolicy", "enum": [ "None", "All", "Explicit" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "Explicit", "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Acknowledgement mode. none = Messa [...]
"ackWait": { "index": 18, "kind": "parameter", "displayName": "Ack Wait", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "After a message is delivered to a consumer, the server waits 30 seconds (default) for an acknowledgemen [...]
"durableName": { "index": 19, "kind": "parameter", "displayName": "Durable 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": "Sets the name to assign to the JetStream durable consumer. Setting this value makes the consumer durable. T [...]
- "maxDeliver": { "index": 20, "kind": "parameter", "displayName": "Max Deliver", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Maximum number of attempts to deliver a message from Nats to a consumer. Once MaxDeliver is reached, the NATS server sto [...]
- "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)" },
- "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" }
+ "manualAck": { "index": 20, "kind": "parameter", "displayName": "Manual Ack", "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": "Whether to allow doing manual acknowledgment via NatsManualAck. If this option is enabled then a [...]
+ "maxDeliver": { "index": 21, "kind": "parameter", "displayName": "Max Deliver", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.nats.NatsConfiguration", "configurationField": "configuration", "description": "Maximum number of attempts to deliver a message from Nats to a consumer. Once MaxDeliver is reached, the NATS server sto [...]
+ "maxMessages": { "index": 22, "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": 23, "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": 24, "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)" },
+ "pullBatchSize": { "index": 25, "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": 26, "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": 27, "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": 28, "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": 29, "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": 30, "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": 31, "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": 32, "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": 33, "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": 34, "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": 35, "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": 36, "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": 37, "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": 38, "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": 39, "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": 40, "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": 41, "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": 42, "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": 43, "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/docs/nats-component.adoc b/components/camel-nats/src/main/docs/nats-component.adoc
index 7ad6f601835e..368adc97304e 100644
--- a/components/camel-nats/src/main/docs/nats-component.adoc
+++ b/components/camel-nats/src/main/docs/nats-component.adoc
@@ -125,5 +125,28 @@ from("nats:mytopic?maxMessages=5&queueName=myqueue")
.to("mock:result");
----
+=== Manual Acknowledgment (JetStream)
+
+When consuming from JetStream, by default messages are automatically acknowledged
+after successful route processing, or negatively acknowledged (redelivered) on failure.
+
+To take full control of acknowledgment, set `manualAck=true` on the consumer endpoint.
+This disables automatic acknowledgment and exposes a `NatsManualAck` object as the
+`CamelNatsManualAck` message header.
+
+[source,java]
+----
+from("nats:mytopic?jetstreamEnabled=true&jetstreamName=mystream&durableName=myconsumer&pullSubscription=false&manualAck=true")
+ .process(exchange -> {
+ // do work ...
+
+ NatsManualAck manualAck = exchange.getIn().getHeader("CamelNatsManualAck", NatsManualAck.class);
+ manualAck.ack();
+ });
+----
+
+If the route completes without calling any NatsManualAck method, the message remains unacknowledged
+and NATS will redeliver it after `ackWait` expires (default 30 seconds).
+
include::spring-boot:partial$starter.adoc[]
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java
new file mode 100644
index 000000000000..7d3b011663ba
--- /dev/null
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java
@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+import java.time.Duration;
+
+import io.nats.client.Message;
+
+class DefaultNatsManualAck implements NatsManualAck {
+
+ private final Message message;
+
+ DefaultNatsManualAck(Message message) {
+ this.message = message;
+ }
+
+ @Override
+ public void ack() {
+ message.ack();
+ }
+
+ @Override
+ public void nak() {
+ message.nak();
+ }
+
+ @Override
+ public void nakWithDelay(Duration delay) {
+ message.nakWithDelay(delay);
+ }
+
+ @Override
+ public void term() {
+ message.term();
+ }
+
+ @Override
+ public void inProgress() {
+ message.inProgress();
+ }
+}
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 ea2fa55136a5..00b489098c7b 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
@@ -83,6 +83,8 @@ public class NatsConfiguration implements Cloneable {
private long nackWait = 5000;
@UriParam(label = "consumer")
private long maxDeliver;
+ @UriParam(label = "consumer", defaultValue = "false")
+ private boolean manualAck;
@UriParam(label = "consumer", defaultValue = "10")
private int poolSize = 10;
@UriParam(label = "common", defaultValue = "true")
@@ -679,4 +681,25 @@ public class NatsConfiguration implements Cloneable {
public void setMaxDeliver(long maxDeliver) {
this.maxDeliver = maxDeliver;
}
+
+ public boolean isManualAck() {
+ return manualAck;
+ }
+
+ /**
+ * Whether to allow doing manual acknowledgment via {@link NatsManualAck}.
+ * <p/>
+ * If this option is enabled then an instance of {@link NatsManualAck} is stored on the
+ * {@link org.apache.camel.Exchange} message header, which allows end users to access this API and perform manual
+ * ack/nak/term operations via the JetStream consumer.
+ * <p/>
+ * When enabled, the automatic acknowledgment on exchange completion is disabled. If the user does not call any ack
+ * method, the message remains unacknowledged and NATS will redeliver it after the ackWait timeout expires.
+ * <p/>
+ * This option is only applicable when JetStream is enabled (jetstreamEnabled=true). It has no effect when
+ * ackPolicy=None since the server acknowledges messages automatically on delivery.
+ */
+ public void setManualAck(boolean manualAck) {
+ this.manualAck = manualAck;
+ }
}
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConstants.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConstants.java
index a897652ca6d8..a0a1c883614f 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConstants.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConstants.java
@@ -38,6 +38,10 @@ public interface NatsConstants {
description = "Number of times this message has been delivered (1 = first, > 1 then message has been redelivered)",
javaType = "long", important = true)
String NATS_DELIVERY_COUNTER = "CamelNatsDeliveryCounter";
+ @Metadata(label = "consumer",
+ description = "The manual acknowledgment handle for JetStream messages (only set when manualAck=true).",
+ javaType = "org.apache.camel.component.nats.NatsManualAck")
+ String NATS_MANUAL_ACK = "CamelNatsManualAck";
String NATS_REQUEST_TIMEOUT_THREAD_PROFILE_NAME = "CamelNatsRequestTimeoutExecutor";
}
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 c36d0be1fdcc..2a1ce6df6113 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
@@ -78,7 +78,15 @@ public class NatsConsumer extends DefaultConsumer {
? this.getEndpoint().getConfiguration().getConnection()
: this.getEndpoint().getConnection();
- this.executor.submit(new NatsConsumingTask(this.connection, this.getEndpoint().getConfiguration()));
+ NatsConfiguration config = this.getEndpoint().getConfiguration();
+ if (config.isManualAck() && !config.isJetstreamEnabled()) {
+ LOG.warn("manualAck=true has no effect without jetstreamEnabled=true; standard NATS has no acknowledgment");
+ }
+ if (config.isManualAck() && config.getAckPolicy() == AckPolicy.None) {
+ LOG.warn(
+ "manualAck=true with ackPolicy=None: the server acknowledges automatically on delivery, manual ack/nak calls will have no effect");
+ }
+ this.executor.submit(new NatsConsumingTask(this.connection, config));
}
@Override
@@ -304,35 +312,38 @@ public class NatsConsumer extends DefaultConsumer {
class CamelNatsMessageHandler implements MessageHandler {
final boolean ackPolicyNone = configuration.getAckPolicy() == AckPolicy.None;
+ final boolean manualAckEnabled = configuration.isManualAck() && configuration.isJetstreamEnabled();
@Override
public void onMessage(Message msg) throws InterruptedException {
LOG.debug("Received Message: {}", msg);
final Exchange exchange = NatsConsumer.this.createExchange(false);
- exchange.getExchangeExtension().addOnCompletion(new SynchronizationAdapter() {
- @Override
- public void onComplete(Exchange exchange) {
- LOG.debug("ACK");
- msg.ack();
- }
-
- @Override
- public void onFailure(Exchange exchange) {
- if (ackPolicyNone) {
- // ACK policy is none which means that we should auto ACK even if the message processed failed in Camel
+ if (!manualAckEnabled) {
+ exchange.getExchangeExtension().addOnCompletion(new SynchronizationAdapter() {
+ @Override
+ public void onComplete(Exchange exchange) {
LOG.debug("ACK");
msg.ack();
- } else {
- LOG.debug("NACK (delay:{})", configuration.getNackWait());
- if (configuration.getNackWait() <= 0) {
- msg.nak();
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ if (ackPolicyNone) {
+ // ACK policy is none which means that we should auto ACK even if the message processed failed in Camel
+ LOG.debug("ACK");
+ msg.ack();
} else {
- msg.nakWithDelay(configuration.getNackWait());
+ LOG.debug("NACK (delay:{})", configuration.getNackWait());
+ if (configuration.getNackWait() <= 0) {
+ msg.nak();
+ } else {
+ msg.nakWithDelay(configuration.getNackWait());
+ }
}
}
- }
- });
+ });
+ }
try {
exchange.getIn().setBody(msg.getData());
exchange.getIn().setHeader(NatsConstants.NATS_REPLY_TO, msg.getReplyTo());
@@ -367,6 +378,10 @@ public class NatsConsumer extends DefaultConsumer {
exchange.getMessage().setPayloadForTrait(MessageTrait.REDELIVERY,
evalRedeliveryMessageTrait(msg, exchange));
+ if (manualAckEnabled) {
+ exchange.getIn().setHeader(NatsConstants.NATS_MANUAL_ACK, new DefaultNatsManualAck(msg));
+ }
+
NatsConsumer.this.processor.process(exchange);
// is there a reply?
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsManualAck.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsManualAck.java
new file mode 100644
index 000000000000..e50ecd463911
--- /dev/null
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsManualAck.java
@@ -0,0 +1,53 @@
+/*
+ * 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;
+
+import java.time.Duration;
+
+/**
+ * Allows manual acknowledgment of JetStream messages when using the NATS consumer with {@code manualAck=true}.
+ *
+ * @see NatsConstants#NATS_MANUAL_ACK
+ */
+public interface NatsManualAck {
+
+ /**
+ * Acknowledge the message.
+ */
+ void ack();
+
+ /**
+ * Negative acknowledge the message. The message will be redelivered immediately.
+ */
+ void nak();
+
+ /**
+ * Negative acknowledge the message with a delay before redelivery.
+ */
+ void nakWithDelay(Duration delay);
+
+ /**
+ * Terminate delivery of this message. The server will stop redelivering it.
+ */
+ void term();
+
+ /**
+ * Signal that processing is still in progress. Resets the {@code ackWait} timer to prevent the server from
+ * redelivering while long-running processing is ongoing.
+ */
+ void inProgress();
+}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerManualAckIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerManualAckIT.java
new file mode 100644
index 000000000000..84ad28d4863f
--- /dev/null
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerManualAckIT.java
@@ -0,0 +1,196 @@
+/*
+ * 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 java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+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.NatsManualAck;
+import org.apache.camel.component.nats.integration.NatsITSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Isolated;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+@Isolated
+public class NatsJetstreamConsumerManualAckIT extends NatsITSupport {
+
+ @EndpointInject("mock:result")
+ protected MockEndpoint mockResultEndpoint;
+
+ @EndpointInject("mock:input")
+ protected MockEndpoint mockInputEndpoint;
+
+ @Test
+ public void testManualAck() throws Exception {
+ mockResultEndpoint.expectedBodiesReceived("Hello World");
+ mockResultEndpoint.expectedHeaderReceived(NatsConstants.NATS_SUBJECT, "mytopic-manualack");
+
+ template.sendBody("direct:send", "Hello World");
+
+ mockResultEndpoint.setAssertPeriod(5000);
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Test
+ public void testManualNakWithDelay() throws Exception {
+ mockResultEndpoint.expectedBodiesReceived("Hello World");
+ mockResultEndpoint.expectedHeaderReceived("counter", 2);
+ mockResultEndpoint.expectedHeaderReceived(NatsConstants.NATS_DELIVERY_COUNTER, 2);
+
+ mockInputEndpoint.expectedMessageCount(2);
+ mockInputEndpoint.message(0).header(NatsConstants.NATS_DELIVERY_COUNTER).isEqualTo(1);
+ mockInputEndpoint.message(1).header(NatsConstants.NATS_DELIVERY_COUNTER).isEqualTo(2);
+
+ template.sendBody("direct:send-nakdelay", "Hello World");
+
+ mockResultEndpoint.setAssertPeriod(5000);
+ mockInputEndpoint.setAssertPeriod(5000);
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Test
+ public void testManualTerm() throws Exception {
+ mockInputEndpoint.expectedMessageCount(1);
+ mockInputEndpoint.expectedHeaderReceived(NatsConstants.NATS_DELIVERY_COUNTER, 1);
+
+ template.sendBody("direct:send-term", "Hello World");
+
+ mockInputEndpoint.setAssertPeriod(5000);
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Test
+ public void testManualNak() throws Exception {
+ mockResultEndpoint.expectedBodiesReceived("Hello World");
+ mockResultEndpoint.expectedHeaderReceived("counter", 2);
+ mockResultEndpoint.expectedHeaderReceived(NatsConstants.NATS_DELIVERY_COUNTER, 2);
+
+ mockInputEndpoint.expectedMessageCount(2);
+ mockInputEndpoint.message(0).header(NatsConstants.NATS_DELIVERY_COUNTER).isEqualTo(1);
+ mockInputEndpoint.message(1).header(NatsConstants.NATS_DELIVERY_COUNTER).isEqualTo(2);
+
+ template.sendBody("direct:send-nak", "Hello World");
+
+ mockResultEndpoint.setAssertPeriod(5000);
+ mockInputEndpoint.setAssertPeriod(5000);
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ String ackUri
+ = "nats:mytopic-manualack?jetstreamEnabled=true&jetstreamName=mystream-manualack&jetstreamAsync=false&durableName=camel-manualack&pullSubscription=false&manualAck=true";
+ String nakUri
+ = "nats:mytopic-manualack-nak?jetstreamEnabled=true&jetstreamName=mystream-manualack-nak&jetstreamAsync=false&durableName=camel-manualack-nak&pullSubscription=false&manualAck=true&nackWait=10";
+ String nakDelayUri
+ = "nats:mytopic-manualack-nakdelay?jetstreamEnabled=true&jetstreamName=mystream-manualack-nakdelay&jetstreamAsync=false&durableName=camel-manualack-nakdelay&pullSubscription=false&manualAck=true";
+ String termUri
+ = "nats:mytopic-manualack-term?jetstreamEnabled=true&jetstreamName=mystream-manualack-term&jetstreamAsync=false&durableName=camel-manualack-term&pullSubscription=false&manualAck=true&nackWait=10&maxDeliver=5";
+
+ from("direct:send")
+ .errorHandler(defaultErrorHandler().maximumRedeliveries(5))
+ .to(ackUri);
+
+ from(ackUri)
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ NatsManualAck manualAck
+ = exchange.getIn().getHeader(NatsConstants.NATS_MANUAL_ACK, NatsManualAck.class);
+ assertNotNull(manualAck);
+ assertInstanceOf(NatsManualAck.class, manualAck);
+ manualAck.ack();
+ }
+ })
+ .to(mockResultEndpoint);
+
+ from("direct:send-nakdelay")
+ .errorHandler(defaultErrorHandler().maximumRedeliveries(5))
+ .to(nakDelayUri);
+
+ final AtomicInteger nakDelayCounter = new AtomicInteger();
+ from(nakDelayUri)
+ .to("mock:input")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ NatsManualAck manualAck
+ = exchange.getIn().getHeader(NatsConstants.NATS_MANUAL_ACK, NatsManualAck.class);
+ if (nakDelayCounter.incrementAndGet() < 2) {
+ manualAck.nakWithDelay(Duration.ofMillis(100));
+ exchange.setRouteStop(true);
+ } else {
+ manualAck.ack();
+ exchange.getMessage().setHeader("counter", nakDelayCounter.intValue());
+ }
+ }
+ })
+ .to(mockResultEndpoint);
+
+ from("direct:send-term")
+ .errorHandler(defaultErrorHandler().maximumRedeliveries(5))
+ .to(termUri);
+
+ from(termUri)
+ .to("mock:input")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ NatsManualAck manualAck
+ = exchange.getIn().getHeader(NatsConstants.NATS_MANUAL_ACK, NatsManualAck.class);
+ manualAck.term();
+ exchange.setRouteStop(true);
+ }
+ });
+
+ from("direct:send-nak")
+ .errorHandler(defaultErrorHandler().maximumRedeliveries(5))
+ .to(nakUri);
+
+ final AtomicInteger counter = new AtomicInteger();
+ from(nakUri)
+ .to("mock:input")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ NatsManualAck manualAck
+ = exchange.getIn().getHeader(NatsConstants.NATS_MANUAL_ACK, NatsManualAck.class);
+ if (counter.incrementAndGet() < 2) {
+ manualAck.nak();
+ exchange.setRouteStop(true);
+ } else {
+ manualAck.ack();
+ exchange.getMessage().setHeader("counter", counter.intValue());
+ }
+ }
+ })
+ .to(mockResultEndpoint);
+ }
+ };
+ }
+}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerManualAckPullIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerManualAckPullIT.java
new file mode 100644
index 000000000000..4360130405c8
--- /dev/null
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerManualAckPullIT.java
@@ -0,0 +1,122 @@
+/*
+ * 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 java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+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.NatsManualAck;
+import org.apache.camel.component.nats.integration.NatsITSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Isolated;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+@Isolated
+public class NatsJetstreamConsumerManualAckPullIT extends NatsITSupport {
+
+ @EndpointInject("mock:result")
+ protected MockEndpoint mockResultEndpoint;
+
+ @EndpointInject("mock:input")
+ protected MockEndpoint mockInputEndpoint;
+
+ @Test
+ public void testManualAck() throws Exception {
+ mockResultEndpoint.expectedBodiesReceived("Hello World");
+ mockResultEndpoint.expectedHeaderReceived(NatsConstants.NATS_SUBJECT, "mytopic-manualack-pull");
+
+ template.sendBody("direct:send", "Hello World");
+
+ mockResultEndpoint.setAssertPeriod(5000);
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Test
+ public void testManualNak() throws Exception {
+ mockResultEndpoint.expectedBodiesReceived("Hello World");
+ mockResultEndpoint.expectedHeaderReceived("counter", 2);
+ mockResultEndpoint.expectedHeaderReceived(NatsConstants.NATS_DELIVERY_COUNTER, 2);
+
+ mockInputEndpoint.expectedMessageCount(2);
+ mockInputEndpoint.message(0).header(NatsConstants.NATS_DELIVERY_COUNTER).isEqualTo(1);
+ mockInputEndpoint.message(1).header(NatsConstants.NATS_DELIVERY_COUNTER).isEqualTo(2);
+
+ template.sendBody("direct:send-nak", "Hello World");
+
+ mockResultEndpoint.setAssertPeriod(5000);
+ mockInputEndpoint.setAssertPeriod(5000);
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ String ackUri
+ = "nats:mytopic-manualack-pull?jetstreamEnabled=true&jetstreamName=mystream-manualack-pull&jetstreamAsync=false&durableName=camel-manualack-pull&pullSubscription=true&pullFetchTimeout=500&manualAck=true";
+ String nakUri
+ = "nats:mytopic-manualack-pull-nak?jetstreamEnabled=true&jetstreamName=mystream-manualack-pull-nak&jetstreamAsync=false&durableName=camel-manualack-pull-nak&pullSubscription=true&pullFetchTimeout=500&manualAck=true&nackWait=10";
+
+ from("direct:send")
+ .errorHandler(defaultErrorHandler().maximumRedeliveries(5))
+ .to(ackUri);
+
+ from(ackUri)
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ NatsManualAck manualAck
+ = exchange.getIn().getHeader(NatsConstants.NATS_MANUAL_ACK, NatsManualAck.class);
+ assertNotNull(manualAck);
+ manualAck.ack();
+ }
+ })
+ .to(mockResultEndpoint);
+
+ from("direct:send-nak")
+ .errorHandler(defaultErrorHandler().maximumRedeliveries(5))
+ .to(nakUri);
+
+ final AtomicInteger counter = new AtomicInteger();
+ from(nakUri)
+ .to("mock:input")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ NatsManualAck manualAck
+ = exchange.getIn().getHeader(NatsConstants.NATS_MANUAL_ACK, NatsManualAck.class);
+ if (counter.incrementAndGet() < 2) {
+ manualAck.nak();
+ exchange.setRouteStop(true);
+ } else {
+ manualAck.ack();
+ exchange.getMessage().setHeader("counter", counter.intValue());
+ }
+ }
+ })
+ .to(mockResultEndpoint);
+ }
+ };
+ }
+}
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 74051188ea21..e7a855382e1c 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
@@ -433,6 +433,33 @@ public interface NatsComponentBuilderFactory {
return this;
}
+
+ /**
+ * Whether to allow doing manual acknowledgment via NatsManualAck. If
+ * this option is enabled then an instance of NatsManualAck is stored on
+ * the org.apache.camel.Exchange message header, which allows end users
+ * to access this API and perform manual ack/nak/term operations via the
+ * JetStream consumer. When enabled, the automatic acknowledgment on
+ * exchange completion is disabled. If the user does not call any ack
+ * method, the message remains unacknowledged and NATS will redeliver it
+ * after the ackWait timeout expires. This option is only applicable
+ * when JetStream is enabled (jetstreamEnabled=true). It has no effect
+ * when ackPolicy=None since the server acknowledges messages
+ * automatically on delivery.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ *
+ * @param manualAck the value to set
+ * @return the dsl builder
+ */
+ default NatsComponentBuilder manualAck(boolean manualAck) {
+ doSetProperty("manualAck", manualAck);
+ return this;
+ }
+
/**
* Maximum number of attempts to deliver a message from Nats to a
* consumer. Once MaxDeliver is reached, the NATS server stops
@@ -875,6 +902,7 @@ public interface NatsComponentBuilderFactory {
case "ackWait": getOrCreateConfiguration((NatsComponent) component).setAckWait((long) value); return true;
case "bridgeErrorHandler": ((NatsComponent) component).setBridgeErrorHandler((boolean) value); return true;
case "durableName": getOrCreateConfiguration((NatsComponent) component).setDurableName((java.lang.String) value); return true;
+ case "manualAck": getOrCreateConfiguration((NatsComponent) component).setManualAck((boolean) value); return true;
case "maxDeliver": getOrCreateConfiguration((NatsComponent) component).setMaxDeliver((long) value); return true;
case "maxMessages": getOrCreateConfiguration((NatsComponent) component).setMaxMessages((java.lang.String) value); return true;
case "nackWait": getOrCreateConfiguration((NatsComponent) component).setNackWait((long) 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 85a1442fb647..ac2af7e2fc0e 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
@@ -605,6 +605,56 @@ public interface NatsEndpointBuilderFactory {
doSetProperty("durableName", durableName);
return this;
}
+ /**
+ * Whether to allow doing manual acknowledgment via NatsManualAck. If
+ * this option is enabled then an instance of NatsManualAck is stored on
+ * the org.apache.camel.Exchange message header, which allows end users
+ * to access this API and perform manual ack/nak/term operations via the
+ * JetStream consumer. When enabled, the automatic acknowledgment on
+ * exchange completion is disabled. If the user does not call any ack
+ * method, the message remains unacknowledged and NATS will redeliver it
+ * after the ackWait timeout expires. This option is only applicable
+ * when JetStream is enabled (jetstreamEnabled=true). It has no effect
+ * when ackPolicy=None since the server acknowledges messages
+ * automatically on delivery.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ *
+ * @param manualAck the value to set
+ * @return the dsl builder
+ */
+ default NatsEndpointConsumerBuilder manualAck(boolean manualAck) {
+ doSetProperty("manualAck", manualAck);
+ return this;
+ }
+ /**
+ * Whether to allow doing manual acknowledgment via NatsManualAck. If
+ * this option is enabled then an instance of NatsManualAck is stored on
+ * the org.apache.camel.Exchange message header, which allows end users
+ * to access this API and perform manual ack/nak/term operations via the
+ * JetStream consumer. When enabled, the automatic acknowledgment on
+ * exchange completion is disabled. If the user does not call any ack
+ * method, the message remains unacknowledged and NATS will redeliver it
+ * after the ackWait timeout expires. This option is only applicable
+ * when JetStream is enabled (jetstreamEnabled=true). It has no effect
+ * when ackPolicy=None since the server acknowledges messages
+ * automatically on delivery.
+ *
+ * The option will be converted to a <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ *
+ * @param manualAck the value to set
+ * @return the dsl builder
+ */
+ default NatsEndpointConsumerBuilder manualAck(String manualAck) {
+ doSetProperty("manualAck", manualAck);
+ return this;
+ }
/**
* Maximum number of attempts to deliver a message from Nats to a
* consumer. Once MaxDeliver is reached, the NATS server stops
@@ -2861,6 +2911,20 @@ public interface NatsEndpointBuilderFactory {
public String natsDeliveryCounter() {
return "CamelNatsDeliveryCounter";
}
+ /**
+ * The manual acknowledgment handle for JetStream messages (only set
+ * when manualAck=true).
+ *
+ * The option is a: {@code
+ * org.apache.camel.component.nats.NatsManualAck} type.
+ *
+ * Group: consumer
+ *
+ * @return the name of the header {@code NatsManualAck}.
+ */
+ public String natsManualAck() {
+ return "CamelNatsManualAck";
+ }
}
static NatsEndpointBuilder endpointBuilder(String componentName, String path) {
class NatsEndpointBuilderImpl extends AbstractEndpointBuilder implements NatsEndpointBuilder, AdvancedNatsEndpointBuilder {
diff --git a/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerAuthService.java b/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerAuthService.java
index 1c8cd672dd48..eb322a363bec 100644
--- a/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerAuthService.java
+++ b/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerAuthService.java
@@ -29,6 +29,7 @@ public class NatsLocalContainerAuthService extends NatsLocalContainerService imp
container
.waitingFor(Wait.forLogMessage(".*Server.*is.*ready.*", 1))
+ .waitingFor(Wait.forListeningPort())
.withCommand("--jetstream", "-DV", "--user", USERNAME, "--pass", PASSWORD);
return container;
diff --git a/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerAuthTokenService.java b/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerAuthTokenService.java
index e1d8a3bc45d0..5f804a61b391 100644
--- a/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerAuthTokenService.java
+++ b/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerAuthTokenService.java
@@ -28,6 +28,7 @@ public class NatsLocalContainerAuthTokenService extends NatsLocalContainerServic
container
.waitingFor(Wait.forLogMessage(".*Server.*is.*ready.*", 1))
+ .waitingFor(Wait.forListeningPort())
.withCommand("--jetstream", "-DV", "-auth", TOKEN);
return container;
diff --git a/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerInfraService.java b/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerInfraService.java
index ab052d5b1540..7fb420363b3f 100644
--- a/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerInfraService.java
+++ b/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerInfraService.java
@@ -55,6 +55,7 @@ public class NatsLocalContainerInfraService implements NatsInfraService, Contain
withNetworkAliases(containerName)
.waitingFor(Wait.forLogMessage(".*Server.*is.*ready.*", 1))
+ .waitingFor(Wait.forListeningPort())
.withCommand("--jetstream");
ContainerEnvironmentUtil.configurePort(this, fixedPort, PORT);
diff --git a/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerTLSAuthService.java b/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerTLSAuthService.java
index 4e165c874db4..24bd5e836b89 100644
--- a/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerTLSAuthService.java
+++ b/test-infra/camel-test-infra-nats/src/main/java/org/apache/camel/test/infra/nats/services/NatsLocalContainerTLSAuthService.java
@@ -28,6 +28,7 @@ public class NatsLocalContainerTLSAuthService extends NatsLocalContainerService
container
.waitingFor(Wait.forLogMessage(".*Server.*is.*ready.*", 1))
+ .waitingFor(Wait.forListeningPort())
.withClasspathResourceMapping("org/apache/camel/test/infra/nats/services", "/nats", BindMode.READ_ONLY)
.withCommand("--jetstream",
"--tls",