You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2021/01/12 09:14:50 UTC
[camel] 01/02: CAMEL-16003: camel-spring-rabbitmq - New component
using spring client
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch rabbitmq
in repository https://gitbox.apache.org/repos/asf/camel.git
commit d8435655a8be1a3cc02eeabb429ef7c32e0751d9
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Jan 12 09:28:37 2021 +0100
CAMEL-16003: camel-spring-rabbitmq - New component using spring client
---
.../catalog/docs/spring-rabbitmq-component.adoc | 6 +-
.../springrabbit/RabbitMQEndpointConfigurer.java | 18 +-
.../springrabbit/RabbitMQEndpointUriFactory.java | 20 ++-
.../component/springrabbit/spring-rabbitmq.json | 6 +-
.../src/main/docs/spring-rabbitmq-component.adoc | 6 +-
.../DefaultListenerContainerFactory.java | 7 +
.../springrabbit/EndpointMessageListener.java | 2 +-
.../component/springrabbit/RabbitMQEndpoint.java | 49 ++++--
.../dsl/RabbitMQEndpointBuilderFactory.java | 195 ++++++++++++---------
.../ROOT/pages/spring-rabbitmq-component.adoc | 6 +-
10 files changed, 204 insertions(+), 111 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/spring-rabbitmq-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/spring-rabbitmq-component.adoc
index 04b35cf..16fa6d5 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/spring-rabbitmq-component.adoc
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/spring-rabbitmq-component.adoc
@@ -91,7 +91,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (23 parameters):
+=== Query Parameters (25 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -101,6 +101,7 @@ with the following path and query parameters:
| *disableReplyTo* (common) | Specifies whether Camel ignores the ReplyTo header in messages. If true, Camel does not send a reply back to the destination specified in the ReplyTo header. You can use this option if you want Camel to consume from a route and you do not want Camel to automatically send back a reply message because another component in your code handles the reply message. You can also use this option if you want to use Camel as a proxy between different message brokers and [...]
| *routingKey* (common) | Routing key. | | String
| *testConnectionOnStartup* (common) | Specifies whether to test the connection on startup. This ensures that when Camel starts that all the JMS consumers have a valid connection to the JMS broker. If a connection cannot be granted then Camel throws an exception on startup. This ensures that Camel is not started with failed connections. The JMS producers is tested as well. | false | boolean
+| *acknowledgeMode* (consumer) | Flag controlling the behaviour of the container with respect to message acknowledgement. The most common usage is to let the container handle the acknowledgements (so the listener doesn't need to know about the channel or the message). Set to AcknowledgeMode.MANUAL if the listener will send the acknowledgements itself using Channel.basicAck(long, boolean). Manual acks are consistent with either a transactional or non-transactional channel, but if you are [...]
| *asyncConsumer* (consumer) | Whether the consumer processes the Exchange asynchronously. If enabled then the consumer may pickup the next message from the queue, while the previous message is being processed asynchronously (by the Asynchronous Routing Engine). This means that messages may be processed not 100% strictly in order. If disabled (as default) then the Exchange is fully processed before the consumer will pickup the next message from the queue. | false | boolean
| *autoDeclare* (consumer) | Specifies whether the consumer should auto declare binding between exchange, queue and routing key when starting. | true | boolean
| *autoStartup* (consumer) | Specifies whether the consumer container should auto-startup. | true | boolean
@@ -110,6 +111,8 @@ with the following path and query parameters:
| *deadLetterQueue* (consumer) | The name of the dead letter queue | | String
| *deadLetterRoutingKey* (consumer) | The routing key for the dead letter exchange | | String
| *exchangeType* (consumer) | The type of the exchange. There are 4 enums and the value can be one of: direct, fanout, headers, topic | direct | String
+| *exclusive* (consumer) | Set to true for an exclusive consumer | false | boolean
+| *noLocal* (consumer) | Set to true for an no-local consumer | false | boolean
| *queues* (consumer) | The queue(s) to use for consuming messages. Multiple queue names can be separated by comma. | | String
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. There are 3 enums and the value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
@@ -119,7 +122,6 @@ with the following path and query parameters:
| *messageConverter* (advanced) | To use a custom MessageConverter so you can be in control how to map to/from a org.springframework.amqp.core.Message. | | MessageConverter
| *messagePropertiesConverter* (advanced) | To use a custom MessagePropertiesConverter so you can be in control how to map to/from a org.springframework.amqp.core.MessageProperties. | | MessagePropertiesConverter
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
-| *transacted* (transaction) | Specifies whether to use transacted mode | false | boolean
|===
// endpoint options: END
diff --git a/components/camel-spring-rabbitmq/src/generated/java/org/apache/camel/component/springrabbit/RabbitMQEndpointConfigurer.java b/components/camel-spring-rabbitmq/src/generated/java/org/apache/camel/component/springrabbit/RabbitMQEndpointConfigurer.java
index 50377af..c44046b 100644
--- a/components/camel-spring-rabbitmq/src/generated/java/org/apache/camel/component/springrabbit/RabbitMQEndpointConfigurer.java
+++ b/components/camel-spring-rabbitmq/src/generated/java/org/apache/camel/component/springrabbit/RabbitMQEndpointConfigurer.java
@@ -21,6 +21,8 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
RabbitMQEndpoint target = (RabbitMQEndpoint) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "acknowledgemode":
+ case "acknowledgeMode": target.setAcknowledgeMode(property(camelContext, org.springframework.amqp.core.AcknowledgeMode.class, value)); return true;
case "args": target.setArgs(property(camelContext, java.util.Map.class, value)); return true;
case "asyncconsumer":
case "asyncConsumer": target.setAsyncConsumer(property(camelContext, boolean.class, value)); return true;
@@ -48,12 +50,15 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
case "exchangePattern": target.setExchangePattern(property(camelContext, org.apache.camel.ExchangePattern.class, value)); return true;
case "exchangetype":
case "exchangeType": target.setExchangeType(property(camelContext, java.lang.String.class, value)); return true;
+ case "exclusive": target.setExclusive(property(camelContext, boolean.class, value)); return true;
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
case "messageconverter":
case "messageConverter": target.setMessageConverter(property(camelContext, org.springframework.amqp.support.converter.MessageConverter.class, value)); return true;
case "messagepropertiesconverter":
case "messagePropertiesConverter": target.setMessagePropertiesConverter(property(camelContext, org.apache.camel.component.springrabbit.MessagePropertiesConverter.class, value)); return true;
+ case "nolocal":
+ case "noLocal": target.setNoLocal(property(camelContext, boolean.class, value)); return true;
case "queues": target.setQueues(property(camelContext, java.lang.String.class, value)); return true;
case "replytimeout":
case "replyTimeout": target.setReplyTimeout(property(camelContext, java.time.Duration.class, value).toMillis()); return true;
@@ -62,7 +67,6 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
case "synchronous": target.setSynchronous(property(camelContext, boolean.class, value)); return true;
case "testconnectiononstartup":
case "testConnectionOnStartup": target.setTestConnectionOnStartup(property(camelContext, boolean.class, value)); return true;
- case "transacted": target.setTransacted(property(camelContext, boolean.class, value)); return true;
default: return false;
}
}
@@ -70,6 +74,8 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
@Override
public Class<?> getOptionType(String name, boolean ignoreCase) {
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "acknowledgemode":
+ case "acknowledgeMode": return org.springframework.amqp.core.AcknowledgeMode.class;
case "args": return java.util.Map.class;
case "asyncconsumer":
case "asyncConsumer": return boolean.class;
@@ -97,12 +103,15 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
case "exchangePattern": return org.apache.camel.ExchangePattern.class;
case "exchangetype":
case "exchangeType": return java.lang.String.class;
+ case "exclusive": return boolean.class;
case "lazystartproducer":
case "lazyStartProducer": return boolean.class;
case "messageconverter":
case "messageConverter": return org.springframework.amqp.support.converter.MessageConverter.class;
case "messagepropertiesconverter":
case "messagePropertiesConverter": return org.apache.camel.component.springrabbit.MessagePropertiesConverter.class;
+ case "nolocal":
+ case "noLocal": return boolean.class;
case "queues": return java.lang.String.class;
case "replytimeout":
case "replyTimeout": return long.class;
@@ -111,7 +120,6 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
case "synchronous": return boolean.class;
case "testconnectiononstartup":
case "testConnectionOnStartup": return boolean.class;
- case "transacted": return boolean.class;
default: return null;
}
}
@@ -120,6 +128,8 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
RabbitMQEndpoint target = (RabbitMQEndpoint) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "acknowledgemode":
+ case "acknowledgeMode": return target.getAcknowledgeMode();
case "args": return target.getArgs();
case "asyncconsumer":
case "asyncConsumer": return target.isAsyncConsumer();
@@ -147,12 +157,15 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
case "exchangePattern": return target.getExchangePattern();
case "exchangetype":
case "exchangeType": return target.getExchangeType();
+ case "exclusive": return target.isExclusive();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
case "messageconverter":
case "messageConverter": return target.getMessageConverter();
case "messagepropertiesconverter":
case "messagePropertiesConverter": return target.getMessagePropertiesConverter();
+ case "nolocal":
+ case "noLocal": return target.isNoLocal();
case "queues": return target.getQueues();
case "replytimeout":
case "replyTimeout": return target.getReplyTimeout();
@@ -161,7 +174,6 @@ public class RabbitMQEndpointConfigurer extends PropertyConfigurerSupport implem
case "synchronous": return target.isSynchronous();
case "testconnectiononstartup":
case "testConnectionOnStartup": return target.isTestConnectionOnStartup();
- case "transacted": return target.isTransacted();
default: return null;
}
}
diff --git a/components/camel-spring-rabbitmq/src/generated/java/org/apache/camel/component/springrabbit/RabbitMQEndpointUriFactory.java b/components/camel-spring-rabbitmq/src/generated/java/org/apache/camel/component/springrabbit/RabbitMQEndpointUriFactory.java
index 6707b6f..8bc18ee 100644
--- a/components/camel-spring-rabbitmq/src/generated/java/org/apache/camel/component/springrabbit/RabbitMQEndpointUriFactory.java
+++ b/components/camel-spring-rabbitmq/src/generated/java/org/apache/camel/component/springrabbit/RabbitMQEndpointUriFactory.java
@@ -20,26 +20,28 @@ public class RabbitMQEndpointUriFactory extends org.apache.camel.support.compone
private static final Set<String> PROPERTY_NAMES;
private static final Set<String> SECRET_PROPERTY_NAMES;
static {
- Set<String> props = new HashSet<>(24);
- props.add("disableReplyTo");
+ Set<String> props = new HashSet<>(26);
props.add("asyncConsumer");
- props.add("deadLetterExchange");
props.add("synchronous");
- props.add("exchangePattern");
props.add("messageConverter");
props.add("exchangeType");
- props.add("transacted");
- props.add("deadLetterExchangeType");
props.add("autoStartup");
+ props.add("noLocal");
+ props.add("bridgeErrorHandler");
+ props.add("deadLetterRoutingKey");
+ props.add("messagePropertiesConverter");
+ props.add("exclusive");
+ props.add("disableReplyTo");
+ props.add("deadLetterExchange");
+ props.add("exchangePattern");
+ props.add("deadLetterExchangeType");
props.add("args");
props.add("lazyStartProducer");
- props.add("bridgeErrorHandler");
props.add("deadLetterQueue");
- props.add("deadLetterRoutingKey");
props.add("connectionFactory");
props.add("testConnectionOnStartup");
+ props.add("acknowledgeMode");
props.add("queues");
- props.add("messagePropertiesConverter");
props.add("exchangeName");
props.add("replyTimeout");
props.add("exceptionHandler");
diff --git a/components/camel-spring-rabbitmq/src/generated/resources/org/apache/camel/component/springrabbit/spring-rabbitmq.json b/components/camel-spring-rabbitmq/src/generated/resources/org/apache/camel/component/springrabbit/spring-rabbitmq.json
index 6eed0e3..83c6251 100644
--- a/components/camel-spring-rabbitmq/src/generated/resources/org/apache/camel/component/springrabbit/spring-rabbitmq.json
+++ b/components/camel-spring-rabbitmq/src/generated/resources/org/apache/camel/component/springrabbit/spring-rabbitmq.json
@@ -47,6 +47,7 @@
"disableReplyTo": { "kind": "parameter", "displayName": "Disable Reply To", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies whether Camel ignores the ReplyTo header in messages. If true, Camel does not send a reply back to the destination specified in the ReplyTo header. You can use this option if you want Camel to consume from a rout [...]
"routingKey": { "kind": "parameter", "displayName": "Routing Key", "group": "common", "label": "common", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Routing key." },
"testConnectionOnStartup": { "kind": "parameter", "displayName": "Test Connection On Startup", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies whether to test the connection on startup. This ensures that when Camel starts that all the JMS consumers have a valid connection to the JMS broker. If a connection cannot be granted then Camel [...]
+ "acknowledgeMode": { "kind": "parameter", "displayName": "Acknowledge Mode", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.springframework.amqp.core.AcknowledgeMode", "enum": [ "NONE", "MANUAL", "AUTO" ], "deprecated": false, "autowired": false, "secret": false, "description": "Flag controlling the behaviour of the container with respect to message acknowledgement. The most common usage is to let the container handle the acknowledgeme [...]
"asyncConsumer": { "kind": "parameter", "displayName": "Async Consumer", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the consumer processes the Exchange asynchronously. If enabled then the consumer may pickup the next message from the queue, while the previous message is being processed asynchronously (by the Asynchronous [...]
"autoDeclare": { "kind": "parameter", "displayName": "Auto Declare", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies whether the consumer should auto declare binding between exchange, queue and routing key when starting." },
"autoStartup": { "kind": "parameter", "displayName": "Auto Startup", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Specifies whether the consumer container should auto-startup." },
@@ -56,6 +57,8 @@
"deadLetterQueue": { "kind": "parameter", "displayName": "Dead Letter Queue", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The name of the dead letter queue" },
"deadLetterRoutingKey": { "kind": "parameter", "displayName": "Dead Letter Routing Key", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The routing key for the dead letter exchange" },
"exchangeType": { "kind": "parameter", "displayName": "Exchange Type", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "enum": [ "direct", "fanout", "headers", "topic" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "direct", "description": "The type of the exchange" },
+ "exclusive": { "kind": "parameter", "displayName": "Exclusive", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Set to true for an exclusive consumer" },
+ "noLocal": { "kind": "parameter", "displayName": "No Local", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Set to true for an no-local consumer" },
"queues": { "kind": "parameter", "displayName": "Queues", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The queue(s) to use for consuming messages. Multiple queue names can be separated by comma." },
"exceptionHandler": { "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 default the con [...]
"exchangePattern": { "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut", "InOptionalOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
@@ -64,7 +67,6 @@
"args": { "kind": "parameter", "displayName": "Args", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, java.lang.Object>", "prefix": "arg.", "multiValue": true, "deprecated": false, "autowired": false, "secret": false, "description": "Specify arguments for configuring the different RabbitMQ concepts, a different prefix is required for each element: arg.exchange. arg.queue. arg.binding. arg.dlq.exchange. arg.dl [...]
"messageConverter": { "kind": "parameter", "displayName": "Message Converter", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.springframework.amqp.support.converter.MessageConverter", "deprecated": false, "autowired": false, "secret": false, "description": "To use a custom MessageConverter so you can be in control how to map to\/from a org.springframework.amqp.core.Message." },
"messagePropertiesConverter": { "kind": "parameter", "displayName": "Message Properties Converter", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.apache.camel.component.springrabbit.MessagePropertiesConverter", "deprecated": false, "autowired": false, "secret": false, "description": "To use a custom MessagePropertiesConverter so you can be in control how to map to\/from a org.springframework.amqp.core.MessageProperties." },
- "synchronous": { "kind": "parameter", "displayName": "Synchronous", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported)." },
- "transacted": { "kind": "parameter", "displayName": "Transacted", "group": "transaction", "label": "transaction", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Specifies whether to use transacted mode" }
+ "synchronous": { "kind": "parameter", "displayName": "Synchronous", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported)." }
}
}
diff --git a/components/camel-spring-rabbitmq/src/main/docs/spring-rabbitmq-component.adoc b/components/camel-spring-rabbitmq/src/main/docs/spring-rabbitmq-component.adoc
index 04b35cf..16fa6d5 100644
--- a/components/camel-spring-rabbitmq/src/main/docs/spring-rabbitmq-component.adoc
+++ b/components/camel-spring-rabbitmq/src/main/docs/spring-rabbitmq-component.adoc
@@ -91,7 +91,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (23 parameters):
+=== Query Parameters (25 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -101,6 +101,7 @@ with the following path and query parameters:
| *disableReplyTo* (common) | Specifies whether Camel ignores the ReplyTo header in messages. If true, Camel does not send a reply back to the destination specified in the ReplyTo header. You can use this option if you want Camel to consume from a route and you do not want Camel to automatically send back a reply message because another component in your code handles the reply message. You can also use this option if you want to use Camel as a proxy between different message brokers and [...]
| *routingKey* (common) | Routing key. | | String
| *testConnectionOnStartup* (common) | Specifies whether to test the connection on startup. This ensures that when Camel starts that all the JMS consumers have a valid connection to the JMS broker. If a connection cannot be granted then Camel throws an exception on startup. This ensures that Camel is not started with failed connections. The JMS producers is tested as well. | false | boolean
+| *acknowledgeMode* (consumer) | Flag controlling the behaviour of the container with respect to message acknowledgement. The most common usage is to let the container handle the acknowledgements (so the listener doesn't need to know about the channel or the message). Set to AcknowledgeMode.MANUAL if the listener will send the acknowledgements itself using Channel.basicAck(long, boolean). Manual acks are consistent with either a transactional or non-transactional channel, but if you are [...]
| *asyncConsumer* (consumer) | Whether the consumer processes the Exchange asynchronously. If enabled then the consumer may pickup the next message from the queue, while the previous message is being processed asynchronously (by the Asynchronous Routing Engine). This means that messages may be processed not 100% strictly in order. If disabled (as default) then the Exchange is fully processed before the consumer will pickup the next message from the queue. | false | boolean
| *autoDeclare* (consumer) | Specifies whether the consumer should auto declare binding between exchange, queue and routing key when starting. | true | boolean
| *autoStartup* (consumer) | Specifies whether the consumer container should auto-startup. | true | boolean
@@ -110,6 +111,8 @@ with the following path and query parameters:
| *deadLetterQueue* (consumer) | The name of the dead letter queue | | String
| *deadLetterRoutingKey* (consumer) | The routing key for the dead letter exchange | | String
| *exchangeType* (consumer) | The type of the exchange. There are 4 enums and the value can be one of: direct, fanout, headers, topic | direct | String
+| *exclusive* (consumer) | Set to true for an exclusive consumer | false | boolean
+| *noLocal* (consumer) | Set to true for an no-local consumer | false | boolean
| *queues* (consumer) | The queue(s) to use for consuming messages. Multiple queue names can be separated by comma. | | String
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. There are 3 enums and the value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
@@ -119,7 +122,6 @@ with the following path and query parameters:
| *messageConverter* (advanced) | To use a custom MessageConverter so you can be in control how to map to/from a org.springframework.amqp.core.Message. | | MessageConverter
| *messagePropertiesConverter* (advanced) | To use a custom MessagePropertiesConverter so you can be in control how to map to/from a org.springframework.amqp.core.MessageProperties. | | MessagePropertiesConverter
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
-| *transacted* (transaction) | Specifies whether to use transacted mode | false | boolean
|===
// endpoint options: END
diff --git a/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/DefaultListenerContainerFactory.java b/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/DefaultListenerContainerFactory.java
index 94daa91..1566cc6 100644
--- a/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/DefaultListenerContainerFactory.java
+++ b/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/DefaultListenerContainerFactory.java
@@ -20,6 +20,9 @@ import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+/**
+ * Default {@link ListenerContainerFactory}.
+ */
public class DefaultListenerContainerFactory implements ListenerContainerFactory {
@Override
@@ -28,6 +31,9 @@ public class DefaultListenerContainerFactory implements ListenerContainerFactory
if (endpoint.getQueues() != null) {
listener.setQueueNames(endpoint.getQueues().split(","));
}
+ listener.setAcknowledgeMode(endpoint.getAcknowledgeMode());
+ listener.setExclusive(endpoint.isExclusive());
+ listener.setNoLocal(endpoint.isNoLocal());
AmqpAdmin admin = endpoint.getComponent().getAmqpAdmin();
if (endpoint.isAutoDeclare() && admin == null) {
@@ -42,4 +48,5 @@ public class DefaultListenerContainerFactory implements ListenerContainerFactory
listener.setShutdownTimeout(endpoint.getComponent().getShutdownTimeout());
return listener;
}
+
}
diff --git a/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/EndpointMessageListener.java b/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/EndpointMessageListener.java
index c43344c..ed15da2 100644
--- a/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/EndpointMessageListener.java
+++ b/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/EndpointMessageListener.java
@@ -105,7 +105,7 @@ public class EndpointMessageListener implements ChannelAwareMessageListener {
// (eg to not consume the next message before the previous has been fully processed)
// but if end user explicit configure consumerAsync=true, then we can process the message
// asynchronously (unless endpoint has been configured synchronous, or we use transaction)
- boolean forceSync = endpoint.isSynchronous() || endpoint.isTransacted();
+ boolean forceSync = endpoint.isSynchronous();
if (forceSync || !isAsync()) {
// must process synchronous if transacted or configured to do so
if (LOG.isTraceEnabled()) {
diff --git a/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/RabbitMQEndpoint.java b/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/RabbitMQEndpoint.java
index b3d5b60..f573e77 100644
--- a/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/RabbitMQEndpoint.java
+++ b/components/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/RabbitMQEndpoint.java
@@ -33,6 +33,7 @@ import org.apache.camel.support.DefaultEndpoint;
import org.apache.camel.util.PropertiesHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.ExchangeBuilder;
@@ -93,10 +94,6 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
private boolean asyncConsumer;
@UriParam(label = "common", description = "Routing key.")
private String routingKey;
- // Transaction related configuration
- @UriParam(label = "transaction",
- description = "Specifies whether to use transacted mode")
- private boolean transacted;
@UriParam(description = "Specifies whether to test the connection on startup."
+ " This ensures that when Camel starts that all the JMS consumers have a valid connection to the JMS broker."
+ " If a connection cannot be granted then Camel throws an exception on startup."
@@ -114,6 +111,18 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
+ " arg.exchange. arg.queue. arg.binding. arg.dlq.exchange. arg.dlq.queue. arg.dlq.binding."
+ " For example to declare a queue with message ttl argument: args=arg.queue.x-message-ttl=60000")
private Map<String, Object> args;
+ @UriParam(label = "consumer",
+ description = "Flag controlling the behaviour of the container with respect to message acknowledgement. The most common usage is to let the container handle the acknowledgements"
+ + " (so the listener doesn't need to know about the channel or the message)."
+ + " Set to AcknowledgeMode.MANUAL if the listener will send the acknowledgements itself using Channel.basicAck(long, boolean). Manual acks are consistent with either a transactional or non-transactional channel,"
+ + " but if you are doing no other work on the channel at the same other than receiving a single message then the transaction is probably unnecessary."
+ + " Set to AcknowledgeMode.NONE to tell the broker not to expect any acknowledgements, and it will assume all messages are acknowledged as soon as they are sent (this is autoack in native Rabbit broker terms)."
+ + " If AcknowledgeMode.NONE then the channel cannot be transactional (so the container will fail on start up if that flag is accidentally set).")
+ private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
+ @UriParam(label = "consumer", description = "Set to true for an exclusive consumer")
+ private boolean exclusive;
+ @UriParam(label = "consumer", description = "Set to true for an no-local consumer")
+ private boolean noLocal;
@UriParam(label = "consumer", description = "The name of the dead letter exchange")
private String deadLetterExchange;
@UriParam(label = "consumer", description = "The name of the dead letter queue")
@@ -164,14 +173,6 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
this.queues = queues;
}
- public boolean isTransacted() {
- return transacted;
- }
-
- public void setTransacted(boolean transacted) {
- this.transacted = transacted;
- }
-
public boolean isAutoStartup() {
return autoStartup;
}
@@ -244,6 +245,30 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
this.args = args;
}
+ public AcknowledgeMode getAcknowledgeMode() {
+ return acknowledgeMode;
+ }
+
+ public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
+ this.acknowledgeMode = acknowledgeMode;
+ }
+
+ public boolean isExclusive() {
+ return exclusive;
+ }
+
+ public void setExclusive(boolean exclusive) {
+ this.exclusive = exclusive;
+ }
+
+ public boolean isNoLocal() {
+ return noLocal;
+ }
+
+ public void setNoLocal(boolean noLocal) {
+ this.noLocal = noLocal;
+ }
+
public String getDeadLetterExchange() {
return deadLetterExchange;
}
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/RabbitMQEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/RabbitMQEndpointBuilderFactory.java
index eb19436..f5361a3 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/RabbitMQEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/RabbitMQEndpointBuilderFactory.java
@@ -179,6 +179,66 @@ public interface RabbitMQEndpointBuilderFactory {
return this;
}
/**
+ * Flag controlling the behaviour of the container with respect to
+ * message acknowledgement. The most common usage is to let the
+ * container handle the acknowledgements (so the listener doesn't need
+ * to know about the channel or the message). Set to
+ * AcknowledgeMode.MANUAL if the listener will send the acknowledgements
+ * itself using Channel.basicAck(long, boolean). Manual acks are
+ * consistent with either a transactional or non-transactional channel,
+ * but if you are doing no other work on the channel at the same other
+ * than receiving a single message then the transaction is probably
+ * unnecessary. Set to AcknowledgeMode.NONE to tell the broker not to
+ * expect any acknowledgements, and it will assume all messages are
+ * acknowledged as soon as they are sent (this is autoack in native
+ * Rabbit broker terms). If AcknowledgeMode.NONE then the channel cannot
+ * be transactional (so the container will fail on start up if that flag
+ * is accidentally set).
+ *
+ * The option is a:
+ * <code>org.springframework.amqp.core.AcknowledgeMode</code> type.
+ *
+ * Group: consumer
+ *
+ * @param acknowledgeMode the value to set
+ * @return the dsl builder
+ */
+ default RabbitMQEndpointConsumerBuilder acknowledgeMode(
+ AcknowledgeMode acknowledgeMode) {
+ doSetProperty("acknowledgeMode", acknowledgeMode);
+ return this;
+ }
+ /**
+ * Flag controlling the behaviour of the container with respect to
+ * message acknowledgement. The most common usage is to let the
+ * container handle the acknowledgements (so the listener doesn't need
+ * to know about the channel or the message). Set to
+ * AcknowledgeMode.MANUAL if the listener will send the acknowledgements
+ * itself using Channel.basicAck(long, boolean). Manual acks are
+ * consistent with either a transactional or non-transactional channel,
+ * but if you are doing no other work on the channel at the same other
+ * than receiving a single message then the transaction is probably
+ * unnecessary. Set to AcknowledgeMode.NONE to tell the broker not to
+ * expect any acknowledgements, and it will assume all messages are
+ * acknowledged as soon as they are sent (this is autoack in native
+ * Rabbit broker terms). If AcknowledgeMode.NONE then the channel cannot
+ * be transactional (so the container will fail on start up if that flag
+ * is accidentally set).
+ *
+ * The option will be converted to a
+ * <code>org.springframework.amqp.core.AcknowledgeMode</code> type.
+ *
+ * Group: consumer
+ *
+ * @param acknowledgeMode the value to set
+ * @return the dsl builder
+ */
+ default RabbitMQEndpointConsumerBuilder acknowledgeMode(
+ String acknowledgeMode) {
+ doSetProperty("acknowledgeMode", acknowledgeMode);
+ return this;
+ }
+ /**
* Whether the consumer processes the Exchange asynchronously. If
* enabled then the consumer may pickup the next message from the queue,
* while the previous message is being processed asynchronously (by the
@@ -409,49 +469,80 @@ public interface RabbitMQEndpointBuilderFactory {
return this;
}
/**
- * The queue(s) to use for consuming messages. Multiple queue names can
- * be separated by comma.
+ * Set to true for an exclusive consumer.
*
- * The option is a: <code>java.lang.String</code> type.
+ * The option is a: <code>boolean</code> type.
*
+ * Default: false
* Group: consumer
*
- * @param queues the value to set
+ * @param exclusive the value to set
* @return the dsl builder
*/
- default RabbitMQEndpointConsumerBuilder queues(String queues) {
- doSetProperty("queues", queues);
+ default RabbitMQEndpointConsumerBuilder exclusive(boolean exclusive) {
+ doSetProperty("exclusive", exclusive);
+ return this;
+ }
+ /**
+ * Set to true for an exclusive consumer.
+ *
+ * The option will be converted to a <code>boolean</code>
+ * type.
+ *
+ * Default: false
+ * Group: consumer
+ *
+ * @param exclusive the value to set
+ * @return the dsl builder
+ */
+ default RabbitMQEndpointConsumerBuilder exclusive(String exclusive) {
+ doSetProperty("exclusive", exclusive);
return this;
}
/**
- * Specifies whether to use transacted mode.
+ * Set to true for an no-local consumer.
*
* The option is a: <code>boolean</code> type.
*
* Default: false
- * Group: transaction
+ * Group: consumer
*
- * @param transacted the value to set
+ * @param noLocal the value to set
* @return the dsl builder
*/
- default RabbitMQEndpointConsumerBuilder transacted(boolean transacted) {
- doSetProperty("transacted", transacted);
+ default RabbitMQEndpointConsumerBuilder noLocal(boolean noLocal) {
+ doSetProperty("noLocal", noLocal);
return this;
}
/**
- * Specifies whether to use transacted mode.
+ * Set to true for an no-local consumer.
*
* The option will be converted to a <code>boolean</code>
* type.
*
* Default: false
- * Group: transaction
+ * Group: consumer
*
- * @param transacted the value to set
+ * @param noLocal the value to set
* @return the dsl builder
*/
- default RabbitMQEndpointConsumerBuilder transacted(String transacted) {
- doSetProperty("transacted", transacted);
+ default RabbitMQEndpointConsumerBuilder noLocal(String noLocal) {
+ doSetProperty("noLocal", noLocal);
+ return this;
+ }
+ /**
+ * The queue(s) to use for consuming messages. Multiple queue names can
+ * be separated by comma.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ *
+ * @param queues the value to set
+ * @return the dsl builder
+ */
+ default RabbitMQEndpointConsumerBuilder queues(String queues) {
+ doSetProperty("queues", queues);
return this;
}
}
@@ -916,37 +1007,6 @@ public interface RabbitMQEndpointBuilderFactory {
doSetProperty("replyTimeout", replyTimeout);
return this;
}
- /**
- * Specifies whether to use transacted mode.
- *
- * The option is a: <code>boolean</code> type.
- *
- * Default: false
- * Group: transaction
- *
- * @param transacted the value to set
- * @return the dsl builder
- */
- default RabbitMQEndpointProducerBuilder transacted(boolean transacted) {
- doSetProperty("transacted", transacted);
- return this;
- }
- /**
- * Specifies whether to use transacted mode.
- *
- * The option will be converted to a <code>boolean</code>
- * type.
- *
- * Default: false
- * Group: transaction
- *
- * @param transacted the value to set
- * @return the dsl builder
- */
- default RabbitMQEndpointProducerBuilder transacted(String transacted) {
- doSetProperty("transacted", transacted);
- return this;
- }
}
/**
@@ -1255,37 +1315,6 @@ public interface RabbitMQEndpointBuilderFactory {
doSetProperty("testConnectionOnStartup", testConnectionOnStartup);
return this;
}
- /**
- * Specifies whether to use transacted mode.
- *
- * The option is a: <code>boolean</code> type.
- *
- * Default: false
- * Group: transaction
- *
- * @param transacted the value to set
- * @return the dsl builder
- */
- default RabbitMQEndpointBuilder transacted(boolean transacted) {
- doSetProperty("transacted", transacted);
- return this;
- }
- /**
- * Specifies whether to use transacted mode.
- *
- * The option will be converted to a <code>boolean</code>
- * type.
- *
- * Default: false
- * Group: transaction
- *
- * @param transacted the value to set
- * @return the dsl builder
- */
- default RabbitMQEndpointBuilder transacted(String transacted) {
- doSetProperty("transacted", transacted);
- return this;
- }
}
/**
@@ -1446,6 +1475,16 @@ public interface RabbitMQEndpointBuilderFactory {
}
}
+ /**
+ * Proxy enum for <code>org.springframework.amqp.core.AcknowledgeMode</code>
+ * enum.
+ */
+ enum AcknowledgeMode {
+ NONE,
+ MANUAL,
+ AUTO;
+ }
+
public interface RabbitMQBuilders {
/**
* Spring RabbitMQ (camel-spring-rabbitmq)
diff --git a/docs/components/modules/ROOT/pages/spring-rabbitmq-component.adoc b/docs/components/modules/ROOT/pages/spring-rabbitmq-component.adoc
index 1f36e4a..614144a 100644
--- a/docs/components/modules/ROOT/pages/spring-rabbitmq-component.adoc
+++ b/docs/components/modules/ROOT/pages/spring-rabbitmq-component.adoc
@@ -93,7 +93,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (23 parameters):
+=== Query Parameters (25 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -103,6 +103,7 @@ with the following path and query parameters:
| *disableReplyTo* (common) | Specifies whether Camel ignores the ReplyTo header in messages. If true, Camel does not send a reply back to the destination specified in the ReplyTo header. You can use this option if you want Camel to consume from a route and you do not want Camel to automatically send back a reply message because another component in your code handles the reply message. You can also use this option if you want to use Camel as a proxy between different message brokers and [...]
| *routingKey* (common) | Routing key. | | String
| *testConnectionOnStartup* (common) | Specifies whether to test the connection on startup. This ensures that when Camel starts that all the JMS consumers have a valid connection to the JMS broker. If a connection cannot be granted then Camel throws an exception on startup. This ensures that Camel is not started with failed connections. The JMS producers is tested as well. | false | boolean
+| *acknowledgeMode* (consumer) | Flag controlling the behaviour of the container with respect to message acknowledgement. The most common usage is to let the container handle the acknowledgements (so the listener doesn't need to know about the channel or the message). Set to AcknowledgeMode.MANUAL if the listener will send the acknowledgements itself using Channel.basicAck(long, boolean). Manual acks are consistent with either a transactional or non-transactional channel, but if you are [...]
| *asyncConsumer* (consumer) | Whether the consumer processes the Exchange asynchronously. If enabled then the consumer may pickup the next message from the queue, while the previous message is being processed asynchronously (by the Asynchronous Routing Engine). This means that messages may be processed not 100% strictly in order. If disabled (as default) then the Exchange is fully processed before the consumer will pickup the next message from the queue. | false | boolean
| *autoDeclare* (consumer) | Specifies whether the consumer should auto declare binding between exchange, queue and routing key when starting. | true | boolean
| *autoStartup* (consumer) | Specifies whether the consumer container should auto-startup. | true | boolean
@@ -112,6 +113,8 @@ with the following path and query parameters:
| *deadLetterQueue* (consumer) | The name of the dead letter queue | | String
| *deadLetterRoutingKey* (consumer) | The routing key for the dead letter exchange | | String
| *exchangeType* (consumer) | The type of the exchange. There are 4 enums and the value can be one of: direct, fanout, headers, topic | direct | String
+| *exclusive* (consumer) | Set to true for an exclusive consumer | false | boolean
+| *noLocal* (consumer) | Set to true for an no-local consumer | false | boolean
| *queues* (consumer) | The queue(s) to use for consuming messages. Multiple queue names can be separated by comma. | | String
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. There are 3 enums and the value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
@@ -121,7 +124,6 @@ with the following path and query parameters:
| *messageConverter* (advanced) | To use a custom MessageConverter so you can be in control how to map to/from a org.springframework.amqp.core.Message. | | MessageConverter
| *messagePropertiesConverter* (advanced) | To use a custom MessagePropertiesConverter so you can be in control how to map to/from a org.springframework.amqp.core.MessageProperties. | | MessagePropertiesConverter
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
-| *transacted* (transaction) | Specifies whether to use transacted mode | false | boolean
|===
// endpoint options: END