You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2017/08/25 12:02:50 UTC
[1/3] camel git commit: CAMEL-11704: Camel-RabbitMQ: Allow passive
queue declaration
Repository: camel
Updated Branches:
refs/heads/master 18f42eaa9 -> 0c55b9c60
CAMEL-11704: Camel-RabbitMQ: Allow passive queue declaration
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a651701e
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a651701e
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a651701e
Branch: refs/heads/master
Commit: a651701eee0f1587d69526fd47d8f96725dd0c1d
Parents: 18f42ea
Author: Christoph Schachinger <ch...@armstrongconsulting.net>
Authored: Fri Aug 25 10:56:07 2017 +0200
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Fri Aug 25 13:55:27 2017 +0200
----------------------------------------------------------------------
.../rabbitmq/RabbitMQDeclareSupport.java | 12 +++++++--
.../component/rabbitmq/RabbitMQEndpoint.java | 28 +++++++++++++++-----
.../rabbitmq/RabbitMQEndpointTest.java | 18 ++++++++-----
3 files changed, 43 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a651701e/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java
index b38eae9..a68c521 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java
@@ -100,7 +100,10 @@ public class RabbitMQDeclareSupport {
}
private void declareExchange(final Channel channel, final String exchange, final String exchangeType, final Map<String, Object> exchangeArgs) throws IOException {
- channel.exchangeDeclare(exchange, exchangeType, endpoint.isDurable(), endpoint.isAutoDelete(), exchangeArgs);
+ if (endpoint.isPassive())
+ channel.exchangeDeclarePassive(exchange);
+ else
+ channel.exchangeDeclare(exchange, exchangeType, endpoint.isDurable(), endpoint.isAutoDelete(), exchangeArgs);
}
private void declareAndBindQueue(final Channel channel,
@@ -111,7 +114,12 @@ public class RabbitMQDeclareSupport {
final Map<String, Object> bindingArgs)
throws IOException {
- channel.queueDeclare(queue, endpoint.isDurable(), endpoint.isExclusive(), endpoint.isAutoDelete(), queueArgs);
+
+ if (endpoint.isPassive())
+ channel.queueDeclarePassive(queue);
+ else
+ channel.queueDeclare(queue, endpoint.isDurable(), endpoint.isExclusive(), endpoint.isAutoDelete(), queueArgs);
+
if (shouldBindQueue()) {
channel.queueBind(queue, exchange, emptyIfNull(routingKey), bindingArgs);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/a651701e/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
index ddfdfe8..5e2be3d 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
@@ -27,13 +27,6 @@ import java.util.concurrent.TimeoutException;
import javax.net.ssl.TrustManager;
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Address;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.Envelope;
-
import org.apache.camel.AsyncEndpoint;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
@@ -46,6 +39,13 @@ import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Address;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.Envelope;
+
/**
* The rabbitmq component allows you produce and consume messages from <a href="http://www.rabbitmq.com/">RabbitMQ</a> instances.
*/
@@ -77,6 +77,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
private boolean durable = true;
@UriParam(label = "common", defaultValue = "false")
private boolean exclusive;
+ @UriParam(label = "common", defaultValue = "false")
+ private boolean passive = false;
@UriParam(label = "producer")
private boolean bridgeEndpoint;
@UriParam(label = "common")
@@ -958,5 +960,17 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
public void setExclusive(boolean exclusive) {
this.exclusive = exclusive;
}
+
+ public boolean isPassive() {
+ return passive;
+ }
+
+ /**
+ * Passive queues depend on the queue already to be available at RabbitMQ.
+ */
+ public void setPassive(boolean passive) {
+ this.passive = passive;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/a651701e/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
index 6b7d737..4ad3257 100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
@@ -27,18 +27,18 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeoutException;
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Address;
-import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.Envelope;
-import com.rabbitmq.client.impl.LongStringHelper;
-
import org.apache.camel.Exchange;
import org.apache.camel.impl.JndiRegistry;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;
import org.mockito.Mockito;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Address;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.impl.LongStringHelper;
+
public class RabbitMQEndpointTest extends CamelTestSupport {
private Envelope envelope = Mockito.mock(Envelope.class);
@@ -355,4 +355,10 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?exclusive=true", RabbitMQEndpoint.class);
assertTrue(endpoint.isExclusive());
}
+
+ @Test
+ public void createEndpointWithPassiveEnabled() throws Exception {
+ RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?passive=true", RabbitMQEndpoint.class);
+ assertTrue(endpoint.isPassive());
+ }
}
[2/3] camel git commit: CAMEL-11704: Fixed CS
Posted by ac...@apache.org.
CAMEL-11704: Fixed CS
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a734ce3c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a734ce3c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a734ce3c
Branch: refs/heads/master
Commit: a734ce3cb541ea3fa0cc59ea033fff887a3fdb8a
Parents: a651701
Author: Andrea Cosentino <an...@gmail.com>
Authored: Fri Aug 25 14:00:19 2017 +0200
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Fri Aug 25 14:00:19 2017 +0200
----------------------------------------------------------------------
.../rabbitmq/RabbitMQDeclareSupport.java | 34 ++--
.../component/rabbitmq/RabbitMQEndpoint.java | 191 +++++++++++--------
.../rabbitmq/RabbitMQEndpointTest.java | 49 ++---
3 files changed, 149 insertions(+), 125 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a734ce3c/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java
index a68c521..90e993e 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQDeclareSupport.java
@@ -39,7 +39,7 @@ public class RabbitMQDeclareSupport {
private void declareAndBindDeadLetterExchangeWithQueue(final Channel channel) throws IOException {
if (endpoint.getDeadLetterExchange() != null) {
// TODO Do we need to setup the args for the DeadLetter?
- declareExchange(channel, endpoint.getDeadLetterExchange(), endpoint.getDeadLetterExchangeType(), Collections.<String, Object>emptyMap());
+ declareExchange(channel, endpoint.getDeadLetterExchange(), endpoint.getDeadLetterExchangeType(), Collections.<String, Object> emptyMap());
declareAndBindQueue(channel, endpoint.getDeadLetterQueue(), endpoint.getDeadLetterExchange(), endpoint.getDeadLetterRoutingKey(), null, null);
}
}
@@ -50,7 +50,8 @@ public class RabbitMQDeclareSupport {
}
if (shouldDeclareQueue()) {
- // need to make sure the queueDeclare is same with the exchange declare
+ // need to make sure the queueDeclare is same with the exchange
+ // declare
declareAndBindQueue(channel, endpoint.getQueue(), endpoint.getExchangeName(), endpoint.getRoutingKey(), resolvedQueueArguments(), endpoint.getBindingArgs());
}
}
@@ -100,26 +101,23 @@ public class RabbitMQDeclareSupport {
}
private void declareExchange(final Channel channel, final String exchange, final String exchangeType, final Map<String, Object> exchangeArgs) throws IOException {
- if (endpoint.isPassive())
- channel.exchangeDeclarePassive(exchange);
- else
- channel.exchangeDeclare(exchange, exchangeType, endpoint.isDurable(), endpoint.isAutoDelete(), exchangeArgs);
+ if (endpoint.isPassive()) {
+ channel.exchangeDeclarePassive(exchange);
+ } else {
+ channel.exchangeDeclare(exchange, exchangeType, endpoint.isDurable(), endpoint.isAutoDelete(), exchangeArgs);
+ }
}
- private void declareAndBindQueue(final Channel channel,
- final String queue,
- final String exchange,
- final String routingKey,
- final Map<String, Object> queueArgs,
+ private void declareAndBindQueue(final Channel channel, final String queue, final String exchange, final String routingKey, final Map<String, Object> queueArgs,
final Map<String, Object> bindingArgs)
- throws IOException {
-
- if (endpoint.isPassive())
- channel.queueDeclarePassive(queue);
- else
- channel.queueDeclare(queue, endpoint.isDurable(), endpoint.isExclusive(), endpoint.isAutoDelete(), queueArgs);
-
+ throws IOException {
+
+ if (endpoint.isPassive()) {
+ channel.queueDeclarePassive(queue);
+ } else {
+ channel.queueDeclare(queue, endpoint.isDurable(), endpoint.isExclusive(), endpoint.isAutoDelete(), queueArgs);
+ }
if (shouldBindQueue()) {
channel.queueBind(queue, exchange, emptyIfNull(routingKey), bindingArgs);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/a734ce3c/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
index 5e2be3d..e3cb28d 100644
--- a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
+++ b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQEndpoint.java
@@ -27,6 +27,13 @@ import java.util.concurrent.TimeoutException;
import javax.net.ssl.TrustManager;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Address;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.Envelope;
+
import org.apache.camel.AsyncEndpoint;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
@@ -39,27 +46,23 @@ import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Address;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.Envelope;
-
/**
- * The rabbitmq component allows you produce and consume messages from <a href="http://www.rabbitmq.com/">RabbitMQ</a> instances.
+ * The rabbitmq component allows you produce and consume messages from
+ * <a href="http://www.rabbitmq.com/">RabbitMQ</a> instances.
*/
-@UriEndpoint(firstVersion = "2.12.0", scheme = "rabbitmq", title = "RabbitMQ", syntax = "rabbitmq:hostname:portNumber/exchangeName",
- consumerClass = RabbitMQConsumer.class, label = "messaging")
+@UriEndpoint(firstVersion = "2.12.0", scheme = "rabbitmq", title = "RabbitMQ", syntax = "rabbitmq:hostname:portNumber/exchangeName", consumerClass = RabbitMQConsumer.class, label = "messaging")
public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
// header to indicate that the message body needs to be de-serialized
public static final String SERIALIZE_HEADER = "CamelSerialize";
- @UriPath @Metadata(required = "true")
+ @UriPath
+ @Metadata(required = "true")
private String hostname;
- @UriPath(defaultValue = "5672") @Metadata(required = "true")
+ @UriPath(defaultValue = "5672")
+ @Metadata(required = "true")
private int portNumber;
- @UriPath @Metadata(required = "true")
+ @UriPath
+ @Metadata(required = "true")
private String exchangeName;
@UriParam(label = "security", defaultValue = ConnectionFactory.DEFAULT_USER, secret = true)
private String username = ConnectionFactory.DEFAULT_USER;
@@ -78,7 +81,7 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
@UriParam(label = "common", defaultValue = "false")
private boolean exclusive;
@UriParam(label = "common", defaultValue = "false")
- private boolean passive = false;
+ private boolean passive;
@UriParam(label = "producer")
private boolean bridgeEndpoint;
@UriParam(label = "common")
@@ -174,11 +177,14 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
private long publisherAcknowledgementsTimeout;
@UriParam(label = "producer")
private boolean guaranteedDeliveries;
- // camel-jms supports this setting but it is not currently configurable in camel-rabbitmq
+ // camel-jms supports this setting but it is not currently configurable in
+ // camel-rabbitmq
private boolean useMessageIDAsCorrelationID = true;
- // camel-jms supports this setting but it is not currently configurable in camel-rabbitmq
+ // camel-jms supports this setting but it is not currently configurable in
+ // camel-rabbitmq
private String replyToType = ReplyToType.Temporary.name();
- // camel-jms supports this setting but it is not currently configurable in camel-rabbitmq
+ // camel-jms supports this setting but it is not currently configurable in
+ // camel-rabbitmq
private String replyTo;
private final RabbitMQMessageConverter messageConverter = new RabbitMQMessageConverter();
@@ -222,7 +228,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
*/
protected String getExchangeName(Message msg) {
String exchangeName = msg.getHeader(RabbitMQConstants.EXCHANGE_NAME, String.class);
- // If it is BridgeEndpoint we should ignore the message header of EXCHANGE_NAME
+ // If it is BridgeEndpoint we should ignore the message header of
+ // EXCHANGE_NAME
if (exchangeName == null || isBridgeEndpoint()) {
exchangeName = getExchangeName();
}
@@ -325,7 +332,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * The consumer uses a Thread Pool Executor with a fixed number of threads. This setting allows you to set that number of threads.
+ * The consumer uses a Thread Pool Executor with a fixed number of threads.
+ * This setting allows you to set that number of threads.
*/
public void setThreadPoolSize(int threadPoolSize) {
this.threadPoolSize = threadPoolSize;
@@ -336,7 +344,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Port number for the host with the running rabbitmq instance or cluster. Default value is 5672.
+ * Port number for the host with the running rabbitmq instance or cluster.
+ * Default value is 5672.
*/
public void setPortNumber(int portNumber) {
this.portNumber = portNumber;
@@ -369,7 +378,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * If we are declaring a durable exchange (the exchange will survive a server restart)
+ * If we are declaring a durable exchange (the exchange will survive a
+ * server restart)
*/
public void setDurable(boolean durable) {
this.durable = durable;
@@ -391,8 +401,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * The exchange name determines which exchange produced messages will sent to.
- * In the case of consumers, the exchange name determines which exchange the queue will bind to.
+ * The exchange name determines which exchange produced messages will sent
+ * to. In the case of consumers, the exchange name determines which exchange
+ * the queue will bind to.
*/
public void setExchangeName(String exchangeName) {
this.exchangeName = exchangeName;
@@ -414,16 +425,16 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * The routing key to use when binding a consumer queue to the exchange.
- * For producer routing keys, you set the header rabbitmq.ROUTING_KEY.
+ * The routing key to use when binding a consumer queue to the exchange. For
+ * producer routing keys, you set the header rabbitmq.ROUTING_KEY.
*/
public void setRoutingKey(String routingKey) {
this.routingKey = routingKey;
}
/**
- * If true the producer will not declare and bind a queue.
- * This can be used for directing messages via an existing routing key.
+ * If true the producer will not declare and bind a queue. This can be used
+ * for directing messages via an existing routing key.
*/
public void setSkipQueueDeclare(boolean skipQueueDeclare) {
this.skipQueueDeclare = skipQueueDeclare;
@@ -435,6 +446,7 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
/**
* If true the queue will not be bound to the exchange after declaring it
+ *
* @return
*/
public boolean isSkipQueueBind() {
@@ -457,7 +469,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * If the bridgeEndpoint is true, the producer will ignore the message header of "rabbitmq.EXCHANGE_NAME" and "rabbitmq.ROUTING_KEY"
+ * If the bridgeEndpoint is true, the producer will ignore the message
+ * header of "rabbitmq.EXCHANGE_NAME" and "rabbitmq.ROUTING_KEY"
*/
public void setBridgeEndpoint(boolean bridgeEndpoint) {
this.bridgeEndpoint = bridgeEndpoint;
@@ -468,8 +481,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * If this option is set, camel-rabbitmq will try to create connection based on the setting of option addresses.
- * The addresses value is a string which looks like "server1:12345, server2:12345"
+ * If this option is set, camel-rabbitmq will try to create connection based
+ * on the setting of option addresses. The addresses value is a string which
+ * looks like "server1:12345, server2:12345"
*/
public void setAddresses(String addresses) {
Address[] addressArray = Address.parseAddresses(addresses);
@@ -542,8 +556,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * To use a custom RabbitMQ connection factory.
- * When this option is set, all connection options (connectionTimeout, requestedChannelMax...) set on URI are not used
+ * To use a custom RabbitMQ connection factory. When this option is set, all
+ * connection options (connectionTimeout, requestedChannelMax...) set on URI
+ * are not used
*/
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
@@ -554,7 +569,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Configure SSL trust manager, SSL should be enabled for this option to be effective
+ * Configure SSL trust manager, SSL should be enabled for this option to be
+ * effective
*/
public void setTrustManager(TrustManager trustManager) {
this.trustManager = trustManager;
@@ -565,7 +581,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Connection client properties (client info used in negotiating with the server)
+ * Connection client properties (client info used in negotiating with the
+ * server)
*/
public void setClientProperties(Map<String, Object> clientProperties) {
this.clientProperties = clientProperties;
@@ -576,7 +593,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Enables connection automatic recovery (uses connection implementation that performs automatic recovery when connection shutdown is not initiated by the application)
+ * Enables connection automatic recovery (uses connection implementation
+ * that performs automatic recovery when connection shutdown is not
+ * initiated by the application)
*/
public void setAutomaticRecoveryEnabled(Boolean automaticRecoveryEnabled) {
this.automaticRecoveryEnabled = automaticRecoveryEnabled;
@@ -587,7 +606,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Network recovery interval in milliseconds (interval used when recovering from network failure)
+ * Network recovery interval in milliseconds (interval used when recovering
+ * from network failure)
*/
public void setNetworkRecoveryInterval(Integer networkRecoveryInterval) {
this.networkRecoveryInterval = networkRecoveryInterval;
@@ -598,7 +618,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Enables connection topology recovery (should topology recovery be performed?)
+ * Enables connection topology recovery (should topology recovery be
+ * performed?)
*/
public void setTopologyRecoveryEnabled(Boolean topologyRecoveryEnabled) {
this.topologyRecoveryEnabled = topologyRecoveryEnabled;
@@ -609,16 +630,18 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Enables the quality of service on the RabbitMQConsumer side.
- * You need to specify the option of prefetchSize, prefetchCount, prefetchGlobal at the same time
+ * Enables the quality of service on the RabbitMQConsumer side. You need to
+ * specify the option of prefetchSize, prefetchCount, prefetchGlobal at the
+ * same time
*/
public void setPrefetchEnabled(boolean prefetchEnabled) {
this.prefetchEnabled = prefetchEnabled;
}
/**
- * The maximum amount of content (measured in octets) that the server will deliver, 0 if unlimited.
- * You need to specify the option of prefetchSize, prefetchCount, prefetchGlobal at the same time
+ * The maximum amount of content (measured in octets) that the server will
+ * deliver, 0 if unlimited. You need to specify the option of prefetchSize,
+ * prefetchCount, prefetchGlobal at the same time
*/
public void setPrefetchSize(int prefetchSize) {
this.prefetchSize = prefetchSize;
@@ -629,8 +652,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * The maximum number of messages that the server will deliver, 0 if unlimited.
- * You need to specify the option of prefetchSize, prefetchCount, prefetchGlobal at the same time
+ * The maximum number of messages that the server will deliver, 0 if
+ * unlimited. You need to specify the option of prefetchSize, prefetchCount,
+ * prefetchGlobal at the same time
*/
public void setPrefetchCount(int prefetchCount) {
this.prefetchCount = prefetchCount;
@@ -641,8 +665,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * If the settings should be applied to the entire channel rather than each consumer
- * You need to specify the option of prefetchSize, prefetchCount, prefetchGlobal at the same time
+ * If the settings should be applied to the entire channel rather than each
+ * consumer You need to specify the option of prefetchSize, prefetchCount,
+ * prefetchGlobal at the same time
*/
public void setPrefetchGlobal(boolean prefetchGlobal) {
this.prefetchGlobal = prefetchGlobal;
@@ -657,7 +682,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Number of concurrent consumers when consuming from broker. (eg similar as to the same option for the JMS component).
+ * Number of concurrent consumers when consuming from broker. (eg similar as
+ * to the same option for the JMS component).
*/
public void setConcurrentConsumers(int concurrentConsumers) {
this.concurrentConsumers = concurrentConsumers;
@@ -668,8 +694,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * If the option is true, camel declare the exchange and queue name and bind them together.
- * If the option is false, camel won't declare the exchange and queue name on the server.
+ * If the option is true, camel declare the exchange and queue name and bind
+ * them together. If the option is false, camel won't declare the exchange
+ * and queue name on the server.
*/
public void setDeclare(boolean declare) {
this.declare = declare;
@@ -735,7 +762,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Set the maximum number of milliseconds to wait for a channel from the pool
+ * Set the maximum number of milliseconds to wait for a channel from the
+ * pool
*/
public void setChannelPoolMaxWait(long channelPoolMaxWait) {
this.channelPoolMaxWait = channelPoolMaxWait;
@@ -746,9 +774,10 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * This flag tells the server how to react if the message cannot be routed to a queue.
- * If this flag is set, the server will return an unroutable message with a Return method.
- * If this flag is zero, the server silently drops the message.
+ * This flag tells the server how to react if the message cannot be routed
+ * to a queue. If this flag is set, the server will return an unroutable
+ * message with a Return method. If this flag is zero, the server silently
+ * drops the message.
* <p/>
* If the header is present rabbitmq.MANDATORY it will override this option.
*/
@@ -761,9 +790,11 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * This flag tells the server how to react if the message cannot be routed to a queue consumer immediately.
- * If this flag is set, the server will return an undeliverable message with a Return method.
- * If this flag is zero, the server will queue the message, but with no guarantee that it will ever be consumed.
+ * This flag tells the server how to react if the message cannot be routed
+ * to a queue consumer immediately. If this flag is set, the server will
+ * return an undeliverable message with a Return method. If this flag is
+ * zero, the server will queue the message, but with no guarantee that it
+ * will ever be consumed.
* <p/>
* If the header is present rabbitmq.IMMEDIATE it will override this option.
*/
@@ -772,17 +803,15 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Specify arguments for configuring the different RabbitMQ concepts, a different prefix is
- * required for each:
+ * Specify arguments for configuring the different RabbitMQ concepts, a
+ * different prefix is required for each:
* <ul>
- * <li>Exchange: arg.exchange.</li>
- * <li>Queue: arg.queue.</li>
- * <li>Binding: arg.binding.</li>
+ * <li>Exchange: arg.exchange.</li>
+ * <li>Queue: arg.queue.</li>
+ * <li>Binding: arg.binding.</li>
* </ul>
* For example to declare a queue with message ttl argument:
- *
* http://localhost:5672/exchange/queue?args=arg.queue.x-message-ttl=60000
- *
*/
public void setArgs(Map<String, Object> args) {
this.args = args;
@@ -820,7 +849,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Key/value args for configuring the queue binding parameters when declare=true
+ * Key/value args for configuring the queue binding parameters when
+ * declare=true
*
* @Deprecated Use args instead e.g arg.binding.foo=bar
*/
@@ -850,7 +880,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Set the configurer for setting the exchange args in Channel.exchangeDeclare
+ * Set the configurer for setting the exchange args in
+ * Channel.exchangeDeclare
*
* @Deprecated Use args instead e.g arg.exchange.x-message-ttl=1000
*/
@@ -859,7 +890,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Set timeout for waiting for a reply when using the InOut Exchange Pattern (in milliseconds)
+ * Set timeout for waiting for a reply when using the InOut Exchange Pattern
+ * (in milliseconds)
*/
public void setRequestTimeout(long requestTimeout) {
this.requestTimeout = requestTimeout;
@@ -888,7 +920,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * When true and an inOut Exchange failed on the consumer side send the caused Exception back in the response
+ * When true and an inOut Exchange failed on the consumer side send the
+ * caused Exception back in the response
*/
public void setTransferException(boolean transferException) {
this.transferException = transferException;
@@ -899,7 +932,9 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * When true, the message will be published with <a href="https://www.rabbitmq.com/confirms.html">publisher acknowledgements</a> turned on
+ * When true, the message will be published with
+ * <a href="https://www.rabbitmq.com/confirms.html">publisher
+ * acknowledgements</a> turned on
*/
public boolean isPublisherAcknowledgements() {
return publisherAcknowledgements;
@@ -910,7 +945,8 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * The amount of time in milliseconds to wait for a basic.ack response from RabbitMQ server
+ * The amount of time in milliseconds to wait for a basic.ack response from
+ * RabbitMQ server
*/
public long getPublisherAcknowledgementsTimeout() {
return publisherAcknowledgementsTimeout;
@@ -921,12 +957,11 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * When true, an exception will be thrown when the message cannot be delivered (basic.return) and the message is
- * marked as mandatory.
- * PublisherAcknowledgement will also be activated in this case
- *
- * See also <a href=https://www.rabbitmq.com/confirms.html">publisher acknowledgements</a> - When will messages be
- * confirmed?
+ * When true, an exception will be thrown when the message cannot be
+ * delivered (basic.return) and the message is marked as mandatory.
+ * PublisherAcknowledgement will also be activated in this case See also <a
+ * href=https://www.rabbitmq.com/confirms.html">publisher
+ * acknowledgements</a> - When will messages be confirmed?
*/
public boolean isGuaranteedDeliveries() {
return guaranteedDeliveries;
@@ -955,12 +990,13 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
}
/**
- * Exclusive queues may only be accessed by the current connection, and are deleted when that connection closes.
+ * Exclusive queues may only be accessed by the current connection, and are
+ * deleted when that connection closes.
*/
public void setExclusive(boolean exclusive) {
this.exclusive = exclusive;
}
-
+
public boolean isPassive() {
return passive;
}
@@ -971,6 +1007,5 @@ public class RabbitMQEndpoint extends DefaultEndpoint implements AsyncEndpoint {
public void setPassive(boolean passive) {
this.passive = passive;
}
-
-
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/a734ce3c/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
index 4ad3257..2373ae7 100644
--- a/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
+++ b/components/camel-rabbitmq/src/test/java/org/apache/camel/component/rabbitmq/RabbitMQEndpointTest.java
@@ -27,18 +27,18 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeoutException;
-import org.apache.camel.Exchange;
-import org.apache.camel.impl.JndiRegistry;
-import org.apache.camel.test.junit4.CamelTestSupport;
-import org.junit.Test;
-import org.mockito.Mockito;
-
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Address;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.impl.LongStringHelper;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+import org.mockito.Mockito;
+
public class RabbitMQEndpointTest extends CamelTestSupport {
private Envelope envelope = Mockito.mock(Envelope.class);
@@ -124,7 +124,7 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
customHeaders.put("byteArrayHeader", "foo".getBytes());
customHeaders.put("longStringHeader", LongStringHelper.asLongString("Some really long string"));
customHeaders.put("timestampHeader", new Timestamp(4200));
- customHeaders.put("byteHeader", new Byte((byte) 0));
+ customHeaders.put("byteHeader", new Byte((byte)0));
customHeaders.put("floatHeader", new Float(42.4242));
customHeaders.put("longHeader", new Long(420000000000000000L));
Mockito.when(properties.getHeaders()).thenReturn(customHeaders);
@@ -140,10 +140,10 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
assertEquals(42.24, exchange.getIn().getHeader("doubleHeader"));
assertEquals(true, exchange.getIn().getHeader("booleanHeader"));
assertEquals(new Date(0), exchange.getIn().getHeader("dateHeader"));
- assertArrayEquals("foo".getBytes(), (byte[]) exchange.getIn().getHeader("byteArrayHeader"));
+ assertArrayEquals("foo".getBytes(), (byte[])exchange.getIn().getHeader("byteArrayHeader"));
assertEquals("Some really long string", exchange.getIn().getHeader("longStringHeader"));
assertEquals(new Timestamp(4200), exchange.getIn().getHeader("timestampHeader"));
- assertEquals(new Byte((byte) 0), exchange.getIn().getHeader("byteHeader"));
+ assertEquals(new Byte((byte)0), exchange.getIn().getHeader("byteHeader"));
assertEquals(new Float(42.4242), exchange.getIn().getHeader("floatHeader"));
assertEquals(new Long(420000000000000000L), exchange.getIn().getHeader("longHeader"));
assertEquals(body, exchange.getIn().getBody());
@@ -154,7 +154,7 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?threadPoolSize=20", RabbitMQEndpoint.class);
assertEquals(20, endpoint.getThreadPoolSize());
- ThreadPoolExecutor executor = assertIsInstanceOf(ThreadPoolExecutor.class, endpoint.createExecutor());
+ ThreadPoolExecutor executor = assertIsInstanceOf(ThreadPoolExecutor.class, endpoint.createExecutor());
assertEquals(20, executor.getCorePoolSize());
}
@@ -209,7 +209,8 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
@Test
public void testMultiArgsPopulateCorrectEndpointProperties() throws Exception {
- RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?arg.exchange.e1=v1&arg.exchange.e2=v2&arg.queue.q1=v3&arg.binding.b1=v4", RabbitMQEndpoint.class);
+ RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?arg.exchange.e1=v1&arg.exchange.e2=v2&arg.queue.q1=v3&arg.binding.b1=v4",
+ RabbitMQEndpoint.class);
assertEquals("Wrong number of args", 4, endpoint.getArgs().size());
assertEquals("Wrong number of args", 1, endpoint.getBindingArgs().size());
assertEquals("Wrong number of args", 2, endpoint.getExchangeArgs().size());
@@ -220,10 +221,8 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
public void testMultiArgsCombinedWithIndividuallySpecified() throws Exception {
// setup two arguments for each rabbit fundamental.
// Configured inline and via named map in the camel registry
- RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange"
- + "?arg.exchange.e1=v1&exchangeArgs=#args"
- + "&arg.queue.q1=v2&queueArgs=#moreArgs"
- + "&arg.binding.b1=v3&bindingArgs=#evenMoreArgs", RabbitMQEndpoint.class);
+ RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange" + "?arg.exchange.e1=v1&exchangeArgs=#args" + "&arg.queue.q1=v2&queueArgs=#moreArgs"
+ + "&arg.binding.b1=v3&bindingArgs=#evenMoreArgs", RabbitMQEndpoint.class);
// The multi-value inline has 3
Map<String, Object> inlineArgs = endpoint.getArgs();
@@ -288,17 +287,9 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
@Test
public void testCreateConnectionFactoryCustom() throws Exception {
- ConnectionFactory connectionFactory = createConnectionFactory("rabbitmq:localhost:1234/exchange"
- + "?username=userxxx"
- + "&password=passxxx"
- + "&connectionTimeout=123"
- + "&requestedChannelMax=456"
- + "&requestedFrameMax=789"
- + "&requestedHeartbeat=987"
- + "&sslProtocol=true"
- + "&automaticRecoveryEnabled=true"
- + "&networkRecoveryInterval=654"
- + "&topologyRecoveryEnabled=false");
+ ConnectionFactory connectionFactory = createConnectionFactory("rabbitmq:localhost:1234/exchange" + "?username=userxxx" + "&password=passxxx" + "&connectionTimeout=123"
+ + "&requestedChannelMax=456" + "&requestedFrameMax=789" + "&requestedHeartbeat=987" + "&sslProtocol=true"
+ + "&automaticRecoveryEnabled=true" + "&networkRecoveryInterval=654" + "&topologyRecoveryEnabled=false");
assertEquals("localhost", connectionFactory.getHost());
assertEquals(1234, connectionFactory.getPort());
@@ -337,13 +328,13 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?skipQueueDeclare=true", RabbitMQEndpoint.class);
assertTrue(endpoint.isSkipQueueDeclare());
}
-
+
@Test
public void createEndpointWithSkipExchangeDeclareEnabled() throws Exception {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?skipExchangeDeclare=true", RabbitMQEndpoint.class);
assertTrue(endpoint.isSkipExchangeDeclare());
}
-
+
@Test
public void createEndpointWithSkipQueueBindEndabled() throws Exception {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?SkipQueueBind=true", RabbitMQEndpoint.class);
@@ -355,7 +346,7 @@ public class RabbitMQEndpointTest extends CamelTestSupport {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?exclusive=true", RabbitMQEndpoint.class);
assertTrue(endpoint.isExclusive());
}
-
+
@Test
public void createEndpointWithPassiveEnabled() throws Exception {
RabbitMQEndpoint endpoint = context.getEndpoint("rabbitmq:localhost/exchange?passive=true", RabbitMQEndpoint.class);
[3/3] camel git commit: CAMEL-11704: Docs
Posted by ac...@apache.org.
CAMEL-11704: Docs
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/0c55b9c6
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/0c55b9c6
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/0c55b9c6
Branch: refs/heads/master
Commit: 0c55b9c60311a54458e169d64b14cb68ca599466
Parents: a734ce3
Author: Andrea Cosentino <an...@gmail.com>
Authored: Fri Aug 25 14:01:23 2017 +0200
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Fri Aug 25 14:01:23 2017 +0200
----------------------------------------------------------------------
components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/0c55b9c6/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc b/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
index c147441..d29472f 100644
--- a/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
+++ b/components/camel-rabbitmq/src/main/docs/rabbitmq-component.adoc
@@ -61,7 +61,7 @@ with the following path and query parameters:
| **exchangeName** | *Required* The exchange name determines which exchange produced messages will sent to. In the case of consumers the exchange name determines which exchange the queue will bind to. | | String
|=======================================================================
-#### Query Parameters (57 parameters):
+#### Query Parameters (58 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|=======================================================================
@@ -76,6 +76,7 @@ with the following path and query parameters:
| **durable** (common) | If we are declaring a durable exchange (the exchange will survive a server restart) | true | boolean
| **exchangeType** (common) | The exchange type such as direct or topic. | direct | String
| **exclusive** (common) | Exclusive queues may only be accessed by the current connection and are deleted when that connection closes. | false | boolean
+| **passive** (common) | Passive queues depend on the queue already to be available at RabbitMQ. | false | boolean
| **queue** (common) | The queue to receive messages from | | String
| **routingKey** (common) | The routing key to use when binding a consumer queue to the exchange. For producer routing keys you set the header rabbitmq.ROUTING_KEY. | | String
| **skipExchangeDeclare** (common) | This can be used if we need to declare the queue but not the exchange | false | boolean
@@ -95,7 +96,7 @@ with the following path and query parameters:
| **bridgeEndpoint** (producer) | If the bridgeEndpoint is true the producer will ignore the message header of rabbitmq.EXCHANGE_NAME and rabbitmq.ROUTING_KEY | false | boolean
| **channelPoolMaxSize** (producer) | Get maximum number of opened channel in pool | 10 | int
| **channelPoolMaxWait** (producer) | Set the maximum number of milliseconds to wait for a channel from the pool | 1000 | long
-| **guaranteedDeliveries** (producer) | When true an exception will be thrown when the message cannot be delivered (basic.return) and the message is marked as mandatory. PublisherAcknowledgement will also be activated in this case See also publisher acknowledgements - When will messages be confirmed | false | boolean
+| **guaranteedDeliveries** (producer) | When true an exception will be thrown when the message cannot be delivered (basic.return) and the message is marked as mandatory. PublisherAcknowledgement will also be activated in this case See also a href=https://www.rabbitmq.com/confirms.htmlpublisher acknowledgements - When will messages be confirmed | false | boolean
| **immediate** (producer) | This flag tells the server how to react if the message cannot be routed to a queue consumer immediately. If this flag is set the server will return an undeliverable message with a Return method. If this flag is zero the server will queue the message but with no guarantee that it will ever be consumed. If the header is present rabbitmq.IMMEDIATE it will override this option. | false | boolean
| **mandatory** (producer) | This flag tells the server how to react if the message cannot be routed to a queue. If this flag is set the server will return an unroutable message with a Return method. If this flag is zero the server silently drops the message. If the header is present rabbitmq.MANDATORY it will override this option. | false | boolean
| **publisherAcknowledgements** (producer) | When true the message will be published with publisher acknowledgements turned on | false | boolean